#!/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 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")
PR_PATH_RE = re.compile(r"/api/v1/repos/atlas/([^/]+)/pulls/([1-9][0-9]*)\Z")
CREATE_PATH_RE = re.compile(r"/api/v1/repos/atlas/([^/]+)/pulls\Z")
READ_PATH_RE = re.compile(r"/api/v1/repos/atlas/([^/]+)(?:/.*)?\Z")
SAFE_REF_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,199}\Z")
MAX_ERROR_BYTES = 8192
DRAFT_TITLE_PREFIX = "WIP: "


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


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:
    if not isinstance(value, str) or not SAFE_REF_RE.fullmatch(value):
        raise PolicyError(f"{name} must be a same-repository branch name")
    if (
        value.startswith(".")
        or value.endswith(("/", ".", ".lock"))
        or "//" in value
        or ".." in value
        or "@{" in value
    ):
        raise PolicyError(f"{name} must be a safe same-repository branch name")
    return value


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 _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 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")
        match = READ_PATH_RE.fullmatch(target.path)
        if not match:
            raise PolicyError("reads are limited to repositories owned by atlas")
        _validate_repo(match.group(1))
        return "read"

    if target.query:
        raise PolicyError("mutating operations cannot include query parameters")
    if normalized_method == "POST":
        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_text(data["body"], "body", 65536, required=False)
        return "create-draft"

    if normalized_method == "PATCH":
        match = PR_PATH_RE.fullmatch(target.path)
        if not match:
            raise PolicyError("PATCH is limited to Atlas draft pull-request metadata")
        _validate_repo(match.group(1))
        if not isinstance(data, dict) or not data or not set(data) <= {"body", "title"}:
            raise PolicyError("draft updates accept only title and body")
        if "title" in data and data["title"] != _draft_title(data["title"]):
            raise PolicyError("updated pull-request titles must preserve draft state")
        if "body" in data:
            _validate_text(data["body"], "body", 65536, required=False)
        return "update-draft"

    raise PolicyError(
        "method is not available: merge, approve, close, delete, and force-push are outside this client"
    )


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/1",
        },
    )


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] = urllib.request.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]
        return redact_bytes(response.read(), token)  # type: ignore[attr-defined]


def read(
    path: str, *, token: str, opener: Callable[..., object] = urllib.request.urlopen
) -> bytes:
    """Read one Atlas repository API resource."""
    return _request("GET", path, None, token=token, opener=opener)


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


def _require_draft_response(value: bytes) -> dict[str, object]:
    """Fail closed unless Forgejo confirms the resulting PR remains a draft."""
    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) or document.get("draft") is not True:
        raise PolicyError(
            "Forgejo did not confirm draft state; human review is required"
        )
    return document


def update_draft(
    repo: str,
    number: int,
    updates: dict[str, str],
    *,
    token: str,
    opener: Callable[..., object] = urllib.request.urlopen,
) -> bytes:
    """Update title/body only after Forgejo confirms the PR is still a draft."""
    if not isinstance(number, int) or isinstance(number, bool) or number < 1:
        raise PolicyError("pull-request number must be positive")
    path = f"/api/v1/repos/{ALLOWED_OWNER}/{_validate_repo(repo)}/pulls/{number}"
    current_raw = _request("GET", path, None, token=token, opener=opener)
    try:
        current = json.loads(current_raw)
    except json.JSONDecodeError as exc:
        raise PolicyError("Forgejo returned invalid pull-request metadata") from exc
    if not isinstance(current, dict) or current.get("draft") is not True:
        raise PolicyError("pull request is not a draft; human review now owns it")
    safe_updates = dict(updates)
    if "title" in safe_updates:
        safe_updates["title"] = _draft_title(safe_updates["title"])
    result = _request("PATCH", path, safe_updates, token=token, opener=opener)
    _require_draft_response(result)
    return result


def _body(args: argparse.Namespace) -> str:
    if args.body_file is None:
        return ""
    return args.body_file.read_text(encoding="utf-8")


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """Parse the three intentionally narrow client 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 an Atlas repository API path"
    )
    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("--title", required=True)
    create_parser.add_argument("--body-file", type=Path)

    update_parser = subparsers.add_parser(
        "update-draft", help="update title/body of an existing draft pull request"
    )
    update_parser.add_argument("repo")
    update_parser.add_argument("number", type=int)
    update_parser.add_argument("--title")
    update_parser.add_argument("--body-file", type=Path)
    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)
    if args.operation == "create-draft":
        data = {
            "base": args.base,
            "body": _body(args),
            "head": args.head,
            "title": _draft_title(args.title),
        }
        return "POST", f"/api/v1/repos/{ALLOWED_OWNER}/{repo}/pulls", data
    if args.number < 1:
        raise PolicyError("pull-request number must be positive")
    data = {}
    if args.title is not None:
        data["title"] = _draft_title(args.title)
    if args.body_file is not None:
        data["body"] = _body(args)
    return "PATCH", f"/api/v1/repos/{ALLOWED_OWNER}/{repo}/pulls/{args.number}", 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 [],
                        "requires_draft_preflight": operation == "update-draft",
                    },
                    sort_keys=True,
                )
            )
            return 0
        token = read_token()
        if args.operation == "read":
            body = read(path, token=token)
        elif args.operation == "create-draft":
            assert isinstance(data, dict)
            body = create_draft(
                args.repo,
                base=str(data["base"]),
                head=str(data["head"]),
                title=str(data["title"]),
                body=str(data["body"]),
                token=token,
            )
        else:
            assert isinstance(data, dict)
            body = update_draft(args.repo, args.number, data, 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):
        # Deliberately omit exception details. File contents, HTTP request
        # objects, and provider errors can carry credential material.
        print(
            "Forgejo request rejected or unavailable; no credential was disclosed",
            file=sys.stderr,
        )
        return 1


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