Submit signature v2: the signature covers the verified ORCID, name and test-tier headers as well as the body, so a signed request cannot be replayed under another identity or tier; the body-only form is refused on submit unless the deploy-window switch is on; smoke scripts sign v2; regression suite
9a4a7ba | #!/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") | |