tegwick
b854e1ed0c
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a09cbb-87c6-7900-a145-4ce53ba9f1a6
63 lines
4.9 KiB
Python
63 lines
4.9 KiB
Python
"""Add the admitted public browser client without exposing or replacing other config."""
|
|
import argparse,copy,importlib.util,json,secrets,hashlib,base64
|
|
from pathlib import Path
|
|
from datetime import datetime,timezone
|
|
from urllib.request import build_opener,ProxyHandler,HTTPRedirectHandler
|
|
from urllib.parse import urlencode,urlparse
|
|
from urllib.error import HTTPError
|
|
import yaml
|
|
spec=importlib.util.spec_from_file_location('rollout','/home/worsch/net-kingdom/sso-mfa/k8s/keycape/approval-clients-rollout.py');r=importlib.util.module_from_spec(spec);spec.loader.exec_module(r)
|
|
CLIENT='informed-decision-approver';r.IDS=(CLIENT,)
|
|
KUBE=['kubectl','--kubeconfig','/home/worsch/.kube/config-railiance01']
|
|
EXPECTED={'clientId':CLIENT,'displayName':'informed-decision approver surface','audience':'approval-engine','redirectUris':['https://decisions.coulomb.social/auth/callback'],'allowedScopes':['openid','approval:read','approval:approve'],'grantTypes':['authorization_code'],'clientType':'public','tenant':'tenant:platform','mfaRequired':True}
|
|
class NoRedirect(HTTPRedirectHandler):
|
|
def redirect_request(self,*args,**kwargs):return None
|
|
|
|
def probe(**overrides):
|
|
query={'response_type':'code','client_id':CLIENT,'redirect_uri':EXPECTED['redirectUris'][0],'scope':' '.join(EXPECTED['allowedScopes']),'state':secrets.token_urlsafe(32),'nonce':secrets.token_urlsafe(32),'code_challenge':base64.urlsafe_b64encode(hashlib.sha256(secrets.token_bytes(48)).digest()).rstrip(b'=').decode(),'code_challenge_method':'S256'};query.update(overrides)
|
|
try:
|
|
response=build_opener(ProxyHandler({}),NoRedirect()).open('https://kc.coulomb.social/authorize?'+urlencode(query),timeout=20)
|
|
except HTTPError as e:response=e
|
|
with response:
|
|
result={'status':response.status}
|
|
if response.status==302:result['redirect_host']=urlparse(response.headers['Location']).hostname
|
|
elif response.headers.get('Content-Type','').startswith('application/json'):
|
|
d=json.load(response);result.update({k:d.get(k) for k in ('error','description','feature')})
|
|
return result
|
|
|
|
def main(args,receipt):
|
|
r.assert_cluster(KUBE)
|
|
candidates=[c for c in yaml.safe_load(Path('/home/worsch/key-cape/config/service-clients.example.yaml').read_text())['clients'] if c['clientId']==CLIENT]
|
|
r.require(candidates==[EXPECTED],'public_registration_contract_drift')
|
|
secret=r.get(KUBE,'secret','keycape-config');deployment=r.get(KUBE,'deployment','keycape')
|
|
raw,config,_=r.pin.issuer_document(secret)
|
|
existing=[c for c in config['clients'] if c['clientId']==CLIENT]
|
|
receipt.update(client_id=CLIENT,existing_registration=bool(existing),before=probe())
|
|
if existing:
|
|
r.require(existing==[EXPECTED],'existing_registration_mismatch');receipt['status']='already_registered';return
|
|
config_value=r.replacement(secret,[EXPECTED]);candidate=copy.deepcopy(deployment['spec'])
|
|
image=next(c['image'] for c in candidate['template']['spec']['containers'] if c['name']=='keycape')
|
|
r.require('@sha256:' in image and candidate['strategy']['type']=='Recreate' and candidate.get('replicas',1)==1,'deployment_pin_required')
|
|
candidate['template'].setdefault('metadata',{}).setdefault('annotations',{})['key-cape.coulomb.social/informed-decision-registration']=datetime.now(timezone.utc).isoformat()
|
|
r.patch_object(KUBE,'secret',secret,'/data/config.yaml',config_value,dry=True)
|
|
r.patch_object(KUBE,'deployment',deployment,'/spec',candidate,dry=True)
|
|
receipt['status']='preflight_passed'
|
|
if not args.apply:return
|
|
r.patch_object(KUBE,'secret',secret,'/data/config.yaml',config_value)
|
|
receipt['phase']='config_applied'
|
|
after=r.get(KUBE,'secret','keycape-config');r.require(after['data']==dict(secret['data'],**{'config.yaml':config_value}),'unrelated_config_changed')
|
|
r.patch_object(KUBE,'deployment',deployment,'/spec',candidate)
|
|
receipt['deployment']=r.ready(KUBE,image)
|
|
receipt.update(image_unchanged=image,unrelated_config_preserved=True)
|
|
checks={'valid_pkce':probe(),'wrong_redirect':probe(redirect_uri='https://example.invalid/callback'),'consume_scope':probe(scope='openid approval:consume'),'missing_pkce':probe(code_challenge='',code_challenge_method='')}
|
|
receipt['checks']=checks
|
|
r.require(checks['valid_pkce']['status']==302,'authorization_did_not_reach_login')
|
|
r.require(all(checks[k]['status'] in (400,403) for k in ('wrong_redirect','consume_scope','missing_pkce')),'authorization_negative_failed')
|
|
receipt['status']='applied_verified'
|
|
|
|
if __name__=='__main__':
|
|
p=argparse.ArgumentParser();p.add_argument('--apply',action='store_true');p.add_argument('--receipt',type=Path,required=True);args=p.parse_args()
|
|
receipt={'observed_at':datetime.now(timezone.utc).isoformat(),'status':'failed','scope':'public browser registration only; no tokens or approval entries issued by this tool'}
|
|
try:main(args,receipt)
|
|
except Exception:print('Registration operation failed; inspect the metadata-only receipt.');raise SystemExit(1) from None
|
|
finally:args.receipt.write_text(json.dumps(receipt,indent=2)+'\n');print(json.dumps(receipt))
|