From 8bbb37006ec0d8deeac759d5326977a3069b7587 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 18:22:58 +1100 Subject: [PATCH] feat(tenancy): route each signed-in request to its tenant's schema (M2b, #93) A request with a bearer session runs in schema tenant_: its file registry, audit log and uploaded tables live there, so another tenant cannot list, read, delete or collide with them, and duplicate detection becomes per tenant. Requests without a session keep the API-key gate and the default schemas. An invalid session is 401, never a fallback. The schema is a context variable set by the async tenant_scope dependency; services read it through uploads_schema()/te_schema(). Tenant schemas get the per-tenant migrations (0004 is shared-only) when created by scripts/create_tenant_admin.py and at every startup. T&E mode for a tenant returns 409 until per-tenant T&E tables (#99). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019YPoXzBSj89ZHgw7nfTmbB --- alembic/versions/0004_control_schema.py | 11 +- api/auth.py | 30 +++- api/db.py | 22 ++- api/routers/audit_routes.py | 10 +- api/routers/csv_routes.py | 22 +-- api/routers/te_routes.py | 11 +- api/services/audit.py | 4 +- api/services/dynamic_loader.py | 3 +- api/services/te_loader.py | 25 ++-- api/tenancy.py | 68 ++++++++++ scripts/create_tenant_admin.py | 7 +- scripts/lint.sh | 2 + tests/test_tenant_isolation.py | 173 ++++++++++++++++++++++++ 13 files changed, 347 insertions(+), 41 deletions(-) create mode 100644 api/tenancy.py create mode 100644 tests/test_tenant_isolation.py diff --git a/alembic/versions/0004_control_schema.py b/alembic/versions/0004_control_schema.py index 78b5a3c..e8f6c17 100644 --- a/alembic/versions/0004_control_schema.py +++ b/alembic/versions/0004_control_schema.py @@ -8,7 +8,7 @@ Create Date: 2026-10-26 """ -from alembic import op +from alembic import context, op from api.config import settings @@ -22,7 +22,14 @@ def _schema() -> str: return op.get_bind().dialect.identifier_preparer.quote(settings.CONTROL_SCHEMA) +def _tenant_run() -> bool: + # Shared objects are created once, not in every tenant schema (api/tenancy.py). + return context.config.attributes.get("scope") == "tenant" + + def upgrade() -> None: + if _tenant_run(): + return s = _schema() op.execute(f"CREATE SCHEMA IF NOT EXISTS {s}") op.execute(f""" @@ -59,4 +66,6 @@ def upgrade() -> None: def downgrade() -> None: + if _tenant_run(): + return op.execute(f"DROP SCHEMA IF EXISTS {_schema()} CASCADE") diff --git a/api/auth.py b/api/auth.py index d164bf1..489909f 100644 --- a/api/auth.py +++ b/api/auth.py @@ -12,12 +12,15 @@ """ import secrets +from collections.abc import AsyncIterator from fastapi import Depends, Header, HTTPException +from starlette.concurrency import run_in_threadpool from api.config import settings from api.db import Conn from api.services.accounts import session_for +from api.tenancy import schema_for, scoped def require_api_key(x_api_key: str = Header(default="")) -> None: @@ -39,11 +42,34 @@ def bearer_token(authorization: str = Header(default="")) -> str: def require_session(token: str = Depends(bearer_token)) -> dict: """The caller's live session (user, tenant, role), or 401 (M2a, #92). - Not yet applied to the data routers; tenant routing (#93) and the role - matrix (#94) switch them over. + Data routers use ``tenant_scope``, which also routes to the tenant's schema. """ + return _live_session(token) + + +def _live_session(token: str) -> dict: with Conn() as conn, conn.cursor() as cur: session = session_for(cur, token) if session is None: raise HTTPException(status_code=401, detail="Session expired or invalid; sign in again") return session + + +async def tenant_scope(authorization: str = Header(default=""), + x_api_key: str = Header(default="")) -> AsyncIterator[dict | None]: + """Route the request to one tenant's schema (M2b, #93). + + With ``Authorization: Bearer `` the request runs in the signed-in + tenant's schema and yields its session. Without one, the shared API key + applies and the request keeps the deployment's default schemas. + + Async on purpose: the schema is a context variable, and only an async + dependency sets it in the context the endpoint's worker thread copies. + """ + if not authorization: + require_api_key(x_api_key) + yield None + return + session = await run_in_threadpool(_live_session, bearer_token(authorization)) + with scoped(schema_for(session["tenant"])): + yield session diff --git a/api/db.py b/api/db.py index f681d2d..9afca94 100644 --- a/api/db.py +++ b/api/db.py @@ -159,8 +159,7 @@ def bootstrap() -> None: from alembic import command from alembic.config import Config - repo_root = Path(__file__).resolve().parent.parent - alembic_ini = repo_root / "alembic.ini" + alembic_ini = alembic_ini_path() if not alembic_ini.exists(): logger.warning( @@ -177,9 +176,28 @@ def bootstrap() -> None: settings.PG_USER, settings.PG_HOST, settings.PG_PORT, settings.PG_DATABASE) command.upgrade(cfg, "head") + _migrate_existing_tenants() logger.info("Alembic migrations complete") +def alembic_ini_path() -> Path: + return Path(__file__).resolve().parent.parent / "alembic.ini" + + +def _migrate_existing_tenants() -> None: + """Bring every tenant schema to head too (decision 6: migrations run across N schemas).""" + from api.tenancy import migrate_tenant + + with Conn() as conn, conn.cursor() as cur: + cur.execute(sql.SQL("SELECT slug FROM {}.tenants ORDER BY id").format( + sql.Identifier(settings.CONTROL_SCHEMA))) + slugs = [r[0] for r in cur.fetchall()] + for slug in slugs: + migrate_tenant(slug) + if slugs: + logger.info("Migrated %d tenant schema(s)", len(slugs)) + + def _bootstrap_fallback_schema() -> None: """Idempotent CREATE TABLE IF NOT EXISTS bootstrap used when Alembic isn't wired up (see bootstrap() above). Mirrors the columns every INSERT/SELECT diff --git a/api/routers/audit_routes.py b/api/routers/audit_routes.py index 7f58ab1..e00acde 100644 --- a/api/routers/audit_routes.py +++ b/api/routers/audit_routes.py @@ -12,14 +12,14 @@ from fastapi import APIRouter, Depends from psycopg2 import sql -from api.auth import require_api_key -from api.config import settings +from api.auth import tenant_scope +from api.tenancy import uploads_schema from api.db import Conn router = APIRouter( prefix="/api/audit", tags=["audit"], - dependencies=[Depends(require_api_key)], + dependencies=[Depends(tenant_scope)], ) @@ -31,7 +31,7 @@ def audit_log(limit: int = 50, offset: int = 0) -> dict: with conn.cursor() as cur: cur.execute( sql.SQL("SELECT COUNT(*) FROM {}.audit_log").format( - sql.Identifier(settings.UPLOADS_SCHEMA) + sql.Identifier(uploads_schema()) ) ) total = cur.fetchone()[0] @@ -39,7 +39,7 @@ def audit_log(limit: int = 50, offset: int = 0) -> dict: sql.SQL( "SELECT id, action, file_id, file_name, table_name, mode, performed_at " "FROM {}.audit_log ORDER BY performed_at DESC LIMIT %s OFFSET %s" - ).format(sql.Identifier(settings.UPLOADS_SCHEMA)), + ).format(sql.Identifier(uploads_schema())), (limit, offset), ) entries = [ diff --git a/api/routers/csv_routes.py b/api/routers/csv_routes.py index af35666..8498a22 100644 --- a/api/routers/csv_routes.py +++ b/api/routers/csv_routes.py @@ -8,8 +8,9 @@ from psycopg2 import sql from pydantic import BaseModel, Field -from api.auth import require_api_key +from api.auth import tenant_scope from api.config import settings +from api.tenancy import current_schema, uploads_schema from api.db import Conn from api.services.audit import coerce_file_id, record_audit, write_audit from api.services.csv_parse import build_preview @@ -23,7 +24,7 @@ router = APIRouter( prefix="/api/csv", tags=["csv"], - dependencies=[Depends(require_api_key)], + dependencies=[Depends(tenant_scope)], ) @@ -71,6 +72,9 @@ def upload(req: UploadRequest) -> dict: if len(req.content.encode("utf-8")) > settings.MAX_UPLOAD_BYTES: raise HTTPException(413, "File too large") if req.mode == "te": + if current_schema() is not None: + raise HTTPException(409, "T&E mode isn't available for tenants yet; it arrives with " + "per-tenant T&E tables (#99). Use dynamic mode.") if not req.targetTable: raise HTTPException(422, "targetTable is required when mode is 'te'") result = upload_te(req.fileName, req.content, req.targetTable) @@ -103,7 +107,7 @@ def list_files() -> list[dict]: "SELECT id, file_name, table_name, mode, row_count, column_names, created_at, " "total_rows, duplicate_rows_skipped, failed_rows " "FROM {}.csv_files ORDER BY created_at DESC, id DESC" - ).format(sql.Identifier(settings.UPLOADS_SCHEMA)) + ).format(sql.Identifier(uploads_schema())) ) return [ { @@ -132,7 +136,7 @@ def table_rows(table_name: str, limit: int = 50) -> dict: # Only serve tables that are registered uploads cur.execute( sql.SQL("SELECT 1 FROM {}.csv_files WHERE table_name = %s").format( - sql.Identifier(settings.UPLOADS_SCHEMA) + sql.Identifier(uploads_schema()) ), (table_name,), ) @@ -140,7 +144,7 @@ def table_rows(table_name: str, limit: int = 50) -> dict: raise HTTPException(404, "Table not found") cur.execute( sql.SQL("SELECT * FROM {}.{} ORDER BY _id LIMIT %s").format( - sql.Identifier(settings.UPLOADS_SCHEMA), sql.Identifier(table_name) + sql.Identifier(uploads_schema()), sql.Identifier(table_name) ), (limit,), ) @@ -167,7 +171,7 @@ def delete_file(file_id: int) -> dict: cur.execute( sql.SQL( "SELECT table_name, mode, file_name FROM {}.csv_files WHERE id = %s" - ).format(sql.Identifier(settings.UPLOADS_SCHEMA)), + ).format(sql.Identifier(uploads_schema())), (file_id,), ) row = cur.fetchone() @@ -181,18 +185,18 @@ def delete_file(file_id: int) -> dict: SELECT 1 FROM information_schema.tables WHERE table_schema = %s AND table_name = %s """, - (settings.UPLOADS_SCHEMA, table_name), + (uploads_schema(), table_name), ) if cur.fetchone() is not None: cur.execute( sql.SQL("DROP TABLE IF EXISTS {}.{}").format( - sql.Identifier(settings.UPLOADS_SCHEMA), + sql.Identifier(uploads_schema()), sql.Identifier(table_name), ) ) cur.execute( sql.SQL("DELETE FROM {}.csv_files WHERE id = %s").format( - sql.Identifier(settings.UPLOADS_SCHEMA) + sql.Identifier(uploads_schema()) ), (file_id,), ) diff --git a/api/routers/te_routes.py b/api/routers/te_routes.py index 1c397cb..edbd0af 100644 --- a/api/routers/te_routes.py +++ b/api/routers/te_routes.py @@ -2,15 +2,16 @@ from fastapi import APIRouter, Depends -from api.auth import require_api_key -from api.config import TE_TABLES, settings +from api.auth import tenant_scope +from api.config import TE_TABLES +from api.tenancy import te_schema from api.db import Conn # BUG-035: auth is applied per-router (see csv_routes.py for the rationale). router = APIRouter( prefix="/api/te", tags=["te"], - dependencies=[Depends(require_api_key)], + dependencies=[Depends(tenant_scope)], ) @@ -30,7 +31,7 @@ def te_tables() -> list[dict]: SELECT table_name FROM information_schema.tables WHERE table_schema = %s AND table_name = ANY(%s) """, - (settings.TE_SCHEMA, tables), + (te_schema(), tables), ) existing = {r[0] for r in cur.fetchall()} cur.execute( @@ -38,7 +39,7 @@ def te_tables() -> list[dict]: SELECT relname, n_live_tup FROM pg_stat_user_tables WHERE schemaname = %s AND relname = ANY(%s) """, - (settings.TE_SCHEMA, tables), + (te_schema(), tables), ) row_counts = {r[0]: max(r[1], 0) for r in cur.fetchall()} return [ diff --git a/api/services/audit.py b/api/services/audit.py index f1d132f..59941e3 100644 --- a/api/services/audit.py +++ b/api/services/audit.py @@ -28,7 +28,7 @@ from psycopg2 import sql -from api.config import settings +from api.tenancy import uploads_schema from api.db import Conn logger = logging.getLogger(__name__) @@ -52,7 +52,7 @@ def write_audit(cur, action: str, file_id: int | None, file_name: str | None, "INSERT INTO {}.audit_log " "(action, file_id, file_name, table_name, mode) " "VALUES (%s, %s, %s, %s, %s)" - ).format(sql.Identifier(settings.UPLOADS_SCHEMA)), + ).format(sql.Identifier(uploads_schema())), (action, file_id, file_name, table_name, mode), ) diff --git a/api/services/dynamic_loader.py b/api/services/dynamic_loader.py index 5f70390..74fdf9c 100644 --- a/api/services/dynamic_loader.py +++ b/api/services/dynamic_loader.py @@ -18,6 +18,7 @@ from psycopg2.extras import execute_values from api.config import settings +from api.tenancy import uploads_schema from api.db import Conn from api.services.csv_parse import ( ALLOWED_TYPES, @@ -103,7 +104,7 @@ def upload_dynamic( "logs": logs, } - schema = settings.UPLOADS_SCHEMA + schema = uploads_schema() try: return _do_upload(file_name, types, overwrite, logs, schema, rows, file_hash) diff --git a/api/services/te_loader.py b/api/services/te_loader.py index 4e36195..be213ce 100644 --- a/api/services/te_loader.py +++ b/api/services/te_loader.py @@ -27,6 +27,7 @@ from psycopg2.extras import execute_values from api.config import TE_TABLES, settings +from api.tenancy import te_schema, uploads_schema from api.db import Conn from api.services.csv_parse import ( cast_value, @@ -56,7 +57,7 @@ def te_table_columns(table: str) -> list[str]: WHERE table_schema = %s AND table_name = %s ORDER BY ordinal_position """, - (settings.TE_SCHEMA, table), + (te_schema(), table), ) return [r[0] for r in cur.fetchall()] @@ -69,7 +70,7 @@ def _te_column_types(cur, table: str) -> dict[str, str]: FROM information_schema.columns WHERE table_schema = %s AND table_name = %s """, - (settings.TE_SCHEMA, table), + (te_schema(), table), ) return {r[0]: r[1] for r in cur.fetchall()} @@ -89,7 +90,7 @@ def _cleanup_prior_registry_entries(cur, file_name, file_hash): sql.SQL( "SELECT table_name, mode FROM {s}.csv_files" " WHERE file_name = %s OR file_hash = %s" - ).format(s=sql.Identifier(settings.UPLOADS_SCHEMA)), + ).format(s=sql.Identifier(uploads_schema())), (file_name, file_hash), ) for table_name, mode in cur.fetchall(): @@ -99,14 +100,14 @@ def _cleanup_prior_registry_entries(cur, file_name, file_hash): # miss the table (it lives in csv_uploads, not public/default). cur.execute( sql.SQL("DROP TABLE IF EXISTS {}.{}").format( - sql.Identifier(settings.UPLOADS_SCHEMA), + sql.Identifier(uploads_schema()), sql.Identifier(table_name), ) ) cur.execute( sql.SQL( "DELETE FROM {s}.csv_files WHERE file_name = %s OR file_hash = %s" - ).format(s=sql.Identifier(settings.UPLOADS_SCHEMA)), + ).format(s=sql.Identifier(uploads_schema())), (file_name, file_hash), ) @@ -154,7 +155,7 @@ def upload_te(file_name: str, content: str, target_table: str) -> dict: if not table_cols: return { "status": "error", - "message": f'T&E table "{target_table}" not found in schema {settings.TE_SCHEMA}. ' + "message": f'T&E table "{target_table}" not found in schema {te_schema()}. ' "Run deploy_all.sh dev first.", "logs": logs, } @@ -162,7 +163,7 @@ def upload_te(file_name: str, content: str, target_table: str) -> dict: if unknown: return { "status": "error", - "message": f'CSV columns not present in {settings.TE_SCHEMA}.{target_table}: {", ".join(unknown)}', + "message": f'CSV columns not present in {te_schema()}.{target_table}: {", ".join(unknown)}', "logs": logs, } _log(logs, "validate_columns", f"All {len(columns)} CSV columns exist on {target_table}") @@ -172,13 +173,13 @@ def upload_te(file_name: str, content: str, target_table: str) -> dict: inserted = 0 insert_stmt = sql.SQL("INSERT INTO {}.{} ({}) VALUES ({})").format( - sql.Identifier(settings.TE_SCHEMA), + sql.Identifier(te_schema()), sql.Identifier(target_table), sql.SQL(", ").join(sql.Identifier(c) for c in columns), sql.SQL(", ").join(sql.Placeholder() for _ in columns), ) batch_insert_stmt = sql.SQL("INSERT INTO {}.{} ({}) VALUES %s").format( - sql.Identifier(settings.TE_SCHEMA), + sql.Identifier(te_schema()), sql.Identifier(target_table), sql.SQL(", ").join(sql.Identifier(c) for c in columns), ) @@ -304,8 +305,8 @@ def _err(row_number: int, column: str, value: str, reason: str) -> None: "INSERT INTO {}.csv_files (file_name, file_hash, table_name, mode, row_count, column_names, " "total_rows, duplicate_rows_skipped, failed_rows) " "VALUES (%s, %s, %s, 'te', %s, %s, %s, 0, %s) RETURNING id" - ).format(sql.Identifier(settings.UPLOADS_SCHEMA)), - (file_name, file_hash, f"{settings.TE_SCHEMA}.{target_table}", inserted, columns, + ).format(sql.Identifier(uploads_schema())), + (file_name, file_hash, f"{te_schema()}.{target_table}", inserted, columns, len(data_rows), failed_row_count), ) except psycopg2.errors.UniqueViolation: @@ -325,7 +326,7 @@ def _err(row_number: int, column: str, value: str, reason: str) -> None: return { "status": "ok", "fileId": str(file_id), - "tableName": f"{settings.TE_SCHEMA}.{target_table}", + "tableName": f"{te_schema()}.{target_table}", "totalRows": len(data_rows), "insertedRows": inserted, "duplicateRowsSkipped": 0, diff --git a/api/tenancy.py b/api/tenancy.py new file mode 100644 index 0000000..0e43efb --- /dev/null +++ b/api/tenancy.py @@ -0,0 +1,68 @@ +"""Per-request tenant schema routing (M2b, #93, BR-36; decision 6: schema per tenant). + +Each tenant's registry, audit log and uploaded tables live in schema +``tenant_``. A request signed in to a tenant runs with that schema set +here; ``uploads_schema()`` and ``te_schema()`` read it. A request without a +session (API-key or dev) keeps the deployment's default schemas. +""" + +from __future__ import annotations + +import threading +from contextlib import contextmanager +from contextvars import ContextVar + +from api.config import settings + +_current: ContextVar[str | None] = ContextVar("tenant_schema", default=None) +_migrate_lock = threading.Lock() + + +def schema_for(slug: str) -> str: + """Schema name for a tenant slug. The slug CHECK in control.tenants keeps it a safe identifier.""" + return f"tenant_{slug}" + + +def current_schema() -> str | None: + return _current.get() + + +def uploads_schema() -> str: + return _current.get() or settings.UPLOADS_SCHEMA + + +def te_schema() -> str: + # ponytail: per-tenant T&E tables arrive in #99; until then a tenant's + # T&E requests are refused before reaching this (csv_routes.upload). + return _current.get() or settings.TE_SCHEMA + + +@contextmanager +def scoped(schema: str): + token = _current.set(schema) + try: + yield + finally: + _current.reset(token) + + +def migrate_tenant(slug: str) -> None: + """Create or upgrade one tenant's schema to the latest per-tenant migration. + + Swaps settings.UPLOADS_SCHEMA while Alembic runs, so it is serialised and + must only be called from startup or a CLI, never inside a request. + """ + from alembic import command + from alembic.config import Config + + from api.db import alembic_ini_path + + with _migrate_lock: + original = settings.UPLOADS_SCHEMA + settings.UPLOADS_SCHEMA = schema_for(slug) + try: + cfg = Config(str(alembic_ini_path())) + cfg.attributes["scope"] = "tenant" + command.upgrade(cfg, "head") + finally: + settings.UPLOADS_SCHEMA = original diff --git a/scripts/create_tenant_admin.py b/scripts/create_tenant_admin.py index 7945620..f75c28f 100644 --- a/scripts/create_tenant_admin.py +++ b/scripts/create_tenant_admin.py @@ -4,7 +4,7 @@ python scripts/create_tenant_admin.py --tenant acme --name "Acme Pty Ltd" --username alice The password is read from TE_ADMIN_PASSWORD if set, otherwise prompted for. -Brings the database to the current migration first. +Brings the database to the current migration first, then creates the tenant's schema. """ from __future__ import annotations @@ -19,6 +19,7 @@ from api import db # noqa: E402 from api.services import accounts # noqa: E402 +from api.tenancy import migrate_tenant, schema_for # noqa: E402 def main() -> int: @@ -40,9 +41,11 @@ def main() -> int: user_id = accounts.create_user(cur, args.username, password) accounts.add_membership(cur, user_id, tenant_id, "tenant_admin") conn.commit() + migrate_tenant(args.tenant) finally: db.close_pool() - print(f"Created tenant '{args.tenant}' with tenant admin '{args.username}'.") + print(f"Created tenant '{args.tenant}' (schema {schema_for(args.tenant)}) " + f"with tenant admin '{args.username}'.") return 0 diff --git a/scripts/lint.sh b/scripts/lint.sh index 7efab81..e533339 100644 --- a/scripts/lint.sh +++ b/scripts/lint.sh @@ -21,6 +21,7 @@ PY_SRC=( api/auth.py api/services/accounts.py api/routers/auth_routes.py + api/tenancy.py scripts/create_tenant_admin.py ) @@ -36,6 +37,7 @@ PY_TESTS=( tests/test_migrations.py tests/test_import_counts.py tests/test_accounts.py + tests/test_tenant_isolation.py ) # Collect scripts/*.py dynamically diff --git a/tests/test_tenant_isolation.py b/tests/test_tenant_isolation.py new file mode 100644 index 0000000..bc4c3d7 --- /dev/null +++ b/tests/test_tenant_isolation.py @@ -0,0 +1,173 @@ +"""Standing cross-tenant negative tests (M2b, #93, BR-36; ADR-0003). + +Two tenants, each with an uploader. Nothing one tenant does may be visible to +or changeable by the other, or by a request that has no session. +Integration: needs a live PostgreSQL. Fails (never skips) without one. +""" + +from __future__ import annotations + +import os +import unittest +import uuid + +import pytest +from fastapi.testclient import TestClient +from psycopg2 import sql + +from api.config import settings +from api.db import Conn +from api.main import app +from api.services import accounts +from api.tenancy import migrate_tenant, schema_for + +_HELP = ("Start PostgreSQL and ensure api/ can reach it (PGHOST/PGPORT/PGUSER/" + "PGPASSWORD/PGDATABASE). See scripts/start-api.ps1.") +_PW = "correct horse battery staple" + + +@pytest.mark.integration +class CrossTenantIsolation(unittest.TestCase): + @classmethod + def setUpClass(cls): + # DELETE is off by default (API_ALLOW_DESTRUCTIVE); enable it so the + # cross-tenant delete attempt reaches the tenant check. + cls._orig_allow_destructive = settings.allow_destructive + settings.allow_destructive = True + cls.ctx = TestClient(app) + try: + cls.client = cls.ctx.__enter__() + except Exception as exc: # noqa: BLE001 + raise AssertionError(f"Could not start the API: {type(exc).__name__}: {exc}\n{_HELP}") from None + tag = f"{os.getpid()}_{uuid.uuid4().hex[:6]}" + cls.slugs = {"a": f"iso_a_{tag}", "b": f"iso_b_{tag}"} + with Conn() as conn: + with conn.cursor() as cur: + for key, slug in cls.slugs.items(): + tid = accounts.create_tenant(cur, slug, f"Tenant {key}") + uid = accounts.create_user(cur, f"{slug}_user", _PW) + accounts.add_membership(cur, uid, tid, "uploader") + conn.commit() + for slug in cls.slugs.values(): + migrate_tenant(slug) + cls.headers = {} + for key, slug in cls.slugs.items(): + res = cls.client.post("/api/auth/login", json={ + "tenant": slug, "username": f"{slug}_user", "password": _PW}) + assert res.status_code == 200, res.text + cls.headers[key] = {"Authorization": f"Bearer {res.json()['token']}"} + + @classmethod + def tearDownClass(cls): + settings.allow_destructive = cls._orig_allow_destructive + with Conn() as conn: + with conn.cursor() as cur: + for slug in cls.slugs.values(): + cur.execute(sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format(sql.Identifier(schema_for(slug)))) + cur.execute(sql.SQL("DELETE FROM {}.tenants WHERE slug = %s").format( + sql.Identifier(settings.CONTROL_SCHEMA)), (slug,)) + cur.execute(sql.SQL("DELETE FROM {}.users WHERE username = %s").format( + sql.Identifier(settings.CONTROL_SCHEMA)), (f"{slug}_user",)) + conn.commit() + cls.ctx.__exit__(None, None, None) + + def _upload(self, who: str, name: str, content: str) -> dict: + res = self.client.post("/api/csv/upload", headers=self.headers[who], + json={"fileName": name, "content": content}) + self.assertEqual(res.status_code, 200, res.text) + return res.json() + + def _files(self, who: str | None) -> list[dict]: + res = self.client.get("/api/csv/files", headers=self.headers.get(who, {})) + self.assertEqual(res.status_code, 200, res.text) + return res.json() + + def _fresh(self) -> tuple[str, str]: + tag = uuid.uuid4().hex[:10] + return f"iso_{tag}.csv", f"k,v\n1,{tag}\n2,{tag}\n" + + def test_upload_lands_in_the_tenants_own_schema(self): + name, content = self._fresh() + res = self._upload("a", name, content) + self.assertEqual(res["status"], "ok", res) + with Conn() as conn, conn.cursor() as cur: + cur.execute("SELECT table_schema FROM information_schema.tables WHERE table_name = %s", + (res["tableName"],)) + self.assertEqual([r[0] for r in cur.fetchall()], [schema_for(self.slugs["a"])]) + + def test_other_tenant_cannot_list_read_or_delete(self): + name, content = self._fresh() + res = self._upload("a", name, content) + table, file_id = res["tableName"], res["fileId"] + self.assertIn(name, [f["file_name"] for f in self._files("a")]) + self.assertNotIn(name, [f["file_name"] for f in self._files("b")]) + read = self.client.get(f"/api/csv/tables/{table}/rows", headers=self.headers["b"]) + self.assertEqual(read.status_code, 404, read.text) + delete = self.client.delete(f"/api/csv/files/{file_id}", headers=self.headers["b"]) + self.assertEqual(delete.status_code, 404, delete.text) + self.assertIn(name, [f["file_name"] for f in self._files("a")], "A's file must survive B's delete") + rows = self.client.get(f"/api/csv/tables/{table}/rows", headers=self.headers["a"]) + self.assertEqual(rows.status_code, 200, rows.text) + + def test_requests_without_a_session_see_no_tenant_data(self): + name, content = self._fresh() + res = self._upload("a", name, content) + self.assertNotIn(name, [f["file_name"] for f in self._files(None)]) + read = self.client.get(f"/api/csv/tables/{res['tableName']}/rows") + self.assertEqual(read.status_code, 404, read.text) + + def test_duplicate_detection_is_per_tenant(self): + name, content = self._fresh() + self.assertEqual(self._upload("a", name, content)["status"], "ok") + self.assertEqual(self._upload("a", "again_" + name, content)["status"], "duplicate_file") + self.assertEqual(self._upload("b", name, content)["status"], "ok", + "the same file in another tenant is not a duplicate") + + def test_audit_log_is_per_tenant(self): + name, content = self._fresh() + self._upload("a", name, content) + + def names(who): + res = self.client.get("/api/audit/log?limit=200", headers=self.headers.get(who, {})) + self.assertEqual(res.status_code, 200, res.text) + return [e["fileName"] for e in res.json()["entries"]] + + self.assertIn(name, names("a")) + self.assertNotIn(name, names("b")) + self.assertNotIn(name, names(None)) + + def test_te_mode_is_refused_for_tenants_until_99(self): + res = self.client.post("/api/csv/upload", headers=self.headers["a"], json={ + "fileName": "t.csv", "content": "name,org_type,country\nX,government,AU\n", + "mode": "te", "targetTable": "organisations"}) + self.assertEqual(res.status_code, 409, res.text) + + def test_invalid_session_is_401_not_a_fallback_to_default(self): + res = self.client.get("/api/csv/files", headers={"Authorization": "Bearer not-a-token"}) + self.assertEqual(res.status_code, 401, res.text) + + def test_tenant_schema_is_at_head_with_its_own_registry(self): + with Conn() as conn, conn.cursor() as cur: + cur.execute(sql.SQL("SELECT version_num FROM {}.alembic_version").format( + sql.Identifier(schema_for(self.slugs["b"])))) + self.assertEqual(cur.fetchone()[0], "0004") + cur.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = %s", + (schema_for(self.slugs["b"]),)) + self.assertTrue({"csv_files", "audit_log"} <= {r[0] for r in cur.fetchall()}) + + def test_concurrent_requests_never_see_another_tenants_schema(self): + from concurrent.futures import ThreadPoolExecutor + + marks = {} + for who in ("a", "b"): + name, content = self._fresh() + self._upload(who, name, content) + marks[who] = name + + def visible(who): + return who, {f["file_name"] for f in self._files(who)} & set(marks.values()) + + with ThreadPoolExecutor(max_workers=4) as pool: + results = list(pool.map(visible, ["a", "b", None] * 20)) + for who, seen in results: + self.assertEqual(seen, {marks[who]} if who else set(), who)