#!/usr/bin/env python3
"""Use the private Atlas Forgejo API through a least-authority PR boundary."""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Callable
from pathlib import Path

CANONICAL_BASE_URL = "https://scm.bstein.dev"
ALLOWED_OWNER = "atlas"
DEFAULT_TOKEN_FILE = Path("/runtime-access/gitea-token")
REPO_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}\Z")
SHA_RE = re.compile(r"[0-9a-fA-F]{40}\Z")
CREATE_PATH_RE = re.compile(r"/api/v1/repos/atlas/([^/]+)/pulls\Z")
MAX_ERROR_BYTES = 8192
MAX_RESPONSE_BYTES = 2 * 1024 * 1024
DRAFT_TITLE_PREFIX = "WIP: "
GIT_BIN = "/usr/bin/git"
SENSITIVE_BODY_PATTERNS = (
    re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
    re.compile(r"(?i)\b(?:password|passwd|secret|token|api[_-]?key)\s*[:=]"),
    re.compile(r"(?i)\bauthorization\s*:\s*(?:bearer|token|basic)\s+\S+"),
    re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9]{20,}|sk-[A-Za-z0-9]{20,})\b"),
    re.compile(r"\b(?:AKIA[0-9A-Z]{16}|AIza[0-9A-Za-z_-]{35})\b"),
    re.compile(r"\beyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}\b"),
)


class PolicyError(ValueError):
    """Raised when a requested Forgejo operation exceeds the safe boundary."""


class RejectRedirectHandler(urllib.request.HTTPRedirectHandler):
    """Reject every redirect before urllib can copy authentication headers."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise PolicyError("Forgejo redirects are not allowed")


_SAFE_OPENER = urllib.request.build_opener(RejectRedirectHandler())


def _safe_urlopen(request: urllib.request.Request, timeout: int):
    """Open one request using a handler that never follows redirects."""
    return _SAFE_OPENER.open(request, timeout=timeout)


def read_token(path: Path = DEFAULT_TOKEN_FILE) -> str:
    """Read the token from the pod-lifetime Vault projection."""
    token = path.read_text(encoding="utf-8").strip()
    if not token:
        raise ValueError(f"runtime credential is empty: {path}")
    return token


def configured_base_url() -> str:
    """Reject attempts to redirect credentials away from the canonical origin."""
    configured = os.environ.get("GITEA_BASE_URL", CANONICAL_BASE_URL).rstrip("/")
    if configured != CANONICAL_BASE_URL:
        raise PolicyError("Forgejo origin is fixed to the private Atlas SCM service")
    return configured


def _validate_repo(repo: str) -> str:
    if not REPO_RE.fullmatch(repo) or repo in {".", ".."}:
        raise PolicyError("repository name is outside the Atlas allowlist")
    return repo


def _validate_ref(value: object, name: str) -> str:
    """Validate the complete Git ref grammar using Git itself."""
    if not isinstance(value, str) or not value or value.startswith("-"):
        raise PolicyError(f"{name} must be a same-repository branch name")
    if any(ord(character) < 32 or ord(character) == 127 for character in value):
        raise PolicyError(f"{name} must be a safe same-repository branch name")
    result = subprocess.run(
        [GIT_BIN, "check-ref-format", f"refs/heads/{value}"],
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    )
    if result.returncode != 0:
        raise PolicyError(f"{name} must be a safe same-repository branch name")
    return value


def _validate_sha(value: object, name: str = "head SHA") -> str:
    if not isinstance(value, str) or not SHA_RE.fullmatch(value):
        raise PolicyError(f"{name} must be a full Git SHA-1")
    return value.lower()


def _validate_text(value: object, name: str, maximum: int, *, required: bool) -> str:
    if not isinstance(value, str):
        raise PolicyError(f"{name} must be text")
    if required and not value.strip():
        raise PolicyError(f"{name} must not be empty")
    if len(value) > maximum or "\x00" in value:
        raise PolicyError(f"{name} exceeds the safe request limit")
    return value


def _validate_body(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
    body = _validate_text(value, "body", 16384, required=False)
    if any(secret and secret in body for secret in forbidden):
        raise PolicyError("pull-request body contains runtime credential material")
    if any(pattern.search(body) for pattern in SENSITIVE_BODY_PATTERNS):
        raise PolicyError("pull-request body resembles credential material")
    return body


def _draft_title(value: object) -> str:
    """Return a bounded title using Gitea's configured default draft prefix."""
    title = _validate_text(value, "title", 251, required=True).strip()
    for prefix in ("WIP:", "[WIP]"):
        if title.upper().startswith(prefix):
            title = title[len(prefix) :].lstrip()
            break
    if not title:
        raise PolicyError("title must contain text after the draft prefix")
    return DRAFT_TITLE_PREFIX + title


