umbragit/UmbraTrans
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Test the property that replaced the one a policy used to prove

Browse Source 
The old file proved a policy refusing a row. With no shared table there is no
policy and nothing to refuse, so the tested property is gone rather than
rewritten - and Umbratools_2.md 5.0 says outright that this trade should be
noticed rather than discovered later. This commit is the noticing.

What is proved instead is stronger, not differently shaped. A statement with no
organization in hand asks for `projects` unqualified and is told the relation
does not exist: not zero rows, an error. That is a different claim from "the
policy filtered it", because a policy that was silently not enforced returned
zero rows and looked fine - which is the failure G2 was, and the reason the old
test had to run as a non-owner role to mean anything.

The new tests keep the non-owner discipline for the part that needs it. Naming
another tenant's schema is refused outright; repointing the search path at it
and using an ordinary name finds no table at all, because a role without USAGE
on a schema does not see its contents when names are resolved. That second one is
the outcome an application actually experiences, so both are asserted rather than
the convenient one.

Positive cases are here because a suite that only proves refusal passes just as
happily against a database where every write is refused: a write that must be
allowed is performed and then counted in both tenants.

What the file cannot prove is written down in its docstring rather than left to
be assumed. The application owns every tenant schema, so it can reach another
tenant's tables by naming that schema, and schema-per-tenant cannot change that
with one role serving every tenant. What it changes is that there is no shared
relation to reach through, and reaching another tenant means writing down another
organization's name - one call site in tenancy.py rather than a predicate in
every query.
This commit is contained in:
umbra audio
2026-10-03 13:33:34 -07:00
parent 79b3484083
commit 4781468181
1 changed files with 489 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files

834
umbratrans_web/tests/test_rls_tenancy.py
View File

