Implement audited lost-factor recovery and track remaining P04 acceptance

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c

scripts/keycape_factor_recovery.py

@@ -0,0 +1,36 @@
+"""Silent attended-owner front door. No provider administrative credential is exported."""
+import argparse,json,os,re,subprocess
+from pathlib import Path
+from keycape_factor_metadata import call
+CLUSTER="a553c742-0115-43d4-99a4-a5ca56fe0786"
+def owner_identity():
+ data=call("token","lookup","-format=json")["data"]
+ policies=set(data.get("policies",[]))|set(data.get("identity_policies",[]))
+ entity=data.get("entity_id","")
+ if "platform-admin" not in policies or not re.fullmatch(r"[0-9a-f-]{36}",entity):raise ValueError("attended_platform_identity_required")
+ return "openbao:"+entity
+
+def execute(args):
+ actor=owner_identity()
+ r=subprocess.run(["kubectl","get","ns","kube-system","-o","json"],capture_output=True,check=True,timeout=20)
+ if json.loads(r.stdout)["metadata"]["uid"]!=CLUSTER:raise ValueError("wrong_cluster")
+ source=(Path(__file__).resolve().parents[2]/"key-cape/scripts/factor_recovery.py").read_text()
+ request=dict(user=args.user,serial=args.serial,realm="coulomb",actor=actor,reference=args.reference,apply=args.apply,expected_version=args.expected_version,identity_verified=args.identity_verified)
+ # The source is code only; identity input travels through stdin, never a shell.
+ r=subprocess.run(["kubectl","-n","mfa","exec","-i","deployment/privacyidea","-c","privacyidea","--","python3","-c",source],input=json.dumps(request).encode(),capture_output=True,timeout=45)
+ result=json.loads(r.stdout)
+ allowed={"success","status","user","serial","realm","actor","reference","scope","operation","version","active","changes_applied","replayed","failure"}
+ if not isinstance(result,dict) or set(result)-allowed:raise ValueError("unexpected_provider_receipt")
+ return result
+
+def main():
+ p=argparse.ArgumentParser(description="Preview or perform one identity-verified factor recovery through attended OpenBao authentication")
+ p.add_argument("--user",required=True);p.add_argument("--serial",required=True);p.add_argument("--reference",required=True)
+ p.add_argument("--apply",action="store_true");p.add_argument("--identity-verified",action="store_true");p.add_argument("--expected-version")
+ p.add_argument("--receipt",type=Path,required=True);a=p.parse_args()
+ if any(os.environ.get(k) for k in ("BAO_TOKEN","VAULT_TOKEN","OPENBAO_TOKEN")):return 2
+ fd=os.open(a.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600);result={"success":False}
+ try:result=execute(a);return 0 if result.get("success") else 1
+ except Exception:result["failure"]="owner_operation_unavailable";return 1
+ finally:os.write(fd,json.dumps(result,indent=2).encode());os.close(fd)
+if __name__=="__main__":raise SystemExit(main())