"""Secret values that pass through tool calls are never echoed back to the model.

Covers stored webhook authorization headers (SEC-02), Action secrets and
submitted credential fields. Credential-issuing operations are the documented
exception: they return the credential they just minted.
"""

from __future__ import annotations

import json
from typing import Any

import pytest
from mcp import Client
from pytest_httpx import HTTPXMock

from aegis_gitea_mcp.gitea.executor import CREDENTIAL_ISSUING_OPERATIONS
from tests.test_server_helpers import API, audit_entries, connected, mock_user, server_settings

STORED_HEADER = "Bearer webhook-target-secret-0123456789"
BASIC_HEADER = "Basic dXNlcjpwYXNzd29yZC1vZi10aGUtd2ViaG9vaw=="
MODES = ["off", "mask", "block"]


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


def hook(header: str | None = STORED_HEADER) -> dict[str, Any]:
    return {
        "id": 3,
        "type": "gitea",
        "active": True,
        "config": {"url": "https://ci.example.org/hook", "content_type": "json"},
        "authorization_header": header,
    }


async def call(client: Client, name: str, arguments: dict[str, Any]) -> tuple[str, Any]:
    result = await client.call_tool(name, arguments)
    return result.content[0].text, result.structured_content  # type: ignore[union-attr]


HOOK_CASES = [
    ("gitea_hook_repo_list_hooks", {"owner": "o", "repo": "r"}, "GET", "/repos/o/r/hooks", True),
    (
        "gitea_hook_repo_get_hook",
        {"owner": "o", "repo": "r", "id": 3},
        "GET",
        "/repos/o/r/hooks/3",
        False,
    ),
    (
        "gitea_hook_repo_edit_hook",
        {"owner": "o", "repo": "r", "id": 3, "body": {"active": True}},
        "PATCH",
        "/repos/o/r/hooks/3",
        False,
    ),
    ("gitea_hook_org_list_hooks", {"org": "acme"}, "GET", "/orgs/acme/hooks", True),
    ("gitea_hook_org_get_hook", {"org": "acme", "id": 3}, "GET", "/orgs/acme/hooks/3", False),
    ("gitea_hook_user_list_hooks", {}, "GET", "/user/hooks", True),
    ("gitea_hook_user_get_hook", {"id": 3}, "GET", "/user/hooks/3", False),
    ("gitea_admin_list_hooks", {}, "GET", "/admin/hooks", True),
    ("gitea_admin_get_hook", {"id": 3}, "GET", "/admin/hooks/3", False),
    (
        "gitea_api_request",
        {"method": "GET", "path": "/repos/o/r/hooks"},
        "GET",
        "/repos/o/r/hooks",
        True,
    ),
]


@pytest.mark.parametrize("mode", MODES)
@pytest.mark.parametrize(("tool", "arguments", "method", "path", "is_list"), HOOK_CASES)
@pytest.mark.parametrize("header", [STORED_HEADER, BASIC_HEADER])
async def test_stored_webhook_authorization_header_is_never_returned(
    httpx_mock: HTTPXMock,
    mode: str,
    tool: str,
    arguments: dict[str, Any],
    method: str,
    path: str,
    is_list: bool,
    header: str,
) -> None:
    mock_user(httpx_mock)
    stored = hook(header)
    httpx_mock.add_response(
        method=method, url=f"{API}{path}", json=[stored, hook(None)] if is_list else stored
    )
    async with connected(server_settings(secret_detection_mode=mode)) as client:
        text, structured = await call(client, tool, arguments)

    assert header not in text
    assert header not in json.dumps(structured)
    assert "[REDACTED]" in text
    # The rest of the hook stays readable.
    assert "ci.example.org" in text
    assert header not in json.dumps(audit_entries())


async def test_empty_authorization_header_stays_visible_as_empty(httpx_mock: HTTPXMock) -> None:
    mock_user(httpx_mock)
    httpx_mock.add_response(url=f"{API}/repos/o/r/hooks/3", json=hook(""))
    async with connected() as client:
        _, structured = await call(
            client, "gitea_hook_repo_get_hook", {"owner": "o", "repo": "r", "id": 3}
        )

    assert structured["data"]["authorization_header"] == ""


async def test_nested_secret_fields_are_redacted_by_key_name(httpx_mock: HTTPXMock) -> None:
    mock_user(httpx_mock)
    httpx_mock.add_response(
        url=f"{API}/repos/o/r/hooks",
        json=[{"config": {"url": "https://x.example", "secret": "plain-shared-secret"}}],
    )
    async with connected(server_settings(secret_detection_mode="off")) as client:
        text, _ = await call(client, "gitea_hook_repo_list_hooks", {"owner": "o", "repo": "r"})

    assert "plain-shared-secret" not in text


