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
414 lines
17 KiB
Python
414 lines
17 KiB
Python
"""Loopback identity, directory and mailbox double for communication journeys.
|
|
|
|
This is a synthetic test peer. It does not send SMTP, store a reusable
|
|
password, or grant application membership from a mailbox event.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import secrets
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import parse_qs, urlencode, urlparse
|
|
|
|
CLIENT_ID = "coulomb-social-test"
|
|
TENANT = "tenant:communication-test"
|
|
ALLOWED_CALLER = "coulomb-social-test"
|
|
|
|
|
|
def challenge_s256(verifier: str) -> str:
|
|
digest = hashlib.sha256(verifier.encode("ascii")).digest()
|
|
return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
|
|
|
|
|
|
def _html(body: str, *, status: int = 200) -> tuple[int, str, bytes, list[tuple[str, str]]]:
|
|
page = (
|
|
"<!doctype html><html><head><meta charset=utf-8><title>Isolated issuer</title></head>"
|
|
f"<body><main>{body}</main></body></html>"
|
|
)
|
|
return status, "text/html; charset=utf-8", page.encode(), []
|
|
|
|
|
|
def _esc(value: str) -> str:
|
|
return (
|
|
value.replace("&", "&")
|
|
.replace("<", "<")
|
|
.replace(">", ">")
|
|
.replace('"', """)
|
|
)
|
|
|
|
|
|
class IsolatedIssuer:
|
|
def __init__(self) -> None:
|
|
from authlib.jose import JsonWebKey
|
|
|
|
self.key = JsonWebKey.generate_key("RSA", 2048, is_private=True)
|
|
self.allowed_redirects: set[str] = set()
|
|
self.codes: dict[str, dict] = {}
|
|
self.messages: list[dict] = []
|
|
self.password_hashes: dict[str, str] = {}
|
|
self._salt = secrets.token_hex(16)
|
|
self._httpd: ThreadingHTTPServer | None = None
|
|
self._thread: threading.Thread | None = None
|
|
self.port = 0
|
|
|
|
@property
|
|
def base(self) -> str:
|
|
return f"http://127.0.0.1:{self.port}"
|
|
|
|
def allow_redirect(self, url: str) -> None:
|
|
self.allowed_redirects.add(url)
|
|
|
|
def start(self) -> None:
|
|
issuer = self
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None: # noqa: N802
|
|
self._dispatch("GET")
|
|
|
|
def do_POST(self) -> None: # noqa: N802
|
|
self._dispatch("POST")
|
|
|
|
def log_message(self, format: str, *args) -> None:
|
|
return
|
|
|
|
def _dispatch(self, method: str) -> None:
|
|
parsed = urlparse(self.path)
|
|
form = self._form() if method == "POST" else {}
|
|
query = {key: values[-1] for key, values in parse_qs(parsed.query).items()}
|
|
status, content_type, body, headers = issuer.handle(
|
|
method, parsed.path, query, form
|
|
)
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
for name, value in headers:
|
|
self.send_header(name, value)
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _form(self) -> dict[str, str]:
|
|
length = int(self.headers.get("Content-Length") or "0")
|
|
raw = self.rfile.read(length).decode() if length else ""
|
|
return {key: values[-1] for key, values in parse_qs(raw).items()}
|
|
|
|
self._httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
self.port = self._httpd.server_address[1]
|
|
self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True)
|
|
self._thread.start()
|
|
|
|
def stop(self) -> None:
|
|
if self._httpd is not None:
|
|
self._httpd.shutdown()
|
|
self._httpd.server_close()
|
|
if self._thread is not None:
|
|
self._thread.join(timeout=5)
|
|
|
|
def remember_password(self, subject: str, password: str) -> None:
|
|
digest = hashlib.sha256(f"{self._salt}|{subject}|{password}".encode()).hexdigest()
|
|
self.password_hashes[subject] = digest
|
|
|
|
def handle(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
query: dict[str, str],
|
|
form: dict[str, str],
|
|
) -> tuple[int, str, bytes, list[tuple[str, str]]]:
|
|
routes = {
|
|
("GET", "/.well-known/openid-configuration"): self._discovery_response,
|
|
("GET", "/jwks"): self._jwks,
|
|
("GET", "/authorize"): lambda: self._authorize("GET", query, form),
|
|
("POST", "/authorize"): lambda: self._authorize("POST", query, form),
|
|
("POST", "/token"): lambda: self._token(form),
|
|
("GET", "/userinfo"): self._userinfo,
|
|
("GET", "/recover"): lambda: self._recover("GET", query, form),
|
|
("POST", "/recover"): lambda: self._recover("POST", query, form),
|
|
("GET", "/mailbox"): lambda: _html(self._mailbox_body()),
|
|
("POST", "/recover/consume"): lambda: self._consume(form, kind="reset"),
|
|
("POST", "/recover/cancel"): lambda: self._cancel(form),
|
|
("POST", "/recover/password"): lambda: self._password(form),
|
|
("GET", "/register"): lambda: self._register("GET", form),
|
|
("POST", "/register"): lambda: self._register("POST", form),
|
|
("POST", "/verify/consume"): lambda: self._consume(form, kind="verification"),
|
|
("POST", "/verify/cancel"): lambda: self._cancel(form),
|
|
("GET", "/invite"): lambda: self._invite("GET", form),
|
|
("POST", "/invite"): lambda: self._invite("POST", form),
|
|
}
|
|
route = routes.get((method, path))
|
|
if route is None:
|
|
return _html("<h1>Not found</h1>", status=404)
|
|
return route()
|
|
|
|
def _discovery(self) -> dict:
|
|
return {
|
|
"issuer": self.base,
|
|
"authorization_endpoint": f"{self.base}/authorize",
|
|
"token_endpoint": f"{self.base}/token",
|
|
"userinfo_endpoint": f"{self.base}/userinfo",
|
|
"jwks_uri": f"{self.base}/jwks",
|
|
"response_types_supported": ["code"],
|
|
"subject_types_supported": ["public"],
|
|
"id_token_signing_alg_values_supported": ["RS256"],
|
|
}
|
|
|
|
def _discovery_response(self) -> tuple:
|
|
body = json.dumps(self._discovery()).encode()
|
|
return 200, "application/json", body, []
|
|
|
|
def _jwks(self) -> tuple:
|
|
body = json.dumps({"keys": [self.key.as_dict(is_private=False)]}).encode()
|
|
return 200, "application/json", body, []
|
|
|
|
def _authorize(self, method: str, query: dict, form: dict) -> tuple:
|
|
fields = form if method == "POST" else query
|
|
client_id = fields.get("client_id", "")
|
|
redirect_uri = fields.get("redirect_uri", "")
|
|
if client_id != CLIENT_ID:
|
|
return _html("<h1>Client refused</h1>", status=400)
|
|
if redirect_uri not in self.allowed_redirects:
|
|
return _html("<h1>Redirect refused</h1>", status=400)
|
|
challenge = fields.get("code_challenge", "")
|
|
if not challenge or fields.get("code_challenge_method") != "S256":
|
|
return _html("<h1>PKCE required</h1>", status=400)
|
|
if method == "GET":
|
|
return _html(self._authorize_form(fields))
|
|
subject = (fields.get("subject") or "").strip()
|
|
email = (fields.get("email") or "").strip()
|
|
tenant = (fields.get("tenant") or "").strip()
|
|
name = (fields.get("name") or subject).strip()
|
|
if tenant != TENANT:
|
|
return _html("<h1>Tenant refused</h1>", status=400)
|
|
if subject == "member-recovered" and subject not in self.password_hashes:
|
|
return _html("<h1>Recovery required</h1>", status=400)
|
|
if not subject or not email:
|
|
return _html("<h1>Subject and email are required</h1>", status=400)
|
|
code = secrets.token_urlsafe(18)
|
|
self.codes[code] = {
|
|
"challenge": challenge,
|
|
"redirect_uri": redirect_uri,
|
|
"client_id": client_id,
|
|
"subject": subject,
|
|
"email": email,
|
|
"name": name,
|
|
"tenant": tenant,
|
|
"expired": fields.get("expired") == "1",
|
|
"used": False,
|
|
}
|
|
location = redirect_uri + ("&" if "?" in redirect_uri else "?") + urlencode(
|
|
{"code": code, "state": fields.get("state", "")}
|
|
)
|
|
return 302, "text/plain", b"", [("Location", location)]
|
|
|
|
def _authorize_form(self, fields: dict[str, str]) -> str:
|
|
hidden = "".join(
|
|
f'<input type="hidden" name="{_esc(key)}" value="{_esc(value)}">'
|
|
for key, value in fields.items()
|
|
if key not in {"subject", "email", "name", "tenant", "expired"}
|
|
)
|
|
return (
|
|
"<h1>Sign in at the isolated issuer</h1>"
|
|
'<form method="post" action="/authorize">'
|
|
f"{hidden}"
|
|
'<label>Subject <input name="subject" value="member-recovered"></label>'
|
|
'<label>Email <input name="email" value="member@example.test"></label>'
|
|
'<label>Name <input name="name" value="Recovered Member"></label>'
|
|
f'<label>Tenant <input name="tenant" value="{TENANT}"></label>'
|
|
'<button type="submit">Continue</button>'
|
|
"</form>"
|
|
)
|
|
|
|
def _token(self, form: dict[str, str]) -> tuple:
|
|
code = self.codes.get(form.get("code", ""))
|
|
verifier = form.get("code_verifier", "")
|
|
if (
|
|
form.get("grant_type") != "authorization_code"
|
|
or code is None
|
|
or code["used"]
|
|
or form.get("client_id") != code["client_id"]
|
|
or form.get("redirect_uri") != code["redirect_uri"]
|
|
or challenge_s256(verifier) != code["challenge"]
|
|
):
|
|
body = json.dumps({"error": "invalid_grant"}).encode()
|
|
return 400, "application/json", body, []
|
|
code["used"] = True
|
|
now = int(time.time())
|
|
exp = now - 120 if code["expired"] else now + 300
|
|
claims = {
|
|
"iss": self.base,
|
|
"sub": code["subject"],
|
|
"aud": CLIENT_ID,
|
|
"exp": exp,
|
|
"iat": now - 10,
|
|
"email": code["email"],
|
|
"name": code["name"],
|
|
"tenant": code["tenant"],
|
|
"azp": CLIENT_ID,
|
|
}
|
|
from authlib.jose import jwt
|
|
|
|
kid = self.key.as_dict(is_private=False).get("kid")
|
|
token = jwt.encode({"alg": "RS256", "kid": kid}, claims, self.key).decode()
|
|
access = secrets.token_urlsafe(18)
|
|
body = json.dumps(
|
|
{
|
|
"access_token": access,
|
|
"token_type": "Bearer",
|
|
"expires_in": 300,
|
|
"id_token": token,
|
|
}
|
|
).encode()
|
|
return 200, "application/json", body, []
|
|
|
|
def _userinfo(self) -> tuple:
|
|
return 200, "application/json", b"{}", []
|
|
|
|
def _add_message(self, *, kind: str, email: str, subject: str, expired: bool) -> None:
|
|
self.messages.append(
|
|
{
|
|
"kind": kind,
|
|
"email": email,
|
|
"subject": subject,
|
|
"handle": secrets.token_urlsafe(18),
|
|
"expired": expired,
|
|
"state": "ready",
|
|
}
|
|
)
|
|
|
|
def _recover(self, method: str, query: dict, form: dict) -> tuple:
|
|
if method == "GET":
|
|
expired = "1" if query.get("expired") == "1" else "0"
|
|
return _html(
|
|
"<h1>Request a reset</h1>"
|
|
'<form method="post" action="/recover">'
|
|
f'<input type="hidden" name="expired" value="{expired}">'
|
|
'<label>Email <input name="email" value="member@example.test"></label>'
|
|
'<button type="submit">Send reset</button></form>'
|
|
)
|
|
email = (form.get("email") or "").strip()
|
|
subject = "member-recovered" if email == "member@example.test" else "other-recovered"
|
|
self._add_message(
|
|
kind="reset", email=email, subject=subject, expired=form.get("expired") == "1"
|
|
)
|
|
return _html(
|
|
"<h1>Reset message is in the local mailbox</h1><a href=\"/mailbox\">Open mailbox</a>"
|
|
)
|
|
|
|
def _mailbox_body(self) -> str:
|
|
parts = ["<h1>Local mailbox</h1>"]
|
|
if not self.messages:
|
|
parts.append("<p>No messages.</p>")
|
|
for message in self.messages:
|
|
label = message["kind"]
|
|
if message["state"] != "ready":
|
|
parts.append(f"<p>{_esc(label)} {_esc(message['state'])}</p>")
|
|
continue
|
|
if label == "invitation":
|
|
parts.append(
|
|
f"<p>{_esc(message['email'])} invitation</p>"
|
|
"<button type=\"button\">Open invitation</button>"
|
|
)
|
|
continue
|
|
action = "/recover/consume" if label == "reset" else "/verify/consume"
|
|
cancel = "/recover/cancel" if label == "reset" else "/verify/cancel"
|
|
button = "Use expired reset" if message["expired"] else f"Use {label}"
|
|
handle = _esc(message["handle"])
|
|
parts.append(
|
|
f'<form method="post" action="{action}">'
|
|
f'<input type="hidden" name="handle" value="{handle}">'
|
|
f"<button>{button}</button></form>"
|
|
f'<form method="post" action="{cancel}">'
|
|
f'<input type="hidden" name="handle" value="{handle}">'
|
|
f"<button>Cancel {label}</button></form>"
|
|
)
|
|
return "".join(parts)
|
|
|
|
def _message(self, handle: str) -> dict | None:
|
|
for message in self.messages:
|
|
if message["handle"] == handle:
|
|
return message
|
|
return None
|
|
|
|
def _consume(self, form: dict, *, kind: str) -> tuple:
|
|
message = self._message(form.get("handle", ""))
|
|
if message is None or message["kind"] != kind:
|
|
return _html("<h1>Unknown handle</h1>", status=400)
|
|
if message["state"] != "ready":
|
|
return _html("<h1>Handle already used</h1>", status=400)
|
|
if message["expired"]:
|
|
message["state"] = "expired"
|
|
return _html("<h1>Handle expired</h1>", status=400)
|
|
message["state"] = "consumed"
|
|
if kind == "reset":
|
|
return _html(
|
|
"<h1>Set a new password</h1>"
|
|
'<form method="post" action="/recover/password">'
|
|
f'<input type="hidden" name="subject" value="{_esc(message["subject"])}">'
|
|
'<label>Password <input name="password" type="password"></label>'
|
|
"<button>Save password</button></form>"
|
|
"<p>The application never receives this password.</p>"
|
|
)
|
|
return _html(
|
|
"<h1>Verification consumed</h1>"
|
|
"<p>This does not sign in to the application.</p>"
|
|
)
|
|
|
|
def _password(self, form: dict) -> tuple:
|
|
subject = (form.get("subject") or "").strip()
|
|
password = form.get("password") or ""
|
|
if not subject or not password:
|
|
return _html("<h1>Password required</h1>", status=400)
|
|
self.remember_password(subject, password)
|
|
return _html("<h1>Password updated</h1>")
|
|
|
|
def _cancel(self, form: dict) -> tuple:
|
|
message = self._message(form.get("handle", ""))
|
|
if message is None or message["state"] != "ready":
|
|
return _html("<h1>Handle already used</h1>", status=400)
|
|
message["state"] = "cancelled"
|
|
return _html("<h1>Handle cancelled</h1>")
|
|
|
|
def _register(self, method: str, form: dict) -> tuple:
|
|
if method == "GET":
|
|
return _html(
|
|
"<h1>Registration sink</h1>"
|
|
"<form method=\"post\">"
|
|
'<label>Email <input name="email" value="new@example.test"></label>'
|
|
"<button>Send verification</button></form>"
|
|
)
|
|
self._add_message(
|
|
kind="verification",
|
|
email=(form.get("email") or "").strip(),
|
|
subject="applicant",
|
|
expired=False,
|
|
)
|
|
return _html(
|
|
"<h1>Verification message is in the local mailbox</h1>"
|
|
'<a href="/mailbox">Open mailbox</a>'
|
|
)
|
|
|
|
def _invite(self, method: str, form: dict) -> tuple:
|
|
if method == "GET":
|
|
return _html(
|
|
"<h1>Invite a member</h1>"
|
|
"<form method=\"post\">"
|
|
'<label>Caller <input name="caller"></label>'
|
|
'<label>Email <input name="email" value="member@example.test"></label>'
|
|
'<label>Suppress <input name="suppress" type="checkbox" value="1"></label>'
|
|
"<button>Send invitation</button></form>"
|
|
)
|
|
if form.get("caller") != ALLOWED_CALLER:
|
|
return _html("<h1>Caller refused</h1>", status=400)
|
|
if form.get("suppress") == "1":
|
|
return _html("<h1>Invitation suppressed</h1>")
|
|
self._add_message(
|
|
kind="invitation",
|
|
email=(form.get("email") or "").strip(),
|
|
subject="invitee",
|
|
expired=False,
|
|
)
|
|
return _html('<h1>Invitation recorded</h1><a href="/mailbox">Open mailbox</a>')
|