cross-tenant-data-leak-tests

| horiz | vert | IDOR | jwt | fk | cache | log
documents       |   X   |   X  |   X  |  X  | -  |   X   |  -
attachments     |   X   |   -  |   X  |  X  | -  |   -   |  -
search_index    |   X   |   -  |   X  |  X  | -  |   -   |  -
audit_log       |   X   |   -  |   -  |  -  | -  |   -   |  X

SELECT rolname, rolsuper, rolbypassrls FROM pg_roles
WHERE rolname = current_user;
-- Expect: rolsuper=f, rolbypassrls=f

def test_get_other_tenant_resource_returns_404(
    self, client, tenant_a_user, tenant_b_resource
):
    client.force_login(tenant_a_user)
    resp = client.get(f"/api/documents/{tenant_b_resource.id}/")
    assert resp.status_code == 404  # not 403 - avoid existence disclosure

def test_list_does_not_include_other_tenant_resources(
    self, client, tenant_a_user, tenant_b_resource
):
    client.force_login(tenant_a_user)
    resp = client.get("/api/documents/")
    assert resp.status_code == 200
    ids = {d["id"] for d in resp.json()["results"]}
    assert tenant_b_resource.id not in ids

def test_tenant_id_in_body_ignored_or_rejected(
    self, client, tenant_a_user, tenant_b
):
    client.force_login(tenant_a_user)
    resp = client.post("/api/documents/", json={
        "tenant_id": str(tenant_b.id),
        "body": "leak"
    })
    if resp.status_code == 201:
        created = resp.json()
        assert created["tenant_id"] != str(tenant_b.id), \
            "Server accepted tenant_id from body - must derive from session"

def test_jwt_for_tenant_a_rejected_on_tenant_b_path(
    self, client, tenant_a_user, tenant_b_resource
):
    token = sign_jwt(tenant_a_user)
    resp = client.get(
        f"/api/documents/{tenant_b_resource.id}/",
        HTTP_AUTHORIZATION=f"Bearer {token}"
    )
    assert resp.status_code in (401, 404)

def test_cannot_create_fk_referencing_other_tenant(
    self, client, tenant_a_user, tenant_b_resource
):
    client.force_login(tenant_a_user)
    # tenant_b_resource exists, but A shouldn't reference it
    resp = client.post("/api/comments/", json={
        "document_id": str(tenant_b_resource.id),
        "body": "comment on other tenant's doc"
    })
    assert resp.status_code in (400, 404)

def test_unique_violation_does_not_disclose_other_tenant_existence(
    self, client, tenant_a_user, tenant_b_resource_with_slug
):
    client.force_login(tenant_a_user)
    # Try to create with same slug as tenant B
    resp = client.post("/api/documents/", json={
        "slug": tenant_b_resource_with_slug.slug,
        "body": "x"
    })
    # Should succeed (RLS scopes the unique check to tenant A)
    # or fail with a non-disclosing error if unique is global
    if resp.status_code in (409, 422):
        assert "tenant_b" not in resp.text.lower()
        assert tenant_b_resource_with_slug.id not in resp.text

def test_storage_presigned_url_path_traversal_denied(
    self, client, tenant_a_user, tenant_b_resource
):
    client.force_login(tenant_a_user)
    presigned = client.get(
        f"/api/documents/{tenant_a_user.tenant_id}/file/"
    ).json()["url"]
    # Modify the URL to point at tenant B's prefix
    leaked_url = presigned.replace(
        str(tenant_a_user.tenant_id),
        str(tenant_b_resource.tenant_id)
    )
    resp = requests.get(leaked_url)
    assert resp.status_code == 403

def test_search_query_must_include_tenant_filter(
    self, opensearch_client, tenant_a, tenant_b_resource
):
    # Direct ES query without tenant filter (simulating leaked path)
    result = opensearch_client.search(
        index="documents",
        body={"query": {"match_all": {}}}
    )
    # Test asserts the API endpoint always adds a tenant filter.
    # The DIRECT search above should not be reachable from any API path.
    # This test ensures no route exists that issues unfiltered search.
    for route in app.url_map.iter_rules():
        if "search" in route.endpoint:
            assert "@tenant_required" in inspect.getsource(
                app.view_functions[route.endpoint]
            )

def test_async_job_reloads_tenant_from_db_not_payload(
    self, tenant_a, tenant_b_resource
):
    # Enqueue a job with a payload pointing at tenant B's resource
    job = enqueue_export(
        resource_id=tenant_b_resource.id,
        # Crafted to spoof - but executor must verify
        tenant_id_claim=tenant_a.id
    )
    result = run_job(job)
    # Executor must derive tenant_id from resource_id, not the payload
    assert result.tenant_id == tenant_b_resource.tenant_id

def test_cache_keys_are_tenant_scoped(self, cache, tenant_a, tenant_b):
    cache.set("user:1", "tenant_a_data", tenant_id=tenant_a.id)
    cache.set("user:1", "tenant_b_data", tenant_id=tenant_b.id)
    assert cache.get("user:1", tenant_id=tenant_a.id) == "tenant_a_data"
    assert cache.get("user:1", tenant_id=tenant_b.id) == "tenant_b_data"

BEGIN;
SET LOCAL ROLE app_user;
SET LOCAL app.tenant_id = '<tenant_a_uuid>';
INSERT INTO documents (id, tenant_id, body)
VALUES (gen_random_uuid(), current_setting('app.tenant_id')::uuid, 'a');

SET LOCAL app.tenant_id = '<tenant_b_uuid>';
SELECT count(*) FROM documents;  -- expect 0

-- Cross-tenant INSERT
INSERT INTO documents (tenant_id, body) VALUES ('<tenant_a_uuid>', 'leak');
-- Expect: ERROR: new row violates row-level security policy for table "documents"
ROLLBACK;

# .github/workflows/tenant-isolation.yml
name: tenant-isolation
on:
  pull_request:
    paths:
      - "**/*.py"
      - "**/migrations/**"
      - ".github/workflows/tenant-isolation.yml"

jobs:
  cross-tenant-tests:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: postgres
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
        ports:
          - 5432:5432
    steps:
      - uses: actions/checkout@v5
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - name: Install
        run: pip install -e ".[test]"
      - name: Create non-superuser role
        env:
          PGPASSWORD: postgres
        run: |
          psql -h localhost -U postgres -d postgres -c "
            CREATE ROLE app_user LOGIN PASSWORD 'app';
          "
      - name: Apply migrations
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost/test
        run: python manage.py migrate
      - name: Run cross-tenant suite
        env:
          DATABASE_URL: postgresql://app_user:app@localhost/test
        run: pytest tests/tenant_isolation/ --tb=short --no-header -v