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

    docker_results = _configure_docker_guardian("/opt/nexus", "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

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