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
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
"""Peer authentication for the mode contract (CONTRACT.md section 7).
|
|
|
|
HMAC-SHA256 over TIMESTAMP + "." + BODY with a pairwise key, sent in three
|
|
headers. Standard library only, so the gateway can use it without the rest
|
|
of `mas_modes`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import time
|
|
from typing import Mapping, Optional
|
|
|
|
PEER = "MAS-Modes-Peer"
|
|
TIMESTAMP = "MAS-Modes-Timestamp"
|
|
SIGNATURE = "MAS-Modes-Signature"
|
|
WINDOW = 300 # seconds
|
|
|
|
|
|
def signature(key: bytes, timestamp: int, body: bytes) -> str:
|
|
return "v1=" + hmac.new(key, str(timestamp).encode() + b"." + body, hashlib.sha256).hexdigest()
|
|
|
|
|
|
def sign(body: bytes, peer: str, key: bytes, timestamp: Optional[int] = None) -> dict:
|
|
"""The headers that authenticate `body` as sent by `peer`."""
|
|
ts = int(time.time()) if timestamp is None else timestamp
|
|
return {PEER: peer, TIMESTAMP: str(ts), SIGNATURE: signature(key, ts, body)}
|
|
|
|
|
|
class AuthError(Exception):
|
|
def __init__(self, code: str, status: int = 401):
|
|
super().__init__(code)
|
|
self.code, self.status = code, status
|
|
|
|
|
|
def verify(headers: Mapping[str, str], body: bytes, keys: Mapping[str, bytes],
|
|
now: Optional[float] = None) -> str:
|
|
"""Return the authenticated peer id, or raise AuthError."""
|
|
get = {k.lower(): v for k, v in headers.items()}.get
|
|
peer, ts, sig = get(PEER.lower()), get(TIMESTAMP.lower()), get(SIGNATURE.lower())
|
|
key = keys.get(peer) if peer else None
|
|
if key is None or not ts or not sig:
|
|
raise AuthError("unauthenticated-peer")
|
|
try:
|
|
ts_i = int(ts)
|
|
except ValueError:
|
|
raise AuthError("unauthenticated-peer") from None
|
|
if not hmac.compare_digest(signature(key, ts_i, body), sig):
|
|
raise AuthError("unauthenticated-peer")
|
|
if abs((time.time() if now is None else now) - ts_i) > WINDOW:
|
|
raise AuthError("replayed")
|
|
return peer
|