Prepare bounded native requester and wrong-identity proofs

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f

scripts/verify_informed_decision_requester.py

@@ -0,0 +1,96 @@
+"""Temporary native proof. Emits only fixed phases and booleans; never raw failures."""
+import json
+import stat
+import sys
+from pathlib import Path
+from urllib.parse import urlencode
+from urllib.request import Request, build_opener, ProxyHandler, HTTPRedirectHandler
+from urllib.error import HTTPError
+from informed_decision.requester import RequesterTokenProvider, CLIENT_ID
+
+BAO = 'http://openbao.openbao.svc:8200/v1/'
+ISSUER = 'https://kc.coulomb.social'
+ROLE = 'informed-decision-sitting-requester-eso'
+POLICY = 'workload-kv-read-informed-decision-sitting-requester-eso'
+SECRET = Path('/var/run/secrets/informed-decision/requester/client-secret')
+
+class NoRedirect(HTTPRedirectHandler):
+    def redirect_request(self, *args, **kwargs): return None
+
+def http(method, url, body=None, token=None, form=False):
+    headers = {'Content-Type': 'application/x-www-form-urlencoded' if form else 'application/json'}
+    if token: headers['X-Vault-Token'] = token
+    raw = body if form else (json.dumps(body).encode() if body is not None else None)
+    req = Request(url, data=raw, headers=headers, method=method)
+    try: response = build_opener(ProxyHandler({}), NoRedirect()).open(req, timeout=5)
+    except HTTPError as error: response = error
+    with response:
+        status = response.code
+        content = response.read(262145)
+        if len(content) > 262144: raise ValueError()
+        return status, json.loads(content) if content else {}
+
+def run(negative=False):
+    receipt = {'complete': False, 'phase': 'start', 'negative_identity': negative}
+    token = None
+    def check(name, condition):
+        receipt['phase'] = name
+        receipt[name] = bool(condition)
+        if not condition: raise ValueError()
+    try:
+        receipt['phase'] = 'login'
+        status, result = http('POST', BAO+'auth/kubernetes/login',
+            {'role': ROLE, 'jwt': Path('/identity/openbao').read_text()})
+        if status == 200: token = result.get('auth', {}).get('client_token')
+        if negative:
+            check('wrong_identity_refused', status == 403 and token is None)
+        else:
+            auth = result.get('auth', {})
+            check('exact_identity', status == 200 and bool(token)
+                  and auth.get('metadata', {}).get('service_account_name') == 'informed-decision-eso'
+                  and auth.get('metadata', {}).get('service_account_namespace') == 'informed-decision')
+            status, result = http('GET', BAO+'auth/token/lookup-self', token=token)
+            data = result.get('data', {})
+            check('exact_policy_and_ttl', status == 200 and data.get('policies') == [POLICY]
+                  and not data.get('identity_policies') and 0 < data.get('ttl', 0) <= 900
+                  and data.get('creation_ttl') == 900 and data.get('explicit_max_ttl') == 900)
+            status, _ = http('GET', BAO+'platform/data/workloads/secrets-engine/approval-requester', token=token)
+            check('sibling_get_refused', status == 403)
+            status, _ = http('LIST', BAO+'platform/metadata/workloads/informed-decision', token=token)
+            check('parent_list_refused', status == 403)
+            status, _ = http('POST', BAO+'auth/kubernetes/login',
+                {'role': ROLE, 'jwt': Path('/identity/wrong-audience').read_text()})
+            check('wrong_audience_refused', status == 403)
+            check('projection_mode', stat.S_IMODE(SECRET.stat().st_mode) == 0o440)
+            secret = SECRET.read_text()
+            status, result = http('GET', BAO+'platform/data/workloads/informed-decision/sitting-requester', token=token)
+            check('exact_path_projection_matches', status == 200 and bool(secret)
+                  and result.get('data', {}).get('data', {}).get('CLIENT_SECRET') == secret)
+            receipt['phase'] = 'requester_exchange'
+            provider = RequesterTokenProvider(ISSUER, str(SECRET))
+            check('signed_create_only_exchange', bool(provider()))
+            for scope in ('approval:approve', 'approval:read', 'approval:consume'):
+                status, result = http('POST', ISSUER+'/token', form=True, body=urlencode({
+                    'grant_type': 'client_credentials', 'client_id': CLIENT_ID,
+                    'client_secret': secret, 'scope': scope}).encode())
+                check(scope.replace(':', '_')+'_refused', status in (400, 403) and 'access_token' not in result)
+        receipt['complete'] = True
+    except BaseException:
+        receipt['complete'] = False
+    finally:
+        if token:
+            try:
+                status, _ = http('POST', BAO+'auth/token/revoke-self', body={}, token=token)
+                receipt['reader_revoked'] = status in (200, 204)
+                status, _ = http('GET', BAO+'auth/token/lookup-self', token=token)
+                receipt['revoked_lookup_refused'] = status == 403
+                if not receipt['reader_revoked'] or not receipt['revoked_lookup_refused']: receipt['complete'] = False
+            except BaseException:
+                receipt['reader_revoked'] = False
+                receipt['complete'] = False
+    return receipt
+
+if __name__ == '__main__':
+    result = run('--negative' in sys.argv)
+    print(json.dumps(result, sort_keys=True))
+    sys.exit(0 if result['complete'] else 1)

tests/test_verify_informed_decision_requester.py

@@ -0,0 +1,34 @@
+import contextlib
+import importlib.util
+import io
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+spec = importlib.util.spec_from_file_location('proof', Path(__file__).resolve().parents[1]/'scripts/verify_informed_decision_requester.py')
+m = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(m)
+
+class ProofTests(unittest.TestCase):
+    def test_negative_identity_is_bounded(self):
+        with patch.object(m.Path, 'read_text', return_value='synthetic-only'), patch.object(m, 'http', return_value=(403, {})):
+            result = m.run(True)
+        self.assertTrue(result['complete'])
+        self.assertTrue(result['wrong_identity_refused'])
+
+    def test_exception_text_never_enters_receipt_or_output(self):
+        output = io.StringIO()
+        with patch.object(m.Path, 'read_text', side_effect=RuntimeError('SECRET-SENTINEL')), contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
+            result = m.run()
+        self.assertFalse(result['complete'])
+        self.assertNotIn('SECRET-SENTINEL', str(result))
+        self.assertEqual(output.getvalue(), '')
+
+    def test_unexpected_admission_revokes_reader_and_still_fails(self):
+        with patch.object(m.Path, 'read_text', return_value='synthetic-only'), patch.object(m, 'http', side_effect=[(200, {'auth': {'client_token': 'synthetic-reader'}}), (204, {}), (403, {})]):
+            result = m.run(True)
+        self.assertFalse(result['complete'])
+        self.assertTrue(result['reader_revoked'])
+        self.assertTrue(result['revoked_lookup_refused'])
+        self.assertNotIn('synthetic-reader', str(result))
+
+if __name__ == '__main__': unittest.main()