tegwick
c59bbf568c
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
64 lines
2.9 KiB
Python
64 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline public selector preflight; no imports of credential/backend clients.
|
|
|
|
Checks current consumer source constants and the producer's already-deployed
|
|
public env/denial selector. This is source metadata, not runtime/owner assent.
|
|
"""
|
|
import argparse
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
LIMIT=128*1024
|
|
|
|
def read(path):
|
|
path=Path(os.path.abspath(path))
|
|
if any(p.is_symlink() for p in (path,*path.parents)):raise ValueError()
|
|
fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK)
|
|
with os.fdopen(fd,'rb') as source:
|
|
info=os.fstat(source.fileno())
|
|
if not stat.S_ISREG(info.st_mode) or not 0<info.st_size<=LIMIT:raise ValueError()
|
|
raw=source.read(LIMIT+1)
|
|
if len(raw)>LIMIT:raise ValueError()
|
|
return raw
|
|
|
|
def unique(pairs):
|
|
result={}
|
|
for key,value in pairs:
|
|
if key in result:raise ValueError()
|
|
result[key]=value
|
|
return result
|
|
|
|
def verify(selector,source,source_sha256):
|
|
if not re.fullmatch('sha256:[a-f0-9]{64}',source_sha256) or 'sha256:'+hashlib.sha256(source).hexdigest()!=source_sha256:raise ValueError()
|
|
needed={'CLIENT_ID','SUBJECT','TENANT','ROLE','SCOPE'};constants={}
|
|
for node in ast.parse(source).body:
|
|
if isinstance(node,ast.Assign):
|
|
for target in node.targets:
|
|
if isinstance(target,ast.Name) and target.id in needed:
|
|
if target.id in constants or not isinstance(node.value,ast.Constant) or not isinstance(node.value.value,str):raise ValueError()
|
|
constants[target.id]=node.value.value
|
|
if set(constants)!=needed:raise ValueError()
|
|
expected={'client_id':constants['CLIENT_ID'],'secret_env':'KEYCAPE_SECRETS_ENGINE_OPENBAO_CLIENT_SECRET',
|
|
'audience':constants['CLIENT_ID'],'scope':constants['SCOPE'],'subject':constants['SUBJECT'],
|
|
'tenant':constants['TENANT'],'roles':constants['ROLE'],'deny_scope':'openbao:admin'}
|
|
if selector!=expected:raise ValueError()
|
|
return {'schema':'secrets-engine.recovery-selector-preflight.v1','status':'source_selector_matches_not_native_assent',
|
|
'source_sha256':source_sha256,'consumer':expected,'credential_source_exercised':False,'network_contacted':False}
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser(description=__doc__);p.add_argument('--selector',required=True);p.add_argument('--source-sha256',required=True)
|
|
args=p.parse_args()
|
|
try:
|
|
proposal=json.loads(read(args.selector),object_pairs_hook=unique)
|
|
if not isinstance(proposal,dict) or not isinstance(proposal.get('consumer'),dict):raise ValueError()
|
|
result=verify(proposal['consumer'],read(ROOT/'src/secrets_engine/service_auth.py'),args.source_sha256)
|
|
except Exception:print('{"status":"source_selector_refused","private_diagnostics_suppressed":true}');return 1
|
|
print(json.dumps(result,indent=2));return 0
|
|
|
|
if __name__=='__main__':raise SystemExit(main())
|