@@ -1,23 +1,53 @@
"""Tenant isolation, exercised against a role that is not the table owner.
"""Tenant isolation, exercised against a schema per organization.
Every earlier check in this project ran as the owner, and RLS does not apply to a
table's owner until `FORCE ROW LEVEL SECURITY` - gate **G2**. So the policies could
be present, syntactically valid, listed in `pg_policies`, and enforcing nothing,
and every test would still have passed. These run as `rls_probe`, a role with
plain table privileges and no ownership, which is the only arrangement in which a
policy can be observed refusing anything.
**This is a different property from the one this file used to test, and the
difference is the point.**
Two behaviours here are worth stating plainly, because they are not the same and
application code gets them wrong:
The previous version proved: *a row-level-security policy refuses a row.* It set
the caller identity, queried a shared table, and asserted the count came back
zero. That was the strongest check the shared-tables model could support, and it
was genuinely worth having - it is the check that found gate G2, the fact that
the application connects as the table owner and so the policies did not apply to
it.
- **An `UPDATE` or `DELETE` that RLS excludes raises nothing.** The statement
succeeds and affects zero rows. Code that checks "did the update raise?" instead
of "did it change a row?" will believe it succeeded.
- **An `INSERT` that violates `WITH CHECK` raises**, with
`new row violates row-level security policy`.
Schema-per-tenant removes the thing that was being tested. There is no shared
``projects`` table, so there is no policy on it and nothing to refuse a row. §5.0
of ``Umbratools_2.md`` names this trade explicitly: *a tested property is traded
for a different one, and that trade should be noticed rather than discovered
later.* This file is the noticing. The old tests are not here in revised form:
keeping them would have meant keeping a shared table.
That asymmetry is a property of PostgreSQL, not of these policies. It is asserted
below so a future change to either is noticed.
What is proved instead, and it is a stronger claim rather than a differently
shaped weaker one:
1. **There is no shared table to leak through.** A statement with no organization
in hand asks for ``projects`` unqualified and is told ``relation "projects"
does not exist``. Not zero rows: an error. The strongest possible mistake -
writing a query and forgetting to scope it - now fails loudly instead of
quietly returning whatever every tenant shares.
2. **A tenant's tables are absent from another tenant's search path**, and the
database is the one saying so: an unqualified ``projects`` resolves to the
caller's own schema, ``current_schemas()`` lists that schema and ``public``
and no other tenant's, and naming another tenant's schema is refused.
3. **A row in one tenant is not reachable from another**, counted rather than
asserted: each tenant's own project is visible in its own context and absent,
by identifier, from the other's.
The positive cases are here too, because a suite that only proves refusal passes
just as happily against a database where every write is refused: a write that must
be allowed is performed, and the rows are counted afterwards in both tenants.
What this does **not** prove, so that it is written down rather than assumed: the
application connects as the owner of every tenant schema, so it can reach any
tenant's tables by naming that schema. Schema-per-tenant cannot change that - not
with one role serving every tenant - and pretending otherwise would be the same
mistake as RLS on a table its owner queries. What it changes is that there is no
shared relation to reach *through*, and that reaching another tenant means
deliberately writing down another organization's schema name, which is one call
site in :mod:`umbratrans_web.tenancy` rather than a predicate in every query.
The privilege assertions therefore run as ``rls_probe``, a role that owns
nothing, because that is the only arrangement in which the database can be
observed refusing.
Requires a role that does not own the tables:
@@ -31,20 +61,28 @@ from __future__ import annotations
import os
import uuid
from contextlib import contextmanager
from typing import Any, Iterator
import pytest
from umbratrans_web.tenancy import tenant_schema_name
ORG_A = uuid.UUID("11111111-1111-1111-1111-111111111111")
ORG_B = uuid.UUID("22222222-2222-2222-2222-222222222222")
USER_A = uuid.UUID("33333333-3333-3333-3333-333333333333")
USER_B = uuid.UUID("44444444-4444-4444-4444-444444444444")
PUBLISHER_A = uuid.UUID("66666666-6666-6666-6666-666666666666")
PROJECT_A = uuid.UUID("aaaaaaaa-0000-0000-0000-000000000001")
PROJECT_B = uuid.UUID("bbbbbbbb-0000-0000-0000-000000000002")
ARTIFACT_A = uuid.UUID("cccccccc-0000-0000-0000-00000000000c")
AUTHOR_A = uuid.UUID("dddddddd-0000-0000-0000-00000000000d")
SCHEMA_A = tenant_schema_name(ORG_A)
SCHEMA_B = tenant_schema_name(ORG_B)
PROBE_ROLE = "rls_probe"
#: Every table migration 0005 moved. The boundary is not about `projects`; it is
#: about all four, and a list of one would let the other three rot unnoticed.
TENANT_TABLES = ("projects", "project_authors", "project_publishers", "invitations")
def _scratch() -> str | None:
"""The disposable database, or None.
@@ -81,336 +119,442 @@ _SCRATCH = _scratch()
pytestmark = pytest.mark.skipif(
not _SCRATCH or not _probe_role_exists(_SCRATCH),
reason=(
"needs a disposable scratch database and a non-owner role to exercise RLS at "
"needs a disposable scratch database and a non-owner role to exercise isolation at "
f"all. Create the role with 'CREATE ROLE {PROBE_ROLE} NOLOGIN; GRANT {PROBE_ROLE} "
"TO umbra;' - RLS does not apply to the table owner, so as its owner these "
"tests can pass while nothing is enforced."
"TO umbra;' - the application owns every tenant schema, so as its owner a "
"cross-tenant read is not something the database will refuse."
),
)
@pytest.fixture
def tenancy():
"""A rolled-back transaction with two organizations and an artifact each.
Everything is undone by the rollback, so these tests can share the database
with the rest of the suite without a teardown that could fail halfway and
leave rows behind.
"""
import psycopg
from psycopg import sql
assert _SCRATCH, "the module-level skip should have prevented this"
connection = psycopg.connect(_SCRATCH)
cursor = connection.cursor()
role = sql.Identifier(PROBE_ROLE)
try:
# GRANT takes an identifier, so it cannot be parameterised either. The
# privileges are plain and additive, which is all a non-owner needs for a
# policy to have anything to bite on.
cursor.execute(
sql.SQL("GRANT USAGE ON SCHEMA public, app TO {}").format(role)
)
cursor.execute(
sql.SQL(
"GRANT SELECT, INSERT, UPDATE ON public.artifacts, public.publishers, "
"public.project_publishers, public.project_authors TO {}"
).format(role)
)
cursor.execute(
sql.SQL(
"GRANT SELECT ON public.organizations, public.users, public.user_roles, "
"public.roles, public.projects TO {}"
).format(role)
)
cursor.execute(
"""
INSERT INTO public.organizations (id, name, slug) VALUES
(%s, 'RLS Probe A', 'rls-probe-a'), (%s, 'RLS Probe B', 'rls-probe-b')
""",
(ORG_A, ORG_B),
)
cursor.execute(
"INSERT INTO public.users (id, email, status, created_at, updated_at) VALUES "
"(%s, 'rls-a@probe.test', 'active', now(), now()), "
"(%s, 'rls-b@probe.test', 'active', now(), now())",
(USER_A, USER_B),
)
# Look the role up rather than inserting one with a fixed id.
#
# The first version inserted `roles` with a hardcoded id and
# `ON CONFLICT DO NOTHING`. The unique constraint is on (scope, name), so
# the insert silently did nothing - `organization_admin` already existed
# with a different id - and the fixture then assigned `user_roles` to a
# role id that was not there. `validate_user_role_scope` then raised
# "role scope does not match assignment scope", and because that fired
# inside the fixture, all eight tests in this file errored rather than
# failing. A fixture that cannot set itself up is worse than no fixture,
# because it looks like the thing under test is broken.
cursor.execute(
"SELECT id FROM public.roles WHERE scope = 'organization' "
"AND name = 'organization_admin'"
)
row = cursor.fetchone()
if row is None:
pytest.fail(
"this database has no organization_admin role; the tenancy tests "
"cannot run against it"
)
role_admin = row[0]
cursor.execute(
"""
INSERT INTO public.user_roles (user_id, role_id, scope, organization_id, status)
SELECT v.who, %s, 'organization', v.org, 'active'
FROM (VALUES (%s::uuid, %s::uuid), (%s::uuid, %s::uuid)) v(who, org)
WHERE NOT EXISTS (SELECT 1 FROM public.user_roles WHERE user_id = v.who)
""",
(role_admin, USER_A, ORG_A, USER_B, ORG_B),
)
cursor.execute(
"INSERT INTO public.projects (id, organization_id, name, slug) VALUES "
"(%s, %s, 'Probe Book A', 'rls-probe-book-a'), "
"(%s, %s, 'Probe Book B', 'rls-probe-book-b')",
(PROJECT_A, ORG_A, PROJECT_B, ORG_B),
)
# A real sha256, because artifacts.content_hash has a format CHECK and a
# probe row that fails it would be refused for the wrong reason.
cursor.execute(
"INSERT INTO public.artifacts (id, project_id, organization_id, kind, "
"content_hash, byte_size, media_type, storage_key) VALUES "
"(%s, %s, %s, 'audio', %s, 300000000, 'audio/wav', 'projects/a/ch01.wav'), "
"(%s, %s, %s, 'audio', %s, 300000000, 'audio/wav', 'projects/b/ch01.wav')",
(ARTIFACT_A, PROJECT_A, ORG_A, "a" * 64,
uuid.uuid4(), PROJECT_B, ORG_B, "b" * 64),
)
cursor.execute(
"INSERT INTO public.publishers (id, name, slug, owner_organization_id, kind) "
"VALUES (%s, 'RLS Probe House', 'rls-probe-house', %s, 'imprint')",
(PUBLISHER_A, ORG_A),
)
connection.commit()
yield connection, cursor
finally:
connection.rollback()
cursor.execute(
"DELETE FROM public.artifacts WHERE project_id IN (%s, %s)", (PROJECT_A, PROJECT_B)
)
cursor.execute(
"DELETE FROM public.project_publishers WHERE project_id IN (%s, %s)",
(PROJECT_A, PROJECT_B),
)
cursor.execute("DELETE FROM public.publishers WHERE id = %s", (PUBLISHER_A,))
cursor.execute("DELETE FROM public.projects WHERE id IN (%s, %s)", (PROJECT_A, PROJECT_B))
cursor.execute(
"DELETE FROM public.user_roles WHERE user_id IN (%s, %s)", (USER_A, USER_B)
)
cursor.execute("DELETE FROM public.users WHERE id IN (%s, %s)", (USER_A, USER_B))
cursor.execute("DELETE FROM public.organizations WHERE id IN (%s, %s)", (ORG_A, ORG_B))
connection.commit()
cursor.close()
connection.close()
def _as(cursor, user_id: uuid.UUID | None) -> None:
"""Become the probe role, then act as ``user_id`` (or nobody).
`SET ROLE` takes an identifier and cannot be parameterised, so it is composed
with psycopg.sql rather than interpolated. `app.user_id` is a GUC and *is*
parameterised - passing it through set_config is also how it is set
everywhere else in the project.
"""
from psycopg import sql
cursor.execute(sql.SQL("SET LOCAL ROLE {}").format(sql.Identifier(PROBE_ROLE)))
cursor.execute(
"SELECT set_config('app.user_id', %s, true)", (str(user_id) if user_id else "",)
)
def test_an_organization_sees_only_its_own_artifacts(tenancy) -> None:
"""The tenant boundary on the datastore, observed rather than asserted.
This is the check that has never been possible in this project before: not
"the adapter's WHERE clause excludes it", which is what everything else has
tested, but "the database itself does not offer the row".
"""
connection, cursor = tenancy
_as(cursor, USER_A)
cursor.execute(
"SELECT count(*) FROM public.artifacts WHERE organization_id = %s", (ORG_A,)
)
assert cursor.fetchone()[0] == 1, "A cannot see its own artifact"
cursor.execute(
"SELECT count(*) FROM public.artifacts WHERE organization_id = %s", (ORG_B,)
)
assert cursor.fetchone()[0] == 0, "A can see B's artifact"
_as(cursor, USER_B)
cursor.execute(
"SELECT count(*) FROM public.artifacts WHERE organization_id = %s", (ORG_B,)
)
assert cursor.fetchone()[0] == 1, "B cannot see its own artifact"
cursor.execute("SELECT count(*) FROM public.artifacts")
assert cursor.fetchone()[0] == 1, (
"B sees more than its own; every row must be filtered by the policy, not by "
"the application"
)
def test_an_anonymous_principal_sees_nothing_at_all(tenancy) -> None:
"""No app.user_id means no rows, on the datastore and on shared reference data."""
connection, cursor = tenancy
_as(cursor, None)
for table in ("artifacts", "publishers", "project_publishers", "project_authors"):
cursor.execute(f"SELECT count(*) FROM public.{table}")
assert cursor.fetchone()[0] == 0, f"{table} is visible without a principal"
def test_publishers_are_shared_but_owned(tenancy) -> None:
"""Both organizations may read an imprint; only its owner may change it.
The asymmetry is the whole design. A publisher is shared reference data - the
same house publishes for many organizations - but `owner_organization_id` is
the one column that crosses tenants, so writing it is not shared.
"""
connection, cursor = tenancy
_as(cursor, USER_B)
cursor.execute("SELECT count(*) FROM public.publishers")
assert cursor.fetchone()[0] == 1, (
"an organization member cannot see an imprint published by another "
"organization; reference data must be readable or the project it belongs "
"to cannot be displayed"
)
# And the write is refused - by filtering, not by raising. See the module note.
cursor.execute(
"UPDATE public.publishers SET name = 'hijacked' WHERE id = %s", (PUBLISHER_A,)
)
assert cursor.rowcount == 0, (
"an organization administrator changed a publisher belonging to another "
"organization"
)
cursor.execute("SELECT name FROM public.publishers WHERE id = %s", (PUBLISHER_A,))
assert cursor.fetchone()[0] == "RLS Probe House"
def test_the_owning_organization_may_change_its_own_publisher(tenancy) -> None:
"""The positive case, because a policy that refuses everything also passes."""
connection, cursor = tenancy
_as(cursor, USER_A)
cursor.execute(
"UPDATE public.publishers SET name = 'RLS Probe House (revised)' WHERE id = %s",
(PUBLISHER_A,),
)
assert cursor.rowcount == 1, "the owning organization could not edit its own publisher"
cursor.execute("SELECT name FROM public.publishers WHERE id = %s", (PUBLISHER_A,))
assert cursor.fetchone()[0] == "RLS Probe House (revised)"
def test_a_cross_tenant_insert_is_refused_with_an_error(tenancy) -> None:
"""INSERT against `WITH CHECK` raises, unlike UPDATE which is filtered.
B writing an artifact into A's project. The error is the point: an INSERT that
cannot be filtered away has to be reported, or the caller believes it wrote
something.
"""
connection, cursor = tenancy
_as(cursor, USER_B)
with pytest.raises(psycopg_errors_insufficient_privilege()):
cursor.execute(
"INSERT INTO public.artifacts (id, project_id, organization_id, kind, "
"content_hash, byte_size, media_type, storage_key) VALUES "
"(%s, %s, %s, 'audio', %s, 1, 'audio/wav', 'x')",
(uuid.uuid4(), PROJECT_A, ORG_A, "c" * 64),
)
def _insert_publisher_claiming(cursor, owner: uuid.UUID | None) -> None:
"""One INSERT, in its own savepoint.
The savepoint is load-bearing. A policy violation aborts the transaction, so
without one the *second* refusal in a test cannot even be attempted - it fails
with "current transaction is aborted" and reads as a policy that allowed
something. The first version of this test had two refusals in one transaction
and reported the second as "DID NOT RAISE" when in fact nothing had run.
"""
cursor.execute("SAVEPOINT one_insert")
try:
cursor.execute(
"INSERT INTO public.publishers (id, name, slug, owner_organization_id, kind) "
"VALUES (%s::uuid, 'Claimed', %s::citext, %s::uuid, 'imprint')",
(uuid.uuid4(), uuid.uuid4().hex[:8], owner),
)
except Exception:
cursor.execute("ROLLBACK TO SAVEPOINT one_insert")
raise
def test_claiming_your_own_organization_as_the_owner_is_allowed(tenancy) -> None:
"""The positive case again, for the cross-tenant column specifically.
Without it, "no organization may insert anything" would satisfy every refusal
below.
"""
connection, cursor = tenancy
_as(cursor, USER_A)
cursor.execute(
"INSERT INTO public.publishers (id, name, slug, owner_organization_id, kind) "
"VALUES (%s::uuid, 'Mine', %s::citext, %s::uuid, 'imprint')",
(uuid.uuid4(), uuid.uuid4().hex[:8], ORG_A),
)
assert cursor.rowcount == 1, "an organization administrator cannot create its own imprint"
def test_claiming_another_organizations_publisher_as_your_own_is_refused(tenancy) -> None:
"""The cross-tenant column cannot be set from inside an organization.
Otherwise an organization administrator could attach a third party's imprint
to their own organization, asserting a corporate relationship that does not
exist - or denying one that does.
"""
connection, cursor = tenancy
_as(cursor, USER_A)
with pytest.raises(psycopg_errors_insufficient_privilege()):
_insert_publisher_claiming(cursor, ORG_B)
# And the same for a peer, which is the case most likely to be waved through
# on the reasoning that "B is not hostile, it is just another tenant".
_as(cursor, USER_B)
with pytest.raises(psycopg_errors_insufficient_privilege()):
_insert_publisher_claiming(cursor, ORG_A)
def test_creating_an_unowned_publisher_needs_a_platform_administrator(tenancy) -> None:
"""NULL owner means "independent", and that is a platform-administrator claim.
With a permissive `OR NULL` this would be the easiest row in the table to
create, and `is_organization_admin(NULL)` returning false rather than NULL is
what makes the refusal work.
"""
connection, cursor = tenancy
_as(cursor, USER_A)
with pytest.raises(psycopg_errors_insufficient_privilege()):
_insert_publisher_claiming(cursor, None)
def psycopg_errors_insufficient_privilege():
"""The exception PostgreSQL raises for a policy violation.
Wrapped rather than imported at module scope so the skip above can be
evaluated without importing psycopg when there is no database.
"""
def _undefined_table():
"""The exception PostgreSQL raises for a name that resolves to nothing."""
import psycopg.errors
return psycopg.errors.InsufficientPrivilege
return psycopg.errors.UndefinedTable
def _insufficient_privilege():
"""The exception PostgreSQL raises for a schema the caller may not use."""
import psycopg.errors
return psycopg.errors.InsufficientPrivilege
def _foreign_key_violation():
"""The exception raised when a row names a parent that is not there."""
import psycopg.errors
return psycopg.errors.ForeignKeyViolation
def _search_path_value(*schemas: str) -> str:
"""The GUC value for a list of schemas, each double-quoted.
``set_config('search_path', ...)`` takes a *string the server parses as a
list*, not SQL, so the quoting has to happen here rather than being left to
``sql.Identifier``. Composing the value with ``sql.Identifier`` instead
produces ``"tenant_x"`` as SQL syntax, which the server reads as the column
name ``tenant_x`` - and the error that comes back names a column, pointing at
nothing to do with the question being asked.
The value is still passed as a bind parameter, so nothing a schema name could
contain would reach the statement.
"""
return ", ".join('"' + part.replace('"', '""') + '"' for part in schemas)
#: The statement that sets it. `set_config` rather than `SET LOCAL search_path`,
#: because `SET` takes no bind parameters - psycopg renders them as `$1` and the
#: grammar rejects that.
SET_SEARCH_PATH = "SELECT set_config('search_path', %s, true)"
class Tenancy:
"""Two organizations, one project each, and a way to look at either.
Everything happens inside one transaction that the fixture rolls back at the
end - including the tenant schemas, because ``CREATE SCHEMA`` is transactional.
That is why this fixture can add two organizations to a shared database without
a teardown that could fail halfway and leave rows behind.
"""
def __init__(self, connection: Any) -> None:
self.connection = connection
@contextmanager
def owner(self, schema: str | None = None) -> Iterator[Any]:
"""A cursor with the search_path a request for ``schema`` would have.
The owner is used to set the fixture up and to *observe* what happened - to
count one tenant's rows from outside it. It is not the role the isolation
assertions run as; see the module docstring.
"""
cursor = self.connection.cursor()
try:
# A tenant context is this tenant's schema *and* public: the platform
# tables have one address and a request needs them. `public` alone is
# the platform context, and is the one with no tenant table in it.
parts = [schema, "public"] if schema else ["public"]
cursor.execute(SET_SEARCH_PATH, (_search_path_value(*parts),))
yield cursor
finally:
cursor.close()
@contextmanager
def probe(self, schema: str) -> Iterator[Any]:
"""A cursor that is the probe role, with only ``schema`` and ``public`` in path."""
from psycopg import sql
cursor = self.connection.cursor()
try:
cursor.execute(sql.SQL("SET LOCAL ROLE {}").format(sql.Identifier(PROBE_ROLE)))
cursor.execute(SET_SEARCH_PATH, (_search_path_value(schema, "public"),))
yield cursor
finally:
# `SET ROLE` is transaction-local, and so is the search path above, so
# both are gone when this transaction ends. RESET ROLE before the
# cursor closes so the fixture's own teardown runs as the owner.
cursor.execute(sql.SQL("RESET ROLE"))
cursor.close()
def refused(self, cursor: Any, expected: type[BaseException], statement: str, params: Any = None) -> BaseException:
"""Run a statement that must be refused, and return the error it raised.
The savepoint is load-bearing, twice over. A failed statement aborts the
transaction, so without one the *second* refusal in a test cannot be
attempted: it fails with "current transaction is aborted" and reads as
though the database had allowed something. A first version of this file had
three refusals in one transaction and reported the last as "DID NOT RAISE"
when in fact nothing had run at all.
And the rollback runs in a ``finally`` rather than inside the ``except``, so
a statement that fails with an error *other* than the one expected does not
leave the transaction aborted for whatever cleans up next. That is not
hypothetical: an over-narrow expectation here produced a second error
naming the cleanup rather than the mistake, which is exactly the shape of
failure this file exists to make impossible to mistake for a pass.
"""
cursor.execute("SAVEPOINT refusal")
try:
try:
cursor.execute(statement, params)
except expected as error:
return error
pytest.fail(f"the database allowed a statement that must be refused: {statement}")
finally:
cursor.execute("ROLLBACK TO SAVEPOINT refusal")
@pytest.fixture
def tenancy():
"""Two organizations with a project each, in their own schemas."""
import psycopg
from psycopg import sql
from psycopg.rows import dict_row
assert _SCRATCH, "the module-level skip should have prevented this"
# dict_row so a count is read as a name rather than a position. Every assertion
# here reads a named column on purpose: a query that changed which column it
# selects should fail loudly instead of quietly returning something else.
connection = psycopg.connect(_SCRATCH, row_factory=dict_row)
harness = Tenancy(connection)
try:
setup = connection.cursor()
# The organizations are inserted first and granted nothing. Creating their
# schemas is the trigger's job, not this fixture's: an organization whose
# schema was made by hand would prove nothing about an organization created
# by the application.
setup.execute(
"INSERT INTO public.organizations (id, name, slug) VALUES (%s, 'Schema Probe A', 'schema-probe-a'), "
"(%s, 'Schema Probe B', 'schema-probe-b')",
(ORG_A, ORG_B),
)
for schema in (SCHEMA_A, SCHEMA_B):
present = setup.execute("SELECT 1 AS found FROM pg_namespace WHERE nspname = %s", (schema,)).fetchone()
assert present is not None, (
f"{schema} does not exist; inserting an organization did not create a schema for it. "
"Migration 0005's trigger is missing, which would make every tenant query raise "
"'relation does not exist' for every organization."
)
for schema, organization, project in ((SCHEMA_A, ORG_A, PROJECT_A), (SCHEMA_B, ORG_B, PROJECT_B)):
with harness.owner(schema) as cursor:
cursor.execute(
"INSERT INTO projects (id, name, slug) VALUES (%s, %s, %s)",
(project, f"Probe Book {organization.hex[:4]}", f"schema-probe-book-{organization.hex[:8]}"),
)
setup.execute(
"INSERT INTO public.authors (id, name, slug) VALUES (%s, 'Schema Probe', %s) ON CONFLICT DO NOTHING",
(AUTHOR_A, "schema-probe-byline"),
)
# The probe may use A's schema and nothing else. No USAGE on B, so a
# statement that names it is refused by the server rather than going
# unread - which is the difference between a boundary and an oversight.
role = sql.Identifier(PROBE_ROLE)
setup.execute(
sql.SQL("GRANT USAGE ON SCHEMA public, app, {} TO {}").format(sql.Identifier(SCHEMA_A), role)
)
setup.execute(
sql.SQL("GRANT SELECT, INSERT ON ALL TABLES IN SCHEMA {} TO {}").format(sql.Identifier(SCHEMA_A), role)
)
setup.close()
yield harness
connection.rollback()
finally:
connection.close()
# ---------------------------------------------------------------------------
# 1. There is no shared table
# ---------------------------------------------------------------------------
def test_a_query_with_no_organization_in_hand_finds_no_project_table(tenancy) -> None:
"""The §5.0 example, observed rather than quoted.
``search_path`` is ``public`` alone - what a request that has not resolved an
organization gets - and an unqualified ``projects`` is refused with ``relation
"projects" does not exist``.
This is the assertion the old file could not make and the new model exists to
make possible. There is no policy to mis-apply, no privilege to grant wrongly
and no row to filter: there is no relation at all. A forgotten scope is an
error rather than a wider result set, which is the difference between a bug
that is found and a bug that is not.
"""
with tenancy.owner() as cursor:
error = tenancy.refused(cursor, _undefined_table(), "SELECT count(*) FROM projects")
assert "projects" in str(error), f"the refusal did not name the missing relation: {error}"
def test_no_shared_copy_of_a_tenant_table_survives_in_public(tenancy) -> None:
"""Asked of the catalogue, because the error above has another possible cause.
A missing relation can also mean a broken ``search_path``, and that would make
the test above pass for the wrong reason. So this asks PostgreSQL whether each
of the four exists in ``public`` at all, and requires that none does. If a
shared copy ever came back, there would be a table two organizations' rows
could both be in, and it would be there whether or not any policy existed.
"""
with tenancy.owner() as cursor:
for table in TENANT_TABLES:
cursor.execute("SELECT to_regclass(%s) AS present", (f"public.{table}",))
assert cursor.fetchone()["present"] is None, f"public.{table} exists and is shared between organizations"
# ---------------------------------------------------------------------------
# 2. A tenant's tables are absent from another tenant's search path
# ---------------------------------------------------------------------------
def test_an_unqualified_name_resolves_to_the_callers_own_tenant(tenancy) -> None:
"""Which schema ``projects`` means, asked of the database.
Not "the count came back right": a count could be right because of a filter
that is one edit away from being wrong. This asks PostgreSQL to resolve the
bare name and names the schema it resolved to, so a change that made A's
context resolve into B's would be caught even if both tenants happened to hold
the same number of rows.
"""
with tenancy.owner(SCHEMA_A) as cursor:
resolved = cursor.execute(
"SELECT n.nspname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
"WHERE c.oid = to_regclass('projects')"
).fetchone()
assert resolved is not None, "projects did not resolve at all in its own tenant's schema"
assert resolved["nspname"] == SCHEMA_A
assert resolved["nspname"] != SCHEMA_B
def test_the_search_path_holds_one_tenant_and_public_and_nothing_else(tenancy) -> None:
"""The list itself, because "absent from the search path" *is* the property.
``current_schemas(false)`` is PostgreSQL's own answer to what an unqualified
name can reach, with the implicitly-searched system schemas excluded. A's
context must list A's schema and ``public``, and B's schema must not be on it.
If it were - by a leftover ``SET``, by a pooled connection that did not reset,
by a database default - then every unqualified query in that connection could
reach B, and nothing else here would see it.
"""
for schema, other in ((SCHEMA_A, SCHEMA_B), (SCHEMA_B, SCHEMA_A)):
with tenancy.owner(schema) as cursor:
path = list(cursor.execute("SELECT current_schemas(false) AS path").fetchone()["path"])
assert path == [schema, "public"], (
f"unexpected search path {path}: it must be this tenant's schema and public, "
"and no other tenant's"
)
assert other not in path
def test_every_tenant_table_belongs_to_exactly_one_of_the_two_tenants(tenancy) -> None:
"""All four, not just ``projects``.
A boundary that holds for the table everybody thinks about and not for the
three nobody does is not a boundary. Each bare name is resolved from each
tenant's context and must land in that tenant's schema.
"""
for schema, other in ((SCHEMA_A, SCHEMA_B), (SCHEMA_B, SCHEMA_A)):
with tenancy.owner(schema) as cursor:
for table in TENANT_TABLES:
resolved = cursor.execute(
"SELECT n.nspname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
"WHERE c.oid = to_regclass(%s)",
(table,),
).fetchone()
assert resolved is not None, f"{table} did not resolve inside {schema}"
assert resolved["nspname"] == schema, (
f"{table} resolved into {resolved['nspname']} from {schema}; it belongs to {other}"
)
# The other tenant has its own copy, which is what makes the first
# assertion a statement about isolation rather than about emptiness.
cursor.execute("SELECT to_regclass(%s) AS present", (f"{other}.{table}",))
assert cursor.fetchone()["present"] is not None, f"{other}.{table} is missing"
def test_naming_another_tenants_schema_is_refused(tenancy) -> None:
"""The privilege half, as a role that owns nothing.
Both organizations are provisioned, A is granted to the probe and B is not. A
statement from A's context that reaches into B is refused by the server. That
is a stronger observation than "the count came back zero", because it means
the database declined rather than that the caller looked in the wrong place.
Three ways in, because two of them are the ones that get written by accident:
a qualified read, a qualified write, and repointing the search path at the
other tenant and then using an ordinary name.
The three do not fail the same way, and the difference is worth knowing.
Naming B's schema outright is refused - *permission denied for schema*. Naming
it in ``search_path`` and then writing an ordinary query finds **no table at
all**, because a role with no ``USAGE`` on a schema does not see its contents
when names are resolved, so the lookup falls through to ``public`` and
``public`` has no ``projects``. The second is the stronger outcome, and it is
the one an application actually experiences.
"""
with tenancy.probe(SCHEMA_A) as cursor:
# The control. Without it, every refusal below would also be produced by a
# probe that cannot read anything at all.
cursor.execute("SELECT count(*) AS total FROM projects")
assert cursor.fetchone()["total"] == 1, (
"the probe cannot read its own tenant's projects, so the refusals below would prove nothing"
)
from psycopg import sql
tenancy.refused(cursor, _insufficient_privilege(), f'SELECT count(*) FROM "{SCHEMA_B}".projects')
tenancy.refused(
cursor,
_insufficient_privilege(),
f'INSERT INTO "{SCHEMA_B}".projects (id, name, slug) VALUES (%s, %s, %s)',
(uuid.uuid4(), "Hijacked", f"hijacked-{uuid.uuid4().hex[:8]}"),
)
# Repointing the path is allowed - it is only a setting - and the refusal
# comes on the query that follows it, which is why the two statements are
# one case rather than two.
cursor.execute(
sql.SQL("SET LOCAL search_path = {}").format(
sql.SQL(", ").join(sql.SQL("{}").format(sql.Identifier(part)) for part in (SCHEMA_B, "public"))
)
)
error = tenancy.refused(cursor, _undefined_table(), "SELECT count(*) FROM projects")
assert "projects" in str(error), f"the refusal did not name the missing relation: {error}"
# ---------------------------------------------------------------------------
# 3. A row in one tenant is not reachable from another
# ---------------------------------------------------------------------------
def test_each_tenant_sees_its_own_project_and_not_the_others(tenancy) -> None:
"""Counted, by identifier, from both sides.
The positive and the negative case together, because either alone is satisfied
by a broken database: "A sees only its own" passes when A sees nothing, and
"A sees its own" passes when A sees everything.
"""
with tenancy.owner(SCHEMA_A) as cursor:
cursor.execute("SELECT count(*) AS total FROM projects")
assert cursor.fetchone()["total"] == 1, "A cannot see its own project"
cursor.execute("SELECT count(*) AS total FROM projects WHERE id = %s", (PROJECT_B,))
assert cursor.fetchone()["total"] == 0, "A can see B's project by identifier"
with tenancy.owner(SCHEMA_B) as cursor:
cursor.execute("SELECT count(*) AS total FROM projects")
assert cursor.fetchone()["total"] == 1, "B cannot see its own project"
cursor.execute("SELECT count(*) AS total FROM projects WHERE id = %s", (PROJECT_A,))
assert cursor.fetchone()["total"] == 0, "B can see A's project by identifier"
def test_the_write_that_must_be_allowed_lands_in_the_writers_own_tenant(tenancy) -> None:
"""The positive case for a write, counted afterwards in *both* tenants.
A boundary that refuses everything satisfies every refusal above, so the write
that must succeed is performed and then counted in the tenant that made it and
in the tenant that must not have it.
"""
new_project = uuid.uuid4()
with tenancy.owner(SCHEMA_A) as cursor:
cursor.execute(
"INSERT INTO projects (id, name, slug) VALUES (%s, %s, %s)",
(new_project, "A Second Book", f"schema-probe-second-{uuid.uuid4().hex[:8]}"),
)
assert cursor.rowcount == 1, "the allowed write was not applied"
cursor.execute("SELECT count(*) AS total FROM projects")
assert cursor.fetchone()["total"] == 2, "the write did not land in the writer's own tenant"
with tenancy.owner(SCHEMA_B) as cursor:
cursor.execute("SELECT count(*) AS total FROM projects")
assert cursor.fetchone()["total"] == 1, (
"a write in one tenant changed another tenant's row count; the schemas are not separate"
)
cursor.execute("SELECT count(*) AS total FROM projects WHERE id = %s", (new_project,))
assert cursor.fetchone()["total"] == 0, "the other tenant can read the new row"
def test_a_tenant_cannot_credit_another_tenants_project(tenancy) -> None:
"""The referential half, which the dropped foreign keys used to enforce.
``artifacts`` and ``jobs`` stay platform-level while ``projects`` is per
tenant, so the three foreign keys that pointed at ``projects`` had to be
dropped; see `docs/tenant-schema-classification.md` §2.5. Inside a tenant the
constraint survives, and it is asserted as behaviour: the write that must
succeed is performed, the write that must fail is performed, and the error it
raises is read.
The failing case is not exotic. It is what a bug looks like: one place
resolving the wrong schema, and another place naming a project it should not
have been able to reach at all.
"""
with tenancy.owner(SCHEMA_A) as cursor:
cursor.execute(
"INSERT INTO project_authors (project_id, author_id, credit_role, lang) VALUES (%s, %s, 'author', 'en')",
(PROJECT_A, AUTHOR_A),
)
assert cursor.rowcount == 1, "a tenant cannot credit its own byline on its own project"
with tenancy.owner(SCHEMA_B) as cursor:
error = tenancy.refused(
cursor,
_foreign_key_violation(),
"INSERT INTO project_authors (project_id, author_id, credit_role, lang) VALUES (%s, %s, 'author', 'en')",
(PROJECT_A, AUTHOR_A),
)
assert "project_authors_project_id_fkey" in str(error), f"the refusal was about the wrong constraint: {error}"