fix(db): tenant scoping for bulk, nested and value-subquery DML shapes

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`;

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

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

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"]

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)

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)
 

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"]