def _split_api_path(path: str) -> urllib.parse.SplitResult:
    target = urllib.parse.urlsplit(path)
    if target.scheme or target.netloc or target.fragment:
        raise PolicyError("API path must be relative to the Atlas SCM origin")
    if not target.path.startswith("/api/v1/"):
        raise PolicyError("API path must start with /api/v1/")
    if "%" in target.path or "//" in target.path or "/../" in f"{target.path}/":
        raise PolicyError("encoded or non-canonical API paths are not allowed")
    return target


def _validate_query(target: urllib.parse.SplitResult, allowed: set[str]) -> None:
    try:
        pairs = urllib.parse.parse_qsl(
            target.query, keep_blank_values=True, strict_parsing=True
        )
    except ValueError as exc:
        raise PolicyError("invalid API query") from exc
    if len({key for key, _ in pairs}) != len(pairs):
        raise PolicyError("duplicate API query parameters are not allowed")
    if any(key not in allowed for key, _ in pairs):
        raise PolicyError("API query parameter is outside the read allowlist")
    values = dict(pairs)
    for name in ("page", "limit"):
        if name not in values:
            continue
        if not values[name].isdigit() or int(values[name]) < 1:
            raise PolicyError(f"{name} must be a positive integer")
    if "limit" in values and int(values["limit"]) > 50:
        raise PolicyError("read limit cannot exceed 50")
    if "state" in values and values["state"] not in {"open", "closed", "all"}:
        raise PolicyError("pull-request state is invalid")


def _authorize_read(target: urllib.parse.SplitResult) -> str:
    """Allow only repository, PR, branch, commit, and status metadata reads."""
    prefix = "/api/v1/repos/atlas/"
    remainder = target.path.removeprefix(prefix)
    if remainder == target.path:
        raise PolicyError("reads are limited to explicit Atlas repository metadata")
    repo, separator, suffix = remainder.partition("/")
    _validate_repo(repo)

    if not separator:
        _validate_query(target, set())
        return "repository"
    if suffix == "pulls":
        _validate_query(target, {"page", "limit", "state"})
        return "pull-list"
    if re.fullmatch(r"pulls/[1-9][0-9]*", suffix):
        _validate_query(target, set())
        return "pull"
    if re.fullmatch(r"pulls/[1-9][0-9]*/(?:commits|files)", suffix):
        _validate_query(target, {"page", "limit"})
        return "pull-evidence"
    if suffix == "branches":
        _validate_query(target, {"page", "limit"})
        return "branch-list"
    branch_match = re.fullmatch(r"branches/([^/]+)", suffix)
    if branch_match:
        _validate_ref(branch_match.group(1), "branch")
        _validate_query(target, set())
        return "branch"
    if suffix == "commits":
        _validate_query(target, {"page", "limit"})
        return "commit-list"
    if re.fullmatch(r"git/commits/[0-9a-fA-F]{40}", suffix):
        _validate_query(target, set())
        return "commit"
    if re.fullmatch(
        r"(?:commits/[0-9a-fA-F]{40}/status|commits/[0-9a-fA-F]{40}/statuses|statuses/[0-9a-fA-F]{40})",
        suffix,
    ):
        _validate_query(target, {"page", "limit"})
        return "status"
    raise PolicyError("repository API route is outside the metadata read allowlist")


def api_url(base_url: str, path: str) -> str:
    """Return a canonical same-origin URL without forwarding credentials."""
    if base_url.rstrip("/") != CANONICAL_BASE_URL:
        raise PolicyError("Forgejo origin is fixed to the private Atlas SCM service")
    target = _split_api_path(path)
    return urllib.parse.urlunsplit(
        ("https", "scm.bstein.dev", target.path, target.query, "")
    )