async def test_credential_issuing_operations_still_return_their_credential(
    httpx_mock: HTTPXMock,
) -> None:
    assert "userCreateToken" in CREDENTIAL_ISSUING_OPERATIONS
    mock_user(httpx_mock)
    minted = "a" * 20 + "0123456789abcdef"
    httpx_mock.add_response(
        method="POST", url=f"{API}/users/alice/tokens", json={"id": 1, "name": "ci", "sha1": minted}
    )
    async with connected() as client:
        text, _ = await call(
            client, "gitea_user_create_token", {"username": "alice", "body": {"name": "ci"}}
        )

    assert minted in text
    # ... but the audit file never contains it.
    assert minted not in json.dumps(audit_entries())


# Action secrets: names and metadata only, never values

VALUE = "Sup3r-Secret-Action-Value!"
SHORT_VALUE = "pw"


def listing_with_values() -> list[dict[str, Any]]:
    """What Gitea sends (names and creation times), plus a hostile ``data`` field."""
    return [
        {"name": "DEPLOY_KEY", "created": "2026-01-01T00:00:00Z", "data": VALUE},
        {"name": "NPM_TOKEN", "created": "2026-01-02T00:00:00Z", "description": "ci"},
    ]


SECRET_LISTINGS = [
    ("gitea_actions_repo_list_secrets", {"owner": "o", "repo": "r"}, "/repos/o/r/actions/secrets"),
    ("gitea_actions_org_list_secrets", {"org": "acme"}, "/orgs/acme/actions/secrets"),
    (
        "gitea_api_request",
        {"method": "GET", "path": "/repos/o/r/actions/secrets"},
        "/repos/o/r/actions/secrets",
    ),
    (
        "gitea_api_request",
        {"method": "GET", "path": "/user/actions/secrets"},
        "/user/actions/secrets",
    ),
]


@pytest.mark.parametrize("mode", MODES)
@pytest.mark.parametrize(("tool", "arguments", "path"), SECRET_LISTINGS)
async def test_listing_secrets_returns_names_and_metadata_never_values(
    httpx_mock: HTTPXMock, mode: str, tool: str, arguments: dict[str, Any], path: str
) -> None:
    mock_user(httpx_mock)
    httpx_mock.add_response(url=f"{API}{path}", json=listing_with_values())
    async with connected(server_settings(secret_detection_mode=mode)) as client:
        text, structured = await call(client, tool, arguments)

    assert VALUE not in text and VALUE not in json.dumps(structured)
    names = [item["name"] for item in structured["data"]]
    assert names == ["DEPLOY_KEY", "NPM_TOKEN"]
    assert structured["data"][0]["created"] == "2026-01-01T00:00:00Z"
    assert structured["data"][1]["description"] == "ci"


async def test_a_secret_path_in_a_repository_is_not_mistaken_for_action_secrets(
    httpx_mock: HTTPXMock,
) -> None:
    mock_user(httpx_mock)
    httpx_mock.add_response(
        url=f"{API}/repos/o/r/contents/secrets", json=[{"name": "a", "data": "kept"}]
    )
    async with connected() as client:
        _, structured = await call(
            client, "gitea_api_request", {"method": "GET", "path": "/repos/o/r/contents/secrets"}
        )

    assert structured["data"][0]["data"] == "kept"


SECRET_WRITES = [
    (
        "gitea_actions_update_repo_secret",
        {"owner": "o", "repo": "r", "secretname": "DEPLOY_KEY"},
        "/repos/o/r/actions/secrets/DEPLOY_KEY",
    ),
    (
        "gitea_actions_update_org_secret",
        {"org": "acme", "secretname": "DEPLOY_KEY"},
        "/orgs/acme/actions/secrets/DEPLOY_KEY",
    ),
    (
        "gitea_actions_update_user_secret",
        {"secretname": "DEPLOY_KEY"},
        "/user/actions/secrets/DEPLOY_KEY",
    ),
]


@pytest.mark.parametrize("mode", MODES)
@pytest.mark.parametrize("value", [VALUE, SHORT_VALUE])
@pytest.mark.parametrize(("tool", "arguments", "path"), SECRET_WRITES)
async def test_updating_a_secret_never_echoes_the_submitted_value(
    httpx_mock: HTTPXMock,
    caplog: pytest.LogCaptureFixture,
    mode: str,
    value: str,
    tool: str,
    arguments: dict[str, Any],
    path: str,
) -> None:
    caplog.set_level("DEBUG")
    mock_user(httpx_mock)
    # Success echoing the value, then a validation error repeating it (as Gitea 422s do).
    httpx_mock.add_response(
        method="PUT",
        url=f"{API}{path}",
        status_code=201,
        json={"echo": value, "name": "DEPLOY_KEY"},
    )
    httpx_mock.add_response(
        method="PUT",
        url=f"{API}{path}",
        status_code=422,
        json={"message": f"invalid secret data: {value}", "errors": [f"data {value} too long"]},
    )
    async with connected(server_settings(secret_detection_mode=mode)) as client:
        ok = await call(client, tool, {**arguments, "body": {"data": value}})
        failed = await call(client, tool, {**arguments, "body": {"data": value}})

    for text, structured in (ok, failed):
        assert value not in text
        assert value not in json.dumps(structured)
    assert "[REDACTED]" in ok[0] and "[REDACTED]" in failed[0]
    assert value not in json.dumps(audit_entries()) and value not in caplog.text


