e77f3989b4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
314 lines
15 KiB
Python
314 lines
15 KiB
Python
#!/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())
|