diff --git a/CHANGELOG.md b/CHANGELOG.md index 471941d40..1558b15e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,6 +53,26 @@ All notable changes to this project are documented in this file. The format is b `account` or whose `azp` is another client (only `oidc` pins both). ### Fixed +- 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()`), + subqueries in an insert's `.values()` / multi-row values / `RETURNING` / + `on_conflict_do_update()` `set_` and `where` (whatever the insert's target; + strict mode with no tenant raises), UPDATE / DELETE embedded as a CTE, UPDATE / DELETE on an aliased Core + table, and a table read only through an UPDATE's `SET` values (`UPDATE … FROM`). +- **Behaviour change:** these shapes now raise `TenantIsolationError`: + - an INSERT into a tenant table embedded in another statement (as a CTE, or + attached with `add_cte()`), unless inside `all_tenants()`; this includes + single-tenant installs, because such rows can be neither checked nor stamped; + - SQLite `INSERT OR REPLACE` into a tenant table while a tenant is bound (use + `on_conflict_do_update()`); + - `Session.bulk_save_objects()`, `bulk_insert_mappings()` and + `bulk_update_mappings()` on a `MultiTenantMixin` model while a tenant is bound + or strict mode is on, unless inside `all_tenants()` (use + `session.execute(insert(Model), rows)` / `update(Model)`). + + A single-tenant install with nothing bound keeps running the last two. See + `docs/framework/multi-tenancy.md`. - **Behaviour change:** `LocaleMiddleware` never picks a language the browser sends with `q=0` ("not acceptable"), including through a regional tag such as `es-MX;q=0` and when `q` follows another parameter or is spelled `Q`; diff --git a/docs/database/mixins.md b/docs/database/mixins.md index 1128fe205..082f56208 100644 --- a/docs/database/mixins.md +++ b/docs/database/mixins.md @@ -110,6 +110,12 @@ for automatic stamping, or use an explicit `all_tenants` bypass for deliberate cross-tenant copies. Strict mode requires a tenant even if the source reads no tenant tables and explicitly supplies `tenant_id` (#693). +What an insert *reads* is scoped like a SELECT: the `from_select()` source and +any subquery in `.values()` or in an `on_conflict_do_update()` arm, into a +tenant table or not. SQLite `OR REPLACE` inserts into a tenant table are +refused while a tenant is bound, and so are the legacy `Session.bulk_*` +methods (see [multi-tenancy](../framework/multi-tenancy.md#what-the-filter-can-see)). + ### Bypass Without an active tenant, strict mode raises `TenantIsolationError` (`MissingTenantError`) for every insert form, even when the rows name diff --git a/docs/framework/multi-tenancy.md b/docs/framework/multi-tenancy.md index 3b71b9b08..08085be08 100644 --- a/docs/framework/multi-tenancy.md +++ b/docs/framework/multi-tenancy.md @@ -16,11 +16,12 @@ tenant a request acts for. | Operation on a `MultiTenantMixin` model | Tenant set | No tenant, strict (`multi_tenant` on) | No tenant, not strict | |---|---|---|---| | `SELECT` | filtered to the tenant | `MissingTenantError` | unfiltered | -| ORM `update()` / `delete()` | filtered to the tenant; `update().values(tenant_id=…)` raises | `MissingTenantError` | unfiltered | +| 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 | | 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 | `MissingTenantError` is a `TenantIsolationError`. Fail closed is the point: a request, job or command that forgot to establish a tenant errors instead of @@ -43,7 +44,20 @@ statement names, so all of these are scoped (#332): stamped, and so is every other Core reference — a subquery, a CTE, a scalar or `exists()` subquery (also inside a join's `ON`), each member of a `union` / `intersect` / `except`, a table an `update()`/`delete()` joins in - through its `WHERE`, and the `select` behind `insert().from_select()`; + through its `WHERE` or its `SET` values (`update(t).values(name=other.c.name)`, + an `UPDATE … FROM`), the `select` behind `insert().from_select()`, and a + subquery in an insert's `.values()` (single or multi-row), its `RETURNING` + or an `on_conflict_do_update()` arm's `set_` / `where`, whatever the insert's + target is — an untenanted table included. A DML on an aliased table + (`delete(Doc.__table__.alias("v"))`) is scoped through the alias; +- ORM **bulk UPDATE by primary key**, `session.execute(update(Doc), [{"id": …}, …])`: + each row's UPDATE gets `WHERE tenant_id = …`, so a primary key of another + tenant updates nothing. SQLAlchemy cannot synchronize the session for that + shape, so the filter does: objects of the bound tenant already loaded get + the new values, as `synchronize_session="evaluate"` would give them; +- a DML embedded in another statement as a CTE (`select(update(t).returning(…).cte())`): + its `UPDATE` / `DELETE` gets `WHERE tenant_id = …` and may not assign + `tenant_id`; - Core **joins**: an outer join's nullable side gets its predicates in the `ON` clause, so `tag.outerjoin(link, …)` keeps tags without links and never counts another tenant's links. A `full=True` join puts both sides' @@ -67,6 +81,22 @@ statement looking for a plain (non-ORM) reference to a scoped table (about 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: + +- 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 + tenant is bound: the conflicting row it deletes may be another tenant's. + Use `on_conflict_do_update()`, which is scoped (`OR IGNORE` is allowed); +- the legacy bulk methods `Session.bulk_save_objects()`, + `bulk_insert_mappings()` and `bulk_update_mappings()` on a tenant model + while scoping is in force (a tenant bound, or strict mode on). They write + without `before_flush` or `do_orm_execute`, so neither guard sees them. + Use `session.execute(insert(Model), rows)` / `session.execute(update(Model), rows)`. + On a single-tenant install with nothing bound they run unscoped, like + every other write there. + Not covered, by design: - raw SQL — `text("SELECT … FROM docs_doc")` is opaque to the filter; scope it diff --git a/framework/db/simple_module_db/_bulk_pk_update.py b/framework/db/simple_module_db/_bulk_pk_update.py new file mode 100644 index 000000000..82ed2374c --- /dev/null +++ b/framework/db/simple_module_db/_bulk_pk_update.py @@ -0,0 +1,70 @@ +"""ORM bulk UPDATE by primary key: ``session.execute(update(Model), [{"id": …}, …])``. + +SQLAlchemy runs that shape as an executemany keyed on the primary key and +does not apply loader criteria to it, so the tenant predicate is added to +the statement's ``WHERE`` instead: a row of another tenant matches nothing +and is left unchanged. + +SQLAlchemy refuses its own identity-map synchronization (``evaluate``) for a +bulk UPDATE with extra ``WHERE`` criteria, so it is switched off for the +statement and done here: objects of the bound tenant already in the session +get the new values, as ``evaluate`` would have given them. +""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy.orm import ORMExecuteState + +from simple_module_db.insert_guard import TENANT_COLUMN, _param_rows + +_NO_SYNC = (False, None) + + +def is_bulk_pk_update(execute_state: ORMExecuteState) -> bool: + return ( + execute_state.is_update + and getattr(execute_state.update_delete_options, "_dml_strategy", None) == "bulk" + ) + + +def scoped_bulk_pk_update( + execute_state: ORMExecuteState, classes: list[type], tenant_id: str +) -> Any: + stmt = execute_state.statement + for cls in classes: + stmt = stmt.where(cls.tenant_id == tenant_id) + execute_state.statement = stmt + if execute_state.execution_options.get("synchronize_session", "auto") in _NO_SYNC: + return None + result = execute_state.invoke_statement(execution_options={"synchronize_session": False}) + _synchronize(execute_state, tenant_id) + return result + + +def _synchronize(execute_state: ORMExecuteState, tenant_id: str) -> None: + """``synchronize_session="evaluate"``, limited to the bound tenant's objects.""" + rows = _param_rows(execute_state) + mapper = execute_state.update_delete_options._subject_mapper + if mapper is None or not rows: + return + pk_keys = [prop.key for prop in mapper._identity_key_props] + identity_map = execute_state.session.identity_map + for row in rows: + if any(key not in row for key in pk_keys): + continue + state = identity_map.fast_get_state( + mapper.identity_key_from_primary_key(tuple(row[key] for key in pk_keys)) + ) + if state is None or state.dict.get(TENANT_COLUMN) != tenant_id: + continue + evaluated = state.unmodified.intersection(set(row).difference(pk_keys)) + for key in evaluated: + if key in state.dict: + state.dict[key] = row[key] + state.manager.dispatch.refresh(state, None, evaluated) + state._commit(state.dict, list(evaluated)) + + +__all__ = ["is_bulk_pk_update", "scoped_bulk_pk_update"] diff --git a/framework/db/simple_module_db/_core_joins.py b/framework/db/simple_module_db/_core_joins.py index ebad7caa1..c9fe2c307 100644 --- a/framework/db/simple_module_db/_core_joins.py +++ b/framework/db/simple_module_db/_core_joins.py @@ -58,6 +58,11 @@ def is_plain_leaf(node: Any) -> bool: return isinstance(base, TableClause) +def is_scoped_leaf(node: Any, names: set[str]) -> bool: + """A plain Core table (or alias of one) whose base table is one of ``names``.""" + return is_plain_leaf(node) and _base_name(node) in names + + def _padded(leaf: Any) -> Any | None: """``leaf``'s row is the NULL padding of an outer join: its primary key is NULL.""" pk = list(leaf.primary_key) diff --git a/framework/db/simple_module_db/_core_walk.py b/framework/db/simple_module_db/_core_walk.py index 3f8b6741a..3b2200b3f 100644 --- a/framework/db/simple_module_db/_core_walk.py +++ b/framework/db/simple_module_db/_core_walk.py @@ -24,7 +24,8 @@ from sqlalchemy.sql.selectable import Alias, Join, Select, TableClause from sqlalchemy.sql.traversals import _get_children -from simple_module_db._core_joins import _base_name, _is_orm, is_plain_leaf +from simple_module_db._core_joins import _base_name, _is_orm, is_scoped_leaf +from simple_module_db.insert_guard import raw_values def orm_less(stmt: Any) -> bool: @@ -46,11 +47,21 @@ def froms_read(stmt: Any) -> list[Any]: """The FROMs a SELECT reads besides its joins, and the extra FROMs of a DML.""" if not isinstance(stmt, Select): target = stmt.table._deannotate() - return [f for f in where_froms(stmt) if f is not target] + return [f for f in (*where_froms(stmt), *set_froms(stmt)) if f is not target] froms = [*stmt.columns_clause_froms, *(f for f in stmt._from_obj if not isinstance(f, Join))] return [*froms, *(where_froms(stmt) if orm_less(stmt) else core_where_froms(stmt))] +def set_froms(stmt: Any) -> list[Any]: + """The FROMs an UPDATE's ``SET`` values pull in: ``values(name=other.c.name)``. + + Such a table is joined in as ``UPDATE ... FROM``, exactly like one named + in the ``WHERE``. A scalar subquery adds no FROM; its own SELECT is + scoped where it is visited. + """ + return [f for value in raw_values(stmt).values() for f in getattr(value, "_from_objects", ())] + + def core_where_froms(stmt: Select) -> list[Any]: """The FROMs a SELECT's ``WHERE`` pulls in through plain (non-ORM) columns. @@ -88,21 +99,28 @@ def references_plain(stmt: Any, names: set[str]) -> bool: seen: dict[int, Any] = {} while stack: el = stack.pop() - if id(el) in seen or _is_orm(el): + if id(el) in seen: + continue + if el is not stmt and _nested_dml_on(el, names): + return True # its target is checked even when ORM-annotated + if _is_orm(el): continue seen[id(el)] = el target = el if isinstance(el, (TableClause, Alias)) else _table_of(el) - if ( - isinstance(target, (TableClause, Alias)) - and is_plain_leaf(target) - and _base_name(target) in names - ): + if isinstance(target, (TableClause, Alias)) and is_scoped_leaf(target, names): return True if el.__visit_name__ != "bindparam": # a bound value: no clause children stack.extend(_children(el)) return False +_DML = ("insert", "update", "delete") + + +def _nested_dml_on(el: Any, names: set[str]) -> bool: + return el.__visit_name__ in _DML and _base_name(el.table) in names + + _SELECT_OMIT = ("_correlate", "_correlate_except") _EMPTYABLE = (tuple, list) diff --git a/framework/db/simple_module_db/bulk_guard.py b/framework/db/simple_module_db/bulk_guard.py new file mode 100644 index 000000000..7e82bd49d --- /dev/null +++ b/framework/db/simple_module_db/bulk_guard.py @@ -0,0 +1,89 @@ +"""Tenant rules for the legacy bulk APIs of ``Session``. + +``bulk_save_objects``, ``bulk_insert_mappings`` and ``bulk_update_mappings`` +write through the persistence layer directly: no ``before_flush``, no +``do_orm_execute``, so neither the flush guard nor the query filter sees +them. On a ``MultiTenantMixin`` model they are refused while tenant scoping +is in force — a tenant is bound, or strict mode is on — and allowed inside +``all_tenants()`` (a deliberate cross-tenant write) and on a non-strict +install with nothing bound, where every write is unscoped anyway. + +The 2.0 forms are scoped and are what to use instead: +``session.execute(insert(Model), rows)`` and +``session.execute(update(Model), rows)``. +""" + +from __future__ import annotations + +import functools +from collections.abc import Callable, Iterable +from typing import Any + +from sqlalchemy import inspect as sa_inspect +from sqlalchemy.orm import Session + +from simple_module_db.mixins import MultiTenantMixin +from simple_module_db.query_filter import strict_configured +from simple_module_db.tenancy import TenantIsolationError, current_tenant_id, is_all_tenants + +_GUARDED_ATTR = "_sm_bulk_guarded" + + +def _mapped_class(mapper: Any) -> type: + return sa_inspect(mapper).class_ + + +def _check(session: Session, method: str, classes: Iterable[type]) -> None: + if is_all_tenants(): + return + if current_tenant_id.get() is None and not strict_configured(session): + return + for cls in classes: + if issubclass(cls, MultiTenantMixin): + raise TenantIsolationError( + f"Session.{method}() on tenant-scoped '{cls.__name__}' is not tenant-scoped; " + "use session.execute(insert(Model), rows) / session.execute(update(Model), " + "rows), or all_tenants() for a deliberate cross-tenant write" + ) + + +def _wrap( + original: Callable[..., Any], method: str, subject: str, classes_of: Callable[..., Any] +) -> Any: + """Check ``subject`` (the first parameter, positional or by keyword) before the call.""" + + @functools.wraps(original) + def guarded(self: Session, *args: Any, **kwargs: Any) -> Any: + if args: + first, rest = args[0], args[1:] + if method == "bulk_save_objects": + first = list(first) # may be a one-shot iterator + _check(self, method, classes_of(first)) + return original(self, first, *rest, **kwargs) + if subject in kwargs: + if method == "bulk_save_objects": + kwargs[subject] = list(kwargs[subject]) + _check(self, method, classes_of(kwargs[subject])) + return original(self, **kwargs) + + return guarded + + +def install_bulk_guard(session_class: type[Session]) -> None: + """Guard the legacy bulk methods of ``session_class``; idempotent.""" + if any(base.__dict__.get(_GUARDED_ATTR) for base in session_class.__mro__): + return + for method, subject, classes_of in ( + ("bulk_save_objects", "objects", lambda objects: {type(o) for o in objects}), + ("bulk_insert_mappings", "mapper", lambda mapper: [_mapped_class(mapper)]), + ("bulk_update_mappings", "mapper", lambda mapper: [_mapped_class(mapper)]), + ): + setattr( + session_class, + method, + _wrap(getattr(session_class, method), method, subject, classes_of), + ) + setattr(session_class, _GUARDED_ATTR, True) + + +__all__ = ["install_bulk_guard"] diff --git a/framework/db/simple_module_db/core_scope.py b/framework/db/simple_module_db/core_scope.py index dc8742b3f..0aa3f28c5 100644 --- a/framework/db/simple_module_db/core_scope.py +++ b/framework/db/simple_module_db/core_scope.py @@ -17,7 +17,10 @@ columns, its FROM list and its ``WHERE`` clause. An UPDATE / DELETE gets ``WHERE`` predicates for the tables its ``WHERE`` -joins in; its target is scoped by the caller. Under strict mode with no +and its ``SET`` values join in. The target of the statement itself is +scoped by the caller; the target of a DML *nested* in it (an +``update(...).cte()`` read by a SELECT) is handed to ``nested``, which +returns that DML's own predicates or refuses it. Under strict mode with no tenant bound, ``predicates_for`` raises: every Core reference fails closed. """ @@ -46,6 +49,7 @@ class _Visitor: def __init__(self, predicates_for: Predicates, orm_covered: Any) -> None: self.predicates_for = predicates_for self.orm_covered = orm_covered + self.dml: list[Any] = [] def _preds(self, candidates: list[Any], skip: list[Any]) -> list[Any]: preds: list[Any] = [] @@ -67,27 +71,51 @@ def select(self, stmt: Any) -> None: candidates = [f for f in froms_read(stmt) if not (outer and is_excluded(f, outer))] _add_where(stmt, [*joined.where, *self._preds(candidates, joined.covered)]) - def dml(self, stmt: Any) -> None: - """The tables an UPDATE / DELETE joins in through its ``WHERE``.""" + def update_delete(self, stmt: Any) -> None: + """The tables an UPDATE / DELETE joins in through its ``WHERE`` or ``SET``.""" _add_where(stmt, self._preds(froms_read(stmt), [])) + self.dml.append(stmt) + + def insert(self, stmt: Any) -> None: + self.dml.append(stmt) def scope_core( - stmt: Any, predicates_for: Predicates, names: set[str], orm_covered: Any = None + stmt: Any, + predicates_for: Predicates, + names: set[str], + orm_covered: Any = None, + nested: Predicates | 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. """ if not names or not references_plain(stmt, names): return stmt visitor = _Visitor(predicates_for, orm_covered) - return visitors.cloned_traverse( + out = visitors.cloned_traverse( stmt, {"stop_on": _executable_options(stmt)}, - {"select": visitor.select, "update": visitor.dml, "delete": visitor.dml}, + { + "select": visitor.select, + "update": visitor.update_delete, + "delete": visitor.update_delete, + "insert": visitor.insert, + }, ) + if nested is not None: + # The traversal is post-order: the root is reached last, so it is + # only known now. The nested clones are this traversal's own. + for dml in visitor.dml: + if dml is not out: + preds = nested(dml) + if dml.__visit_name__ != "insert": + _add_where(dml, preds) + return out __all__ = ["scope_core"] diff --git a/framework/db/simple_module_db/insert_guard.py b/framework/db/simple_module_db/insert_guard.py index 953c82f70..ab0adb49b 100644 --- a/framework/db/simple_module_db/insert_guard.py +++ b/framework/db/simple_module_db/insert_guard.py @@ -38,9 +38,14 @@ def _column_name(key: Any) -> str: return getattr(key, "key", None) or getattr(key, "name", None) or str(key) -def _statement_values(execute_state: ORMExecuteState) -> dict[str, Any]: - raw = getattr(execute_state.statement, "_values", None) or {} - return {_column_name(k): _bound_value(v) for k, v in raw.items()} +def raw_values(stmt: Any) -> dict[Any, Any]: + """The unprocessed ``SET`` / ``VALUES`` mapping of a DML, keyed by column.""" + return getattr(stmt, "_values", None) or dict(getattr(stmt, "_ordered_values", None) or ()) + + +def statement_values(stmt: Any) -> dict[str, Any]: + """The ``SET`` / ``VALUES`` of a DML, ``.values()`` and ``.ordered_values()`` alike.""" + return {_column_name(k): _bound_value(v) for k, v in raw_values(stmt).items()} def _param_rows(execute_state: ORMExecuteState) -> list[dict[str, Any]]: @@ -52,7 +57,7 @@ def _param_rows(execute_state: ORMExecuteState) -> list[dict[str, Any]]: def assigned_tenant_ids(execute_state: ORMExecuteState) -> list[Any]: found = [] - stmt_values = _statement_values(execute_state) + stmt_values = statement_values(execute_state.statement) if TENANT_COLUMN in stmt_values: found.append(stmt_values[TENANT_COLUMN]) found.extend(row[TENANT_COLUMN] for row in _param_rows(execute_state) if TENANT_COLUMN in row) @@ -81,6 +86,8 @@ def guard_insert( # the bound tenant, else the install's fallback (#380). bound = current_tenant_id.get() if not bypass: + if bound is not None: + _refuse_replace(execute_state.statement) _guard_conflict_update(execute_state, bound) tenant_id = None if bypass else bound stamp = bound or fallback @@ -95,7 +102,7 @@ def guard_insert( raise _foreign_insert(value, tenant_id) rows = _param_rows(execute_state) - stmt_has_tenant = TENANT_COLUMN in _statement_values(execute_state) + stmt_has_tenant = TENANT_COLUMN in statement_values(execute_state.statement) missing = ( [r for r in rows if TENANT_COLUMN not in r] if rows else ([] if stmt_has_tenant else [None]) ) @@ -177,11 +184,31 @@ def _guard_multi_values(execute_state: ORMExecuteState, tenant_id: str | None, s return None +def _refuse_replace(stmt: Any) -> None: + """SQLite ``INSERT OR REPLACE``: a conflict deletes the existing row, whoever owns it. + + The row it replaces is never checked against the bound tenant, so the + prefix is refused on a tenant table outside ``all_tenants()``; use + ``on_conflict_do_update()``, which is scoped. + """ + for prefix, _dialect in getattr(stmt, "_prefixes", ()): + if "REPLACE" in str(prefix).upper(): + raise TenantIsolationError( + "INSERT OR REPLACE is not tenant-scoped; use on_conflict_do_update()" + ) + + +def conflict_arm(stmt: Any) -> Any: + """The ``ON CONFLICT DO UPDATE`` arm of an insert, or ``None``.""" + arm = getattr(stmt, "_post_values_clause", None) + return arm if isinstance(arm, (PgConflictUpdate, SqliteConflictUpdate)) else None + + def _guard_conflict_update(execute_state: ORMExecuteState, tenant_id: str | None) -> None: """Conflict updates are UPDATEs: preserve ownership and scope their target.""" stmt = execute_state.statement - arm = getattr(stmt, "_post_values_clause", None) - if not isinstance(arm, (PgConflictUpdate, SqliteConflictUpdate)): + arm = conflict_arm(stmt) + if arm is None: return if any( # excluded.tenant_id is harmless only once the arm is scoped below. diff --git a/framework/db/simple_module_db/insert_scope.py b/framework/db/simple_module_db/insert_scope.py new file mode 100644 index 000000000..e6d53de32 --- /dev/null +++ b/framework/db/simple_module_db/insert_scope.py @@ -0,0 +1,117 @@ +"""Scope what an INSERT reads: its SELECT source, CTEs, and subqueries in values / RETURNING. + +``insert(...).from_select(..., select(...))`` reads like any SELECT, and so +does a subquery placed in ``.values()`` (single or multi-row) or in an +``ON CONFLICT DO UPDATE`` arm's ``set_`` / ``where`` or in ``RETURNING``. +Those subqueries are scoped whatever the insert's target is — an untenanted +table is no less a place another tenant's data could be copied to. + +The values and RETURNING of an insert sit outside the reach of loader +criteria, so an ORM subquery there is deannotated first and scoped as a Core +one: every tenant / soft-deletable table it reads gets its predicates (or, +strict with no tenant, raises). +""" + +from __future__ import annotations + +from collections.abc import Callable +from copy import copy +from typing import Any + +from sqlalchemy.sql import visitors +from sqlalchemy.sql.annotation import _deep_deannotate +from sqlalchemy.sql.elements import BindParameter, ClauseElement +from sqlalchemy.util import immutabledict + +from simple_module_db.insert_guard import conflict_arm + +Scope = Callable[[Any], Any] + +_READS = frozenset({"select", "compound_select", "insert", "update", "delete"}) + + +def _reads_rows(value: Any) -> bool: + """``value`` holds a statement of its own (a scalar subquery, ``exists()``…).""" + if not isinstance(value, ClauseElement) or isinstance(value, BindParameter): + return False + return any(el.__visit_name__ in _READS for el in visitors.iterate(value)) + + +def _expr(value: Any, scope: Scope) -> Any: + if not _reads_rows(value): + return value + plain = _deep_deannotate(value) + scoped = scope(plain) + return value if scoped is plain else scoped + + +def _row(row: Any, scope: Scope) -> Any: + if isinstance(row, dict): + return {k: _expr(v, scope) for k, v in row.items()} + return type(row)(_expr(v, scope) for v in row) + + +def _arm(arm: Any, scope: Scope) -> Any: + if arm is None: + return None + values = [(k, _expr(v, scope)) for k, v in arm.update_values_to_set] + where = arm.update_whereclause + scoped_where = None if where is None else _expr(where, scope) + unchanged = all(a[1] is b[1] for a, b in zip(values, arm.update_values_to_set, strict=True)) + if unchanged and scoped_where is where: + return arm + out = copy(arm) # no generative setter for an existing conflict arm + out.update_values_to_set = values + out.update_whereclause = scoped_where + return out + + +def scope_insert(stmt: Any, scope: Scope) -> Any: + """``stmt`` with its SELECT source, CTEs, RETURNING and value subqueries scoped.""" + # The parts a Core traversal must visit: the SELECT source, an ``add_cte()`` + # CTE and RETURNING subqueries. Values and conflict arms are handled below. + # Gated, because the traversal clones even a plain insert. + if ( + getattr(stmt, "select", None) is not None + or getattr(stmt, "_independent_ctes", None) + or any(_reads_rows(c) for c in stmt._returning) + ): + stmt = scope(stmt) + if stmt._values is not None: + # The clone holds a plain dict; tenant stamping calls ``.union()`` on it. + stmt._values = immutabledict(stmt._values) + values = stmt._values + multi = stmt._multi_values + arm = conflict_arm(stmt) + returning = stmt._returning + new_returning = tuple(_expr(c, scope) for c in returning) + new_values = {k: _expr(v, scope) for k, v in values.items()} if values else values + new_multi = tuple([_row(r, scope) for r in group] for group in multi) if multi else multi + new_arm = _arm(arm, scope) + values_same = not values or all(new_values[k] is v for k, v in values.items()) + multi_same = not multi or all( + _same_row(a, b) + for ga, gb in zip(new_multi, multi, strict=True) + for a, b in zip(ga, gb, strict=True) + ) + returning_same = all(a is b for a, b in zip(new_returning, returning, strict=True)) + if values_same and multi_same and returning_same and new_arm is arm: + return stmt + out = stmt._generate() + out._values = immutabledict(new_values) if new_values else new_values + out._multi_values = new_multi + if new_arm is not arm: + out._post_values_clause = new_arm + if not returning_same: + out._returning = new_returning + return out + + +def _same_row(a: Any, b: Any) -> bool: + pairs = ( + zip(a.values(), b.values(), strict=True) if isinstance(a, dict) else zip(a, b, strict=True) + ) + return all(x is y for x, y in pairs) + + +__all__ = ["scope_insert"] diff --git a/framework/db/simple_module_db/listeners.py b/framework/db/simple_module_db/listeners.py index 1543d264b..e13eec735 100644 --- a/framework/db/simple_module_db/listeners.py +++ b/framework/db/simple_module_db/listeners.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import Session 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 from simple_module_db.mixins import AuditMixin, SoftDeleteMixin, VersionedMixin from simple_module_db.query_filter import bind_engine_policy, engine_policy, filter_statements @@ -137,6 +138,8 @@ def attach_session_listeners(session_class: type[Session]) -> None: ("do_orm_execute", audit_statement), ): event.listen(session_class, name, fn) + # The legacy bulk_* methods reach no session event: guard them directly. + install_bulk_guard(session_class) setattr(session_class, _LISTENERS_ATTACHED, True) diff --git a/framework/db/simple_module_db/model_registry.py b/framework/db/simple_module_db/model_registry.py index eb0dad1fd..a4b41ae17 100644 --- a/framework/db/simple_module_db/model_registry.py +++ b/framework/db/simple_module_db/model_registry.py @@ -51,10 +51,3 @@ def register(cls: type) -> None: @event.listens_for(Mapper, "mapper_configured") def _on_mapper_configured(mapper: Mapper, cls: type) -> None: register(cls) - - -def is_plain_table(from_obj: Any, names: set[str]) -> bool: - """A Core table (``Model.__table__``), not an ORM-annotated occurrence.""" - return getattr(from_obj, "name", None) in names and "parententity" not in getattr( - from_obj, "_annotations", {} - ) diff --git a/framework/db/simple_module_db/query_filter.py b/framework/db/simple_module_db/query_filter.py index ebc0845cd..89eb3053e 100644 --- a/framework/db/simple_module_db/query_filter.py +++ b/framework/db/simple_module_db/query_filter.py @@ -6,8 +6,9 @@ ``in_()`` / scalar subquery or ``select(func.count()).select_from(Model)`` is filtered too. Core references to ``Model.__table__`` (or an alias of it) get explicit predicates at any depth — joins, subqueries, CTEs, compound -selects, the extra FROM of an UPDATE / DELETE, an INSERT's SELECT source — -see ``core_scope``. +selects, the extra FROM of an UPDATE / DELETE (from its ``WHERE`` or its +``SET``), an INSERT's SELECT source and the subqueries in its values, and a +DML embedded as a CTE — see ``core_scope`` and ``insert_scope``. """ from __future__ import annotations @@ -20,9 +21,17 @@ from sqlalchemy.orm import ORMExecuteState, Session, with_loader_criteria from simple_module_db import model_registry as registry -from simple_module_db._core_joins import _base_name +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.core_scope import scope_core 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 ( + nested_dml_guard, + orm_covered, + scoped_names, + subquery_predicates, +) from simple_module_db.tenancy import ( ALL_TENANTS_OPTION, DEFAULT_TENANT_ID, @@ -142,9 +151,10 @@ def filter_statements(execute_state: ORMExecuteState) -> Any: named.append(mapper.class_) execute_state.statement = scope_core( execute_state.statement, - _subquery_predicates(skip_soft_delete, skip_tenant, tenant_id, strict), - _scoped_names(skip_soft_delete, skip_tenant), - _orm_covered(skip_soft_delete, skip_tenant, tenant_id), + subquery_predicates(skip_soft_delete, skip_tenant, tenant_id, strict), + scoped_names(skip_soft_delete, skip_tenant), + orm_covered(skip_soft_delete, skip_tenant, tenant_id), + nested_dml_guard(skip_tenant, tenant_id, strict), ) if not skip_soft_delete: options.extend( @@ -156,69 +166,29 @@ def filter_statements(execute_state: ORMExecuteState) -> Any: if options: execute_state.statement = execute_state.statement.options(*options) + if named and tenant_id is not None and not skip_tenant and is_bulk_pk_update(execute_state): + return scoped_bulk_pk_update(execute_state, named, tenant_id) return None def _scope_insert_source(execute_state: ORMExecuteState) -> None: - """``insert(...).from_select(..., select(t...))``: the source reads like any SELECT.""" - if getattr(execute_state.statement, "select", None) is None: - return + """What an INSERT reads — its SELECT source, its value subqueries — like any SELECT.""" options = execute_state.execution_options skip_soft_delete = options.get("include_deleted", False) skip_tenant = options.get(ALL_TENANTS_OPTION, False) or is_all_tenants() + names = scoped_names(skip_soft_delete, skip_tenant) + if not names: + return tenant_id = current_tenant_id.get() strict = strict_configured(execute_state.session) - execute_state.statement = scope_core( + predicates_for = subquery_predicates(skip_soft_delete, skip_tenant, tenant_id, strict) + nested = nested_dml_guard(skip_tenant, tenant_id, strict) + execute_state.statement = scope_insert( execute_state.statement, - _subquery_predicates(skip_soft_delete, skip_tenant, tenant_id, strict), - _scoped_names(skip_soft_delete, skip_tenant), + lambda el: scope_core(el, predicates_for, names, nested=nested), ) -def _scoped_names(skip_soft_delete: bool, skip_tenant: bool) -> set[str]: - names: set[str] = set() - if not skip_soft_delete: - names |= registry.soft_delete_table_names - if not skip_tenant: - names |= registry.tenant_table_names - return names - - -def _orm_covered(skip_soft_delete: bool, skip_tenant: bool, tenant_id: str | None) -> set[type]: - """Classes whose every predicate the loader criteria put in a join's ``ON`` (#417). - - Strict mode without a tenant gives no tenant ``ON`` predicate worth the - name, so a tenant class is covered only once a tenant is bound: a bare - reference to it keeps refusing. - """ - covered = set(registry.soft_delete_classes) if not skip_soft_delete else set() - if skip_tenant or tenant_id is not None: - covered |= registry.tenant_classes - else: - covered -= registry.tenant_classes - return covered - - -def _subquery_predicates( - skip_soft_delete: bool, skip_tenant: bool, tenant_id: str | None, strict: bool -): - def predicates_for(from_obj: Any) -> list[Any]: - # ``aliased(Model)`` / ``table.alias()`` is named after the alias, not - # the table: resolve to the underlying table, filter on the alias. - name = _base_name(from_obj) - preds = [] - if not skip_soft_delete and name in registry.soft_delete_table_names: - preds.append(from_obj.c.is_deleted.is_(False)) - if not skip_tenant and name in registry.tenant_table_names: - if tenant_id is not None: - preds.append(from_obj.c.tenant_id == tenant_id) - elif strict: - raise missing_tenant_error(name, "SELECT") - return preds - - return predicates_for - - def _dml_target(execute_state: ORMExecuteState, names: set[str]) -> list[Any]: """The target of a Core UPDATE / DELETE on ``Model.__table__``, if scoped. @@ -228,7 +198,8 @@ def _dml_target(execute_state: ORMExecuteState, names: set[str]) -> list[Any]: if execute_state.is_select: return [] target = getattr(execute_state.statement, "table", None) - return [target] if registry.is_plain_table(target, names) else [] + # ``delete(Model.__table__.alias("v"))`` is named after the alias. + return [target] if target is not None and is_scoped_leaf(target, names) else [] def _tenant_criteria( @@ -242,7 +213,7 @@ def _tenant_criteria( if not strict: return [] if named or core: - raise missing_tenant_error(named[0].__name__ if named else core[0].name, op) + raise missing_tenant_error(named[0].__name__ if named else str(_base_name(core[0])), op) # Indirect references (joins, subqueries) with no tenant bound match # nothing — tenant_id is NOT NULL — rather than every tenant. return [ diff --git a/framework/db/simple_module_db/scope_predicates.py b/framework/db/simple_module_db/scope_predicates.py new file mode 100644 index 000000000..1794bb003 --- /dev/null +++ b/framework/db/simple_module_db/scope_predicates.py @@ -0,0 +1,102 @@ +"""The predicates ``core_scope`` attaches, built from one statement's tenant scope. + +``query_filter`` decides *whether* a statement is scoped (bound tenant, strict +mode, the ``all_tenants`` / ``include_deleted`` bypasses); this module turns +that decision into the callables ``core_scope`` applies to each Core +reference it finds: a table read (``subquery_predicates``) and a DML nested +inside another statement, such as an ``update(...).cte()`` under a SELECT +(``nested_dml_guard``). +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from simple_module_db import model_registry as registry +from simple_module_db._core_joins import _base_name +from simple_module_db.insert_guard import TENANT_COLUMN, statement_values +from simple_module_db.tenancy import TenantIsolationError, missing_tenant_error + + +def scoped_names(skip_soft_delete: bool, skip_tenant: bool) -> set[str]: + names: set[str] = set() + if not skip_soft_delete: + names |= registry.soft_delete_table_names + if not skip_tenant: + names |= registry.tenant_table_names + return names + + +def orm_covered(skip_soft_delete: bool, skip_tenant: bool, tenant_id: str | None) -> set[type]: + """Classes whose every predicate the loader criteria put in a join's ``ON`` (#417). + + Strict mode without a tenant gives no tenant ``ON`` predicate worth the + name, so a tenant class is covered only once a tenant is bound: a bare + reference to it keeps refusing. + """ + covered = set(registry.soft_delete_classes) if not skip_soft_delete else set() + if skip_tenant or tenant_id is not None: + covered |= registry.tenant_classes + else: + covered -= registry.tenant_classes + return covered + + +def subquery_predicates( + skip_soft_delete: bool, skip_tenant: bool, tenant_id: str | None, strict: bool +) -> Callable[[Any], list[Any]]: + def predicates_for(from_obj: Any) -> list[Any]: + # ``aliased(Model)`` / ``table.alias()`` is named after the alias, not + # the table: resolve to the underlying table, filter on the alias. + name = _base_name(from_obj) + preds = [] + if not skip_soft_delete and name in registry.soft_delete_table_names: + preds.append(from_obj.c.is_deleted.is_(False)) + if not skip_tenant and name in registry.tenant_table_names: + if tenant_id is not None: + preds.append(from_obj.c.tenant_id == tenant_id) + elif strict: + raise missing_tenant_error(name, "SELECT") + return preds + + return predicates_for + + +def nested_dml_guard( + skip_tenant: bool, tenant_id: str | None, strict: bool +) -> Callable[[Any], list[Any]]: + """Rules for an INSERT / UPDATE / DELETE embedded in another statement (a CTE). + + Such a DML never reaches the filter as its own statement, so its target + is handled here: an UPDATE / DELETE gets ``tenant_id = `` (and + may not assign ``tenant_id``), and an INSERT on a tenant table is refused + — its rows could be neither checked nor stamped. Strict mode with no + tenant raises; ``all_tenants`` leaves them alone. Soft-delete never + applies to a DML target. + """ + + def check(stmt: Any) -> list[Any]: + name = _base_name(stmt.table) + if skip_tenant or name not in registry.tenant_table_names: + return [] + op = stmt.__visit_name__.upper() + if op == "INSERT": + # Refused with or without a bound tenant: its rows can be neither + # checked nor stamped with the fallback tenant. + raise TenantIsolationError( + f"INSERT into tenant-scoped '{name}' inside another statement is not " + "tenant-scoped; run it as a statement of its own" + ) + if tenant_id is None: + if strict: + raise missing_tenant_error(name, op) + return [] + if op == "UPDATE" and TENANT_COLUMN in statement_values(stmt): + raise TenantIsolationError("Cannot change tenant_id of existing rows") + return [stmt.table.c.tenant_id == tenant_id] + + return check + + +__all__ = ["nested_dml_guard", "orm_covered", "scoped_names", "subquery_predicates"] diff --git a/framework/db/tests/_db_dml_scope_models.py b/framework/db/tests/_db_dml_scope_models.py new file mode 100644 index 000000000..eabd0c2a7 --- /dev/null +++ b/framework/db/tests/_db_dml_scope_models.py @@ -0,0 +1,99 @@ +"""Models and seed for the DML / INSERT shape tests (``test_db_dml_*``, ``test_db_insert_*``). + +Tenant ``a`` owns doc 1 (live) and doc 4 (trashed); tenant ``b`` owns doc 2 +(live), doc 3 (trashed) and its only ``Other`` row. ``Proj`` is untenanted. +""" + +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, select +from sqlmodel import Field + +Base = create_module_base("dml_scope") + + +class ScopeProj(Base, table=True): # ty: ignore[unsupported-base] + __tablename__ = "dml_scope_proj" + id: int = Field(primary_key=True) + note: str | None = None + + +class ScopeDoc(Base, MultiTenantMixin, SoftDeleteMixin, table=True): # ty: ignore[unsupported-base] + __tablename__ = "dml_scope_doc" + id: int = Field(primary_key=True) + name: str = "d" + note: str | None = None + + +class ScopeOther(Base, MultiTenantMixin, table=True): # ty: ignore[unsupported-base] + __tablename__ = "dml_scope_other" + id: int = Field(primary_key=True) + name: str = "o" + + +doc, proj, other = ScopeDoc.__table__, ScopeProj.__table__, ScopeOther.__table__ + + +async def seed(session): + session.add_all([ScopeProj(id=1, note="p1"), ScopeProj(id=2, note="p2")]) + await session.flush() + with tenant_context("a"): + session.add_all([ScopeDoc(id=1, name="a-doc"), ScopeDoc(id=4, name="a-trash")]) + await session.flush() + with tenant_context("b"): + session.add_all([ScopeDoc(id=2, name="b-secret"), ScopeDoc(id=3, name="b-trash")]) + session.add(ScopeOther(id=1, name="b-other")) + await session.flush() + with all_tenants(): + for trashed in (await session.execute(select(ScopeDoc).where(ScopeDoc.id > 2))).scalars(): + trashed.is_deleted = True + await session.flush() + await session.commit() + + +@pytest.fixture +async def scope_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_db(): + """A single-tenant install: strict mode off.""" + async for session in _tenant_session(Base.metadata, strict=False, seed=seed): + yield session + + +async def docs(session): + """Every doc row as ``(id, tenant_id, name, note)``, ignoring both filters.""" + stmt = select(doc.c.id, doc.c.tenant_id, doc.c.name, doc.c.note).order_by(doc.c.id) + stmt = stmt.execution_options(all_tenants=True, include_deleted=True) + return [tuple(r) for r in (await session.execute(stmt)).all()] + + +async def projs(session): + return [tuple(r) for r in (await session.execute(select(proj).order_by(proj.c.id))).all()] + + +@pytest.fixture +def captured_sql(scope_db): + """The SQL statements ``scope_db``'s engine sends, in order.""" + found: list[str] = [] + engine = scope_db.bind.sync_engine + + def capture(conn, cursor, statement, params, context, executemany): + found.append(statement) + + event.listen(engine, "before_cursor_execute", capture) + yield found + event.remove(engine, "before_cursor_execute", capture) + + +def requires_sqlite(session, reason: str) -> None: + """Skip unless ``session`` runs on SQLite (dialect-specific syntax).""" + if session.bind.dialect.name != "sqlite": + pytest.skip(reason) diff --git a/framework/db/tests/test_db_dml_target_scope.py b/framework/db/tests/test_db_dml_target_scope.py new file mode 100644 index 000000000..0e93701b8 --- /dev/null +++ b/framework/db/tests/test_db_dml_target_scope.py @@ -0,0 +1,136 @@ +"""UPDATE / DELETE shapes whose target or source the tenant filter has to resolve. + +* ORM bulk UPDATE by primary key (``execute(update(M), [{"id": ...}])``); +* ``update().ordered_values()`` assigning ``tenant_id``; +* a DML on an aliased Core table (``delete(M.__table__.alias("v"))``); +* a table read only through an UPDATE's ``SET`` (``UPDATE ... FROM``). +""" + +import pytest +from _db_dml_scope_models import ScopeDoc, ScopeOther, ScopeProj, doc, docs, other, proj, scope_db +from simple_module_db.tenancy import TenantIsolationError, all_tenants, no_tenant, tenant_context +from sqlalchemy import delete, select, update + +__all__ = ["scope_db"] + +# ``UPDATE ... FROM`` with no join condition is the shape under test. +_cartesian = pytest.mark.filterwarnings("ignore:UPDATE statement has a cartesian product") + + +async def test_bulk_pk_update_leaves_other_tenant_rows(scope_db): + with tenant_context("a"): + await scope_db.execute(update(ScopeDoc), [{"id": 2, "note": "x"}, {"id": 1, "note": "y"}]) + rows = await docs(scope_db) + assert (2, "b", "b-secret", None) in rows + assert (1, "a", "a-doc", "y") in rows + + +async def test_bulk_pk_update_synchronizes_own_tenant_objects(scope_db): + with tenant_context("a"): + own = await scope_db.get(ScopeDoc, 1) + await scope_db.execute(update(ScopeDoc), [{"id": 1, "note": "synced"}]) + assert own.note == "synced" + + +@pytest.mark.parametrize("sync", [False, "evaluate"]) +async def test_bulk_pk_update_explicit_synchronize(scope_db, sync): + stmt = update(ScopeDoc).execution_options(synchronize_session=sync) + with tenant_context("a"): + await scope_db.execute(stmt, [{"id": 2, "note": "x"}, {"id": 1, "note": "y"}]) + rows = await docs(scope_db) + assert (2, "b", "b-secret", None) in rows and (1, "a", "a-doc", "y") in rows + + +async def test_bulk_pk_update_reaches_trashed_rows_of_own_tenant(scope_db): + with tenant_context("a"): + await scope_db.execute(update(ScopeDoc), [{"id": 4, "note": "restored"}]) + assert (4, "a", "a-trash", "restored") in await docs(scope_db) + + +async def test_bulk_pk_update_unbound_strict_fails_closed(scope_db): + with no_tenant(), pytest.raises(TenantIsolationError): + await scope_db.execute(update(ScopeDoc), [{"id": 2, "note": "x"}]) + + +async def test_bulk_pk_update_all_tenants(scope_db): + with all_tenants(): + await scope_db.execute(update(ScopeDoc), [{"id": 2, "note": "x"}]) + assert (2, "b", "b-secret", "x") in await docs(scope_db) + + +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +async def test_ordered_values_cannot_move_rows(scope_db, orm): + column = ScopeDoc.tenant_id if orm else doc.c.tenant_id + target = ScopeDoc if orm else doc + stmt = update(target).ordered_values((column, "b")) + with tenant_context("a"), pytest.raises(TenantIsolationError): + await scope_db.execute(stmt.execution_options(synchronize_session=False)) + assert (1, "a", "a-doc", None) in await docs(scope_db) + + +async def test_ordered_values_without_tenant_is_scoped(scope_db): + with tenant_context("a"): + await scope_db.execute(update(doc).ordered_values((doc.c.note, "z"))) + rows = await docs(scope_db) + assert {r[0] for r in rows if r[3] == "z"} == {1, 4} + + +@pytest.mark.parametrize("operation", ["update", "delete"]) +async def test_aliased_core_target_is_scoped(scope_db, operation): + view = doc.alias("v") + stmt = update(view).values(note="hit") if operation == "update" else delete(view) + with tenant_context("a"): + result = await scope_db.execute(stmt) + assert result.rowcount == 2 # docs 1 and 4: trashed rows are writable + assert (2, "b", "b-secret", None) in await docs(scope_db) + + +async def test_aliased_core_target_unbound_strict_fails_closed(scope_db): + with no_tenant(), pytest.raises(TenantIsolationError): + await scope_db.execute(delete(doc.alias("v"))) + + +async def test_aliased_core_target_cannot_move_rows(scope_db): + view = doc.alias("v") + with tenant_context("a"), pytest.raises(TenantIsolationError): + await scope_db.execute(update(view).values(tenant_id="b")) + + +@_cartesian +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +async def test_update_from_set_source_is_scoped(scope_db, orm): + source = ScopeOther.name if orm else other.c.name + stmt = update(doc).where(doc.c.id == 1).values(note=source) + with tenant_context("a"): + await scope_db.execute(stmt) + assert (1, "a", "a-doc", None) in await docs(scope_db) + + +@_cartesian +async def test_update_from_set_source_own_tenant(scope_db): + with tenant_context("b"): + await scope_db.execute(update(doc).where(doc.c.id == 2).values(note=other.c.name)) + assert (2, "b", "b-secret", "b-other") in await docs(scope_db) + + +async def test_update_from_set_source_unbound_strict_fails_closed(scope_db): + stmt = update(ScopeOther).values(name=select(doc.c.name).where(doc.c.id == 2).scalar_subquery()) + with no_tenant(), pytest.raises(TenantIsolationError): + await scope_db.execute(stmt) + + +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +@pytest.mark.parametrize("kind", ["update", "delete"]) +async def test_dml_returning_subquery_is_scoped(scope_db, orm, kind): + secret = ( + select(ScopeDoc.name).where(ScopeDoc.id == 2) + if orm + else select(doc.c.name).where(doc.c.id == 2) + ).scalar_subquery() + target = ScopeProj if orm else proj + stmt = update(target).values(note="x") if kind == "update" else delete(target) + with tenant_context("a"): + result = await scope_db.execute( + stmt.where(target.id == 1 if orm else proj.c.id == 1).returning(secret) + ) + assert result.scalar_one() is None diff --git a/framework/db/tests/test_db_insert_value_subqueries.py b/framework/db/tests/test_db_insert_value_subqueries.py new file mode 100644 index 000000000..407aead3a --- /dev/null +++ b/framework/db/tests/test_db_insert_value_subqueries.py @@ -0,0 +1,133 @@ +"""A subquery inside an INSERT's values reads like any SELECT: tenant and soft-delete scoped. + +Covers ``.values()``, multi-row ``.values([...])`` and the ``set_`` / ``where`` +of an ``ON CONFLICT DO UPDATE`` arm, Core and ORM subqueries alike, whatever +the insert's target is. +""" + +import pytest +from _db_dml_scope_models import ( + ScopeDoc, + ScopeProj, + doc, + docs, + lax_db, + projs, + requires_sqlite, + scope_db, +) +from simple_module_db.tenancy import TenantIsolationError, all_tenants, no_tenant, tenant_context +from sqlalchemy import String, cast, func, insert, select +from sqlalchemy.dialects.sqlite import insert as sqlite_insert + +__all__ = ["lax_db", "scope_db"] + + +def _name_of(doc_id, orm): + if orm: + return select(ScopeDoc.name).where(ScopeDoc.id == doc_id).scalar_subquery() + return select(doc.c.name).where(doc.c.id == doc_id).scalar_subquery() + + +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +async def test_values_subquery_into_tenant_table(scope_db, orm): + with tenant_context("a"): + await scope_db.execute(insert(ScopeDoc).values(id=10, note=_name_of(2, orm))) + await scope_db.execute(insert(ScopeDoc).values(id=11, note=_name_of(1, orm))) + rows = await docs(scope_db) + assert (10, "a", "d", None) in rows + assert (11, "a", "d", "a-doc") in rows + + +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +async def test_values_subquery_into_untenanted_table(scope_db, orm): + with tenant_context("a"): + await scope_db.execute(insert(ScopeProj).values(id=10, note=_name_of(2, orm))) + await scope_db.execute(insert(ScopeProj).values(id=11, note=_name_of(1, orm))) + assert (await projs(scope_db))[2:] == [(10, None), (11, "a-doc")] + + +async def test_values_subquery_skips_trashed_rows(scope_db): + with tenant_context("a"): + await scope_db.execute(insert(ScopeProj).values(id=10, note=_name_of(4, orm=False))) + await scope_db.execute( + insert(ScopeProj) + .values(id=11, note=_name_of(4, orm=True)) + .execution_options(include_deleted=True) + ) + assert (await projs(scope_db))[2:] == [(10, None), (11, "a-trash")] + + +async def test_values_aggregate_subquery_counts_own_tenant(scope_db): + count = select(func.count()).select_from(doc).scalar_subquery() + with tenant_context("b"): + await scope_db.execute(insert(ScopeProj).values(id=10, note=cast(count, String))) + assert (await projs(scope_db))[2] == (10, "1") + + +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +async def test_multi_values_subquery(scope_db, orm): + rows = [{"id": 10, "note": _name_of(2, orm)}, {"id": 11, "note": _name_of(1, orm)}] + with tenant_context("a"): + await scope_db.execute(insert(ScopeProj).values(rows)) + assert (await projs(scope_db))[2:] == [(10, None), (11, "a-doc")] + + +@pytest.mark.parametrize("target", [ScopeProj, ScopeDoc], ids=["untenanted", "tenant"]) +async def test_values_subquery_unbound_strict_fails_closed(scope_db, target): + with no_tenant(), pytest.raises(TenantIsolationError): + await scope_db.execute(insert(target).values(id=12, note=_name_of(2, orm=False))) + + +async def test_values_subquery_unbound_lax_reads_unfiltered(lax_db): + await lax_db.execute(insert(ScopeProj).values(id=12, note=_name_of(2, orm=False))) + assert (await projs(lax_db))[2] == (12, "b-secret") + + +async def test_values_subquery_all_tenants_reads_every_tenant(scope_db): + with tenant_context("a"), all_tenants(): + await scope_db.execute(insert(ScopeProj).values(id=13, note=_name_of(2, orm=True))) + assert (await projs(scope_db))[2] == (13, "b-secret") + + +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +async def test_upsert_set_subquery(scope_db, orm): + requires_sqlite(scope_db, "SQLite upsert syntax") + stmt = sqlite_insert(ScopeDoc).values(id=1, name="x") + stmt = stmt.on_conflict_do_update(index_elements=["id"], set_={"note": _name_of(2, orm)}) + with tenant_context("a"): + await scope_db.execute(stmt) + assert (1, "a", "a-doc", None) in await docs(scope_db) + + +async def test_upsert_where_subquery(scope_db): + requires_sqlite(scope_db, "SQLite upsert syntax") + probe = select(doc.c.id).where(doc.c.name == "b-secret").exists() + stmt = sqlite_insert(ScopeDoc).values(id=1, name="x") + stmt = stmt.on_conflict_do_update(index_elements=["id"], set_={"note": "hit"}, where=probe) + with tenant_context("a"): + await scope_db.execute(stmt) + assert (1, "a", "a-doc", None) in await docs(scope_db) + + +async def test_plain_values_unchanged(scope_db): + stmt = insert(ScopeProj).values(id=14, note="plain") + with tenant_context("a"): + await scope_db.execute(stmt) + assert (await projs(scope_db))[2] == (14, "plain") + + +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +async def test_returning_subquery_is_scoped(scope_db, orm): + with tenant_context("a"): + result = await scope_db.execute( + insert(ScopeProj).values(id=15, note="r").returning(_name_of(2, orm)) + ) + assert result.scalar_one() is None + + +async def test_returning_subquery_unbound_strict_fails_closed(scope_db): + with no_tenant(), pytest.raises(TenantIsolationError): + await scope_db.execute( + insert(ScopeProj).values(id=16, note="r").returning(_name_of(2, False)) + ) diff --git a/framework/db/tests/test_db_nested_dml_and_bulk_api.py b/framework/db/tests/test_db_nested_dml_and_bulk_api.py new file mode 100644 index 000000000..20ebee3c9 --- /dev/null +++ b/framework/db/tests/test_db_nested_dml_and_bulk_api.py @@ -0,0 +1,176 @@ +"""DML embedded as a CTE, ``INSERT OR REPLACE`` and the legacy ``Session.bulk_*`` APIs. + +SQLite cannot run a data-modifying CTE, so those tests assert on the SQL the +engine is sent; on Postgres the statement runs and the rows are checked too. +""" + +import pytest +from _db_dml_scope_models import ( + ScopeDoc, + ScopeProj, + captured_sql, + doc, + docs, + lax_db, + requires_sqlite, + scope_db, +) +from simple_module_db.tenancy import TenantIsolationError, all_tenants, no_tenant, tenant_context +from sqlalchemy import delete, insert, select, update +from sqlalchemy.exc import DBAPIError + +__all__ = ["captured_sql", "lax_db", "scope_db"] + + +async def _run(session, stmt): + try: + await session.execute(stmt) + except DBAPIError: + if session.bind.dialect.name != "sqlite": + raise + await session.rollback() + + +def _scoped_where(sql, keyword): + """The DML inside the CTE has a ``tenant_id`` predicate before its RETURNING.""" + found = [q.split(keyword, 1)[1] for q in sql if keyword in q] + return bool(found) and "tenant_id" in found[0].split("RETURNING")[0] + + +@pytest.mark.parametrize("operation", ["update", "delete"]) +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +async def test_dml_cte_target_is_scoped(scope_db, captured_sql, operation, orm): + target = ScopeDoc if orm else doc + returning = ScopeDoc.id if orm else doc.c.id + stmt = update(target).values(note="cte") if operation == "update" else delete(target) + with tenant_context("a"): + await _run(scope_db, select(stmt.returning(returning).cte("w").c.id)) + assert _scoped_where(captured_sql, f"{operation.upper()} "), captured_sql + assert (2, "b", "b-secret", None) in await docs(scope_db) + + +async def test_dml_cte_unbound_strict_fails_closed(scope_db): + cte = update(doc).values(note="cte").returning(doc.c.id).cte("w") + with no_tenant(), pytest.raises(TenantIsolationError): + await scope_db.execute(select(cte.c.id)) + + +async def test_dml_cte_cannot_move_rows(scope_db): + cte = update(doc).values(tenant_id="b").returning(doc.c.id).cte("w") + with tenant_context("a"), pytest.raises(TenantIsolationError): + await scope_db.execute(select(cte.c.id)) + + +@pytest.mark.parametrize("orm", [False, True], ids=["core", "orm"]) +async def test_insert_cte_is_refused(scope_db, orm): + target = ScopeDoc if orm else doc + returning = ScopeDoc.id if orm else doc.c.id + cte = insert(target).values(id=30, tenant_id="b").returning(returning).cte("i") + with tenant_context("a"), pytest.raises(TenantIsolationError): + await scope_db.execute(select(cte.c.id)) + + +async def test_insert_cte_is_refused_unbound_on_single_tenant_install(lax_db): + cte = insert(doc).values(id=31, name="x").returning(doc.c.id).cte("i") + with pytest.raises(TenantIsolationError): + await lax_db.execute(select(cte.c.id)) + + +async def test_dml_cte_all_tenants_is_left_alone(scope_db, captured_sql): + cte = update(doc).values(note="cte").returning(doc.c.id).cte("w") + with all_tenants(): + await _run(scope_db, select(cte.c.id)) + assert not _scoped_where(captured_sql, "UPDATE ") + + +@pytest.mark.parametrize("prefix", ["OR REPLACE", "or replace"]) +async def test_insert_or_replace_is_refused(scope_db, prefix): + stmt = insert(doc).prefix_with(prefix).values(id=2, name="replacement") + with tenant_context("a"), pytest.raises(TenantIsolationError): + await scope_db.execute(stmt) + assert (2, "b", "b-secret", None) in await docs(scope_db) + + +async def test_insert_or_ignore_is_allowed(scope_db): + requires_sqlite(scope_db, "SQLite prefix") + with tenant_context("a"): + await scope_db.execute(insert(doc).prefix_with("OR IGNORE").values(id=2, name="x")) + assert (2, "b", "b-secret", None) in await docs(scope_db) + + +async def test_insert_or_replace_all_tenants(scope_db): + requires_sqlite(scope_db, "SQLite prefix") + stmt = insert(doc).prefix_with("OR REPLACE").values(id=2, name="new", tenant_id="b") + with all_tenants(): + await scope_db.execute(stmt) + assert (2, "b", "new", None) in await docs(scope_db) + + +_LEGACY = { + "bulk_update_mappings": lambda s: s.bulk_update_mappings(ScopeDoc, [{"id": 2, "note": "x"}]), + "bulk_insert_mappings": lambda s: s.bulk_insert_mappings( + ScopeDoc, [{"id": 20, "name": "x", "tenant_id": "b"}] + ), + "bulk_save_objects": lambda s: s.bulk_save_objects([ScopeDoc(id=20, name="x", tenant_id="b")]), +} + + +@pytest.mark.parametrize("method", sorted(_LEGACY)) +async def test_legacy_bulk_api_refused_while_scoped(scope_db, method): + with tenant_context("a"), pytest.raises(TenantIsolationError, match=method): + await scope_db.run_sync(_LEGACY[method]) + with no_tenant(), pytest.raises(TenantIsolationError): + await scope_db.run_sync(_LEGACY[method]) + assert [r[0] for r in await docs(scope_db)] == [1, 2, 3, 4] + + +@pytest.mark.parametrize("method", sorted(_LEGACY)) +async def test_legacy_bulk_api_allowed_in_all_tenants(scope_db, method): + with all_tenants(): + await scope_db.run_sync(_LEGACY[method]) + await scope_db.flush() + + +async def test_legacy_bulk_api_allowed_unbound_on_single_tenant_install(lax_db): + await lax_db.run_sync(_LEGACY["bulk_update_mappings"]) + assert (2, "b", "b-secret", "x") in await docs(lax_db) + + +async def test_legacy_bulk_api_untenanted_model_unaffected(scope_db): + with tenant_context("a"): + await scope_db.run_sync(lambda s: s.bulk_insert_mappings(ScopeProj, [{"id": 9}])) + + +async def test_legacy_bulk_api_keyword_arguments(scope_db): + with tenant_context("a"): + await scope_db.run_sync( + lambda s: s.bulk_insert_mappings(mapper=ScopeProj, mappings=[{"id": 9}]) + ) + await scope_db.run_sync(lambda s: s.bulk_save_objects(objects=iter([ScopeProj(id=10)]))) + with pytest.raises(TenantIsolationError): + await scope_db.run_sync( + lambda s: s.bulk_insert_mappings(mapper=ScopeDoc, mappings=[{"id": 20}]) + ) + with pytest.raises(TenantIsolationError): + await scope_db.run_sync(lambda s: s.bulk_save_objects(objects=iter([ScopeDoc(id=21)]))) + + +async def test_insert_add_cte_dml_is_scoped(scope_db, captured_sql): + cte = update(doc).values(note="cte").returning(doc.c.id).cte("w") + stmt = insert(ScopeProj.__table__).values(id=7, note="z").add_cte(cte) + with tenant_context("a"): + await _run(scope_db, stmt) + assert _scoped_where(captured_sql, "UPDATE "), captured_sql + + +async def test_insert_add_cte_insert_on_tenant_table_is_refused(scope_db): + cte = insert(doc).values(id=32, tenant_id="a").returning(doc.c.id).cte("i") + stmt = insert(ScopeProj.__table__).values(id=8, note="z").add_cte(cte) + with tenant_context("a"), pytest.raises(TenantIsolationError): + await scope_db.execute(stmt) + + +async def test_insert_cte_all_tenants_is_left_alone(scope_db): + cte = insert(doc).values(id=33, tenant_id="b").returning(doc.c.id).cte("i") + with all_tenants(): + await _run(scope_db, select(cte.c.id)) diff --git a/scripts/bench_core_scope.py b/scripts/bench_core_scope.py index 12d4daaf8..b21c7e397 100644 --- a/scripts/bench_core_scope.py +++ b/scripts/bench_core_scope.py @@ -20,7 +20,7 @@ from simple_module_db.base import create_module_base from simple_module_db.core_scope import scope_core from simple_module_db.mixins import MultiTenantMixin, SoftDeleteMixin -from simple_module_db.query_filter import _orm_covered, _scoped_names, _subquery_predicates +from simple_module_db.scope_predicates import orm_covered, scoped_names, subquery_predicates from sqlalchemy import exists, func, select from sqlalchemy.orm import aliased, configure_mappers, selectinload from sqlmodel import Field, Relationship @@ -88,9 +88,9 @@ def main(number: int = 2000) -> None: configure_mappers() # fills model_registry's scoped table names current_tenant_id.set("A") - preds = _subquery_predicates(False, False, "A", False) - names = _scoped_names(False, False) - covered = _orm_covered(False, False, "A") + preds = subquery_predicates(False, False, "A", False) + names = scoped_names(False, False) + covered = orm_covered(False, False, "A") def call(stmt: Any) -> Any: return scope_core(stmt, preds, names, covered) diff --git a/skills/simple-module-database/SKILL.md b/skills/simple-module-database/SKILL.md index 608784c9e..a6badacb7 100644 --- a/skills/simple-module-database/SKILL.md +++ b/skills/simple-module-database/SKILL.md @@ -79,6 +79,7 @@ async def create_order(session: AsyncSession, payload: OrderCreate) -> Order: ## Pitfalls - **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. - **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