File size: 5,550 Bytes
add315b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | #!/usr/bin/env python3
"""Regression checks for the identity-bound submit signature (2026-09-29,
audit item 12e). Offline: no network, no mail, no pings, no files written.
Run from the repo root:
.venv/bin/python intake/scripts/test_hmac_identity.py
Covers: a v2 signature covers the timestamp, the ORCID, name and test-tier
headers and the body, so changing, adding or removing any of them fails; the
submit endpoint refuses the body-only v1 form unless the deploy-window switch
is on; the other signed endpoints keep accepting v1 (their identity is inside
the signed body); header values with line breaks or non-ASCII are refused; the
smoke scripts sign v2 in the same layout.
"""
from __future__ import annotations
import hashlib
import hmac
import inspect
import os
import sys
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
os.environ["INTAKE_HMAC_SECRET"] = "test-secret-not-used-anywhere"
failures: list[str] = []
def check(cond: bool, name: str) -> None:
print((" ok " if cond else " FAIL ") + name)
if not cond:
failures.append(name)
from fastapi import HTTPException # noqa: E402
from starlette.datastructures import Headers # noqa: E402
from intake import intake_server as srv # noqa: E402
srv.HMAC_SECRET = b"test-secret-not-used-anywhere"
KEY = srv.HMAC_SECRET
BODY = b"--b\r\nContent-Disposition: form-data; name=\"orcid\"\r\n\r\n0000-0002-1825-0097\r\n--b--\r\n"
ORCID, NAME, TIER = "0000-0002-1825-0097", "Ada%20Lovelace", "t3"
class Req:
def __init__(self, headers: dict):
self.headers = Headers(headers=headers)
def v2(ts: str, orcid: str, name: str, tier: str, body: bytes) -> str:
msg = ("\n".join(["icsac-v2", ts, orcid, name, tier]) + "\n").encode() + body
return "v2=" + hmac.new(KEY, msg, hashlib.sha256).hexdigest()
def v1(ts: str, body: bytes) -> str:
return "sha256=" + hmac.new(KEY, f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
def hdrs(ts, sig, orcid=ORCID, name=NAME, tier=TIER):
h = {"x-icsac-timestamp": ts, "x-icsac-signature": sig}
if orcid is not None:
h["x-icsac-auth-orcid"] = orcid
if name is not None:
h["x-icsac-auth-name"] = name
if tier is not None:
h["x-icsac-test-tier"] = tier
return h
def outcome(headers: dict, body: bytes = BODY, bind: bool = True) -> str:
try:
return srv._verify_hmac(Req(headers), body, bind_identity=bind)
except HTTPException as exc:
return f"{exc.status_code}:{exc.detail}"
ts = str(int(time.time()))
good = v2(ts, ORCID, NAME, TIER, BODY)
check(outcome(hdrs(ts, good)) == "v2", "v2 with all three identity headers verifies")
check(outcome(hdrs(ts, good, orcid="0000-0001-5109-3700")) == "401:bad signature", "another ORCID under the same signature fails")
check(outcome(hdrs(ts, good, name="Someone%20Else")) == "401:bad signature", "another name under the same signature fails")
check(outcome(hdrs(ts, good, tier=None)) == "401:bad signature", "dropping the tier header fails")
check(outcome(hdrs(ts, good, tier="t2")) == "401:bad signature", "changing the tier header fails")
check(outcome(hdrs(ts, good), body=BODY + b"x") == "401:bad signature", "a changed body fails")
no_tier = v2(ts, ORCID, NAME, "", BODY)
check(outcome(hdrs(ts, no_tier, tier=None)) == "v2", "a production request (no tier header) verifies with the tier signed empty")
check(outcome(hdrs(ts, no_tier, tier="t3")) == "401:bad signature", "adding a tier header to a production signature fails")
old = str(int(time.time()) - 301)
check(outcome(hdrs(old, v2(old, ORCID, NAME, TIER, BODY))) == "401:stale signature", "a v2 signature past the skew window fails")
check(outcome(hdrs(ts, v2(ts, ORCID, "A%0AB", TIER, BODY), name="AéB")) == "401:bad identity header", "a non-ASCII identity header is refused")
srv.SUBMIT_SIG_MODE = "v2"
check(outcome(hdrs(ts, v1(ts, BODY))) == "401:signature does not cover the identity headers", "submit refuses v1 by default")
srv.SUBMIT_SIG_MODE = "both"
check(outcome(hdrs(ts, v1(ts, BODY))) == "v1", "submit accepts v1 in the deploy window (INTAKE_SUBMIT_SIG=both)")
check(outcome(hdrs(ts, good)) == "v2", "submit accepts v2 in the deploy window")
srv.SUBMIT_SIG_MODE = "v2"
check(outcome(hdrs(ts, v1(ts, BODY), orcid=None, name=None, tier=None), bind=False) == "v1", "approve and sponsor-logo keep v1 (identity is in their signed body)")
check(outcome({"x-icsac-timestamp": ts, "x-icsac-signature": "md5=abc"}) == "401:missing signature", "an unknown signature scheme is refused")
check(outcome(hdrs(ts, "v2=" + "0" * 64)) == "401:bad signature", "a forged v2 value fails")
src = inspect.getsource(srv.api_submit)
check("bind_identity=True" in src, "the submit endpoint binds identity")
for ep in (getattr(srv, "api_approve_record", None), srv.api_sponsor_logo):
if ep is not None:
check("bind_identity=True" not in inspect.getsource(ep), f"{ep.__name__} does not demand identity headers")
check(srv.SUBMIT_SIG_MODE == "v2" and os.environ.get("INTAKE_SUBMIT_SIG", "v2") == "v2", "the default mode is v2")
for script in ("smoke_test_tier_2.sh", "smoke_test_test_mode.sh"):
text = (ROOT / "intake" / "scripts" / script).read_text()
check('printf "icsac-v2\\n%s\\n%s\\n%s\\n%s\\n"' in text and "x-icsac-signature: v2=$SIG" in text,
f"{script} signs v2 in the server's layout")
print()
if failures:
print(f"FAILED: {len(failures)}")
for f in failures:
print(" - " + f)
sys.exit(1)
print("all checks passed")
|