def authorize_request(method: str, path: str, data: object | None) -> str:
    """Validate one request and return its bounded operation name."""
    normalized_method = method.upper()
    target = _split_api_path(path)
    if normalized_method == "GET":
        if data is not None:
            raise PolicyError("read operations cannot include a request body")
        return _authorize_read(target)
    if target.query:
        raise PolicyError("mutating operations cannot include query parameters")
    if normalized_method != "POST":
        raise PolicyError("only read and create-draft operations are available")
    match = CREATE_PATH_RE.fullmatch(target.path)
    if not match:
        raise PolicyError("POST is limited to creating an Atlas draft pull request")
    _validate_repo(match.group(1))
    if not isinstance(data, dict) or set(data) != {"base", "body", "head", "title"}:
        raise PolicyError("draft creation accepts only base, body, head, and title")
    _validate_ref(data["base"], "base")
    _validate_ref(data["head"], "head")
    if data["title"] != _draft_title(data["title"]):
        raise PolicyError("new pull requests must use the Gitea draft-title prefix")
    _validate_body(data["body"])
    return "create-draft"


def build_request(
    method: str,
    path: str,
    *,
    base_url: str,
    token: str,
    data: object | None = None,
) -> urllib.request.Request:
    """Build an authorized request without putting the token in its URL or body."""
    authorize_request(method, path, data)
    payload = None
    if data is not None:
        payload = json.dumps(data, separators=(",", ":")).encode("utf-8")
    return urllib.request.Request(
        api_url(base_url, path),
        data=payload,
        method=method.upper(),
        headers={
            "Accept": "application/json",
            "Authorization": f"token {token}",
            "Content-Type": "application/json",
            "User-Agent": "hermes-atlas-pr-client/2",
        },
    )


def redact_bytes(value: bytes, token: str) -> bytes:
    """Remove exact runtime-token and authorization-header values from output."""
    redacted = value.replace(token.encode("utf-8"), b"[REDACTED]")
    return re.sub(
        rb"(?i)(authorization\s*[:=]\s*)(?:token|bearer)?\s*[^\s,;\"}]+",
        rb"\1[REDACTED]",
        redacted,
    )


def _request(
    method: str,
    path: str,
    data: object | None,
    *,
    token: str,
    opener: Callable[..., object] = _safe_urlopen,
) -> bytes:
    request = build_request(
        method, path, base_url=configured_base_url(), token=token, data=data
    )
    with opener(request, timeout=30) as response:  # type: ignore[attr-defined]
        body = response.read(MAX_RESPONSE_BYTES + 1)  # type: ignore[attr-defined]
    if len(body) > MAX_RESPONSE_BYTES:
        raise PolicyError("Forgejo response exceeds the safe size limit")
    return redact_bytes(body, token)


def read(
    path: str, *, token: str, opener: Callable[..., object] = _safe_urlopen
) -> bytes:
    """Read one explicitly allowed Atlas repository metadata resource."""
    return _request("GET", path, None, token=token, opener=opener)


def _nested(document: dict[str, object], *keys: str) -> object:
    value: object = document
    for key in keys:
        if not isinstance(value, dict) or key not in value:
            raise PolicyError("Forgejo omitted required pull-request evidence")
        value = value[key]
    return value


def _require_create_response(
    value: bytes,
    *,
    repo: str,
    base: str,
    head: str,
    head_sha: str,
    title: str,
    body: str,
) -> dict[str, object]:
    """Require exact server evidence for the intended open draft handoff."""
    try:
        document = json.loads(value)
    except json.JSONDecodeError as exc:
        raise PolicyError("Forgejo returned invalid pull-request metadata") from exc
    if not isinstance(document, dict):
        raise PolicyError("Forgejo returned invalid pull-request metadata")
    number = document.get("number")
    if not isinstance(number, int) or isinstance(number, bool) or number < 1:
        raise PolicyError("Forgejo omitted a valid pull-request number")
    full_name = f"{ALLOWED_OWNER}/{repo}"
    expected_html = f"{CANONICAL_BASE_URL}/{full_name}/pulls/{number}"
    checks = (
        document.get("state") == "open",
        document.get("draft") is True,
        document.get("merged") is False,
        document.get("html_url") == expected_html,
        document.get("url") == expected_html,
        document.get("title") == title,
        document.get("body") == body,
        _nested(document, "base", "ref") == base,
        _nested(document, "base", "repo", "full_name") == full_name,
        _nested(document, "head", "ref") == head,
        _nested(document, "head", "sha") == head_sha,
        _nested(document, "head", "repo", "full_name") == full_name,
    )
    if not all(checks):
        raise PolicyError("Forgejo did not confirm the exact draft pull-request state")
    return document


