"""Real Approval/Access servers; every admission/key/identity is synthetic local. The positive fixture policy is separately named and exact-descriptor scoped. It never edits either existing unadmitted production task policy or their false gate. """ import base64 from contextlib import contextmanager from dataclasses import replace from datetime import datetime,timedelta,timezone import hashlib from http.client import HTTPConnection from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer import json import os from pathlib import Path import shutil import socket import subprocess import threading import time from types import SimpleNamespace from wsgiref.simple_server import make_server,WSGIRequestHandler from cryptography import x509 from cryptography.hazmat.primitives import serialization,hashes from cryptography.hazmat.primitives.asymmetric import ed25519,rsa from cryptography.x509.oid import NameOID ACCESS=Path(os.environ.get('ACCESS_ENGINE_SOURCE_ROOT','/home/worsch/access-engine')) APPROVAL=Path(os.environ.get('APPROVAL_ENGINE_SOURCE_ROOT','/home/worsch/approval-engine')) SECRETS=Path(os.environ.get('SECRETS_ENGINE_SOURCE_ROOT','/home/worsch/secrets-engine')) FIXTURE_PACKAGE='platform.disposable-task-seam.fixture' FIXTURE_VERSION='v1' PRINCIPAL='system:serviceaccount:secrets-engine:secrets-engine' def private(path,text):path.write_text(text);path.chmod(0o600);return path def free_port(): with socket.socket() as sock:sock.bind(('127.0.0.1',0));return sock.getsockname()[1] def http(origin,method,path,body=None,token=None): from urllib.parse import urlsplit parsed=urlsplit(origin);conn=HTTPConnection(parsed.hostname,parsed.port,timeout=4) try: headers={'Content-Type':'application/json'} if token is not None:headers['Authorization']='Bearer '+token conn.request(method,path,body=json.dumps(body) if body is not None else None,headers=headers) response=conn.getresponse();raw=response.read(1024*1024) return response.status,json.loads(raw) if raw else {} finally:conn.close() class QuietWSGI(WSGIRequestHandler): def log_message(self,*args):pass @contextmanager def approval_server(directory): from approval_engine.api import App from approval_engine.auth import Identity,StaticTokenAuthenticator from approval_engine.store import Engine service='synthetic-source-service';human='synthetic-source-human';no_read='synthetic-source-no-read' def identity(subject,kind,scopes): return Identity(subject,'https://synthetic.invalid',('approval-engine',),kind,'tenant:platform', frozenset(),frozenset(scopes),{'level':'aal2','methods':['synthetic']}, 'fixture:synthetic-identity',tenant_source='directory' if kind=='human' else 'registration') identities={service:identity('secrets-engine','service',['approval:create','approval:read','approval:consume']), human:identity('synthetic-founder','human',['approval:approve','approval:read']), no_read:identity('platform-fixture-confirmation','service',['approval:consume'])} engine=Engine(str(directory/'approval.sqlite'));app=App(engine,StaticTokenAuthenticator(identities)) calls=[] def logged(environ,start_response): calls.append((environ['REQUEST_METHOD'],environ['PATH_INFO'])) return app(environ,start_response) server=make_server('127.0.0.1',0,logged,handler_class=QuietWSGI) thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() try: yield SimpleNamespace(origin='http://127.0.0.1:'+str(server.server_port),engine=engine, service_token=service,human_token=human,no_read_token=no_read,calls=calls) finally:server.shutdown();server.server_close();thread.join(5);engine.close() @contextmanager def reviewer_server(directory): # Actual Access TokenReview client consumes this independently admitted # synthetic reviewer. This is not a Kubernetes authentication proof. exp=int(time.time())+900 payload=base64.urlsafe_b64encode(json.dumps({'exp':exp}).encode()).decode().rstrip('=') token='synthetic.'+payload+'.synthetic';reviewer='synthetic-reviewer';calls=[] class Handler(BaseHTTPRequestHandler): def log_message(self,*args):pass def do_POST(self): size=int(self.headers.get('Content-Length','0'));data=json.loads(self.rfile.read(size)) valid=(self.path=='/apis/authentication.k8s.io/v1/tokenreviews' and self.headers.get('Authorization')=='Bearer '+reviewer and data.get('spec',{}).get('token')==token and 'flex-auth' in data.get('spec',{}).get('audiences',[])) calls.append(valid) result={'apiVersion':'authentication.k8s.io/v1','kind':'TokenReview', 'status':{'authenticated':valid,'audiences':['flex-auth'], 'user':{'username':PRINCIPAL}}} raw=json.dumps(result).encode();self.send_response(201) self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(raw)));self.end_headers();self.wfile.write(raw) server=ThreadingHTTPServer(('127.0.0.1',0),Handler);thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() # The actual reviewer constructor requires a valid CA even for its explicitly # synthetic HTTP endpoint; this CA never trusts an estate endpoint. key=rsa.generate_private_key(public_exponent=65537,key_size=2048) name=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,'synthetic-local-reviewer')]) now=datetime.now(timezone.utc) cert=x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key()).serial_number(x509.random_serial_number()).not_valid_before(now-timedelta(minutes=1)).not_valid_after(now+timedelta(days=1)).sign(key,hashes.SHA256()) ca=directory/'fixture-ca.pem';ca.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) reviewer_file=private(directory/'reviewer.token',reviewer) try:yield SimpleNamespace(origin='http://127.0.0.1:'+str(server.server_port),caller_token=token,reviewer_file=reviewer_file,ca=ca,calls=calls) finally:server.shutdown();server.server_close();thread.join(5) def fixture_policy(directory,descriptor,request): policy=directory/'fixture-policy.md' fixture=directory/'fixture-cases.json' fixture.write_text(json.dumps([{'id':'synthetic-claim-free-deny','request':request, 'expect':{'effect':'deny','reason':'fixture_unapproved'}}])) document='''--- id: platform.disposable-task-seam.fixture name: Disposable exact task seam fixture namespace: secrets-engine:secret-catalog-lane version: v1 status: proposed package: fixture.platform.task_seam allow_ttl: 5m actions: [execute-ccr] owner: team:synthetic-fixture fixtures: [fixture-cases.json] caring: {profile: caring-0.4.0-rc2, enforce: false} activation: {mode: local} --- Synthetic local policy only. No production task policy is edited or enabled. ```rego default decision := {"effect":"deny","reason":"fixture_unapproved"} decision := {"effect":"allow","reason":"synthetic_exact_task_approved"} { input.action == "execute-ccr" input.tenant == "tenant:platform" input.subject.id == "secrets-engine" input.subject.type == "service" input.resource.system == "secrets-engine" input.context.human_control == true input.context.approval.valid_now == true input.context.approval.consumed == false input.context.approval.binding.human_control == true input.context.execution_contract == DESCRIPTOR } ``` ''' + ''' ```rego test import data.fixture.platform.task_seam test_missing_approval_denies { task_seam.decision.effect == "deny" with input as {} } ``` ''' document=document.replace('DESCRIPTOR',json.dumps(descriptor,separators=(',',':'))) policy.write_text(document) return policy @contextmanager def access_server(directory,descriptor,request,reviewer,*,actual_policy=False): binary=Path(os.environ.get('PLATFORM_ACCESS_FIXTURE_BINARY','/tmp/netkingdom-platform-wave4-access')) verifier=Path(os.environ.get('PLATFORM_PDP_FIXTURE_VERIFIER','/tmp/netkingdom-platform-wave4-verify')) if not binary.is_file() or not verifier.is_file():raise RuntimeError('build source fixture binaries first') registry=directory/'registry.json';registry.write_bytes((ACCESS/'examples/secrets-engine/registry_snapshot.json').read_bytes()) policy=ACCESS/'examples/secrets-engine-service-task-execution/policy_package.md' if actual_policy else fixture_policy(directory,descriptor,request) requestfile=directory/'claim-free-request.json';requestfile.write_text(json.dumps(request)) # Trusted owner-local source CLI derives the real package pin independently # from the network responder. Its DENY is used only as digest provenance. offline=subprocess.run([str(binary),'check','--registry',str(registry),'--policy',str(policy),'--request',str(requestfile)],capture_output=True,text=True,timeout=15) if offline.returncode!=0:raise RuntimeError('source policy fixture invalid: '+offline.stderr[:300]) produced=json.loads(offline.stdout);provenance=produced['provenance'] key=ed25519.Ed25519PrivateKey.generate() seed=key.private_bytes(serialization.Encoding.Raw,serialization.PrivateFormat.Raw,serialization.NoEncryption()) public=key.public_key().public_bytes(serialization.Encoding.Raw,serialization.PublicFormat.Raw) b64=lambda value:base64.urlsafe_b64encode(value).decode().rstrip('=') seedfile=private(directory/'signing.seed',b64(seed)) trust={name:provenance[name] for name in ['evaluator','mode','policy_package','policy_version','policy_package_digest']} trust.update(schema='secrets-engine.pdp-trust.v1',keys={'synthetic-owner-key':b64(public)}) trustfile=directory/'pdp-trust.json';trustfile.write_text(json.dumps(trust)) port=free_port();origin='http://127.0.0.1:'+str(port) args=[str(binary),'serve','--addr','127.0.0.1:'+str(port),'--registry',str(registry),'--policy',str(policy), '--envelope-signing-key-file',str(seedfile),'--envelope-signing-kid','synthetic-owner-key', '--caller-auth-mode','enforce','--caller-audience','flex-auth','--caller-kubernetes-url',reviewer.origin, '--caller-reviewer-token-file',str(reviewer.reviewer_file),'--caller-ca-file',str(reviewer.ca), '--caller-binding','secrets-engine='+PRINCIPAL] process=subprocess.Popen(args,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) for _ in range(100): if process.poll() is not None:raise RuntimeError('source Access server exited') try: with socket.create_connection(('127.0.0.1',port),timeout=.1):break except OSError:time.sleep(.05) else:process.kill();process.wait();raise RuntimeError('source Access server failed to start') try:yield SimpleNamespace(origin=origin,trustfile=trustfile,verifier=verifier, package=provenance['policy_package'],version=provenance['policy_version'], process=process,offline_envelope=produced) finally: process.terminate() try:process.wait(timeout=5) except subprocess.TimeoutExpired:process.kill();process.wait()