Establish scoped KeyCape factor custody and verified automatic renewal

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

tests/test_keycape_factor_activation.py

@@ -0,0 +1,15 @@
+import importlib.util,unittest
+from pathlib import Path
+spec=importlib.util.spec_from_file_location("activation",Path(__file__).resolve().parents[1]/"scripts/keycape_factor_activate.py")
+m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
+class ActivationTests(unittest.TestCase):
+ def test_preserves_other_bytes_and_is_idempotent(self):
+  old=b"other: secret\nprivacyidea:\n  baseURL: http://provider\n  adminToken: expired\n  realm: coulomb\n  requireForAll: true\nclients:\n  - secret: unchanged\n"
+  new=m.rewrite_config(old)
+  self.assertEqual(old.replace(b"adminToken: expired",b"adminTokenFile: /etc/keycape-factor/admin-token"),new)
+  self.assertEqual(new,m.rewrite_config(new))
+ def test_rejects_ambiguous_source(self):
+  for text in (b"other: value\n",b"privacyidea:\n  adminToken: one\n  adminToken: two\n",b"privacyidea:\n  adminTokenFile: /unexpected\n"):
+   with self.assertRaises(ValueError):m.rewrite_config(text)
+ def test_refuses_unexpected_deployment(self):
+  with self.assertRaises(ValueError):m.deployment_patch({"metadata":{"uid":"replacement"}})

tests/test_keycape_factor_metadata.py

@@ -0,0 +1,31 @@
+import importlib.util,unittest
+from pathlib import Path
+from unittest.mock import patch
+spec=importlib.util.spec_from_file_location("factor_metadata",Path(__file__).resolve().parents[1]/"scripts/keycape_factor_metadata.py")
+m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
+class FactorMetadataTests(unittest.TestCase):
+ def test_exact_role_and_policy_scope(self):
+  self.assertEqual(["keycape-factor-eso"],m.ROLE_CONFIG["bound_service_account_names"])
+  self.assertEqual(["sso"],m.ROLE_CONFIG["bound_service_account_namespaces"])
+  self.assertTrue(m.ROLE_CONFIG["token_no_default_policy"])
+  self.assertEqual(900,m.ROLE_CONFIG["token_max_ttl"])
+  policy=(m.ROOT/"openbao/policies"/(m.POLICY+".hcl")).read_text()
+  self.assertNotIn("*",policy);self.assertNotIn("issuer",policy.splitlines()[1:])
+  self.assertEqual(3,policy.count('capabilities = ["read"]'))
+ def test_denied_capabilities_prevent_writes(self):
+  calls=[]
+  def fake(*args,**kwargs):
+   calls.append(args)
+   if args[:2]==("secrets","list"):return {"platform/":{"type":"kv","options":{"version":"2"}}}
+   if args[:2]==("auth","list"):return {"kubernetes/":{"type":"kubernetes"}}
+   return ["deny"]
+  with patch.object(m,"call",side_effect=fake):
+   with self.assertRaises(RuntimeError):m.execute(True)
+  self.assertFalse(any(c[0]=="write" for c in calls))
+ def test_mismatched_role_fails_readback(self):
+  self.assertTrue(m.verify_role(dict(m.ROLE_CONFIG)))
+  self.assertFalse(m.verify_role(dict(m.ROLE_CONFIG,token_policies=["root"])))
+
+ def test_policy_readback_accepts_cli_and_api_shapes(self):
+  self.assertEqual("expected",m.policy_text({"policy":"expected"}))
+  self.assertEqual("expected",m.policy_text({"data":{"policy":"expected"}}))

tests/test_keycape_factor_provider.py

