2e2c31d237
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
55 lines
3.8 KiB
Python
55 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Silent attended custody and dedicated provider bootstrap; no personal passwords."""
|
|
import argparse,json,os,secrets,subprocess,time,re
|
|
from pathlib import Path
|
|
from keycape_factor_metadata import call
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
ISSUER="platform/data/workloads/net-kingdom/keycape-factor-issuer"
|
|
TOKEN="platform/data/workloads/net-kingdom/keycape-factor-read"
|
|
REQUEST="RPF-WP-0040"
|
|
USER="keycape-factor-reader"
|
|
|
|
def read_optional(path):
|
|
p=subprocess.run(["bao","read","-format=json",path],capture_output=True,text=True,timeout=25)
|
|
if p.returncode:
|
|
# Only explicit 404 permits creation. Permission or transport failures never mean absent.
|
|
if "Code: 404" in p.stderr or "No value found" in p.stderr:return None
|
|
raise RuntimeError("custody read failed")
|
|
return json.loads(p.stdout)["data"]
|
|
|
|
def execute():
|
|
cluster=json.loads(subprocess.check_output(["kubectl","get","ns","kube-system","-o","json"]))
|
|
if cluster["metadata"]["uid"]!="a553c742-0115-43d4-99a4-a5ca56fe0786":raise RuntimeError("wrong cluster")
|
|
issuer=read_optional(ISSUER)
|
|
if issuer is None:
|
|
secret={"USERNAME":USER,"PASSWORD":secrets.token_urlsafe(48),"REQUEST":REQUEST}
|
|
call("write","-format=json",ISSUER,"-",payload={"options":{"cas":0},"data":secret})
|
|
issuer=read_optional(ISSUER)
|
|
secret=issuer["data"]
|
|
if set(secret)!={"USERNAME","PASSWORD","REQUEST"} or secret["USERNAME"]!=USER or secret["REQUEST"]!=REQUEST:raise RuntimeError("issuer custody mismatch")
|
|
source=(ROOT/"scripts/keycape_factor_provider.py").read_text()
|
|
p=subprocess.run(["kubectl","-n","mfa","exec","-i","deployment/privacyidea","-c","privacyidea","--","python3","-c",source],input=json.dumps({"operation":"bootstrap","username":USER,"password":secret["PASSWORD"]}),text=True,capture_output=True,timeout=60)
|
|
if p.returncode:
|
|
failure=json.loads(p.stdout)
|
|
permitted={"admin policy baseline changed","existing managed policy differs","service name already belongs to another setup","service authentication failed","unsupported service token lifetime","factor lookup failed","service has unexpected administration rights","provider response too large"}
|
|
reason=failure.get("failure")
|
|
raise RuntimeError(reason if reason in permitted or (isinstance(reason,str) and re.fullmatch(r"factor lookup failed [0-9]{3} (True|False) (True|False) (True|False)",reason)) else "provider setup failed without secret-bearing detail")
|
|
result=json.loads(p.stdout)
|
|
if not result.get("policy_read_denied") or result["expires_at"]<=time.time()+300:raise RuntimeError("provider verification incomplete")
|
|
current=read_optional(TOKEN)
|
|
if current and current["data"].get("REQUEST")!=REQUEST:raise RuntimeError("token custody belongs to another request")
|
|
cas=current["metadata"]["version"] if current else 0
|
|
call("write","-format=json",TOKEN,"-",payload={"options":{"cas":cas},"data":{"TOKEN":result["token"],"EXPIRES_AT":result["expires_at"],"REQUEST":REQUEST}})
|
|
stored=read_optional(TOKEN)
|
|
if stored["data"]["TOKEN"]!=result["token"]:raise RuntimeError("token custody readback failed")
|
|
return {"issuer_path":ISSUER,"token_path":TOKEN,"token_version":stored["metadata"]["version"],"expires_at":result["expires_at"],"provider_policy_read_denied":True,"cross_user_factor_visible":result["cross_user_factor_visible"],"values_emitted":False}
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser();p.add_argument("--receipt",type=Path,required=True);args=p.parse_args()
|
|
if any(os.environ.get(k) for k in ["OPENBAO_TOKEN","BAO_TOKEN","VAULT_TOKEN"]):return 2
|
|
fd=os.open(args.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600);result={"success":False}
|
|
try:result.update(execute());result["success"]=True;return 0
|
|
except RuntimeError as e:result["failure"]=str(e);return 1
|
|
except Exception as e:result["failure_type"]=type(e).__name__;return 1
|
|
finally:os.write(fd,json.dumps(result,indent=2).encode());os.close(fd)
|
|
if __name__=="__main__":raise SystemExit(main())
|