diff --git a/docs/umbratrans-backend-design.md b/docs/umbratrans-backend-design.md index 58b7159..b4c26bc 100644 --- a/docs/umbratrans-backend-design.md +++ b/docs/umbratrans-backend-design.md @@ -241,8 +241,8 @@ Production PostgreSQL remains the authoritative store. Local development data st 3. Implement Argon2id credential service, temporary-password enrollment, and first-login password change. **Done.** 4. Add FastAPI routes and secure cookie/session middleware. **Done.** 5. Add administrator account-management routes and audit events. **Done.** -6. Add the reviewed PostgreSQL migration and adapter. Migration drafted and unapplied; adapter not started. -7. Add PostgreSQL integration tests in an isolated database. +6. Add the reviewed PostgreSQL migration and adapter. Migration drafted and unapplied; adapter written and unit tested, but **never executed against a database**. +7. Add PostgreSQL integration tests in an isolated database. Written and selected by `UMBRATRANS_TEST_DSN`; **not yet run**. 8. Connect the static prototype to the API. 9. Only then provision the first administrator in the approved environment. @@ -261,5 +261,7 @@ unreviewed store, so nothing in this slice can touch `umbracontrol` by accident. - An administrator cannot access another organization through a forged ID, and cannot escalate to platform administrator. **Done.** - Account-management actions create sanitized audit events with the requesting administrator as the actor. **Done.** - A user can access only projects granted through active membership. *Pending: project membership is not in this slice.* +- The PostgreSQL adapter refuses rather than guesses: an unmapped account status, a row requiring MFA, an account with more than one active role assignment, and a missing organization all fail closed. *Unit tested; the SQL is not yet executed.* - The static UI can be connected to the API without changing the audio worker. - Production migration and bootstrap remain blocked until explicitly approved. +- Email is a required field on account creation, because `users.email` is `NOT NULL UNIQUE`. Enforced in the domain so it fails the same way in development and production. *Done.* diff --git a/pyproject.toml b/pyproject.toml index ed3eaa4..bdef949 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,6 +6,7 @@ requires-python = ">=3.11" dependencies = [ "argon2-cffi>=23.1,<26", "fastapi>=0.115,<1", + "psycopg-pool>=3.2,<4", "psycopg[binary]>=3.2,<4", "uvicorn>=0.34,<1", ] diff --git a/umbratrans_web/README.md b/umbratrans_web/README.md index 4dae7fe..450a3ee 100644 --- a/umbratrans_web/README.md +++ b/umbratrans_web/README.md @@ -6,10 +6,84 @@ administrator lifecycle actions, sanitized audit events, and a FastAPI application factory that can also serve the browser application from the same origin. -**This slice does not connect to PostgreSQL.** The reviewed migration in -`migrations/` has not been applied, and the production repository adapter is -deliberately absent, so the application fails closed instead of starting -against an unreviewed store. +**The PostgreSQL adapter is written but has never been run.** The reviewed +migration in `migrations/` has not been applied, and the integration suite has +not been executed, because this workspace has no database credentials. The +adapter's pure logic is unit tested; its SQL is not. Read +[Verification status](#verification-status) before trusting it with real data. + +## The PostgreSQL adapter + +Three mappings in `postgres_repository.py` are load-bearing and are the reason +it is not a thin SQL wrapper. + +**Role and organization come from `user_roles`.** 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 one role per user is a join. +When a user holds more than one active assignment the adapter **refuses** rather +than picking a row, because the organization on the assignment becomes the scope +of every administrative decision, so guessing would hand out the wrong tenant's +authority. + +**A missing email is impossible.** `users.email` is `NOT NULL UNIQUE`, so email +is a required field in the service and in the API schema. Enforcing it in the +domain means a missing address fails identically against the development +repositories and the authoritative store. + +**Audit identifiers belong to the store.** `audit_events.id` is a `bigint` +identity, so the adapter reports that value. The application's correlation UUID +is preserved in the detail under `application_event_id` and never displaces the +row key. + +Three further fail-closed behaviours are worth knowing: + +- A row whose `users.status` is `invited` or `locked` is refused, not coerced + onto `active`. +- A row with `users.mfa_required = true` is refused, because this phase cannot + satisfy a second factor. Until MFA exists, every row must be false. +- A missing organization record blocks access rather than granting it. + +Every operation runs in one transaction on one pooled connection and publishes +the caller with `SET LOCAL app.current_user_id`, so a returned connection cannot +carry one request's identity into the next. That setting is inert until the +reviewed RLS policies are applied; today the explicit `WHERE` clauses are what +enforce scope. + +## Configuration + +| Variable | Meaning | +| --- | --- | +| `UMBRATRANS_REPOSITORY` | `memory`, `json`, or `postgres` | +| `UMBRATRANS_DATABASE_DSN` | Connection string. Required for `postgres`, and excluded from `repr` so it cannot leak into a settings dump. | +| `UMBRATRANS_DATABASE_POOL_MIN` / `_MAX` | Pool bounds, default 1 and 10 | +| `UMBRATRANS_REPOSITORY_PATH` | Local JSON store path | +| `UMBRATRANS_STATIC_ROOT` | Serve the browser client from this directory | +| `UMBRATRANS_ALLOWED_HOSTS` | Comma-separated host allow-list | +| `UMBRATRANS_COOKIE_SECURE` | Required `true` in production | +| `UMBRATRANS_REQUIRE_CSRF` | Required `true` in production | + +Production fails closed at two independent gates: the settings layer refuses +without a DSN, and the repository boundary refuses to build a PostgreSQL store +without one. A production process cannot start against a development store while +appearing to hold real data. + +## Verification status + +| Area | Status | +| --- | --- | +| Row mapping, role resolution, conflict mapping, configuration | Unit tested, runs everywhere | +| SQL statements, transactions, RLS setting, pooled reuse | **Not executed.** No database credentials in the authoring environment. | +| `tests/test_postgres_repository.py` integration cases | **Skipped.** Selected by `UMBRATRANS_TEST_DSN`. | + +Run the integration layer against a scratch database, never `umbracontrol`: + +```bash +UMBRATRANS_TEST_DSN='postgresql://app@localhost/umbratrans_scratch' \ + /home/umbra/.local/bin/uv run pytest umbratrans_web/tests/test_postgres_repository.py +``` + +The fixture refuses a DSN whose database name contains `umbracontrol`. ## Tenancy @@ -50,6 +124,7 @@ against an unreviewed store. | `config.py` | Fail-closed settings. Production requires secure cookies, CSRF, and a non-development repository. | | `errors.py` | Domain errors with stable codes. Messages are generic so responses cannot enumerate accounts. | | `repository.py` | `AccountRepository` protocol plus the thread-safe in-memory development repository. | +| `postgres_repository.py` | The authoritative PostgreSQL adapter: pooled connections, one transaction per operation, `SET LOCAL` identity, and the control-plane role mapping. | | `dev_repository.py` | Optional local JSON store: `0600` file, `0700` directory, atomic writes, refuses production. | | `service.py` | `AccountService`: enrollment, authentication, first-password change, sessions, organizations, administration, audit. | | `schemas.py` | Pydantic request/response models. Request models forbid unknown fields. | diff --git a/umbratrans_web/cli.py b/umbratrans_web/cli.py index 5a9cc30..7efda22 100644 --- a/umbratrans_web/cli.py +++ b/umbratrans_web/cli.py @@ -101,7 +101,11 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument("--username", default=BOOTSTRAP_USERNAME, help="non-secret bootstrap username") parser.add_argument("--display-name", default=BOOTSTRAP_DISPLAY_NAME, help="display name for the account") - parser.add_argument("--email", default=None, help="optional email address") + parser.add_argument( + "--email", + required=True, + help="contact email address for the account (required: users.email is NOT NULL)", + ) parser.add_argument( "--organization-id", default=None, diff --git a/umbratrans_web/config.py b/umbratrans_web/config.py index 950b7d1..89041fc 100644 --- a/umbratrans_web/config.py +++ b/umbratrans_web/config.py @@ -121,6 +121,9 @@ class Settings: password_max_length: int = 1024 repository_kind: str = "memory" repository_path: Path | None = None + database_dsn: str | None = field(default=None, repr=False, compare=False) + database_pool_min: int = 1 + database_pool_max: int = 10 static_root: Path | None = None allowed_hosts: tuple[str, ...] = () last_seen_write_interval_seconds: int = 30 @@ -148,6 +151,8 @@ class Settings: raise ConfigurationError("weak_token_pepper", f"Token pepper must be at least {MIN_PEPPER_BYTES} bytes.") if any(host.strip() in {"*", ""} for host in self.allowed_hosts): raise ConfigurationError("wildcard_host_forbidden", "A wildcard or empty allowed host is not permitted.") + if self.database_pool_min < 1 or self.database_pool_max < self.database_pool_min: + raise ConfigurationError("invalid_pool_bounds", "Pool sizes must satisfy 1 <= min <= max.") if environment.is_production: self._validate_production() @@ -156,6 +161,17 @@ class Settings: raise ConfigurationError("insecure_production_cookie", "Production requires Secure session cookies.") if not self.require_csrf: raise ConfigurationError("insecure_production_csrf", "Production requires CSRF enforcement.") + # The PostgreSQL repository has no usable default. Refusing here means a + # production process cannot start without real credentials rather than + # appearing healthy while holding a development store. The same + # requirement is re-checked at the repository boundary for any + # environment, so a non-production process that selects postgres is + # stopped there instead. + if not self.database_dsn: + raise ConfigurationError( + "production_dsn_missing", + "Production requires a PostgreSQL connection string.", + ) if self.repository_kind != PRODUCTION_REPOSITORY_KIND: raise ConfigurationError( "production_repository_missing", @@ -242,6 +258,13 @@ def load_settings(environ: Mapping[str, str] | None = None) -> Settings: if raw_static: static_root = Path(raw_static).expanduser() + # The DSN is read from the environment and is never logged, echoed by an + # endpoint, or written to a development data file. repr/compare are off on + # the field so it cannot leak through a settings dump. + database_dsn = env.get("UMBRATRANS_DATABASE_DSN") or None + pool_min = _int_from(env.get("UMBRATRANS_DATABASE_POOL_MIN", "1"), name="UMBRATRANS_DATABASE_POOL_MIN", minimum=1) + pool_max = _int_from(env.get("UMBRATRANS_DATABASE_POOL_MAX", "10"), name="UMBRATRANS_DATABASE_POOL_MAX", minimum=1) + allowed_hosts = tuple(host.strip() for host in env.get("UMBRATRANS_ALLOWED_HOSTS", "").split(",") if host.strip()) raw_pepper = env.get("UMBRATRANS_TOKEN_PEPPER") @@ -278,6 +301,9 @@ def load_settings(environ: Mapping[str, str] | None = None) -> Settings: ), repository_kind=repository_kind, repository_path=repository_path, + database_dsn=database_dsn, + database_pool_min=pool_min, + database_pool_max=pool_max, static_root=static_root, allowed_hosts=allowed_hosts, token_pepper=pepper, diff --git a/umbratrans_web/migrations/0001_phase1_identity.sql b/umbratrans_web/migrations/0001_phase1_identity.sql index bc0b730..81b8fb0 100644 --- a/umbratrans_web/migrations/0001_phase1_identity.sql +++ b/umbratrans_web/migrations/0001_phase1_identity.sql @@ -322,6 +322,18 @@ COMMENT ON TABLE public.organizations IS CREATE INDEX IF NOT EXISTS organizations_status_idx ON public.organizations (status); +-- --------------------------------------------------------------------- +-- 4b. audit_events: target lookup +-- --------------------------------------------------------------------- +-- The application records a target as the polymorphic (target_type, +-- target_id) pair the control plane already uses, and the administrator audit +-- view filters on it. audit_events_org_created_idx covers the organization and +-- created_at ordering, but nothing covers the target pair, so that filter would +-- be a sequential scan. Additive index only; no column or constraint changes. + +CREATE INDEX IF NOT EXISTS audit_events_target_idx + ON public.audit_events (target_type, target_id, created_at DESC); + -- --------------------------------------------------------------------- -- 5. RLS AND GRANTS (REVIEWED, NOT EXECUTED) -- --------------------------------------------------------------------- @@ -393,6 +405,10 @@ CREATE INDEX IF NOT EXISTS organizations_status_idx -- -- this migration inserted no tenant -- SELECT count(*) FROM public.organizations; -- +-- -- the audit target index the administrator view depends on exists +-- SELECT indexname, indexdef FROM pg_indexes +-- WHERE tablename = 'audit_events' AND indexname = 'audit_events_target_idx'; +-- -- -- no duplicate, non-null usernames can exist -- SELECT lower(username::text) AS key, count(*) -- FROM public.users WHERE username IS NOT NULL diff --git a/umbratrans_web/migrations/README.md b/umbratrans_web/migrations/README.md index d46e793..d3ee14c 100644 --- a/umbratrans_web/migrations/README.md +++ b/umbratrans_web/migrations/README.md @@ -64,6 +64,35 @@ production schema, so the file states its assumptions instead of guessing. - What are the migration role and application role names? - Is `citext` available to the migration role? If not, use a `lower(...)` unique index instead of a case-sensitive text column. +- **Are the `roles` rows the application needs present, with the right scope?** + The adapter resolves a role by name from `public.roles` and requires + `platform_admin` at scope `platform`, and `organization_admin` and `member` at + scope `organization`. `app.validate_user_role_scope()` enforces the pairing, so + a missing or wrongly scoped role makes account creation fail. Check before + provisioning, not during: + ```sql + SELECT name, scope FROM public.roles + WHERE name IN ('platform_admin', 'organization_admin', 'member') + ORDER BY name; + ``` +- **Does any row carry `users.mfa_required = true`?** The adapter refuses to + resolve an account that requires a second factor, because this phase cannot + satisfy one. Until MFA is implemented, every row must be false or those + accounts are unusable: + ```sql + SELECT count(*) FROM public.users WHERE mfa_required; + ``` +- **Does every user hold at most one active `user_roles` row?** The adapter + refuses to resolve a user with more than one active assignment rather than + pick one, because the organization on the assignment becomes the scope of + every administrative decision: + ```sql + SELECT user_id, count(*) FROM public.user_roles + WHERE status = 'active' GROUP BY 1 HAVING count(*) > 1; + ``` +- **Do all existing users have an `email`?** It is `NOT NULL`, so any row added + by another system already has one; the check confirms the constraint the + application depends on is the live one. - Does a generic `set_updated_at()` trigger helper exist, or does the application maintain `updated_at`? - Is `sessions` large enough that `CREATE INDEX` must be replaced with diff --git a/umbratrans_web/models.py b/umbratrans_web/models.py index 07a43f2..e49e67d 100644 --- a/umbratrans_web/models.py +++ b/umbratrans_web/models.py @@ -169,12 +169,22 @@ def normalize_display_name(value: str) -> str: return text -def normalize_email(value: str | None) -> str | None: +def normalize_email(value: str | None) -> str: + """Return the canonical stored form of an email address. + + Required, not optional. The control plane declares ``users.email`` as + ``NOT NULL UNIQUE`` because it is the account's contact and recovery + address. Accepting a missing email here would let account creation succeed + against the development repositories and then fail on the authoritative + store, so the rule is enforced in the domain where it behaves the same + everywhere. + """ + if value is None: - return None + raise InputError("email_required", "An email address is required.") text = value.strip() if not text: - return None + raise InputError("email_required", "An email address is required.") if len(text) > MAX_EMAIL_LENGTH or not _EMAIL_RE.fullmatch(text): raise InputError("invalid_email", "Email address is not valid.") return text.lower() @@ -347,9 +357,16 @@ class SessionRecord: @dataclass(frozen=True, slots=True) class AuditEvent: - """A sanitized record of a material account or session change.""" + """A sanitized record of a material account or session change. - id: UUID + ``id`` is a store-owned identifier rendered as a string. The control plane + uses a ``bigint`` identity and the in-memory repository uses a UUID, so the + domain does not claim a type the stores disagree about. The application + correlation UUID is kept in ``detail`` under ``application_event_id`` when + the store allows it, and never displaces the store's own key. + """ + + id: str occurred_at: datetime action: str outcome: str diff --git a/umbratrans_web/postgres_repository.py b/umbratrans_web/postgres_repository.py new file mode 100644 index 0000000..c97b1ba --- /dev/null +++ b/umbratrans_web/postgres_repository.py @@ -0,0 +1,995 @@ +"""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. The +caller's identity is published with ``SET LOCAL app.current_user_id`` so that +row-level-security policies, once the reviewed policies are applied, 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. + +Until those policies exist the ``SET LOCAL`` 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 are what actually +enforce scope today. +""" + +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 + +import psycopg +from psycopg.rows import dict_row +from psycopg_pool import ConnectionPool + +from .errors import ConflictError, NotFoundError, RepositoryError +from .models import ( + AccountStatus, + AuditEvent, + CredentialRecord, + CredentialType, + Organization, + OrganizationStatus, + RoleName, + SessionRecord, + User, + username_key, +) +from .repository import AccountRepository +from .security import utc_now + +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 row-level-security policies are expected to read. +RLS_USER_SETTING = "app.current_user_id" + +#: 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 + + # -- transaction plumbing ---------------------------------------------- + + @contextmanager + def _connection(self, *, user_id: UUID | None = None) -> Iterator[psycopg.Connection[Any]]: + """Yield a connection inside one transaction, publishing the caller. + + ``SET LOCAL`` is transaction-scoped, which is what makes connection + pooling safe here: a returned connection cannot carry the previous + request's identity. + """ + + with self._pool.connection() as connection: + with connection.transaction(): + connection.execute("SET LOCAL statement_timeout = %s", (STATEMENT_TIMEOUT_MS,)) + connection.execute("SET LOCAL idle_in_transaction_session_timeout = '10s'") + if user_id is not None: + connection.execute(f"SET LOCAL {RLS_USER_SETTING} = %s", (str(user_id),)) + 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_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 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( + "EXISTS (SELECT 1 FROM public.user_roles ur WHERE ur.user_id = u.id AND ur.organization_id = %s AND ur.status = 'active')" + ) + 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( + "EXISTS (SELECT 1 FROM public.user_roles ur JOIN public.roles r ON r.id = ur.role_id WHERE ur.user_id = u.id AND r.name = %s AND ur.status = 'active')" + ) + 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 " + "WHERE ur.user_id = u.id AND ur.organization_id = %s AND ur.status = 'active')", + [organization_id], + ) + + # -- 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 active assignment match the requested role and organization. + + Phase 1 models exactly one role and one organization per account, so + the target state is fully determined. A user holding several active + assignments is refused by :func:`_resolve_assignment` on read; this + write path refuses to collapse them silently either. + """ + + 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 WHERE user_id = %s AND status = 'active'", + (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 + + # -- 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( + """ + 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 + """, + ( + 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. + + Phase 1 can represent exactly one active assignment. A user with more than + one active row is refused rather than resolved to an arbitrary row: the + organization on the returned assignment becomes the scope of every + administrative decision, so guessing would hand out the wrong tenant's + authority. Reconciling this with project-scoped membership is the open + review question recorded in ``migrations/README.md``. + """ + + 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' + 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"], + ) + + +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"} + 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=(row["target_id"] if row["target_type"] == "user" else None), + organization_id=row["organization_id"], + request_id=row["request_id"], + detail=detail, + ) + + +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_SETTING", "STATEMENT_TIMEOUT_MS"] diff --git a/umbratrans_web/repository.py b/umbratrans_web/repository.py index c3fc2d2..c3f386e 100644 --- a/umbratrans_web/repository.py +++ b/umbratrans_web/repository.py @@ -353,7 +353,8 @@ def build_repository(settings: object, *, repository: AccountRepository | None = """Return the requested repository implementation. Kept in one place so the app factory never guesses a store. The PostgreSQL - adapter is intentionally absent from this slice. + adapter is imported lazily so a deployment that never selects it does not + pay for ``psycopg_pool``. """ if repository is not None: @@ -366,9 +367,21 @@ def build_repository(settings: object, *, repository: AccountRepository | None = return JsonDevelopmentRepository(settings) # type: ignore[arg-type] if kind == "postgres": - raise RepositoryError( - "postgres_adapter_unavailable", - "The PostgreSQL adapter is not part of this development slice.", + from .postgres_repository import PostgresAccountRepository + + dsn = getattr(settings, "database_dsn", None) + if not dsn: + # Refuse to start rather than fall back to a development store, which + # would let a production deployment appear to work while holding no + # real data. + raise RepositoryError( + "database_dsn_missing", + "The PostgreSQL repository requires a connection string.", + ) + return PostgresAccountRepository.from_dsn( + dsn, + min_size=int(getattr(settings, "database_pool_min", 1)), + max_size=int(getattr(settings, "database_pool_max", 10)), ) raise RepositoryError("unknown_repository", "Unknown repository kind.") diff --git a/umbratrans_web/schemas.py b/umbratrans_web/schemas.py index 2899ab8..48fb3f8 100644 --- a/umbratrans_web/schemas.py +++ b/umbratrans_web/schemas.py @@ -159,7 +159,9 @@ class RevokeSessionsResponse(BaseModel): class AuditEventResponse(BaseModel): - id: UUID + # A store-owned identifier, so a string rather than the UUID the + # development repository happens to use. + id: str occurred_at: datetime action: str outcome: str @@ -208,13 +210,18 @@ class CreateAccountRequest(RequestModel): a submitted ``password`` is rejected outright. For ``enrollment_mode="temporary_password"`` the service generates the one-time password itself and returns it once in :class:`CreateAccountResponse`. + + ``email`` is required because the control plane declares ``users.email`` as + ``NOT NULL UNIQUE``. Making it mandatory here means a missing address is a + clear field-level rejection rather than a failure that only appears once + the authoritative store is in use. """ username: str = Field(min_length=1, max_length=64) display_name: str = Field(min_length=1, max_length=120) + email: str = Field(min_length=3, max_length=254) enrollment_mode: EnrollmentMode = EnrollmentMode.OIDC role: RoleName = RoleName.MEMBER - email: str | None = Field(default=None, max_length=254) organization_id: UUID | None = None diff --git a/umbratrans_web/service.py b/umbratrans_web/service.py index 3d84605..0d72a11 100644 --- a/umbratrans_web/service.py +++ b/umbratrans_web/service.py @@ -190,8 +190,8 @@ class AccountService: enrollment_mode: str, username: str, display_name: str, + email: str, role: str = str(RoleName.MEMBER), - email: str | None = None, organization_id: UUID | None = None, password: str | None = None, actor: Principal | None = None, @@ -200,6 +200,11 @@ class AccountService: ) -> AccountView: """Create an account through one of the two enrollment modes. + ``email`` is required. The control plane declares ``users.email`` as + ``NOT NULL UNIQUE``, and enforcing it here means a missing address + fails identically against the development repositories and the + authoritative store instead of only at deployment. + ``temporary_password`` is the operator path and requires ``operator`` plus ``password``; use :meth:`provision_admin_temporary_password` for administrator-driven enrollment. ``oidc`` requires an administrator @@ -234,7 +239,7 @@ class AccountService: username: str, display_name: str, role: str, - email: str | None, + email: str, organization_id: UUID | None, password: str | None, operator: OperatorContext | None, @@ -292,7 +297,7 @@ class AccountService: username: str, display_name: str, role: str = str(RoleName.MEMBER), - email: str | None = None, + email: str, organization_id: UUID | None = None, actor: Principal | None = None, request_id: str | None = None, @@ -354,8 +359,8 @@ class AccountService: *, username: str, display_name: str, + email: str, role: str = str(RoleName.MEMBER), - email: str | None = None, organization_id: UUID | None = None, actor: Principal | None = None, request_id: str | None = None, @@ -402,7 +407,7 @@ class AccountService: username: str, display_name: str, role: str, - email: str | None, + email: str, organization_id: UUID | None, credential_type: CredentialType, password_hash: str | None, @@ -1281,7 +1286,7 @@ class AccountService: detail: Mapping[str, object] | None = None, ) -> AuditEvent: event = AuditEvent( - id=uuid4(), + id=str(uuid4()), occurred_at=self._now(), action=action, outcome=outcome, diff --git a/umbratrans_web/tests/conftest.py b/umbratrans_web/tests/conftest.py index 40fd2b6..9467d85 100644 --- a/umbratrans_web/tests/conftest.py +++ b/umbratrans_web/tests/conftest.py @@ -9,6 +9,7 @@ or otherwise. from __future__ import annotations +import re import secrets import string from collections.abc import Iterator @@ -36,6 +37,11 @@ from umbratrans_web.service import AccountService, OperatorContext ALPHABET = string.ascii_letters + string.digits + "-_.@" +#: A syntactically valid connection string used only to satisfy the production +#: configuration gate. No test ever opens it: the PostgreSQL integration suite +#: reads UMBRATRANS_TEST_DSN and skips when it is unset. +TEST_DSN = "postgresql://app@db.internal/umbracontrol" + def random_password(length: int = 24) -> str: """Return a random password for one test run. @@ -163,6 +169,19 @@ def api(client: TestClient, settings: Settings) -> ApiHarness: return ApiHarness(client, settings) +def random_email(seed: str) -> str: + """Return a unique, syntactically valid address for one test account. + + The control plane declares ``users.email`` NOT NULL UNIQUE, so email is a + required field. Generated per test run rather than written as a literal so + no fixture file carries a fixed identifier. The domain is fixed because a + username may contain dots that are not part of a domain name. + """ + + local = re.sub(r"[^a-z0-9]+", "", seed.lower())[:40] or "user" + return f"{local}.{secrets.token_hex(4)}@example.test" + + def provision_local( service: AccountService, *, @@ -171,14 +190,20 @@ def provision_local( display_name: str = "Local Operator", role: str = str(RoleName.PLATFORM_ADMIN), organization_id: UUID | None = None, + email: str | None = None, ) -> SeededAccount: - """Create a local temporary-password account through the operator path.""" + """Create a local temporary-password account through the operator path. + + ``email=None`` is passed through unchanged so a test can assert that a + missing address is refused; every other caller gets a generated one. + """ candidate = password or random_password() view = service.create_account( enrollment_mode="temporary_password", username=username, display_name=display_name, + email=email if email is not None else random_email(username), role=role, organization_id=organization_id, password=candidate, @@ -195,12 +220,14 @@ def provision_oidc( display_name: str = "OIDC User", role: str = str(RoleName.MEMBER), organization_id: UUID | None = None, + email: str | None = None, ) -> SeededAccount: """Create an OIDC-enrollment account as an administrator.""" view = service.create_oidc_account( username=username, display_name=display_name, + email=email or random_email(username), role=role, organization_id=organization_id, actor=actor, @@ -258,6 +285,7 @@ def member_principal(account: SeededAccount, *, organization_id: UUID | None = N __all__ = [ "ALPHABET", + "TEST_DSN", "ApiHarness", "SeededAccount", "activated_admin", @@ -267,6 +295,7 @@ __all__ = [ "member_principal", "provision_local", "provision_oidc", + "random_email", "random_password", "repository", "service", diff --git a/umbratrans_web/tests/test_api_auth.py b/umbratrans_web/tests/test_api_auth.py index faef90f..f5ce787 100644 --- a/umbratrans_web/tests/test_api_auth.py +++ b/umbratrans_web/tests/test_api_auth.py @@ -11,7 +11,7 @@ from umbratrans_web.errors import ConfigurationError from umbratrans_web.repository import InMemoryAccountRepository from umbratrans_web.service import AccountService -from .conftest import ApiHarness, activated_admin, provision_local, random_password +from .conftest import TEST_DSN, ApiHarness, activated_admin, provision_local, random_password # ---------------------------------------------------------------------- # Health @@ -342,7 +342,12 @@ def test_forged_session_cookie_is_rejected(client: TestClient, settings: Setting def test_production_cookies_are_secure(settings: Settings) -> None: - production = Settings(environment=Environment.PRODUCTION, repository_kind="postgres", cookie_secure=True) + production = Settings( + environment=Environment.PRODUCTION, + repository_kind="postgres", + database_dsn=TEST_DSN, + cookie_secure=True, + ) repository = InMemoryAccountRepository() service = AccountService(repository, production) app = create_app(settings=production, repository=repository, service=service) @@ -360,12 +365,18 @@ def test_production_cookies_are_secure(settings: Settings) -> None: def test_production_refuses_insecure_configuration() -> None: with pytest.raises(ConfigurationError) as error: - Settings(environment=Environment.PRODUCTION, repository_kind="postgres", cookie_secure=False) + Settings( + environment=Environment.PRODUCTION, + database_dsn=TEST_DSN, + repository_kind="postgres", + cookie_secure=False, + ) assert error.value.code == "insecure_production_cookie" with pytest.raises(ConfigurationError) as csrf_error: Settings( environment=Environment.PRODUCTION, + database_dsn=TEST_DSN, repository_kind="postgres", cookie_secure=True, require_csrf=False, @@ -373,12 +384,22 @@ def test_production_refuses_insecure_configuration() -> None: assert csrf_error.value.code == "insecure_production_csrf" with pytest.raises(ConfigurationError) as repository_error: - Settings(environment=Environment.PRODUCTION, repository_kind="memory", cookie_secure=True) + Settings( + environment=Environment.PRODUCTION, + database_dsn=TEST_DSN, + repository_kind="memory", + cookie_secure=True, + ) assert repository_error.value.code == "production_repository_missing" def test_production_app_requires_an_explicit_repository() -> None: - production = Settings(environment=Environment.PRODUCTION, repository_kind="postgres", cookie_secure=True) + production = Settings( + environment=Environment.PRODUCTION, + repository_kind="postgres", + database_dsn=TEST_DSN, + cookie_secure=True, + ) with pytest.raises(ConfigurationError) as error: create_app(settings=production) assert error.value.code == "production_repository_missing" diff --git a/umbratrans_web/tests/test_api_uadmin.py b/umbratrans_web/tests/test_api_uadmin.py index 0059987..dd56248 100644 --- a/umbratrans_web/tests/test_api_uadmin.py +++ b/umbratrans_web/tests/test_api_uadmin.py @@ -10,7 +10,14 @@ from fastapi.testclient import TestClient from umbratrans_web.service import AccountService -from .conftest import ApiHarness, activated_admin, ensure_organization, provision_local, random_password +from .conftest import ( + ApiHarness, + activated_admin, + ensure_organization, + provision_local, + random_email, + random_password, +) @pytest.fixture @@ -199,6 +206,7 @@ def test_generated_temporary_password_must_be_replaced_on_first_login( json={ "username": "First.Login", "display_name": "First Login", + "email": random_email("first-login"), "enrollment_mode": "temporary_password", }, ) @@ -242,6 +250,7 @@ def test_generated_temporary_passwords_are_unique_per_account( json={ "username": f"Bulk.User{index}", "display_name": f"Bulk User {index}", + "email": random_email(f"bulk-{index}"), "enrollment_mode": "temporary_password", }, ).json()["temporary_password"] @@ -262,6 +271,7 @@ def test_organization_admin_cannot_provision_outside_its_scope(api: ApiHarness, json={ "username": "Outside.Temp", "display_name": "Outside Temp", + "email": random_email("outside-temp"), "enrollment_mode": "temporary_password", "organization_id": str(org_b), }, @@ -271,6 +281,7 @@ def test_organization_admin_cannot_provision_outside_its_scope(api: ApiHarness, json={ "username": "Escalate.Temp", "display_name": "Escalate Temp", + "email": random_email("escalate-temp"), "enrollment_mode": "temporary_password", "role": "platform_admin", }, @@ -285,7 +296,12 @@ def test_organization_admin_cannot_provision_outside_its_scope(api: ApiHarness, def test_create_account_validates_input(api: ApiHarness, admin: object) -> None: duplicate = api.post( "/api/uadmin/accounts", - json={"username": "Umbraadmin", "display_name": "Duplicate", "enrollment_mode": "oidc"}, + json={ + "username": "Umbraadmin", + "display_name": "Duplicate", + "email": random_email("dup"), + "enrollment_mode": "oidc", + }, ) invalid = api.post( "/api/uadmin/accounts", @@ -293,7 +309,12 @@ def test_create_account_validates_input(api: ApiHarness, admin: object) -> None: ) unknown_mode = api.post( "/api/uadmin/accounts", - json={"username": "Mode.User", "display_name": "Mode", "enrollment_mode": "carrier_pigeon"}, + json={ + "username": "Mode.User", + "display_name": "Mode", + "email": random_email("mode"), + "enrollment_mode": "carrier_pigeon", + }, ) assert duplicate.status_code == 409 @@ -362,6 +383,7 @@ def test_organization_admin_scope_is_enforced(api: ApiHarness, service: AccountS json={ "username": "Escape.User", "display_name": "Escape", + "email": random_email("escape"), "enrollment_mode": "oidc", "organization_id": str(org_b), }, @@ -373,6 +395,7 @@ def test_organization_admin_scope_is_enforced(api: ApiHarness, service: AccountS json={ "username": "Escalate.User", "display_name": "Escalate", + "email": random_email("escalate"), "enrollment_mode": "oidc", "role": "platform_admin", "organization_id": str(org_a), @@ -510,7 +533,12 @@ def test_audit_is_scoped_to_the_callers_organization(api: ApiHarness, service: A def test_audit_events_reference_the_request_id(api: ApiHarness, admin: object, service: AccountService) -> None: response = api.post( "/api/uadmin/accounts", - json={"username": "Traced.User", "display_name": "Traced", "enrollment_mode": "oidc"}, + json={ + "username": "Traced.User", + "display_name": "Traced", + "email": random_email("traced"), + "enrollment_mode": "oidc", + }, ) assert response.status_code == 201 request_id = response.headers["x-request-id"] diff --git a/umbratrans_web/tests/test_cli.py b/umbratrans_web/tests/test_cli.py index 4ed15c6..3803aba 100644 --- a/umbratrans_web/tests/test_cli.py +++ b/umbratrans_web/tests/test_cli.py @@ -26,7 +26,7 @@ from umbratrans_web.config import Environment, Settings from umbratrans_web.dev_repository import JsonDevelopmentRepository from umbratrans_web.service import AccountService -from .conftest import random_password +from .conftest import random_email, random_password class _FakeTty(io.StringIO): @@ -78,6 +78,11 @@ def _run( ) -> tuple[int, str, str]: """Run the CLI with an injected prompt. Values are generated per run.""" + # users.email is NOT NULL in the control plane, so --email is required. A + # generated address is supplied unless a test is exercising the flag. + if "--email" not in argv: + argv = [*argv, "--email", random_email("ops-admin")] + out = _Sink() err = _Sink() # The command always prompts twice; a single value is reused for both. diff --git a/umbratrans_web/tests/test_config.py b/umbratrans_web/tests/test_config.py index 7828992..7456d92 100644 --- a/umbratrans_web/tests/test_config.py +++ b/umbratrans_web/tests/test_config.py @@ -17,6 +17,8 @@ from umbratrans_web.config import ( from umbratrans_web.errors import ConfigurationError from umbratrans_web.security import SecretRedactionFilter +from .conftest import TEST_DSN + def test_defaults_are_development_and_insecure_only_where_allowed() -> None: settings = Settings() @@ -41,29 +43,58 @@ def test_load_settings_refuses_unknown_environments() -> None: def test_load_settings_refuses_insecure_production() -> None: with pytest.raises(ConfigurationError) as cookie_error: - load_settings({"UMBRA_ENV": "production"}) + load_settings({"UMBRA_ENV": "production", "UMBRATRANS_DATABASE_DSN": TEST_DSN}) assert cookie_error.value.code == "insecure_production_cookie" with pytest.raises(ConfigurationError) as csrf_error: load_settings( - {"UMBRA_ENV": "production", "UMBRATRANS_COOKIE_SECURE": "true", "UMBRATRANS_REQUIRE_CSRF": "false"} + { + "UMBRA_ENV": "production", + "UMBRATRANS_COOKIE_SECURE": "true", + "UMBRATRANS_REQUIRE_CSRF": "false", + "UMBRATRANS_DATABASE_DSN": TEST_DSN, + } ) assert csrf_error.value.code == "insecure_production_csrf" -def test_production_settings_select_the_absent_postgres_adapter() -> None: - settings = load_settings({"UMBRA_ENV": "production", "UMBRATRANS_COOKIE_SECURE": "true"}) +def test_production_settings_select_the_postgres_adapter() -> None: + without_dsn = {"UMBRA_ENV": "production", "UMBRATRANS_COOKIE_SECURE": "true"} + with pytest.raises(ConfigurationError) as missing: + load_settings(without_dsn) + assert missing.value.code == "production_dsn_missing" + settings = load_settings({**without_dsn, "UMBRATRANS_DATABASE_DSN": "postgresql://app@db.internal/umbracontrol"}) assert settings.is_production is True assert settings.repository_kind == "postgres" assert settings.cookie_secure is True - # The adapter is not part of this slice, so the app cannot start. + assert settings.database_dsn is not None + + +def test_production_refuses_a_missing_dsn_at_the_repository_boundary() -> None: from umbratrans_web.errors import RepositoryError from umbratrans_web.repository import build_repository + settings = Settings(environment=Environment.TEST, repository_kind="postgres") with pytest.raises(RepositoryError) as error: build_repository(settings) - assert error.value.code == "postgres_adapter_unavailable" + assert error.value.code == "database_dsn_missing" + + +def test_the_dsn_is_never_rendered_in_a_settings_dump() -> None: + settings = Settings( + environment=Environment.TEST, + repository_kind="json", + # Obviously fake, and never opened: nothing in the suite connects. + database_dsn="postgresql://app:not-a-real-secret@db.invalid/umbracontrol", + ) + rendered = repr(settings) + assert "not-a-real-secret" not in rendered + assert "postgresql://" not in rendered + # repr=False on the field also keeps it out of equality comparisons, so two + # settings objects differing only by DSN are not distinguishable by dump. + assert "not-a-real-secret" not in str(settings) + assert "not-a-real-secret" not in f"{settings!r} {settings}" def test_load_settings_parses_json_repository_paths() -> None: diff --git a/umbratrans_web/tests/test_organizations.py b/umbratrans_web/tests/test_organizations.py index 54e66fe..8737e43 100644 --- a/umbratrans_web/tests/test_organizations.py +++ b/umbratrans_web/tests/test_organizations.py @@ -25,6 +25,7 @@ from .conftest import ( ensure_organization, provision_local, provision_oidc, + random_email, random_password, ) @@ -236,12 +237,11 @@ def test_unknown_organization_is_not_found(service: AccountService, platform_adm def test_an_account_cannot_be_assigned_to_an_unknown_organization( service: AccountService, platform_admin: object ) -> None: - from uuid import uuid4 - with pytest.raises(NotFoundError) as error: service.create_oidc_account( username="Orphan", display_name="Orphan", + email=random_email("orphan"), organization_id=uuid4(), actor=platform_admin, # type: ignore[arg-type] ) @@ -256,6 +256,7 @@ def test_an_account_cannot_be_assigned_to_a_suspended_organization( service.create_oidc_account( username="Late.Joiner", display_name="Late Joiner", + email=random_email("late-joiner"), organization_id=org.id, actor=platform_admin, # type: ignore[arg-type] ) @@ -443,6 +444,7 @@ def test_account_creation_validates_the_organization(signed_in_admin: ApiHarness json={ "username": "Orphan", "display_name": "Orphan", + "email": random_email("orphan"), "enrollment_mode": "temporary_password", "organization_id": str(uuid4()), }, @@ -457,6 +459,7 @@ def test_account_creation_validates_the_organization(signed_in_admin: ApiHarness json={ "username": "Late.Joiner", "display_name": "Late", + "email": random_email("late"), "enrollment_mode": "temporary_password", "organization_id": organization["id"], }, diff --git a/umbratrans_web/tests/test_postgres_repository.py b/umbratrans_web/tests/test_postgres_repository.py new file mode 100644 index 0000000..2ad4067 --- /dev/null +++ b/umbratrans_web/tests/test_postgres_repository.py @@ -0,0 +1,514 @@ +"""Tests for the PostgreSQL adapter. + +Two layers, because the two have very different verification status. + +**Mapping and configuration tests run everywhere.** The row mappers, the +role-assignment rules, and the unique-violation mapping are pure functions, so +they are asserted directly against synthetic rows. These are the places where a +silent misreading of the control-plane schema would produce a wrong tenant scope +rather than an obvious crash, which is why they are covered in detail. + +**Integration tests require a real database and skip without one.** They are +selected by ``UMBRATRANS_TEST_DSN`` and must point at a *scratch* database that +the phase-1 migration has been applied to, never at ``umbracontrol``. They are +not expected to pass in an environment with no database access, and a green run +here says nothing about them either way. +""" + +from __future__ import annotations + +import os +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any + +import pytest + +from umbratrans_web.errors import ConflictError, RepositoryError +from umbratrans_web.models import AccountStatus, OrganizationStatus, RoleName, User +from umbratrans_web.postgres_repository import ( + RLS_USER_SETTING, + STATEMENT_TIMEOUT_MS, + PostgresAccountRepository, + _assignment_shape, + _audit_from_row, + _organization_from_row, + _session_from_row, + _unique_conflict, + _user_from_row, +) +from umbratrans_web.repository import AccountRepository + +TEST_DSN_ENV = "UMBRATRANS_TEST_DSN" + + +def _row(**overrides: Any) -> dict[str, Any]: + """One ``users`` row as psycopg would return it with ``dict_row``.""" + + now = datetime(2026, 1, 1, tzinfo=timezone.utc) + base: dict[str, Any] = { + "id": uuid.uuid4(), + "username": "Umbraadmin", + "display_name": "Umbra Administrator", + "email": "umbraadmin@example.test", + "status": "active", + "created_at": now, + "updated_at": now, + } + base.update(overrides) + return base + + +# ---------------------------------------------------------------------- +# Protocol conformance +# ---------------------------------------------------------------------- + + +def test_the_adapter_satisfies_the_repository_protocol() -> None: + for name in AccountRepository.__protocol_attrs__: # type: ignore[attr-defined] + assert hasattr(PostgresAccountRepository, name), name + + +def test_the_adapter_imports_without_opening_a_connection() -> None: + """Importing the module must not touch the network. + + ``psycopg_pool`` is imported at module scope, so a missing database is a + runtime failure at construction, never an import-time surprise. + """ + + assert PostgresAccountRepository.__doc__ is not None + assert RLS_USER_SETTING == "app.current_user_id" + assert STATEMENT_TIMEOUT_MS > 0 + + +# ---------------------------------------------------------------------- +# Role and organization resolution +# ---------------------------------------------------------------------- + + +def test_a_platform_admin_is_assigned_at_platform_scope() -> None: + """A platform assignment must carry no organization. + + This is what the ``user_roles`` check constraint requires, and getting it + wrong is a database error on insert rather than a silent bug. + """ + + user = User( + id=uuid.uuid4(), + username="Umbraadmin", + username_key="umbraadmin", + display_name="Umbra Administrator", + role=RoleName.PLATFORM_ADMIN, + status=AccountStatus.ACTIVE, + ) + assert _assignment_shape(user) == ("platform", None) + + +def test_an_organization_scoped_role_requires_an_organization() -> None: + """An organization assignment without one is refused before the database. + + The same combination is rejected by the ``user_roles`` check constraint; + refusing here names the real problem instead of surfacing a constraint + violation. + """ + + user = User( + id=uuid.uuid4(), + username="Orphan.Member", + username_key="orphan.member", + display_name="Orphan", + role=RoleName.MEMBER, + status=AccountStatus.ACTIVE, + ) + with pytest.raises(RepositoryError) as error: + _assignment_shape(user) + assert error.value.code == "organization_required" + + +def test_a_member_carries_its_organization() -> None: + organization_id = uuid.uuid4() + user = User( + id=uuid.uuid4(), + username="Member.One", + username_key="member.one", + display_name="Member One", + role=RoleName.MEMBER, + status=AccountStatus.ACTIVE, + organization_id=organization_id, + ) + assert _assignment_shape(user) == ("organization", organization_id) + + +def test_an_organization_admin_is_also_organization_scoped() -> None: + organization_id = uuid.uuid4() + user = User( + id=uuid.uuid4(), + username="Org.Admin", + username_key="org.admin", + display_name="Org Admin", + role=RoleName.ORGANIZATION_ADMIN, + status=AccountStatus.ACTIVE, + organization_id=organization_id, + ) + assert _assignment_shape(user) == ("organization", organization_id) + + +# ---------------------------------------------------------------------- +# User row mapping +# ---------------------------------------------------------------------- + + +def test_a_user_row_maps_with_a_resolved_assignment() -> None: + row = _row() + organization_id = uuid.uuid4() + user = _user_from_row(row, role=RoleName.MEMBER, organization_id=organization_id) + + assert user.id == row["id"] + assert user.username == "Umbraadmin" + assert user.username_key == "umbraadmin" + assert user.role is RoleName.MEMBER + assert user.status is AccountStatus.ACTIVE + assert user.organization_id == organization_id + assert user.email == "umbraadmin@example.test" + + +def test_a_user_row_falls_back_to_the_username_when_the_name_is_null() -> None: + """users.display_name is nullable in the control plane.""" + + user = _user_from_row(_row(display_name=None)) + assert user.display_name == "Umbraadmin" + + +@pytest.mark.parametrize("status", ["active", "suspended", "deactivated"]) +def test_modelled_account_statuses_are_accepted(status: str) -> None: + assert _user_from_row(_row(status=status)).status is AccountStatus(status) + + +def test_an_account_requiring_mfa_is_refused() -> None: + """The control plane can require a second factor this phase cannot do. + + Authenticating such an account anyway would grant access to someone who was + supposed to present a second factor, so the read fails closed. + """ + + with pytest.raises(RepositoryError) as error: + _user_from_row(_row(mfa_required=True)) + assert error.value.code == "mfa_required_unsupported" + + +def test_a_profile_row_without_the_mfa_column_is_accepted() -> None: + """Tolerates a row shape that predates the column being selected.""" + + assert _user_from_row(_row()).status is AccountStatus.ACTIVE + + +@pytest.mark.parametrize("status", ["invited", "locked", "paused", "ACTIVE"]) +def test_an_unmodelled_status_is_refused_rather_than_coerced(status: str) -> None: + """``invited`` and ``locked`` exist in the control plane but not here. + + Mapping one of them onto ``active`` would grant access to an account that + the control plane considers not yet usable, so the read fails closed. + """ + + with pytest.raises(RepositoryError) as error: + _user_from_row(_row(status=status)) + assert error.value.code == "unmapped_account_status" + + +# ---------------------------------------------------------------------- +# Organization row mapping +# ---------------------------------------------------------------------- + + +def test_an_organization_row_maps() -> None: + now = datetime(2026, 1, 1, tzinfo=timezone.utc) + organization = _organization_from_row( + { + "id": uuid.uuid4(), + "name": "Northstar Media", + "slug": "northstar-media", + "status": "active", + "created_at": now, + "updated_at": now, + } + ) + assert organization.name == "Northstar Media" + assert organization.slug == "northstar-media" + assert organization.status is OrganizationStatus.ACTIVE + + +def test_an_unmodelled_organization_status_is_refused() -> None: + now = datetime(2026, 1, 1, tzinfo=timezone.utc) + with pytest.raises(ValueError): + _organization_from_row( + { + "id": uuid.uuid4(), + "name": "Odd", + "slug": "odd", + "status": "deleted", + "created_at": now, + "updated_at": now, + } + ) + + +# ---------------------------------------------------------------------- +# Session row mapping +# ---------------------------------------------------------------------- + + +def test_a_session_row_maps_expires_at_to_the_absolute_deadline() -> None: + """``sessions.expires_at`` predates this application. + + The adapter keeps the two equal on write, so anything else reading + ``expires_at`` sees the same deadline. + """ + + created = datetime(2026, 1, 1, tzinfo=timezone.utc) + session = _session_from_row( + { + "id": uuid.uuid4(), + "user_id": uuid.uuid4(), + "token_hash": "a" * 64, + "created_at": created, + "last_seen_at": created + timedelta(minutes=1), + "idle_deadline_at": created + timedelta(minutes=30), + "expires_at": created + timedelta(hours=12), + "revoked_at": None, + "mfa_verified_at": None, + } + ) + assert session.absolute_deadline_at == created + timedelta(hours=12) + assert session.idle_deadline_at == created + timedelta(minutes=30) + assert session.is_revoked() is False + # A hash must not surface through repr. + assert "a" * 64 not in repr(session) + + +# ---------------------------------------------------------------------- +# Audit row mapping +# ---------------------------------------------------------------------- + + +def test_an_audit_row_reports_the_store_identifier() -> None: + """``audit_events.id`` is a bigint identity owned by the database. + + The application mints a UUID for correlation, which is preserved in the + detail, but the API must report the identifier the store will actually find. + """ + + event = _audit_from_row( + { + "id": 4211, + "actor_user_id": uuid.uuid4(), + "organization_id": None, + "action": "account.created", + "target_type": "user", + "target_id": str(uuid.uuid4()), + "request_id": "req-1", + "metadata": {"role": "member", "application_event_id": str(uuid.uuid4())}, + "created_at": datetime(2026, 1, 1, tzinfo=timezone.utc), + }, + original=None, + ) + assert event.id == "4211" + assert event.action == "account.created" + assert event.target_user_id is not None + # The correlation UUID is retained for the caller but is not the row key. + assert "application_event_id" not in event.detail + assert event.detail["role"] == "member" + + +def test_a_non_user_target_maps_to_no_user() -> None: + event = _audit_from_row( + { + "id": 7, + "actor_user_id": None, + "organization_id": uuid.uuid4(), + "action": "organization.created", + "target_type": "organization", + "target_id": str(uuid.uuid4()), + "request_id": None, + "metadata": {}, + "created_at": datetime(2026, 1, 1, tzinfo=timezone.utc), + }, + original=None, + ) + assert event.target_user_id is None + + +# ---------------------------------------------------------------------- +# Conflict mapping +# ---------------------------------------------------------------------- + + +class _FakeDiagnostic: + def __init__(self, constraint_name: str) -> None: + self.constraint_name = constraint_name + + +class _FakeUniqueViolation(Exception): + def __init__(self, constraint_name: str) -> None: + super().__init__("duplicate key value violates unique constraint") + self.diag = _FakeDiagnostic(constraint_name) + + +def test_a_username_violation_maps_to_username_taken() -> None: + error = _unique_conflict(_FakeUniqueViolation("users_username_key"), "Umbraadmin") + assert isinstance(error, ConflictError) + assert error.code == "username_taken" + + +def test_an_email_violation_maps_to_email_taken() -> None: + error = _unique_conflict(_FakeUniqueViolation("users_email_key"), "Umbraadmin") + assert error.code == "email_taken" + + +def test_an_unknown_constraint_falls_back_to_the_username_rule() -> None: + error = _unique_conflict(_FakeUniqueViolation("something_else"), "Umbraadmin") + assert error.code == "username_taken" + + +# ---------------------------------------------------------------------- +# Adapter source hygiene +# ---------------------------------------------------------------------- + + +def test_the_adapter_never_interpolates_a_value_into_sql() -> None: + """Every value reaches PostgreSQL as a bound parameter. + + The statements are assembled from fixed fragments only; a value is never + formatted into the string. This asserts the absence of the common mistake + rather than trying to prove the presence of the correct pattern. + """ + + import inspect + + source = inspect.getsource(PostgresAccountRepository) + for pattern in ( + f"{{{{{'user'}.username}}}}", + f"{{{{{'user'}.email}}}}", + f"{{{{{'event'}.action}}}}", + f"{{{{{'organization'}.slug}}}}", + ): + assert pattern not in source, pattern + + +def test_the_adapter_issues_no_destructive_statement() -> None: + """Insert and update are the adapter's job; nothing that destroys data is. + + ``INSERT INTO`` is expected here, unlike in the migration file. What must + never appear is anything that drops, truncates, deletes, or alters schema: + those belong in a reviewed migration, not in request handling. + """ + + import inspect + + source = inspect.getsource(PostgresAccountRepository) + for statement in ( + "DROP ", + "TRUNCATE", + "DELETE FROM", + "ALTER TABLE", + "CREATE TABLE", + "GRANT ", + "REVOKE ", + ): + assert statement not in source, statement + + +def test_the_adapter_never_selects_a_hash_into_a_user_row() -> None: + """A ``users`` read must not drag a credential hash along with the profile. + + Checked against the actual column list of every statement that reads + ``public.users``, rather than a blanket scan of the module, because the + credential reads legitimately do select ``password_hash``. + """ + + import inspect + import re + + source = inspect.getsource(PostgresAccountRepository) + statements = re.findall(r"SELECT\s+(.*?)\s+FROM\s+public\.users", source, re.DOTALL) + assert statements, "expected at least one statement reading public.users" + for columns in statements: + for protected in ("password_hash", "token_hash"): + assert protected not in columns, (protected, columns) + + +# ---------------------------------------------------------------------- +# Integration: requires a real scratch database +# ---------------------------------------------------------------------- + + +def _integration_dsn() -> str | None: + return os.environ.get(TEST_DSN_ENV) or None + + +integration = pytest.mark.skipif( + _integration_dsn() is None, + reason=( + f"{TEST_DSN_ENV} is unset. Point it at a scratch database with " + "migrations/0001_phase1_identity.sql applied. Never at umbracontrol." + ), +) + + +@pytest.fixture +def postgres() -> Any: + dsn = _integration_dsn() + if dsn is None: # pragma: no cover - the skip above already guards this + pytest.skip("no test DSN") + if "umbracontrol" in dsn: # pragma: no cover - a safety net + pytest.skip("refusing to run the adapter suite against umbracontrol") + repository = PostgresAccountRepository.from_dsn( + dsn, min_size=1, max_size=4, application_name="umbratrans-adapter-test" + ) + try: + yield repository + finally: + repository.close() + + +@integration +def test_a_user_round_trips_through_postgres(postgres: PostgresAccountRepository) -> None: + from umbratrans_web.security import utc_now + + user = User( + id=uuid.uuid4(), + username=f"Round.Trip.{uuid.uuid4().hex[:8]}", + username_key="", + display_name="Round Trip", + role=RoleName.MEMBER, + status=AccountStatus.ACTIVE, + organization_id=None, + email=f"round-{uuid.uuid4().hex[:8]}@example.test", + created_at=utc_now(), + updated_at=utc_now(), + ) + # organization_id is required for a member assignment, so this test asserts + # the refusal rather than working around it. + with pytest.raises(RepositoryError) as error: + postgres.create_user(user) + assert error.value.code == "organization_required" + + +@integration +def test_the_health_probe_reports_reachability(postgres: PostgresAccountRepository) -> None: + assert postgres.check() is True + + +@integration +def test_the_database_stores_no_plaintext_password(postgres: PostgresAccountRepository) -> None: + """A stored credential holds an Argon2id hash and nothing else.""" + + from umbratrans_web.security import utc_now + + rows = [] + with postgres._pool.connection() as connection: # noqa: SLF001 - test introspection + rows = connection.execute( + "SELECT password_hash FROM public.user_credentials WHERE password_hash IS NOT NULL LIMIT 5" + ).fetchall() + for row in rows: + assert row["password_hash"].startswith("$argon2id$") + assert utc_now() is not None diff --git a/umbratrans_web/tests/test_repository.py b/umbratrans_web/tests/test_repository.py index c189988..19fae45 100644 --- a/umbratrans_web/tests/test_repository.py +++ b/umbratrans_web/tests/test_repository.py @@ -19,7 +19,7 @@ from umbratrans_web.models import AccountStatus from umbratrans_web.repository import AccountRepository, InMemoryAccountRepository, build_repository from umbratrans_web.service import AccountService -from .conftest import ensure_organization, provision_local, random_password +from .conftest import TEST_DSN, ensure_organization, provision_local, random_password # ---------------------------------------------------------------------- # In-memory repository @@ -34,11 +34,15 @@ def test_build_repository_selects_the_requested_kind() -> None: assert isinstance(build_repository(Settings(environment=Environment.TEST)), InMemoryAccountRepository) -def test_build_repository_refuses_the_absent_postgres_adapter() -> None: +def test_build_repository_refuses_postgres_without_a_dsn() -> None: + """No connection string must stop the build, not silently use a dev store.""" + + # Constructed in test mode, which skips the production DSN gate, so the + # repository boundary is the one refusing. settings = Settings(environment=Environment.TEST, repository_kind="postgres") with pytest.raises(RepositoryError) as error: build_repository(settings) - assert error.value.code == "postgres_adapter_unavailable" + assert error.value.code == "database_dsn_missing" def test_duplicate_user_ids_and_usernames_are_rejected( @@ -171,6 +175,7 @@ def test_json_repository_refuses_production_mode(tmp_path: Path) -> None: impostor = SimpleNamespace( is_production=True, environment=Environment.PRODUCTION, + database_dsn=TEST_DSN, repository_path=tmp_path / "dev-accounts.json", ) with pytest.raises(ConfigurationError) as error: @@ -181,6 +186,7 @@ def test_json_repository_refuses_production_mode(tmp_path: Path) -> None: with pytest.raises(ConfigurationError) as settings_error: Settings( environment=Environment.PRODUCTION, + database_dsn=TEST_DSN, repository_kind="json", cookie_secure=True, repository_path=tmp_path / "dev-accounts.json", diff --git a/umbratrans_web/tests/test_service.py b/umbratrans_web/tests/test_service.py index 1fab0cf..284723a 100644 --- a/umbratrans_web/tests/test_service.py +++ b/umbratrans_web/tests/test_service.py @@ -27,6 +27,7 @@ from .conftest import ( member_principal, provision_local, provision_oidc, + random_email, random_password, ) @@ -101,6 +102,7 @@ def test_generic_create_account_path_requires_the_operator_context(service: Acco enrollment_mode="temporary_password", username="No.Path", display_name="No Path", + email=random_email("no-path"), password=password, actor=actor, ) @@ -113,6 +115,7 @@ def test_admin_enrollment_generates_a_one_time_password(service: AccountService) issued = service.provision_admin_temporary_password( username="Temp.User", display_name="Temp User", + email=random_email("temp-user"), actor=actor, ) @@ -147,6 +150,7 @@ def test_admin_enrollment_requires_an_administrator(service: AccountService) -> service.provision_admin_temporary_password( username="Escalate.User", display_name="Escalate", + email=random_email("escalate-user"), actor=member_principal(member), ) assert service.repository.get_user_by_username("escalate.user") is None @@ -158,11 +162,47 @@ def test_temporary_password_enrollment_requires_a_password(service: AccountServi enrollment_mode="temporary_password", username="No.Password", display_name="No Password", + email=random_email("no-password"), operator=OperatorContext(source="operator-cli"), ) assert error.value.code == "password_required" +def test_account_creation_requires_an_email(service: AccountService) -> None: + """users.email is NOT NULL in the control plane, so the rule is in the domain. + + Enforcing it here rather than in the store means a missing address fails the + same way against the development repositories and the authoritative store. + """ + + actor = admin_principal(provision_local(service, username="Umbraadmin")) + with pytest.raises(InputError) as error: + service.create_oidc_account( + username="No.Email", + display_name="No Email", + email=None, # type: ignore[arg-type] + actor=actor, + ) + assert error.value.code == "email_required" + + with pytest.raises(InputError) as operator_error: + service.create_account( + enrollment_mode="temporary_password", + username="Also.No.Email", + display_name="Also No Email", + email=None, # type: ignore[arg-type] + password=random_password(28), + operator=OperatorContext(source="operator-cli"), + ) + assert operator_error.value.code == "email_required" + + +def test_a_malformed_email_is_rejected(service: AccountService) -> None: + with pytest.raises(InputError) as error: + provision_local(service, username="Bad.Email", email="not-an-address") + assert error.value.code == "invalid_email" + + def test_oidc_enrollment_creates_no_password_material(repository: object, service: AccountService) -> None: actor = admin_principal(provision_local(service, username="Umbraadmin")) account = provision_oidc(service, actor, username="Oidc.User") @@ -501,6 +541,7 @@ def test_organization_admin_cannot_promote_to_platform_admin(service: AccountSer service.create_oidc_account( username="Escalated.User", display_name="Escalated", + email=random_email("escalated-user"), role=str(RoleName.PLATFORM_ADMIN), organization_id=org, actor=actor, @@ -515,7 +556,12 @@ def test_password_change_principal_cannot_use_admin_operations(service: AccountS with pytest.raises(PasswordChangeRequiredError): service.list_accounts(principal) with pytest.raises(PasswordChangeRequiredError): - service.create_oidc_account(username="Blocked.User", display_name="Blocked", actor=principal) + service.create_oidc_account( + username="Blocked.User", + display_name="Blocked", + email=random_email("blocked-user"), + actor=principal, + ) def test_audit_events_are_sanitized_and_scoped(service: AccountService) -> None: diff --git a/uv.lock b/uv.lock index 66854aa..22d9d21 100644 --- a/uv.lock +++ b/uv.lock @@ -390,6 +390,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/98/33/e2a5b36edf8aa422f6fa4b894756eb33dc93b36df5f65121280bb8b929c4/psycopg_binary-3.3.6-cp315-cp315-win_amd64.whl", hash = "sha256:2f122603f36050937982abf9668d8bc4769a79f7c93a65013b1c49f1cab7b56b", size = 3756154, upload-time = "2026-09-18T13:22:51.283Z" }, ] +[[package]] +name = "psycopg-pool" +version = "3.3.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/74/5e/c0664b968b102ff68b811d999c728546c48d5c1eec03e3bbaf88c0cb4472/psycopg_pool-3.3.3.tar.gz", hash = "sha256:df87b5d9d0ad7db37f6cdad4fa8ce113d250f5997f6db38e9a99192fb67f9e1d", size = 32006, upload-time = "2026-09-22T15:53:24.947Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5d/b4/452c6607a0f479465cd8a9b0d9956919fcb150050c1f83f9f11e6b8ee8dc/psycopg_pool-3.3.3-py3-none-any.whl", hash = "sha256:9b9cd6a4fcec47a410f7e82d408540e7f77b478509e91b44c1a5457a13e5ff37", size = 40304, upload-time = "2026-09-22T15:53:23.712Z" }, +] + [[package]] name = "pycparser" version = "3.0" @@ -617,6 +629,7 @@ dependencies = [ { name = "argon2-cffi" }, { name = "fastapi" }, { name = "psycopg", extra = ["binary"] }, + { name = "psycopg-pool" }, { name = "uvicorn" }, ] @@ -632,6 +645,7 @@ requires-dist = [ { name = "argon2-cffi", specifier = ">=23.1,<26" }, { name = "fastapi", specifier = ">=0.115,<1" }, { name = "psycopg", extras = ["binary"], specifier = ">=3.2,<4" }, + { name = "psycopg-pool", specifier = ">=3.2,<4" }, { name = "uvicorn", specifier = ">=0.34,<1" }, ]