Implement S3 service assurance and admission checks

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

scripts/capture_service_observation.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Collect only selected status fields over the existing railiance01 SSH lane."""
+import argparse
+from datetime import datetime, timezone
+import json
+import re
+import os
+import tempfile
+from pathlib import Path
+import shlex
+import subprocess
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / 'scripts'))
+from service_assurance import read_contract, admission
+
+
+def query(args, allowed_codes=(0,)):
+    result = subprocess.run(['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10',
+        'railiance01', shlex.join(['kubectl', '--request-timeout=10s', *args])],
+        capture_output=True, text=True, timeout=25)
+    if result.returncode not in allowed_codes:
+        raise ValueError('query unavailable')
+    return json.loads(result.stdout)
+
+
+def memory_bytes(value):
+    match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)(Ki|Mi|Gi|Ti|K|M|G|T)?", value)
+    if not match:
+        raise ValueError('invalid memory quantity')
+    unit = match.group(2) or ''
+    multipliers = {'': 1, 'Ki': 1024, 'Mi': 1024**2, 'Gi': 1024**3,
+        'Ti': 1024**4, 'K': 1000, 'M': 1000**2, 'G': 1000**3, 'T': 1000**4}
+    return float(match.group(1)) * multipliers[unit]
+
+
+def capture():
+    contract = read_contract(ROOT / 'assurance/service-contract.json')
+    uid = query(['get', 'namespace', 'kube-system', '-o', 'go-template={{printf "%q" .metadata.uid}}'])
+    if uid != contract['cluster_uid']:
+        raise ValueError('wrong cluster')
+    signals = {}
+    def add(name, result, when=None):
+        signals[name] = {'result': result, 'observed_at': when or datetime.now(timezone.utc).isoformat()}
+    baseline = json.loads((ROOT / 'assurance/admission-baseline.json').read_text())
+    if admission() != baseline:
+        raise ValueError('admission baseline drift')
+    for cell in ('apps-pg', 'platform-pg', 'platform-pg-2'):
+        names = [cell + '.' + suffix for suffix in ('ready', 'backup', 'wal')]
+        try:
+            d = query(['get', 'cluster', cell, '-n', 'databases', '-o',
+                'go-template={"ready":{{.status.readyInstances}},"instances":{{.spec.instances}},"lastBackup":{{printf "%q" .status.lastSuccessfulBackup}},"conditions":[{{range $i,$v := .status.conditions}}{{if $i}},{{end}}{"type":{{printf "%q" $v.type}},"status":{{printf "%q" $v.status}}}{{end}}]}'])
+            add(names[0], 'pass' if d['ready'] == d['instances'] and d['ready'] > 0 else 'fail')
+            if d['lastBackup']:
+                add(names[1], 'pass', d['lastBackup'])
+            add(names[2], 'pass' if any(c == {'type': 'ContinuousArchiving', 'status': 'True'} for c in d['conditions']) else 'fail')
+        except (ValueError, KeyError, TypeError, subprocess.TimeoutExpired):
+            for name in names:
+                add(name, 'unavailable')
+        try:
+            metrics = query(['get', '--raw', '/apis/metrics.k8s.io/v1beta1/namespaces/databases/pods/' + cell + '-1'])
+            if not metrics['containers']:
+                raise ValueError('empty metrics')
+            memory = sum(memory_bytes(c['usage']['memory']) for c in metrics['containers'])
+            # Same read-only pg_stat_activity aggregate as the package ops surface,
+            # narrowed to a count: no SQL text, usernames, database rows or values.
+            count = query(['exec', '-n', 'databases', cell + '-1', '-c', 'postgres', '--',
+                'psql', '-U', 'postgres', '-d', 'postgres', '-Atqc',
+                'SELECT count(*) FROM pg_stat_activity'])
+            limits = baseline['cells'][cell]
+            add(cell + '.headroom', 'pass' if type(count) is int and count >= 0
+                and count <= limits['max_connections'] * 0.8
+                and memory <= memory_bytes(limits['memory_limit']) * 0.8 else 'fail',
+                metrics['timestamp'])
+        except (OSError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired):
+            add(cell + '.headroom', 'unavailable')
+    try:
+        status = query(['exec', '-n', 'openbao', 'openbao-0', '--',
+            'bao', 'status', '-format=json'], allowed_codes=(0, 2))
+        add('openbao.seal', 'pass' if status.get('sealed') is False else 'fail')
+    except (OSError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired):
+        add('openbao.seal', 'unavailable')
+    try:
+        rows = query(['get', 'externalsecrets', '-A', '-o',
+            'go-template=[{{range $i,$v := .items}}{{if $i}},{{end}}{"refresh":{{printf "%q" $v.status.refreshTime}},"conditions":[{{range $j,$c := $v.status.conditions}}{{if $j}},{{end}}{"type":{{printf "%q" $c.type}},"status":{{printf "%q" $c.status}}}{{end}}]}{{end}}]'])
+        ready = bool(rows) and all(any(c == {'type': 'Ready', 'status': 'True'} for c in row['conditions']) for row in rows)
+        # Empty/missing refresh is a failure, never an apparently healthy zero age.
+        refresh = min((r['refresh'] for r in rows), default='')
+        add('eso.ready', 'pass' if ready else 'fail')
+        if refresh:
+            add('eso.refresh', 'pass', refresh)
+    except (ValueError, KeyError, TypeError, subprocess.TimeoutExpired):
+        add('eso.ready', 'unavailable')
+        add('eso.refresh', 'unavailable')
+    # No token, Secret, application data/logs or seal/unseal mutation.
+    # Native restore and offsite receipts remain separate attended evidence.
+    return {'schema': 'railiance-platform.observation.v1', 'cluster_uid': uid,
+        'captured_at': datetime.now(timezone.utc).isoformat(), 'signals': signals}
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--output', type=Path)
+    args = parser.parse_args()
+    try:
+        data = json.dumps(capture(), indent=2) + '\n'
+        if args.output:
+            temporary = None
+            try:
+                with tempfile.NamedTemporaryFile(mode='w', dir=args.output.parent, delete=False) as f:
+                    temporary = f.name
+                    f.write(data)
+                    f.flush()
+                    os.fsync(f.fileno())
+                os.replace(temporary, args.output)
+                temporary = None
+            finally:
+                if temporary:
+                    os.unlink(temporary)
+            print(json.dumps({'captured': True}))
+        else:
+            print(data, end='')
+    except (OSError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired):
+        print(json.dumps({'schema': 'railiance-platform.assurance-error.v1', 'error': 'capture-unavailable-or-wrong-cluster'}))
+        sys.exit(2)

scripts/service_assurance.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env python3
+"""Read-only S3 admission and bounded evidence evaluation. Never fetch Secrets."""
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timezone
+import hashlib
+import importlib.util
+import json
+import math
+from pathlib import Path
+import sys
+
+import yaml
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def load_module(name, path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def timestamp(value):
+    if not isinstance(value, str):
+        raise ValueError('invalid timestamp')
+    result = datetime.fromisoformat(value.replace('Z', '+00:00'))
+    if result.tzinfo is None:
+        raise ValueError('timezone required')
+    return result.astimezone(timezone.utc)
+
+
+def digest(path):
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def admission(root=ROOT, package=None):
+    package = package or root.parent / 'rapp-postgres'
+    local = load_module('apps_capacity', root / 'tools/verify_apps_pg_capacity.py')
+    native = load_module('postgres_consumers', package / 'scripts/render_consumers.py')
+    manifests = [root / 'helm' / (name + '-cluster.yaml') for name in local.REVIEWED_CELLS]
+    local.verify(manifests)
+    paths = sorted((package / 'consumers').glob('*.yaml'))
+    if not paths:
+        raise ValueError('package consumers missing')
+    consumers = [native.load(path) for path in paths]
+    native.validate_cluster_capacity(consumers)
+    registry = json.loads((root / 'assurance/placement-owners.json').read_text())
+    cells = {}
+    source_files = manifests + paths + [root / 'tools/verify_apps_pg_capacity.py',
+        root / 'assurance/placement-owners.json', package / 'scripts/render_consumers.py']
+    for manifest in manifests:
+        doc = yaml.safe_load(manifest.read_text())
+        name, spec = doc['metadata']['name'], doc['spec']
+        roles = spec.get('managed', {}).get('roles', [])
+        names = sorted(r['name'] for r in roles)
+        cells[name] = {'consumers': names, 'ceiling': local.MAX_CONSUMERS,
+            'retention_days': int(spec['backup']['retentionPolicy'].removesuffix('d')),
+            'instances': spec['instances'], 'memory_limit': spec['resources']['limits']['memory'],
+            'max_connections': int(spec['postgresql']['parameters']['max_connections']),
+            'service_classes': {n: registry[n]['service_class'] for n in names},
+            'connection_limits': {r['name']: r['connectionLimit'] for r in roles}}
+        if cells[name]['retention_days'] != 30:
+            raise ValueError('apps-pg retention differs from disclosed 30-day contract')
+    for name in sorted(native.ALLOWED_CLUSTERS):
+        manifest = package / 'helm' / (name + '-cluster.yaml')
+        spec = yaml.safe_load(manifest.read_text())['spec']
+        source_files.append(manifest)
+        selected = [d for d in consumers if native.cluster_name(d) == name]
+        horizon = native.erasure_horizon(selected)
+        retention = int(spec['backup']['retentionPolicy'].removesuffix('d'))
+        if retention != horizon['instanceRetentionDays']:
+            raise ValueError('package retention differs from effective consumer horizon')
+        if any(not d['honoured'] for d in horizon['consumers'].values()):
+            raise ValueError('consumer retention request cannot be honoured on shared cell')
+        cells[name] = {'consumers': sorted(horizon['consumers']),
+            'ceiling': native.MAX_CONSUMERS_PER_CLUSTER, 'retention_days': retention,
+            'instances': spec['instances'], 'memory_limit': spec['resources']['limits']['memory'],
+            'max_connections': int(spec['postgresql']['parameters']['max_connections']),
+            'service_classes': horizon['serviceClasses']}
+    names = {n for c in cells.values() for n in c['consumers']}
+    if names != set(registry) or any(not r.get('owner') or not r.get('evidence') for r in registry.values()):
+        raise ValueError('missing or stale placement owner')
+    for name, cell in cells.items():
+        for consumer in cell['consumers']:
+            if registry[consumer]['cell'] != name:
+                raise ValueError('placement owner cell mismatch')
+        cell['owners'] = {n: registry[n]['owner'] for n in cell['consumers']}
+    return {'schema': 'railiance-platform.admission.v1', 'basis': 'source-declarations',
+        'cells': cells, 'sources': {str(p.relative_to(root.parent)): digest(p) for p in source_files}}
+
+
+def evaluate(contract, observation, now):
+    """Classify a value-safe observation, not certify its underlying truth."""
+    if set(observation) != {'schema', 'cluster_uid', 'captured_at', 'signals'}:
+        raise ValueError('invalid observation envelope')
+    if observation['schema'] != 'railiance-platform.observation.v1':
+        raise ValueError('unsupported observation version')
+    if observation['cluster_uid'] != contract['cluster_uid']:
+        raise ValueError('wrong cluster')
+    capture = timestamp(observation['captured_at'])
+    capture_age = (now - capture).total_seconds()
+    if capture_age < 0:
+        raise ValueError('future capture')
+    specs = contract['signals']
+    if not isinstance(observation['signals'], dict) or set(observation['signals']) - set(specs):
+        raise ValueError('unknown signals')
+    results = {}
+    for name, spec in specs.items():
+        sample = observation['signals'].get(name)
+        state = 'missing'
+        if sample is not None:
+            if not isinstance(sample, dict) or set(sample) != {'observed_at', 'result'}:
+                raise ValueError('invalid signal fields')
+            if sample['result'] not in ('pass', 'fail', 'unavailable'):
+                raise ValueError('invalid result')
+            observed = timestamp(sample['observed_at'])
+            age = (now - observed).total_seconds()
+            if observed > capture:
+                raise ValueError('signal newer than capture')
+            if sample['result'] == 'unavailable':
+                state = 'unavailable'
+            elif sample['result'] == 'fail':
+                state = 'failed'
+            elif capture_age > contract['capture_max_age_seconds']:
+                state = 'stale'
+            elif age > spec['max_age_seconds']:
+                state = 'stale'
+            else:
+                state = 'healthy'
+        results[name] = {'state': state, 'owner': spec['owner']}
+    return {'schema': 'railiance-platform.assurance-signal.v1',
+        'cluster_uid': contract['cluster_uid'], 'evaluated_at': now.isoformat(),
+        'signals': results, 'transport': 'unmonitored',
+        'guarantees': 'unsupported', 'threshold_status': 'local-diagnostic-only',
+        'healthy': all(r['state'] == 'healthy' for r in results.values())}
+
+
+def read_contract(path):
+    doc = json.loads(path.read_text())
+    if doc['schema'] != 'railiance-platform.assurance-contract.v1' or not doc['signals']:
+        raise ValueError('invalid contract')
+    for budget in [doc['capture_max_age_seconds'], *(s['max_age_seconds'] for s in doc['signals'].values())]:
+        if type(budget) not in (int, float) or not math.isfinite(budget) or budget <= 0:
+            raise ValueError('invalid freshness budget')
+    if any(not s.get('owner') for s in doc['signals'].values()):
+        raise ValueError('signal owner missing')
+    return doc
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    sub = parser.add_subparsers(dest='command', required=True)
+    a = sub.add_parser('admission')
+    a.add_argument('--package', type=Path, default=ROOT.parent / 'rapp-postgres')
+    a.add_argument('--check', type=Path)
+    e = sub.add_parser('evaluate')
+    e.add_argument('observation', type=Path)
+    e.add_argument('--contract', type=Path, default=ROOT / 'assurance/service-contract.json')
+    e.add_argument('--now', help='Explicit UTC time for deterministic replay')
+    args = parser.parse_args(argv)
+    try:
+        if args.command == 'admission':
+            report = admission(package=args.package)
+            if args.check and report != json.loads(args.check.read_text()):
+                raise ValueError('admission source or disclosure drift; review before refreshing')
+            code = 0
+        else:
+            now = timestamp(args.now) if args.now else datetime.now(timezone.utc)
+            report = evaluate(read_contract(args.contract), json.loads(args.observation.read_text()), now)
+            code = 0 if report['healthy'] else 1
+    except (OSError, ValueError, KeyError, TypeError, AttributeError, yaml.YAMLError):
+        # Do not print parser errors, raw input, values, source paths or subprocess stderr.
+        print(json.dumps({'schema': 'railiance-platform.assurance-error.v1', 'error': 'invalid-or-unavailable-input'}))
+        return 2
+    print(json.dumps(report, indent=2, allow_nan=False))
+    return code
+
+
+if __name__ == '__main__':
+    sys.exit(main())