90 lines
9.1 KiB
Python
90 lines
9.1 KiB
Python
"""Disposable synthetic proof: actual Approval App HTTPS -> actual Access serve.
|
|
No native config, credential environment, approval or backend effect is used.
|
|
"""
|
|
import base64, hashlib, http.server, io, json, os, pathlib, socket, ssl, subprocess, sys, tempfile, threading, time
|
|
from datetime import datetime, timedelta, timezone
|
|
from wsgiref.simple_server import make_server, WSGIRequestHandler
|
|
from cryptography import x509
|
|
from cryptography.x509.oid import NameOID
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
sys.path.insert(0, '/home/worsch/approval-engine')
|
|
sys.path.insert(0, '/home/worsch/secrets-engine/src')
|
|
from approval_engine.api import App
|
|
from approval_engine.auth import Identity, StaticTokenAuthenticator
|
|
from approval_engine.store import Engine
|
|
from secrets_engine.authorization import request_digest, approval_binding_digest
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError
|
|
|
|
OUT=pathlib.Path(sys.argv[1]);OUT.mkdir(exist_ok=True)
|
|
def digest(raw):return 'sha256:'+hashlib.sha256(raw).hexdigest()
|
|
def write(path,raw,mode=0o600):path.write_bytes(raw);path.chmod(mode)
|
|
def identity(subject,kind,scopes):return Identity(subject=subject,issuer='synthetic-test-only',audiences=('approval-engine',),principal_type=kind,tenant='tenant:platform',roles=frozenset(),scopes=frozenset(scopes),assurance={'level':'synthetic'},evidence_ref='synthetic-test-only',tenant_source='directory')
|
|
class Quiet(WSGIRequestHandler):
|
|
def log_message(self,*args):pass
|
|
class Review(http.server.BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
raw=self.rfile.read(int(self.headers['Content-Length']));body=json.loads(raw)
|
|
assert self.path=='/apis/authentication.k8s.io/v1/tokenreviews'
|
|
assert self.headers['Authorization']=='Bearer synthetic-reviewer'
|
|
assert body['spec']['audiences']==['flex-auth','access-engine']
|
|
self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(json.dumps({'status':{'authenticated':True,'audiences':['flex-auth'],'user':{'username':'system:serviceaccount:secrets-engine:secrets-engine'}}}).encode())
|
|
def log_message(self,*args):pass
|
|
with tempfile.TemporaryDirectory(prefix='access-actual-python-source-') as folder:
|
|
root=pathlib.Path(folder);now=datetime.now(timezone.utc)
|
|
key=rsa.generate_private_key(public_exponent=65537,key_size=2048);name=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,'localhost')]);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(hours=1)).add_extension(x509.SubjectAlternativeName([x509.DNSName('localhost')]),critical=False).sign(key,hashes.SHA256()))
|
|
ca=root/'ca.pem';private=root/'tls-key.pem';write(ca,cert.public_bytes(serialization.Encoding.PEM));write(private,key.private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption()))
|
|
tls=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER);tls.load_cert_chain(ca,private)
|
|
ctx=ssl.create_default_context(cafile=str(ca))
|
|
def start(server):server.socket=tls.wrap_socket(server.socket,server_side=True);threading.Thread(target=server.serve_forever,daemon=True).start();return 'https://localhost:'+str(server.server_port)
|
|
review=http.server.HTTPServer(('127.0.0.1',0),Review);review_origin=start(review)
|
|
engine=Engine(root/'approval.sqlite3',freshness_ttl=60)
|
|
auth=StaticTokenAuthenticator({'synthetic-creator':identity('secrets-engine','service',{'approval:create'}),'synthetic-human':identity('synthetic-human','human',{'approval:approve'}),'synthetic-reader':identity('synthetic-access-reader','service',{'approval:read'})})
|
|
app=App(engine,auth)
|
|
source=make_server('127.0.0.1',0,app,handler_class=Quiet);origin=start(source)
|
|
def source_call(method,path,body,token):
|
|
request=Request(origin+path,data=json.dumps(body).encode() if body is not None else None,method=method,headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'})
|
|
with urlopen(request,context=ctx,timeout=3) as response:return response.status,json.load(response)
|
|
# Reuse existing owner interoperability contract generator; it has synthetic metadata only.
|
|
generated=subprocess.check_output([sys.executable,'/tmp/access-trusted-b2-b3/tools/service_task_execution_fixture.py'],env={'PATH':os.defpath})
|
|
request=json.loads(generated)['request'];request['subject']['tenant']='tenant:platform';request['resource']['tenant']='tenant:platform';request['context']['purpose']='actual Python App source-only synthetic proof';del request['context']['approval']
|
|
native={'action':request['action'],'actor':'secrets-engine','principal':'secrets-engine','purpose':request['context']['purpose'],'target':request['resource']}
|
|
aid='aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee'
|
|
status,created=source_call('POST','/v1/approvals',{'id':aid,'binding':native,'validity':{'not_before':(now-timedelta(minutes=1)).isoformat(),'expires_at':(now+timedelta(minutes=3)).isoformat()},'required_count':1,'pdp_digest':approval_binding_digest(request),'pdp_path':True,'human_control':True},'synthetic-creator');assert status==201
|
|
status,_=source_call('POST',f'/v1/approvals/{aid}/entries',{},'synthetic-human');assert status==200
|
|
status,claim=source_call('GET',f'/v1/approvals/{aid}/claim',None,'synthetic-reader');assert status==200 and claim['valid_now'] is True and claim['binding']['human_control'] is True
|
|
request['context']['approval']=claim
|
|
seed=root/'seed';write(seed,base64.urlsafe_b64encode(bytes([19])*32).rstrip(b'='),0o440)
|
|
reviewer_token=root/'review-token';write(reviewer_token,b'synthetic-reviewer',0o440)
|
|
reader_token=root/'reader-token';write(reader_token,b'synthetic-reader',0o440)
|
|
registry=root/'registry.json';write(registry,b'{}')
|
|
config=root/'source.json';write(config,json.dumps({'source_id':'synthetic-actual-approval-app','source_revision':'actual-owner-python-source-fixture','origin':origin,'store_tenant':'tenant:platform','ca_file':str(ca),'token_file':str(reader_token)}).encode())
|
|
listener=socket.socket();listener.bind(('127.0.0.1',0));port=listener.getsockname()[1];listener.close()
|
|
record=root/'record.jsonl';url=f'http://127.0.0.1:{port}'
|
|
command=['/tmp/netkingdom-access-trusted-r1','serve','--addr',f'127.0.0.1:{port}','--registry',str(registry),'--policy','/tmp/access-trusted-b2-b3/examples/secrets-engine-service-task-execution-v2/policy_package.md','--log',str(record),'--caller-auth-mode','enforce','--caller-audience','flex-auth','--caller-kubernetes-url',review_origin,'--caller-reviewer-token-file',str(reviewer_token),'--caller-ca-file',str(ca),'--caller-binding','secrets-engine=system:serviceaccount:secrets-engine:secrets-engine','--envelope-signing-key-file',str(seed),'--envelope-signing-kid','synthetic-v2','--approval-source-file',str(config)]
|
|
child=subprocess.Popen(command,env={},stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
|
|
try:
|
|
for _ in range(100):
|
|
try:
|
|
with urlopen(url+'/healthz',timeout=.2):break
|
|
except Exception:time.sleep(.02)
|
|
jwt='synthetic.'+base64.urlsafe_b64encode(json.dumps({'exp':int((now+timedelta(minutes=2)).timestamp())}).encode()).decode().rstrip('=')+'.synthetic'
|
|
def check(value):
|
|
req=Request(url+'/v1/check',data=json.dumps(value).encode(),headers={'Authorization':'Bearer '+jwt,'Content-Type':'application/json'})
|
|
try:
|
|
with urlopen(req,timeout=5) as response:return response.status,response.read()
|
|
except HTTPError as error:return error.code,error.read()
|
|
status,raw=check(request);assert status==200,(status,raw)
|
|
response=json.loads(raw);assert response['effect']=='allow' and response['signature']['mode']=='signed'
|
|
assert response['binding']['context']==request['context'];assert response['binding']['submitted_request_digest']==request_digest(request);assert response['binding']['approval_binding_digest']==approval_binding_digest(request)
|
|
assert response['provenance']['approval_observation']['source_id']=='synthetic-actual-approval-app'
|
|
saved=record.read_bytes();assert json.loads(saved)==response
|
|
# Actual producer state transition is fetched anew and cannot be rescued by carried claim.
|
|
engine.revoke(aid)
|
|
refused,_=check(request);assert refused!=200 and record.read_bytes()==saved
|
|
for name,value in [('request.json',request),('response.json',response),('source-record.json',json.loads(saved)),('original-approval-claim.json',claim)]:write(OUT/name,(json.dumps(value,indent=2)+'\n').encode(),0o644)
|
|
receipt={'scope':'disposable synthetic only','actual_source':'Approval App over TLS with authenticated create, configured synthetic human entry, read; actual SQLite engine','actual_access':'production serve binary from frozen B2/B3 R1','allow':True,'original_context_unchanged':True,'published_digest_match':True,'returned_source_record_match':True,'actual_source_revoke_refused':True,'no_record_after_refusal':True,'native_admission':False,'files_sha256':{p.name:digest(p.read_bytes()) for p in OUT.iterdir() if p.suffix=='.json'}}
|
|
write(OUT/'receipt.json',(json.dumps(receipt,indent=2)+'\n').encode(),0o644);print(json.dumps(receipt,indent=2))
|
|
finally:
|
|
child.terminate();child.wait(timeout=5);source.shutdown();review.shutdown();engine.close()
|