diff --git a/SCOPE.md b/SCOPE.md index 778cf32..b9a31e8 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -115,3 +115,10 @@ active repository set using the existing Forge/Hub verification and receipts. Missing active repositories or limits smaller than the set fail closed; partial budget/failure results still require caller coverage checks. This does not install the disabled schedule or add any credential, transport or producer. + +The runtime has a default-disabled fixed-policy Custodian snapshot endpoint, +reusing existing Forge read settings and targeted native freshness receipts. +It prepares bounded exact-commit workplan/decision bundles, not a general file +service. It is not deployed or admitted to Activity Core. Producer cancellation, +combined daily timing, receiver integration and reviewed downstream private-source +access remain unfinished; see `docs/custodian-fleet-input-contract.md`. diff --git a/docs/custodian-fleet-input-contract.md b/docs/custodian-fleet-input-contract.md index 076942a..12298ba 100644 --- a/docs/custodian-fleet-input-contract.md +++ b/docs/custodian-fleet-input-contract.md @@ -129,3 +129,33 @@ Local tests use temporary Git repositories and prove committed-byte export despi dirty files, strict path selection, hashes/completion marker, immutable names, invalid-request refusal, selected symlink refusal and size bounds. They perform no network, production reads, schedule changes or Hub writes. + +## Wind-down checkpoint: disabled remote capture source + +`custodian_snapshot.py` and the existing runtime now prepare a fixed-policy +`POST /ports/repositories/custodian-snapshot` endpoint. Default is404/disabled. +Configuration requires `REPO_MANAGER_CUSTODIAN_SNAPSHOT_ENABLED=true`, +`REPO_MANAGER_CUSTODIAN_SNAPSHOT_POLICY` and +`REPO_MANAGER_CUSTODIAN_SNAPSHOT_HUB`; existing PublisherSettings supplies the +authorized Forge reader. Existing optional API-token admission is preserved. +No caller paths/refs/repos/commands are used. No source endpoint is deployed. + +Capture targets the complete configured set through native freshness, then reads +allowlisted files at those exact commits through Forge contents API. It verifies +Git blob identity, refuses unexpected/symlink files and returns a content- +addressed manifest/receipt/file bundle. Default hard payload bound8MiB and +per-response bound3MiB fail closed. Publisher state has no new persistent volume. + +Validation: 47 publisher/capture/freshness/exporter tests passed with mocked HTTP; +one existing Starlette/httpx deprecation warning. Source lint review remains a +restart check. The source endpoint uses a process-local concurrency lock and +240-second native budget; it does NOT cancel producer work on client disconnect +or retain durable in-progress receipt state. Do not treat it as ready for the +proposed45-second combined daily pre-run phase. Existing15-repo capture timing +has not been measured. Activity Core receiver is standalone and unintegrated; +see its `k8s/custodian-release/RESTART.md` for exact restart steps. + +The core-hub publisher's existing CCR0031 reader remains in that namespace. New +Activity Core access to private workplan/decision content and publisher-to-Hub +egress require the explicit source-reviewed data/network delta before live +promotion. No credential route, access scope or schedule was changed here. diff --git a/src/repo_manager/custodian_snapshot.py b/src/repo_manager/custodian_snapshot.py new file mode 100644 index 0000000..3b1f8c8 --- /dev/null +++ b/src/repo_manager/custodian_snapshot.py @@ -0,0 +1,112 @@ +"""Opt-in, bounded fixed-policy capture using the publisher's existing Forge reader.""" +from __future__ import annotations + +import base64 +from dataclasses import replace +import hashlib +import json +from pathlib import Path, PurePosixPath +import tempfile +import time +from urllib.parse import quote + +import httpx + +from repo_manager.freshness_receipts import load_receipts +from repo_manager.freshness_runner import RunConfig, run + + +def capture(policy: dict, cfg: RunConfig, *, hub_transport=None, forge_transport=None) -> dict: + if policy.get("schema") != "activity-core.custodian-input-policy.v1" or not cfg.forge_token: + raise ValueError("fixed capture policy and existing authenticated reader required") + repos = tuple(policy["repositories"]) + if not repos or len(repos) > 15: + raise ValueError("capture policy repository bound exceeded") + deadline = time.monotonic() + cfg.max_seconds + # Cap in-memory capture independently of the consumer's broader retention bound. + cap = min(int(policy["max_snapshot_bytes"]), 8 * 1024 * 1024) + total = 0 + files = {} + manifest = {"schema": "repo-manager.fleet-input.v1", "freshness_verified": False, + "repositories": []} + with tempfile.TemporaryDirectory(prefix="rmgr-custodian-capture-") as temporary: + selected_cfg = replace(cfg, repo_slugs=repos, limit=len(repos), dry_run=False, + receipts_path=Path(temporary) / "receipts.jsonl") + summary = run(selected_cfg, hub_transport=hub_transport, forge_transport=forge_transport) + receipts = load_receipts(selected_cfg.receipts_path) + accepted = {r["repo_slug"]: r for r in receipts if r["outcome"] in {"applied", "noop"}} + if summary["aborted"] or summary["deferred"] or set(accepted) != set(repos): + raise ValueError("incomplete targeted freshness capture") + manifest["observed_at"] = summary["finished_at"] + with httpx.Client(base_url=cfg.forge_base.rstrip("/"), follow_redirects=False, + headers={"Authorization": f"token {cfg.forge_token}"}, + transport=forge_transport) as forge: + def fetch(slug, relative, commit, optional=False): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ValueError("capture deadline exceeded") + url = f"/api/v1/repos/{quote(cfg.forge_owner, safe='')}/{quote(slug, safe='')}/contents/{quote(relative, safe='/')}" + # Streaming bound applies even if upstream lies about Content-Length. + with forge.stream("GET", url, params={"ref": commit}, timeout=min(15, remaining)) as response: + if response.status_code == 404 and optional: + return [] + if response.status_code != 200: + raise ValueError("pinned Forge content unavailable") + raw = bytearray() + for chunk in response.iter_bytes(): + raw.extend(chunk) + if len(raw) > 3 * 1024 * 1024: + raise ValueError("Forge content response exceeds limit") + return json.loads(raw) + + for slug in sorted(repos): + commit = accepted[slug]["forge_commit"] + records = [] + queue = [("workplans", False), ("decisions", True)] + visited = set() + while queue: + directory, optional = queue.pop() + if directory in visited or len(visited) >= 256: + raise ValueError("duplicate or excessive source directories") + visited.add(directory) + entries = fetch(slug, directory, commit, optional) + if not isinstance(entries, list) or len(entries) > 10000: + raise ValueError("invalid source directory") + for entry in entries: + relative = entry.get("path", "") + path = PurePosixPath(relative) + if path.is_absolute() or ".." in path.parts or str(path.parent) != directory: + raise ValueError("source path escapes selected directory") + if entry.get("type") == "dir" and directory.startswith("workplans"): + queue.append((relative, False)) + elif path.suffix == ".md": + if entry.get("type") != "file": + raise ValueError("source record is not regular file") + item = fetch(slug, relative, commit) + if item.get("type") != "file" or item.get("encoding") != "base64": + raise ValueError("invalid source file representation") + data = base64.b64decode(item["content"], validate=True) + total += len(data) + if len(data) > 2 * 1024 * 1024 or total > cap: + raise ValueError("capture content limit exceeded") + blob = hashlib.sha1(b"blob " + str(len(data)).encode() + b"\0" + data).hexdigest() + if blob != entry.get("sha") or blob != item.get("sha"): + raise ValueError("source blob identity mismatch") + key = slug + "/" + relative + if key in files: + raise ValueError("duplicate source record") + files[key] = base64.b64encode(data).decode() + records.append({"path": relative, "bytes": len(data), + "sha256": hashlib.sha256(data).hexdigest()}) + if not any(r["path"].startswith("workplans/") for r in records): + raise ValueError("required workplan records absent") + manifest["repositories"].append({"slug": slug, "commit": commit, + "files": sorted(records, key=lambda r: r["path"])}) + raw_manifest = json.dumps(manifest, sort_keys=True, indent=2) + "\n" + raw_receipts = json.dumps(receipts, sort_keys=True, indent=2) + "\n" + manifest_hash = hashlib.sha256(raw_manifest.encode()).hexdigest() + receipts_hash = hashlib.sha256(raw_receipts.encode()).hexdigest() + return {"schema": "repo-manager.custodian-capture.v1", + "snapshot_id": hashlib.sha256((manifest_hash + receipts_hash).encode()).hexdigest(), + "manifest": raw_manifest, "manifest_sha256": manifest_hash, + "receipts": raw_receipts, "receipts_sha256": receipts_hash, "files": files} diff --git a/src/repo_manager/runtime.py b/src/repo_manager/runtime.py index cdac4ab..4c43b7e 100644 --- a/src/repo_manager/runtime.py +++ b/src/repo_manager/runtime.py @@ -5,6 +5,8 @@ from __future__ import annotations import hmac import os import secrets +import threading +import json from contextlib import asynccontextmanager from dataclasses import dataclass from pathlib import Path @@ -32,6 +34,8 @@ class PublisherSettings: forgejo_timeout_seconds: float = 10.0 forgejo_retries: int = 2 remote_workers: int = 8 + custodian_snapshot_policy: Path | None = None + custodian_snapshot_hub: str | None = None @classmethod def from_env(cls) -> PublisherSettings: @@ -51,6 +55,8 @@ class PublisherSettings: raise ValueError("Forgejo token file is empty") return cls( registry_path=Path(registry) if registry else None, + custodian_snapshot_policy=Path(os.environ["REPO_MANAGER_CUSTODIAN_SNAPSHOT_POLICY"]) if os.getenv("REPO_MANAGER_CUSTODIAN_SNAPSHOT_ENABLED") == "true" else None, + custodian_snapshot_hub=os.getenv("REPO_MANAGER_CUSTODIAN_SNAPSHOT_HUB"), cursor_secret=os.getenv("REPO_MANAGER_CURSOR_SECRET") or secrets.token_urlsafe(32), api_token=os.getenv("REPO_MANAGER_API_TOKEN"), page_size=int(os.getenv("REPO_MANAGER_PROJECTION_PAGE_SIZE", "100")), @@ -149,6 +155,34 @@ def create_app(*, settings: PublisherSettings | None = None) -> FastAPI: status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc) ) from exc + capture_lock = threading.Lock() + + @app.post("/ports/repositories/custodian-snapshot", tags=["port.repo"]) + def custodian_snapshot(authorization: str | None = Header(default=None)) -> dict: + # No caller paths, refs, repo names, URLs or commands are accepted. + if resolved.custodian_snapshot_policy is None or not resolved.custodian_snapshot_hub: + raise HTTPException(status_code=404, detail="snapshot publication disabled") + if resolved.api_token is not None and not hmac.compare_digest( + authorization or "", f"Bearer {resolved.api_token}" + ): + raise HTTPException(status_code=401, detail="unauthorized") + if not capture_lock.acquire(blocking=False): + raise HTTPException(status_code=409, detail="capture already running") + try: + from repo_manager.custodian_snapshot import capture + from repo_manager.freshness_runner import RunConfig + policy = json.loads(resolved.custodian_snapshot_policy.read_text()) + return capture(policy, RunConfig( + limit=len(policy["repositories"]), max_seconds=240, + api_base=resolved.custodian_snapshot_hub, + forge_base=resolved.forgejo_base_url or "https://forgejo.coulomb.social", + forge_token=resolved.forgejo_token, + )) + except Exception as exc: + raise HTTPException(status_code=503, detail="bounded snapshot capture failed") from exc + finally: + capture_lock.release() + return app diff --git a/tests/test_custodian_snapshot.py b/tests/test_custodian_snapshot.py new file mode 100644 index 0000000..38eb254 --- /dev/null +++ b/tests/test_custodian_snapshot.py @@ -0,0 +1,65 @@ +import base64 +import hashlib +from pathlib import Path + +import httpx +import pytest +from fastapi.testclient import TestClient + +from repo_manager.custodian_snapshot import capture +from repo_manager.freshness_runner import RunConfig +from repo_manager.runtime import PublisherSettings, create_app +from test_freshness_runner import A, FakeHub + + +def fixture_transport(tamper=False): + def handler(request): + path = request.url.path + if path.endswith('/example'): + return httpx.Response(200, json={'default_branch': 'main'}) + if '/branches/' in path: + return httpx.Response(200, json={'commit': {'id': A}}) + assert request.url.params['ref'] == A + data = b'workplan' + sha = hashlib.sha1(b'blob 8\0' + data).hexdigest() + if path.endswith('/contents/workplans'): + return httpx.Response(200, json=[{'path': 'workplans/task.md', 'type': 'file', 'sha': sha}]) + if path.endswith('/contents/decisions'): + return httpx.Response(404) + return httpx.Response(200, json={'type': 'file', 'encoding': 'base64', 'sha': sha, + 'content': base64.b64encode(b'changed' if tamper else data).decode()}) + return httpx.MockTransport(handler) + + +def test_capture_uses_exact_verified_commits_and_no_local_checkout(): + policy = {'schema': 'activity-core.custodian-input-policy.v1', 'repositories': ['example'], 'max_snapshot_bytes': 1024} + result = capture(policy, RunConfig(limit=1, max_seconds=30, forge_token='fixture'), + hub_transport=httpx.MockTransport(FakeHub([('example', 'active')])), + forge_transport=fixture_transport()) + assert result['schema'] == 'repo-manager.custodian-capture.v1' + assert list(result['files']) == ['example/workplans/task.md'] + assert base64.b64decode(result['files']['example/workplans/task.md']) == b'workplan' + assert hashlib.sha256(result['manifest'].encode()).hexdigest() == result['manifest_sha256'] + + +def test_capture_refuses_blob_mismatch(): + policy = {'schema': 'activity-core.custodian-input-policy.v1', 'repositories': ['example'], 'max_snapshot_bytes': 1024} + with pytest.raises(ValueError, match='blob identity'): + capture(policy, RunConfig(limit=1, max_seconds=30, forge_token='fixture'), + hub_transport=httpx.MockTransport(FakeHub([('example', 'active')])), + forge_transport=fixture_transport(tamper=True)) + + +def test_snapshot_endpoint_disabled_by_default(): + with TestClient(create_app(settings=PublisherSettings())) as client: + assert client.post('/ports/repositories/custodian-snapshot').status_code == 404 + + +def test_snapshot_endpoint_uses_existing_auth_before_capture(): + settings = PublisherSettings(api_token='fixture', custodian_snapshot_policy=Path('/missing'), + custodian_snapshot_hub='http://fixture') + with TestClient(create_app(settings=settings)) as client: + assert client.post('/ports/repositories/custodian-snapshot').status_code == 401 + result = client.post('/ports/repositories/custodian-snapshot', headers={'Authorization': 'Bearer fixture'}) + assert result.status_code == 503 + assert result.json()['detail'] == 'bounded snapshot capture failed'