jcv-dev
551b42eab5
Close the gaps between the documented contract (API-STANDARD, TENANCY, ENTITLEMENTS) and the implementation, and remove committed build artifacts. Security: - verify + process inbound webhooks (HMAC/handler verify, size limit, timestamp tolerance, idempotency via InboundEvent.external_id) - real 2FA login flow (pre-auth challenge; tokens only after verify/recovery) - wire HybridPermission into security/audit views; add API-key rotate and users/<id>/permissions|roles endpoints - tenant-scoped throttling on by default; webhook replay protection - verify MercadoPago webhook signatures - login brute-force guard, configurable password policy, real ALTCHA PoW Correctness: - apply verified billing webhooks idempotently (subscription/entitlement/ invoice/PaymentTransaction); scheduled payment lifecycle jobs - capture audit update diffs automatically; add audit retention purge - working notification retries, per-channel rate limits, log retention - pluggable virus scanner, upload-size limit, pipeline toggle - feature rollout %/environment targeting; settings-driven registrations - workflow guards (instance cap, route depth, self-assignment, clone on re-entry) - wire every previously-dead INFRASYNTH_* setting; drop truly dead ones Delivery: - README + CHANGELOG; CI format check + coverage gate - keep test media out of the tree; untrack .coverage, __pycache__, egg-info, docs/ and invoice artifacts
251 lines
9.7 KiB
Python
251 lines
9.7 KiB
Python
import json
|
|
|
|
import pytest
|
|
|
|
from infrasynth.webhooks.models import (
|
|
InboundEndpoint,
|
|
InboundEvent,
|
|
OutboundDelivery,
|
|
OutboundEndpoint,
|
|
OutboundSubscription,
|
|
)
|
|
from infrasynth.webhooks.signature import sign_payload
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
OUTBOUND_ENDPOINTS_URL = "/api/v1/webhooks/outbound/endpoints/"
|
|
OUTBOUND_SUBSCRIPTIONS_URL = "/api/v1/webhooks/outbound/subscriptions/"
|
|
OUTBOUND_DELIVERIES_URL = "/api/v1/webhooks/outbound/deliveries/"
|
|
INBOUND_ENDPOINTS_URL = "/api/v1/webhooks/inbound/endpoints/"
|
|
INBOUND_EVENTS_URL = "/api/v1/webhooks/inbound/events/"
|
|
|
|
|
|
@pytest.fixture
|
|
def endpoint():
|
|
return OutboundEndpoint.objects.create(
|
|
name="Target",
|
|
url="https://example.com/hook",
|
|
secret="test-secret",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def subscription(endpoint):
|
|
return OutboundSubscription.objects.create(endpoint=endpoint, event_name="evt.test")
|
|
|
|
|
|
@pytest.fixture
|
|
def inbound_endpoint():
|
|
return InboundEndpoint.objects.create(
|
|
name="Stripe",
|
|
slug="stripe",
|
|
source="stripe",
|
|
secret="wh-secret",
|
|
handler="infrasynth.webhooks.inbound.handlers.HMACInboundHandler",
|
|
)
|
|
|
|
|
|
class TestOutboundEndpointViewSet:
|
|
def test_list(self, authenticated_client, endpoint):
|
|
response = authenticated_client.get(OUTBOUND_ENDPOINTS_URL)
|
|
assert response.status_code == 200
|
|
assert response.data["count"] == 1
|
|
assert response.data["results"][0]["name"] == "Target"
|
|
|
|
def test_requires_auth(self, api_client, endpoint):
|
|
assert api_client.get(OUTBOUND_ENDPOINTS_URL).status_code == 401
|
|
|
|
def test_create(self, authenticated_client):
|
|
response = authenticated_client.post(
|
|
OUTBOUND_ENDPOINTS_URL,
|
|
{
|
|
"name": "New",
|
|
"url": "https://example.com/new",
|
|
"secret": "s",
|
|
},
|
|
format="json",
|
|
)
|
|
assert response.status_code == 201
|
|
assert OutboundEndpoint.objects.filter(name="New").exists()
|
|
|
|
def test_update_and_delete(self, authenticated_client, endpoint):
|
|
response = authenticated_client.patch(
|
|
f"{OUTBOUND_ENDPOINTS_URL}{endpoint.id}/",
|
|
{"name": "Renamed"},
|
|
format="json",
|
|
)
|
|
assert response.status_code == 200
|
|
response = authenticated_client.delete(f"{OUTBOUND_ENDPOINTS_URL}{endpoint.id}/")
|
|
assert response.status_code == 204
|
|
assert not OutboundEndpoint.objects.filter(pk=endpoint.pk).exists()
|
|
|
|
|
|
class TestOutboundSubscriptionViewSet:
|
|
def test_list(self, authenticated_client, subscription):
|
|
response = authenticated_client.get(OUTBOUND_SUBSCRIPTIONS_URL)
|
|
assert response.status_code == 200
|
|
assert response.data["count"] == 1
|
|
assert response.data["results"][0]["event_name"] == "evt.test"
|
|
|
|
def test_list_includes_endpoint_info(self, authenticated_client, subscription):
|
|
response = authenticated_client.get(OUTBOUND_SUBSCRIPTIONS_URL)
|
|
assert response.data["results"][0]["endpoint_info"]["name"] == "Target"
|
|
|
|
def test_create(self, authenticated_client, endpoint):
|
|
response = authenticated_client.post(
|
|
OUTBOUND_SUBSCRIPTIONS_URL,
|
|
{"endpoint": endpoint.id, "event_name": "evt.new"},
|
|
format="json",
|
|
)
|
|
assert response.status_code == 201
|
|
assert OutboundSubscription.objects.filter(event_name="evt.new").exists()
|
|
|
|
def test_requires_auth(self, api_client, subscription):
|
|
assert api_client.get(OUTBOUND_SUBSCRIPTIONS_URL).status_code == 401
|
|
|
|
def test_duplicate_event_rejected(self, authenticated_client, subscription):
|
|
response = authenticated_client.post(
|
|
OUTBOUND_SUBSCRIPTIONS_URL,
|
|
{
|
|
"endpoint": subscription.endpoint_id,
|
|
"event_name": "evt.test",
|
|
},
|
|
format="json",
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
class TestOutboundDeliveryViewSet:
|
|
def test_list(self, authenticated_client, subscription):
|
|
OutboundDelivery.objects.create(subscription=subscription, payload={"a": 1}, status="success")
|
|
response = authenticated_client.get(OUTBOUND_DELIVERIES_URL)
|
|
assert response.status_code == 200
|
|
assert response.data["count"] == 1
|
|
assert response.data["results"][0]["payload"] == {"a": 1}
|
|
|
|
def test_does_not_allow_create(self, authenticated_client):
|
|
assert authenticated_client.post(OUTBOUND_DELIVERIES_URL, {}, format="json").status_code == 405
|
|
|
|
def test_requires_auth(self, api_client):
|
|
assert api_client.get(OUTBOUND_DELIVERIES_URL).status_code == 401
|
|
|
|
|
|
class TestInboundEndpointViewSet:
|
|
def test_list(self, authenticated_client, inbound_endpoint):
|
|
response = authenticated_client.get(INBOUND_ENDPOINTS_URL)
|
|
assert response.status_code == 200
|
|
assert response.data["count"] == 1
|
|
assert response.data["results"][0]["slug"] == "stripe"
|
|
|
|
def test_create(self, authenticated_client):
|
|
response = authenticated_client.post(
|
|
INBOUND_ENDPOINTS_URL,
|
|
{
|
|
"name": "GitHub",
|
|
"slug": "github",
|
|
"source": "github",
|
|
"secret": "s",
|
|
"handler": "helpdesk.webhook_handlers.JiraWebhookHandler",
|
|
},
|
|
format="json",
|
|
)
|
|
assert response.status_code == 201
|
|
assert InboundEndpoint.objects.filter(slug="github").exists()
|
|
|
|
def test_requires_auth(self, api_client, inbound_endpoint):
|
|
assert api_client.get(INBOUND_ENDPOINTS_URL).status_code == 401
|
|
|
|
|
|
class TestInboundEventViewSet:
|
|
def test_list(self, authenticated_client, inbound_endpoint):
|
|
InboundEvent.objects.create(
|
|
endpoint=inbound_endpoint,
|
|
event_type="invoice.paid",
|
|
raw_payload={"id": "inv_1"},
|
|
)
|
|
response = authenticated_client.get(INBOUND_EVENTS_URL)
|
|
assert response.status_code == 200
|
|
assert response.data["count"] == 1
|
|
assert response.data["results"][0]["event_type"] == "invoice.paid"
|
|
|
|
def test_requires_auth(self, api_client, inbound_endpoint):
|
|
assert api_client.get(INBOUND_EVENTS_URL).status_code == 401
|
|
|
|
|
|
class TestInboundReceiveView:
|
|
URL = "/api/v1/webhooks/inbound/receive/{slug}/"
|
|
|
|
def _post(self, client, endpoint, payload, headers=None, *, secret=None, sign=True, event_id=None):
|
|
headers = dict(headers or {})
|
|
if event_id:
|
|
headers["X-Event-Id"] = event_id
|
|
body = json.dumps({"payload": payload, "headers": headers})
|
|
http_headers = {}
|
|
if sign:
|
|
http_headers["HTTP_X_WEBHOOK_SIGNATURE"] = sign_payload(secret or endpoint.secret, body)
|
|
return client.post(
|
|
self.URL.format(slug=endpoint.slug),
|
|
data=body,
|
|
content_type="application/json",
|
|
**http_headers,
|
|
)
|
|
|
|
def test_receive_creates_event(self, api_client, inbound_endpoint):
|
|
response = self._post(api_client, inbound_endpoint, {"type": "invoice.paid"}, {"X-Event-Type": "invoice.paid"})
|
|
assert response.status_code == 201
|
|
event = InboundEvent.objects.get(endpoint=inbound_endpoint)
|
|
assert event.event_type == "invoice.paid"
|
|
assert event.raw_payload == {"type": "invoice.paid"}
|
|
assert event.is_verified is True
|
|
assert event.is_processed is True
|
|
|
|
def test_receive_is_public(self, api_client, inbound_endpoint):
|
|
response = self._post(api_client, inbound_endpoint, {}, {})
|
|
assert response.status_code == 201
|
|
|
|
def test_receive_rejects_bad_signature(self, api_client, inbound_endpoint):
|
|
response = self._post(api_client, inbound_endpoint, {"a": 1}, sign=False)
|
|
assert response.status_code == 401
|
|
assert response.data["code"] == "AUTH_INVALID_SIGNATURE"
|
|
assert not InboundEvent.objects.exists()
|
|
|
|
def test_receive_rejects_wrong_secret(self, api_client, inbound_endpoint):
|
|
response = self._post(api_client, inbound_endpoint, {"a": 1}, secret="wrong")
|
|
assert response.status_code == 401
|
|
assert not InboundEvent.objects.exists()
|
|
|
|
def test_receive_is_idempotent_by_external_id(self, api_client, inbound_endpoint):
|
|
first = self._post(api_client, inbound_endpoint, {"a": 1}, event_id="evt_1")
|
|
second = self._post(api_client, inbound_endpoint, {"a": 1}, event_id="evt_1")
|
|
assert first.status_code == 201
|
|
assert second.status_code == 200
|
|
assert InboundEvent.objects.filter(endpoint=inbound_endpoint, external_id="evt_1").count() == 1
|
|
|
|
def test_receive_unknown_slug_404(self, api_client, inbound_endpoint):
|
|
response = api_client.post(
|
|
self.URL.format(slug="unknown"),
|
|
{"payload": {}, "headers": {}},
|
|
format="json",
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
def test_receive_inactive_endpoint_404(self, api_client, inbound_endpoint):
|
|
inbound_endpoint.is_active = False
|
|
inbound_endpoint.save(update_fields=["is_active"])
|
|
response = self._post(api_client, inbound_endpoint, {}, {})
|
|
assert response.status_code == 404
|
|
|
|
def test_receive_missing_payload_400(self, api_client, inbound_endpoint):
|
|
response = api_client.post(self.URL.format(slug="stripe"), {}, format="json")
|
|
assert response.status_code == 400
|
|
|
|
def test_receive_emits_signal(self, api_client, inbound_endpoint):
|
|
from infrasynth.webhooks.signals import inbound_event_received
|
|
|
|
sent = []
|
|
receiver = lambda **kw: sent.append(kw) # noqa: E731
|
|
inbound_event_received.connect(receiver, weak=False)
|
|
self._post(api_client, inbound_endpoint, {"a": 1}, {})
|
|
assert sent
|
|
assert sent[0]["payload"] == {"a": 1}
|
|
inbound_event_received.disconnect(receiver)
|