Nexus Agent
c9e08799f9
Docker 向导提示 nx cron/宿主机路径；裸机 cron 含 db_backup；验收脚本检查 cron 与 MySQL 备份能力。 Co-authored-by: Cursor <cursoragent@cursor.com>
577 lines
20 KiB
Python
577 lines
20 KiB
Python
"""P0/P2 security regression unit tests (no live server required)."""
|
||
|
||
from pathlib import Path
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
from starlette.requests import Request
|
||
|
||
from server.api.dependencies import check_dangerous_command
|
||
from server.application.services.script_jobs import master_callback_url
|
||
from server.infrastructure.redis.script_execution_store import _parse_server_ids_json
|
||
from server.infrastructure.telegram import sanitize_external_message
|
||
|
||
|
||
def test_dangerous_command_rm_rf_detected():
|
||
warnings = check_dangerous_command("rm -rf /")
|
||
assert warnings
|
||
|
||
|
||
def test_dangerous_command_safe_empty():
|
||
assert check_dangerous_command("echo hello") == []
|
||
|
||
|
||
def test_parse_server_ids_invalid_json():
|
||
assert _parse_server_ids_json("not-json") == []
|
||
|
||
|
||
def test_parse_server_ids_valid():
|
||
assert _parse_server_ids_json("[1, 2]") == [1, 2]
|
||
|
||
|
||
def test_master_callback_url_rejects_http_non_local(monkeypatch):
|
||
monkeypatch.setattr(
|
||
"server.application.services.script_jobs.settings.API_BASE_URL",
|
||
"http://api.example.com",
|
||
)
|
||
with pytest.raises(ValueError, match="HTTPS"):
|
||
master_callback_url()
|
||
|
||
|
||
def test_master_callback_url_allows_localhost_http(monkeypatch):
|
||
monkeypatch.setattr(
|
||
"server.application.services.script_jobs.settings.API_BASE_URL",
|
||
"http://localhost:8600",
|
||
)
|
||
assert master_callback_url().endswith("/api/agent/script-callback")
|
||
|
||
|
||
def test_sanitize_external_message_redacts_mysql_url():
|
||
raw = "connect failed: mysql+pymysql://nexus:SecretPass@db.internal:3306/nexus"
|
||
cleaned = sanitize_external_message(raw)
|
||
assert "SecretPass" not in cleaned
|
||
assert "mysql+pymysql://" not in cleaned
|
||
assert "<connection-url-redacted>" in cleaned
|
||
|
||
|
||
def test_sanitize_external_message_redacts_bearer_token():
|
||
raw = "auth failed Bearer eyJhbGciOiJIUzI1NiJ9.payload.sig"
|
||
cleaned = sanitize_external_message(raw)
|
||
assert "eyJhbGci" not in cleaned
|
||
assert "Bearer <redacted>" in cleaned
|
||
|
||
|
||
def test_sanitize_external_message_truncates_long_text():
|
||
assert len(sanitize_external_message("x" * 500)) <= 240
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_jwt_middleware_blocks_missing_token(monkeypatch):
|
||
from server.api import auth_jwt
|
||
|
||
monkeypatch.setattr(auth_jwt, "_is_install_mode", lambda: False)
|
||
|
||
inner = AsyncMock()
|
||
middleware = auth_jwt.JwtAuthMiddleware(app=inner)
|
||
scope = {
|
||
"type": "http",
|
||
"method": "GET",
|
||
"path": "/api/servers/",
|
||
"headers": [],
|
||
"query_string": b"",
|
||
}
|
||
|
||
async def receive():
|
||
return {"type": "http.request", "body": b"", "more_body": False}
|
||
|
||
sent: list[dict] = []
|
||
|
||
async def send(message):
|
||
sent.append(message)
|
||
|
||
await middleware(scope, receive, send)
|
||
starts = [m for m in sent if m.get("type") == "http.response.start"]
|
||
assert starts and starts[0].get("status") == 401
|
||
inner.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_jwt_middleware_skips_public_login(monkeypatch):
|
||
from server.api import auth_jwt
|
||
|
||
monkeypatch.setattr(auth_jwt, "_is_install_mode", lambda: False)
|
||
|
||
inner = AsyncMock()
|
||
middleware = auth_jwt.JwtAuthMiddleware(app=inner)
|
||
scope = {
|
||
"type": "http",
|
||
"method": "POST",
|
||
"path": "/api/auth/login",
|
||
"headers": [],
|
||
"query_string": b"",
|
||
}
|
||
|
||
async def receive():
|
||
return {"type": "http.request", "body": b"", "more_body": False}
|
||
|
||
async def send(message):
|
||
pass
|
||
|
||
await middleware(scope, receive, send)
|
||
inner.assert_called_once()
|
||
|
||
|
||
def test_public_path_install_and_logout():
|
||
from server.api.auth_jwt import _is_public_path
|
||
|
||
assert _is_public_path("/api/install/status") is True
|
||
assert _is_public_path("/api/auth/logout") is True
|
||
|
||
|
||
def _cdn_offenders_in_text(path: Path, text: str) -> bool:
|
||
return "cdn.jsdelivr" in text or "unpkg.com" in text
|
||
|
||
|
||
def test_html_pages_do_not_load_external_cdn():
|
||
app_dir = Path(__file__).resolve().parent.parent / "web" / "app"
|
||
offenders = []
|
||
for html in app_dir.glob("*.html"):
|
||
if _cdn_offenders_in_text(html, html.read_text(encoding="utf-8")):
|
||
offenders.append(html.name)
|
||
assert offenders == [], f"CDN still referenced in: {offenders}"
|
||
|
||
|
||
def test_app_root_js_do_not_load_external_cdn():
|
||
app_dir = Path(__file__).resolve().parent.parent / "web" / "app"
|
||
offenders = []
|
||
for js in app_dir.glob("*.js"):
|
||
if _cdn_offenders_in_text(js, js.read_text(encoding="utf-8")):
|
||
offenders.append(js.name)
|
||
assert offenders == [], f"CDN still referenced in: {offenders}"
|
||
|
||
|
||
def test_monaco_vendor_loader_present():
|
||
loader = (
|
||
Path(__file__).resolve().parent.parent
|
||
/ "web"
|
||
/ "app"
|
||
/ "vendor"
|
||
/ "monaco"
|
||
/ "vs"
|
||
/ "loader.js"
|
||
)
|
||
assert loader.is_file(), "Run: python scripts/sync_frontend_vendor.py (needs network)"
|
||
|
||
|
||
def test_vendor_manifest_lists_core_assets():
|
||
manifest_path = Path(__file__).resolve().parent.parent / "web" / "app" / "vendor" / "manifest.json"
|
||
assert manifest_path.is_file()
|
||
import json
|
||
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
for name in ("tailwindcss-browser.js", "alpinejs.min.js"):
|
||
assert name in manifest["files"]
|
||
assert (manifest_path.parent / name).is_file()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_install_test_credentials_rejects_when_env_exists(monkeypatch, tmp_path):
|
||
from fastapi import HTTPException
|
||
from server.api import install as install_api
|
||
from server.api.install import InitDbRequest, test_credentials
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("NEXUS_SECRET_KEY=test\n", encoding="utf-8")
|
||
monkeypatch.setattr(install_api, "ENV_FILE", env_file)
|
||
monkeypatch.setattr(install_api, "INSTALL_LOCK", tmp_path / ".install_locked")
|
||
|
||
req = InitDbRequest(db_pass="secret")
|
||
with pytest.raises(HTTPException) as exc:
|
||
await test_credentials(req)
|
||
assert exc.value.status_code == 400
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_install_test_credentials_all_pass(monkeypatch, tmp_path):
|
||
from server.api import install as install_api
|
||
from server.api.install import InitDbRequest, test_credentials
|
||
|
||
monkeypatch.setattr(install_api, "ENV_FILE", tmp_path / ".env")
|
||
monkeypatch.setattr(install_api, "INSTALL_LOCK", tmp_path / ".install_locked")
|
||
|
||
async def _mysql_ok(_url):
|
||
return True, "✓ MySQL 连接正常"
|
||
|
||
monkeypatch.setattr(install_api, "_test_mysql_url", _mysql_ok)
|
||
monkeypatch.setattr(install_api, "_test_redis_url", lambda _url: (True, "✓ Redis 连接正常"))
|
||
|
||
req = InitDbRequest(db_pass="secret")
|
||
out = await test_credentials(req)
|
||
assert out["all_pass"] is True
|
||
assert len(out["checks"]) == 2
|
||
|
||
|
||
def test_build_redis_url_password_with_leading_colon():
|
||
from server.api.install import InitDbRequest, _build_redis_url
|
||
|
||
req = InitDbRequest(
|
||
db_pass="db",
|
||
redis_host="1Panel-redis-x",
|
||
redis_pass="redis_WP3NyN",
|
||
)
|
||
assert _build_redis_url(req) == "redis://:redis_WP3NyN@1Panel-redis-x:6379/0"
|
||
|
||
|
||
def test_build_redis_url_with_default_user():
|
||
from server.api.install import InitDbRequest, _build_redis_url
|
||
|
||
req = InitDbRequest(
|
||
db_pass="db",
|
||
redis_host="1Panel-redis-x",
|
||
redis_user="default",
|
||
redis_pass="redis_WP3NyN",
|
||
)
|
||
assert _build_redis_url(req) == "redis://default:redis_WP3NyN@1Panel-redis-x:6379/0"
|
||
|
||
|
||
def test_resolve_redis_url_falls_back_to_colon_pass(monkeypatch):
|
||
from server.api.install import InitDbRequest, _resolve_redis_url, _sanitize_redis_request
|
||
|
||
monkeypatch.setenv("NEXUS_DOCKER_WRITE_ENV", "1")
|
||
req = InitDbRequest(
|
||
db_pass="db",
|
||
redis_host="1Panel-redis-x",
|
||
redis_user="root",
|
||
redis_pass="secret",
|
||
)
|
||
calls: list[str] = []
|
||
|
||
def fake_test(url: str) -> tuple[bool, str]:
|
||
calls.append(url)
|
||
return url.startswith("redis://:secret@"), "ok" if url.startswith("redis://:secret@") else "fail"
|
||
|
||
monkeypatch.setattr("server.api.install._test_redis_url", fake_test)
|
||
ok, msg, url = _resolve_redis_url(_sanitize_redis_request(req))
|
||
assert ok is True
|
||
assert url == "redis://:secret@1Panel-redis-x:6379/0"
|
||
assert "仅密码" in msg
|
||
assert not any(u.startswith("redis://root:") for u in calls)
|
||
|
||
|
||
def test_redis_url_candidates_never_tries_root_in_docker_mode(monkeypatch):
|
||
from server.api.install import InitDbRequest, _redis_url_candidates, _sanitize_redis_request
|
||
|
||
monkeypatch.setenv("NEXUS_DOCKER_WRITE_ENV", "1")
|
||
req = InitDbRequest(
|
||
db_pass="db",
|
||
redis_host="1Panel-redis-x",
|
||
redis_user="root",
|
||
redis_pass="secret",
|
||
)
|
||
labels = [label for label, url in _redis_url_candidates(_sanitize_redis_request(req))]
|
||
urls = [url for _, url in _redis_url_candidates(_sanitize_redis_request(req))]
|
||
assert "root" not in " ".join(labels).lower()
|
||
assert not any("root:" in u for u in urls)
|
||
|
||
|
||
def test_configure_guardian_docker_mode_skips_supervisor(monkeypatch):
|
||
from server.api import install as install_api
|
||
from server.api.install import _configure_docker_guardian, _configure_guardian
|
||
|
||
monkeypatch.setenv("NEXUS_DOCKER_WRITE_ENV", "1")
|
||
monkeypatch.setenv("NEXUS_API_BASE_URL", "https://api.example.com")
|
||
monkeypatch.setattr(
|
||
install_api.Path,
|
||
"is_file",
|
||
lambda self: str(self) == "/.dockerenv",
|
||
)
|
||
|
||
monkeypatch.setenv("NEXUS_HOST_ROOT", "/opt/nexus")
|
||
docker_results = _configure_docker_guardian("/app", "8600")
|
||
docker_text = "\n".join(docker_results)
|
||
assert "Supervisor" not in docker_text
|
||
assert "self_monitor" in docker_text
|
||
assert "OpenResty" in docker_text or "反代" in docker_text
|
||
assert "install_ops_cron" in docker_text or "nx cron" in docker_text
|
||
assert "/opt/nexus" in docker_text
|
||
|
||
bare_results = _configure_guardian("/opt/nexus", "8600")
|
||
assert bare_results == docker_results
|
||
|
||
|
||
def test_redis_repair_request_uses_install_state_password(tmp_path, monkeypatch):
|
||
from server.api import install as install_api
|
||
from server.api.install import _redis_repair_request
|
||
|
||
state_file = tmp_path / ".install_state.json"
|
||
state_file.write_text(
|
||
'{"redis_pass":"secret-from-state","step":4}',
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(install_api, "STATE_FILE", state_file)
|
||
|
||
req = _redis_repair_request("redis://1Panel-redis-x:6379/0")
|
||
assert req is not None
|
||
assert req.redis_pass == "secret-from-state"
|
||
assert req.redis_host == "1panel-redis-x"
|
||
|
||
|
||
def test_init_db_request_from_redis_url_legacy_and_colon_pass():
|
||
from server.api.install import _init_db_request_from_redis_url
|
||
|
||
legacy = _init_db_request_from_redis_url("redis://badpass@1Panel-redis-x:6379/0")
|
||
assert legacy is not None
|
||
assert legacy.redis_pass == "badpass"
|
||
assert legacy.redis_user == ""
|
||
|
||
colon = _init_db_request_from_redis_url("redis://:secret@1Panel-redis-x:6379/0")
|
||
assert colon is not None
|
||
assert colon.redis_pass == "secret"
|
||
assert colon.redis_user == ""
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_connection_check_repairs_bad_redis_url(monkeypatch, tmp_path):
|
||
from server.api import install as install_api
|
||
from server.api.install import connection_check
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text(
|
||
'NEXUS_DATABASE_URL="mysql+aiomysql://n:pass@h:3306/n"\n'
|
||
'NEXUS_REDIS_URL="redis://root:secret@1Panel-redis-x:6379/0"\n',
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(install_api, "ENV_FILE", env_file)
|
||
monkeypatch.setattr(install_api, "INSTALL_LOCK", tmp_path / ".install_locked")
|
||
|
||
async def _mysql_ok(_url):
|
||
return True, "✓ MySQL"
|
||
|
||
calls: list[str] = []
|
||
|
||
def fake_redis(url: str) -> tuple[bool, str]:
|
||
calls.append(url)
|
||
if url.startswith("redis://:secret@"):
|
||
return True, "✓ Redis 连接正常（仅密码 (:pass@)）"
|
||
return False, "auth fail"
|
||
|
||
monkeypatch.setattr(install_api, "_test_mysql_url", _mysql_ok)
|
||
monkeypatch.setattr(install_api, "_test_redis_url", fake_redis)
|
||
monkeypatch.setenv("NEXUS_DOCKER_WRITE_ENV", "1")
|
||
|
||
out = await connection_check()
|
||
assert out["all_pass"] is True
|
||
assert out.get("redis_repaired") is True
|
||
assert "redis://:secret@" in env_file.read_text(encoding="utf-8")
|
||
|
||
|
||
def test_install_reject_repeat_init_when_env_exists(monkeypatch, tmp_path):
|
||
from fastapi import HTTPException
|
||
from server.api import install as install_api
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("NEXUS_SECRET_KEY=test\n", encoding="utf-8")
|
||
monkeypatch.setattr(install_api, "ENV_FILE", env_file)
|
||
monkeypatch.setattr(install_api, "INSTALL_LOCK", tmp_path / ".install_locked")
|
||
|
||
with pytest.raises(HTTPException) as exc:
|
||
install_api._reject_if_env_exists()
|
||
assert exc.value.status_code == 403
|
||
|
||
|
||
def test_install_create_admin_allowed_when_env_not_locked(monkeypatch, tmp_path):
|
||
from server.api import install as install_api
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text("NEXUS_SECRET_KEY=test\n", encoding="utf-8")
|
||
monkeypatch.setattr(install_api, "ENV_FILE", env_file)
|
||
monkeypatch.setattr(install_api, "INSTALL_LOCK", tmp_path / ".install_locked")
|
||
|
||
install_api._reject_if_locked() # should not raise
|
||
|
||
|
||
def test_script_aggregate_empty_server_ids_pending():
|
||
from server.application.services.script_job_callback import _aggregate_execution_status
|
||
|
||
status = _aggregate_execution_status(
|
||
[],
|
||
{"1": {"phase": "pending"}, "2": {"phase": "done", "exit_code": 0}},
|
||
)
|
||
assert status == "running"
|
||
|
||
|
||
def test_script_aggregate_empty_server_ids_no_pending():
|
||
from server.application.services.script_job_callback import _aggregate_execution_status
|
||
|
||
assert _aggregate_execution_status([], {}) == "failed"
|
||
|
||
|
||
def test_archive_install_wizard_html(monkeypatch, tmp_path):
|
||
from server.api import install as install_api
|
||
|
||
app_dir = tmp_path / "web" / "app"
|
||
app_dir.mkdir(parents=True)
|
||
html = app_dir / "install.html"
|
||
html.write_text("<html>wizard</html>", encoding="utf-8")
|
||
monkeypatch.setattr(install_api, "INSTALL_HTML", html)
|
||
monkeypatch.setattr(install_api, "INSTALL_HTML_BAK", app_dir / "install.html.bak")
|
||
|
||
install_api._archive_install_wizard_html()
|
||
|
||
assert not html.is_file()
|
||
assert (app_dir / "install.html.bak").read_text(encoding="utf-8") == "<html>wizard</html>"
|
||
|
||
|
||
def test_sync_lock_to_persist_volume(monkeypatch, tmp_path):
|
||
from server.api import install as install_api
|
||
|
||
persist_dir = tmp_path / "state"
|
||
persist_dir.mkdir()
|
||
lock_file = tmp_path / ".install_locked"
|
||
lock_file.write_text("locked at test\n", encoding="utf-8")
|
||
monkeypatch.setattr(install_api, "INSTALL_LOCK", lock_file)
|
||
monkeypatch.setenv("NEXUS_PERSIST_DIR", str(persist_dir))
|
||
|
||
install_api._sync_lock_to_persist_volume()
|
||
|
||
dest = persist_dir / ".install_locked"
|
||
assert dest.is_file()
|
||
assert dest.read_text(encoding="utf-8") == "locked at test\n"
|
||
|
||
|
||
def test_install_reject_when_locked(monkeypatch, tmp_path):
|
||
from fastapi import HTTPException
|
||
from server.api import install as install_api
|
||
|
||
monkeypatch.setattr(install_api, "ENV_FILE", tmp_path / ".env")
|
||
lock_file = tmp_path / ".install_locked"
|
||
lock_file.write_text("locked\n", encoding="utf-8")
|
||
monkeypatch.setattr(install_api, "INSTALL_LOCK", lock_file)
|
||
|
||
with pytest.raises(HTTPException) as exc:
|
||
install_api._reject_if_locked()
|
||
assert exc.value.status_code == 403
|
||
|
||
|
||
def test_install_verify_token_rejects_missing(monkeypatch, tmp_path):
|
||
from fastapi import HTTPException
|
||
from server.api import install as install_api
|
||
|
||
state_file = tmp_path / ".install_state.json"
|
||
state_file.write_text('{"install_token":"abc123"}', encoding="utf-8")
|
||
monkeypatch.setattr(install_api, "STATE_FILE", state_file)
|
||
|
||
with pytest.raises(HTTPException) as exc:
|
||
install_api._verify_install_token("")
|
||
assert exc.value.status_code == 403
|
||
|
||
with pytest.raises(HTTPException) as exc:
|
||
install_api._verify_install_token("wrong")
|
||
assert exc.value.status_code == 403
|
||
|
||
install_api._verify_install_token("abc123")
|
||
|
||
|
||
def test_public_install_state_redacts_secrets():
|
||
from server.api.install import _public_install_state
|
||
|
||
state = {
|
||
"step": 4,
|
||
"db_host": "1Panel-mysql-x",
|
||
"install_token": "secret-token-abc",
|
||
"redis_pass": "redis-secret",
|
||
"db_port": "3306",
|
||
}
|
||
public = _public_install_state(state)
|
||
assert "install_token" not in public
|
||
assert "redis_pass" not in public
|
||
assert public.get("has_token") is True
|
||
assert public["db_host"] == "1Panel-mysql-x"
|
||
|
||
|
||
def test_public_install_state_no_has_token_when_missing():
|
||
from server.api.install import _public_install_state
|
||
|
||
public = _public_install_state({"step": 2, "db_host": "h"})
|
||
assert "has_token" not in public
|
||
|
||
|
||
def test_clear_state_redis_pass(tmp_path, monkeypatch):
|
||
from server.api import install as install_api
|
||
from server.api.install import _clear_state_redis_pass
|
||
|
||
state_file = tmp_path / ".install_state.json"
|
||
state_file.write_text(
|
||
'{"redis_pass":"secret","step":4,"install_token":"tok"}',
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(install_api, "STATE_FILE", state_file)
|
||
_clear_state_redis_pass()
|
||
import json
|
||
|
||
data = json.loads(state_file.read_text(encoding="utf-8"))
|
||
assert data.get("redis_pass") == ""
|
||
assert data.get("install_token") == "tok"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_connection_check_clears_redis_pass_after_repair(monkeypatch, tmp_path):
|
||
from server.api import install as install_api
|
||
from server.api.install import connection_check
|
||
|
||
env_file = tmp_path / ".env"
|
||
env_file.write_text(
|
||
'NEXUS_DATABASE_URL="mysql+aiomysql://n:pass@h:3306/n"\n'
|
||
'NEXUS_REDIS_URL="redis://root:secret@1Panel-redis-x:6379/0"\n',
|
||
encoding="utf-8",
|
||
)
|
||
state_file = tmp_path / ".install_state.json"
|
||
state_file.write_text(
|
||
'{"redis_pass":"secret","step":4,"install_token":"tok"}',
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(install_api, "ENV_FILE", env_file)
|
||
monkeypatch.setattr(install_api, "STATE_FILE", state_file)
|
||
monkeypatch.setattr(install_api, "INSTALL_LOCK", tmp_path / ".install_locked")
|
||
|
||
async def _mysql_ok(_url):
|
||
return True, "✓ MySQL"
|
||
|
||
def fake_redis(url: str) -> tuple[bool, str]:
|
||
if url.startswith("redis://:secret@"):
|
||
return True, "✓ Redis"
|
||
return False, "auth fail"
|
||
|
||
monkeypatch.setattr(install_api, "_test_mysql_url", _mysql_ok)
|
||
monkeypatch.setattr(install_api, "_test_redis_url", fake_redis)
|
||
monkeypatch.setenv("NEXUS_DOCKER_WRITE_ENV", "1")
|
||
|
||
out = await connection_check()
|
||
assert out["all_pass"] is True
|
||
import json
|
||
|
||
data = json.loads(state_file.read_text(encoding="utf-8"))
|
||
assert data.get("redis_pass") == ""
|
||
|
||
|
||
def test_init_db_mysql_error_uses_db_port_not_nameerror(monkeypatch, tmp_path):
|
||
"""Regression: init-db except branch must use int db_port, not undefined name."""
|
||
import inspect
|
||
|
||
from server.api import install as install_api
|
||
|
||
source = inspect.getsource(install_api.init_db)
|
||
assert "db_port = int(req.db_port)" in source
|
||
assert "_mysql_connect_error_detail(req.db_host, db_port)" in source
|
||
|
||
|
||
def test_resolve_nexus_push_source_path_rejects_system_prefix(tmp_path, monkeypatch):
|
||
from server.utils.posix_paths import PosixPathError, resolve_nexus_push_source_path
|
||
|
||
allowed = tmp_path / "push_src"
|
||
allowed.mkdir()
|
||
assert resolve_nexus_push_source_path(str(allowed)) == str(allowed.resolve())
|
||
|
||
with pytest.raises(PosixPathError, match="系统路径"):
|
||
resolve_nexus_push_source_path("/etc")
|