umbragit/UmbraTrans
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
79b34840836c4b55f24ba91c941a1bebdca3df47
UmbraTrans/umbratrans_web/postgres_repository.py
umbra audio 79b3484083 Reach the tenant tables by name, not by a filter somebody remembers 
Every project statement now says `projects` unqualified and takes its
organization through search_path. A shared table filtered by
`organization_id = %s` is a filter somebody has to remember, and there is always
a write path that would not have it; an unqualified name resolves into the
caller's own schema or it does not resolve at all. The strongest possible
mistake - forgetting to scope a query - now raises instead of returning a wider
result set.

Project.organization_id survives in the model and is filled in from the schema
the answer came from. Callers still see which tenant a row belongs to; it is
derived from where the row is rather than from what the row says about itself.

Two consequences that are worth more than the mechanical change:

  - assign_user_to_project now takes the organization as an argument. It used to
    read it off the project row and copy it into user_roles, and there is no
    column to read now. user_roles is platform-level, so it cannot be written to
    by looking inside a tenant schema - so the caller, which has already resolved
    the project, has to say it. The protocol declares it, and the in-memory store
    accepts and ignores it rather than being the one implementation with a
    different signature.

  - A platform administrator listing projects without naming an organization now
    gets the answer assembled across tenant schemas, because there is no shared
    table to ask. That is one round trip per organization and it is paid only by
    that one caller; every per-tenant path, which is what an ordinary request
    does, is still a single statement. The merge reproduces the ordering and the
    total a shared table would have given, so paging across the boundary is the
    same as paging inside one.
2026-10-03 13:32:46 -07:00

1767 lines
76 KiB
Python
Raw Blame History

