From 6f4b87d44e43a4b4c6402000a989e09373c80174 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 4 Oct 2026 23:42:15 +0200 Subject: [PATCH] Align native proof with KeyCape Basic authentication and corrected image Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f --- .../infd-requester-proof-before-basic.json | 14 ++++++++++++++ docs/evidence/infd-requester-proof-wrong-ns.json | 6 ++++++ docs/evidence/infd-requester-proof-wrong-sa.json | 6 ++++++ .../informed-decision/native-proof.json | 10 +++++----- scripts/verify_informed_decision_requester.py | 12 +++++++----- 5 files changed, 38 insertions(+), 10 deletions(-) create mode 100644 docs/evidence/infd-requester-proof-before-basic.json create mode 100644 docs/evidence/infd-requester-proof-wrong-ns.json create mode 100644 docs/evidence/infd-requester-proof-wrong-sa.json diff --git a/docs/evidence/infd-requester-proof-before-basic.json b/docs/evidence/infd-requester-proof-before-basic.json new file mode 100644 index 0000000..924b9cc --- /dev/null +++ b/docs/evidence/infd-requester-proof-before-basic.json @@ -0,0 +1,14 @@ +{ + "complete": false, + "exact_identity": true, + "exact_path_projection_matches": true, + "exact_policy_and_ttl": true, + "negative_identity": false, + "parent_list_refused": true, + "phase": "requester_exchange", + "projection_mode": true, + "reader_revoked": true, + "revoked_lookup_refused": true, + "sibling_get_refused": true, + "wrong_audience_refused": true +} diff --git a/docs/evidence/infd-requester-proof-wrong-ns.json b/docs/evidence/infd-requester-proof-wrong-ns.json new file mode 100644 index 0000000..4974fec --- /dev/null +++ b/docs/evidence/infd-requester-proof-wrong-ns.json @@ -0,0 +1,6 @@ +{ + "complete": true, + "negative_identity": true, + "phase": "wrong_identity_refused", + "wrong_identity_refused": true +} diff --git a/docs/evidence/infd-requester-proof-wrong-sa.json b/docs/evidence/infd-requester-proof-wrong-sa.json new file mode 100644 index 0000000..4974fec --- /dev/null +++ b/docs/evidence/infd-requester-proof-wrong-sa.json @@ -0,0 +1,6 @@ +{ + "complete": true, + "negative_identity": true, + "phase": "wrong_identity_refused", + "wrong_identity_refused": true +} 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 index d7e907d..c6a0940 100644 --- 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 @@ -10,7 +10,7 @@ "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" + "proof.py": "\"\"\"Temporary native proof. Emits only fixed phases and booleans; never raw failures.\"\"\"\nimport base64\nimport json\nimport stat\nimport sys\nfrom pathlib import Path\nfrom urllib.parse import urlencode, quote_plus\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, client_auth=None):\n headers = {'Content-Type': 'application/x-www-form-urlencoded' if form else 'application/json'}\n if token: headers['X-Vault-Token'] = token\n if client_auth: headers['Authorization'] = 'Basic ' + client_auth\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 credentials = base64.b64encode((quote_plus(CLIENT_ID)+':'+quote_plus(secret)).encode()).decode()\n for scope in ('approval:approve', 'approval:read', 'approval:consume'):\n status, result = http('POST', ISSUER+'/token', form=True, client_auth=credentials, body=urlencode({\n 'grant_type': 'client_credentials', '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" } }, { @@ -89,7 +89,7 @@ "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" + "proof.py": "\"\"\"Temporary native proof. Emits only fixed phases and booleans; never raw failures.\"\"\"\nimport base64\nimport json\nimport stat\nimport sys\nfrom pathlib import Path\nfrom urllib.parse import urlencode, quote_plus\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, client_auth=None):\n headers = {'Content-Type': 'application/x-www-form-urlencoded' if form else 'application/json'}\n if token: headers['X-Vault-Token'] = token\n if client_auth: headers['Authorization'] = 'Basic ' + client_auth\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 credentials = base64.b64encode((quote_plus(CLIENT_ID)+':'+quote_plus(secret)).encode()).decode()\n for scope in ('approval:approve', 'approval:read', 'approval:consume'):\n status, result = http('POST', ISSUER+'/token', form=True, client_auth=credentials, body=urlencode({\n 'grant_type': 'client_credentials', '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" } }, { @@ -205,7 +205,7 @@ "containers": [ { "name": "proof", - "image": "forgejo.coulomb.social/coulomb/informed-decision@sha256:2768bd1da3e651cc0be2d0f1ae6eea1af4eb057b31ef63ebb03b004c2ed76c3d", + "image": "forgejo.coulomb.social/coulomb/informed-decision@sha256:804d89fb0bb3d691720380eedde4f361133bada717d1773c2eb349004346ebf1", "command": [ "python", "/proof/proof.py" @@ -328,7 +328,7 @@ "containers": [ { "name": "proof", - "image": "forgejo.coulomb.social/coulomb/informed-decision@sha256:2768bd1da3e651cc0be2d0f1ae6eea1af4eb057b31ef63ebb03b004c2ed76c3d", + "image": "forgejo.coulomb.social/coulomb/informed-decision@sha256:804d89fb0bb3d691720380eedde4f361133bada717d1773c2eb349004346ebf1", "command": [ "python", "/proof/proof.py", @@ -434,7 +434,7 @@ "containers": [ { "name": "proof", - "image": "forgejo.coulomb.social/coulomb/informed-decision@sha256:2768bd1da3e651cc0be2d0f1ae6eea1af4eb057b31ef63ebb03b004c2ed76c3d", + "image": "forgejo.coulomb.social/coulomb/informed-decision@sha256:804d89fb0bb3d691720380eedde4f361133bada717d1773c2eb349004346ebf1", "command": [ "python", "/proof/proof.py", diff --git a/scripts/verify_informed_decision_requester.py b/scripts/verify_informed_decision_requester.py index 3cfa6d9..e78a063 100644 --- a/scripts/verify_informed_decision_requester.py +++ b/scripts/verify_informed_decision_requester.py @@ -1,9 +1,10 @@ """Temporary native proof. Emits only fixed phases and booleans; never raw failures.""" +import base64 import json import stat import sys from pathlib import Path -from urllib.parse import urlencode +from urllib.parse import urlencode, quote_plus from urllib.request import Request, build_opener, ProxyHandler, HTTPRedirectHandler from urllib.error import HTTPError from informed_decision.requester import RequesterTokenProvider, CLIENT_ID @@ -17,9 +18,10 @@ 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): +def http(method, url, body=None, token=None, form=False, client_auth=None): headers = {'Content-Type': 'application/x-www-form-urlencoded' if form else 'application/json'} if token: headers['X-Vault-Token'] = token + if client_auth: headers['Authorization'] = 'Basic ' + client_auth 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) @@ -69,10 +71,10 @@ def run(negative=False): receipt['phase'] = 'requester_exchange' provider = RequesterTokenProvider(ISSUER, str(SECRET)) check('signed_create_only_exchange', bool(provider())) + credentials = base64.b64encode((quote_plus(CLIENT_ID)+':'+quote_plus(secret)).encode()).decode() 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()) + status, result = http('POST', ISSUER+'/token', form=True, client_auth=credentials, body=urlencode({ + 'grant_type': 'client_credentials', 'scope': scope}).encode()) check(scope.replace(':', '_')+'_refused', status in (400, 403) and 'access_token' not in result) receipt['complete'] = True except BaseException: