atlas-iac/services/hermes/scm-common/scripts/gitea_api.py

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

from __future__ import annotations

import argparse
import base64
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

import deadline_http
from gitea_api_policy import (
    PolicyError,
    _draft_title,
    _reject_forbidden,
    _validate_body,
    _validate_pr_number,
    _validate_pr_number_segment,
    _validate_query,
    _validate_ref,
    _validate_ref_bounds,
    _validate_repo,
    _validate_sha,
)

CANONICAL_BASE_URL = "https://scm.bstein.dev"
ALLOWED_OWNER = "atlas"
DEFAULT_TOKEN_FILE = Path("/vault/secrets/gitea-token")
CREATE_PATH_RE = re.compile(r"/api/v1/repos/atlas/([^/]+)/pulls\Z")
MAX_ERROR_BYTES = 8192
MAX_RESPONSE_BYTES = 2 * 1024 * 1024
MAX_API_TARGET_LENGTH = 768
MAX_API_PATH_LENGTH = 512
RAW_API_TARGET_RE = re.compile(r"[A-Za-z0-9/_.?&=-]+\Z")


def _safe_urlopen(request: urllib.request.Request, timeout: int):
    """Open one exchange under a killable absolute wall-clock deadline.

    The helper process rejects redirects before urllib can copy
    authentication headers, and its whole connect/send/read lifetime
    shares one hard deadline.
    """
    return deadline_http.open_bounded(
        request, maximum=MAX_RESPONSE_BYTES, 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)
    if configured != CANONICAL_BASE_URL:
        raise PolicyError("Forgejo origin is fixed to the private Atlas SCM service")
    return configured


def _split_api_path(path: str) -> urllib.parse.SplitResult:
    if not isinstance(path, str) or len(path) > MAX_API_TARGET_LENGTH:
        raise PolicyError("API target exceeds the safe size limit")
    if (
        not path.isascii()
        or any(ord(character) < 32 or ord(character) == 127 for character in path)
        or "\\" in path
        or not RAW_API_TARGET_RE.fullmatch(path)
    ):
        raise PolicyError("API target contains non-canonical raw characters")
    target = urllib.parse.urlsplit(path)
    canonical = urllib.parse.urlunsplit(("", "", target.path, target.query, ""))
    if canonical != path:
        raise PolicyError("API target is not in exact canonical form")
    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 len(target.path) > MAX_API_PATH_LENGTH:
        raise PolicyError("API path exceeds the safe size limit")
    segments = target.path.split("/")
    if "//" in target.path or any(segment in {".", ".."} for segment in segments):
        raise PolicyError("encoded or non-canonical API paths are not allowed")
    return target


def _authorize_read(
    target: urllib.parse.SplitResult, *, forbidden: tuple[str, ...] = ()
) -> str:
    """Allow only identity, repository, PR, branch, commit, and status reads."""
    if target.path == "/api/v1/user":
        # The broker holds the only forge credential, so this answers with the
        # broker identity's own account metadata (name, is_admin) and nothing
        # else; acceptance tooling uses it to prove that identity holds no
        # administrative rights. No query, no other /user route.
        _validate_query(target, set())
        return "identity"
    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("/")
    _reject_forbidden(repo, "repository", forbidden)
    _validate_repo(repo)

    if not separator:
        _validate_query(target, set())
        return "repository"
    if suffix == "pulls":
        _validate_query(target, {"page", "limit", "state"})
        return "pull-list"
    content_match = re.fullmatch(r"pulls/([^/]+)\.(?:patch|diff)", suffix)
    if content_match:
        _validate_pr_number_segment(content_match.group(1))
        raise PolicyError("repository API route is outside the metadata read allowlist")
    pull_match = re.fullmatch(r"pulls/([^/]+)", suffix)
    if pull_match:
        _validate_pr_number_segment(pull_match.group(1))
        _validate_query(target, set())
        return "pull"
    evidence_match = re.fullmatch(r"pulls/([^/]+)/(?:commits|files)", suffix)
    if evidence_match:
        _validate_pr_number_segment(evidence_match.group(1))
        _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 != 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,
    *,
    forbidden: tuple[str, ...] = (),
) -> 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, forbidden=forbidden)
    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")
    _reject_forbidden(match.group(1), "repository", forbidden)
    _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")
    if data["title"] != _draft_title(data["title"], forbidden=forbidden):
        raise PolicyError("new pull requests must use the Gitea draft-title prefix")
    _validate_body(data["body"], forbidden=forbidden)
    _validate_ref_bounds(data["base"], "base", forbidden=forbidden)
    _validate_ref_bounds(data["head"], "head", forbidden=forbidden)
    _validate_ref(data["base"], "base", forbidden=forbidden)
    _validate_ref(data["head"], "head", forbidden=forbidden)
    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, forbidden=(token,))
    return _build_request(method, path, base_url=base_url, token=token, data=data)


