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
81 lines
3.6 KiB
Python
81 lines
3.6 KiB
Python
"""Tenant isolation suite (AGENTS.backend-packages.md §11).
|
|
|
|
Proves tenant A cannot read, write, update, or delete tenant B's rows and that
|
|
cross-tenant access returns 404 through the API.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from infrasynth.notifications.models import NotificationDispatch
|
|
from infrasynth.scheduler.models import ScheduledTask
|
|
from infrasynth.security.models import Grant
|
|
from infrasynth.tenancy.context import tenant_context
|
|
from infrasynth.tenancy.models import Tenant
|
|
from infrasynth.webhooks.models import OutboundEndpoint
|
|
from infrasynth.workflows.models import Workflow
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
@pytest.fixture
|
|
def other_tenant():
|
|
return Tenant.objects.create(slug="other-ws", name="Other Workspace")
|
|
|
|
|
|
def _seed(tenant):
|
|
ScheduledTask.all_objects.create(
|
|
tenant=tenant, name="task", task_path="tests.helpers.noop_task", schedule_type="manual"
|
|
)
|
|
OutboundEndpoint.all_objects.create(tenant=tenant, name="ep", url="https://x.test/h", secret="s")
|
|
Workflow.all_objects.create(tenant=tenant, slug="wf", name="WF")
|
|
NotificationDispatch.all_objects.create(tenant=tenant, recipient="a@b.c", channel="email", subject="s", body="b")
|
|
|
|
|
|
class TestManagerIsolation:
|
|
def test_tenant_a_cannot_read_tenant_b(self, tenant, other_tenant):
|
|
_seed(tenant)
|
|
with tenant_context(other_tenant):
|
|
assert ScheduledTask.objects.count() == 0
|
|
assert OutboundEndpoint.objects.count() == 0
|
|
assert Workflow.objects.count() == 0
|
|
assert NotificationDispatch.objects.count() == 0
|
|
|
|
def test_tenant_b_cannot_read_tenant_a(self, tenant, other_tenant):
|
|
_seed(other_tenant)
|
|
with tenant_context(tenant):
|
|
assert ScheduledTask.objects.count() == 0
|
|
assert OutboundEndpoint.objects.count() == 0
|
|
|
|
def test_cross_tenant_update_is_invisible(self, tenant, other_tenant):
|
|
_seed(other_tenant)
|
|
with tenant_context(tenant), pytest.raises(ScheduledTask.DoesNotExist):
|
|
ScheduledTask.objects.get(name="task")
|
|
|
|
def test_cross_tenant_delete_is_invisible(self, tenant, other_tenant):
|
|
_seed(other_tenant)
|
|
with tenant_context(tenant):
|
|
assert ScheduledTask.objects.filter(name="task").delete()[0] == 0
|
|
with tenant_context(other_tenant):
|
|
assert ScheduledTask.objects.filter(name="task").exists()
|
|
|
|
def test_grant_scoped(self, tenant, other_tenant, user):
|
|
Grant.all_objects.create(tenant=tenant, user=user, codename="a")
|
|
Grant.all_objects.create(tenant=other_tenant, user=user, codename="b")
|
|
with tenant_context(tenant):
|
|
assert list(Grant.objects.values_list("codename", flat=True)) == ["a"]
|
|
|
|
|
|
class TestCrossTenantApiReturns404:
|
|
def test_other_tenant_task_is_404(self, authenticated_client, tenant, other_tenant):
|
|
task = ScheduledTask.all_objects.create(
|
|
tenant=other_tenant, name="secret", task_path="tests.helpers.noop_task", schedule_type="manual"
|
|
)
|
|
# authenticated_client's context tenant is `tenant`, so B's row is invisible.
|
|
assert authenticated_client.get(f"/api/v1/scheduler/tasks/{task.pk}/").status_code == 404
|
|
assert authenticated_client.delete(f"/api/v1/scheduler/tasks/{task.pk}/").status_code == 404
|
|
|
|
def test_other_tenant_endpoint_is_404(self, authenticated_client, other_tenant):
|
|
endpoint = OutboundEndpoint.all_objects.create(
|
|
tenant=other_tenant, name="secret", url="https://x.test/h", secret="s"
|
|
)
|
|
assert authenticated_client.get(f"/api/v1/webhooks/outbound/endpoints/{endpoint.pk}/").status_code == 404
|