tegwick
bdf200a600
CONTRACT.md section 7 (founder-reviewed): pairwise HMAC-SHA256 over TIMESTAMP.BODY in MAS-Modes-* headers, 300 s window; master commands only from the configured master; statuses only at the handoff's tokenized URL from its peer; primed webhooks must name a known peer. mas_modes.signing, signed contract.post, authenticating json_server, EmailMode/stubs/gateway signing. conformance/auth.json vectors + 7 authorization flows. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 5787@bnt-lap001 Assistant-Session: 5db26b82-138b-40cc-9856-4d5eb0f752fd
83 lines
3.5 KiB
Python
83 lines
3.5 KiB
Python
"""Peer authentication (CONTRACT.md section 7, MAS-WP-0011)."""
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from mas_modes import _load
|
|
from mas_modes.contract import message, post
|
|
from mas_modes.email_mode import EmailMode, RoleConfig
|
|
from mas_modes.harness import ROOT, Harness
|
|
|
|
STATUTE = ROOT / "mas-modes-v0.1-spec/conformance/statute.json"
|
|
|
|
|
|
def test_vectors_match_an_independent_hmac():
|
|
for v in _load("conformance/auth.json")["vectors"]:
|
|
mac = hmac.new(bytes.fromhex(v["key_hex"]), f"{v['timestamp']}.".encode() + v["body"].encode(),
|
|
hashlib.sha256).hexdigest()
|
|
assert v["signature"] == "v1=" + mac
|
|
|
|
|
|
def test_signed_events_required_when_configured():
|
|
h = Harness(STATUTE, {"ticket": {"role": "master"}}, sign_events=True)
|
|
try:
|
|
assert h.mail("ticket", "hello").status == 201 # the gateway signs
|
|
h.signer = None
|
|
d = h.mail("ticket", "unsigned")
|
|
assert d.status == 401 and d.response["error"] == "unauthenticated-peer"
|
|
h.signer = ("gateway", b"\0" * 32)
|
|
assert h.mail("ticket", "bad key").status == 401
|
|
finally:
|
|
h.close()
|
|
|
|
|
|
def test_unsigned_events_only_without_event_peer():
|
|
h = Harness(STATUTE, {"ticket": {"role": "master"}})
|
|
try:
|
|
assert h.mail("ticket", "loopback, unsigned").status == 201
|
|
h.signer = ("cms-1", h.keys["cms-1"]) # signed, but not by a gateway
|
|
assert h.mail("ticket", "x").status == 403
|
|
finally:
|
|
h.close()
|
|
|
|
|
|
def test_stubs_accept_only_email():
|
|
h = Harness(STATUTE, {"case": {"role": "slave"}})
|
|
try:
|
|
req = message("handoff.request", {"address": "a@b", "key": "a"}, handoff_id="h", mode="callback",
|
|
details={}, status_url="http://127.0.0.1:9/x")
|
|
url = h.email.roles["case"].master_url
|
|
assert post(url, message("case.intake", intake_ref="i", resource_type="case", event={}))[0] == 401
|
|
assert post(url, message("case.intake", intake_ref="i", resource_type="case", event={}),
|
|
("cms-1", h.keys["cms-1"]))[0] == 401 # the master's key is not email's identity
|
|
mode_url = h.callback_url
|
|
assert post(mode_url, req)[0] == 401
|
|
assert post(mode_url, req, ("email", b"\1" * 32))[0] == 401
|
|
assert post(mode_url, req, ("email", h.keys["callback-1"]))[0] == 202
|
|
finally:
|
|
h.close()
|
|
|
|
|
|
def test_email_refuses_config_without_keys():
|
|
with pytest.raises(ValueError, match="no key"):
|
|
EmailMode({"case": RoleConfig("slave", master_url="http://x", master_peer="cms-1")}, keys={})
|
|
with pytest.raises(ValueError, match="no key"):
|
|
EmailMode({"case": RoleConfig("slave", master_url="http://x", master_peer="cms-1",
|
|
transfers={"webhook": "http://y", "peer": "nobody"})},
|
|
keys={"cms-1": b"k" * 32})
|
|
|
|
|
|
def test_status_urls_are_unguessable_and_distinct():
|
|
h = Harness(STATUTE, {"ticket": {"role": "master", "fields": ("phone", "time", "scenario")}},
|
|
scenario="silent")
|
|
try:
|
|
h.mail("ticket!callback", "Phone: 1\nTime: 2")
|
|
h.mail("ticket!callback", "Phone: 1\nTime: 2")
|
|
urls = [m["status_url"] for m in h.callback.received]
|
|
tokens = [u.rsplit("/", 1)[1] for u in urls]
|
|
assert len(set(tokens)) == 2 and all(len(t) >= 22 for t in tokens) # >= 128 bits
|
|
assert json.dumps(h.callback.received).count(tokens[0]) == 1 # not echoed elsewhere
|
|
finally:
|
|
h.close()
|