"""PostgreSQL adapter for the account and organization repository.
This is the authoritative store. It is written against the control-plane schema
in ``001_control_plane.sql`` plus the phase-1 identity migration in
``migrations/0001_phase1_identity.sql``.
Three mappings in this adapter are load-bearing and are the reason it is not a
thin SQL wrapper. Each is documented at its implementation.
1. **Role and organization come from ``user_roles``, not from ``users``.**
The control plane stores no role column and no organization column on
``users``. It assigns a role through ``user_roles`` with a scope, and
``app.validate_user_role_scope()`` enforces that the role's scope matches the
assignment. Resolving a single role per user is therefore a join, and
:func:`_resolve_assignment` refuses to guess when a user holds more than one
active assignment, because picking one silently would grant the wrong scope.
2. **A missing email is impossible, not optional.** ``users.email`` is
``NOT NULL UNIQUE``. The service already requires one; this adapter is the
second line of defence for rows written by another system.
3. **Audit identifiers are the database's.** ``audit_events.id`` is
``bigint GENERATED ALWAYS AS IDENTITY``, so the adapter returns that value
and not the UUID the in-memory repository mints. The API therefore reports
audit events by their store identifier.
Connection handling
-------------------
Every operation runs inside one transaction on one pooled connection, and two
things are published onto it before the statement runs.
``SET LOCAL app.user_id`` carries the caller's identity, so the row-level
security policies that still guard the platform tables see a server-derived value
rather than anything a client controls. ``SET LOCAL`` is scoped to the
transaction, so a pooled connection cannot leak one caller's identity into the
next request.
``SET LOCAL search_path`` carries the caller's **organization**, resolved into
the schema that holds its tables. That is the isolation model: one schema per
organization, no table of tenant data shared, so an organization's ``projects``
table is found by being in the search path rather than by being filtered by a
predicate. A statement with no organization in hand gets ``public`` alone, and
therefore finds no tenant table at all.
Both are set with ``set_config``, which takes bind parameters. ``SET`` itself
takes none - psycopg renders them as ``$1`` and the grammar rejects that.
Until the policies are applied to a non-owner role the ``app.user_id`` setting is
inert and authorization is enforced by the explicit ``WHERE`` clauses below.
That is defence in depth being staged, not a substitute: the ``WHERE`` clauses
enforce scope today, and the ``search_path`` enforces the schema boundary.
"""
from __future__ import annotations
import json
import logging
from contextlib import contextmanager
from datetime import datetime
from typing import Any, Iterator, Mapping, Sequence
from uuid import UUID, uuid4
import psycopg
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from .errors import ConflictError, InputError, NotFoundError, RepositoryError
from .models import (
VERIFICATION_PURPOSE_EMAIL,
AccountStatus,
AuditEvent,
CredentialRecord,
CredentialType,
Organization,
OrganizationStatus,
Project,
ProjectStatus,
RoleName,
SessionRecord,
User,
VerificationToken,
normalize_email,
username_key,
)
from .repository import AccountRepository
from .security import utc_now
from .tenancy import tenant_schema_name, tenant_search_path
LOGGER = logging.getLogger("umbratrans_web.postgres")
#: ``users.status`` allows more values than the application models. Anything
#: outside the application's set is refused on read rather than coerced, so an
#: unrecognised row surfaces as an error instead of being treated as active.
_ACCOUNT_STATUSES: Mapping[str, AccountStatus] = {status.value: status for status in AccountStatus}
#: The GUC that the row-level-security policy helpers read.
#:
#: This name is not a guess and it is not cosmetic. Every policy helper in the
#: control plane resolves the caller through ``app.current_user_id()``, and that
#: function is defined as ``current_setting('app.user_id', true)``. Setting a
#: similarly named GUC instead leaves every helper resolving to NULL, which makes
#: every policy return false and every query return zero rows: a silent total
#: lockout that looks like an empty database rather than an error.
#: :meth:`PostgresAccountRepository.verify_rls_contract` proves the two agree at
#: startup, for exactly that reason.
RLS_USER_SETTING = "app.user_id"
#: The function the policies call to resolve the caller.
RLS_USER_FUNCTION = "app.current_user_id"
#: The GUC that selects which tenant schema a statement sees.
#:
#: Named here rather than interpolated because this is the one setting a
#: statement's *target* depends on: `search_path` is the only way a single
#: statement template can reach a table whose address is not known until the
#: caller's organization has been resolved. Everything else in this adapter
#: names `public.` explicitly, because the platform tables have one address.
SEARCH_PATH_SETTING = "search_path"
#: The function the migration uses to name a tenant schema, and the one this
#: adapter is checked against at startup.
TENANT_SCHEMA_FUNCTION = "app.tenant_schema_name"
#: Statement timeout applied to every transaction, so one slow query cannot pin
#: a pooled connection indefinitely.
STATEMENT_TIMEOUT_MS = 5_000
class PostgresAccountRepository(AccountRepository):
"""Thread-safe :class:`AccountRepository` backed by PostgreSQL.
The pool must be opened by the caller. Pass ``open=False`` and call
:meth:`open` yourself so startup can fail before the process serves a
request.
"""
def __init__(self, pool: ConnectionPool[Any], *, open_pool: bool = True) -> None:
if open_pool:
pool.open(wait=True, timeout=10)
self._pool = pool
# -- lifecycle -----------------------------------------------------------
@classmethod
def from_dsn(
cls,
dsn: str,
*,
min_size: int = 1,
max_size: int = 10,
application_name: str = "umbratrans-web",
) -> PostgresAccountRepository:
"""Build a repository from a DSN.
The DSN is expected to come from the environment or a secret store. It
is never logged, never returned by an endpoint, and never written to a
development data file.
"""
pool = ConnectionPool(
conninfo=dsn,
min_size=min_size,
max_size=max_size,
kwargs={"application_name": application_name, "row_factory": dict_row},
open=False,
# Every checkout is wrapped in a transaction by _connection, so a
# returned connection never carries uncommitted work.
reset=_reset_connection,
)
return cls(pool)
def open(self, *, wait: bool = True, timeout: float = 10) -> None:
self._pool.open(wait=wait, timeout=timeout)
def close(self) -> None:
self._pool.close()
def check(self) -> bool:
"""Return whether the database answers a trivial query.
Used by the health probe. It reports reachability only and never
surfaces a version, host, or error string.
"""
try:
with self._connection() as connection:
connection.execute("SELECT 1")
return True
except Exception: # noqa: BLE001 - any failure means "not usable"
LOGGER.warning("database health check failed", exc_info=True)
return False
def verify_tenant_schema_contract(self) -> None:
"""Prove the adapter and the database agree on a tenant schema's name.
The name is derived from the organization id in two languages:
``umbratrans_web.tenancy.tenant_schema_name`` for the connection, and
``app.tenant_schema_name`` for the migration that created the schemas.
Nothing in SQL joins the two together, and the failure mode of a
disagreement is silent and total - every statement would resolve into a
schema that does not exist, so every project request for every
organization would raise "relation does not exist", and the error would
name the statement rather than the cause. That reads as an empty
database, which is the one diagnosis nobody reaches for.
So this asks the database what it calls one organization's schema and
refuses to serve unless it is what this adapter would have chosen. Run at
startup, for the same reason :meth:`verify_rls_contract` is.
"""
probe = uuid4()
expected = tenant_schema_name(probe)
with self._connection() as connection:
row = connection.execute(f"SELECT {TENANT_SCHEMA_FUNCTION}(%s) AS resolved", (probe,)).fetchone()
resolved = row["resolved"] if row is not None else None
if resolved != expected:
raise RepositoryError(
"tenant_schema_mismatch",
"The database derives a different tenant schema name from this adapter, so every "
"tenant table would be looked for in a schema that does not exist. Refusing to "
"start.",
)
LOGGER.info("tenant schema naming contract verified")
def verify_rls_contract(self) -> None:
"""Prove the adapter and the policy helpers agree on the caller's identity.
The policies resolve the caller by calling ``app.current_user_id()``.
The adapter publishes that identity as a session GUC. Nothing in the SQL
connects the two, so a mismatch is possible and the failure mode is
silent: every policy evaluates false, every query returns zero rows, and
the service reports an empty database rather than an error.
So this runs a real probe: set the GUC, ask the policy function what it
sees, and refuse to continue unless the two agree. Called at startup,
because discovering this in production as a total lockout is far worse
than refusing to start.
"""
probe = uuid4()
with self._connection() as connection:
connection.execute("SELECT set_config(%s, %s, true)", (RLS_USER_SETTING, str(probe)))
row = connection.execute(f"SELECT {RLS_USER_FUNCTION}() AS resolved").fetchone()
resolved = row["resolved"] if row is not None else None
if resolved != probe:
raise RepositoryError(
"rls_identity_mismatch",
"The row-level-security helper does not read the session setting this "
"adapter publishes, so every policy would deny access. Refusing to start.",
)
LOGGER.info("row-level-security identity contract verified setting=%s", RLS_USER_SETTING)
# -- transaction plumbing ----------------------------------------------
@contextmanager
def _connection(
self, *, user_id: UUID | None = None, organization_id: UUID | None = None
) -> Iterator[psycopg.Connection[Any]]:
"""Yield a connection inside one transaction, publishing caller and tenant.
``organization_id`` is the isolation model. It is resolved into a schema
name by :func:`umbratrans_web.tenancy.tenant_search_path` and set as
``SET LOCAL search_path``, so an unqualified ``projects`` in this
transaction means *this organization's* projects, and there is no
unqualified ``projects`` at all when no organization was given.
Both settings are ``SET LOCAL`` and therefore transaction-scoped, which is
what makes connection pooling safe: a returned connection cannot carry the
previous request's identity or the previous request's tenant.
The search path is set on **every** transaction, including the ones with
no organization, where it becomes ``public`` alone. Leaving it alone
instead would let the connection's own default apply, and a default that
happened to name a tenant schema would be a silent leak with no code
anywhere mentioning it.
"""
# ``SET`` is a utility statement and takes no bind parameters at all -
# psycopg renders them as ``$1`` and the grammar rejects that. The first
# statement therefore carries a literal. It is coerced to ``int`` here so
# the interpolation cannot carry SQL, and the value is a module constant
# anyway. Verified against a real server, not assumed.
timeout_ms = int(STATEMENT_TIMEOUT_MS)
search_path = tenant_search_path(organization_id)
with self._pool.connection() as connection:
with connection.transaction():
connection.execute(f"SET LOCAL statement_timeout = {timeout_ms}")
connection.execute("SET LOCAL idle_in_transaction_session_timeout = '10s'")
if user_id is not None:
# ``set_config`` is the parameterisable form and the right way
# to set a namespaced GUC, so the identity stays a bound value.
# The third argument is ``true`` for local scope, which is what
# makes pooling safe: the identity is discarded when the
# transaction ends, so a returned connection cannot carry the
# previous request's caller.
connection.execute(
"SELECT set_config(%s, %s, true)",
(RLS_USER_SETTING, str(user_id)),
)
connection.execute("SELECT set_config(%s, %s, true)", (SEARCH_PATH_SETTING, search_path))
yield connection
@contextmanager
def _for_user(self, user_id: UUID | None) -> Iterator[psycopg.Connection[Any]]:
with self._connection(user_id=user_id) as connection:
yield connection
# -- users ---------------------------------------------------------------
def create_user(self, user: User) -> User:
"""Insert the profile and its role assignment in one transaction.
``users`` carries no role or organization column, so the role is written
as a ``user_roles`` row. Both statements share a transaction: a user
without a resolvable role could not authenticate, so committing one
without the other would create an account that can never be used.
"""
now = user.created_at or utc_now()
with self._connection() as connection:
try:
connection.execute(
"""
INSERT INTO public.users (
id, email, display_name, username, status,
mfa_required, created_at, updated_at
)
VALUES (%s, %s, %s, %s, %s, false, %s, %s)
""",
(
user.id,
user.email,
user.display_name,
user.username,
user.status.value,
now,
now,
),
)
except psycopg.errors.UniqueViolation as error:
raise _unique_conflict(error, user.username) from error
self._insert_assignment(connection, user=user, now=now)
return user
def save_user(self, user: User) -> User:
"""Update the mutable profile fields.
``username`` and ``email`` are intentionally not written. Both are
identity keys under unique constraints, and neither has a reviewed
change path in this phase, so an accidental edit here would move an
account's login identifier with no audit event behind it.
"""
with self._connection() as connection:
updated = connection.execute(
"""
UPDATE public.users
SET display_name = %s,
status = %s,
updated_at = %s
WHERE id = %s
RETURNING id
""",
(user.display_name, user.status.value, user.updated_at or utc_now(), user.id),
).fetchone()
if updated is None:
raise NotFoundError("account_not_found")
self._sync_assignment(connection, user=user)
return user
def get_user(self, user_id: UUID) -> User | None:
return self._fetch_user("u.id = %s", (user_id,))
def get_user_by_email(self, key: str) -> User | None:
"""Look up an account by its email address. This is the login path.
The column comparison is left to the database. ``users.email`` is
``citext``, so equality is already case-insensitive there, which means
case-insensitivity does not depend on this process normalising
correctly. The value is still normalised, because a malformed address
cannot match any row and rejecting it here avoids a pointless query.
Validation failure is "no such account", not an error: the login form
must not tell an attacker which addresses exist.
"""
try:
candidate = normalize_email(key)
except InputError:
return None
return self._fetch_user("u.email = %s", (candidate,))
def get_user_by_username(self, key: str) -> User | None:
try:
normalized = username_key(key)
except Exception: # noqa: BLE001 - a malformed key is simply "no match"
return None
return self._fetch_user("u.username::text = %s", (normalized,))
def count_users(self, organization_id: UUID | None = None) -> int:
where, params = self._user_scope(organization_id)
with self._connection() as connection:
row = connection.execute(f"SELECT count(*) AS total FROM public.users u {where}", params).fetchone()
return int(row["total"]) if row else 0
def count_users_with_role(self, role: RoleName) -> int:
"""Count active assignments of one role.
Counts ``user_roles`` rather than ``users``, because a role lives on the
assignment and not on the profile. The join is left inner on purpose: a
user with no resolvable assignment cannot administer anything, so it must
not satisfy a check that asks whether an administrator exists.
This is what decides whether first-run setup is still required, so it runs
on every health probe. That makes it the one query here that cannot be
left broken without taking the whole deployment's health check with it.
"""
with self._connection() as connection:
row = connection.execute(
f"""
SELECT count(*) AS total
FROM public.user_roles ur
JOIN public.roles r ON r.id = ur.role_id
WHERE r.name = %s
AND ur.status = 'active'
AND {TENANT_ROW_PREDICATE}
""",
(str(role),),
).fetchone()
return int(row["total"]) if row else 0
def find_users(
self,
*,
organization_id: UUID | None = None,
status: AccountStatus | None = None,
role: str | None = None,
limit: int = 50,
offset: int = 0,
) -> tuple[list[User], int]:
"""Return a page of accounts.
Filtering by organization is a join through ``user_roles`` because
``users`` holds no organization column. The aggregate count is computed
in the same statement shape so the page total and the rows cannot
disagree.
"""
conditions: list[str] = []
params: list[Any] = []
if organization_id is not None:
conditions.append(
f"EXISTS (SELECT 1 FROM public.user_roles ur WHERE ur.user_id = u.id "
f"AND ur.organization_id = %s AND ur.status = 'active' AND {TENANT_ROW_PREDICATE})"
)
params.append(organization_id)
if status is not None:
conditions.append("u.status = %s")
params.append(status.value)
if role is not None:
conditions.append(
f"EXISTS (SELECT 1 FROM public.user_roles ur JOIN public.roles r ON r.id = ur.role_id "
f"WHERE ur.user_id = u.id AND r.name = %s AND ur.status = 'active' AND {TENANT_ROW_PREDICATE})"
)
params.append(role)
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
with self._connection() as connection:
total_row = connection.execute(f"SELECT count(*) AS total FROM public.users u {where}", params).fetchone()
total = int(total_row["total"]) if total_row else 0
if total == 0:
return [], 0
rows = connection.execute(
f"""
SELECT u.id, u.username, u.display_name, u.email, u.status,
u.mfa_required, u.created_at, u.updated_at
FROM public.users u
{where}
ORDER BY u.created_at DESC, u.id
LIMIT %s OFFSET %s
""",
[*params, max(limit, 0), max(offset, 0)],
).fetchall()
return [_user_from_row(row) for row in rows], total
def _fetch_user(self, predicate: str, params: Sequence[Any]) -> User | None:
with self._connection() as connection:
row = connection.execute(
f"""
SELECT u.id, u.username, u.display_name, u.email, u.status,
u.mfa_required, u.created_at, u.updated_at
FROM public.users u
WHERE {predicate}
""",
list(params),
).fetchone()
if row is None:
return None
return _user_with_assignment(connection, row)
def _user_scope(self, organization_id: UUID | None) -> tuple[str, list[Any]]:
if organization_id is None:
return "", []
return (
"WHERE EXISTS (SELECT 1 FROM public.user_roles ur "
f"WHERE ur.user_id = u.id AND ur.organization_id = %s "
f"AND ur.status = 'active' AND {TENANT_ROW_PREDICATE})",
[organization_id],
)
# -- projects ------------------------------------------------------------
#
# Every statement below names `projects` **unqualified**, and the organization
# reaches the right one through `search_path` rather than through a predicate.
# That is deliberate and it is the whole point: a shared table filtered by
# `organization_id = %s` is a filter somebody has to remember, and there is
# always a write path that would not have it. An unqualified name resolves into
# the caller's own schema or it does not resolve at all.
#
# `Project.organization_id` therefore comes from the schema the row was found
# in rather than from a column. It is still filled in - the model and the API
# both carry it, and a caller needs to be able to see which tenant a row
# belongs to - but it is derived from where the row is rather than from what
# the row says about itself.
def create_project(self, project: Project) -> Project:
"""Insert into the organization's own schema.
The organization comes from the row: a project cannot be created without
one, so there is nothing to ask for separately.
"""
with self._connection(organization_id=project.organization_id) as connection:
try:
connection.execute(
"""
INSERT INTO projects (
id, name, slug, status, created_by, created_at, updated_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s)
""",
(
project.id,
project.name,
project.slug,
str(project.status),
project.created_by,
project.created_at,
project.updated_at or project.created_at,
),
)
except psycopg.errors.UniqueViolation as error:
raise _unique_conflict(error, project.slug) from None
return project
def get_project(self, project_id: UUID, *, organization_id: UUID | None = None) -> Project | None:
"""Fetch one project.
With an organization, the lookup runs in that organization's schema, so a
project belonging to another tenant is not merely filtered out of a result
set: there is no row to read. A project identifier is unique on its own, so
an unscoped lookup against a shared table would fetch another tenant's row
and then decide it was not the caller's - a read that should never have
happened, and one that a later change to the check could turn into a leak.
Without one - a platform administrator who has not named an organization -
there is no single schema to look in, so the tenant schemas are searched in
turn. That is a platform operation by definition: the caller is asking for
"any project's", and there is no statement that can answer it without
naming every organization.
"""
if organization_id is not None:
with self._connection(organization_id=organization_id) as connection:
row = connection.execute(
f"SELECT {_PROJECT_COLUMNS} FROM projects p WHERE p.id = %s", (project_id,)
).fetchone()
if row is not None:
return _project_from_row(row, organization_id=organization_id)
return None
for candidate in self._organizations_with_tenant_schema():
with self._connection(organization_id=candidate) as connection:
row = connection.execute(
f"SELECT {_PROJECT_COLUMNS} FROM projects p WHERE p.id = %s", (project_id,)
).fetchone()
if row is not None:
return _project_from_row(row, organization_id=candidate)
return None
def get_project_by_slug(self, slug: str, *, organization_id: UUID) -> Project | None:
"""Fetch by handle, case-insensitively.
The comparison is left to the ``citext`` column rather than casting it to
``text``. ``slug::text = %s`` is an ordinary text comparison, so it is
case-*sensitive*, and ``/Book-One`` would fail to find ``book-one`` while
appearing to work for every other store in the system. The cast looks like
a type accommodation for the parameter and is in fact a silent change of
behaviour.
The organization is no longer part of the predicate, because it is part of
the schema: the same slug in another tenant is a different table.
"""
with self._connection(organization_id=organization_id) as connection:
row = connection.execute(
f"SELECT {_PROJECT_COLUMNS} FROM projects p WHERE p.slug = %s",
(str(slug),),
).fetchone()
return _project_from_row(row, organization_id=organization_id) if row else None
def save_project(self, project: Project) -> Project:
with self._connection(organization_id=project.organization_id) as connection:
row = connection.execute(
"""
UPDATE projects
SET name = %s, slug = %s, status = %s, updated_at = %s
WHERE id = %s
RETURNING id
""",
(
project.name,
project.slug,
str(project.status),
project.updated_at or utc_now(),
project.id,
),
).fetchone()
if row is None:
raise NotFoundError("project_not_found")
return project
def find_projects(
self,
*,
organization_id: UUID | None = None,
statuses: Sequence[ProjectStatus] | None = None,
limit: int = 50,
offset: int = 0,
) -> tuple[list[Project], int]:
"""List projects, from one tenant's schema or from all of them.
The scoped form is one statement in one schema. The unscoped form - a
platform administrator who has not named an organization - is assembled
across tenant schemas, because there is no shared table to ask.
"""
if organization_id is None:
return self._find_projects_across_tenants(statuses=statuses, limit=limit, offset=offset)
conditions: list[str] = []
params: list[Any] = []
if statuses:
conditions.extend(_status_conditions(statuses))
params.extend(_status_params(statuses))
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
window = max(offset, 0) + max(limit, 0)
with self._connection(organization_id=organization_id) as connection:
total_row = connection.execute(f"SELECT count(*) AS total FROM projects p {where}", params).fetchone()
total = int(total_row["total"]) if total_row else 0
if total == 0:
return [], 0
rows = connection.execute(
f"SELECT {_PROJECT_COLUMNS} FROM projects p {where} "
"ORDER BY p.created_at DESC, p.id LIMIT %s OFFSET %s",
[*params, window, max(offset, 0)],
).fetchall()
return [_project_from_row(row, organization_id=organization_id) for row in rows], total
def _find_projects_across_tenants(
self,
*,
statuses: Sequence[ProjectStatus] | None = None,
limit: int = 50,
offset: int = 0,
) -> tuple[list[Project], int]:
"""The platform-wide listing, assembled one tenant schema at a time.
Each tenant contributes its own total and its own first ``offset + limit``
rows; merging those and slicing gives the same answer a single shared table
would have. Asking each tenant for more than that is unnecessary, because a
row outside its tenant's first window cannot be inside the global one.
That is one round trip per organization, which at a hundred organizations
is a hundred round trips. It is the price of not sharing the table, and it
is paid only by a platform administrator listing across every tenant - the
per-tenant path above, which is everything an ordinary request does, is one
statement.
"""
window = max(offset, 0) + max(limit, 0)
collected: list[tuple[datetime, str, Project]] = []
total = 0
for candidate in self._organizations_with_tenant_schema():
conditions: list[str] = []
params: list[Any] = []
if statuses:
conditions.extend(_status_conditions(statuses))
params.extend(_status_params(statuses))
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
with self._connection(organization_id=candidate) as connection:
total_row = connection.execute(f"SELECT count(*) AS total FROM projects p {where}", params).fetchone()
total += int(total_row["total"]) if total_row else 0
rows = connection.execute(
f"SELECT {_PROJECT_COLUMNS} FROM projects p {where} ORDER BY p.created_at DESC, p.id LIMIT %s",
[*params, window],
).fetchall()
collected.extend(
(row["created_at"], str(row["id"]), _project_from_row(row, organization_id=candidate))
for row in rows
)
# The ordering a shared table would have given: newest first, identifier
# breaking ties, so a page boundary does not depend on which tenant
# answered first.
collected.sort(key=lambda item: (item[0], item[1]), reverse=True)
page = [project for _, _, project in collected][max(offset, 0) : max(offset, 0) + max(limit, 0)]
return page, total
def count_projects(self, *, organization_id: UUID | None = None) -> int:
if organization_id is None:
return sum(
self.count_projects(organization_id=candidate)
for candidate in self._organizations_with_tenant_schema()
)
with self._connection(organization_id=organization_id) as connection:
row = connection.execute("SELECT count(*) AS total FROM projects").fetchone()
return int(row["total"]) if row else 0
def _organizations_with_tenant_schema(self) -> list[UUID]:
"""Every organization that has a tenant schema, for the cross-tenant paths.
The join against ``pg_namespace`` is not decoration. An organization can
exist without a schema if it was inserted by something that bypassed the
trigger, and a cross-tenant listing that pointed at a missing schema would
fail on that one organization and return nothing for any of them - a single
half-provisioned tenant taking the whole platform view down with it.
"""
with self._connection() as connection:
rows = connection.execute(
"SELECT o.id FROM public.organizations o "
"JOIN pg_namespace n ON n.nspname = app.tenant_schema_name(o.id) "
"ORDER BY o.id"
).fetchall()
return [row["id"] for row in rows]
def list_project_member_ids(self, project_id: UUID) -> list[UUID]:
"""Project-scoped assignments, from the same table the roles live in.
Scoped to ``scope = 'project'`` because a user's organization-scoped
assignment must not be read as membership of every project in the tenant.
That is the mistake that would make one assignment grant project work.
"""
with self._connection() as connection:
rows = connection.execute(
"SELECT user_id FROM public.user_roles "
"WHERE project_id = %s AND scope = 'project' AND status = 'active' "
"ORDER BY created_at, user_id",
(project_id,),
).fetchall()
return [row["user_id"] for row in rows]
#: The project-scoped role an assignment grants.
#:
#: There is no ``member`` role at project scope. The control plane seeds
#: ``translator``, ``proofreader``, ``project_manager``, ``project_owner``,
#: ``release_manager``, ``audio_director``, and ``viewer`` - all scoped to
#: ``project`` - and those name jobs rather than participation. ``viewer`` is
#: used because it is the least privileged of them and is the only one that
#: claims nothing about what the assignee is doing there.
#:
#: This is provisional and deliberately so. An assignment currently means "is
#: working on this", and the workflow that would give it a job title is not
#: defined yet. When it is, this constant is where the role changes, and the
#: alternative to changing it is a migration - so it is named here rather than
#: buried in a query.
PROJECT_ASSIGNMENT_ROLE = "viewer"
def assign_user_to_project(self, project_id: UUID, user_id: UUID, *, organization_id: UUID) -> bool:
"""Add a project-scoped assignment, or report that there already is one.
Idempotent rather than erroring, because assigning the same account twice
is a no-op an administrator can reasonably make. The insert is conditional
so two concurrent requests cannot both add one.
``organization_id`` is required here rather than derived, and that is a
consequence of the isolation model rather than a preference. This statement
used to read ``p.organization_id`` off the project row and copy it into
``user_roles``. There is no column to read now - the schema the project was
found in *is* the organization - and ``user_roles`` is a platform table, so
it cannot be written to by looking inside a tenant schema. The caller has
already resolved the project and therefore already knows.
"""
with self._connection(organization_id=organization_id) as connection:
row = connection.execute(
"""
INSERT INTO public.user_roles (user_id, role_id, scope, organization_id, project_id, status)
SELECT %s, r.id, 'project', %s, p.id, 'active'
FROM projects p
JOIN public.roles r ON r.name = %s AND r.scope = 'project'
WHERE p.id = %s
ON CONFLICT DO NOTHING
RETURNING id
""",
(user_id, organization_id, self.PROJECT_ASSIGNMENT_ROLE, project_id),
).fetchone()
if row is not None:
return True
# Nothing inserted: either it was already there, or the project has no
# member-scoped project role seeded, or the project does not exist.
existing = connection.execute(
"SELECT 1 AS present FROM public.user_roles "
"WHERE user_id = %s AND project_id = %s AND scope = 'project' AND status = 'active'",
(user_id, project_id),
).fetchone()
if existing is not None:
return False
project_row = connection.execute("SELECT 1 AS present FROM projects WHERE id = %s", (project_id,)).fetchone()
if project_row is None:
raise NotFoundError("project_not_found")
raise RepositoryError(
"project_role_not_seeded",
f"No project-scoped {self.PROJECT_ASSIGNMENT_ROLE!r} role is present in the roles table.",
)
def unassign_user_from_project(self, project_id: UUID, user_id: UUID) -> bool:
with self._connection() as connection:
rows = connection.execute(
"DELETE FROM public.user_roles "
"WHERE user_id = %s AND project_id = %s AND scope = 'project' AND status = 'active' "
"RETURNING id",
(user_id, project_id),
).fetchall()
return bool(rows)
# -- role assignments ---------------------------------------------------
def _insert_assignment(self, connection: psycopg.Connection[Any], *, user: User, now: datetime) -> None:
scope, organization_id = _assignment_shape(user)
role_row = connection.execute(
"SELECT id, scope FROM public.roles WHERE name = %s", (str(user.role),)
).fetchone()
if role_row is None:
raise RepositoryError(
"role_not_seeded",
"The requested role is not present in the roles table.",
)
if role_row["scope"] != scope:
# app.validate_user_role_scope() would reject this anyway; failing
# here names the actual problem instead of surfacing a trigger error.
raise RepositoryError(
"role_scope_mismatch",
"The requested role does not have the scope required for this assignment.",
)
connection.execute(
"""
INSERT INTO public.user_roles (
user_id, role_id, scope, organization_id, project_id,
status, created_at, created_by
)
VALUES (%s, %s, %s, %s, NULL, 'active', %s, %s)
""",
(user.id, role_row["id"], scope, organization_id, now, user.id),
)
def _sync_assignment(self, connection: psycopg.Connection[Any], *, user: User) -> None:
"""Make the account's *tenant* assignment match the requested role and organization.
One role and one organization per account, so the target state is fully
determined.
Only tenant rows are read and written. An account assigned to a project
holds a second active ``user_roles`` row that says who is working on what;
it is not a second tenancy, and it must survive this call untouched.
Counting it would make every role change on an assigned member fail, and
a narrow ``UPDATE`` that reached it would strip the assignment while
appearing to change a role.
Several *tenant* rows are still refused, on the same grounds as
:func:`_resolve_assignment`: this path will not collapse them silently.
"""
scope, organization_id = _assignment_shape(user)
role_row = connection.execute(
"SELECT id, scope FROM public.roles WHERE name = %s", (str(user.role),)
).fetchone()
if role_row is None:
raise RepositoryError("role_not_seeded", "The requested role is not present in the roles table.")
if role_row["scope"] != scope:
raise RepositoryError(
"role_scope_mismatch",
"The requested role does not have the scope required for this assignment.",
)
existing = connection.execute(
"SELECT id, role_id, organization_id FROM public.user_roles ur "
f"WHERE ur.user_id = %s AND ur.status = 'active' AND {TENANT_ROW_PREDICATE}",
(user.id,),
).fetchall()
if len(existing) > 1:
raise RepositoryError(
"ambiguous_role_assignment",
"The account holds more than one active role assignment, which this phase cannot represent.",
)
if existing:
if existing[0]["role_id"] == role_row["id"] and existing[0]["organization_id"] == organization_id:
return
connection.execute(
"UPDATE public.user_roles SET role_id = %s, organization_id = %s WHERE id = %s",
(role_row["id"], organization_id, existing[0]["id"]),
)
return
connection.execute(
"""
INSERT INTO public.user_roles (
user_id, role_id, scope, organization_id, project_id, status, created_at, created_by
)
VALUES (%s, %s, %s, %s, NULL, 'active', now(), %s)
""",
(user.id, role_row["id"], scope, organization_id, user.id),
)
# -- organizations -------------------------------------------------------
def create_organization(self, organization: Organization) -> Organization:
with self._connection() as connection:
try:
connection.execute(
"""
INSERT INTO public.organizations (id, name, slug, status, created_at, updated_at)
VALUES (%s, %s, %s, %s, %s, %s)
""",
(
organization.id,
organization.name,
organization.slug,
organization.status.value,
organization.created_at or utc_now(),
organization.updated_at or utc_now(),
),
)
except psycopg.errors.UniqueViolation as error:
raise ConflictError("organization_slug_taken", "That organization slug is already in use.") from error
return organization
def save_organization(self, organization: Organization) -> Organization:
with self._connection() as connection:
row = connection.execute(
"""
UPDATE public.organizations
SET name = %s, slug = %s, status = %s, updated_at = %s
WHERE id = %s
RETURNING id
""",
(
organization.name,
organization.slug,
organization.status.value,
organization.updated_at or utc_now(),
organization.id,
),
).fetchone()
if row is None:
raise NotFoundError("organization_not_found")
return organization
def get_organization(self, organization_id: UUID) -> Organization | None:
with self._connection() as connection:
row = connection.execute(
"SELECT id, name, slug, status, created_at, updated_at FROM public.organizations WHERE id = %s",
(organization_id,),
).fetchone()
return _organization_from_row(row) if row else None
def get_organization_by_slug(self, slug: str) -> Organization | None:
with self._connection() as connection:
row = connection.execute(
"SELECT id, name, slug, status, created_at, updated_at FROM public.organizations WHERE slug::text = %s",
(slug.casefold(),),
).fetchone()
return _organization_from_row(row) if row else None
def find_organizations(
self,
*,
status: OrganizationStatus | None = None,
organization_ids: tuple[UUID, ...] | None = None,
limit: int = 50,
offset: int = 0,
) -> tuple[list[Organization], int]:
conditions: list[str] = []
params: list[Any] = []
if organization_ids is not None:
if not organization_ids:
return [], 0
conditions.append("id = ANY(%s)")
params.append([str(value) for value in organization_ids])
if status is not None:
conditions.append("status = %s")
params.append(status.value)
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
with self._connection() as connection:
total_row = connection.execute(
f"SELECT count(*) AS total FROM public.organizations {where}", params
).fetchone()
total = int(total_row["total"]) if total_row else 0
if total == 0:
return [], 0
rows = connection.execute(
f"""
SELECT id, name, slug, status, created_at, updated_at
FROM public.organizations
{where}
ORDER BY created_at DESC, id
LIMIT %s OFFSET %s
""",
[*params, max(limit, 0), max(offset, 0)],
).fetchall()
return [_organization_from_row(row) for row in rows], total
# -- credentials --------------------------------------------------------
def get_credential(self, user_id: UUID) -> CredentialRecord | None:
with self._connection() as connection:
row = connection.execute(
"""
SELECT id, user_id, credential_type, password_hash, must_change_password,
failed_login_count, locked_until, last_login_at,
password_changed_at, created_at, updated_at
FROM public.user_credentials
WHERE user_id = %s
""",
(user_id,),
).fetchone()
return _credential_from_row(row) if row else None
def save_credential(self, credential: CredentialRecord) -> CredentialRecord:
"""Write the credential, or insert it when the account has none.
The check constraint on ``user_credentials`` independently enforces that
a local password carries a hash and an OIDC credential does not, so a
programming error here is refused by the database rather than stored.
"""
with self._connection() as connection:
updated = connection.execute(
"""
UPDATE public.user_credentials
SET credential_type = %s,
password_hash = %s,
must_change_password = %s,
failed_login_count = %s,
locked_until = %s,
last_login_at = %s,
password_changed_at = %s,
updated_at = %s
WHERE user_id = %s
RETURNING id
""",
(
credential.credential_type.value,
credential.password_hash,
credential.must_change_password,
credential.failed_login_count,
credential.locked_until,
credential.last_login_at,
credential.password_changed_at,
credential.updated_at or utc_now(),
credential.user_id,
),
).fetchone()
if updated is not None:
return credential
try:
connection.execute(
"""
INSERT INTO public.user_credentials (
id, user_id, credential_type, password_hash, must_change_password,
failed_login_count, locked_until, last_login_at, password_changed_at,
created_at, updated_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
(
credential.id,
credential.user_id,
credential.credential_type.value,
credential.password_hash,
credential.must_change_password,
credential.failed_login_count,
credential.locked_until,
credential.last_login_at,
credential.password_changed_at,
credential.created_at or utc_now(),
credential.updated_at or utc_now(),
),
)
except psycopg.errors.UniqueViolation as error:
raise ConflictError("credential_exists", "This account already has a credential.") from error
# The control plane's users.auth_provider is its own field and this
# phase does not treat it as an identity source: the canonical
# credential type lives in user_credentials. Mirroring it here keeps
# the two from disagreeing for anything that does read it.
connection.execute(
"UPDATE public.users SET auth_provider = %s WHERE id = %s",
(credential.credential_type.value, credential.user_id),
)
return credential
# -- sessions -----------------------------------------------------------
def create_session(self, session: SessionRecord) -> SessionRecord:
"""Insert a session, writing both deadline columns.
``sessions.expires_at`` predates this application and is the absolute
deadline for anything else that reads it, so it is kept equal to
``absolute_deadline_at`` rather than left to default.
"""
with self._connection() as connection:
connection.execute(
"""
INSERT INTO public.sessions (
id, user_id, token_hash, expires_at, revoked_at,
created_at, last_seen_at, idle_deadline_at, mfa_verified_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
(
session.id,
session.user_id,
session.token_hash,
session.absolute_deadline_at,
session.revoked_at,
session.created_at,
session.last_seen_at,
session.idle_deadline_at,
session.mfa_verified_at,
),
)
# users.last_login_at is the control plane's own signal; keep it in
# step so anything reading it sees the same value.
connection.execute(
"UPDATE public.users SET last_login_at = %s WHERE id = %s", (session.created_at, session.user_id)
)
return session
def get_session(self, session_id: UUID) -> SessionRecord | None:
return self._fetch_session("s.id = %s", (session_id,))
def get_session_by_token_hash(self, token_hash: str) -> SessionRecord | None:
return self._fetch_session("s.token_hash = %s", (token_hash,))
def save_session(self, session: SessionRecord) -> SessionRecord:
with self._connection() as connection:
row = connection.execute(
"""
UPDATE public.sessions
SET last_seen_at = %s,
idle_deadline_at = %s,
expires_at = %s,
revoked_at = %s
WHERE id = %s
RETURNING id
""",
(
session.last_seen_at,
session.idle_deadline_at,
session.absolute_deadline_at,
session.revoked_at,
session.id,
),
).fetchone()
if row is None:
raise NotFoundError("session_not_found")
return session
def list_sessions(self, user_id: UUID, *, include_revoked: bool = False) -> list[SessionRecord]:
clause = "" if include_revoked else "AND s.revoked_at IS NULL"
with self._connection() as connection:
rows = connection.execute(
f"""
SELECT s.id, s.user_id, s.token_hash, s.created_at, s.last_seen_at,
s.idle_deadline_at, s.expires_at, s.revoked_at, s.mfa_verified_at
FROM public.sessions s
WHERE s.user_id = %s {clause}
ORDER BY s.created_at DESC
""",
(user_id,),
).fetchall()
return [_session_from_row(row) for row in rows]
def count_sessions(self, user_id: UUID) -> int:
return len(self.list_sessions(user_id))
def revoke_sessions(
self,
user_id: UUID,
*,
reason: str,
now: datetime,
exclude_session_id: UUID | None = None,
) -> int:
"""Revoke live sessions in one statement.
The ``reason`` is written into a comment-free column only: the
control-plane ``sessions`` table has no reason column, so it is returned
to the caller for the audit record rather than persisted here.
"""
clause = "AND id <> %s" if exclude_session_id is not None else ""
params: list[Any] = [now, user_id]
if exclude_session_id is not None:
params.append(exclude_session_id)
with self._connection() as connection:
rows = connection.execute(
f"""
UPDATE public.sessions
SET revoked_at = %s
WHERE user_id = %s AND revoked_at IS NULL {clause}
RETURNING id
""",
params,
).fetchall()
LOGGER.info("revoked sessions user_id=%s count=%s reason=%s", user_id, len(rows), reason)
return len(rows)
def _fetch_session(self, predicate: str, params: Sequence[Any]) -> SessionRecord | None:
with self._connection() as connection:
row = connection.execute(
f"""
SELECT s.id, s.user_id, s.token_hash, s.created_at, s.last_seen_at,
s.idle_deadline_at, s.expires_at, s.revoked_at, s.mfa_verified_at
FROM public.sessions s
WHERE {predicate}
""",
list(params),
).fetchone()
return _session_from_row(row) if row else None
# -- verification tokens ------------------------------------------------
def create_verification_token(self, token: VerificationToken) -> VerificationToken:
"""Store a single-use token in ``bootstrap_tokens``.
That table predates this application and was unused, which is why email
validation needs no migration. It has no ``purpose`` column, so every row
in it is treated as an email-verification token. Phase 1 issues no other
kind, and if a second purpose is ever added the column has to be added
with it: without a purpose in the row, a token issued for one purpose
would be honoured for another. :func:`_verification_from_row` therefore
always reports the email-verification purpose rather than reading one.
"""
with self._connection() as connection:
connection.execute(
"""
INSERT INTO public.bootstrap_tokens (
id, user_id, token_hash, expires_at, consumed_at, created_at, purpose
)
VALUES (%s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (id) DO NOTHING
""",
(
token.id,
token.user_id,
token.token_hash,
token.expires_at,
token.consumed_at,
token.created_at,
token.purpose,
),
)
return token
def get_live_verification_token(
self,
token_hash: str,
*,
purpose: str = VERIFICATION_PURPOSE_EMAIL,
now: datetime | None = None,
) -> VerificationToken | None:
"""Return an unconsumed, unexpired token for this hash *and purpose*.
Both are matched in the query rather than checked afterwards. A
password-reset token and an email-verification token are both a hash
against a user, so a lookup that ignored purpose would accept one where
the other is required: a verification link would set a password, and a
reset link would confirm an address. The ``purpose`` column added by
migration 0002 exists for this.
Expiry is filtered in SQL rather than in Python so a token that expired
while the process was down is still refused, and so the decision is made
against the same clock the row was written with.
``now`` is accepted and deliberately unused. The in-memory repository has
to be told what time it is, and the two have to agree about the contract,
but here the comparison is ``expires_at > now()`` inside the query: the
database's clock, against the database's own timestamp. Substituting the
application's clock would make a token appear live or dead depending on
which repository answered, which is the divergence this whole argument
exists to prevent.
"""
with self._connection() as connection:
row = connection.execute(
"""
SELECT id, user_id, token_hash, expires_at, consumed_at, created_at, purpose
FROM public.bootstrap_tokens
WHERE token_hash = %s
AND purpose = %s
AND consumed_at IS NULL
AND expires_at > now()
""",
(token_hash, purpose),
).fetchone()
return _verification_from_row(row) if row else None
def mark_verification_token_consumed(self, token_id: UUID, *, now: datetime) -> bool:
"""Consume a token, returning whether this call was the one that did it.
The update carries ``consumed_at IS NULL`` in its predicate, so the
database decides the winner. Two simultaneous requests cannot both
observe success, which is what makes the link single-use; deciding this in
Python after a read would not.
"""
with self._connection() as connection:
row = connection.execute(
"""
UPDATE public.bootstrap_tokens
SET consumed_at = %s
WHERE id = %s
AND consumed_at IS NULL
RETURNING id
""",
(now, token_id),
).fetchone()
return row is not None
def delete_expired_verification_tokens(self, *, now: datetime) -> int:
"""Delete tokens that are spent or long expired, returning how many."""
with self._connection() as connection:
row = connection.execute(
"""
DELETE FROM public.bootstrap_tokens
WHERE consumed_at IS NOT NULL
OR expires_at <= %s
RETURNING id
""",
(now,),
).fetchall()
return len(row)
# -- audit --------------------------------------------------------------
def append_audit_event(self, event: AuditEvent) -> AuditEvent:
"""Append one audit event.
The application mints a UUID for correlation, but ``audit_events.id`` is
a database identity, so the row identifier comes from the database. The
application UUID is preserved in ``metadata`` so a client that saw the
in-memory identifier can still correlate a test run.
"""
with self._connection() as connection:
row = connection.execute(
# The whole row comes back, not just the identifier: the store
# defaults and normalises these columns, so the returned row is
# what was actually written and is what a later read would see.
# Returning only ``id`` left the mapper reading keys the row did
# not have.
"""
INSERT INTO public.audit_events (
actor_user_id, organization_id, action, target_type, target_id,
request_id, metadata, created_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s::jsonb, %s)
RETURNING id, actor_user_id, organization_id, action, target_type,
target_id, request_id, metadata, created_at
""",
(
event.actor_user_id,
event.organization_id,
event.action,
"user" if event.target_user_id else None,
str(event.target_user_id) if event.target_user_id else None,
event.request_id,
json.dumps({**dict(event.detail), "application_event_id": str(event.id)}),
event.occurred_at,
),
).fetchone()
return _audit_from_row(row, event)
def list_audit_events(
self,
*,
organization_id: UUID | None = None,
target_user_id: UUID | None = None,
actor_user_id: UUID | None = None,
limit: int = 50,
offset: int = 0,
) -> tuple[list[AuditEvent], int]:
conditions: list[str] = []
params: list[Any] = []
if organization_id is not None:
conditions.append("organization_id = %s")
params.append(organization_id)
if target_user_id is not None:
# The application's target_user_id maps onto the polymorphic
# (target_type, target_id) pair the control plane uses.
conditions.append("target_type = 'user' AND target_id = %s")
params.append(str(target_user_id))
if actor_user_id is not None:
conditions.append("actor_user_id = %s")
params.append(actor_user_id)
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
with self._connection() as connection:
total_row = connection.execute(
f"SELECT count(*) AS total FROM public.audit_events {where}", params
).fetchone()
total = int(total_row["total"]) if total_row else 0
if total == 0:
return [], 0
rows = connection.execute(
f"""
SELECT id, actor_user_id, organization_id, action, target_type, target_id,
request_id, metadata, created_at
FROM public.audit_events
{where}
ORDER BY created_at DESC, id DESC
LIMIT %s OFFSET %s
""",
[*params, max(limit, 0), max(offset, 0)],
).fetchall()
return [_audit_from_row(row, None) for row in rows], total
# ---------------------------------------------------------------------------
# Row mapping
# ---------------------------------------------------------------------------
def _reset_connection(connection: psycopg.Connection[Any]) -> None:
"""Clear per-transaction state before a pooled connection is reused."""
connection.rollback()
def _assignment_shape(user: User) -> tuple[str, UUID | None]:
"""Return the ``user_roles`` scope and organization for a user.
A platform administrator is assigned at platform scope with no organization,
which is what the ``user_roles`` check constraint requires. Everyone else
needs an organization, because an organization-scoped assignment without one
is rejected by the same constraint.
"""
if user.role is RoleName.PLATFORM_ADMIN:
return "platform", None
if user.organization_id is None:
raise RepositoryError(
"organization_required",
"An organization-scoped role requires an organization.",
)
return "organization", user.organization_id
def _resolve_assignment(connection: psycopg.Connection[Any], user_id: UUID) -> tuple[RoleName, UUID | None]:
"""Resolve one role and organization for a user, or refuse.
An account may hold several active ``user_roles`` rows, because two different
kinds of row live in that table:
* a **tenant** row, at ``platform`` or ``organization`` scope, saying which
organization the account belongs to and what it may do there;
* a **project** row, at ``project`` scope, saying who is working on which
project.
Only the first kind answers "which tenant is this?", so only the first kind is
read here. Project rows are excluded by scope rather than by organisation,
because a project row carries an organization too and filtering on the
column rather than the scope would be indistinguishable from the tenant row
it sits beside.
Counting all rows here would refuse to resolve any assigned member, which
locks them out of authentication entirely - the failure mode is a member who
worked on a book and then could not sign in.
One tenant row remains the rule, and one tenant row is still ambiguous: the
organization returned scopes every administrative decision, so a second tenant
row is refused rather than resolved to whichever sorted first, which would
hand out the wrong organization's authority. Project membership does not
create that ambiguity and is not a reason to refuse.
"""
rows = connection.execute(
"""
SELECT r.name, ur.organization_id, ur.scope
FROM public.user_roles ur
JOIN public.roles r ON r.id = ur.role_id
WHERE ur.user_id = %s
AND ur.status = 'active'
AND ur.scope <> 'project'
ORDER BY ur.scope, ur.created_at, ur.id
""",
(user_id,),
).fetchall()
if not rows:
raise RepositoryError(
"no_active_role",
"The account has no active role assignment, so it cannot be authorized.",
)
if len(rows) > 1:
raise RepositoryError(
"ambiguous_role_assignment",
"The account holds more than one active role assignment, which this phase cannot represent.",
)
row = rows[0]
try:
role = RoleName(row["name"])
except ValueError as error:
raise RepositoryError("unknown_role", "The account has a role this application does not model.") from error
return role, row["organization_id"]
def _user_with_assignment(connection: psycopg.Connection[Any], row: Mapping[str, Any]) -> User:
role, organization_id = _resolve_assignment(connection, row["id"])
return _user_from_row(row, role=role, organization_id=organization_id)
def _user_from_row(
row: Mapping[str, Any], *, role: RoleName | None = None, organization_id: UUID | None = None
) -> User:
if row.get("mfa_required"):
# The control plane can mark an account as requiring a second factor.
# This application cannot satisfy that, so it refuses rather than
# authenticating someone who was supposed to present a second factor.
# The deployment gate is that no row carries mfa_required = true until
# MFA is implemented; see migrations/README.md.
raise RepositoryError(
"mfa_required_unsupported",
"The account requires multi-factor authentication, which this application does not implement.",
)
status = row["status"]
if status not in _ACCOUNT_STATUSES:
# 'invited' and 'locked' exist in the control plane but are not states
# this application models. Failing closed beats mapping one of them
# onto 'active' by accident.
raise RepositoryError(
"unmapped_account_status",
"The account has a status this application does not model.",
)
return User(
id=row["id"],
username=row["username"],
username_key=username_key(row["username"]),
display_name=row["display_name"] or row["username"],
role=role if role is not None else RoleName.MEMBER,
status=_ACCOUNT_STATUSES[status],
organization_id=organization_id,
email=row["email"],
created_at=row["created_at"],
updated_at=row["updated_at"],
)
#: The predicate that selects an account's *tenancy* rows and excludes project
#: rows.
#:
#: ``user_roles`` holds two different statements. A tenant row (scope ``platform``
#: or ``organization``) says which organization the account belongs to and what it
#: may do there. A project row (scope ``project``) says who is working on which
#: project. Project rows also carry an organization, so they cannot be told apart
#: by that column - only by scope.
#:
#: Every query that asks "which organization is this account in", or "what role
#: does it hold there", must apply this. Leaving one out is not a style
#: difference: it made role resolution refuse any assigned member, locking them
#: out of sign-in, because such an account now holds two active rows.
#:
#: Written for a table aliased ``ur``.
TENANT_ROW_PREDICATE = "ur.scope <> 'project'"
def _status_conditions(statuses: Sequence[ProjectStatus]) -> list[str]:
"""The predicate for a status filter, refusing a value the model does not have.
Every status the model knows is rendered, so a value outside the database's own
check constraint is refused by the store rather than silently matching nothing.
"No projects have status 'paused'" and "that status does not exist" are
different answers and only one of them is true.
"""
wanted = [str(status) for status in ProjectStatus]
for status in statuses:
if str(status) not in wanted:
raise RepositoryError("unmodelled_project_status", f"Unrecognised project status {status!r}.")
return ["p.status = ANY(%s)"]
def _status_params(statuses: Sequence[ProjectStatus]) -> list[Any]:
return [[str(status) for status in statuses]]
#: The columns a project is read with.
#:
#: ``organization_id`` is **not** among them, and that is the migration talking:
#: the column is gone from the table, because the schema the row was found in
#: already says which organization owns it. The model still carries an organization
#: and callers still see one, so :func:`_project_from_row` fills it in from the
#: schema the answer came from.
_PROJECT_COLUMNS = "p.id, p.name, p.slug, p.status, p.created_by, p.created_at, p.updated_at"
def _project_from_row(row: Mapping[str, Any], *, organization_id: UUID) -> Project:
"""Map a ``projects`` row, taking the organization from the schema it is in.
The status is mapped through the same table of modelled values the accounts
use, and an unrecognised value is refused rather than coerced. The database's
check constraint allows exactly the four the model declares, so a row outside
them would mean the schema and the application disagree, and guessing which is
right is how a project ends up quietly archived.
"""
return Project(
id=row["id"],
organization_id=organization_id,
name=row["name"],
slug=str(row["slug"]),
status=_project_status(row["status"]),
created_by=row["created_by"],
created_at=row["created_at"],
updated_at=row["updated_at"],
)
def _project_status(value: str) -> ProjectStatus:
try:
return ProjectStatus(value)
except ValueError:
raise RepositoryError(
"unmodelled_project_status", f"The store holds a project status this version does not know: {value!r}."
) from None
def _organization_from_row(row: Mapping[str, Any]) -> Organization:
return Organization(
id=row["id"],
name=row["name"],
slug=row["slug"],
status=OrganizationStatus(row["status"]),
created_at=row["created_at"],
updated_at=row["updated_at"],
)
def _credential_from_row(row: Mapping[str, Any]) -> CredentialRecord:
return CredentialRecord(
id=row["id"],
user_id=row["user_id"],
credential_type=CredentialType(row["credential_type"]),
password_hash=row["password_hash"],
must_change_password=row["must_change_password"],
failed_login_count=row["failed_login_count"],
locked_until=row["locked_until"],
last_login_at=row["last_login_at"],
password_changed_at=row["password_changed_at"],
created_at=row["created_at"],
updated_at=row["updated_at"],
)
def _session_from_row(row: Mapping[str, Any]) -> SessionRecord:
return SessionRecord(
id=row["id"],
user_id=row["user_id"],
token_hash=row["token_hash"],
created_at=row["created_at"],
last_seen_at=row["last_seen_at"],
idle_deadline_at=row["idle_deadline_at"],
absolute_deadline_at=row["expires_at"],
revoked_at=row["revoked_at"],
mfa_verified_at=row["mfa_verified_at"],
)
def _audit_from_row(row: Mapping[str, Any], original: AuditEvent | None) -> AuditEvent:
"""Map an ``audit_events`` row, keeping the store's identifier authoritative.
The application mints a UUID for correlation and the store mints a bigint
identity. The store's value is what the API reports, and the application
UUID is preserved in ``detail`` so a caller that saw it on a write can
still correlate a test run.
"""
metadata = row.get("metadata") or {}
if isinstance(metadata, str):
metadata = json.loads(metadata)
metadata = dict(metadata)
detail = {key: str(value) for key, value in metadata.items() if key != "application_event_id"}
# ``target_id`` is a text column that also holds non-user targets, so it is
# narrowed on ``target_type`` and then parsed. A ``user`` row whose text is not
# a UUID is a store inconsistency; it becomes no target rather than an
# exception raised while mapping an audit record, which would fail the write
# that was already committed.
target_user_id: UUID | None = None
if row["target_type"] == "user" and row["target_id"]:
try:
target_user_id = UUID(str(row["target_id"]))
except ValueError:
target_user_id = None
return AuditEvent(
id=str(row["id"]),
occurred_at=row["created_at"],
action=row["action"],
outcome=(original.outcome if original is not None else str(metadata.get("outcome", "success"))),
actor_user_id=row["actor_user_id"],
target_user_id=target_user_id,
organization_id=row["organization_id"],
request_id=row["request_id"],
detail=detail,
)
def _verification_from_row(row: Mapping[str, Any]) -> VerificationToken:
"""Map a ``bootstrap_tokens`` row, reading the purpose from the row.
The purpose used to be hardcoded here because the table had no such column.
Migration 0002 added it, and it is read now, because a token that reports the
wrong purpose is exactly the substitution the column exists to prevent.
"""
return VerificationToken(
id=row["id"],
user_id=row["user_id"],
token_hash=row["token_hash"],
expires_at=row["expires_at"],
created_at=row["created_at"],
consumed_at=row["consumed_at"],
purpose=row["purpose"],
)
def _unique_conflict(error: Exception, username: str) -> ConflictError:
"""Map a unique violation onto the right domain conflict.
``users`` has two unique constraints that matter here: the username added by
the phase-1 migration and the pre-existing email. The constraint name is
used rather than parsing the driver message, which is not stable.
"""
diagnostic = getattr(error, "diag", None)
constraint = getattr(diagnostic, "constraint_name", "") or ""
if "email" in constraint:
return ConflictError("email_taken", "That email address is already in use.")
return ConflictError("username_taken", "That username is already in use.")
__all__ = [
"PostgresAccountRepository",
"RLS_USER_FUNCTION",
"RLS_USER_SETTING",
"STATEMENT_TIMEOUT_MS",
]
Reference in New Issue View Git Blame Copy Permalink