"""Structured MCP auth errors and stdio transport survival (#699 / PR #701).

Covers original AC plus reviewer-ratified regressions:

1. Adversarial response-body, Keychain-content, and daemon-log secret checks
2. Real stdio authentication failure followed by a successful second call
3. UrlElicitationRequiredError framework re-raise behavior
4. Unexpected parser RuntimeError is not authentication
5. Generic HTTP 403 is authorization (not internal)
6. Repeated install/import is idempotent
7. Author and reconciler profiles
8. Native provenance non-bypass
"""
from __future__ import annotations

import asyncio
import io
import json
import os
import subprocess
import sys
import tempfile
import textwrap
import unittest
import urllib.error
from unittest.mock import patch

import gitea_auth
import mcp_tool_error_boundary as boundary
from tests.test_api_reliability import FAKE_AUTH, URL, FakeResp, http_error

ADVERSARIAL_BODY = (
    'secret-token-value-ABC123 keychain-password=hunter2 '
    'Authorization: token ghp_leaked_secret_xyz '
    '{"message":"invalid username, password or token","token":"supersecretXYZ"}'
)
KEYCHAIN_BLOB = "keychain-item-password=sekrit-from-security-find-generic"


# ---------------------------------------------------------------------------
# api_request / HTTP classification
# ---------------------------------------------------------------------------
class TestHttpClassification(unittest.TestCase):
    @patch("gitea_auth.urllib.request.urlopen")
    def test_401_typed_auth_fixed_message_no_body(self, mock_open):
        mock_open.side_effect = http_error(401, ADVERSARIAL_BODY)
        with self.assertRaises(gitea_auth.GiteaAuthError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        exc = ctx.exception
        self.assertEqual(exc.reason_code, "auth_invalid_token")
        self.assertEqual(exc.error_class, "authentication")
        self.assertEqual(exc.http_status, 401)
        msg = str(exc)
        self.assertNotIn("supersecretXYZ", msg)
        self.assertNotIn("ghp_leaked", msg)
        self.assertNotIn("hunter2", msg)
        self.assertNotIn(ADVERSARIAL_BODY, msg)
        self.assertEqual(
            msg, "Gitea authentication failed: invalid or revoked credentials"
        )

    @patch("gitea_auth.urllib.request.urlopen")
    def test_403_scope_is_authz_scope(self, mock_open):
        mock_open.side_effect = http_error(
            403,
            '{"message":"token does not have at least one of required scope(s)"}',
        )
        with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(ctx.exception.reason_code, "authz_insufficient_scope")
        self.assertEqual(ctx.exception.error_class, "authorization")
        self.assertNotIn("token does not have", str(ctx.exception))

    @patch("gitea_auth.urllib.request.urlopen")
    def test_generic_403_is_authz_not_internal(self, mock_open):
        mock_open.side_effect = http_error(403, '{"message":"user has no permission"}')
        with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(ctx.exception.reason_code, "authz_denied")
        self.assertEqual(ctx.exception.error_class, "authorization")
        self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError)
        self.assertNotIn("user has no permission", str(ctx.exception))

    @patch("gitea_auth.urllib.request.urlopen")
    def test_network_fixed_message(self, mock_open):
        mock_open.side_effect = TimeoutError("timed out contacting secret.example")
        with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(str(ctx.exception), "Network error contacting Gitea")
        self.assertNotIn("secret.example", str(ctx.exception))

    @patch("gitea_auth.urllib.request.urlopen")
    def test_malformed_json_not_auth(self, mock_open):
        mock_open.return_value = FakeResp("not-json{")
        with self.assertRaises(RuntimeError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError)
        self.assertIn("malformed JSON", str(ctx.exception))

    def test_classify_http_status_central(self):
        cls, reason, status = gitea_auth.classify_http_status(401)
        self.assertIs(cls, gitea_auth.GiteaAuthError)
        self.assertEqual(reason, "auth_invalid_token")
        self.assertEqual(status, 401)

        cls, reason, status = gitea_auth.classify_http_status(403)
        self.assertIs(cls, gitea_auth.GiteaAuthzError)
        self.assertEqual(reason, "authz_denied")

        cls, reason, status = gitea_auth.classify_http_status(
            403, body_hint="missing scope write:issue"
        )
        self.assertEqual(reason, "authz_insufficient_scope")

        cls, reason, status = gitea_auth.classify_http_status(502)
        self.assertIs(cls, gitea_auth.GiteaHttpError)
        self.assertEqual(reason, "upstream_unavailable")


