"""Trusted execution adapters that consume ScopeWrit tickets before acting."""

from __future__ import annotations

import ipaddress
import re
import socket
import ssl

# Intentional: this module is the explicit subprocess enforcement boundary.
import subprocess  # nosec B404
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Generic, TypeVar
from urllib.parse import urlsplit

import httpx

from .canonical import sha256_bytes, sha256_value
from .engine import ScopeWritEngine
from .errors import GatewayError
from .models import ActionDecision, ActionProposal, Decision
from .resources import normalize_resource

T = TypeVar("T")
_HTTP_HEADER_NAME = re.compile(r"^[!#$%&'*+.^_`|~0-9A-Za-z-]+$")


@dataclass(frozen=True)
class GuardedResult(Generic[T]):
    decision: ActionDecision
    value: T


class GuardedExecutor:
    """Wrap arbitrary MCP/cloud/browser callbacks behind a consumed ticket."""

    def __init__(self, engine: ScopeWritEngine, *, adapter: str):
        self.engine = engine
        self.adapter = adapter

    def execute(
        self,
        *,
        attestation: str,
        proposal: ActionProposal,
        operation: Callable[[], T],
        approval: str | None = None,
        outcome: Callable[[T], Mapping[str, Any]] | None = None,
    ) -> GuardedResult[T]:
        decision = self.engine.authorize(
            attestation=attestation,
            proposal=proposal,
            approval=approval,
        )
        if decision.decision != Decision.ALLOW or decision.ticket is None:
            raise GatewayError(
                f"action {decision.decision.value.lower()}: {','.join(decision.reasons)}",
                decision=decision,
            )
        self.engine.consume_ticket(
            decision.ticket,
            proposal=proposal,
            adapter=self.adapter,
        )
        try:
            value = operation()
        except Exception as exc:
            self.engine.record_execution(
                proposal=proposal,
                outcome={
                    "status": "error",
                    "error_type": type(exc).__name__,
                },
            )
            raise
        self.engine.record_execution(
            proposal=proposal,
            outcome=(dict(outcome(value)) if outcome is not None else {"status": "completed"}),
        )
        return GuardedResult(decision=decision, value=value)


class GuardedHTTPClient:
    """Explicit HTTP egress gateway.

    Resolve once, authorize that address set, and connect to a selected literal
    address while preserving the logical Host and TLS hostname. Each request
    has its own connection and explicit headers; cookies and TLS sessions are
    not reused across virtual hosts. This is an explicit gateway, not OS-level
    socket interception. An injected transport is a trusted adapter component.
    """

    def __init__(
        self,
        engine: ScopeWritEngine,
        *,
        timeout_seconds: float = 10.0,
        max_request_bytes: int = 256 * 1024,
        max_response_bytes: int = 1024 * 1024,
        transport: httpx.BaseTransport | None = None,
        tls_context: ssl.SSLContext | None = None,
    ):
        self.engine = engine
        if max_request_bytes < 0 or max_response_bytes < 1:
            raise ValueError("HTTP body limits must be positive")
        self.max_request_bytes = max_request_bytes
        self.max_response_bytes = max_response_bytes
        self._timeouts = httpx.Timeout(timeout_seconds).as_dict()
        if tls_context is not None and (
            not tls_context.check_hostname or tls_context.verify_mode != ssl.CERT_REQUIRED
        ):
            raise ValueError("TLS context must require certificates and hostname verification")
        # HTTPX keys its pool by the physical origin. With address pinning,
        # distinct hostnames can share that origin, so never reuse a connection.
        self._transport = transport or httpx.HTTPTransport(
            verify=tls_context if tls_context is not None else True,
            trust_env=False,
            limits=httpx.Limits(max_connections=32, max_keepalive_connections=0),
            retries=0,
        )

    @staticmethod
    def resolve(url: str) -> tuple[str, ...]:
        normalized = normalize_resource(url)
        if normalized.scheme not in {"http", "https"}:
            raise GatewayError("HTTP adapter only accepts http or https resources")
        parsed = urlsplit(normalized.value)
        port = parsed.port or (443 if parsed.scheme == "https" else 80)
        try:
            addresses = {
                item[4][0]
                for item in socket.getaddrinfo(
                    parsed.hostname,
                    port,
                    type=socket.SOCK_STREAM,
                )
            }
        except socket.gaierror as exc:
            raise GatewayError("resource host could not be resolved") from exc
        if not addresses:
            raise GatewayError("resource host resolved to no addresses")
        return tuple(sorted(addresses))

    def request(
        self,
        *,
        attestation: str,
        run_id: str,
        request_id: str,
        method: str,
        url: str,
        justification: str | None = None,
        headers: Mapping[str, str] | None = None,
        content: bytes | None = None,
        approval: str | None = None,
    ) -> GuardedResult[httpx.Response]:
        normalized_headers: dict[str, str] = {}
        forbidden_headers = {
            "host",
            ":authority",
            "connection",
            "content-length",
            "proxy-authorization",
            "transfer-encoding",
        }
        for name, value in (headers or {}).items():
            normalized_name = name.strip().lower()
            if (
                not normalized_name
                or not _HTTP_HEADER_NAME.fullmatch(normalized_name)
                or normalized_name in forbidden_headers
                or normalized_name in normalized_headers
            ):
                raise GatewayError(f"HTTP header is forbidden or ambiguous: {name}")
            if any(
                ord(character) == 127 or (ord(character) < 32 and character != "\t")
                for character in value
            ):
                raise GatewayError("HTTP header values contain control characters")
            normalized_headers[normalized_name] = value
        logical_url = httpx.URL(normalize_resource(url).value)
        try:
            addresses = [ipaddress.ip_address(value) for value in self.resolve(str(logical_url))]
        except ValueError as exc:
            raise GatewayError("resolver returned an invalid address") from exc
        if not addresses or any(getattr(address, "scope_id", None) for address in addresses):
            raise GatewayError("resolver must return unscoped IP addresses")
        resolved_ips = tuple(
            str(address)
            for address in sorted(
                set(addresses), key=lambda address: (address.version, address.packed)
            )
        )
        connection_ip = resolved_ips[0]
        body = content or b""
        if len(body) > self.max_request_bytes:
            raise GatewayError("HTTP request body exceeds the configured limit")
        proposal = ActionProposal(
            request_id=request_id,
            run_id=run_id,
            action="network.http",
            resource=str(logical_url),
            method=method,
            attributes={
                "resolved_ips": list(resolved_ips),
                "connection_ip": connection_ip,
                "header_names": sorted(normalized_headers),
                "headers_sha256": sha256_value(normalized_headers),
                "content_sha256": sha256_bytes(body),
                "content_bytes": len(body),
            },
            justification=justification,
        )
        executor = GuardedExecutor(self.engine, adapter="http")

        def operation() -> httpx.Response:
            # Use the documented HTTP Core SNI extension with the checked IP
            # in the connection URL. Never resolve the logical hostname again.
            request = httpx.Request(
                method,
                logical_url.copy_with(host=connection_ip),
                headers={
                    **normalized_headers,
                    "host": logical_url.netloc.decode("ascii"),
                    "connection": "close",
                },
                content=content,
                extensions={
                    "sni_hostname": logical_url.raw_host.decode("ascii"),
                    "timeout": self._timeouts,
                },
            )
            upstream = self._transport.handle_request(request)
            try:
                chunks: list[bytes] = []
                size = 0
                for chunk in upstream.iter_bytes(chunk_size=64 * 1024):
                    size += len(chunk)
                    if size > self.max_response_bytes:
                        raise GatewayError("HTTP response body exceeds the configured limit")
                    chunks.append(chunk)
                response_headers = httpx.Headers(upstream.headers)
                # iter_bytes has decoded the content. Do not ask Response to
                # decode it again, or preserve compressed/chunked lengths.
                response_headers.pop("transfer-encoding", None)
                if method.upper() != "HEAD" and upstream.status_code != 304:
                    response_headers.pop("content-encoding", None)
                    response_headers["content-length"] = str(size)
                # HEAD and 304 describe the selected representation; their
                # length/encoding refer to a body that was not transferred.
                return httpx.Response(
                    status_code=upstream.status_code,
                    headers=response_headers,
                    content=b"".join(chunks),
                    request=httpx.Request(
                        method, logical_url, headers=normalized_headers, content=content
                    ),
                    extensions={**upstream.extensions, "scopewrit_connection_ip": connection_ip},
                )
            finally:
                upstream.close()

        return executor.execute(
            attestation=attestation,
            proposal=proposal,
            approval=approval,
            operation=operation,
            outcome=lambda response: {
                "status": "completed",
                "http_status": response.status_code,
                "response_body_sha256": sha256_bytes(response.content),
                "response_bytes": len(response.content),
                "redirect_location_present": "location" in response.headers,
                "connection_ip": connection_ip,
                "tls_server_name": logical_url.raw_host.decode("ascii")
                if logical_url.scheme == "https"
                else None,
            },
        )

    def close(self) -> None:
        self._transport.close()

    def __enter__(self) -> GuardedHTTPClient:
        return self

    def __exit__(self, *_: object) -> None:
        self.close()