async def test_the_raw_tool_also_hides_a_submitted_secret_value(httpx_mock: HTTPXMock) -> None:
    mock_user(httpx_mock)
    httpx_mock.add_response(
        method="PUT",
        url=f"{API}/repos/o/r/actions/secrets/K",
        status_code=422,
        json={"message": f"bad {VALUE}"},
    )
    async with connected() as client:
        text, _ = await call(
            client,
            "gitea_api_request",
            {
                "method": "PUT",
                "path": "/repos/o/r/actions/secrets/K",
                "body": {"data": VALUE},
            },
        )

    assert VALUE not in text and "[REDACTED]" in text


# Other submitted credential fields

CREDENTIAL_BODIES = [
    (
        "gitea_admin_create_user",
        {},
        "POST",
        "/admin/users",
        {"username": "bob", "email": "bob@example.com", "password": "Pa55word-Of-Bob"},
    ),
    (
        "gitea_hook_repo_create_hook",
        {"owner": "o", "repo": "r"},
        "POST",
        "/repos/o/r/hooks",
        {"type": "gitea", "authorization_header": "Bearer hook-target-token-123", "config": {}},
    ),
    (
        "gitea_repo_migrate",
        {},
        "POST",
        "/repos/migrate",
        {
            "clone_addr": "https://example.org/x.git",
            "repo_name": "x",
            "auth_token": "migration-source-token-123",
            "auth_password": "migration-source-pass-123",
            "aws_secret_access_key": "aws-secret-access-key-value-123",
        },
    ),
    (
        "gitea_api_request",
        {},
        "POST",
        "/orgs/acme/hooks",
        {"config": {"secret": "shared-hook-secret-123", "url": "https://x.example"}},
    ),
]


def secret_values(body: Any) -> list[str]:
    found: list[str] = []
    if isinstance(body, dict):
        for key, item in body.items():
            if isinstance(item, str) and key in {
                "password",
                "authorization_header",
                "auth_token",
                "auth_password",
                "aws_secret_access_key",
                "secret",
            }:
                found.append(item)
            else:
                found.extend(secret_values(item))
    return found


@pytest.mark.parametrize("mode", MODES)
@pytest.mark.parametrize(("tool", "arguments", "method", "path", "body"), CREDENTIAL_BODIES)
async def test_submitted_credentials_are_redacted_from_results(
    httpx_mock: HTTPXMock,
    mode: str,
    tool: str,
    arguments: dict[str, Any],
    method: str,
    path: str,
    body: dict[str, Any],
) -> None:
    mock_user(httpx_mock)
    secrets = secret_values(body)
    assert secrets
    httpx_mock.add_response(
        method=method,
        url=f"{API}{path}",
        status_code=422,
        json={"message": "rejected: " + " ".join(secrets), "errors": secrets},
    )
    request = (
        {"method": method, "path": path, "body": body}
        if tool == "gitea_api_request"
        else {**arguments, "body": body}
    )
    async with connected(server_settings(secret_detection_mode=mode)) as client:
        text, structured = await call(client, tool, request)

    for secret in secrets:
        assert secret not in text and secret not in json.dumps(structured)
        assert secret not in json.dumps(audit_entries())
    assert "[REDACTED]" in text


# Credential-issuing operations: the credential goes to the caller, nowhere else


async def test_issued_credentials_reach_the_caller_but_not_the_audit_file_or_the_logs(
    httpx_mock: HTTPXMock, caplog: pytest.LogCaptureFixture
) -> None:
    caplog.set_level("DEBUG")
    mock_user(httpx_mock)
    token = "9f8e7d6c5b4a39281706f5e4d3c2b1a099887766"
    registration = "runner-registration-token-0123456789ABCDEF"
    httpx_mock.add_response(
        method="POST", url=f"{API}/users/alice/tokens", json={"id": 1, "name": "ci", "sha1": token}
    )
    httpx_mock.add_response(
        method="POST",
        url=f"{API}/admin/actions/runners/registration-token",
        headers={"token": registration},
        status_code=200,
    )
    async with connected(server_settings(secret_detection_mode="block")) as client:
        created, _ = await call(
            client, "gitea_user_create_token", {"username": "alice", "body": {"name": "ci"}}
        )
        runner, _ = await call(client, "gitea_admin_create_runner_registration_token", {})

    assert token in created and registration in runner  # the documented exemption (W-07)
    audit_text = json.dumps(audit_entries())
    for secret in (token, registration):
        assert secret not in audit_text
        assert secret not in caplog.text