diff --git a/SCOPE.md b/SCOPE.md index fe83a6e..46a43b9 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -390,3 +390,8 @@ freshness receipts. The reviewed candidate in `k8s/custodian-release/` is outsid active kustomization; placeholders prevent claiming an accepted image/snapshot. Both definitions stay disabled. No runtime publication, retention deletion or schedule activation is implemented by these source artifacts. + +Standalone `custodian_capture.py` prepares bounded publisher response reception +and retained descriptor advancement. It is not registered as another bounded +operation and is not integrated into the daily command. The checkpoint and +remaining timeout/transport/access work are in `k8s/custodian-release/RESTART.md`. diff --git a/k8s/custodian-release/RESTART.md b/k8s/custodian-release/RESTART.md new file mode 100644 index 0000000..2c0a5be --- /dev/null +++ b/k8s/custodian-release/RESTART.md @@ -0,0 +1,63 @@ +# Runtime delivery checkpoint — 2026-10-19 + +User requested wind-down. No deployment, credential read, network-policy apply +or schedule activation occurred. Both daily and weekly definitions stay disabled. + +Committed prerequisites: Activity Core fed119a (pinned image bundle, required +freshness admission), 3b25aea (immutable descriptor selection); Repo Manager +3e2726f (offline exporter), 510ba0b (targeted freshness). + +Latest source preparation adds Repo Manager's opt-in fixed-policy snapshot +endpoint and Activity Core's standalone `custodian_capture.py` receiver. The +endpoint defaults disabled, reuses the publisher's existing Forge credential, +and accepts no caller repository/ref/path/command. Receiver validates and retains +a complete captured bundle before replacing its regular descriptor. It is NOT +wired into the daily operation or definition. The deployment candidate therefore +does NOT deliver recurring captures yet. No new API or secret is live. + +Validation already performed: Repo Manager publisher/capture/freshness/exporter +tests: 47 passed (one existing Starlette/httpx deprecation warning). Activity Core +receiver + prior focused tests: 79 passed / 1 failed when a second bounded stage +was temporarily added. Failure was the owner's explicit one-operation-per- +definition contract. That unsupported wiring was removed entirely from registry, +activity dispatch, definition and expected registry test. Do not reintroduce it. +The four standalone receiver tests passed within that run. Earlier descriptor +checkpoint: 76 focused tests passed. See final checkpoint report for the minimal +post-removal check. No container build or runtime acceptance is claimed. + +Restart work: + +1. Inspect clean Git state and actual installed source bytes before promotion. + Review endpoint/receiver source and run lint plus focused tests before extending. +2. Preserve the single bounded operation. Integrate capture as a fixed pre-run + phase only with a defensible shared timeout: proposed <=45 seconds capture + plus <=240 seconds owner command within existing300-second Temporal envelope. + Current standalone HTTP client has250-second per-I/O timeout and publisher + budget240 seconds; these are NOT suitable for that combined envelope yet. +3. Prove cancellation/deadline behavior. Current endpoint's process-local lock + refuses concurrent capture with409 but HTTP disconnect does not cancel its + synchronous owner work, and lock state is not durable/across replicas. Define + retained in-progress receipt/retry behavior or stop producer work; do not claim + this solved. Measure actual15-repo capture completion, fail on partial coverage, + never silently widen freshness age/timeouts to hide it. +4. Complete reviewed transport placement. Existing core-hub publisher endpoint + has no adopted NetworkPolicy ingress from Activity Core or Hub egress. The + precise downstream15-repo private workplan/decision disclosure and port8020/ + Hub8000 delta is in `publisher-access-delta.md`; no live approval is implied. + Prepare actual wrapper manifests only after that exact review. Preserve + CCR0031 identity/namespace; do not add a worker Forge credential. +5. Receiver requires a writable snapshot path. Current candidate mounts the + snapshot parent read-only for commands; it has no receiver write alias/binding. + Review a narrow publisher-output mount and verify both paths refer to the + same retained store before wiring. Keep private state0700 and immutable names. +6. Update Dockerfile test list if receiver becomes integrated. Current pinned + package/descriptor tests are included; standalone receiver test is not yet + added to image-stage tests. Rebuild immutable image and verify mounted bytes. +7. Only after source/runtime review: storage capacity/backup, scoped network + admission, rollout headroom/rollback/abort owner, then separate first-run and + unattended receipts. Retention settings remain recommended policy; no pruning + is installed. Weekly collector coverage and acceptance remain separate. + +Endpoint fixed payload cap is8MiB (consumer snapshot policy64MiB upper bound), +response cap16MiB; neither is a measured production requirement. These fail closed +and need owner sizing evidence. Runtime source has no new resource requests. diff --git a/src/activity_core/custodian_capture.py b/src/activity_core/custodian_capture.py new file mode 100644 index 0000000..73b4a88 --- /dev/null +++ b/src/activity_core/custodian_capture.py @@ -0,0 +1,115 @@ +"""Receive fixed-policy owner snapshots; no Forge credential or arbitrary destination.""" +import base64 +import fcntl +import hashlib +import json +import os +from pathlib import Path, PurePosixPath +import re +import shutil +import tempfile +from urllib.parse import urlsplit + +import httpx + +from activity_core.custodian_inputs import _read, admit +from activity_core.custodian_operations import _path, _private_state + + +def receive(bundle, state, snapshots, policy_path): + state, snapshots = Path(state), Path(snapshots) + snapshot_id = bundle.get('snapshot_id', '') + if bundle.get('schema') != 'repo-manager.custodian-capture.v1' or not re.fullmatch('[0-9a-f]{64}', snapshot_id): + raise ValueError('invalid captured snapshot contract') + manifest_raw, receipts_raw = bundle['manifest'].encode(), bundle['receipts'].encode() + mh, rh = hashlib.sha256(manifest_raw).hexdigest(), hashlib.sha256(receipts_raw).hexdigest() + if mh != bundle['manifest_sha256'] or rh != bundle['receipts_sha256'] or snapshot_id != hashlib.sha256((mh + rh).encode()).hexdigest(): + raise ValueError('capture digest mismatch') + policy = json.loads(_read(policy_path)) + manifest = json.loads(manifest_raw) + expected = {row['slug'] + '/' + entry['path'] for row in manifest['repositories'] for entry in row['files']} + if set(bundle['files']) != expected: + raise ValueError('capture files differ from manifest') + inputs = _path(str(state / 'inputs'), directory=True) + receipts = inputs / (snapshot_id + '.receipts.json') + selected = snapshots / snapshot_id + temporary = Path(tempfile.mkdtemp(prefix='.capture-', dir=snapshots)) + try: + total = 0 + for name, encoded in bundle['files'].items(): + relative = PurePosixPath(name) + if relative.is_absolute() or '..' in relative.parts or len(relative.parts) < 3: + raise ValueError('capture path escapes snapshot') + data = base64.b64decode(encoded, validate=True) + total += len(data) + if total > min(policy['max_snapshot_bytes'], 8 * 1024 * 1024): + raise ValueError('capture exceeds size policy') + target = temporary / relative + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(data) + target.chmod(0o400) + (temporary / 'manifest.json').write_bytes(manifest_raw) + (temporary / 'COMPLETE').write_text(mh + '\n') + # Validate receipts before advancing any descriptor. Retained name is immutable. + _path(str(receipts), existing=False) + if receipts.exists(): + if receipts.read_bytes() != receipts_raw: + raise ValueError('retained receipt conflict') + else: + with receipts.open('xb') as stream: + stream.write(receipts_raw) + admit(temporary, policy_path, receipts, policy['repositories']) + _path(str(selected), directory=True, existing=False) + if selected.exists(): + if _read(selected / 'manifest.json') != manifest_raw: + raise ValueError('retained snapshot conflict') + admit(selected, policy_path, receipts, policy['repositories']) + else: + temporary.rename(selected) + descriptor = {'schema': 'activity-core.custodian-snapshot-selection.v1', 'snapshot_id': snapshot_id, + 'manifest_sha256': mh, 'receipts_sha256': rh} + current = _path(str(inputs / 'current.json'), existing=False) + fd, staged = tempfile.mkstemp(prefix='.descriptor-', dir=inputs) + try: + with os.fdopen(fd, 'w') as stream: + json.dump(descriptor, stream, sort_keys=True) + stream.flush() + os.fsync(stream.fileno()) + os.replace(staged, current) + finally: + if os.path.exists(staged): + os.unlink(staged) + return {'kind': 'custodian_snapshot_capture', 'status': 'completed', 'snapshot_id': snapshot_id, + 'manifest_sha256': mh, 'receipts_sha256': rh, 'repository_count': len(policy['repositories'])} + finally: + if temporary.exists(): + shutil.rmtree(temporary) + + +def run_capture(params, *, transport=None): + if set(params) - {'evidence_sinks'}: + raise ValueError('capture accepts no definition command, path or endpoint') + state = _private_state() + snapshots = _path(os.environ.get('ACTIVITY_CORE_CUSTODIAN_SNAPSHOT_WRITE_ROOT'), directory=True) + policy = _path(os.environ.get('ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY')) + base = os.environ.get('ACTIVITY_CORE_CUSTODIAN_PUBLISHER_URL', '') + url = urlsplit(base) + if url.scheme not in {'http', 'https'} or not url.hostname or url.username or url.password or url.query or url.fragment or url.path not in {'', '/'}: + raise ValueError('explicit non-credential publisher base URL required') + lock = _path(str(state / 'capture.lock'), existing=False) + with lock.open('w') as stream: + fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB) + try: + with httpx.Client(transport=transport, follow_redirects=False, timeout=250) as client: + with client.stream('POST', base.rstrip('/') + '/ports/repositories/custodian-snapshot') as response: + if response.status_code != 200: + raise ValueError('owner snapshot publisher unavailable') + data = bytearray() + for chunk in response.iter_bytes(): + data.extend(chunk) + if len(data) > 16 * 1024 * 1024: + raise ValueError('capture response exceeds size bound') + return receive(json.loads(data), state, snapshots, policy) + except Exception as exc: + # Never persist raw HTTP payloads, paths or credential-bearing error text. + raise ValueError('bounded owner snapshot reception failed') from exc diff --git a/tests/test_custodian_capture.py b/tests/test_custodian_capture.py new file mode 100644 index 0000000..e70709b --- /dev/null +++ b/tests/test_custodian_capture.py @@ -0,0 +1,74 @@ +import base64 +from datetime import datetime, timezone +import hashlib +import json + +import httpx +import pytest + +from activity_core.custodian_capture import receive, run_capture + + +@pytest.fixture +def capture_runtime(tmp_path, monkeypatch): + state, snapshots = tmp_path / 'state', tmp_path / 'snapshots' + state.mkdir(mode=0o700) + (state / 'inputs').mkdir() + snapshots.mkdir() + policy = state / 'policy.json' + policy.write_text(json.dumps({'schema': 'activity-core.custodian-input-policy.v1', 'repositories': ['example'], + 'max_snapshot_bytes': 1024, 'max_observation_age_seconds': 7200, 'max_future_skew_seconds': 300, + 'expected_hub_instance': 'railiance01'})) + for name, value in {'STATE_ROOT': state, 'SNAPSHOT_WRITE_ROOT': snapshots, 'INPUT_POLICY': policy, + 'PUBLISHER_URL': 'http://publisher.invalid'}.items(): + monkeypatch.setenv('ACTIVITY_CORE_CUSTODIAN_' + name, str(value)) + manifest = {'schema': 'repo-manager.fleet-input.v1', 'repositories': [{'slug': 'example', 'commit': 'a' * 40, + 'files': [{'path': 'workplans/task.md', 'bytes': 4, 'sha256': hashlib.sha256(b'task').hexdigest()}]}]} + receipts = [{'schema': 'repo-manager.freshness-receipt.v1', 'repo_slug': 'example', 'outcome': 'applied', + 'finished_at': datetime.now(timezone.utc).isoformat(), + 'hub': {'verified_primary': True, 'instance_role': 'primary', 'instance_label': 'railiance01'}, + **{key: 'a' * 40 for key in ('forge_commit', 'expected_commit', 'derived_commit', 'projected_commit')}}] + m, r = json.dumps(manifest), json.dumps(receipts) + mh, rh = hashlib.sha256(m.encode()).hexdigest(), hashlib.sha256(r.encode()).hexdigest() + bundle = {'schema': 'repo-manager.custodian-capture.v1', 'manifest': m, 'receipts': r, + 'manifest_sha256': mh, 'receipts_sha256': rh, + 'snapshot_id': hashlib.sha256((mh + rh).encode()).hexdigest(), + 'files': {'example/workplans/task.md': base64.b64encode(b'task').decode()}} + return state, snapshots, policy, bundle + + +def test_receiver_retains_before_selecting_and_identical_retry(capture_runtime): + state, snapshots, policy, bundle = capture_runtime + first = receive(bundle, state, snapshots, policy) + assert receive(bundle, state, snapshots, policy) == first + descriptor = json.loads((state / 'inputs/current.json').read_text()) + assert descriptor['snapshot_id'] == bundle['snapshot_id'] + assert (snapshots / bundle['snapshot_id'] / 'example/workplans/task.md').read_text() == 'task' + + +def test_failed_capture_does_not_advance_descriptor(capture_runtime): + state, snapshots, policy, bundle = capture_runtime + (state / 'inputs/current.json').write_text('unchanged') + bundle['files']['example/workplans/task.md'] = base64.b64encode(b'bad').decode() + with pytest.raises(ValueError): + receive(bundle, state, snapshots, policy) + assert (state / 'inputs/current.json').read_text() == 'unchanged' + assert not (snapshots / bundle['snapshot_id']).exists() + + +def test_fixed_endpoint_transport_sends_no_forge_credential(capture_runtime): + bundle = capture_runtime[3] + def handler(request): + assert request.method == 'POST' + assert request.url.path == '/ports/repositories/custodian-snapshot' + assert 'authorization' not in request.headers + return httpx.Response(200, json=bundle) + assert run_capture({}, transport=httpx.MockTransport(handler))['status'] == 'completed' + + +def test_transport_error_sanitized(capture_runtime): + def handler(request): + return httpx.Response(500, text='sensitive upstream details') + with pytest.raises(ValueError, match='bounded owner snapshot reception failed') as error: + run_capture({}, transport=httpx.MockTransport(handler)) + assert 'sensitive' not in str(error.value)