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
78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
"""TenantMiddleware resolution and rejection (TENANCY.md §3.2)."""
|
|
|
|
import pytest
|
|
from django.contrib.auth.models import AnonymousUser
|
|
from django.http import HttpResponse
|
|
from django.test import RequestFactory
|
|
|
|
from infrasynth.tenancy.context import get_current_tenant
|
|
from infrasynth.tenancy.middleware import TenantMiddleware
|
|
from infrasynth.tenancy.models import Tenant, TenantMembership
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
TENANCY_ON = {
|
|
"ENABLED": True,
|
|
"REQUIRE_TENANT_BY_DEFAULT": True,
|
|
"TENANT_CLAIM": "tenant",
|
|
"TENANT_ALLOWLIST_PATHS": ["/api/v1/auth/login/", "/api/v1/auth/select-workspace/"],
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def tenancy_enabled(settings):
|
|
settings.INFRASYNTH_TENANCY = TENANCY_ON
|
|
|
|
|
|
def _call(request):
|
|
return TenantMiddleware(lambda r: HttpResponse("ok"))(request)
|
|
|
|
|
|
class TestTenantMiddleware:
|
|
def test_binds_tenant_from_token_claim(self, user, tenant):
|
|
from infrasynth.tenancy.context import tenant_context
|
|
|
|
req = RequestFactory().get("/api/v1/scheduler/tasks/")
|
|
req.user = user
|
|
req.auth = {"tenant": str(tenant.pk)}
|
|
with tenant_context(None):
|
|
resp = _call(req)
|
|
assert resp.status_code == 200
|
|
assert req.tenant == tenant
|
|
assert get_current_tenant() is None # previous context restored
|
|
|
|
def test_single_membership_auto_selected(self, user, tenant):
|
|
req = RequestFactory().get("/api/v1/scheduler/tasks/")
|
|
req.user = user
|
|
resp = _call(req)
|
|
assert resp.status_code == 200
|
|
assert req.tenant == tenant
|
|
|
|
def test_rejects_when_no_tenant_and_multiple_memberships(self, user, tenant):
|
|
other = Tenant.objects.create(slug="second", name="Second")
|
|
TenantMembership.objects.create(tenant=other, user=user, is_active=True)
|
|
req = RequestFactory().get("/api/v1/scheduler/tasks/")
|
|
req.user = user
|
|
resp = _call(req)
|
|
assert resp.status_code == 403
|
|
assert b"AUTH_TENANT_REQUIRED" in resp.content
|
|
|
|
def test_rejects_revoked_membership_immediately(self, user, tenant):
|
|
TenantMembership.objects.filter(tenant=tenant, user=user).update(is_active=False)
|
|
req = RequestFactory().get("/api/v1/scheduler/tasks/")
|
|
req.user = user
|
|
req.auth = {"tenant": str(tenant.pk)}
|
|
resp = _call(req)
|
|
assert resp.status_code == 403
|
|
assert b"AUTH_MEMBERSHIP_REVOKED" in resp.content
|
|
|
|
def test_allowlisted_path_passes_without_tenant(self):
|
|
req = RequestFactory().get("/api/v1/auth/login/")
|
|
req.user = AnonymousUser()
|
|
assert _call(req).status_code == 200
|
|
|
|
def test_unauthenticated_request_is_not_403ed(self):
|
|
req = RequestFactory().get("/api/v1/scheduler/tasks/")
|
|
req.user = AnonymousUser()
|
|
# Auth classes own the 401; middleware must not turn it into a 403.
|
|
assert _call(req).status_code == 200
|