def shell_resource(argv: Sequence[str]) -> str:
    if not argv or not argv[0]:
        raise GatewayError("shell adapter requires a non-empty argv")
    executable_path = Path(argv[0])
    if not executable_path.is_absolute():
        raise GatewayError("shell adapter requires an absolute executable path")
    if not executable_path.is_file():
        raise GatewayError("shell executable does not exist")
    executable = executable_path.name.lower()
    if not executable:
        raise GatewayError("shell executable name is ambiguous")
    subcommand = "run"
    if len(argv) > 1 and argv[1] and not argv[1].startswith("-"):
        subcommand = argv[1].lower()
    return f"shell://{executable}/{subcommand}"


def guarded_subprocess(
    engine: ScopeWritEngine,
    *,
    attestation: str,
    run_id: str,
    request_id: str,
    argv: Sequence[str],
    cwd: str | Path | None = None,
    approval: str | None = None,
    justification: str | None = None,
    timeout_seconds: float = 30,
) -> GuardedResult[subprocess.CompletedProcess[str]]:
    canonical_argv = [str(Path(argv[0]).resolve()), *argv[1:]]
    resource = shell_resource(canonical_argv)
    proposal = ActionProposal(
        request_id=request_id,
        run_id=run_id,
        action="shell.exec",
        resource=resource,
        attributes={
            "argv_sha256": sha256_value(canonical_argv),
            "cwd_sha256": sha256_value(str(Path(cwd).resolve())) if cwd else None,
        },
        justification=justification,
    )
    executor = GuardedExecutor(engine, adapter="subprocess")
    # Exact canonical argv is principal-signed and shell expansion is off.
    return executor.execute(
        attestation=attestation,
        proposal=proposal,
        approval=approval,
        operation=lambda: subprocess.run(
            canonical_argv,
            cwd=cwd,
            shell=False,  # nosec B603
            check=False,
            capture_output=True,
            text=True,
            timeout=timeout_seconds,
        ),
        outcome=lambda result: {
            "status": "completed",
            "exit_code": result.returncode,
            "stdout_sha256": sha256_value(result.stdout),
            "stderr_sha256": sha256_value(result.stderr),
        },
    )
