From 03f18d8258a208cb1f5b0432360dcd139cf0db4c Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 4 Oct 2026 23:40:38 +0200 Subject: [PATCH] Use KeyCape client-secret Basic authentication for requester exchange Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f --- informed_decision/requester.py | 13 +++++++++---- tests/test_requester.py | 21 ++++++++++++++++++--- 2 files changed, 27 insertions(+), 7 deletions(-) diff --git a/informed_decision/requester.py b/informed_decision/requester.py index 03c7894..f62dd9f 100644 --- a/informed_decision/requester.py +++ b/informed_decision/requester.py @@ -3,8 +3,9 @@ No access token or credential cache, environment delivery, refresh grant or browser identity. Public issuer endpoints and claims are pinned locally. """ +import base64 from pathlib import Path -from urllib.parse import urlencode +from urllib.parse import urlencode, quote_plus import time import jwt @@ -36,10 +37,14 @@ class RequesterTokenProvider: secret = raw.decode("ascii") if not secret or len(raw) > 4096 or any(c.isspace() for c in secret): raise ValueError() + credentials = base64.b64encode( + (quote_plus(CLIENT_ID) + ":" + quote_plus(secret)).encode("ascii") + ).decode("ascii") status, result = self.transport.request("POST", self.issuer + "/token", - headers={"Content-Type": "application/x-www-form-urlencoded"}, - body=urlencode({"grant_type": "client_credentials", "client_id": CLIENT_ID, - "client_secret": secret, "scope": "approval:create"}).encode()) + headers={"Content-Type": "application/x-www-form-urlencoded", + "Authorization": "Basic " + credentials}, + body=urlencode({"grant_type": "client_credentials", + "scope": "approval:create"}).encode()) if (status != 200 or not isinstance(result.get("token_type"), str) or result["token_type"].lower() != "bearer" or "refresh_token" in result): diff --git a/tests/test_requester.py b/tests/test_requester.py index b079a86..f00b9a7 100644 --- a/tests/test_requester.py +++ b/tests/test_requester.py @@ -1,6 +1,7 @@ +import base64 import json import time -from urllib.parse import parse_qs +from urllib.parse import parse_qs, unquote_plus from cryptography.hazmat.primitives.asymmetric import rsa import jwt @@ -29,7 +30,14 @@ def lane(tmp_path): assert url == ISSUER + "/jwks" return 200, {"keys": [jwk]} assert url == ISSUER + "/token" - self.seen.append(parse_qs(body.decode())) + fields = parse_qs(body.decode()) + assert set(fields) == {"grant_type", "scope"} + scheme, encoded = headers["Authorization"].split(" ", 1) + assert scheme == "Basic" + client_id, credential = base64.b64decode(encoded, validate=True).decode().split(":", 1) + assert unquote_plus(client_id) == CLIENT_ID + fields["decoded_credential"] = unquote_plus(credential) + self.seen.append(fields) return 200, {"token_type": "Bearer", "access_token": jwt.encode( claims, private, algorithm="RS256", headers={"kid": "test"})} transport = Transport() @@ -43,7 +51,7 @@ def test_exchange_reopens_projected_rotation(lane, tmp_path): replacement.write_text("synthetic-second") replacement.replace(secret) provider() - assert [call["client_secret"] for call in transport.seen] == [["synthetic-first"], ["synthetic-second"]] + assert [call["decoded_credential"] for call in transport.seen] == ["synthetic-first", "synthetic-second"] assert all(call["scope"] == ["approval:create"] and call["grant_type"] == ["client_credentials"] for call in transport.seen) secret.unlink() @@ -88,3 +96,10 @@ def test_invalid_custody_never_exchanges(lane, secret): with pytest.raises(RequesterError): provider() assert transport.seen == [] + + +def test_basic_auth_form_encodes_special_characters(lane): + provider, path, _, transport, _ = lane + path.write_text("synthetic+%:/=&") + provider() + assert transport.seen[-1]["decoded_credential"] == "synthetic+%:/=&"