def create_draft(
    repo: str,
    *,
    base: str,
    head: str,
    head_sha: str,
    title: str,
    body: str,
    token: str,
    opener: Callable[..., object] = _safe_urlopen,
) -> bytes:
    """Create and verify a same-repository draft pull request for human review."""
    repo = _validate_repo(repo)
    base = _validate_ref(base, "base")
    head = _validate_ref(head, "head")
    head_sha = _validate_sha(head_sha)
    title = _draft_title(title)
    body = _validate_body(body, forbidden=(token,))
    data = {"base": base, "body": body, "head": head, "title": title}
    result = _request(
        "POST",
        f"/api/v1/repos/{ALLOWED_OWNER}/{repo}/pulls",
        data,
        token=token,
        opener=opener,
    )
    _require_create_response(
        result,
        repo=repo,
        base=base,
        head=head,
        head_sha=head_sha,
        title=title,
        body=body,
    )
    return result


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """Parse the intentionally narrow read and create-draft operations."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="validate and describe the request without reading a token or using the network",
    )
    subparsers = parser.add_subparsers(dest="operation", required=True)
    read_parser = subparsers.add_parser("read", help="read allowed Atlas metadata")
    read_parser.add_argument("path")
    create_parser = subparsers.add_parser(
        "create-draft", help="create a same-repository draft pull request"
    )
    create_parser.add_argument("repo")
    create_parser.add_argument("--base", required=True)
    create_parser.add_argument("--head", required=True)
    create_parser.add_argument("--head-sha", required=True)
    create_parser.add_argument("--title", required=True)
    create_parser.add_argument("--body", default="")
    return parser.parse_args(argv)


def _operation(args: argparse.Namespace) -> tuple[str, str, object | None]:
    if args.operation == "read":
        return "GET", args.path, None
    repo = _validate_repo(args.repo)
    _validate_sha(args.head_sha)
    data = {
        "base": _validate_ref(args.base, "base"),
        "body": _validate_body(args.body),
        "head": _validate_ref(args.head, "head"),
        "title": _draft_title(args.title),
    }
    return "POST", f"/api/v1/repos/{ALLOWED_OWNER}/{repo}/pulls", data


def _write_body(body: bytes) -> None:
    if body:
        sys.stdout.buffer.write(body)
        if not body.endswith(b"\n"):
            sys.stdout.buffer.write(b"\n")


def main(argv: list[str] | None = None) -> int:
    """Validate, optionally describe, and execute one safe Forgejo operation."""
    token = ""
    try:
        args = parse_args(argv)
        method, path, data = _operation(args)
        operation = authorize_request(method, path, data)
        if args.dry_run:
            print(
                json.dumps(
                    {
                        "dry_run": True,
                        "host": "scm.bstein.dev",
                        "operation": operation,
                        "owner": ALLOWED_OWNER,
                        "path": path,
                        "request_fields": sorted(data)
                        if isinstance(data, dict)
                        else [],
                    },
                    sort_keys=True,
                )
            )
            return 0
        token = read_token()
        if args.operation == "read":
            body = read(path, token=token)
        else:
            assert isinstance(data, dict)
            body = create_draft(
                args.repo,
                base=str(data["base"]),
                head=str(data["head"]),
                head_sha=args.head_sha,
                title=str(data["title"]),
                body=str(data["body"]),
                token=token,
            )
        _write_body(body)
        return 0
    except urllib.error.HTTPError as exc:
        body = redact_bytes(exc.read(MAX_ERROR_BYTES), token) if token else b""
        print(f"Forgejo request failed with HTTP {exc.code}", file=sys.stderr)
        if body:
            sys.stderr.buffer.write(body)
            if not body.endswith(b"\n"):
                sys.stderr.buffer.write(b"\n")
        return 1
    except (OSError, PolicyError, ValueError, json.JSONDecodeError):
        print(
            "Forgejo request rejected or unavailable; no credential was disclosed",
            file=sys.stderr,
        )
        return 1


if __name__ == "__main__":
    raise SystemExit(main())