feat(abonnement): signature + mapping des webhooks Frisbii (#90)

src/subscriptions/webhooks.py

@@ -0,0 +1,48 @@
+import hashlib
+import hmac
+from datetime import datetime, timezone
+
+
+def verify_signature(payload: dict, secret: str) -> bool:
+    # Schéma Reepay/Frisbii : HMAC-SHA256(secret, timestamp + id), hex.
+    # À confirmer dans la doc webhooks Frisbii lors de l'intégration.
+    if not secret:
+        return False
+    timestamp = payload.get("timestamp", "")
+    event_id = payload.get("id", "")
+    received = payload.get("signature", "")
+    expected = hmac.new(
+        secret.encode(), (timestamp + event_id).encode(), hashlib.sha256
+    ).hexdigest()
+    return hmac.compare_digest(expected, received)
+
+
+def _in_future(value) -> bool:
+    if not value:
+        return False
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")) > datetime.now(
+            timezone.utc
+        )
+    except ValueError:
+        return False
+
+
+def map_subscription(sub: dict) -> tuple[str, str | None]:
+    """Mappe un objet subscription Frisbii vers (status, current_period_end).
+
+    Noms de champs Reepay (à confirmer) : state, trial_end, expires,
+    is_cancelled, next_period_start.
+    """
+    state = sub.get("state")
+    if state == "expired":
+        return "expired", sub.get("expires")
+    if sub.get("is_cancelled") or state == "cancelled":
+        return "cancelled", sub.get("expires")
+    if _in_future(sub.get("trial_end")):
+        return "trial", sub.get("trial_end")
+    if state == "active":
+        return "active", sub.get("next_period_start")
+    if state == "pending":
+        return "pending", None
+    return (state or "pending"), sub.get("next_period_start")

tests/subscriptions/test_webhooks.py

@@ -0,0 +1,67 @@
+import hashlib
+import hmac
+from datetime import datetime, timedelta, timezone
+
+from src.subscriptions import webhooks
+
+
+def _sign(secret, timestamp, event_id):
+    return hmac.new(
+        secret.encode(), (timestamp + event_id).encode(), hashlib.sha256
+    ).hexdigest()
+
+
+def test_verify_signature_accepts_valid():
+    payload = {"id": "evt_1", "timestamp": "2026-06-25T10:00:00Z"}
+    payload["signature"] = _sign("s3cr3t", payload["timestamp"], payload["id"])
+    assert webhooks.verify_signature(payload, "s3cr3t") is True
+
+
+def test_verify_signature_rejects_tampered():
+    payload = {
+        "id": "evt_1",
+        "timestamp": "2026-06-25T10:00:00Z",
+        "signature": "deadbeef",
+    }
+    assert webhooks.verify_signature(payload, "s3cr3t") is False
+
+
+def test_verify_signature_rejects_without_secret():
+    payload = {"id": "evt_1", "timestamp": "t", "signature": "x"}
+    assert webhooks.verify_signature(payload, "") is False
+
+
+def _future():
+    return (datetime.now(timezone.utc) + timedelta(days=2)).isoformat()
+
+
+def test_map_trial():
+    status, end = webhooks.map_subscription(
+        {"state": "active", "trial_end": _future(), "next_period_start": _future()}
+    )
+    assert status == "trial"
+
+
+def test_map_active():
+    nxt = _future()
+    status, end = webhooks.map_subscription(
+        {"state": "active", "next_period_start": nxt}
+    )
+    assert status == "active"
+    assert end == nxt
+
+
+def test_map_cancelled():
+    exp = _future()
+    status, end = webhooks.map_subscription(
+        {"state": "active", "is_cancelled": True, "expires": exp}
+    )
+    assert status == "cancelled"
+    assert end == exp
+
+
+def test_map_expired():
+    status, end = webhooks.map_subscription(
+        {"state": "expired", "expires": "2020-01-01T00:00:00Z"}
+    )
+    assert status == "expired"