diff --git a/docs/evidence/infd-requester-proof-before-ingress-peer.json b/docs/evidence/infd-requester-proof-before-ingress-peer.json new file mode 100644 index 0000000..924b9cc --- /dev/null +++ b/docs/evidence/infd-requester-proof-before-ingress-peer.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/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 c6a0940..fa61d74 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 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" + "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\nfrom informed_decision.http_transport import JSONTransport\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 class ObservedTransport:\n def request(self, method, url, **kwargs):\n step = 'issuer_token_request' if method == 'POST' else 'issuer_jwks_request'\n receipt['phase'] = step\n status, result = JSONTransport().request(method, url, **kwargs)\n receipt[step+'_ok'] = status == 200\n return status, result\n provider = RequesterTokenProvider(ISSUER, str(SECRET), transport=ObservedTransport())\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" } }, { @@ -71,12 +71,41 @@ ] }, { + "to": [ + { + "ipBlock": { + "cidr": "92.205.62.239/32" + } + } + ], "ports": [ { "protocol": "TCP", "port": 443 } ] + }, + { + "to": [ + { + "namespaceSelector": { + "matchLabels": { + "kubernetes.io/metadata.name": "kube-system" + } + }, + "podSelector": { + "matchLabels": { + "app.kubernetes.io/name": "traefik" + } + } + } + ], + "ports": [ + { + "protocol": "TCP", + "port": 8443 + } + ] } ] } @@ -89,7 +118,7 @@ "namespace": "state-hub" }, "data": { - "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" + "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\nfrom informed_decision.http_transport import JSONTransport\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 class ObservedTransport:\n def request(self, method, url, **kwargs):\n step = 'issuer_token_request' if method == 'POST' else 'issuer_jwks_request'\n receipt['phase'] = step\n status, result = JSONTransport().request(method, url, **kwargs)\n receipt[step+'_ok'] = status == 200\n return status, result\n provider = RequesterTokenProvider(ISSUER, str(SECRET), transport=ObservedTransport())\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" } }, { @@ -150,12 +179,41 @@ ] }, { + "to": [ + { + "ipBlock": { + "cidr": "92.205.62.239/32" + } + } + ], "ports": [ { "protocol": "TCP", "port": 443 } ] + }, + { + "to": [ + { + "namespaceSelector": { + "matchLabels": { + "kubernetes.io/metadata.name": "kube-system" + } + }, + "podSelector": { + "matchLabels": { + "app.kubernetes.io/name": "traefik" + } + } + } + ], + "ports": [ + { + "protocol": "TCP", + "port": 8443 + } + ] } ] } diff --git a/scripts/verify_informed_decision_requester.py b/scripts/verify_informed_decision_requester.py index e78a063..c814f5d 100644 --- a/scripts/verify_informed_decision_requester.py +++ b/scripts/verify_informed_decision_requester.py @@ -8,6 +8,7 @@ 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 +from informed_decision.http_transport import JSONTransport BAO = 'http://openbao.openbao.svc:8200/v1/' ISSUER = 'https://kc.coulomb.social' @@ -69,7 +70,14 @@ def run(negative=False): 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)) + class ObservedTransport: + def request(self, method, url, **kwargs): + step = 'issuer_token_request' if method == 'POST' else 'issuer_jwks_request' + receipt['phase'] = step + status, result = JSONTransport().request(method, url, **kwargs) + receipt[step+'_ok'] = status == 200 + return status, result + provider = RequesterTokenProvider(ISSUER, str(SECRET), transport=ObservedTransport()) 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'):