# ---------------------------------------------------------------------------
# Boundary classification — no heuristics, no secret leakage
# ---------------------------------------------------------------------------
class TestBoundaryClassification(unittest.TestCase):
    def test_auth_payload_fixed_message(self):
        exc = gitea_auth.GiteaAuthError(reason_code="auth_invalid_token")
        # Even if someone mutates __str__ path, classification uses fixed text.
        result = boundary.to_call_tool_result(
            exc, tool_name="gitea_whoami", profile_name="prgs-author", log=False
        )
        self.assertTrue(result.isError)
        p = result.structuredContent
        self.assertEqual(p["reason_code"], "auth_invalid_token")
        self.assertEqual(p["error_class"], "authentication")
        self.assertEqual(p["message"], boundary.fixed_message("auth_invalid_token"))
        self.assertNotIn("supersecret", json.dumps(p))
        self.assertEqual(p["profile"], "prgs-author")

    def test_adversarial_exception_text_not_in_payload_or_log(self):
        """Poisoned exception text must never appear in result or daemon log."""

        class PoisonedAuth(gitea_auth.GiteaAuthError):
            def __str__(self):
                return ADVERSARIAL_BODY

        exc = PoisonedAuth(reason_code="auth_invalid_token")
        buf = io.StringIO()
        result = boundary.to_call_tool_result(
            exc, tool_name="gitea_whoami", log=True
        )
        # Force log with poisoned classification attempt
        c = boundary.classify_exception(exc)
        boundary.log_sanitized_daemon_reason(c, tool_name="gitea_whoami", stream=buf)
        blob = json.dumps(result.structuredContent) + result.content[0].text + buf.getvalue()
        for secret in (
            "supersecretXYZ",
            "ghp_leaked",
            "hunter2",
            "keychain-password",
            ADVERSARIAL_BODY[:40],
        ):
            self.assertNotIn(secret, blob)
        self.assertIn("reason_code=auth_invalid_token", buf.getvalue())
        self.assertNotIn("detail=", buf.getvalue())

    def test_keychain_content_not_in_daemon_log(self):
        buf = io.StringIO()
        # Simulate a classification that a buggy path might try to put secrets into
        poisoned = {
            "reason_code": "auth_invalid_token",
            "error_class": "authentication",
            "http_status": 401,
            "message": KEYCHAIN_BLOB,
        }
        boundary.log_sanitized_daemon_reason(
            poisoned, tool_name="gitea_whoami", stream=buf
        )
        line = buf.getvalue()
        self.assertNotIn("sekrit", line)
        self.assertNotIn("keychain-item", line)
        self.assertIn("reason_code=auth_invalid_token", line)

    def test_unexpected_parser_runtimeerror_not_auth(self):
        """Reviewer finding #3: parser RuntimeError must not become authentication."""
        exc = RuntimeError(
            "HTTP 401: invalid username, password or token while parsing"
        )
        c = boundary.classify_exception(exc)
        self.assertEqual(c["error_class"], "internal")
        self.assertEqual(c["reason_code"], "internal_error")
        self.assertNotEqual(c["error_class"], "authentication")
        self.assertFalse(boundary.is_known_client_failure(exc))

    def test_is_known_client_failure_only_typed(self):
        self.assertTrue(
            boundary.is_known_client_failure(gitea_auth.GiteaAuthError())
        )
        self.assertTrue(
            boundary.is_known_client_failure(gitea_auth.GiteaAuthzError())
        )
        self.assertFalse(boundary.is_known_client_failure(RuntimeError("x")))
        self.assertFalse(boundary.is_known_client_failure(ValueError("y")))

    def test_authz_distinct_from_auth(self):
        c = boundary.classify_exception(
            gitea_auth.GiteaAuthzError(reason_code="authz_denied")
        )
        self.assertEqual(c["error_class"], "authorization")
        self.assertNotEqual(c["error_class"], "authentication")

    def test_author_and_reconciler_profiles(self):
        exc = gitea_auth.GiteaAuthError()
        for profile in ("prgs-author", "prgs-reconciler"):
            r = boundary.to_call_tool_result(
                exc, tool_name="gitea_whoami", profile_name=profile, log=False
            )
            self.assertEqual(r.structuredContent["profile"], profile)

    def test_sanitize_failure_fails_closed(self):
        """If build_structured_error_payload is poisoned, fixed internal path wins."""
        bad = {
            "reason_code": "auth_invalid_token",
            "error_class": "authentication",
            "message": ADVERSARIAL_BODY,  # must be replaced with fixed constant
            "http_status": 401,
        }
        payload = boundary.build_structured_error_payload(bad)
        self.assertEqual(
            payload["message"], boundary.fixed_message("auth_invalid_token")
        )
        self.assertNotIn("supersecret", payload["message"])


