"""Credentials never reach results, errors, logs, the audit file or ``repr`` output.

Every secret the server knows (the Gitea token, the MCP bearer, the OAuth client secret,
the store key, a user's Gitea access and refresh tokens and our own issued tokens) is
planted in Gitea's answers, successful and failed, and in the incoming requests.
"""

from __future__ import annotations

import json
import logging
from typing import Any

import pytest
from pytest_httpx import HTTPXMock

from aegis_gitea_mcp.audit import get_audit_logger
from aegis_gitea_mcp.identity import Identity
from aegis_gitea_mcp.oauth.provider import PendingAuthorization
from aegis_gitea_mcp.oauth.store import SessionRecord
from aegis_gitea_mcp.oauth.upstream import UpstreamTokens
from aegis_gitea_mcp.transports import AccessLogQueryFilter, run_http
from tests.test_oauth_helpers import (
    API,
    APP_CLIENT_SECRET,
    BEARER_TOKEN,
    ENCRYPTION_KEY,
    GITEA_TOKEN,
    USER_REFRESH,
    USER_TOKEN,
    browser,
    build_oauth_app,
    mcp_client,
    oauth_settings,
    sign_in,
)


@pytest.fixture(autouse=True)
def _env(mock_env: None) -> None:
    """The audit logger and the rate limiter read the process-wide settings."""


def static_secrets() -> list[str]:
    return [GITEA_TOKEN, BEARER_TOKEN, APP_CLIENT_SECRET, ENCRYPTION_KEY, USER_TOKEN, USER_REFRESH]


def everything_secret(extra: list[str]) -> str:
    return " ".join([*static_secrets(), *extra])


def assert_clean(text: str, secrets: list[str]) -> None:
    for secret in secrets:
        assert secret not in text, f"secret leaked: {secret[:6]}..."


@pytest.mark.parametrize("mode", ["off", "mask", "block"])
async def test_no_secret_reaches_a_result_the_audit_file_or_the_logs(
    httpx_mock: HTTPXMock,
    caplog: pytest.LogCaptureFixture,
    capsys: pytest.CaptureFixture[str],
    mode: str,
) -> None:
    caplog.set_level(logging.DEBUG)
    oauth = build_oauth_app(oauth_settings(secret_detection_mode=mode))
    flow = sign_in(browser(oauth.app), httpx_mock)
    # The client secret the SDK issues at registration is plain hex without a recognisable
    # prefix: it is stored (encrypted) but cannot be pattern masked, and Gitea never sees it.
    issued = [flow.tokens["access_token"], flow.tokens["refresh_token"]]
    # SECRET_DETECTION_MODE=off disables pattern matching, which is what protects our own
    # issued agm_ tokens (only hashes of them live on the server, so they cannot be redacted
    # exactly); every other secret is redacted exactly in all modes.
    secrets = static_secrets() if mode == "off" else [*static_secrets(), *issued]
    # The admin identity is not the OAuth user: Gitea never shows it that user's tokens, so
    # only the server's own secrets (and, with pattern masking, our issued tokens) apply.
    admin_secrets = [GITEA_TOKEN, BEARER_TOKEN, APP_CLIENT_SECRET, ENCRYPTION_KEY]
    blob = everything_secret(issued)

    # The admin identity's lookup, then success and failure bodies that echo every secret.
    httpx_mock.add_response(url=f"{API}/user", json={"login": "admin"}, is_reusable=True)
    echo = {"is_reusable": True}
    httpx_mock.add_response(
        url=f"{API}/repos/octo/ok", json={"description": blob, blob: "key"}, **echo
    )
    httpx_mock.add_response(
        url=f"{API}/repos/octo/bad",
        status_code=422,
        json={"message": blob, "errors": [blob]},
        **echo,
    )
    httpx_mock.add_response(
        url=f"{API}/repos/octo/text", text=blob, headers={"content-type": "text/plain"}, **echo
    )
    httpx_mock.add_response(url=f"{API}/repos/octo/boom", status_code=500, text=blob, **echo)

    results: list[Any] = []
    # One app per identity: the SDK's session manager can only run once per app.
    apps = [
        (oauth.app, flow.tokens["access_token"]),
        (build_oauth_app(oauth.settings).app, BEARER_TOKEN),
    ]
    for app, token in apps:
        async with mcp_client(app, token) as mcp:
            for repo in ("ok", "bad", "boom"):
                results.append(
                    await mcp.call_tool("gitea_repo_get", {"owner": "octo", "repo": repo})
                )
            results.append(
                await mcp.call_tool(
                    "gitea_api_request", {"method": "GET", "path": "/repos/octo/text"}
                )
            )

    texts = [result.content[0].text for result in results]
    per_identity = [
        (results[:4], secrets),
        (results[4:], [*admin_secrets, *(issued if mode != "off" else [])]),
    ]
    for group, forbidden in per_identity:
        for result in group:
            assert_clean(result.content[0].text, forbidden)
            assert_clean(json.dumps(result.structured_content), forbidden)
    assert any("[REDACTED]" in text for text in texts)  # the redaction happened, not an empty body
    everything = [*secrets, *admin_secrets]
    assert_clean(get_audit_logger().log_path.read_text(encoding="utf-8"), everything)
    assert_clean(caplog.text, everything)
    captured = capsys.readouterr()
    assert_clean(captured.out + captured.err, everything)