@@ -0,0 +1,21 @@
+import base64,importlib.util,json,time,unittest
+from pathlib import Path
+from unittest.mock import patch
+spec=importlib.util.spec_from_file_location("factor_provider",Path(__file__).resolve().parents[1]/"scripts/keycape_factor_provider.py")
+m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
+class FactorProviderTests(unittest.TestCase):
+ def check(self,role="admin",auth_status=True,expiry=3600,policy_status=403,listing_status=True):
+  token="synthetic."+base64.urlsafe_b64encode(json.dumps({"exp":int(time.time())+expiry}).encode()).decode().rstrip("=")+".signature"
+  def request(path,payload=None,token=None):
+   if path=="/auth":return 200,{"result":{"status":auth_status,"value":{"role":role,"username":m.USER,"token":self.token}}}
+   if path=="/policy/":return policy_status,{}
+   return 200,{"result":{"status":listing_status,"value":{"tokens":[{}],"count":1}}}
+  self.token=token
+  with patch.object(m,"request",side_effect=request):return m.token_result("synthetic-password-not-a-real-secret")
+ def test_success_requires_scoped_admin_and_expiry(self):
+  r=self.check();self.assertTrue(r["policy_read_denied"]);self.assertTrue(r["cross_user_factor_visible"])
+ def test_rejects_user_token_expiry_and_privilege_leak(self):
+  for kwargs in [dict(role="user"),dict(auth_status=False),dict(expiry=-1),dict(expiry=9000),dict(policy_status=200),dict(policy_status=500),dict(listing_status=False)]:
+   with self.subTest(kwargs=kwargs):
+    with self.assertRaises(RuntimeError):self.check(**kwargs)
+ def test_no_redirect_forwarding(self):self.assertIsNone(m.NoRedirect().redirect_request(None,None,None,None,None,None))

tests/test_keycape_factor_renewal.py

@@ -0,0 +1,33 @@
+import base64,importlib.util,json,time,unittest
+from pathlib import Path
+from unittest.mock import patch
+spec=importlib.util.spec_from_file_location("factor_renew",Path(__file__).resolve().parents[1]/"scripts/keycape_factor_renew.py")
+m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
+class FactorRenewalTests(unittest.TestCase):
+ def drive(self,provider_failure=False,publish_failure=False,revoke_failure=False):
+  self.published=False;token="synthetic."+base64.urlsafe_b64encode(json.dumps({"exp":int(time.time())+3600}).encode()).decode()+".sig"
+  def http(url,body=None,headers=None,method=None,form=False):
+   if url.endswith('/auth/kubernetes/login'):return 200,{"auth":{"client_token":"synthetic-bao-session"}}
+   if url.endswith('/auth/token/revoke-self'):return (403 if revoke_failure else 204),{}
+   if url.endswith(m.ISSUER):return 200,{"data":{"data":{"REQUEST":m.PROVENANCE,"USERNAME":m.USER,"PASSWORD":"synthetic-service-password"}}}
+   if url.endswith('/auth'):
+    return (401,{}) if provider_failure else (200,{"result":{"status":True,"value":{"role":"admin","username":m.USER,"token":token}}})
+   if '/token/?' in url:return 200,{"result":{"status":True,"value":{"tokens":[{}],"count":1}}}
+   if url.endswith('/policy/'):return 403,{}
+   if url.endswith(m.TARGET):
+    if body:
+     if publish_failure:return 400,{}
+     self.assertEqual(1,body['options']['cas']);self.assertNotIn('PASSWORD',body['data']);self.published=True;return 200,{}
+    return 200,{"data":{"data":{"REQUEST":m.PROVENANCE,"TOKEN":token if self.published else "synthetic-old"},"metadata":{"version":2 if self.published else 1}}}
+   raise AssertionError('Unexpected endpoint')
+  with patch.object(m,'http',side_effect=http),patch.object(Path,'read_text',return_value='synthetic-kubernetes-jwt'):return m.run()
+ def test_renewal_publishes_verified_token_and_revokes_session(self):
+  r=self.drive();self.assertTrue(r['success']);self.assertEqual(2,r['kv_version']);self.assertTrue(r['session_revoked'])
+  self.assertNotIn('synthetic',json.dumps(r))
+ def test_rejected_provider_retains_current_token(self):
+  r=self.drive(provider_failure=True);self.assertFalse(r['success']);self.assertFalse(self.published);self.assertTrue(r['session_revoked'])
+  self.assertTrue(self.drive()['success'])
+ def test_cas_failure_is_not_success(self):
+  r=self.drive(publish_failure=True);self.assertFalse(r['success']);self.assertFalse(self.published);self.assertEqual('publish',r['phase'])
+ def test_cleanup_failure_is_reported(self):
+  r=self.drive(revoke_failure=True);self.assertFalse(r['success']);self.assertEqual('session_cleanup',r['phase'])