#!/usr/bin/env python3 """Silent attended CCR-2026-0056 foundation, admission and projection. Default output is an offline proposal. Never changes the application sender. An incomplete attempt requires inspection, not an automatic retry or rollback. """ from __future__ import annotations import argparse import hashlib import http.client import json import os from pathlib import Path import socket import subprocess import sys import tempfile import time from datetime import datetime, timezone import email_connect_smtp_delivery as admission from state_hub_preflight_lane import command, data, bao, capabilities ROOT = admission.ROOT BUNDLE = ROOT / 'openbao/email-connect-platform-smtp/foundation.json' NS = 'email-connect' SA = admission.LANE['service_account'] KUBECONFIG = '/home/worsch/.kube/config-railiance01' require = admission.require def digest(value): return 'sha256:' + hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest() def prepare(): ccr, owner, owner_digest = admission.prepare() bundle = json.loads(BUNDLE.read_text()) proposal = dict(schema='platform.email-connect-smtp-foundation-proposal.v1', foundation=bundle, admission_digest=owner_digest, kubeconfig=KUBECONFIG, controller_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest()) return ccr, proposal, digest(proposal) class Native: def __init__(self): self.kube = ['kubectl', '--kubeconfig', KUBECONFIG] def get(self, kind, name, namespace=NS): args = self.kube + ['get', kind, name, '--ignore-not-found', '-o', 'json'] if namespace: args += ['-n', namespace] result = command(args) return data(result) if result.stdout.strip() else None def target_metadata(self, name): # Kubernetes content negotiation excludes Secret values at the server. # Use a private local socket, one fixed GET path and no full-object fallback. require(name == 'email-connect-platform-smtp', 'target_secret_scope_drift') path = '/api/v1/namespaces/email-connect/secrets/'+name with tempfile.TemporaryDirectory(prefix='email-connect-metadata-') as directory: sockpath = str(Path(directory)/'api.sock') proc = subprocess.Popen(self.kube + ['proxy', '--unix-socket='+sockpath, '--accept-paths=^'+path+'$', '--reject-methods=^(POST|PUT|PATCH|DELETE)$'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) conn = None try: for _ in range(50): if Path(sockpath).exists(): break require(proc.poll() is None, 'metadata_proxy_failed') time.sleep(0.1) sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(10) try: sock.connect(sockpath) except BaseException: sock.close() raise conn = http.client.HTTPConnection('localhost', timeout=10) conn.sock = sock conn.request('GET', path, headers={'Accept': 'application/json;as=PartialObjectMetadata;g=meta.k8s.io;v=v1'}) response = conn.getresponse() if response.status == 404: return None require(response.status == 200, 'metadata_query_failed') value = json.load(response) require(value.get('kind') == 'PartialObjectMetadata' and value.get('apiVersion') == 'meta.k8s.io/v1' and 'data' not in value, 'metadata_only_response_required') return value['metadata'] finally: if conn: conn.close() proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill(); proc.wait(timeout=5) def create(self, resource): return data(command(self.kube + ['create', '-f', '-', '-o', 'json'], payload=resource)) def jwt(self, service_account=SA, audience='openbao'): return command(self.kube + ['-n', NS, 'create', 'token', service_account, '--audience='+audience, '--duration=10m']).stdout.decode().strip() def token_review(self, jwt): return data(command(self.kube + ['create', '-f', '-', '-o', 'json'], payload={ 'apiVersion': 'authentication.k8s.io/v1', 'kind': 'TokenReview', 'spec': {'token': jwt, 'audiences': ['openbao']}}))['status'] def wait(self, resource): command(self.kube + ['-n', NS, 'wait', '--for=condition=Ready', resource['kind']+'/'+resource['metadata']['name'], '--timeout=45s']) def exact_resource(actual, wanted): require(actual is not None, 'resource_missing') require(actual['apiVersion'] == wanted['apiVersion'] and actual['kind'] == wanted['kind'], 'resource_drift') require(all(actual['metadata'].get(k) == v for k, v in wanted['metadata'].items()), 'resource_identity_drift') if wanted['kind'] == 'ServiceAccount': require(actual.get('automountServiceAccountToken') is False and not actual.get('secrets') and not actual.get('imagePullSecrets'), 'service_account_drift') else: require(actual['spec'] == wanted['spec'], 'resource_spec_drift') require(not actual['metadata'].get('deletionTimestamp'), 'resource_deleting') def protected_application(native, bundle): deployment = native.get('deployment', 'email-connect') require(deployment['metadata']['uid'] == bundle['deployment_uid'] and deployment['metadata']['generation'] == bundle['deployment_generation'], 'application_changed') values = [] for kind, name in [('deployment', 'email-connect'), ('configmap', 'email-connect-config'), ('externalsecret', 'email-connect-runtime')]: obj = deployment if kind == 'deployment' else native.get(kind, name) values.append({'uid': obj['metadata']['uid'], 'spec': obj.get('spec'), 'data': obj.get('data')}) return digest(values) def verify_login(native): def login(sa=SA, audience='openbao', negative=False): result = bao(['write', '-format=json', 'auth/kubernetes/login', '-'], payload={'role': SA, 'jwt': native.jwt(sa, audience)}, allow_failure=negative) if negative: if result.returncode == 0: revoke(data(result)['auth']['client_token']) raise admission.Refused('negative_login_succeeded') require(b'400' in result.stderr or b'403' in result.stderr, 'negative_login_inconclusive') return None return data(result)['auth'] def revoke(token): bao(['write', '-f', 'auth/token/revoke-self'], token=token) check = bao(['token', 'lookup', '-format=json'], token=token, allow_failure=True) require(check.returncode != 0 and b'403' in check.stderr, 'revocation_unconfirmed') auth = login() token = auth['client_token'] try: require(auth['token_policies'] == [SA] and auth.get('policies', [SA]) == [SA] and 0 < auth['lease_duration'] <= 900, 'effective_policy_or_ttl_drift') path = admission.LANE['kv_path'] paths = [path, path.replace('/data/', '/metadata/', 1), path+'-unrelated', 'auth/token/create', 'auth/token/lookup-self', 'auth/token/revoke-self'] expected = [['read'], ['deny'], ['deny'], ['deny'], ['read'], ['update']] require(capabilities(token, paths) == dict(zip(paths, expected)), 'capability_drift') mailbox = data(bao(['read', '-format=json', path], token=token))['data'] admission.lane_module().validate_entry(mailbox) finally: revoke(token) login('default', negative=True) login(audience='wrong-audience', negative=True) return dict(native_consumer_auth_verified=True, exact_read_verified=True, mutation_sibling_metadata_denied=True, bounded_ttl=True, other_service_account_denied=True, wrong_audience_denied=True, workload_token_revocation_verified=True) def run(proposal, receipt, checkpoint, native=None): native = native or Native() bundle = proposal['foundation'] status = data(bao(['status', '-format=json'])) require(status.get('cluster_id') == admission.OPENBAO_CLUSTER_ID and status.get('initialized') is True and status.get('sealed') is False, 'openbao_instance_mismatch') policies = data(bao(['token', 'lookup', '-format=json']))['data']['policies'] require('platform-admin' in policies and 'root' not in policies, 'attended_platform_admin_required') for name, uid in [('kube-system', bundle['cluster_uid']), (NS, bundle['namespace_uid'])]: obj = native.get('namespace', name, namespace=None) require(obj and obj['metadata']['uid'] == uid and not obj['metadata'].get('deletionTimestamp'), 'cluster_or_namespace_identity_mismatch') before = protected_application(native, bundle) resources = bundle['resources'] require([x['kind'] for x in resources] == ['ServiceAccount', 'SecretStore', 'ExternalSecret'], 'foundation_contract_drift') for x in resources: require(native.get(x['kind'], x['metadata']['name']) is None, 'target_exists_requires_inspection') target = resources[-1]['spec']['target']['name'] require(native.target_metadata(target) is None, 'target_secret_exists_requires_inspection') receipt.update(phase='creating_service_account', mutation_outcome='unconfirmed') checkpoint(receipt) account = native.create(resources[0]) exact_resource(account, resources[0]) review = native.token_review(native.jwt()) require(review.get('authenticated') is True and review.get('audiences') == ['openbao'] and review['user']['username'] == 'system:serviceaccount:'+NS+':'+SA and review['user']['uid'] == account['metadata']['uid'], 'native_binding_unproven') receipt.update(phase='binding_verified', service_account_uid=account['metadata']['uid'], native_binding_verified=True) checkpoint(receipt) validator = admission.module('email_connect_foundation_ccr', ROOT/'scripts/credential-change.py') validator.confirm_binding(admission.CCR, reviewer='attended-platform-operator', comment='CCR-2026-0056 foundation: pinned cluster and namespace, exact new service account and native audience TokenReview verified; receipt '+receipt['proposal_digest']) ccr, _, owner_digest = admission.prepare() admission.execution_guard(ccr, proposal['admission_digest'], owner_digest) owner_receipt = {} def owner_checkpoint(value): receipt.update(phase='owner_admission', owner_admission=dict(value)) checkpoint(receipt) admission.admit(owner_receipt, owner_checkpoint) receipt.update(phase='native_login_verification') checkpoint(receipt) receipt.update(verify_login(native)) for wanted in resources[1:]: receipt.update(phase='creating_'+wanted['kind'].lower()) checkpoint(receipt) created = native.create(wanted) exact_resource(created, wanted) native.wait(wanted) live = native.get(wanted['kind'], wanted['metadata']['name']) exact_resource(live, wanted) require(live['metadata']['uid'] == created['metadata']['uid'] and any(c.get('type') == 'Ready' and c.get('status') == 'True' for c in live.get('status', {}).get('conditions', [])), 'projection_not_ready') if wanted['kind'] == 'ExternalSecret': require(bool(live['status'].get('refreshTime')), 'projection_refresh_unproven') metadata = native.target_metadata(target) require(metadata and any(o.get('uid') == live['metadata']['uid'] and o.get('controller') is True for o in metadata.get('ownerReferences', [])), 'projection_ownership_unproven') receipt['projection_refresh_time'] = live['status']['refreshTime'] require(protected_application(native, bundle) == before, 'application_changed') receipt.update(status='foundation_admitted_projection_ready', phase='completed', mutation_outcome='verified', projection_ready=True, application_sender_changed=False) checkpoint(receipt) def execute(ccr, proposal, actual, expected, path, runner=run): require(actual == expected, 'reviewed_digest_required') require(ccr['status'] == 'approved', 'approved_ccr_required') info = path.parent.lstat() require(path.is_absolute() and path.parent.is_dir() and not path.parent.is_symlink() and info.st_uid == os.getuid() and not info.st_mode & 0o077, 'private_receipt_parent_required') fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) receipt = dict(schema='platform.email-connect-smtp-foundation.v1', status='started', phase='preflight', proposal_digest=actual, started_at=datetime.now(timezone.utc).isoformat(), mutation_outcome='not_started', email_sent=False, application_sender_changed=False) def checkpoint(value): raw = (json.dumps(value, sort_keys=True)+'\n').encode() while raw: n = os.write(fd, raw) if n <= 0: raise OSError('receipt_write_failed') raw = raw[n:] os.fsync(fd) try: checkpoint(receipt) parent = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(parent) finally: os.close(parent) runner(proposal, receipt, checkpoint) require(receipt['status'] == 'foundation_admitted_projection_ready', 'success_receipt_required') result = 0 except Exception: receipt.update(status='failed_requires_inspection', reason='contained_foundation_failure') result = 1 finally: key = 'interrupted_at' if receipt['status'] == 'started' else 'completed_at' receipt[key] = datetime.now(timezone.utc).isoformat() try: checkpoint(receipt) finally: os.close(fd) return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--execute', action='store_true') parser.add_argument('--expect-digest') parser.add_argument('--receipt', type=Path) args = parser.parse_args() ccr, proposal, value = prepare() if not args.execute: print(json.dumps(dict(proposal_digest=value, proposal=proposal, ccr_status=ccr['status']), indent=2)) return 0 require(args.receipt is not None, 'private_receipt_required') return execute(ccr, proposal, value, args.expect_digest, args.receipt) if __name__ == '__main__': if '--execute' in sys.argv: with open(os.devnull, 'w') as sink: os.dup2(sink.fileno(), 1); os.dup2(sink.fileno(), 2) try: code = main() except Exception: code = 1 sys.exit(code) sys.exit(main())