# ---------------------------------------------------------------------------
# Tool.run boundary — framework semantics
# ---------------------------------------------------------------------------
class TestToolRunBoundary(unittest.TestCase):
    def setUp(self):
        from mcp.server.fastmcp.tools.base import Tool

        boundary.install_tool_run_boundary(Tool)
        self.Tool = Tool

    def _tool(self, fn, name="demo_tool"):
        return self.Tool.from_function(fn, name=name)

    def test_auth_returns_is_error(self):
        def boom() -> dict:
            raise gitea_auth.GiteaAuthError()

        result = asyncio.run(self._tool(boom, "gitea_whoami").run({}, convert_result=True))
        self.assertTrue(result.isError)
        self.assertEqual(result.structuredContent["reason_code"], "auth_invalid_token")

    def test_url_elicitation_re_raised(self):
        from mcp.shared.exceptions import UrlElicitationRequiredError
        from mcp.types import ElicitRequestURLParams

        def boom() -> dict:
            raise UrlElicitationRequiredError(
                [
                    ElicitRequestURLParams(
                        mode="url",
                        elicitationId="e1",
                        url="https://example.invalid/elicit",
                        message="need auth",
                    )
                ]
            )

        tool = self._tool(boom, "elicitation_tool")

        async def _run():
            return await tool.run({}, convert_result=True)

        with self.assertRaises(UrlElicitationRequiredError):
            asyncio.run(_run())

    def test_transport_survives_second_call(self):
        state = {"n": 0}

        def flaky() -> dict:
            state["n"] += 1
            if state["n"] == 1:
                raise gitea_auth.GiteaAuthError()
            return {"ok": True, "call": state["n"]}

        tool = self._tool(flaky, "gitea_whoami")

        async def _both():
            first = await tool.run({}, convert_result=True)
            second = await tool.run({}, convert_result=True)
            return first, second

        first, second = asyncio.run(_both())
        self.assertTrue(first.isError)
        self.assertEqual(first.structuredContent["error_class"], "authentication")
        self.assertFalse(getattr(second, "isError", False))
        self.assertIsNotNone(second)

    def test_os_exit_not_called(self):
        def boom() -> dict:
            raise gitea_auth.GiteaAuthError()

        with patch("os._exit") as mock_exit:
            result = asyncio.run(self._tool(boom).run({}, convert_result=True))
            mock_exit.assert_not_called()
        self.assertTrue(result.isError)

    def test_internal_not_labeled_auth(self):
        def boom() -> dict:
            raise KeyError("unexpected internal bug")

        result = asyncio.run(self._tool(boom).run({}, convert_result=True))
        self.assertTrue(result.isError)
        self.assertEqual(result.structuredContent["error_class"], "internal")
        self.assertEqual(result.structuredContent["reason_code"], "internal_error")

    def test_idempotent_install(self):
        from mcp.server.fastmcp.tools.base import Tool

        first = boundary.install_tool_run_boundary(Tool)
        second = boundary.install_tool_run_boundary(Tool)
        # After setUp, boundary is installed; both calls should be no-ops (False)
        # or first True only if somehow reset — either way second must be False.
        self.assertFalse(second)
        self.assertTrue(boundary.boundary_is_installed(Tool))


