"""Burki transfer webhook offline example. Public dummy key/body only. This verifies the current transfer signing contract without generating a real transfer. A production receiver needs independently reviewed HTTP/body limits, organization routing, timestamp policy and transactional durable event handling. """ import hashlib import hmac import json import re import uuid DUMMY_KEY = "1" * 64 # Deliberately public fixture, never a real verification key. NOW = 1_800_000_000 WINDOW = 300 # Illustrative receiver policy, not a Burki-configured guarantee. MAX_BODY = 65_536 # Illustrative receiver limit, not a sender payload guarantee. ATTEMPT = "00000000-0000-4000-8000-000000000001" EVENT_ID = ATTEMPT + ":7" RAW = json.dumps({"id": EVENT_ID, "status": "failed", "type": "transfer_failed"}, sort_keys=True, separators=(",", ":")).encode() EVENT_TYPES = { "transfer_started", "transfer_admitted", "transfer_prepared", "transfer_bridged", "transfer_cancelled", "transfer_failed", "transfer_ended", "transfer_reclaimed", "refer_completed", } def headers(body, timestamp=NOW, event_id=EVENT_ID): digest = hmac.new( DUMMY_KEY.encode("utf-8"), str(timestamp).encode("ascii") + b"." + body, hashlib.sha256, ).hexdigest() return { "X-Burki-Timestamp": str(timestamp), "X-Burki-Signature": "v1=" + digest, "X-Burki-Event-Id": event_id, } def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("duplicate_json_key") result[key] = value return result def authenticate(body, incoming, now=NOW): if not isinstance(body, bytes) or len(body) > MAX_BODY: raise ValueError("invalid_body_size") timestamp = incoming.get("X-Burki-Timestamp", "") signature = incoming.get("X-Burki-Signature", "") if not re.fullmatch(r"[0-9]{1,12}", timestamp): raise ValueError("invalid_timestamp") if abs(now - int(timestamp)) > WINDOW: raise ValueError("stale_or_future_timestamp") if not re.fullmatch(r"v1=[0-9a-f]{64}", signature): raise ValueError("invalid_signature_format") expected = hmac.new( DUMMY_KEY.encode("utf-8"), timestamp.encode("ascii") + b"." + body, hashlib.sha256, ).hexdigest() if not hmac.compare_digest(expected, signature[3:]): raise ValueError("signature_mismatch") # Parse only after verifying the exact received bytes. The ID header alone # is not included in the signature, so bind it to the authenticated body ID. payload = json.loads(body, object_pairs_hook=unique_object) event_id = payload.get("id") if isinstance(payload, dict) else None if not isinstance(event_id, str) or not event_id or event_id != incoming.get("X-Burki-Event-Id"): raise ValueError("event_id_mismatch") # Current sender constructs attempt UUID + ':' + positive event-row ID. match = re.fullmatch(r"([0-9a-f-]{36}):([1-9][0-9]{0,18})", event_id) if not match: raise ValueError("invalid_event_id") try: canonical_attempt = str(uuid.UUID(match[1])) except ValueError: raise ValueError("invalid_event_id") from None if canonical_attempt != match[1]: raise ValueError("invalid_event_id") if not isinstance(payload.get("type"), str) or payload["type"] not in EVENT_TYPES: raise ValueError("unsupported_event_type") status = payload.get("status") if not isinstance(status, str) or not 1 <= len(status) <= 64: raise ValueError("invalid_status") if "attempt_id" in payload and payload["attempt_id"] != canonical_attempt: raise ValueError("attempt_id_mismatch") # Consume an explicit field projection. Actual sender has additional fields; # receiving them does not authorize treating arbitrary details as commands. return {"id": event_id, "type": payload["type"], "status": status} def rejected(body, incoming, reason, now=NOW): try: authenticate(body, incoming, now) except ValueError as error: assert str(error) == reason, (str(error), reason) return raise AssertionError("Expected rejection: " + reason) def main(): checks = [] payload = authenticate(RAW, headers(RAW)) assert payload["type"] == "transfer_failed" and payload["status"] == "failed" checks.append("authentic failure remains a failure, not a bridged success") rejected(RAW + b" ", headers(RAW), "signature_mismatch") checks.append("whitespace/body change fails original signature") rewritten = json.dumps(json.loads(RAW)).encode() rejected(rewritten, headers(RAW), "signature_mismatch") checks.append("parse/reserialize before verification fails original signature") rejected(RAW, headers(RAW, NOW - WINDOW - 1), "stale_or_future_timestamp") rejected(RAW, headers(RAW, NOW + WINDOW + 1), "stale_or_future_timestamp") checks.append("stale and excessively future timestamps rejected") bad = headers(RAW) bad["X-Burki-Timestamp"] = str(NOW - 1) rejected(RAW, bad, "signature_mismatch") checks.append("changing timestamp without resigning fails") rejected(RAW, headers(RAW, event_id="attacker-changed-header"), "event_id_mismatch") checks.append("unsigned ID header cannot override signed body ID") bad = headers(RAW) bad["X-Burki-Signature"] = "v1=" + "0" * 64 rejected(RAW, bad, "signature_mismatch") checks.append("wrong signature rejected") # UTF-8 text key and hex-decoded key are not equivalent. wrong = hmac.new(bytes.fromhex(DUMMY_KEY), str(NOW).encode() + b"." + RAW, hashlib.sha256).hexdigest() bad = headers(RAW) bad["X-Burki-Signature"] = "v1=" + wrong rejected(RAW, bad, "signature_mismatch") checks.append("hex-decoded key bytes are wrong for current sender contract") malformed = json.dumps({"id": "not-an-event", "type": "transfer_failed", "status": "failed"}).encode() rejected(malformed, headers(malformed, event_id="not-an-event"), "invalid_event_id") checks.append("malformed signed event identity rejected") unknown = RAW.replace(b"transfer_failed", b"execute_arbitrary_command") rejected(unknown, headers(unknown), "unsupported_event_type") checks.append("unsupported signed event type rejected") wrong_type = json.dumps({"id": EVENT_ID, "type": ["transfer_failed"], "status": "failed"}).encode() rejected(wrong_type, headers(wrong_type), "unsupported_event_type") bad_status = json.dumps({"id": EVENT_ID, "type": "transfer_failed", "status": None}).encode() rejected(bad_status, headers(bad_status), "invalid_status") checks.append("malformed signed field types rejected") wrong_attempt = json.dumps({"id": EVENT_ID, "type": "transfer_failed", "status": "failed", "attempt_id": "different-attempt"}).encode() rejected(wrong_attempt, headers(wrong_attempt), "attempt_id_mismatch") checks.append("optional attempt identity must match signed event identity") duplicate = RAW[:-1] + b',"status":"bridged"}' rejected(duplicate, headers(duplicate), "duplicate_json_key") checks.append("duplicate signed JSON fields rejected") oversized = b" " * (MAX_BODY + 1) rejected(oversized, headers(oversized), "invalid_body_size") checks.append("illustrative receiver body limit enforced") # Demonstrate only. A set does not provide multi-process durable deduplication. seen = set() applied = 0 for timestamp in (NOW, NOW + 1): verified = authenticate(RAW, headers(RAW, timestamp), now=NOW + 1) trusted_organization = "example-org" # From receiver route/key, not body. scoped_id = (trusted_organization, verified["id"]) if scoped_id not in seen: seen.add(scoped_id) applied += 1 assert applied == 1 checks.append("freshly signed retry retains same event identity; demo applies once") print(json.dumps({"status": "passed", "offline_only": True, "checks": checks, "production_delivery_or_transfer_proven": False}, indent=2)) if __name__ == "__main__": main()