diff --git a/CHANGELOG.md b/CHANGELOG.md index 25668e7aa..370e90827 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -67,6 +67,33 @@ All notable changes to this project are documented in this file. The format is b - ORM restores now emit `restored` audit entries with translated action labels and ORM primary-key updates include the old and new key in their diffs (behaviour change: both used to leave no entry) (#811). +- Tenant and soft-delete scoping in `simple_module_db` now also applies to: + relationships whose `secondary` is the table of a mapped `MultiTenantMixin` / + `SoftDeleteMixin` model (a bare association `Table` is not filtered), in + `joinedload()`, `selectinload()`, `subqueryload()` and joins through the + relationship (lazy loads already were; writes through it skip the + soft-delete condition, like every write); Core statements on + `Model.__table__` run before any ORM use has configured the mappers; and + sessions on a derived engine (`engine.execution_options(...)`), bound per + model (`binds={...}`) or routing through an overridden `get_bind()`, which + now keep the engine's strict mode, fallback tenant and audit sink. + `simple_module_db` now requires SQLAlchemy 2.0.2 or later. +- **Behaviour change:** these shapes now raise `TenantIsolationError`: + - flushing an UPDATE or DELETE of an object the session did not load or + insert (`make_transient_to_detached()`, `merge(load=False)`, a detached + object passed to `session.add()`) whose row belongs to another tenant — + its `tenant_id` is read from the table first (only for such objects; a + found owner is read once, a missing row is read again on the next flush); + - `aliased(Model, )` whose selectable reads a tenant-scoped + or soft-deletable table through Core, other than the entity's own table + (or the entity's own table with `GROUP BY`, `HAVING`, a window function + or any function other than a row-wise one such as `lower`), unless inside `all_tenants()` (with + `include_deleted=True`); build the selectable from ORM entities instead. + The same holds for `aliased(Model, )` whose join holds a scoped + table other than the entity's own, and for `aliased(Model, text(…).columns(…))` + (bare, subquery or CTE) on a tenant-scoped or soft-deletable `Model`. +- `subqueryload()` on a relationship between tenant-scoped or soft-deletable + models no longer fails with a duplicate `anon_1` alias (#816). - Tenant scoping in `simple_module_db` now applies to more statement shapes: ORM bulk UPDATE by primary key (`session.execute(update(Model), rows)`), `update().ordered_values()` (assigning `tenant_id` raises like `.values()`), diff --git a/docs/database/mixins.md b/docs/database/mixins.md index 082f56208..c1309141f 100644 --- a/docs/database/mixins.md +++ b/docs/database/mixins.md @@ -89,6 +89,18 @@ Adds: When `SM_MULTI_TENANT=true` and a request has an active tenant, the `do_orm_execute` listener attaches `with_loader_criteria(cls, cls.tenant_id == tenant_id)` to every SELECT touching a `MultiTenantMixin` table. +A tenant-scoped (or soft-deletable) model used as a relationship's link table, +`relationship(..., secondary=Link.__table__)`, is filtered in every loader — +lazy, `joinedload`, `selectinload`, `subqueryload` — and in joins through the +relationship (writes through it skip the soft-delete condition). Only a link +table mapped by a mixin class is filtered: a bare association `Table` with a +`tenant_id` column is not, like every unmapped table. +`aliased(Model, )` over a selectable that reads a scoped table through Core is +refused unless that table is the entity's own (and the selectable does not +group, aggregate or use a window function); build such selectables from ORM +entities. Details in +[multi-tenancy](../framework/multi-tenancy.md#what-the-filter-can-see). + ### Automatic stamping SQLite and Postgres `INSERT ... ON CONFLICT DO UPDATE` arms obey the same @@ -98,7 +110,7 @@ predicate). Conflicts with another tenant's row leave it unchanged. An explicit `all_tenants()` / `execution_options(all_tenants=True)` bypass permits deliberate cross-tenant upserts (#749). -INSERTs populate `tenant_id` from the `current_tenant_id` contextvar. Creating a row for a different tenant than the active one — or changing `tenant_id` on an existing row — raises `TenantIsolationError`. +INSERTs populate `tenant_id` from the `current_tenant_id` contextvar. Creating a row for a different tenant than the active one — or changing `tenant_id` on an existing row — raises `TenantIsolationError`. Flushing an UPDATE or DELETE of an object the session did not load or insert (`make_transient_to_detached()`, `merge(load=False)`, a detached object passed to `session.add()`) first reads its row's `tenant_id`, once per identity; a row of another tenant raises `TenantIsolationError`. `insert(Model).from_select(...)` and Core table targets follow the same rules: when the target column list omits `tenant_id`, the guard adds it and a constant diff --git a/docs/framework/multi-tenancy.md b/docs/framework/multi-tenancy.md index 08085be08..29357f185 100644 --- a/docs/framework/multi-tenancy.md +++ b/docs/framework/multi-tenancy.md @@ -19,7 +19,7 @@ tenant a request acts for. | ORM `update()` / `delete()` (also bulk UPDATE by primary key) | filtered to the tenant; assigning `tenant_id` (`.values()` or `.ordered_values()`) raises | `MissingTenantError` | unfiltered | | `session.add` + flush | `tenant_id` filled in; a different explicit value raises | `MissingTenantError`, even with an explicit `tenant_id` | `tenant_id` filled in with the install's fallback tenant unless set | | ORM `insert(Model)` (bulk / `.values()`) | `tenant_id` filled in; a different explicit value raises | `MissingTenantError`, even when every row sets `tenant_id` | `tenant_id` filled in with the install's fallback tenant unless set | -| Flushing a change to, or a delete of, a loaded object | only if it belongs to the bound tenant | `MissingTenantError` | allowed | +| Flushing a change to, or a delete of, a persistent object | only if its row belongs to the bound tenant (an object made persistent in memory has its owner read from the table first) | `MissingTenantError` | allowed | | Changing `tenant_id` | raises | raises | raises (only an `all_tenants()` block may move a row) | | Legacy `Session.bulk_save_objects` / `bulk_insert_mappings` / `bulk_update_mappings` | `TenantIsolationError` | `TenantIsolationError` | allowed, unscoped | @@ -33,7 +33,20 @@ Tenant criteria — and soft-delete criteria, on reads — are attached for **every** tenant-scoped / soft-deletable model, not only the entities a statement names, so all of these are scoped (#332): -- `select(Model)`, relationship loads, ORM `update()`/`delete()`/`insert()`; +- `select(Model)`, ORM `update()`/`delete()`/`insert()`; +- relationship loads — lazy, `joinedload`, `selectinload`, `subqueryload` — + and joins or `any()` through a relationship. A relationship whose + `secondary` is the table of a `MultiTenantMixin` / `SoftDeleteMixin` model + (`secondary=Link.__table__`) filters the link rows too: its join condition + carries `(:all_tenants OR link.tenant_id = :tenant)` and + `(:include_deleted OR link.is_deleted IS false)`, bound on each execution + from the bound tenant, `all_tenants()` and the statement's + `all_tenants` / `include_deleted` options. With no tenant under strict mode + it matches no link row. An `update()` / `delete()` through the relationship + (`delete(Parent).where(Parent.links.any())`) keeps the tenant condition and + drops the soft-delete one, like every write. A bare association `Table` + that no mapped mixin class owns is not filtered, `tenant_id` column or not, + like every other unmapped table: map it with the mixins to have it scoped; - a table as a **join target**: `select(Project).join(Doc)`; - subqueries: `select(Doc.id).where(...).exists()`, a bare Core `exists().where(Doc.x == ...)`, `in_(select(Doc.x))`, scalar subqueries; @@ -73,6 +86,33 @@ statement names, so all of these are scoped (#332): loader criteria sit in `WHERE`, and a table with no primary key, whose padding cannot be told apart from a real row. +`aliased(Model, )` is rendered from the selectable it was +built over, which the filter does not rewrite. Built from ORM entities +(`aliased(Proj, select(Proj).where(Proj.id.in_(select(Doc.proj_id))).subquery())`) +it is scoped by the loader criteria. One that reads only the entity's own +tables (`aliased(Doc, select(Doc.__table__).where(…).subquery())`; for joined +inheritance, the tables of its mapper) is scoped on the rows it returns, unless +it may combine rows first: `GROUP BY`, `HAVING`, a window function, or any +function other than a row-wise one such as `lower`, `coalesce` or `abs`. So is `aliased(Doc, Doc.__table__.join(tag))`, +a `Join` of the entity's own table with tables that are not scoped. Any other +Core read of a scoped table inside it, and SQL text under an entity of a +scoped table, is refused (see below). + +The engine's policy (strict mode, the fallback tenant, the audit sink) is +found through a derived engine (`engine.execution_options(...)`) and through a +session bound per model (`binds={...}`, the strictest bind wins). A model's +table is known to the filter as soon as the class is mapped, so a Core +statement on `Model.__table__` is scoped even before any ORM use configures +the mappers. + +The filter runs on `Session` executions. A statement run on a plain +`Connection` (`connection.execute(select(Model))`, or Core on `Model.__table__`) +is not scoped: no loader criteria, no Core predicates. Only the `secondary` +conditions above apply to it, because they sit in the relationship's join and +are read on every engine from the bound tenant, the statement's options and the +engine's strict mode (off when no policy is bound). Run statements through a +session. + With no tenant under strict mode, a statement that *names* a tenant model, or reads a tenant table through Core anywhere in it, raises; an indirect ORM reference (a join, a subquery) matches nothing. The cost is one walk of the @@ -82,8 +122,22 @@ and rewritten. Both are skipped when no tenant-scoped or soft-deletable model is installed. Refused outside `all_tenants()` (`TenantIsolationError`), because the rows -they write cannot be checked: - +they read or write cannot be checked: + +- `aliased(Model, )` whose selectable reads a tenant-scoped + or soft-deletable table through Core — `select(Doc.__table__)` joined to + another table, a nested `select(doc.c.x)`, an `exists()` over one — other + than the entity's own table in its own `FROM`. Build the selectable from + ORM entities, or run it under `all_tenants()` (with + `include_deleted=True` where soft-delete applies). A relationship loader's + own statement (`subqueryload` wraps the parent query) is not refused: it + was scoped when the parent query ran; +- `aliased(Model, )` (`Proj.__table__.join(Doc.__table__)`) whose join + reads a tenant-scoped or soft-deletable table through Core, other than the + entity's own table joined to unscoped tables (a second occurrence of the + entity's own table, an alias of it included, counts as another table); +- `aliased(Model, text(…).columns(…))`, bare or as a subquery or CTE, when + `Model` is tenant-scoped or soft-deletable: the filter cannot read SQL text; - an `INSERT` into a tenant table embedded in another statement as a CTE — run it as a statement of its own; - SQLite `insert(…).prefix_with("OR REPLACE")` on a tenant table while a @@ -113,6 +167,16 @@ tenant. The flush refuses to write or delete it, but reading it is not prevented: give each tenant its own session (or `session.expunge_all()` between tenants) in jobs and CLI loops. Per-request sessions are unaffected. +The unit of work updates and deletes by primary key, so the flush trusts an +object's `tenant_id` only when a session loaded or inserted it. An object +attached as already persistent — `make_transient_to_detached()`, +`merge(load=False)`, a transient copy given another row's primary key, any +detached object passed to `session.add()` — has its owner read from the table +before the flush writes it, and a row of another tenant raises +`TenantIsolationError`. A found owner is read once per identity; a missing row +is read again on the next flush. On a non-strict install with no bound tenant +the flush guard does not check owners, as before. + ## Acting outside a request ```python diff --git a/framework/db/pyproject.toml b/framework/db/pyproject.toml index 42e501729..c64471235 100644 --- a/framework/db/pyproject.toml +++ b/framework/db/pyproject.toml @@ -29,7 +29,7 @@ dependencies = [ # URL to +psycopg2 because Alembic's online migrations are sync-only. "psycopg2-binary>=2.9,<3", "simple_module_core==0.0.42", - "sqlalchemy[asyncio]>=2.0,<3", + "sqlalchemy[asyncio]>=2.0.2,<3", "sqlmodel>=0.0.22,<0.1", ] diff --git a/framework/db/simple_module_db/_core_walk.py b/framework/db/simple_module_db/_core_walk.py index 3b2200b3f..232d29b01 100644 --- a/framework/db/simple_module_db/_core_walk.py +++ b/framework/db/simple_module_db/_core_walk.py @@ -21,7 +21,14 @@ from typing import Any from sqlalchemy.sql.elements import ClauseElement, ColumnClause, ColumnElement -from sqlalchemy.sql.selectable import Alias, Join, Select, TableClause +from sqlalchemy.sql.selectable import ( + Alias, + CompoundSelect, + Join, + Select, + TableClause, + TextualSelect, +) from sqlalchemy.sql.traversals import _get_children from simple_module_db._core_joins import _base_name, _is_orm, is_scoped_leaf @@ -86,14 +93,18 @@ def core_where_froms(stmt: Select) -> list[Any]: return found -def references_plain(stmt: Any, names: set[str]) -> bool: +def references_plain(stmt: Any, names: set[str], opaque: frozenset[int] = frozenset()) -> bool: """Whether ``stmt`` reads a scoped table outside the ORM's annotations. ORM-annotated elements are not descended into: their tables are the - loader criteria's business, and every ORM column leads to a plain table. + loader criteria's business, and every ORM column leads to a plain table — + except the selectable of an ``aliased(Model, )`` entity, + which is whatever the caller built and is walked like any FROM, unless its + id is in ``opaque`` (SQL text under a scoped entity counts as a read). The answer depends only on *which* elements are reachable, so an element - met twice is expanded once. ``seen`` keeps the objects, not just their - ids, so a temporary child list cannot free one and recycle its id. + met twice — ORM-annotated and opaque ones included — is looked at once. + ``seen`` keeps the objects, not just their ids, so a temporary child list + cannot free one and recycle its id. """ stack = [stmt] seen: dict[int, Any] = {} @@ -103,9 +114,16 @@ def references_plain(stmt: Any, names: set[str]) -> bool: continue if el is not stmt and _nested_dml_on(el, names): return True # its target is checked even when ORM-annotated + seen[id(el)] = el if _is_orm(el): + selectable = aliased_selectable(el) + if selectable is not None and id(selectable) not in opaque: + if reads_text_as_scoped(el._annotations["parententity"], selectable, names): + return True # SQL text: nothing in it can be seen, let alone scoped + stack.append(selectable) + continue + if id(el) in opaque: continue - seen[id(el)] = el target = el if isinstance(el, (TableClause, Alias)) else _table_of(el) if isinstance(target, (TableClause, Alias)) and is_scoped_leaf(target, names): return True @@ -114,6 +132,36 @@ def references_plain(stmt: Any, names: set[str]) -> bool: return False +def aliased_selectable(el: Any) -> Any | None: + """The selectable an ``aliased(Model, selectable)`` entity of ``el`` reads, if any. + + A subquery / CTE of a SELECT (or of SQL text), a bare ``text().columns()`` + or a ``Join`` of tables: whatever the caller built. ``aliased(Model)`` (an + alias of the mapped table) and other entities give ``None``: their tables + are covered by the loader criteria. + """ + insp = el._annotations.get("parententity") + if insp is None or not insp.is_aliased_class: + return None + selectable = insp.selectable + if isinstance(selectable, (Join, TextualSelect)): + return selectable + inner = getattr(selectable, "element", None) + return selectable if isinstance(inner, (Select, CompoundSelect, TextualSelect)) else None + + +def is_textual(selectable: Any) -> bool: + """``text(...).columns(...)``, bare or as a subquery / CTE.""" + return isinstance(selectable, TextualSelect) or isinstance( + getattr(selectable, "element", None), TextualSelect + ) + + +def reads_text_as_scoped(insp: Any, selectable: Any, names: set[str]) -> bool: + """An entity of a scoped table whose rows come from SQL text, which no filter can read.""" + return is_textual(selectable) and _base_name(insp.mapper.local_table) in names + + _DML = ("insert", "update", "delete") diff --git a/framework/db/simple_module_db/_engine_policies.py b/framework/db/simple_module_db/_engine_policies.py new file mode 100644 index 000000000..1d834f281 --- /dev/null +++ b/framework/db/simple_module_db/_engine_policies.py @@ -0,0 +1,80 @@ +"""Per-engine tenancy policy, resolved through derived engines and session binds. + +Keyed by engine rather than held in a module global so two ``DatabaseState``s +in one process (tests, a CLI next to an app) cannot switch each other's +strict mode off. ``engine.execution_options(...)`` returns an ``OptionEngine`` +that proxies the engine it came from, and a session may be bound per model or +table (``binds={...}``) instead of to one engine: both resolve to the policy +of the base engine. +""" + +from __future__ import annotations + +from typing import Any +from weakref import WeakKeyDictionary + +from sqlalchemy.engine import Connection, Engine +from sqlalchemy.exc import UnboundExecutionError +from sqlalchemy.orm import Session +from sqlalchemy.orm.exc import UnmappedClassError + +from simple_module_db import model_registry as registry + +_POLICY_KEY = "sm_engine_policy" +_engine_policy: WeakKeyDictionary[Engine, Any] = WeakKeyDictionary() +_NO_POLICY = object() # cached for a session none of whose binds has a policy +_bound = 0 # bumped by ``set_policy``: a memoized answer older than it is stale + + +def set_policy(engine: Engine, policy: Any) -> None: + """Attach ``policy`` to ``engine`` and invalidate every session's memoized answer.""" + global _bound + _engine_policy[engine] = policy + _bound += 1 + + +def policy_of(bind: Any) -> Any: + """The policy of ``bind`` (an engine, a derived engine or a connection), or ``None``.""" + if isinstance(bind, Connection): + bind = bind.engine + while isinstance(bind, Engine): + found = _engine_policy.get(bind) + if found is not None: + return found + bind = getattr(bind, "_proxied", None) # ``OptionEngine`` -> its base engine + return None + + +def session_policy(session: Session) -> Any: + """The policy for ``session``: the strictest of the binds ``get_bind`` resolves. + + Asked through the public ``Session.get_bind`` — once without a mapper and + once per tenant-scoped model — so per-model binds and a subclass that + overrides ``get_bind`` keep the strict policy of the engine they route to. + A class that is no longer mapped (or has no bind) is skipped, never raised. + The answer (a policy, or none) is memoized on ``session.info`` for the + session's ``bind``, the number of tenant models and the policy bindings + made; a bind added with ``bind_mapper()`` after the first statement is not + seen. + """ + count = (len(registry.tenant_classes), _bound) + cached = session.info.get(_POLICY_KEY) + if cached is not None and cached[0] is session.bind and cached[1] == count: + return None if cached[2] is _NO_POLICY else cached[2] + found: dict[int, Any] = {} + for mapper in (None, *registry.tenant_classes): + try: + bind = session.get_bind(mapper=mapper) + except (UnboundExecutionError, UnmappedClassError): + continue + policy = policy_of(bind) + if policy is not None: + found.setdefault(id(policy), policy) + policies = list(found.values()) + strict = [p for p in policies if getattr(p, "tenant_strict", False)] + policy = (strict or policies or [None])[0] + session.info[_POLICY_KEY] = (session.bind, count, _NO_POLICY if policy is None else policy) + return policy + + +__all__ = ["policy_of", "session_policy", "set_policy"] diff --git a/framework/db/simple_module_db/_entity_selectables.py b/framework/db/simple_module_db/_entity_selectables.py new file mode 100644 index 000000000..e467c8b1b --- /dev/null +++ b/framework/db/simple_module_db/_entity_selectables.py @@ -0,0 +1,218 @@ +"""``aliased(Model, )`` entities in a statement (#816). + +The ORM renders such an entity from the selectable it was built over, found +through its annotations — not from the statement's FROM list. So +``core_scope`` must not clone one: the clone would be rendered *next to* the +original, under the same name (``subqueryload()`` is built this way). + +What the selectable reads is therefore never rewritten. It is either trusted +or refused: + +* a relationship loader's statement (``subqueryload``) wraps the parent + statement, which was scoped when it ran — trusted; +* a selectable that reads only the entity's own table, in its own ``FROM``, + is covered by the entity's loader criteria (``include_aliases``) on the + rows it returns — allowed; so is a ``Join`` of the entity's own table with + tables that are not scoped; +* any other selectable that reads a tenant-scoped or soft-deletable table + through Core (``select(Model.__table__)``, a nested ``select(t.c.x)``, a + join) is refused with ``TenantIsolationError``. Built from ORM entities + (``select(Model)``) it is scoped by the loader criteria and runs; +* SQL text (``text().columns()``, bare or as a subquery / CTE) under an + entity of a scoped table is refused: no filter can read what it selects. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from sqlalchemy.sql import visitors +from sqlalchemy.sql.elements import FunctionFilter, Over, WithinGroup +from sqlalchemy.sql.functions import FunctionElement +from sqlalchemy.sql.selectable import ( + Alias, + CompoundSelect, + FromClause, + FromGrouping, + Join, + Select, +) + +from simple_module_db._core_joins import _base_name, _is_orm, is_plain_leaf, is_scoped_leaf +from simple_module_db._core_walk import ( + aliased_selectable, + froms_read, + is_textual, + reads_text_as_scoped, + references_plain, +) +from simple_module_db.tenancy import TenantIsolationError + +AliasedGuard = Callable[[Any, Any, set[str]], None] + +# Functions that read one row at a time; any other (an aggregate such as ``max``, +# ``any_value`` or a custom ``GenericFunction``) may combine every tenant's rows +# even without a GROUP BY, so it makes the own-rows shortcut decline. Names, since +# ``func.`` builds a plain ``Function`` for any name SQLAlchemy has no class for. +_ROW_WISE = frozenset( + { + "abs", "ceil", "ceiling", "char_length", "coalesce", "concat", "current_date", + "current_time", "current_timestamp", "floor", "ifnull", "instr", "length", + "localtime", "localtimestamp", "lower", "ltrim", "now", "nullif", "replace", + "round", "rtrim", "substr", "substring", "trim", "upper", + } +) # fmt: skip + + +class StatementParts: + """One pass over a statement: its ORM options and its aliased-entity selectables.""" + + def __init__(self, stmt: Any) -> None: + self.options: list[Any] = [] + self.entities: dict[int, tuple[Any, Any]] = {} # id(selectable) -> (insp, selectable) + annotated: list[Any] = [] + for el in visitors.iterate(stmt): + self.options.extend(getattr(el, "_with_options", ())) + if _is_orm(el): + selectable = aliased_selectable(el) + if selectable is not None: + self.entities.setdefault( + id(selectable), (el._annotations["parententity"], selectable) + ) + if isinstance(el, FromClause): + annotated.append(el) + # The annotated copies of an entity's selectable (``select_from(P)``) + # are as much the entity's as the selectable itself. + self.keep = [sel for _insp, sel in self.entities.values()] + if self.entities: + self.keep += [el for el in annotated if id(el._deannotate()) in self.entities] + self.opaque = frozenset(self.entities) + + +def check_entities(parts: StatementParts, names: set[str], guard: AliasedGuard | None) -> None: + if guard is None: + return + for insp, selectable in parts.entities.values(): + if references_plain(selectable, names) or reads_text_as_scoped(insp, selectable, names): + guard(insp, selectable, names) + + +def refuse_aliased_core(insp: Any, selectable: Any, names: set[str]) -> None: + """The guard for statements a caller wrote (not relationship loads).""" + entity = insp.mapper.class_.__name__ + if is_textual(selectable): + raise TenantIsolationError( + f"aliased({entity}) over SQL text is not tenant-scoped: the filter cannot read " + "what the text selects; build the selectable from ORM entities (select(Model)) " + "or run it under all_tenants() (with include_deleted=True for trashed rows)" + ) + if _own_rows_only(insp, selectable, names): + return + leaves = (el for el in visitors.iterate(selectable) if is_scoped_leaf(el, names)) + tables = sorted({str(_base_name(el)) for el in leaves}) + raise TenantIsolationError( + f"aliased({entity}) over a selectable that reads " + f"{', '.join(repr(t) for t in tables) or 'a scoped table'} through Core is not " + "tenant-scoped; build the selectable from ORM entities (select(Model)) or run " + "it under all_tenants() (with include_deleted=True for trashed rows)" + ) + + +def _own_rows_only(insp: Any, selectable: Any, names: set[str]) -> bool: + """``aliased(Doc, select(doc).where(doc.c.x == ...).subquery())``: one SELECT, own table. + + No join, no grouping, window function or function outside a small set of + row-wise ones, no scoped table other than the entity's own (those of its + mapper, so a joined-inheritance entity's base table counts), and no nested + SELECT reading one — so every row it returns is a row of the entity's + tables, and the entity's loader criteria, applied to the alias, filter + exactly those rows. + """ + if isinstance(selectable, Join): + return _own_join_only(insp, selectable, names) + inner = selectable.element + if not isinstance(inner, Select) or inner._setup_joins: + return False + if any(isinstance(f, Join) for f in inner._from_obj): + return False + # Grouping, aggregates and window functions combine rows *before* the entity's + # criteria filter the result, so a row of this tenant could carry other tenants' data. + if inner._group_by_clauses or inner._having_criteria or _combines_rows(inner): + return False + own = _own_tables(insp) + if any(is_scoped_leaf(f, names) and f not in own for f in froms_read(inner)): + return False + return not any( + el is not inner and isinstance(el, (Select, CompoundSelect)) and references_plain(el, names) + for el in visitors.iterate(inner) + ) + + +def _own_join_only(insp: Any, join: Join, names: set[str]) -> bool: + """``aliased(Doc, doc.join(tag))``: the entity's own table(s) joined to unscoped tables. + + Every row carries one row of the entity's tables, which its loader + criteria filter. The entity's tables are those of its mapper (and of a + ``with_polymorphic`` entity's mappers), so the flat alias the ORM builds + for joined inheritance counts too. Each may appear once: a second copy of + one, any other scoped table, a subquery or text on a side, or an ``ON`` + clause with a nested SELECT reading a scoped table through Core is not + covered. + """ + own = _own_tables(insp) + met: set[Any] = set() + stack: list[Any] = [join] + while stack: + node = stack.pop() + if isinstance(node, FromGrouping): + stack.append(node.element) + elif isinstance(node, Join): + stack.extend((node.left, node.right)) + nested = ( + el + for el in visitors.iterate(node.onclause) + if isinstance(el, (Select, CompoundSelect)) + ) + if any(references_plain(el, names) for el in nested): + return False + elif not is_plain_leaf(node): + return False + elif is_scoped_leaf(node, names): + base = _table_of_leaf(node) + if base not in own or base in met: + return False + met.add(base) + return True + + +def _own_tables(insp: Any) -> set[Any]: + """The tables of the entity's mapper and of a ``with_polymorphic`` entity's mappers.""" + return {t for m in (insp.mapper, *insp.with_polymorphic_mappers) for t in m.tables} + + +def _table_of_leaf(node: Any) -> Any: + while isinstance(node, Alias): + node = node.element + return node + + +def _combines_rows(inner: Select) -> bool: + """A function in ``inner`` itself that is not row-wise (a nested SELECT is its own).""" + stack = list(inner.get_children()) + while stack: + el = stack.pop() + if isinstance(el, (Select, CompoundSelect)): + continue + if isinstance(el, (Over, WithinGroup, FunctionFilter)): + return True + if ( + isinstance(el, FunctionElement) + and str(getattr(el, "name", "")).lower() not in _ROW_WISE + ): + return True + stack.extend(el.get_children()) + return False + + +__all__ = ["AliasedGuard", "StatementParts", "check_entities", "refuse_aliased_core"] diff --git a/framework/db/simple_module_db/_persisted_owner.py b/framework/db/simple_module_db/_persisted_owner.py new file mode 100644 index 000000000..839171c9a --- /dev/null +++ b/framework/db/simple_module_db/_persisted_owner.py @@ -0,0 +1,86 @@ +"""The tenant a flushed row belongs to *in the database*, not only in memory. + +The unit of work updates and deletes by primary key alone, so the flush guard +compares the bound tenant with the object's ``tenant_id``. That value is +trustworthy when the session loaded the object from (or inserted it into) the +database — but an object can also become persistent without either: +``make_transient_to_detached()``, ``merge(load=False)``, a transient copy given +another row's primary key, or any detached object passed to ``session.add()``. +Its ``tenant_id`` is then whatever was set in memory. + +Every such object enters the session through ``detached_to_persistent``, and +only those are marked (``mark_unverified``): loads and inserts cost nothing. +A marked object has its owner read from the table (``persisted_owner``) before +the flush may write it. A found owner is remembered with the identity it was +read for, so it costs one ``SELECT``, once; a missing row is not remembered and +is read again on the next flush (a row inserted since must still be checked). +""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy import and_, select +from sqlalchemy import inspect as sa_inspect +from sqlalchemy.orm import Session + +from simple_module_db.insert_guard import TENANT_COLUMN +from simple_module_db.mixins import MultiTenantMixin + +_OWNER_KEY = "sm_unverified_owner" +_UNCHECKED = (None, None) # ``state.key`` is never ``None`` for a persistent object + + +def mark_unverified(session: Session, obj: Any) -> None: + """``detached_to_persistent``: the session did not load or insert ``obj``.""" + if isinstance(obj, MultiTenantMixin): + sa_inspect(obj).info[_OWNER_KEY] = _UNCHECKED + + +def owner_mark(obj: Any) -> Any: + """``obj``'s mark, for ``restore_owner_mark`` after an expunge and re-add.""" + return sa_inspect(obj).info.get(_OWNER_KEY) + + +def restore_owner_mark(obj: Any, mark: Any) -> None: + info = sa_inspect(obj).info + if mark is None: + info.pop(_OWNER_KEY, None) + else: + info[_OWNER_KEY] = mark + + +def is_known(obj: Any, owner: str | None) -> bool: + """``obj`` was loaded or inserted by a session, or its owner was read for this identity.""" + state = sa_inspect(obj) + mark = state.info.get(_OWNER_KEY) + if mark is None: + return True + key, persisted = mark + return key is not None and key == state.key and persisted == owner + + +def persisted_owner(session: Session, obj: Any) -> tuple[bool, str | None]: + """``(found, tenant_id)`` of ``obj``'s row, read with Core (never tenant-filtered).""" + state = sa_inspect(obj) + mapper = state.mapper + if state.key is None: + return False, None + column = mapper.columns[TENANT_COLUMN] + match = [col == value for col, value in zip(mapper.primary_key, state.key[1], strict=True)] + stmt = select(column).select_from(mapper.persist_selectable).where(and_(*match)) + rows = session.connection(bind_arguments={"mapper": mapper}).execute(stmt).all() + if not rows: # not memoized: a row inserted before the next flush is read then + return False, None + owner = rows[0][0] + state.info[_OWNER_KEY] = (state.key, owner) + return True, owner + + +__all__ = [ + "is_known", + "mark_unverified", + "owner_mark", + "persisted_owner", + "restore_owner_mark", +] diff --git a/framework/db/simple_module_db/_secondary_scope.py b/framework/db/simple_module_db/_secondary_scope.py new file mode 100644 index 000000000..8d8123053 --- /dev/null +++ b/framework/db/simple_module_db/_secondary_scope.py @@ -0,0 +1,145 @@ +"""Tenant / soft-delete criteria for a relationship's ``secondary`` table. + +``relationship(secondary=Link.__table__)`` joins a plain table that loader +criteria never see: ``joinedload``, ``selectinload``, ``subqueryload`` and a +``join(Parent.links)`` build that join at compile time, from the +relationship itself. (A lazy load names the table in its ``WHERE``, where +``core_scope`` finds it.) So once a mapper is configured, every relationship +whose ``secondary`` is the table of a ``MultiTenantMixin`` / +``SoftDeleteMixin`` model gets its ``secondaryjoin`` extended with + +* ``:all_tenants OR secondary.tenant_id = :tenant``, and +* ``:include_deleted OR secondary.is_deleted IS false``. + +The values are bound per execution, not per statement: the parameters are +``callable_`` binds, read when the statement's parameters are built, from +the scope ``record_execution_scope`` records just before — from the statement's +execution options (``all_tenants`` / ``include_deleted``), the bound tenant +and the engine's strict mode. Strict mode with no tenant matches no link row. +Like the main filter, an UPDATE / DELETE skips the soft-delete criterion: writes +may touch trashed rows. The scope is cleared after each execution, failed +ones included. + +A bare association ``Table`` that no mapped mixin class owns is not scoped, +``tenant_id`` column or not, like every other unmapped table. +""" + +from __future__ import annotations + +from contextvars import ContextVar +from typing import Any, NamedTuple + +from sqlalchemy import Boolean, String, and_, bindparam, event, or_ +from sqlalchemy.engine import Engine +from sqlalchemy.orm import Mapper + +from simple_module_db import model_registry as registry +from simple_module_db._core_joins import _base_name +from simple_module_db._engine_policies import policy_of +from simple_module_db.tenancy import ALL_TENANTS_OPTION, current_tenant_id, is_all_tenants + +INCLUDE_DELETED_OPTION = "include_deleted" +_SCOPED_MARK = "_sm_secondary_scoped" + + +class _Scope(NamedTuple): + skip_tenant: bool + tenant_id: str | None + skip_deleted: bool + + +_scope: ContextVar[_Scope | None] = ContextVar("sm_secondary_scope", default=None) +_any_scoped = False # whether any relationship has been extended yet + + +def _current() -> _Scope: + found = _scope.get() + if found is not None: + return found + tenant_id = current_tenant_id.get() + return _Scope(is_all_tenants() or tenant_id is None, tenant_id, False) + + +def secondary_criteria(table: Any) -> list[Any]: + name = _base_name(table) + criteria: list[Any] = [] + if name in registry.tenant_table_names: + skip = bindparam("sm_secondary_all_tenants", type_=Boolean(), callable_=_skip_tenant) + tenant = bindparam("sm_secondary_tenant_id", type_=String(), callable_=_tenant_id) + criteria.append(or_(skip, table.c.tenant_id == tenant)) + if name in registry.soft_delete_table_names: + skip = bindparam("sm_secondary_include_deleted", type_=Boolean(), callable_=_skip_deleted) + criteria.append(or_(skip, table.c.is_deleted.is_(False))) + return criteria + + +def _skip_tenant() -> bool: + return _current().skip_tenant + + +def _tenant_id() -> str | None: + return _current().tenant_id + + +def _skip_deleted() -> bool: + return _current().skip_deleted + + +def scope_relationship(prop: Any) -> None: + """Extend ``prop``'s ``secondaryjoin`` with its secondary table's criteria; idempotent.""" + if prop.secondary is None or getattr(prop, _SCOPED_MARK, False): + return + criteria = secondary_criteria(prop.secondary) + if not criteria: + return + join_condition = prop._join_condition + join_condition.secondaryjoin = and_(join_condition.secondaryjoin, *criteria) + prop.secondaryjoin = join_condition.secondaryjoin + setattr(prop, _SCOPED_MARK, True) + global _any_scoped + _any_scoped = True + + +@event.listens_for(Mapper, "mapper_configured") +def _on_mapper_configured(mapper: Mapper, cls: type) -> None: + for prop in mapper.relationships: + if prop.parent is mapper: + scope_relationship(prop) + + +def record_execution_scope( + conn: Any, clause: Any, multiparams: Any, params: Any, options: Any +) -> None: + """``before_execute``: record the scope the criteria above read for this execution.""" + if not _any_scoped: + return + policy = policy_of(conn) + strict = bool(policy is not None and policy.tenant_strict) + tenant_id = current_tenant_id.get() + skip_tenant = ( + bool(options.get(ALL_TENANTS_OPTION)) + or is_all_tenants() + or (tenant_id is None and not strict) + ) + is_write = bool(getattr(clause, "is_update", False) or getattr(clause, "is_delete", False)) + skip_deleted = is_write or bool(options.get(INCLUDE_DELETED_OPTION)) + _scope.set(_Scope(skip_tenant, tenant_id, skip_deleted)) + + +def clear_execution_scope(*_args: Any) -> None: + """``after_execute`` / ``handle_error``: forget the scope, so nothing outside an + execution reads it.""" + if _scope.get() is not None: + _scope.set(None) + + +# On the ``Engine`` class, so every engine records the scope — bound to a policy or +# not. A per-engine listener left an unbound engine reading the previous +# execution's scope and ignoring its options. Each execution sets the scope +# before its parameters are built and clears it afterwards, or when it fails +# (``handle_error``), so a failed one does not leave its scope behind. +event.listen(Engine, "before_execute", record_execution_scope) +event.listen(Engine, "after_execute", clear_execution_scope) +event.listen(Engine, "handle_error", clear_execution_scope) + +__all__: list[str] = [] diff --git a/framework/db/simple_module_db/core_scope.py b/framework/db/simple_module_db/core_scope.py index 0aa3f28c5..35acb165d 100644 --- a/framework/db/simple_module_db/core_scope.py +++ b/framework/db/simple_module_db/core_scope.py @@ -32,14 +32,10 @@ from simple_module_db._core_joins import Predicates, is_plain_leaf, same_from, scope_joins from simple_module_db._core_walk import froms_read, references_plain +from simple_module_db._entity_selectables import AliasedGuard, StatementParts, check_entities from simple_module_db._outer_joins import is_excluded, orm_outer_join_tables -def _executable_options(stmt: Any) -> list[Any]: - """ORM options anywhere in ``stmt``: shared as they are, never cloned.""" - return [o for el in visitors.iterate(stmt) for o in getattr(el, "_with_options", ())] - - def _add_where(stmt: Any, preds: list[Any]) -> None: if preds: # ``stmt`` is this traversal's own clone: mutating it is safe stmt._where_criteria = (*stmt._where_criteria, *preds) @@ -86,20 +82,28 @@ def scope_core( names: set[str], orm_covered: Any = None, nested: Predicates | None = None, + aliased: AliasedGuard | None = None, ) -> Any: """Return ``stmt`` with every plain scoped table filtered, or ``stmt`` itself. ``names`` are the scoped table names. ``orm_covered`` are the classes the caller gives loader criteria (for the #417 exclusion), or ``None``. ``nested`` gives the predicates for the target of a DML embedded in - ``stmt`` (or raises); ``stmt``'s own target is the caller's. + ``stmt`` (or raises); ``stmt``'s own target is the caller's. ``aliased`` + vets the selectable of each ``aliased(Model, )`` entity that + reads a scoped table (``_entity_selectables``); such a selectable, and the + ORM options, are shared as they are, never cloned. """ if not names or not references_plain(stmt, names): return stmt + parts = StatementParts(stmt) + check_entities(parts, names, aliased) + if parts.entities and not references_plain(stmt, names, parts.opaque): + return stmt visitor = _Visitor(predicates_for, orm_covered) out = visitors.cloned_traverse( stmt, - {"stop_on": _executable_options(stmt)}, + {"stop_on": [*parts.options, *parts.keep]}, { "select": visitor.select, "update": visitor.update_delete, diff --git a/framework/db/simple_module_db/flush_guard.py b/framework/db/simple_module_db/flush_guard.py index b0b4e4565..f6076b76a 100644 --- a/framework/db/simple_module_db/flush_guard.py +++ b/framework/db/simple_module_db/flush_guard.py @@ -5,6 +5,9 @@ added with ``session.add``, and objects that came out of the identity map — ``session.get`` answers from it without SQL, so a session reused across a ``tenant_context`` switch can hand back, and then write, another tenant's row. +An object the session did not load or insert (``make_transient_to_detached``, +``merge(load=False)``, any detached object passed to ``session.add()``) has its +owner read from the table first (``_persisted_owner``). """ from __future__ import annotations @@ -12,6 +15,7 @@ from sqlalchemy import inspect as sa_inspect from sqlalchemy.orm import Session +from simple_module_db._persisted_owner import is_known, persisted_owner from simple_module_db.mixins import MultiTenantMixin from simple_module_db.query_filter import fallback_tenant_id, is_strict, strict_configured from simple_module_db.tenancy import ( @@ -81,12 +85,35 @@ def guard_flush(session: Session) -> None: raise TenantIsolationError("Cannot change tenant_id of an existing object") owner = _owner(obj) op = "DELETE" if is_deleted else "UPDATE" - if tenant_id is not None and owner != tenant_id: - raise TenantIsolationError( - f"Cannot {op} an object of tenant '{owner}' in context of tenant '{tenant_id}'" - ) - if tenant_id is None and strict: - raise missing_tenant_error(type(obj).__name__, op) + if tenant_id is None: + if strict: + raise missing_tenant_error(type(obj).__name__, op) + continue + if owner != tenant_id: + raise _foreign_owner_error(op, owner, tenant_id) + _check_persisted_owner(session, obj, owner, tenant_id, op) + + +def _foreign_owner_error(op: str, owner: str | None, tenant_id: str) -> TenantIsolationError: + return TenantIsolationError( + f"Cannot {op} an object of tenant '{owner}' in context of tenant '{tenant_id}'" + ) + + +def _check_persisted_owner( + session: Session, obj: MultiTenantMixin, owner: str | None, tenant_id: str, op: str +) -> None: + """The unit of work writes by primary key: the row itself must be the bound tenant's. + + Only an object the session did not load or insert costs a read — one made + persistent in memory, whose ``tenant_id`` proves nothing. A found owner is + read once per identity; a missing row is read again on the next flush. + """ + if is_known(obj, owner): + return + found, persisted = persisted_owner(session, obj) + if found and persisted != tenant_id: + raise _foreign_owner_error(op, persisted, tenant_id) __all__ = ["guard_flush"] diff --git a/framework/db/simple_module_db/listeners.py b/framework/db/simple_module_db/listeners.py index e13eec735..a4d69b94a 100644 --- a/framework/db/simple_module_db/listeners.py +++ b/framework/db/simple_module_db/listeners.py @@ -12,6 +12,7 @@ from sqlalchemy import inspect as sa_inspect from sqlalchemy.orm import Session +from simple_module_db._persisted_owner import mark_unverified, owner_mark, restore_owner_mark from simple_module_db.bulk_audit import audit_statement from simple_module_db.bulk_guard import install_bulk_guard from simple_module_db.flush_guard import guard_flush @@ -130,6 +131,7 @@ def attach_session_listeners(session_class: type[Session]) -> None: for name, fn in ( ("before_flush", _before_flush_listener), ("after_flush", _mark_session_written), + ("detached_to_persistent", mark_unverified), ("after_flush_postexec", _after_flush_audit), ("do_orm_execute", _mark_dml_written), ("do_orm_execute", filter_statements), @@ -203,13 +205,15 @@ def _before_flush_listener( hard_delete_ids = session.info.pop(_HARD_DELETE_KEY, frozenset()) for obj in list(session.deleted): if isinstance(obj, SoftDeleteMixin) and not _is_purge(obj, hard_delete_ids): - # Cancel the hard delete + # Cancel the hard delete; the re-add keeps the object's owner mark. + mark = owner_mark(obj) session.expunge(obj) # Merge back as modified with soft-delete fields set obj.is_deleted = True obj.deleted_at = now obj.deleted_by = user_id session.add(obj) + restore_owner_mark(obj, mark) _log_entity_change( obj, diff --git a/framework/db/simple_module_db/model_registry.py b/framework/db/simple_module_db/model_registry.py index a4b41ae17..20de8d41f 100644 --- a/framework/db/simple_module_db/model_registry.py +++ b/framework/db/simple_module_db/model_registry.py @@ -1,6 +1,6 @@ """Every mapped tenant-scoped and soft-deletable model, known to the filters. -Filled as mappers configure (and topped up from any statement that names a +Filled as mappers are constructed (and topped up from any statement that names a model), so criteria can be attached for models a statement does *not* name: a join target, a subquery, a count (#332). """ @@ -51,3 +51,11 @@ def register(cls: type) -> None: @event.listens_for(Mapper, "mapper_configured") def _on_mapper_configured(mapper: Mapper, cls: type) -> None: register(cls) + + +@event.listens_for(Mapper, "after_mapper_constructed") +def _on_mapper_constructed(mapper: Mapper, cls: type) -> None: + # Registered as soon as the class is mapped, not when mappers configure: + # a Core statement over ``Model.__table__`` run before any ORM use would + # otherwise find the table unknown, and unscoped. + register(cls) diff --git a/framework/db/simple_module_db/query_filter.py b/framework/db/simple_module_db/query_filter.py index 89eb3053e..d025bd3ac 100644 --- a/framework/db/simple_module_db/query_filter.py +++ b/framework/db/simple_module_db/query_filter.py @@ -15,15 +15,19 @@ from collections.abc import Callable from typing import Any -from weakref import WeakKeyDictionary -from sqlalchemy.engine import Connection, Engine +from sqlalchemy.engine import Engine from sqlalchemy.orm import ORMExecuteState, Session, with_loader_criteria +from simple_module_db import _secondary_scope # noqa: F401 (registers the execution-scope listener) from simple_module_db import model_registry as registry from simple_module_db._bulk_pk_update import is_bulk_pk_update, scoped_bulk_pk_update from simple_module_db._core_joins import _base_name, is_scoped_leaf +from simple_module_db._engine_policies import _engine_policy as _engine_policy # re-exported +from simple_module_db._engine_policies import session_policy, set_policy +from simple_module_db._entity_selectables import refuse_aliased_core from simple_module_db.core_scope import scope_core +from simple_module_db.insert_guard import TENANT_COLUMN as TENANT_COLUMN # re-exported from simple_module_db.insert_guard import assigned_tenant_ids, guard_insert from simple_module_db.insert_scope import scope_insert from simple_module_db.scope_predicates import ( @@ -41,15 +45,8 @@ missing_tenant_error, ) -TENANT_COLUMN = "tenant_id" - -# Per-engine tenancy policy. Keyed by engine rather than held in a module -# global so two ``DatabaseState``s in one process (tests, a CLI next to an -# app) cannot switch each other's strict mode off. Values expose -# ``tenant_strict`` (and optionally ``default_tenant_id``) — a -# ``DatabaseState``, or an ``EngineTenancy`` for a bare sync engine such as -# the Celery worker's. -_engine_policy: WeakKeyDictionary[Engine, Any] = WeakKeyDictionary() +# ``_engine_policy`` (imported above) is re-exported: the per-engine policy map lives in +# ``_engine_policies``; modules/background_tasks/tests/test_worker_tenancy.py reads it here. class EngineTenancy: @@ -75,16 +72,17 @@ def __init__( def bind_engine_policy(engine: Engine, policy: Any) -> None: - """Attach a policy (anything with ``tenant_strict``) to a sync engine.""" - _engine_policy[engine] = policy + """Attach a policy (anything with ``tenant_strict``) to a sync engine. + + Engines derived from it (``engine.execution_options(...)``) and sessions + bound to it per model (``binds={...}``) resolve to the same policy. + """ + set_policy(engine, policy) def engine_policy(session: Session) -> Any: """The policy bound to this session's engine, or ``None``.""" - bind = session.bind - if isinstance(bind, Connection): - bind = bind.engine - return _engine_policy.get(bind) if isinstance(bind, Engine) else None + return session_policy(session) def strict_configured(session: Session) -> bool: @@ -155,6 +153,8 @@ def filter_statements(execute_state: ORMExecuteState) -> Any: scoped_names(skip_soft_delete, skip_tenant), orm_covered(skip_soft_delete, skip_tenant, tenant_id), nested_dml_guard(skip_tenant, tenant_id, strict), + # A relationship loader's statement wraps one that was scoped when it ran. + None if execute_state.is_relationship_load else refuse_aliased_core, ) if not skip_soft_delete: options.extend( @@ -185,7 +185,9 @@ def _scope_insert_source(execute_state: ORMExecuteState) -> None: nested = nested_dml_guard(skip_tenant, tenant_id, strict) execute_state.statement = scope_insert( execute_state.statement, - lambda el: scope_core(el, predicates_for, names, nested=nested), + lambda el: scope_core( + el, predicates_for, names, nested=nested, aliased=refuse_aliased_core + ), ) diff --git a/framework/db/tests/_db_read_scope_models.py b/framework/db/tests/_db_read_scope_models.py new file mode 100644 index 000000000..b15e653da --- /dev/null +++ b/framework/db/tests/_db_read_scope_models.py @@ -0,0 +1,118 @@ +"""Models and seed for the read-scope tests (relationship loaders, aliased entities). + +Parent 1 belongs to tenant ``a``. Its children (``parent_id = 1``): 1 and 2 +(``a``), 3 (``a``, trashed) and 4 (``b``). Its link rows (``secondary``): +1 → child 1 (``a``), 2 → child 2 (``b``) and 3 → child 2 (``a``, trashed). +""" + +from collections.abc import Iterator + +import pytest +from _models import _tenant_session +from simple_module_db.base import create_module_base +from simple_module_db.mixins import MultiTenantMixin, SoftDeleteMixin +from simple_module_db.tenancy import all_tenants, tenant_context +from sqlalchemy import event, update +from sqlalchemy.orm import relationship +from sqlmodel import Field, Relationship + +Base = create_module_base("read_scope") + + +class RsParent(Base, MultiTenantMixin, table=True): # ty: ignore[unsupported-base] + __tablename__ = "read_scope_parent" + id: int = Field(primary_key=True) + kids: list["RsChild"] = Relationship( + sa_relationship=relationship("RsChild", viewonly=True, order_by="RsChild.id") + ) + linked: list["RsChild"] = Relationship( + sa_relationship=relationship( + "RsChild", + secondary=lambda: RsLink.__table__, + primaryjoin="RsParent.id == RsLink.parent_id", + secondaryjoin="RsLink.child_id == RsChild.id", + viewonly=True, + order_by="RsChild.id", + ) + ) + + +class RsChild(Base, MultiTenantMixin, SoftDeleteMixin, table=True): # ty: ignore[unsupported-base] + __tablename__ = "read_scope_child" + id: int = Field(primary_key=True) + parent_id: int | None = Field(default=None, foreign_key="read_scope_parent.id") + + +class RsLink(Base, MultiTenantMixin, SoftDeleteMixin, table=True): # ty: ignore[unsupported-base] + __tablename__ = "read_scope_link" + id: int = Field(primary_key=True) + parent_id: int = Field(foreign_key="read_scope_parent.id") + child_id: int = Field(foreign_key="read_scope_child.id") + + +class RsProj(Base, table=True): # ty: ignore[unsupported-base] + """Untenanted, for ``aliased(RsProj, )``.""" + + __tablename__ = "read_scope_proj" + id: int = Field(primary_key=True) + child_id: int | None = None + + +parent, child, link, proj = ( + RsParent.__table__, + RsChild.__table__, + RsLink.__table__, + RsProj.__table__, +) + + +async def seed(session): + session.add_all([RsProj(id=1, child_id=1), RsProj(id=4, child_id=4)]) + with tenant_context("a"): + session.add(RsParent(id=1)) + await session.flush() + session.add_all([RsChild(id=i, parent_id=1) for i in (1, 2, 3)]) + await session.flush() + session.add_all( + [RsLink(id=1, parent_id=1, child_id=1), RsLink(id=3, parent_id=1, child_id=2)] + ) + await session.flush() + with tenant_context("b"): + session.add(RsChild(id=4, parent_id=1)) + await session.flush() + session.add(RsLink(id=2, parent_id=1, child_id=2)) + await session.flush() + with all_tenants(): + await session.execute(update(child).where(child.c.id == 3).values(is_deleted=True)) + await session.execute(update(link).where(link.c.id == 3).values(is_deleted=True)) + await session.commit() + + +@pytest.fixture +async def read_db(): + """Strict mode on, as with ``multi_tenant``.""" + async for session in _tenant_session(Base.metadata, strict=True, seed=seed): + yield session + + +@pytest.fixture +async def lax_read_db(): + """A single-tenant install: strict mode off.""" + async for session in _tenant_session(Base.metadata, strict=False, seed=seed): + yield session + + +@pytest.fixture +def read_sql(read_db) -> Iterator[list[tuple[str, object]]]: + """Every ``(statement, params)`` sent to ``read_db``'s engine while the test runs.""" + seen: list[tuple[str, object]] = [] + engine = read_db.bind.sync_engine + + def capture(conn, cursor, statement, params, context, executemany): + seen.append((statement, params)) + + event.listen(engine, "before_cursor_execute", capture) + try: + yield seen + finally: + event.remove(engine, "before_cursor_execute", capture) diff --git a/framework/db/tests/test_db_aliased_join_text_scope.py b/framework/db/tests/test_db_aliased_join_text_scope.py new file mode 100644 index 000000000..d8fe339ed --- /dev/null +++ b/framework/db/tests/test_db_aliased_join_text_scope.py @@ -0,0 +1,115 @@ +"""``aliased(Model, )`` and ``aliased(Model, )``: allowed only where scoped. + +A join that reads only the entity's own table besides unscoped ones is +filtered by the entity's loader criteria; any other scoped table in it is +refused like other Core reads. SQL text under an entity of a scoped table +is refused outright: no filter can read it. Both run under ``all_tenants()`` +(with ``include_deleted=True``). +""" + +import pytest +from _db_read_scope_models import RsChild, RsProj, child, proj, read_db +from simple_module_db._entity_selectables import _own_join_only, _own_rows_only +from simple_module_db.tenancy import TenantIsolationError, all_tenants, tenant_context +from sqlalchemy import ForeignKey, Integer, String, inspect, select, text +from sqlalchemy.orm import DeclarativeBase, aliased, mapped_column, with_polymorphic +from sqlalchemy.sql.selectable import Join + +__all__ = ["read_db"] + + +def _child_text(): + names = ", ".join(c.name for c in child.c) + return text(f"SELECT {names} FROM read_scope_child").columns(*child.c) + + +async def test_own_table_joined_to_an_unscoped_table_is_scoped(read_db): + child_alias = aliased(RsChild, child.join(proj, proj.c.child_id == child.c.id)) + with tenant_context("a"): + assert (await read_db.execute(select(child_alias.id))).scalars().all() == [1] + with tenant_context("b"): + assert (await read_db.execute(select(child_alias.id))).scalars().all() == [4] + + +async def test_join_reading_another_scoped_table_is_refused(read_db): + proj_alias = aliased(RsProj, proj.join(child, proj.c.child_id == child.c.id)) + with tenant_context("a"), pytest.raises(TenantIsolationError, match="read_scope_child"): + await read_db.execute(select(proj_alias.id)) + + +async def test_own_table_joined_to_its_own_alias_is_refused(read_db): + other = child.alias("other") + child_alias = aliased(RsChild, child.join(other, other.c.id == child.c.parent_id)) + with tenant_context("a"), pytest.raises(TenantIsolationError): + await read_db.execute(select(child_alias.id)) + + +async def test_refused_join_runs_under_all_tenants_and_include_deleted(read_db): + proj_alias = aliased(RsProj, proj.join(child, proj.c.child_id == child.c.id)) + stmt = select(proj_alias.id).order_by(proj_alias.id).execution_options(include_deleted=True) + with all_tenants(): + assert (await read_db.execute(stmt)).scalars().all() == [1, 4] + + +@pytest.mark.parametrize("form", ["bare", "subquery", "cte"]) +async def test_text_under_a_scoped_entity_is_refused(read_db, form): + textual = _child_text() + sel = {"bare": textual, "subquery": textual.subquery(), "cte": textual.cte()}[form] + child_alias = aliased(RsChild, sel) + with tenant_context("a"), pytest.raises(TenantIsolationError, match="SQL text"): + await read_db.execute(select(child_alias.id)) + + +async def test_text_under_a_scoped_entity_runs_under_all_tenants(read_db): + child_alias = aliased(RsChild, _child_text().subquery()) + stmt = select(child_alias.id).execution_options(include_deleted=True) + with all_tenants(): + assert sorted((await read_db.execute(stmt)).scalars()) == [1, 2, 3, 4] + + +async def test_text_under_an_unscoped_entity_runs(read_db): + textual = text("SELECT id, child_id FROM read_scope_proj").columns(proj.c.id, proj.c.child_id) + proj_alias = aliased(RsProj, textual.subquery()) + with tenant_context("a"): + assert sorted((await read_db.execute(select(proj_alias.id))).scalars()) == [1, 4] + + +class _InheritBase(DeclarativeBase): + pass + + +class _Animal(_InheritBase): + __tablename__ = "aliased_join_animal" + id = mapped_column(Integer, primary_key=True) + kind = mapped_column(String) + __mapper_args__ = {"polymorphic_on": kind, "polymorphic_identity": "animal"} # noqa: RUF012 + + +class _Dog(_Animal): + __tablename__ = "aliased_join_dog" + id = mapped_column(ForeignKey("aliased_join_animal.id"), primary_key=True) + __mapper_args__ = {"polymorphic_identity": "dog"} # noqa: RUF012 + + +@pytest.mark.parametrize( + "make", [lambda: aliased(_Dog, flat=True), lambda: with_polymorphic(_Animal, [_Dog], flat=True)] +) +def test_orm_built_flat_join_of_an_inheriting_entity_is_its_own(make): + # The ORM aliases each table of a joined-inheritance mapper once: still the entity's rows. + insp = inspect(make()) + names = {"aliased_join_animal", "aliased_join_dog"} + assert isinstance(insp.selectable, Join) + assert _own_join_only(insp, insp.selectable, names) + + +def test_subquery_over_an_inheriting_entitys_own_tables_is_its_own(): + # Both tables are _Dog's (its mapper's base and local table): not another table. + animal, dog = _Animal.__table__, _Dog.__table__ + sub = select(animal.c.id, animal.c.kind).where(animal.c.id == dog.c.id).subquery() + insp = inspect(aliased(_Dog, sub)) + names = {"aliased_join_animal", "aliased_join_dog"} + assert _own_rows_only(insp, insp.selectable, names) + other = animal.alias("other") + second = select(animal.c.id, animal.c.kind).where(animal.c.id == other.c.id).subquery() + insp = inspect(aliased(_Dog, second)) + assert not _own_rows_only(insp, insp.selectable, names) diff --git a/framework/db/tests/test_db_aliased_selectable_scope.py b/framework/db/tests/test_db_aliased_selectable_scope.py new file mode 100644 index 000000000..3f72a95ce --- /dev/null +++ b/framework/db/tests/test_db_aliased_selectable_scope.py @@ -0,0 +1,121 @@ +"""``aliased(Model, )``: scoped through ORM entities, else refused. + +The ORM renders such an entity from the selectable it was built over, so the +filter cannot rewrite what that selectable reads. Built from ORM entities it +is scoped by the loader criteria; reading only the entity's own table it is +scoped on the rows it returns; any other Core read of a scoped table inside +it is refused (``_entity_selectables``). +""" + +import pytest +from _db_read_scope_models import RsChild, RsProj, child, lax_read_db, proj, read_db +from simple_module_db.tenancy import TenantIsolationError, all_tenants, tenant_context +from sqlalchemy import exists, func, select +from sqlalchemy.orm import aliased + +__all__ = ["lax_read_db", "read_db"] + + +def _proj_over_core_child(): + """Projects whose child is tenant ``b``'s: a Core read inside the selectable.""" + inner = select(child.c.id).where(child.c.tenant_id == "b") + return aliased(RsProj, select(proj).where(proj.c.child_id.in_(inner)).subquery()) + + +@pytest.mark.parametrize("form", ["subquery", "cte"]) +async def test_core_read_inside_aliased_selectable_is_refused(read_db, form): + sub = select(proj).where(proj.c.child_id.in_(select(child.c.id))) + proj_alias = aliased(RsProj, sub.subquery() if form == "subquery" else sub.cte()) + with tenant_context("a"), pytest.raises(TenantIsolationError, match="read_scope_child"): + await read_db.execute(select(proj_alias.id)) + + +async def test_refused_shape_also_refused_on_a_single_tenant_install(lax_read_db): + # Soft-delete still applies, and the selectable cannot be filtered for it. + with pytest.raises(TenantIsolationError): + await lax_read_db.execute(select(_proj_over_core_child().id)) + + +async def test_refused_shape_runs_under_all_tenants_and_include_deleted(read_db): + stmt = select(_proj_over_core_child().id).execution_options(include_deleted=True) + with all_tenants(): + assert (await read_db.execute(stmt)).scalars().all() == [4] + + +async def test_orm_read_inside_aliased_selectable_is_scoped(read_db): + inner = select(RsChild.id).where(RsChild.tenant_id == "b") + proj_alias = aliased(RsProj, select(RsProj).where(RsProj.child_id.in_(inner)).subquery()) + with tenant_context("a"): + assert (await read_db.execute(select(proj_alias.id))).scalars().all() == [] + with tenant_context("b"): + assert (await read_db.execute(select(proj_alias.id))).scalars().all() == [4] + + +async def test_own_table_selectable_is_scoped_by_the_entity_criteria(read_db): + child_alias = aliased(RsChild, select(child).where(child.c.parent_id == 1).subquery()) + with tenant_context("a"): + assert sorted((await read_db.execute(select(child_alias.id))).scalars()) == [1, 2] + with tenant_context("b"): + assert sorted((await read_db.execute(select(child_alias.id))).scalars()) == [4] + + +async def test_own_table_with_a_nested_core_read_is_refused(read_db): + probe = exists(select(child.c.id).where(child.c.tenant_id == "b")) + child_alias = aliased(RsChild, select(child).where(probe).subquery()) + with tenant_context("a"), pytest.raises(TenantIsolationError): + await read_db.execute(select(child_alias.id)) + + +async def test_own_table_joined_to_another_scoped_table_is_refused(read_db): + other = child.alias("other") + joined = select(child).join(other, other.c.id == child.c.parent_id) + child_alias = aliased(RsChild, joined.subquery()) + with tenant_context("a"), pytest.raises(TenantIsolationError): + await read_db.execute(select(child_alias.id)) + + +async def test_aliased_selectable_via_select_from_is_not_rendered_twice(read_db): + inner = select(RsChild.id).where(RsChild.parent_id == 1) + proj_alias = aliased(RsProj, select(RsProj).where(RsProj.child_id.in_(inner)).subquery("p")) + stmt = ( + select(proj_alias.id) + .select_from(proj_alias) + .where(proj_alias.child_id.in_(select(child.c.id))) + ) + with tenant_context("a"): + assert (await read_db.execute(stmt)).scalars().all() == [1] + + +# Each combines rows before the entity's criteria filter the result. +_COMBINING = { + "group by": lambda: select(child).group_by(child.c.parent_id), + "having": lambda: select(child).having(func.count(child.c.id) > 0), + "window": lambda: select( + child, func.max(child.c.id).over(partition_by=child.c.parent_id).label("top") + ), + "aggregate": lambda: select( + func.max(child.c.id).label("id"), func.min(child.c.tenant_id).label("tenant_id") + ), + # Not a known aggregate name: an unknown function is not trusted to be row-wise. + "unlisted aggregate": lambda: select( + func.any_value(child.c.id).label("id"), func.any_value(child.c.tenant_id).label("tenant_id") + ), +} + + +@pytest.mark.parametrize("shape", list(_COMBINING)) +async def test_own_table_selectable_that_combines_rows_is_refused(read_db, shape): + child_alias = aliased(RsChild, _COMBINING[shape]().subquery()) + with tenant_context("a"), pytest.raises(TenantIsolationError, match="read_scope_child"): + await read_db.execute(select(child_alias.id)) + + +async def test_row_wise_function_and_nested_unscoped_aggregate_stay_allowed(read_db): + top_proj = select(func.max(proj.c.child_id)).scalar_subquery() + row_wise = func.coalesce(func.lower(child.c.tenant_id), "") != "z" + sub = select(child).where(func.abs(child.c.id) > 0, row_wise, child.c.id <= top_proj) + child_alias = aliased(RsChild, sub.subquery()) + with tenant_context("a"): + assert sorted((await read_db.execute(select(child_alias.id))).scalars()) == [1, 2] + with tenant_context("b"): + assert sorted((await read_db.execute(select(child_alias.id))).scalars()) == [4] diff --git a/framework/db/tests/test_db_engine_policy_resolution.py b/framework/db/tests/test_db_engine_policy_resolution.py new file mode 100644 index 000000000..d3e0c1949 --- /dev/null +++ b/framework/db/tests/test_db_engine_policy_resolution.py @@ -0,0 +1,204 @@ +"""The engine policy follows derived engines and per-model binds; tables register eagerly. + +``engine.execution_options(...)`` gives an ``OptionEngine`` proxying the +policy's engine, and a session may be bound per model (``binds={...}``): +both keep strict mode. A model's table is known to the filters as soon as +the class is mapped, before any ORM use configures the mappers. +""" + +import subprocess +import sys +import textwrap + +import pytest +from _db_dml_scope_models import ScopeDoc, ScopeProj, scope_db +from simple_module_db import model_registry +from simple_module_db._engine_policies import policy_of, session_policy +from simple_module_db.query_filter import EngineTenancy, _engine_policy, bind_engine_policy +from simple_module_db.tenancy import TenantIsolationError, no_tenant, tenant_context +from sqlalchemy import create_engine, inspect, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import Session + +__all__ = ["scope_db"] + + +def _sync_class(scope_db): + return type(scope_db.sync_session) + + +async def test_option_engine_keeps_strict_mode(scope_db): + engine = scope_db.bind.execution_options(isolation_level="SERIALIZABLE") + async with AsyncSession(engine, sync_session_class=_sync_class(scope_db)) as session: + with no_tenant(), pytest.raises(TenantIsolationError): + await session.execute(select(ScopeDoc)) + with tenant_context("b"): + assert (await session.execute(select(ScopeDoc.id))).scalars().all() == [2] + + +async def test_per_model_binds_keep_strict_mode(scope_db): + engine = scope_db.bind + binds = {ScopeDoc: engine, ScopeProj: engine} + async with AsyncSession(binds=binds, sync_session_class=_sync_class(scope_db)) as session: + with no_tenant(), pytest.raises(TenantIsolationError): + await session.execute(select(ScopeDoc)) + + +async def test_connection_bound_session_keeps_strict_mode(scope_db): + engine = scope_db.bind.execution_options(logging_token="x") + async with engine.connect() as conn: + session = AsyncSession(conn, sync_session_class=_sync_class(scope_db)) + with no_tenant(), pytest.raises(TenantIsolationError): + await session.execute(select(ScopeDoc)) + await session.close() + + +def test_policy_lookup_resolves_through_option_engines(): + engine = create_engine("sqlite://") + policy = EngineTenancy(tenant_strict=True) + bind_engine_policy(engine, policy) + derived = engine.execution_options(isolation_level="AUTOCOMMIT").execution_options(x=1) + assert policy_of(derived) is policy + assert _engine_policy[engine] is policy # the map callers import is the live one + assert session_policy(Session(binds={ScopeDoc: derived})) is policy + assert session_policy(Session()) is None + engine.dispose() + + +def test_strictest_bind_wins_for_a_session_with_several_engines(): + lax, strict = create_engine("sqlite://"), create_engine("sqlite://") + bind_engine_policy(lax, EngineTenancy(tenant_strict=False)) + bind_engine_policy(strict, EngineTenancy(tenant_strict=True)) + session = Session(binds={ScopeProj: lax, ScopeDoc: strict}) + assert session_policy(session).tenant_strict is True + lax.dispose() + strict.dispose() + + +def test_session_overriding_get_bind_keeps_the_routed_engines_policy(): + lax, strict = create_engine("sqlite://"), create_engine("sqlite://") + bind_engine_policy(lax, EngineTenancy(tenant_strict=False)) + bind_engine_policy(strict, EngineTenancy(tenant_strict=True)) + + class RoutingSession(Session): + def get_bind(self, mapper=None, **kwargs): + return strict if mapper is not None and inspect(mapper).class_ is ScopeDoc else lax + + assert session_policy(RoutingSession()).tenant_strict is True + assert session_policy(RoutingSession(bind=lax)).tenant_strict is True + lax.dispose() + strict.dispose() + + +def test_session_policy_is_memoized_per_bind(): + first, second = create_engine("sqlite://"), create_engine("sqlite://") + bind_engine_policy(first, EngineTenancy(tenant_strict=False)) + bind_engine_policy(second, EngineTenancy(tenant_strict=True)) + session = Session(bind=first) + assert session_policy(session).tenant_strict is False + session.bind = second + assert session_policy(session).tenant_strict is True + first.dispose() + second.dispose() + + +def test_policy_bound_after_a_memoized_answer_is_seen(): + engine = create_engine("sqlite://") + session = Session(bind=engine) + assert session_policy(session) is None + policy = EngineTenancy(tenant_strict=True) + bind_engine_policy(engine, policy) + assert session_policy(session) is policy + engine.dispose() + + +@pytest.mark.parametrize("bound", [True, False]) +def test_unmapped_class_among_tenant_models_is_skipped(monkeypatch, bound): + class _NotMapped: # e.g. a model whose registry was disposed + pass + + engine = create_engine("sqlite://") + policy = EngineTenancy(tenant_strict=True) + bind_engine_policy(engine, policy) + stale = {*model_registry.tenant_classes, _NotMapped} + monkeypatch.setattr(model_registry, "tenant_classes", stale) + session = Session(bind=engine) if bound else Session(binds={ScopeDoc: engine}) + assert session_policy(session) is policy + engine.dispose() + + +def test_session_without_a_policy_is_memoized_too(): + engine, other = create_engine("sqlite://"), create_engine("sqlite://") # no policy bound + calls = [] + + class CountingSession(Session): + def get_bind(self, mapper=None, **kwargs): + calls.append(mapper) + return self.bind + + session = CountingSession(bind=engine) + assert session_policy(session) is None + first = len(calls) + assert first > 0 + assert session_policy(session) is None + assert len(calls) == first # the "no policy" answer was reused + session.bind = other + assert session_policy(session) is None + assert len(calls) > first # a new bind is asked again + engine.dispose() + other.dispose() + + +_EAGER_SCRIPT = textwrap.dedent( + """ + import sqlalchemy as sa + from sqlalchemy.orm import Session + from sqlmodel import Field + from simple_module_db.base import create_module_base + from simple_module_db.listeners import attach_session_listeners + from simple_module_db.mixins import MultiTenantMixin, SoftDeleteMixin + from simple_module_db.query_filter import EngineTenancy, bind_engine_policy + from simple_module_db.tenancy import TenantIsolationError, tenant_context + + Base = create_module_base("eager") + + class Thing(Base, MultiTenantMixin, SoftDeleteMixin, table=True): + __tablename__ = "eager_thing" + id: int = Field(primary_key=True) + + engine = sa.create_engine("sqlite://") + Base.metadata.create_all(engine) + bind_engine_policy(engine, EngineTenancy(tenant_strict=True)) + + class S(Session): + pass + + attach_session_listeners(S) + t = Thing.__table__ + with engine.begin() as c: # Core seeding: never configures the mappers + c.execute(sa.insert(t), [ + {"id": 1, "tenant_id": "a", "is_deleted": False}, + {"id": 2, "tenant_id": "b", "is_deleted": False}, + {"id": 3, "tenant_id": "a", "is_deleted": True}, + ]) + stmt = sa.select(t.c.id).order_by(t.c.id) + with S(engine) as s: + with tenant_context("a"): + ids = list(s.execute(stmt).scalars()) + try: + s.execute(stmt) + raised = False + except TenantIsolationError: + raised = True + print("RESULT", ids, raised) + """ +) + + +def test_core_statement_is_scoped_before_mappers_configure(): + # A fresh interpreter: in this one the mappers are long configured. + proc = subprocess.run( + [sys.executable, "-c", _EAGER_SCRIPT], capture_output=True, text=True, timeout=120 + ) + assert proc.returncode == 0, proc.stderr + assert "RESULT [1] True" in proc.stdout, proc.stdout diff --git a/framework/db/tests/test_db_flush_persisted_owner.py b/framework/db/tests/test_db_flush_persisted_owner.py new file mode 100644 index 000000000..826e26d85 --- /dev/null +++ b/framework/db/tests/test_db_flush_persisted_owner.py @@ -0,0 +1,179 @@ +"""A flushed UPDATE / DELETE must hit a row the bound tenant owns *in the database*. + +The unit of work writes by primary key, so an object made persistent in +memory — ``make_transient_to_detached``, ``merge(load=False)``, a transient +copy given another row's key — must not carry an in-memory ``tenant_id`` onto +another tenant's row. Its owner is read from the table first; objects loaded +or inserted through the session are trusted and cost no extra read. +""" + +import pytest +from _db_dml_scope_models import ScopeDoc, captured_sql, docs, scope_db +from simple_module_db._persisted_owner import _OWNER_KEY +from simple_module_db.flush_guard import guard_flush +from simple_module_db.tenancy import TenantIsolationError, all_tenants, tenant_context +from sqlalchemy import inspect as sa_inspect +from sqlalchemy import select +from sqlalchemy.orm import make_transient, make_transient_to_detached +from sqlalchemy.orm.exc import StaleDataError + +__all__ = ["captured_sql", "scope_db"] + +B_SECRET = (2, "b", "b-secret", None) + + +def _in_memory(doc_id: int, tenant: str, name: str) -> ScopeDoc: + obj = ScopeDoc(id=doc_id, name=name, tenant_id=tenant, is_deleted=False) + make_transient_to_detached(obj) + return obj + + +@pytest.mark.parametrize("op", ["update", "delete"]) +async def test_detached_object_with_mismatched_owner_is_refused(scope_db, op): + with tenant_context("a"): + obj = _in_memory(2, "a", "b-secret") + scope_db.add(obj) + if op == "update": + obj.name = "edited" + else: + await scope_db.delete(obj) + with pytest.raises(TenantIsolationError, match="tenant 'b'"): + await scope_db.flush() + await scope_db.rollback() + assert B_SECRET in await docs(scope_db) + + +async def test_merge_without_load_with_mismatched_owner_is_refused(scope_db): + with tenant_context("a"): + merged = await scope_db.merge(_in_memory(2, "a", "b-secret"), load=False) + merged.name = "edited" + with pytest.raises(TenantIsolationError): + await scope_db.flush() + await scope_db.rollback() + assert B_SECRET in await docs(scope_db) + + +async def test_loaded_object_given_another_rows_key_is_refused(scope_db): + with tenant_context("a"): + own = await scope_db.get(ScopeDoc, 1) + assert own is not None + scope_db.expunge(own) + make_transient(own) + own.id = 2 + make_transient_to_detached(own) + scope_db.add(own) + own.name = "edited" + with pytest.raises(TenantIsolationError): + await scope_db.flush() + await scope_db.rollback() + assert B_SECRET in await docs(scope_db) + + +async def test_detached_object_of_the_bound_tenant_is_written(scope_db): + with tenant_context("a"): + own = _in_memory(1, "a", "a-doc") + scope_db.add(own) + own.note = "edited" + await scope_db.flush() + assert (1, "a", "a-doc", "edited") in await docs(scope_db) + + +async def test_missing_row_is_left_to_the_unit_of_work(scope_db): + with tenant_context("a"): + ghost = _in_memory(99, "a", "ghost") + scope_db.add(ghost) + ghost.note = "x" + with pytest.raises(StaleDataError): + await scope_db.flush() + + +async def test_all_tenants_block_is_not_checked(scope_db): + with all_tenants(): + obj = _in_memory(2, "b", "b-secret") + scope_db.add(obj) + obj.note = "admin" + await scope_db.flush() + assert (2, "b", "b-secret", "admin") in await docs(scope_db) + + +async def test_loaded_and_inserted_objects_cost_no_extra_read(scope_db, captured_sql): + with tenant_context("a"): + loaded = (await scope_db.execute(select(ScopeDoc).where(ScopeDoc.id == 1))).scalar_one() + fresh = ScopeDoc(id=5, name="new") + scope_db.add(fresh) + await scope_db.flush() + captured_sql.clear() + loaded.note = "edited" + fresh.note = "edited" + await scope_db.flush() + await scope_db.delete(fresh) + await scope_db.flush() + assert captured_sql and all(sql.startswith("UPDATE") for sql in captured_sql) + + +async def test_detached_object_costs_one_read_then_none(scope_db, captured_sql): + with tenant_context("a"): + own = _in_memory(1, "a", "a-doc") + scope_db.add(own) + own.note = "first" + captured_sql.clear() + await scope_db.flush() + own.note = "second" + await scope_db.flush() + assert [sql.split()[0] for sql in captured_sql] == ["SELECT", "UPDATE", "UPDATE"] + + +async def test_loaded_and_inserted_objects_are_not_marked(scope_db): + """Normal loads and inserts do no per-row bookkeeping; only re-attached objects do.""" + with tenant_context("a"): + loaded = (await scope_db.execute(select(ScopeDoc).where(ScopeDoc.id == 1))).scalar_one() + fresh = ScopeDoc(id=6, name="new") + scope_db.add(fresh) + await scope_db.flush() + reattached = _in_memory(3, "a", "x") + scope_db.add(reattached) + assert _OWNER_KEY not in sa_inspect(loaded).info + assert _OWNER_KEY not in sa_inspect(fresh).info + assert _OWNER_KEY in sa_inspect(reattached).info + + +async def test_missing_row_is_read_again_on_the_next_flush(scope_db, captured_sql): + with tenant_context("a"): + ghost = _in_memory(99, "a", "ghost") + scope_db.add(ghost) + ghost.note = "x" + captured_sql.clear() + await scope_db.run_sync(guard_flush) + await scope_db.run_sync(guard_flush) + assert [sql.split()[0] for sql in captured_sql] == ["SELECT", "SELECT"] + + +async def test_row_inserted_after_a_missing_read_is_checked(scope_db): + with tenant_context("a"): + obj = _in_memory(99, "a", "late") + scope_db.add(obj) + obj.note = "x" + await scope_db.run_sync(guard_flush) # no row yet: left to the unit of work + with tenant_context("b"): + await scope_db.run_sync(_insert_doc_99_for_b) + with tenant_context("a"), pytest.raises(TenantIsolationError, match="tenant 'b'"): + await scope_db.run_sync(guard_flush) + + +def _insert_doc_99_for_b(session): + table = ScopeDoc.__table__ + session.connection().execute( + table.insert().values(id=99, name="b-late", tenant_id="b", is_deleted=False) + ) + + +async def test_soft_delete_of_a_loaded_object_costs_no_read_later(scope_db, captured_sql): + with tenant_context("a"): + own = await scope_db.get(ScopeDoc, 1) + assert own is not None + await scope_db.delete(own) + await scope_db.flush() # soft delete: expunged and re-added internally + captured_sql.clear() + own.note = "after" + await scope_db.flush() + assert [sql.split()[0] for sql in captured_sql] == ["UPDATE"] diff --git a/framework/db/tests/test_db_relationship_loader_scope.py b/framework/db/tests/test_db_relationship_loader_scope.py new file mode 100644 index 000000000..d4c0b0e89 --- /dev/null +++ b/framework/db/tests/test_db_relationship_loader_scope.py @@ -0,0 +1,106 @@ +"""Every relationship loader applies the same tenant / soft-delete scope (#816). + +A plain relationship (``kids``) and one through a ``secondary`` link table +(``linked``) must load exactly the rows the scope lets the statement see, +whether the collection is loaded lazily, joined, by ``selectinload`` or by +``subqueryload``. The link table is a tenant-scoped, soft-deletable table: +its rows are filtered like the children's. +""" + +import re + +import pytest +from _db_read_scope_models import RsChild, RsParent, lax_read_db, link, read_db, read_sql +from simple_module_db.tenancy import TenantIsolationError, all_tenants, no_tenant, tenant_context +from sqlalchemy import select +from sqlalchemy.orm import joinedload, selectinload, subqueryload + +__all__ = ["lax_read_db", "read_db", "read_sql"] + +LOADERS = { + "lazy": None, + "joinedload": joinedload, + "selectinload": selectinload, + "subqueryload": subqueryload, +} + +# (context, execution options) -> expected (kids, linked) of parent 1 +SCOPES = { + "tenant": ("a", {}, [1, 2], [1]), + "tenant-include-deleted": ("a", {"include_deleted": True}, [1, 2, 3], [1, 2]), + "all-tenants": (None, {}, [1, 2, 4], [1, 2]), +} + + +async def _load(session, attr, loader, options): + stmt = select(RsParent).execution_options(**options) + if loader is not None: + stmt = stmt.options(loader(getattr(RsParent, attr))) + parent = (await session.execute(stmt)).unique().scalar_one() + ids = await session.run_sync(lambda _sync: [c.id for c in getattr(parent, attr)]) + return sorted(set(ids)) + + +@pytest.mark.parametrize("loader", list(LOADERS)) +@pytest.mark.parametrize("attr", ["kids", "linked"]) +@pytest.mark.parametrize("scope", list(SCOPES)) +async def test_relationship_loader_matrix(read_db, loader, attr, scope): + tenant, options, kids, linked = SCOPES[scope] + if loader == "lazy" and options: + pytest.skip("execution options belong to the statement, not to later lazy loads") + expected = kids if attr == "kids" else linked + with all_tenants() if tenant is None else tenant_context(tenant): + assert await _load(read_db, attr, LOADERS[loader], options) == expected + + +@pytest.mark.parametrize("loader", ["joinedload", "selectinload", "subqueryload"]) +async def test_secondary_loader_under_statement_all_tenants_option(read_db, loader): + with tenant_context("a"): + found = await _load(read_db, "linked", LOADERS[loader], {"all_tenants": True}) + assert found == [1, 2] + + +@pytest.mark.parametrize("loader", list(LOADERS)) +async def test_single_tenant_install_sees_every_live_link(lax_read_db, loader): + with no_tenant(): + assert await _load(lax_read_db, "linked", LOADERS[loader], {}) == [1, 2] + + +async def test_strict_without_tenant_refuses(read_db): + with no_tenant(), pytest.raises(TenantIsolationError): + await read_db.execute(select(RsParent).options(joinedload(RsParent.linked))) + + +async def test_join_through_secondary_relationship_is_scoped(read_db): + stmt = select(RsParent.id, RsChild.id).join(RsParent.linked).order_by(RsChild.id) + with tenant_context("a"): + assert (await read_db.execute(stmt)).all() == [(1, 1)] + + +async def test_any_through_secondary_relationship_is_scoped(read_db): + stmt = select(RsParent.id).where(RsParent.linked.any(RsChild.id == 2)) + with tenant_context("a"): + assert (await read_db.execute(stmt)).all() == [] + found = await read_db.execute(stmt.execution_options(include_deleted=True)) + assert found.all() == [(1,)] + + +async def test_secondary_criteria_are_bound_per_execution(read_db, read_sql): + """One cached compiled form; the values come from each execution's scope.""" + for tenant in ("a", "b"): + with tenant_context(tenant): + stmt = select(RsParent).options(selectinload(RsParent.linked)) + await read_db.execute(stmt.execution_options(all_tenants=True)) + read_db.expunge_all() + sql = [(statement, params) for statement, params in read_sql if link.name in statement] + assert len(sql) == 2 and sql[0][0] == sql[1][0] + assert [ + list(params).count(tenant) for (_sql, params), tenant in zip(sql, "ab", strict=True) + ] == [1, 1] + + +async def test_subqueryload_renders_its_parent_subquery_once(read_db, read_sql): + with tenant_context("a"): + await read_db.execute(select(RsParent).options(subqueryload(RsParent.kids))) + loader_sql = read_sql[-1][0] + assert len(re.findall(r"\) AS anon_1\b", loader_sql)) == 1, loader_sql diff --git a/framework/db/tests/test_db_secondary_scope_execution.py b/framework/db/tests/test_db_secondary_scope_execution.py new file mode 100644 index 000000000..150e691dc --- /dev/null +++ b/framework/db/tests/test_db_secondary_scope_execution.py @@ -0,0 +1,95 @@ +"""The ``secondary`` criteria read a scope recorded per execution, on every engine. + +``_secondary_scope`` listens on the ``Engine`` class: an engine with no policy +bound still records each execution's scope (and its options), and the scope is +cleared once the execution is done, or has failed. An UPDATE / DELETE skips the link table's +soft-delete criterion, like the main filter skips soft-delete on writes. +""" + +import asyncio + +import pytest +from _db_read_scope_models import Base, RsChild, RsParent, child, link, parent, read_db +from simple_module_db import _secondary_scope +from simple_module_db.tenancy import all_tenants, tenant_context +from sqlalchemy import create_engine, insert, select, text, update +from sqlalchemy.exc import OperationalError +from sqlalchemy.pool import StaticPool + +__all__ = ["read_db"] + + +async def test_update_through_secondary_ignores_trashed_link_rows_like_the_filter(read_db): + # Child 2 is linked by link 3 (tenant a, trashed) and link 2 (tenant b). + stmt = update(RsParent).where(RsParent.linked.any(RsChild.id == 2)).values(id=RsParent.id) + with tenant_context("a"): + assert (await read_db.execute(stmt)).rowcount == 1 + with tenant_context("c"): + assert (await read_db.execute(stmt)).rowcount == 0 # tenant criteria still apply + probe = select(RsParent.id).where(RsParent.linked.any(RsChild.id == 2)) + with tenant_context("a"): + assert (await read_db.execute(probe)).all() == [] # a read keeps soft-delete + + +async def test_scope_is_cleared_after_each_execution(read_db): + with tenant_context("a"): + await read_db.execute(select(RsParent.id).join(RsParent.linked)) + assert _secondary_scope._scope.get() is None + + async def child_task(): + return _secondary_scope._scope.get() + + assert await asyncio.create_task(child_task()) is None + + +def _unbound_engine(): + engine = create_engine("sqlite://", poolclass=StaticPool) + Base.metadata.create_all(engine) + with engine.begin() as conn: + conn.execute(insert(parent), [{"id": 1, "tenant_id": "a"}]) + conn.execute( + insert(child), + [ + {"id": 1, "parent_id": 1, "tenant_id": "a", "is_deleted": False}, + {"id": 2, "parent_id": 1, "tenant_id": "a", "is_deleted": False}, + {"id": 4, "parent_id": 1, "tenant_id": "b", "is_deleted": False}, + ], + ) + conn.execute( + insert(link), + [ + {"id": 1, "parent_id": 1, "child_id": 1, "tenant_id": "a", "is_deleted": False}, + {"id": 2, "parent_id": 1, "child_id": 2, "tenant_id": "b", "is_deleted": False}, + {"id": 3, "parent_id": 1, "child_id": 2, "tenant_id": "a", "is_deleted": True}, + ], + ) + return engine + + +def test_engine_without_a_policy_applies_only_the_secondary_conditions(): + """On a plain ``Connection`` no loader criteria apply; the link conditions do.""" + engine = _unbound_engine() + linked = select(RsChild.id).join(RsParent.linked).order_by(RsChild.id) + kids = select(RsChild.id).join(RsParent.kids).order_by(RsChild.id) + try: + with engine.connect() as conn, tenant_context("a"): + assert conn.execute(linked).scalars().all() == [1] + options = {"all_tenants": True, "include_deleted": True} + everything = conn.execute(linked.execution_options(**options)).scalars().all() + assert everything == [1, 2, 2] + # The children themselves are not filtered: tenant b's child 4 is read. + assert conn.execute(kids).scalars().all() == [1, 2, 4] + finally: + engine.dispose() + + +def test_scope_is_cleared_after_a_failed_execution(): + engine = _unbound_engine() + failing = select(RsParent.id).join(RsParent.linked).where(text("no_such_column = 1")) + try: + with engine.connect() as conn, all_tenants(): + with pytest.raises(OperationalError): + conn.execute(failing) + assert _secondary_scope._scope.get() is None + finally: + engine.dispose() diff --git a/skills/simple-module-database/SKILL.md b/skills/simple-module-database/SKILL.md index a6badacb7..850150a55 100644 --- a/skills/simple-module-database/SKILL.md +++ b/skills/simple-module-database/SKILL.md @@ -80,6 +80,8 @@ async def create_order(session: AsyncSession, payload: OrderCreate) -> Order: - **Setting `tenant_id` manually.** `MultiTenantMixin` populates it on insert from request state (`current_tenant_id`). Setting it to a *different* tenant than the active context raises `TenantIsolationError`; changing it on an existing row raises too. - **Legacy bulk APIs on tenant models.** `Session.bulk_save_objects()`, `bulk_insert_mappings()` and `bulk_update_mappings()` raise `TenantIsolationError` on a `MultiTenantMixin` model while a tenant is bound or strict mode is on; use `session.execute(insert(Model), rows)` / `session.execute(update(Model), rows)`, which are scoped. SQLite `INSERT OR REPLACE` into a tenant table and an INSERT embedded as a CTE are refused the same way. +- **`aliased(Model, )` over Core.** A selectable built with `select(Other.__table__)` (or any Core read of a tenant-scoped / soft-deletable table other than the entity's own, or of the entity's own table with `GROUP BY`, `HAVING`, an aggregate or a window function) raises `TenantIsolationError`; build it from ORM entities (`select(Other)`), which the loader criteria scope. Relationships with `secondary=Link.__table__` are scoped in every loader (lazy, joined, selectin, subquery) only when `Link` is a mapped class using the mixins; a bare association `Table` is never filtered. +- **Objects made persistent in memory.** `make_transient_to_detached()` / `merge(load=False)` objects, and any detached object passed to `session.add()`, have their row's `tenant_id` read (once) before a flush writes them; another tenant's row raises `TenantIsolationError`. Load the row instead. - **Querying `WHERE is_deleted = false` by hand.** The listener already does it. Use `execution_options(include_deleted=True)` only when you genuinely want deleted rows. ## Related skills diff --git a/uv.lock b/uv.lock index 410fc9853..943c04756 100644 --- a/uv.lock +++ b/uv.lock @@ -3271,7 +3271,7 @@ requires-dist = [ { name = "asyncpg", specifier = ">=0.30,<0.33" }, { name = "psycopg2-binary", specifier = ">=2.9,<3" }, { name = "simple-module-core", editable = "framework/core" }, - { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0,<3" }, + { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0.2,<3" }, { name = "sqlmodel", specifier = ">=0.0.22,<0.1" }, ]