2e2c31d237
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
44 lines
2.5 KiB
Python
44 lines
2.5 KiB
Python
"""Native proof job: only boolean scope and projection observations leave the pod."""
|
|
import json,time,sys,urllib.parse
|
|
from pathlib import Path
|
|
sys.path.insert(0,"/worker")
|
|
from renew import http,BAO,PI,TARGET,ISSUER,success
|
|
result={"success":False};session=None
|
|
try:
|
|
jwt=Path("/var/run/keycape-factor/token").read_text().strip()
|
|
code,body=http(BAO+"/auth/kubernetes/login",{"role":"keycape-factor-renewer","jwt":jwt})
|
|
result["wrong_service_account_denied"]=code in (400,403)
|
|
code,body=http(BAO+"/auth/kubernetes/login",{"role":"keycape-factor-workload-kv-read","jwt":jwt})
|
|
if code!=200:raise ValueError()
|
|
session=body["auth"]["client_token"];headers={"X-Vault-Token":session}
|
|
code,_=http(BAO+"/"+ISSUER,headers=headers);result["issuer_password_denied"]=code==403
|
|
code,_=http(BAO+"/platform/data/workloads/net-kingdom/unrelated",headers=headers);result["sibling_secret_denied"]=code==403
|
|
code,data=http(BAO+"/"+TARGET,headers=headers)
|
|
if code!=200:raise ValueError()
|
|
initial=Path("/factor/admin-token").read_text().strip()
|
|
result["mounted_token_matches_custody"]=initial==data["data"]["data"]["TOKEN"]
|
|
listing=success(*http(PI+"/token/?tokenrealm=coulomb&active=True&pagesize=1",headers={"Authorization":initial}))
|
|
user=listing["tokens"][0]["username"]
|
|
query=urllib.parse.urlencode({"user":user,"realm":"coulomb","active":"True"})
|
|
listing=success(*http(PI+"/token/?"+query,headers={"Authorization":initial}))
|
|
result["keycape_user_lookup_accepted"]=listing["count"]>0 and any(x.get("active") is True for x in listing["tokens"])
|
|
code,_=http(PI+"/policy/",headers={"Authorization":initial});result["provider_administration_denied"]=code in (401,403)
|
|
print(json.dumps({"phase":"awaiting_rotation",**result}),flush=True)
|
|
for _ in range(48):
|
|
time.sleep(5)
|
|
current=Path("/factor/admin-token").read_text().strip()
|
|
if current!=initial:
|
|
listing=success(*http(PI+"/token/?"+query,headers={"Authorization":current}))
|
|
result["rotated_projection_accepted"]=listing["count"]>0
|
|
break
|
|
result["success"]=all(v for k,v in result.items() if k!="success") and result.get("rotated_projection_accepted",False)
|
|
except Exception:pass
|
|
finally:
|
|
if session:
|
|
try:
|
|
code,_=http(BAO+"/auth/token/revoke-self",headers={"X-Vault-Token":session},method="POST")
|
|
result["session_revoked"]=code in (200,204)
|
|
except Exception:result["session_revoked"]=False
|
|
if not result["session_revoked"]:result["success"]=False
|
|
print(json.dumps(result),flush=True)
|
|
raise SystemExit(0 if result["success"] else 1)
|