net-kingdom/tools/emission-cadence-profile/receiver_readback.py

#!/usr/bin/env python3
"""Independent receiver readback for a retained cadence capture (NK-WP-0035-T04).

Reads the capture's `capture-provenance.json` (its `receiver_readback` request:
source, tenant, stream instance, window and event IDs) and `events.jsonl`, then
asks audit-core directly, under the operator's own reader token:

- `GET /v1/reconciliation` for per-class counts in the same window and instance;
- `GET /v1/events/<id>` for every captured event ID.

The output is a value-safe receipt with the counts from both sides, a presence
check per event ID and a verdict (`agree` or `diverge`). It keeps no event
payloads, and the token never enters the receipt. Agreement proves neither
completeness nor that King's Guard accepts the stream (KG-WP-0005-T03).

Reach the ClusterIP receiver with your own port-forward, for example
`kubectl -n audit-core port-forward svc/audit-core 18080:8080`.
"""
import argparse
import collections
import json
import os
import stat
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

SCHEMA = 'nk-wp-0035-receiver-readback/v1'


def load_capture(directory):
    provenance = json.loads((directory / 'capture-provenance.json').read_text())
    request = provenance['receiver_readback']
    events = [json.loads(line) for line in (directory / 'events.jsonl').read_text().splitlines() if line.strip()]
    if any(e.get('stream_instance_id') != request['stream_instance_id'] for e in events):
        raise ValueError('capture events span another stream instance')
    return request, collections.Counter(e['event_class'] for e in events)


def read_token(path):
    info = path.lstat()
    if not stat.S_ISREG(info.st_mode) or info.st_mode & 0o077:
        raise ValueError('token file must be a regular file readable only by its owner (0600)')
    token = path.read_text().strip()
    if not token or any(c.isspace() for c in token):
        raise ValueError('token file must hold exactly one bearer token')
    return token


def readback(request, captured, get):
    """get(path_with_query) -> (status, json). Returns the value-safe receipt."""
    query = urllib.parse.urlencode({'source': request['source'], 'tenant': request['tenant'],
                                    'since': request['since'], 'until': request['until'],
                                    'stream_instance_id': request['stream_instance_id']})
    status, body = get('/v1/reconciliation?' + query)
    findings = []
    if status != 200:
        findings.append(f'reconciliation returned {status}: {body.get("error", "")}')
        receiver, basis = {}, None
    else:
        receiver = {row['class']: row['count'] for row in body.get('counts', [])}
        # audit-core marks cross-source answers so they cannot pass for the source's own check.
        basis = body.get('basis')
        if basis != 'independent_reader':
            findings.append(f'reconciliation basis is {basis!r}, not independent_reader')
        for cls in sorted(set(receiver) | set(captured)):
            if receiver.get(cls, 0) != captured.get(cls, 0):
                findings.append(f'class {cls}: captured {captured.get(cls, 0)}, receiver {receiver.get(cls, 0)}')
    presence = {}
    for event_id in request['event_ids']:
        status, body = get('/v1/events/' + urllib.parse.quote(event_id, safe=''))
        presence[event_id] = status == 200
        if status != 200:
            findings.append(f'event {event_id[:12]}… returned {status}')
        elif (body.get('source'), body.get('tenant')) not in ((request['source'], request['tenant']), (None, None)):
            findings.append(f'event {event_id[:12]}… source/tenant differs')
    return {'schema': SCHEMA, 'request': request, 'captured_counts': dict(captured),
            'receiver_counts': receiver, 'receiver_basis': basis, 'event_presence': presence,
            'verdict': 'agree' if not findings else 'diverge', 'findings': findings,
            'payloads_retained': False, 'token_retained': False,
            'means': 'agreement proves neither completeness nor King\'s Guard acceptance'}


def http_get(base, token):
    def get(path):
        req = urllib.request.Request(base.rstrip('/') + path, headers={'Authorization': 'Bearer ' + token})
        try:
            with urllib.request.urlopen(req, timeout=15) as response:
                return response.status, json.loads(response.read() or b'{}')
        except urllib.error.HTTPError as exc:
            try:
                return exc.code, json.loads(exc.read() or b'{}')
            except ValueError:
                return exc.code, {}
    return get


def main(argv=None):
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument('--capture-dir', type=Path, required=True)
    p.add_argument('--base-url', default='http://127.0.0.1:18080')
    p.add_argument('--token-file', type=Path, required=True)
    p.add_argument('--output', type=Path, required=True)
    args = p.parse_args(argv)
    request, captured = load_capture(args.capture_dir)
    receipt = readback(request, captured, http_get(args.base_url, read_token(args.token_file)))
    fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
    with os.fdopen(fd, 'w') as out:
        json.dump(receipt, out, indent=2, sort_keys=True)
        out.write('\n')
    print(f"{receipt['verdict']}: {len(receipt['findings'])} finding(s); receipt {args.output}")
    return 0 if receipt['verdict'] == 'agree' else 1


if __name__ == '__main__':
    sys.exit(main())