reuse-surface/reuse_surface/hub/webhooks.py

38 lines
1.6 KiB
Python
Raw Normal View History

REUSE-WP-0019-T02: hub recompose staleness tracking + Forgejo webhook reuse_surface/hub/store.py: compose_state table tracking composed_at/stale, updated only on a *forced* recompose (refresh=true, webhook, future scheduled fallback) -- a plain GET still serves current best-effort data but never silently reports itself as freshly composed. reuse_surface/hub/webhooks.py: constant-time HMAC-SHA256 signature verification (fails closed on an empty/unconfigured secret) and path-only push-payload inspection (never parses file content, per design principle 2 -- webhook only decides whether to trigger a pull-based recompose). New endpoint POST /v1/webhooks/forgejo, accepting both X-Forgejo-Signature and X-Gitea-Signature headers since sibling repos migrate independently. GET /v1/federated and POST /v1/federated/compose now share an asyncio.Lock with the webhook so concurrent recompose triggers coalesce instead of overlapping. No separate /v1/recompose route was added -- POST /v1/federated/compose already did that job from earlier hub work; specs/FederationHubAPI.md now documents this explicitly instead of duplicating it. 28 new pytest cases (128 total pass). Live-verified: ran the actual hub service locally and sent a real HMAC-signed webhook over HTTP, confirming the full webhook-to-recompose path, signature rejection, and the irrelevant-path no-op. Does NOT deploy to the live reuse.coulomb.social hub -- that needs a container rebuild/push/k8s rollout, a separate production-deployment action out of scope here without explicit sign-off. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-07 18:24:55 +02:00
from __future__ import annotations
import hashlib
import hmac
from typing import Any
# Forgejo (Gitea-compatible) signs webhook payloads with HMAC-SHA256 over the
# raw request body, sent as a hex digest in X-Forgejo-Signature (Forgejo) or
# X-Gitea-Signature (Gitea) -- both forges use the same scheme during the
# transition, so both header names are accepted.
SIGNATURE_HEADERS = ("X-Forgejo-Signature", "X-Gitea-Signature")
RELEVANT_PATH_PREFIX = "registry/indexes/"
def verify_signature(secret: str, body: bytes, signature: str | None) -> bool:
"""Constant-time HMAC-SHA256 verification. Returns False (never raises)
for a missing/malformed signature or a misconfigured (empty) secret --
callers must treat False as 'reject', not 'skip verification'."""
if not secret or not signature:
return False
expected = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature.strip())
def push_touches_registry_index(payload: dict[str, Any]) -> bool:
"""True if any commit in a Forgejo/Gitea push-event payload adds,
modifies, or removes a path under registry/indexes/. Only inspects
paths -- never parses file content (design principle 2: no
push-parsing of payloads into registry state, the webhook only
triggers a pull-based recompose)."""
for commit in payload.get("commits") or []:
for key in ("added", "modified", "removed"):
for path in commit.get(key) or []:
if path.startswith(RELEVANT_PATH_PREFIX):
return True
return False