def _build_request(
    method: str,
    path: str,
    *,
    base_url: str,
    token: str,
    data: object | None = None,
) -> urllib.request.Request:
    """Build one already-authorized upstream request."""
    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 _reject_response_credential(value: bytes, token: str) -> None:
    """Fail closed if an upstream body reflects any ordinary token encoding."""
    raw = token.encode("utf-8")
    forms = {
        raw,
        base64.b64encode(raw),
        base64.b64encode(b"hermes-automation:" + raw),
        urllib.parse.quote(token, safe="").encode("ascii"),
    }
    if any(form and form in value for form in forms):
        raise PolicyError("Forgejo response contains runtime credential material")


def _request(
    method: str,
    path: str,
    data: object | None,
    *,
    token: str,
    opener: Callable[..., object] = _safe_urlopen,
    authorized: bool = False,
    expected_status: int,
) -> bytes:
    builder = _build_request if authorized else build_request
    request = builder(method, path, base_url=configured_base_url(), token=token, data=data)
    with opener(request, timeout=30) as response:  # type: ignore[attr-defined]
        status = getattr(response, "status", None)
        if status is None and hasattr(response, "getcode"):
            status = response.getcode()  # type: ignore[attr-defined]
        if status != expected_status:
            raise PolicyError("Forgejo returned an unexpected HTTP status")
        if response.headers.get_content_type() != "application/json":  # type: ignore[attr-defined]
            raise PolicyError("Forgejo returned an unexpected response type")
        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")
    _reject_response_credential(body, token)
    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, expected_status=200
    )


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 = _validate_pr_number(document.get("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."""
    _reject_forbidden(repo, "repository", (token,))
    repo = _validate_repo(repo)
    title = _draft_title(title, forbidden=(token,))
    body = _validate_body(body, forbidden=(token,))
    head_sha = _validate_sha(head_sha)
    _validate_ref_bounds(base, "base", forbidden=(token,))
    _validate_ref_bounds(head, "head", forbidden=(token,))
    base = _validate_ref(base, "base", forbidden=(token,))
    head = _validate_ref(head, "head", 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,
        authorized=True,
        expected_status=201,
    )
    _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, *, validate_refs: bool = True
) -> tuple[str, str, object | None]:
    if args.operation == "read":
        return "GET", args.path, None
    repo = _validate_repo(args.repo)
    _validate_sha(args.head_sha)
    body = _validate_body(args.body)
    title = _draft_title(args.title)
    _validate_ref_bounds(args.base, "base")
    _validate_ref_bounds(args.head, "head")
    base = _validate_ref(args.base, "base") if validate_refs else args.base
    head = _validate_ref(args.head, "head") if validate_refs else args.head
    data = {
        "base": base,
        "body": body,
        "head": head,
        "title": 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."""
    try:
        args = parse_args(argv)
        if args.dry_run:
            method, path, data = _operation(args, validate_refs=False)
            operation = authorize_request(method, path, data)
            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
        method, path, data = _operation(args, validate_refs=False)
        from scm_broker_client import create_draft as broker_create_draft
        from scm_broker_client import read as broker_read

        if args.operation == "read":
            body = broker_read(path)
        else:
            assert isinstance(data, dict)
            body = broker_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"]),
            )
        _write_body(body)
        return 0
    except urllib.error.HTTPError as exc:
        exc.read(MAX_ERROR_BYTES)
        print(f"SCM broker request failed with HTTP {exc.code}", file=sys.stderr)
        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())