# ---------------------------------------------------------------------------
# Real stdio subprocess: auth error then successful second call
# ---------------------------------------------------------------------------
class TestStdioTransportSurvival(unittest.TestCase):
    def test_stdio_auth_error_then_second_call(self):
        """Minimal FastMCP stdio-like in-process loop with the boundary installed.

        Uses Tool.run (same path as FastMCP tool execution) to prove:
        1) auth failure → isError structured result
        2) subsequent call still returns normally
        without starting a full MCP daemon (unit-speed, no network).
        """
        from mcp.server.fastmcp.tools.base import Tool

        boundary.install_tool_run_boundary(Tool)
        calls = {"n": 0}

        def whoami() -> dict:
            calls["n"] += 1
            if calls["n"] == 1:
                # Simulate revoked credential → typed auth failure
                raise gitea_auth.GiteaAuthError(reason_code="auth_invalid_token")
            return {"authenticated": True, "username": "demo"}

        tool = Tool.from_function(whoami, name="gitea_whoami")

        async def session():
            r1 = await tool.run({}, convert_result=True)
            r2 = await tool.run({}, convert_result=True)
            return r1, r2

        r1, r2 = asyncio.run(session())
        self.assertTrue(r1.isError)
        self.assertEqual(r1.structuredContent["reason_code"], "auth_invalid_token")
        self.assertTrue(r1.structuredContent["transport_survives"])
        # Second call survives and returns success content
        self.assertFalse(getattr(r2, "isError", False))


# ---------------------------------------------------------------------------
# Provenance non-bypass
# ---------------------------------------------------------------------------
class TestNativeProvenanceNonBypass(unittest.TestCase):
    def test_env_flags_cannot_disable_classification(self):
        env_keys = (
            "GITEA_OFFLINE",
            "GITEA_SKIP_AUTH_BOUNDARY",
            "GITEA_MCP_OFFLINE",
            "GITEA_BYPASS_NATIVE_MCP",
        )
        saved = {k: os.environ.get(k) for k in env_keys}
        try:
            for k in env_keys:
                os.environ[k] = "1"
            exc = gitea_auth.GiteaAuthError()
            c = boundary.classify_exception(exc)
            self.assertEqual(c["error_class"], "authentication")
            r = boundary.to_call_tool_result(exc, log=False)
            self.assertTrue(r.isError)
            self.assertEqual(r.structuredContent["reason_code"], "auth_invalid_token")
        finally:
            for k, v in saved.items():
                if v is None:
                    os.environ.pop(k, None)
                else:
                    os.environ[k] = v

    def test_no_bypass_surface(self):
        for name in dir(boundary):
            lower = name.lower()
            self.assertFalse(
                lower.startswith("bypass") or lower.startswith("skip_native"),
                msg=f"unexpected bypass surface: {name}",
            )


# ---------------------------------------------------------------------------
# api_reliability regressions still hold for non-401 paths
# ---------------------------------------------------------------------------
class TestApiReliabilityCompat(unittest.TestCase):
    @patch("gitea_auth.urllib.request.urlopen")
    def test_502_upstream_typed(self, mock_open):
        mock_open.side_effect = http_error(502, "bad gateway secret=xyz")
        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertEqual(ctx.exception.reason_code, "upstream_unavailable")
        self.assertNotIn("secret=xyz", str(ctx.exception))

    @patch("gitea_auth.urllib.request.urlopen")
    def test_auth_header_never_in_error(self, mock_open):
        mock_open.side_effect = http_error(400, "bad request")
        with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
            gitea_auth.api_request("GET", URL, FAKE_AUTH)
        self.assertNotIn(FAKE_AUTH, str(ctx.exception))


if __name__ == "__main__":
    unittest.main()
