126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Call the private Atlas Gitea API through a runtime-only token boundary."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
DEFAULT_BASE_URL = "https://scm.bstein.dev"
|
|
DEFAULT_TOKEN_FILE = Path("/runtime-access/gitea-token")
|
|
ALLOWED_METHODS = ("GET", "POST", "PUT", "PATCH", "DELETE")
|
|
|
|
|
|
def read_token(path: Path = DEFAULT_TOKEN_FILE) -> str:
|
|
"""Read and validate the token from its in-memory Vault projection."""
|
|
token = path.read_text(encoding="utf-8").strip()
|
|
if not token:
|
|
raise ValueError(f"runtime credential is empty: {path}")
|
|
return token
|
|
|
|
|
|
def api_url(base_url: str, path: str) -> str:
|
|
"""Return a same-origin Gitea API URL for a validated API path."""
|
|
base = urllib.parse.urlsplit(base_url.rstrip("/"))
|
|
target = urllib.parse.urlsplit(path)
|
|
if base.scheme not in {"http", "https"} or not base.netloc:
|
|
raise ValueError("GITEA_BASE_URL must be an absolute HTTP(S) URL")
|
|
if target.scheme or target.netloc or target.fragment:
|
|
raise ValueError("API path must be relative to the configured Gitea origin")
|
|
if not target.path.startswith("/api/v1/"):
|
|
raise ValueError("API path must start with /api/v1/")
|
|
return urllib.parse.urlunsplit(
|
|
(base.scheme, base.netloc, target.path, target.query, "")
|
|
)
|
|
|
|
|
|
def build_request(
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
base_url: str,
|
|
token: str,
|
|
data: object | None = None,
|
|
) -> urllib.request.Request:
|
|
"""Build one authenticated request without placing the token in its URL."""
|
|
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,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Authorization": f"token {token}",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "hermes-atlas-operator/1",
|
|
},
|
|
)
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
"""Parse a bounded method, API path, and optional JSON request body."""
|
|
parser = argparse.ArgumentParser(
|
|
description="Call the Atlas Gitea API using the runtime Vault token."
|
|
)
|
|
parser.add_argument("method", choices=ALLOWED_METHODS)
|
|
parser.add_argument("path", help="Gitea path beginning with /api/v1/")
|
|
data_group = parser.add_mutually_exclusive_group()
|
|
data_group.add_argument("--data-json", help="JSON object/array request body")
|
|
data_group.add_argument(
|
|
"--data-file", type=Path, help="path to a JSON request body"
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def load_data(args: argparse.Namespace) -> object | None:
|
|
"""Decode the optional JSON body without involving a shell expansion."""
|
|
if args.data_json is not None:
|
|
return json.loads(args.data_json)
|
|
if args.data_file is not None:
|
|
return json.loads(args.data_file.read_text(encoding="utf-8"))
|
|
return None
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
"""Execute the request, print only its response body, and return HTTP status."""
|
|
args = parse_args(argv)
|
|
try:
|
|
request = build_request(
|
|
args.method,
|
|
args.path,
|
|
base_url=os.environ.get("GITEA_BASE_URL", DEFAULT_BASE_URL),
|
|
token=read_token(),
|
|
data=load_data(args),
|
|
)
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
body = response.read()
|
|
if body:
|
|
sys.stdout.buffer.write(body)
|
|
if not body.endswith(b"\n"):
|
|
sys.stdout.buffer.write(b"\n")
|
|
return 0
|
|
except urllib.error.HTTPError as exc:
|
|
body = exc.read(65536)
|
|
print(f"Gitea API returned 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, ValueError, json.JSONDecodeError) as exc:
|
|
print(f"Gitea API request failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|