421 lines
15 KiB
Python
Executable File
421 lines
15 KiB
Python
Executable File
#!/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())
|