tegwick
268a9421ff
Some checks failed
Use-case regression gate / usecase-regressions (push) Has been cancelled
Exercise reset, registration and invitation against an isolated issuer and mailbox. The nodes stay unestablished and do not contact production. Assistant: grok Assistant-Session: 01a12cc2-e3b2-77a3-8aef-9cc60e577fee
136 lines
5.1 KiB
Python
136 lines
5.1 KiB
Python
"""Fast checks for the loopback communication double. No browser and no SMTP."""
|
|
|
|
import http.client
|
|
import json
|
|
from urllib.parse import urlencode
|
|
|
|
from tests.isolated_issuer import ALLOWED_CALLER, CLIENT_ID, IsolatedIssuer, challenge_s256
|
|
|
|
|
|
def test_code_replay_pkce_expiry_and_caller_denial():
|
|
issuer = IsolatedIssuer()
|
|
issuer.start()
|
|
try:
|
|
redirect = "http://127.0.0.1:9/callback"
|
|
issuer.allow_redirect(redirect)
|
|
verifier = "verifier-value-with-enough-length-for-pkce"
|
|
connection = http.client.HTTPConnection("127.0.0.1", issuer.port, timeout=5)
|
|
|
|
def post(path: str, form: dict) -> tuple[int, str, dict]:
|
|
raw = urlencode(form)
|
|
connection.request(
|
|
"POST",
|
|
path,
|
|
raw,
|
|
{"Content-Type": "application/x-www-form-urlencoded"},
|
|
)
|
|
response = connection.getresponse()
|
|
payload = response.read().decode()
|
|
headers = {key.lower(): value for key, value in response.getheaders()}
|
|
return response.status, payload, headers
|
|
|
|
status, _payload, headers = post(
|
|
"/authorize",
|
|
{
|
|
"client_id": CLIENT_ID,
|
|
"redirect_uri": redirect,
|
|
"code_challenge": challenge_s256(verifier),
|
|
"code_challenge_method": "S256",
|
|
"state": "state-1",
|
|
"subject": "member-other",
|
|
"email": "member@example.test",
|
|
"name": "Other Member",
|
|
"tenant": "tenant:communication-test",
|
|
},
|
|
)
|
|
assert status == 302
|
|
location = headers["location"]
|
|
code = location.split("code=")[1].split("&")[0]
|
|
token_form = {
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": redirect,
|
|
"client_id": CLIENT_ID,
|
|
"code_verifier": verifier,
|
|
}
|
|
status, payload, _headers = post("/token", token_form)
|
|
assert status == 200
|
|
assert "id_token" in json.loads(payload)
|
|
status, payload, _headers = post("/token", token_form)
|
|
assert status == 400
|
|
assert json.loads(payload)["error"] == "invalid_grant"
|
|
status, _payload, headers = post(
|
|
"/authorize",
|
|
{
|
|
"client_id": CLIENT_ID,
|
|
"redirect_uri": redirect,
|
|
"code_challenge": challenge_s256(verifier),
|
|
"code_challenge_method": "S256",
|
|
"state": "state-2",
|
|
"subject": "member-other",
|
|
"email": "member@example.test",
|
|
"name": "Other Member",
|
|
"tenant": "tenant:communication-test",
|
|
},
|
|
)
|
|
assert status == 302
|
|
second = headers["location"].split("code=")[1].split("&")[0]
|
|
bad = dict(token_form)
|
|
bad["code"] = second
|
|
bad["code_verifier"] = "wrong-verifier-value-with-enough-length"
|
|
status, _payload, _headers = post("/token", bad)
|
|
assert status == 400
|
|
|
|
status, payload, _headers = post(
|
|
"/authorize",
|
|
{
|
|
"client_id": "intruder",
|
|
"redirect_uri": redirect,
|
|
"code_challenge": challenge_s256(verifier),
|
|
"code_challenge_method": "S256",
|
|
"state": "x",
|
|
"subject": "member-other",
|
|
"email": "member@example.test",
|
|
"tenant": "tenant:communication-test",
|
|
},
|
|
)
|
|
assert status == 400
|
|
assert "Client refused" in payload
|
|
|
|
status, payload, _headers = post(
|
|
"/recover", {"email": "member@example.test", "expired": "1"}
|
|
)
|
|
assert status == 200
|
|
handle = issuer.messages[-1]["handle"]
|
|
status, payload, _headers = post("/recover/consume", {"handle": handle})
|
|
assert status == 400
|
|
assert "Handle expired" in payload
|
|
status, payload, _headers = post("/recover/consume", {"handle": handle})
|
|
assert status == 400
|
|
assert "Handle already used" in payload
|
|
|
|
status, payload, _headers = post(
|
|
"/invite", {"caller": "intruder", "email": "member@example.test"}
|
|
)
|
|
assert status == 400
|
|
assert "Caller refused" in payload
|
|
assert not any(item["kind"] == "invitation" for item in issuer.messages)
|
|
status, payload, _headers = post(
|
|
"/invite",
|
|
{"caller": ALLOWED_CALLER, "email": "member@example.test", "suppress": "1"},
|
|
)
|
|
assert status == 200
|
|
assert "Invitation suppressed" in payload
|
|
assert not any(item["kind"] == "invitation" for item in issuer.messages)
|
|
status, _payload, _headers = post(
|
|
"/invite", {"caller": ALLOWED_CALLER, "email": "member@example.test"}
|
|
)
|
|
assert status == 200
|
|
status, _payload, _headers = post(
|
|
"/invite", {"caller": ALLOWED_CALLER, "email": "member@example.test"}
|
|
)
|
|
assert status == 200
|
|
assert sum(item["kind"] == "invitation" for item in issuer.messages) == 2
|
|
connection.close()
|
|
finally:
|
|
issuer.stop()
|