""" AUTH.3E — forgejo_source_loader.py =================================== Forgejo Source Loader für den SAVE-Executor. Lädt Content aus der Forgejo Source of Truth, exakt commit-bound. Eigenschaften: * Commit-bound: liest Content aus genau dem source_commit (kein Branch-Latest). * object_id -> Pfad: leitet den Repo-relativen Pfad aus dem vault_path ab (kanonisch RELATIV, AUTH.4C2 PATH CONTRACT REPAIR OPTION A), konsistent mit C5C (_vault_path). * Keine Write-Fähigkeit. Keine automatische Upstream-Integration. * FAIL CLOSED bei: commit missing, object missing, hash mismatch, provenance mismatch, path mismatch, Forgejo unavailable. * Read-only git-Befehle (git show / git cat-file) gegen den lokalen Clone ODER HTTP-API (public Repo, kein Credential nötig). Zugriff: Repo nexo312/trading-system-docs ist PUBLIC -> kein Credential nötig. """ from __future__ import annotations import hashlib import os import subprocess from typing import Any, Dict, Optional # Forgejo-Repo (autoritative Source of Truth) DEFAULT_FORGEJO_REPO = "nexo312/trading-system-docs" ENV_FORGEJO_REPO = "C5_FORGEJO_REPO" # Lokaler Clone-Pfad (falls vorhanden) — sonst HTTP-API DEFAULT_CLONE_PATH = "/data/forgejo-clone" ENV_CLONE_PATH = "C5_FORGEJO_CLONE" # Forgejo HTTP-API Basis (public Repo, read-only) DEFAULT_FORGEJO_BASE = "http://forgejo-c4u8yyi1eaz1gepn3pqmr5fb:3000" ENV_FORGEJO_BASE = "C5_FORGEJO_BASE" class SourceLoaderError(Exception): pass class SourceUnavailableError(SourceLoaderError): pass class SourceMismatchError(SourceLoaderError): pass class ForgejoSourceLoader: """ Lädt Content aus Forgejo, exakt commit-bound. load(source_commit, object_id, vault_path) -> content (str) """ def __init__(self, repo: Optional[str] = None, clone_path: Optional[str] = None, forgejo_base: Optional[str] = None): self.repo = (repo or os.environ.get(ENV_FORGEJO_REPO) or DEFAULT_FORGEJO_REPO) self.clone_path = (clone_path or os.environ.get(ENV_CLONE_PATH) or DEFAULT_CLONE_PATH) self.forgejo_base = (forgejo_base or os.environ.get(ENV_FORGEJO_BASE) or DEFAULT_FORGEJO_BASE).rstrip("/") # -- Pfad-Ableitung ----------------------------------------------------- def _rel_path(self, vault_path: str) -> str: """Leitet den Repo-relativen Pfad aus dem vault_path ab. KANONISCH (AUTH.4C2 PATH CONTRACT REPAIR, OPTION A): vault_path ist bereits der RELATIVE Pfad unter dem Vault-Root (z.B. 'tolaria/auth4c2-canary.md'). Er wird unverändert als rel_path verwendet — KEIN /app/vault/-Prefix-Stripping mehr. Konsistent mit C5C _vault_path (rel_path unter Vault-Root). """ vp = vault_path or "" rel = vp.lstrip("/") if not rel: raise SourceMismatchError("vault_path ergibt keinen rel_path") return rel # -- Content-Load (commit-bound) ---------------------------------------- def load(self, source_commit: str, object_id: str, vault_path: str) -> str: """Lädt Content aus dem exakten Commit. Returns: autoritativer Content (str). Raises: SourceUnavailableError / SourceMismatchError (FAIL CLOSED). """ rel_path = self._rel_path(vault_path) # 1. Commit-bound lesen (kein Branch-Latest) content = self._read_from_commit(source_commit, rel_path) if content is None: raise SourceUnavailableError( f"object {rel_path} nicht in commit {source_commit}") # 2. object_id-Konsistenz prüfen (Frontmatter id: object/) self._validate_object_id(content, object_id, rel_path) return content def _read_from_commit(self, commit: str, rel_path: str) -> Optional[str]: """Liest Datei aus exakt dem Commit. Bevorzugt lokalen Clone, sonst HTTP.""" # Versuche lokalen Clone (falls gemountet) if os.path.isdir(self.clone_path): try: return self._read_from_clone(commit, rel_path) except SourceUnavailableError: # Clone nicht verfügbar -> Fallback auf HTTP-API pass # HTTP-API (public Repo, read-only) return self._read_from_http(commit, rel_path) def _read_from_clone(self, commit: str, rel_path: str) -> Optional[str]: """git show : gegen lokalen Clone.""" try: proc = subprocess.run( ["git", "-C", self.clone_path, "show", f"{commit}:{rel_path}"], capture_output=True, text=True, timeout=30) except (subprocess.SubprocessError, OSError) as e: raise SourceUnavailableError(f"git show fehlgeschlagen: {e}") if proc.returncode != 0: return None # object nicht in commit return proc.stdout def _read_from_http(self, commit: str, rel_path: str) -> Optional[str]: """HTTP-API: GET /api/v1/repos/{repo}/raw/{commit}/{rel_path} (public).""" import urllib.error import urllib.request url = (f"{self.forgejo_base}/api/v1/repos/{self.repo}" f"/raw/{commit}/{rel_path}") req = urllib.request.Request(url, method="GET") try: with urllib.request.urlopen(req, timeout=30) as resp: return resp.read().decode("utf-8") except urllib.error.HTTPError as e: if e.code in (404, 410): return None # object nicht in commit raise SourceUnavailableError(f"Forgejo HTTP {e.code} auf {rel_path}") except (urllib.error.URLError, TimeoutError, OSError) as e: raise SourceUnavailableError(f"Forgejo nicht erreichbar: {e}") # -- object_id-Konsistenz ---------------------------------------------- def _validate_object_id(self, content: str, object_id: str, rel_path: str) -> None: """Prüft, dass das Frontmatter die erwartete object_id enthält.""" fm = self._parse_frontmatter(content) fm_id = fm.get("id") if not fm_id: raise SourceMismatchError( f"object {rel_path} hat keine id im Frontmatter") # object_id im Job ist 'object/' oder ''; Frontmatter id # ist 'object/'. Normalisiere. expected = object_id if not expected.startswith("object/"): expected = f"object/{expected}" if fm_id != expected: raise SourceMismatchError( f"object_id mismatch: Frontmatter={fm_id} Job={expected}") @staticmethod def _parse_frontmatter(content: str) -> Dict[str, Any]: """Minimaler Frontmatter-Parser (--- ... ---).""" if not content.startswith("---"): return {} end = content.find("\n---", 3) if end == -1: return {} fm_text = content[3:end] fm: Dict[str, Any] = {} for line in fm_text.splitlines(): if ":" in line: k, _, v = line.partition(":") fm[k.strip()] = v.strip().strip('"').strip("'") return fm # -- Hash/Provenance (RECOMPUTE) --------------------------------------- @staticmethod def content_hash(content: str) -> str: """SHA-256 des fachlichen Bodies (nach Frontmatter).""" body = content if content.startswith("---"): end = content.find("\n---", 3) if end != -1: body = content[end + 4:] return hashlib.sha256(body.encode("utf-8")).hexdigest() @staticmethod def provenance_hash(content: str) -> str: """SHA-256 des gesamten Contents (konsistent mit save_executor_core).""" return hashlib.sha256(content.encode("utf-8")).hexdigest()