Nexus/tests/test_security_unit.py

"""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_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_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_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")