async def test_exceptions_during_a_call_do_not_leak_secrets(
    httpx_mock: HTTPXMock, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    caplog.set_level(logging.DEBUG)
    oauth = build_oauth_app()
    secrets = static_secrets()

    async def explode(*args: Any, **kwargs: Any) -> None:
        raise RuntimeError("failure carrying " + " ".join(secrets))

    monkeypatch.setattr(oauth.extras.executor, "execute", explode)
    httpx_mock.add_response(url=f"{API}/user", json={"login": "admin"}, is_reusable=True)

    async with mcp_client(oauth.app, BEARER_TOKEN) as mcp:
        result = await mcp.call_tool("gitea_repo_get", {"owner": "o", "repo": "r"})

    assert result.is_error
    assert_clean(result.content[0].text, secrets)
    assert_clean(caplog.text, secrets)
    assert_clean(get_audit_logger().log_path.read_text(encoding="utf-8"), secrets)
    assert "RuntimeError" not in result.content[0].text  # no exception details by default


async def test_validation_errors_never_contain_argument_values(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    planted = "sekret-value-" + GITEA_TOKEN

    async with mcp_client(oauth.app, BEARER_TOKEN) as mcp:
        results = [
            await mcp.call_tool("gitea_repo_get", {"owner": planted + "?x#y", "repo": "r"}),
            await mcp.call_tool("gitea_repo_get", {"owner": 12345, "repo": planted}),
            await mcp.call_tool("gitea_repo_get", {"owner": "o", "repo": "r", "extra": planted}),
            await mcp.call_tool("gitea_api_request", {"method": planted, "path": "/x"}),
            await mcp.call_tool("gitea_api_request", {"method": "GET", "path": planted}),
            await mcp.call_tool(
                "gitea_api_request",
                {"method": "GET", "path": "/user", "sudo": "not valid " + planted},
            ),
        ]

    for result in results:
        assert result.is_error
        assert planted not in result.content[0].text
        assert GITEA_TOKEN not in result.content[0].text
    assert planted not in get_audit_logger().log_path.read_text(encoding="utf-8")


# Access logs


def uvicorn_access_record(path_with_query: str) -> logging.LogRecord:
    return logging.LogRecord(
        name="uvicorn.access",
        level=logging.INFO,
        pathname=__file__,
        lineno=1,
        msg='%s - "%s %s HTTP/%s" %d',
        args=("127.0.0.1:5000", "GET", path_with_query, "1.1", 302),
        exc_info=None,
    )


@pytest.mark.parametrize(
    "path",
    [
        "/oauth/callback?code=one-time-gitea-code&state=state-secret-value",
        "/oauth/consent?request=pending-request-secret",
        "/authorize?client_id=c&code_challenge=abc&state=client-state-secret",
    ],
)
def test_access_log_lines_drop_the_query_string(path: str) -> None:
    record = uvicorn_access_record(path)

    assert AccessLogQueryFilter().filter(record)

    line = record.getMessage()
    assert "[redacted]" in line
    assert path.split("?")[0] in line
    for secret in (
        "one-time-gitea-code",
        "state-secret-value",
        "pending-request-secret",
        "client-state-secret",
        "code_challenge",
    ):
        assert secret not in line


def test_the_runner_installs_the_access_log_filter(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    from aegis_gitea_mcp import transports

    access_logger = logging.getLogger("uvicorn.access")
    monkeypatch.setattr(access_logger, "filters", [])
    monkeypatch.setattr(transports.uvicorn, "run", lambda *a, **k: None)
    caplog.set_level(logging.INFO, logger="uvicorn.access")

    assert run_http(oauth_settings()) == 0
    access_logger.info(
        '%s - "%s %s HTTP/%s" %d',
        "127.0.0.1:5000",
        "GET",
        "/oauth/callback?code=one-time-gitea-code&state=state-secret-value",
        "1.1",
        302,
    )

    assert "one-time-gitea-code" not in caplog.text
    assert "[redacted]" in caplog.text


# repr


def test_repr_of_settings_and_secret_holding_objects_contains_no_secret() -> None:
    settings = oauth_settings()
    identity = Identity(login="alice", mode="oauth", token=USER_TOKEN, redact=(USER_REFRESH,))
    session = SessionRecord(
        session_id="session-id-value",
        client_id="c1",
        login="alice",
        is_admin=False,
        gitea_access_token=USER_TOKEN,
        gitea_refresh_token=USER_REFRESH,
        gitea_expires_at=None,
        scopes=["gitea"],
        resource="https://mcp.example.com/mcp",
        created_at=0.0,
        previous_refresh_hash="hash-of-previous-refresh",
    )
    upstream = UpstreamTokens(access_token=USER_TOKEN, refresh_token=USER_REFRESH, expires_at=None)
    pending = PendingAuthorization(
        client_id="c1",
        client_name="Client",
        redirect_uri="https://client.example/cb",
        redirect_uri_provided_explicitly=True,
        code_challenge="challenge",
        client_state="client-state-secret",
        scopes=["gitea"],
        resource="https://mcp.example.com/mcp",
        code_verifier="pkce-verifier-secret",
        csrf_hash="csrf-hash-secret",
        binding_hash="binding-hash-secret",
    )

    rendered = [
        repr(settings),
        str(settings),
        repr(identity),
        repr(session),
        repr(upstream),
        repr(pending),
        repr(settings.gitea_token),
        repr(settings.model_dump()),
    ]

    for text in rendered:
        assert_clean(text, [*static_secrets(), "client-state-secret", "pkce-verifier-secret"])
        assert "csrf-hash-secret" not in text and "binding-hash-secret" not in text
        assert "hash-of-previous-refresh" not in text
    assert "**********" in repr(settings.gitea_token)


def test_secrets_in_object_keys_are_masked_like_values() -> None:
    from aegis_gitea_mcp.security import sanitize_data

    token = "agm_" + "A1b2C3d4E5f6G7h8I9j0K1l2M3n4O5p6Q7r8S9t0U1v"
    nested = {"outer": [{token: "value", "plain": token}]}

    masked = sanitize_data(nested, mode="mask")
    blocked = sanitize_data(nested, mode="block")

    assert token not in json.dumps(masked) and token not in json.dumps(blocked)
    assert masked["outer"][0]["plain"] != token
    assert blocked != nested


# Configuration errors and a real stdio process


@pytest.mark.parametrize(
    ("variable", "value"),
    [
        ("GITEA_URL", "https://user:" + GITEA_TOKEN + "@gitea.example.com"),
        ("GITEA_URL", "https://gitea.example.com?token=" + GITEA_TOKEN),
        ("MCP_BEARER_TOKEN", "short-" + "x" * 5),
        ("PUBLIC_BASE_URL", "http://mcp.example.com/" + GITEA_TOKEN),
        ("OAUTH_SCOPES", GITEA_TOKEN),
        ("TOKEN_ENCRYPTION_KEY", "tooshort-" + ENCRYPTION_KEY[:5]),
        ("MCP_PORT", GITEA_TOKEN),
        ("MCP_PATH", GITEA_TOKEN),
        ("FORWARDED_ALLOW_IPS", GITEA_TOKEN),
        ("LOG_LEVEL", GITEA_TOKEN),
    ],
)
def test_configuration_errors_name_the_setting_but_never_echo_its_value(
    variable: str,
    value: str,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    from aegis_gitea_mcp import cli
    from aegis_gitea_mcp.cli import main

    monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
    monkeypatch.setenv("GITEA_TOKEN", GITEA_TOKEN)
    monkeypatch.setenv(variable, value)
    monkeypatch.setattr(cli, "configure_logging", lambda level: None)

    assert main(["serve", "--transport", "http"]) == 2

    err = capsys.readouterr().err
    assert variable in err
    assert value not in err and GITEA_TOKEN not in err
    assert "Traceback" not in err


async def test_a_real_stdio_process_keeps_every_secret_out_of_stdout_stderr_and_the_audit_file(
    tmp_path: Any,
) -> None:
    import sys

    from mcp import Client
    from mcp.client.stdio import StdioServerParameters, stdio_client

    secrets = [GITEA_TOKEN, BEARER_TOKEN, APP_CLIENT_SECRET, ENCRYPTION_KEY]
    env = {
        "GITEA_URL": "http://127.0.0.1:9",  # nothing listens: every call fails
        "GITEA_TOKEN": GITEA_TOKEN,
        "MCP_BEARER_TOKEN": BEARER_TOKEN,
        "GITEA_OAUTH_CLIENT_SECRET": APP_CLIENT_SECRET,
        "TOKEN_ENCRYPTION_KEY": ENCRYPTION_KEY,
        "DATA_DIR": str(tmp_path / "state"),
        "AUDIT_LOG_PATH": str(tmp_path / "state" / "audit.log"),
        "REQUEST_TIMEOUT_SECONDS": "2",
        "LOG_LEVEL": "DEBUG",
    }
    params = StdioServerParameters(
        command=sys.executable,
        args=["-m", "aegis_gitea_mcp.cli", "serve", "--transport", "stdio"],
        env=env,
        cwd=str(tmp_path),
    )
    stderr_path = tmp_path / "stderr.log"
    with stderr_path.open("w", encoding="utf-8") as errlog:
        async with Client(stdio_client(params, errlog=errlog)) as client:
            results = [
                await client.call_tool("gitea_repo_get", {"owner": "o", "repo": "r"}),
                await client.call_tool("gitea_repo_get", {"owner": 5}),
                await client.call_tool("gitea_api_request", {"method": "GET", "path": "/user"}),
            ]

    for result in results:
        assert result.is_error
        assert_clean(result.content[0].text, secrets)  # type: ignore[union-attr]
    assert_clean(stderr_path.read_text(encoding="utf-8"), secrets)
    assert_clean((tmp_path / "state" / "audit.log").read_text(encoding="utf-8"), secrets)