Nick
4eefc15837
- scripts/gitea_auth.py: single credential ladder (env -> 0600 token file -> git credential fill); 6 consumers migrated off direct os.environ reads - rune_operator: consolidate 5 duplicated failure blocks into _terminalize_execution_failure; fix 3 fault-injection gaps: (1) dual-module PreparedTool identity split crashed approval-path saves, (2) fallback-model failure escaped send_task (created/pending orphan), (3) running-transition save uncaught (running-with-no-owner) - envelope task.status now agrees with canonical ledger (awaiting_approval) - orphan task_ac35f4e287c8 terminalized failed with durable evidence - tests: gitea auth ladder (6), lifecycle fault matrix (9 boundaries + restart/reload + helper contract); 1280 green in ci-venv; ratchet 352/80
113 lines
3.6 KiB
Python
113 lines
3.6 KiB
Python
"""gitea_auth.py — canonical Gitea API authentication (handoff 2026-09-11 §H).
|
|
|
|
One deterministic helper for authenticated Gitea API access. Never prints
|
|
credentials. Fails clearly when unavailable. Credential ladder:
|
|
|
|
1. GITEA_TOKEN environment variable (preferred for CI/deployment)
|
|
2. private token file (GITEA_TOKEN_FILE, default
|
|
~/.config/opencode/secrets/gitea-token, must be mode 0600)
|
|
3. git credential fill (uses the credential helper / store)
|
|
|
|
Never parses credentials from git remote URLs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
DEFAULT_SERVER = "https://gitea.thingscouldgetdicey.online"
|
|
DEFAULT_REPO = "conductor/rune"
|
|
DEFAULT_TOKEN_FILE = Path("~/.config/opencode/secrets/gitea-token").expanduser()
|
|
GITEA_HOST = "gitea.thingscouldgetdicey.online"
|
|
|
|
|
|
class GiteaCredentialError(RuntimeError):
|
|
"""Raised when no Gitea credential is available. Message never contains one."""
|
|
|
|
|
|
def _token_from_file() -> str:
|
|
path = Path(os.environ.get("GITEA_TOKEN_FILE", DEFAULT_TOKEN_FILE)).expanduser()
|
|
if not path.is_file():
|
|
return ""
|
|
mode = stat.S_IMODE(path.stat().st_mode)
|
|
if mode & 0o077:
|
|
raise PermissionError(f"Gitea token file must be mode 0600: {path}")
|
|
return path.read_text(encoding="utf-8").strip()
|
|
|
|
|
|
def _token_from_credential_helper() -> str:
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "credential", "fill"],
|
|
input=f"protocol=https\nhost={GITEA_HOST}\nusername=conductor\n\n",
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
for line in result.stdout.splitlines():
|
|
if line.startswith("password="):
|
|
return line[len("password=") :]
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
pass
|
|
return ""
|
|
|
|
|
|
def has_token() -> bool:
|
|
"""True if any rung of the credential ladder answers. Never raises."""
|
|
try:
|
|
get_token()
|
|
return True
|
|
except Exception: # noqa: BLE001 — probe-only: any failure means "no credential available"
|
|
return False
|
|
|
|
|
|
def get_token() -> str:
|
|
"""Obtain the Gitea API token. Never prints the token."""
|
|
token = os.environ.get("GITEA_TOKEN", "").strip()
|
|
if token:
|
|
return token
|
|
token = _token_from_file()
|
|
if token:
|
|
return token
|
|
token = _token_from_credential_helper()
|
|
if token:
|
|
return token
|
|
raise GiteaCredentialError(
|
|
"Gitea credentials unavailable: set GITEA_TOKEN, create a mode-0600 "
|
|
f"token file at {DEFAULT_TOKEN_FILE}, or configure a git credential "
|
|
f"helper for {GITEA_HOST}"
|
|
)
|
|
|
|
|
|
def api(
|
|
path: str,
|
|
method: str = "GET",
|
|
data: dict | None = None,
|
|
server: str = DEFAULT_SERVER,
|
|
repo: str = DEFAULT_REPO,
|
|
) -> tuple[int, Any]:
|
|
"""Make an authenticated Gitea API request. Returns (status_code, parsed_body)."""
|
|
token = get_token()
|
|
url = f"{server}/api/v1/repos/{repo}{path}"
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(data).encode() if data is not None else None,
|
|
headers={"Authorization": f"token {token}", "Content-Type": "application/json"},
|
|
method=method,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
body, status = r.read(), r.status
|
|
except urllib.error.HTTPError as e:
|
|
body, status = e.read(), e.code
|
|
try:
|
|
return status, (json.loads(body) if body else {})
|
|
except json.JSONDecodeError:
|
|
return status, {"_raw": body.decode("utf-8", "replace")[:200]}
|