2e2c31d237
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
57 lines
4.7 KiB
Python
57 lines
4.7 KiB
Python
"""Guarded activation; credentials remain in memory and subprocess stdin."""
|
|
import argparse,base64,copy,json,re,subprocess,time
|
|
CLUSTER="a553c742-0115-43d4-99a4-a5ca56fe0786"
|
|
DEPLOYMENT="99ddd83c-cb3f-4847-bcf8-35f1aa87627f"
|
|
IMAGE="forgejo.coulomb.social/coulomb/key-cape@sha256:c9eb584d60efecfe00e1745a7e8cd3ebb4ae0f94faa715ef9594cd58a5dcebb6"
|
|
MOUNT={"name":"factor-token","mountPath":"/etc/keycape-factor","readOnly":True}
|
|
VOLUME={"name":"factor-token","secret":{"secretName":"keycape-factor-read","defaultMode":288,"items":[{"key":"admin-token","path":"admin-token"}]}}
|
|
def kubectl(*args,payload=None):
|
|
r=subprocess.run(["kubectl",*args],input=json.dumps(payload).encode() if payload is not None else None,stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=40)
|
|
if r.returncode:raise RuntimeError("Kubernetes operation failed")
|
|
return json.loads(r.stdout) if r.stdout else {}
|
|
def rewrite_config(raw):
|
|
text=raw.decode();match=re.search(r"(?m)^privacyidea:[^\n]*\n(?:(?:[ \t]+[^\n]*|)[\n]|[ \t]+[^\n]*$)*",text)
|
|
if not match:raise ValueError("privacyidea section missing")
|
|
section=match.group()
|
|
if re.search(r"(?m)^ adminTokenFile:",section):
|
|
if not re.search(r"(?m)^ adminTokenFile: /etc/keycape-factor/admin-token$",section) or re.search(r"(?m)^ adminToken:",section):raise ValueError("unexpected credential configuration")
|
|
return raw
|
|
section,n=re.subn(r"(?m)^ adminToken:[^\n]*$"," adminTokenFile: /etc/keycape-factor/admin-token",section)
|
|
if n!=1:raise ValueError("expected exactly one inline token")
|
|
return (text[:match.start()]+section+text[match.end():]).encode()
|
|
def deployment_patch(obj):
|
|
if obj["metadata"]["uid"]!=DEPLOYMENT:raise ValueError("deployment identity mismatch")
|
|
spec=obj["spec"]["template"]["spec"];containers=spec["containers"]
|
|
i=next(i for i,c in enumerate(containers) if c["name"]=="keycape")
|
|
if containers[i]["image"]!=IMAGE:raise ValueError("deployment image mismatch")
|
|
mounts=copy.deepcopy(containers[i].get("volumeMounts",[]));volumes=copy.deepcopy(spec.get("volumes",[]))
|
|
for entries,expected in ((mounts,MOUNT),(volumes,VOLUME)):
|
|
existing=[x for x in entries if x["name"]==expected["name"]]
|
|
if existing and existing!=[expected]:raise ValueError("existing factor mount differs")
|
|
if not existing:entries.append(expected)
|
|
return [{"op":"test","path":"/metadata/uid","value":DEPLOYMENT},{"op":"test","path":"/metadata/resourceVersion","value":obj["metadata"]["resourceVersion"]},{"op":"add","path":f"/spec/template/spec/containers/{i}/volumeMounts","value":mounts},{"op":"add","path":"/spec/template/spec/volumes","value":volumes}]
|
|
def run(apply):
|
|
if kubectl("get","ns","kube-system","-o","json")["metadata"]["uid"]!=CLUSTER:raise ValueError("cluster identity mismatch")
|
|
es=kubectl("-n","sso","get","externalsecret","keycape-factor-read","-o","json")
|
|
if not any(x["type"]=="Ready" and x["status"]=="True" for x in es.get("status",{}).get("conditions",[])):raise ValueError("delivery not ready")
|
|
delivered=kubectl("-n","sso","get","secret","keycape-factor-read","-o","json")
|
|
expiry=int(base64.b64decode(delivered["data"]["expires-at"]))
|
|
if expiry-time.time()<600:raise ValueError("delivered token too near expiry")
|
|
secret=kubectl("-n","sso","get","secret","keycape-config","-o","json")
|
|
old=secret["data"]["config.yaml"];new=base64.b64encode(rewrite_config(base64.b64decode(old))).decode()
|
|
patch=[{"op":"test","path":"/metadata/uid","value":secret["metadata"]["uid"]},{"op":"test","path":"/metadata/resourceVersion","value":secret["metadata"]["resourceVersion"]},{"op":"replace","path":"/data/config.yaml","value":new}]
|
|
deployment=kubectl("-n","sso","get","deployment","keycape","-o","json");dp=deployment_patch(deployment)
|
|
for kind,name,ops in (("secret","keycape-config",patch),("deployment","keycape",dp)):
|
|
kubectl("-n","sso","patch",kind,name,"--type=json","--patch-file=/dev/stdin","--dry-run=server","-o","json",payload=ops)
|
|
if apply:
|
|
kubectl("-n","sso","patch","secret","keycape-config","--type=json","--patch-file=/dev/stdin","-o","json",payload=patch)
|
|
kubectl("-n","sso","patch","deployment","keycape","--type=json","--patch-file=/dev/stdin","-o","json",payload=dp)
|
|
current=kubectl("-n","sso","get","secret","keycape-config","-o","json")
|
|
expected=dict(secret["data"],**{"config.yaml":new})
|
|
if current["data"]!=expected:raise ValueError("config readback mismatch")
|
|
return {"success":True,"applied":apply,"unrelated_config_preserved":True,"credential_source":"mounted_file","expires_at":expiry}
|
|
if __name__=="__main__":
|
|
p=argparse.ArgumentParser();p.add_argument("--apply",action="store_true");a=p.parse_args()
|
|
try:result=run(a.apply)
|
|
except Exception:result={"success":False,"failure":"activation failed; inspect sanitized resource status"}
|
|
print(json.dumps(result));raise SystemExit(0 if result["success"] else 1)
|