diff --git a/docs/provisioning-windows/2026-10-04-1000/informed-decision/native-proof.json b/docs/provisioning-windows/2026-10-04-1000/informed-decision/native-proof.json new file mode 100644 index 0000000..d7e907d --- /dev/null +++ b/docs/provisioning-windows/2026-10-04-1000/informed-decision/native-proof.json @@ -0,0 +1,511 @@ +{ + "apiVersion": "v1", + "kind": "List", + "items": [ + { + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": { + "name": "infd-requester-proof", + "namespace": "informed-decision" + }, + "data": { + "proof.py": "\"\"\"Temporary native proof. Emits only fixed phases and booleans; never raw failures.\"\"\"\nimport json\nimport stat\nimport sys\nfrom pathlib import Path\nfrom urllib.parse import urlencode\nfrom urllib.request import Request, build_opener, ProxyHandler, HTTPRedirectHandler\nfrom urllib.error import HTTPError\nfrom informed_decision.requester import RequesterTokenProvider, CLIENT_ID\n\nBAO = 'http://openbao.openbao.svc:8200/v1/'\nISSUER = 'https://kc.coulomb.social'\nROLE = 'informed-decision-sitting-requester-eso'\nPOLICY = 'workload-kv-read-informed-decision-sitting-requester-eso'\nSECRET = Path('/var/run/secrets/informed-decision/requester/client-secret')\n\nclass NoRedirect(HTTPRedirectHandler):\n def redirect_request(self, *args, **kwargs): return None\n\ndef http(method, url, body=None, token=None, form=False):\n headers = {'Content-Type': 'application/x-www-form-urlencoded' if form else 'application/json'}\n if token: headers['X-Vault-Token'] = token\n raw = body if form else (json.dumps(body).encode() if body is not None else None)\n req = Request(url, data=raw, headers=headers, method=method)\n try: response = build_opener(ProxyHandler({}), NoRedirect()).open(req, timeout=5)\n except HTTPError as error: response = error\n with response:\n status = response.code\n content = response.read(262145)\n if len(content) > 262144: raise ValueError()\n return status, json.loads(content) if content else {}\n\ndef run(negative=False):\n receipt = {'complete': False, 'phase': 'start', 'negative_identity': negative}\n token = None\n def check(name, condition):\n receipt['phase'] = name\n receipt[name] = bool(condition)\n if not condition: raise ValueError()\n try:\n receipt['phase'] = 'login'\n status, result = http('POST', BAO+'auth/kubernetes/login',\n {'role': ROLE, 'jwt': Path('/identity/openbao').read_text()})\n if status == 200: token = result.get('auth', {}).get('client_token')\n if negative:\n check('wrong_identity_refused', status == 403 and token is None)\n else:\n auth = result.get('auth', {})\n check('exact_identity', status == 200 and bool(token)\n and auth.get('metadata', {}).get('service_account_name') == 'informed-decision-eso'\n and auth.get('metadata', {}).get('service_account_namespace') == 'informed-decision')\n status, result = http('GET', BAO+'auth/token/lookup-self', token=token)\n data = result.get('data', {})\n check('exact_policy_and_ttl', status == 200 and data.get('policies') == [POLICY]\n and not data.get('identity_policies') and 0 < data.get('ttl', 0) <= 900\n and data.get('creation_ttl') == 900 and data.get('explicit_max_ttl') == 900)\n status, _ = http('GET', BAO+'platform/data/workloads/secrets-engine/approval-requester', token=token)\n check('sibling_get_refused', status == 403)\n status, _ = http('LIST', BAO+'platform/metadata/workloads/informed-decision', token=token)\n check('parent_list_refused', status == 403)\n status, _ = http('POST', BAO+'auth/kubernetes/login',\n {'role': ROLE, 'jwt': Path('/identity/wrong-audience').read_text()})\n check('wrong_audience_refused', status == 403)\n check('projection_mode', stat.S_IMODE(SECRET.stat().st_mode) == 0o440)\n secret = SECRET.read_text()\n status, result = http('GET', BAO+'platform/data/workloads/informed-decision/sitting-requester', token=token)\n check('exact_path_projection_matches', status == 200 and bool(secret)\n and result.get('data', {}).get('data', {}).get('CLIENT_SECRET') == secret)\n receipt['phase'] = 'requester_exchange'\n provider = RequesterTokenProvider(ISSUER, str(SECRET))\n check('signed_create_only_exchange', bool(provider()))\n for scope in ('approval:approve', 'approval:read', 'approval:consume'):\n status, result = http('POST', ISSUER+'/token', form=True, body=urlencode({\n 'grant_type': 'client_credentials', 'client_id': CLIENT_ID,\n 'client_secret': secret, 'scope': scope}).encode())\n check(scope.replace(':', '_')+'_refused', status in (400, 403) and 'access_token' not in result)\n receipt['complete'] = True\n except BaseException:\n receipt['complete'] = False\n finally:\n if token:\n try:\n status, _ = http('POST', BAO+'auth/token/revoke-self', body={}, token=token)\n receipt['reader_revoked'] = status in (200, 204)\n status, _ = http('GET', BAO+'auth/token/lookup-self', token=token)\n receipt['revoked_lookup_refused'] = status == 403\n if not receipt['reader_revoked'] or not receipt['revoked_lookup_refused']: receipt['complete'] = False\n except BaseException:\n receipt['reader_revoked'] = False\n receipt['complete'] = False\n return receipt\n\nif __name__ == '__main__':\n result = run('--negative' in sys.argv)\n print(json.dumps(result, sort_keys=True))\n sys.exit(0 if result['complete'] else 1)\n" + } + }, + { + "apiVersion": "networking.k8s.io/v1", + "kind": "NetworkPolicy", + "metadata": { + "name": "infd-requester-proof", + "namespace": "informed-decision" + }, + "spec": { + "podSelector": { + "matchLabels": { + "verification": "infd-requester-proof" + } + }, + "policyTypes": [ + "Ingress", + "Egress" + ], + "ingress": [], + "egress": [ + { + "to": [ + { + "namespaceSelector": { + "matchLabels": { + "kubernetes.io/metadata.name": "kube-system" + } + } + } + ], + "ports": [ + { + "protocol": "UDP", + "port": 53 + }, + { + "protocol": "TCP", + "port": 53 + } + ] + }, + { + "to": [ + { + "namespaceSelector": { + "matchLabels": { + "kubernetes.io/metadata.name": "openbao" + } + } + } + ], + "ports": [ + { + "protocol": "TCP", + "port": 8200 + } + ] + }, + { + "ports": [ + { + "protocol": "TCP", + "port": 443 + } + ] + } + ] + } + }, + { + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": { + "name": "infd-requester-proof", + "namespace": "state-hub" + }, + "data": { + "proof.py": "\"\"\"Temporary native proof. Emits only fixed phases and booleans; never raw failures.\"\"\"\nimport json\nimport stat\nimport sys\nfrom pathlib import Path\nfrom urllib.parse import urlencode\nfrom urllib.request import Request, build_opener, ProxyHandler, HTTPRedirectHandler\nfrom urllib.error import HTTPError\nfrom informed_decision.requester import RequesterTokenProvider, CLIENT_ID\n\nBAO = 'http://openbao.openbao.svc:8200/v1/'\nISSUER = 'https://kc.coulomb.social'\nROLE = 'informed-decision-sitting-requester-eso'\nPOLICY = 'workload-kv-read-informed-decision-sitting-requester-eso'\nSECRET = Path('/var/run/secrets/informed-decision/requester/client-secret')\n\nclass NoRedirect(HTTPRedirectHandler):\n def redirect_request(self, *args, **kwargs): return None\n\ndef http(method, url, body=None, token=None, form=False):\n headers = {'Content-Type': 'application/x-www-form-urlencoded' if form else 'application/json'}\n if token: headers['X-Vault-Token'] = token\n raw = body if form else (json.dumps(body).encode() if body is not None else None)\n req = Request(url, data=raw, headers=headers, method=method)\n try: response = build_opener(ProxyHandler({}), NoRedirect()).open(req, timeout=5)\n except HTTPError as error: response = error\n with response:\n status = response.code\n content = response.read(262145)\n if len(content) > 262144: raise ValueError()\n return status, json.loads(content) if content else {}\n\ndef run(negative=False):\n receipt = {'complete': False, 'phase': 'start', 'negative_identity': negative}\n token = None\n def check(name, condition):\n receipt['phase'] = name\n receipt[name] = bool(condition)\n if not condition: raise ValueError()\n try:\n receipt['phase'] = 'login'\n status, result = http('POST', BAO+'auth/kubernetes/login',\n {'role': ROLE, 'jwt': Path('/identity/openbao').read_text()})\n if status == 200: token = result.get('auth', {}).get('client_token')\n if negative:\n check('wrong_identity_refused', status == 403 and token is None)\n else:\n auth = result.get('auth', {})\n check('exact_identity', status == 200 and bool(token)\n and auth.get('metadata', {}).get('service_account_name') == 'informed-decision-eso'\n and auth.get('metadata', {}).get('service_account_namespace') == 'informed-decision')\n status, result = http('GET', BAO+'auth/token/lookup-self', token=token)\n data = result.get('data', {})\n check('exact_policy_and_ttl', status == 200 and data.get('policies') == [POLICY]\n and not data.get('identity_policies') and 0 < data.get('ttl', 0) <= 900\n and data.get('creation_ttl') == 900 and data.get('explicit_max_ttl') == 900)\n status, _ = http('GET', BAO+'platform/data/workloads/secrets-engine/approval-requester', token=token)\n check('sibling_get_refused', status == 403)\n status, _ = http('LIST', BAO+'platform/metadata/workloads/informed-decision', token=token)\n check('parent_list_refused', status == 403)\n status, _ = http('POST', BAO+'auth/kubernetes/login',\n {'role': ROLE, 'jwt': Path('/identity/wrong-audience').read_text()})\n check('wrong_audience_refused', status == 403)\n check('projection_mode', stat.S_IMODE(SECRET.stat().st_mode) == 0o440)\n secret = SECRET.read_text()\n status, result = http('GET', BAO+'platform/data/workloads/informed-decision/sitting-requester', token=token)\n check('exact_path_projection_matches', status == 200 and bool(secret)\n and result.get('data', {}).get('data', {}).get('CLIENT_SECRET') == secret)\n receipt['phase'] = 'requester_exchange'\n provider = RequesterTokenProvider(ISSUER, str(SECRET))\n check('signed_create_only_exchange', bool(provider()))\n for scope in ('approval:approve', 'approval:read', 'approval:consume'):\n status, result = http('POST', ISSUER+'/token', form=True, body=urlencode({\n 'grant_type': 'client_credentials', 'client_id': CLIENT_ID,\n 'client_secret': secret, 'scope': scope}).encode())\n check(scope.replace(':', '_')+'_refused', status in (400, 403) and 'access_token' not in result)\n receipt['complete'] = True\n except BaseException:\n receipt['complete'] = False\n finally:\n if token:\n try:\n status, _ = http('POST', BAO+'auth/token/revoke-self', body={}, token=token)\n receipt['reader_revoked'] = status in (200, 204)\n status, _ = http('GET', BAO+'auth/token/lookup-self', token=token)\n receipt['revoked_lookup_refused'] = status == 403\n if not receipt['reader_revoked'] or not receipt['revoked_lookup_refused']: receipt['complete'] = False\n except BaseException:\n receipt['reader_revoked'] = False\n receipt['complete'] = False\n return receipt\n\nif __name__ == '__main__':\n result = run('--negative' in sys.argv)\n print(json.dumps(result, sort_keys=True))\n sys.exit(0 if result['complete'] else 1)\n" + } + }, + { + "apiVersion": "networking.k8s.io/v1", + "kind": "NetworkPolicy", + "metadata": { + "name": "infd-requester-proof", + "namespace": "state-hub" + }, + "spec": { + "podSelector": { + "matchLabels": { + "verification": "infd-requester-proof" + } + }, + "policyTypes": [ + "Ingress", + "Egress" + ], + "ingress": [], + "egress": [ + { + "to": [ + { + "namespaceSelector": { + "matchLabels": { + "kubernetes.io/metadata.name": "kube-system" + } + } + } + ], + "ports": [ + { + "protocol": "UDP", + "port": 53 + }, + { + "protocol": "TCP", + "port": 53 + } + ] + }, + { + "to": [ + { + "namespaceSelector": { + "matchLabels": { + "kubernetes.io/metadata.name": "openbao" + } + } + } + ], + "ports": [ + { + "protocol": "TCP", + "port": 8200 + } + ] + }, + { + "ports": [ + { + "protocol": "TCP", + "port": 443 + } + ] + } + ] + } + }, + { + "apiVersion": "v1", + "kind": "ServiceAccount", + "metadata": { + "name": "informed-decision-eso", + "namespace": "state-hub", + "labels": { + "verification": "infd-requester-proof" + } + }, + "automountServiceAccountToken": false + }, + { + "apiVersion": "batch/v1", + "kind": "Job", + "metadata": { + "name": "infd-requester-proof", + "namespace": "informed-decision" + }, + "spec": { + "backoffLimit": 0, + "activeDeadlineSeconds": 90, + "ttlSecondsAfterFinished": 600, + "template": { + "metadata": { + "labels": { + "verification": "infd-requester-proof" + } + }, + "spec": { + "serviceAccountName": "informed-decision-eso", + "automountServiceAccountToken": false, + "restartPolicy": "Never", + "securityContext": { + "runAsNonRoot": true, + "runAsUser": 10001, + "runAsGroup": 10001, + "fsGroup": 10001, + "seccompProfile": { + "type": "RuntimeDefault" + } + }, + "containers": [ + { + "name": "proof", + "image": "forgejo.coulomb.social/coulomb/informed-decision@sha256:2768bd1da3e651cc0be2d0f1ae6eea1af4eb057b31ef63ebb03b004c2ed76c3d", + "command": [ + "python", + "/proof/proof.py" + ], + "resources": { + "requests": { + "cpu": "5m", + "memory": "32Mi" + }, + "limits": { + "cpu": "100m", + "memory": "128Mi" + } + }, + "securityContext": { + "readOnlyRootFilesystem": true, + "allowPrivilegeEscalation": false, + "capabilities": { + "drop": [ + "ALL" + ] + } + }, + "volumeMounts": [ + { + "name": "proof", + "mountPath": "/proof", + "readOnly": true + }, + { + "name": "identity", + "mountPath": "/identity", + "readOnly": true + }, + { + "name": "requester", + "mountPath": "/var/run/secrets/informed-decision/requester", + "readOnly": true + } + ] + } + ], + "volumes": [ + { + "name": "proof", + "configMap": { + "name": "infd-requester-proof" + } + }, + { + "name": "identity", + "projected": { + "defaultMode": 288, + "sources": [ + { + "serviceAccountToken": { + "path": "openbao", + "audience": "openbao", + "expirationSeconds": 600 + } + }, + { + "serviceAccountToken": { + "path": "wrong-audience", + "audience": "informed-decision-verification-wrong-audience", + "expirationSeconds": 600 + } + } + ] + } + }, + { + "name": "requester", + "secret": { + "secretName": "informed-decision-sitting-requester", + "defaultMode": 288, + "items": [ + { + "key": "client-secret", + "path": "client-secret" + } + ] + } + } + ] + } + } + } + }, + { + "apiVersion": "batch/v1", + "kind": "Job", + "metadata": { + "name": "infd-requester-proof-wrong-sa", + "namespace": "informed-decision" + }, + "spec": { + "backoffLimit": 0, + "activeDeadlineSeconds": 90, + "ttlSecondsAfterFinished": 600, + "template": { + "metadata": { + "labels": { + "verification": "infd-requester-proof" + } + }, + "spec": { + "serviceAccountName": "review", + "automountServiceAccountToken": false, + "restartPolicy": "Never", + "securityContext": { + "runAsNonRoot": true, + "runAsUser": 10001, + "runAsGroup": 10001, + "fsGroup": 10001, + "seccompProfile": { + "type": "RuntimeDefault" + } + }, + "containers": [ + { + "name": "proof", + "image": "forgejo.coulomb.social/coulomb/informed-decision@sha256:2768bd1da3e651cc0be2d0f1ae6eea1af4eb057b31ef63ebb03b004c2ed76c3d", + "command": [ + "python", + "/proof/proof.py", + "--negative" + ], + "resources": { + "requests": { + "cpu": "5m", + "memory": "32Mi" + }, + "limits": { + "cpu": "100m", + "memory": "128Mi" + } + }, + "securityContext": { + "readOnlyRootFilesystem": true, + "allowPrivilegeEscalation": false, + "capabilities": { + "drop": [ + "ALL" + ] + } + }, + "volumeMounts": [ + { + "name": "proof", + "mountPath": "/proof", + "readOnly": true + }, + { + "name": "identity", + "mountPath": "/identity", + "readOnly": true + } + ] + } + ], + "volumes": [ + { + "name": "proof", + "configMap": { + "name": "infd-requester-proof" + } + }, + { + "name": "identity", + "projected": { + "defaultMode": 288, + "sources": [ + { + "serviceAccountToken": { + "path": "openbao", + "audience": "openbao", + "expirationSeconds": 600 + } + }, + { + "serviceAccountToken": { + "path": "wrong-audience", + "audience": "informed-decision-verification-wrong-audience", + "expirationSeconds": 600 + } + } + ] + } + } + ] + } + } + } + }, + { + "apiVersion": "batch/v1", + "kind": "Job", + "metadata": { + "name": "infd-requester-proof-wrong-ns", + "namespace": "state-hub" + }, + "spec": { + "backoffLimit": 0, + "activeDeadlineSeconds": 90, + "ttlSecondsAfterFinished": 600, + "template": { + "metadata": { + "labels": { + "verification": "infd-requester-proof" + } + }, + "spec": { + "serviceAccountName": "informed-decision-eso", + "automountServiceAccountToken": false, + "restartPolicy": "Never", + "securityContext": { + "runAsNonRoot": true, + "runAsUser": 10001, + "runAsGroup": 10001, + "fsGroup": 10001, + "seccompProfile": { + "type": "RuntimeDefault" + } + }, + "containers": [ + { + "name": "proof", + "image": "forgejo.coulomb.social/coulomb/informed-decision@sha256:2768bd1da3e651cc0be2d0f1ae6eea1af4eb057b31ef63ebb03b004c2ed76c3d", + "command": [ + "python", + "/proof/proof.py", + "--negative" + ], + "resources": { + "requests": { + "cpu": "5m", + "memory": "32Mi" + }, + "limits": { + "cpu": "100m", + "memory": "128Mi" + } + }, + "securityContext": { + "readOnlyRootFilesystem": true, + "allowPrivilegeEscalation": false, + "capabilities": { + "drop": [ + "ALL" + ] + } + }, + "volumeMounts": [ + { + "name": "proof", + "mountPath": "/proof", + "readOnly": true + }, + { + "name": "identity", + "mountPath": "/identity", + "readOnly": true + } + ] + } + ], + "volumes": [ + { + "name": "proof", + "configMap": { + "name": "infd-requester-proof" + } + }, + { + "name": "identity", + "projected": { + "defaultMode": 288, + "sources": [ + { + "serviceAccountToken": { + "path": "openbao", + "audience": "openbao", + "expirationSeconds": 600 + } + }, + { + "serviceAccountToken": { + "path": "wrong-audience", + "audience": "informed-decision-verification-wrong-audience", + "expirationSeconds": 600 + } + } + ] + } + } + ] + } + } + } + } + ] +} diff --git a/scripts/verify_informed_decision_requester.py b/scripts/verify_informed_decision_requester.py new file mode 100644 index 0000000..3cfa6d9 --- /dev/null +++ b/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) diff --git a/tests/test_verify_informed_decision_requester.py b/tests/test_verify_informed_decision_requester.py new file mode 100644 index 0000000..a3930c5 --- /dev/null +++ b/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()