Use KeyCape client-secret Basic authentication for requester exchange

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
This commit is contained in:
tegwick 2026-10-04 23:40:38 +02:00
parent 929f6cdff6
commit 03f18d8258
2 changed files with 27 additions and 7 deletions

View file

@ -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):

View file

@ -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+%:/=&"