umbragit/UmbraTrans
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Run the PostgreSQL integration suite against the real schema

Browse Source 
The integration suite had never been executed. Running it against a
PostgreSQL 16 instance carrying an exact copy of the control-plane
schema found seven real defects, all of which were invisible without a
database.

Migration:

- `ADD CONSTRAINT ... UNIQUE USING INDEX` failed when the index was
  already attached, for both `organizations_slug_key` and
  `users_username_key`. Because the file runs in one transaction, that
  error rolled back every other change, leaving a half-applied schema
  rather than an obviously untouched one. Both are now guarded, so the
  file applies cleanly to the real schema and is idempotent on re-apply.

The adapter:

- `get_user_by_email` was never implemented. The protocol stub returned
  `None`, so email login found no account against a real database while
  working perfectly against the in-memory one.
- The four verification-token methods were never implemented either, so
  email verification links never worked against PostgreSQL.
- The RLS identity GUC was wrong. `app.current_user_id()` reads
  `app.user_id`, not `app.current_user_id`; setting the similarly named
  GUC left every policy helper resolving to NULL, so every policy denied
  and every query returned zero rows. Added `verify_rls_contract()`,
  called at startup, which proves the two agree. It has teeth: pointed at
  a wrong GUC it refuses.
- `SET` takes no bind parameters, so the statement timeout was a syntax
  error on every transaction. The identity GUC now uses `set_config`,
  which is parameterisable and keeps the value bound.
- `append_audit_event` returned only `id` and then mapped the whole row,
  and passed a text `target_id` where a `UUID` was required.

The domain:

- A member with no organization is storable in the development JSON
  store and rejected by `user_roles_check`, so the same call succeeded in
  development and failed in production. Enforced in the service now, and
  43 tests were relying on the JSON store's leniency.
- `create_account` accepted `require_email_verification` and silently
  dropped it on the operator path, returning an unverified account to a
  caller that had asked for a verified one. Now refused, with the reason.
- All 26 protocol stubs returned `None` instead of raising, so a missing
  method was indistinguishable from an empty database. They now raise, and
  a conformance test checks each repository implements every declared
  method rather than merely inheriting it. The old conformance test used
  `hasattr`, which inheritance always satisfies; it passed while five
  methods were missing.
- The bootstrap CLI refused an unsatisfiable role only after the operator
  had typed a one-time password.

The migration README now records which review questions are answered and
which remain, including that RLS is enabled but not forced, so the
adapter's own scoping is currently the only tenant control and the
isolation tests below it are still outstanding.

255 pass without a database, 263 against one.
This commit is contained in:
umbra audio
2026-09-26 21:55:00 -07:00
parent 0461dff9d2
commit 64af39eb68
12 changed files with 1033 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
umbratrans_web/cli.py
View File

@@ -166,10 +166,19 @@ def bootstrap_admin(
)
return EXIT_REFUSED_PRODUCTION
# Checked before the password prompt, not after. Every role except
# platform_admin is organization-scoped, so such a request cannot be
# satisfied. Reporting it afterwards would mean the operator types a
# one-time password and only then learns it was never going to be used.
refusal = _unsatisfiable_assignment(args)
if refusal is not None:
print(f"Refusing to run: {refusal}", file=err)
return EXIT_ERROR
try:
_require_interactive(stream)
except OperatorRefusalError as refusal:
print(f"Refusing to run: {refusal.message}", file=err)
except OperatorRefusalError as refusal_error:
print(f"Refusing to run: {refusal_error.message}", file=err)
return EXIT_NOT_INTERACTIVE
password = _read_confirmed_password(prompt, err=err)
@@ -246,6 +255,23 @@ def _read_confirmed_password(prompt: Callable[[str], str], *, err: TextIO) -> st
return first
def _unsatisfiable_assignment(args: argparse.Namespace) -> str | None:
"""Return why the requested role and organization cannot both be honoured, if so.
``user_roles_check`` in the control plane makes an organization-scoped
assignment without an organization impossible, and every role other than
``platform_admin`` is organization-scoped. Naming the mismatch here keeps the
reason in front of the operator instead of buried in a domain error.
"""
if args.role == str(RoleName.PLATFORM_ADMIN) or args.organization_id:
return None
return (
f"role {args.role!r} is scoped to an organization, so --organization-id is required. "
"Only a platform administrator can be created without one."
)
def _provision(settings: Settings, *, args: argparse.Namespace, password: str) -> AccountView:
"""Create the account in the local development repository."""

57
umbratrans_web/migrations/0001_phase1_identity.sql
View File

@@ -162,9 +162,31 @@ COMMENT ON COLUMN public.users.username IS
CREATE UNIQUE INDEX IF NOT EXISTS users_username_key
ON public.users (username);
ALTER TABLE public.users
ADD CONSTRAINT users_username_key
UNIQUE USING INDEX users_username_key;
-- Attaching the index is guarded, not unconditional, so the file can be
-- applied more than once. An index that is already attached to a constraint
-- cannot be attached again, and because this migration runs in one transaction
-- the resulting error would roll back every other change in the file, leaving a
-- half-applied schema rather than an obviously untouched one.
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1
FROM pg_constraint c
JOIN pg_class t ON t.oid = c.conrelid
JOIN pg_namespace n ON n.oid = t.relnamespace
WHERE n.nspname = 'public'
AND t.relname = 'users'
AND c.conname = 'users_username_key'
) THEN
ALTER TABLE public.users
ADD CONSTRAINT users_username_key
UNIQUE USING INDEX users_username_key;
RAISE NOTICE 'attached users_username_key to users';
ELSE
RAISE NOTICE 'users_username_key already present; leaving it alone';
END IF;
END
$$;
-- ---------------------------------------------------------------------
-- 2. user_credentials
@@ -311,9 +333,32 @@ CREATE TABLE IF NOT EXISTS public.organizations (
CREATE UNIQUE INDEX IF NOT EXISTS organizations_slug_key
ON public.organizations (slug);
ALTER TABLE public.organizations
ADD CONSTRAINT organizations_slug_key
UNIQUE USING INDEX organizations_slug_key;
-- Attaching the index is guarded, not unconditional. When the control plane
-- already created organizations it also already created this constraint, and
-- ADD CONSTRAINT against an index that is already attached fails with "index is
-- already associated with a constraint". Found by running this migration against
-- the real schema: unguarded, that error aborts the whole transaction, so the
-- rest of the migration would silently not apply.
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1
FROM pg_constraint c
JOIN pg_class t ON t.oid = c.conrelid
JOIN pg_namespace n ON n.oid = t.relnamespace
WHERE n.nspname = 'public'
AND t.relname = 'organizations'
AND c.conname = 'organizations_slug_key'
) THEN
ALTER TABLE public.organizations
ADD CONSTRAINT organizations_slug_key
UNIQUE USING INDEX organizations_slug_key;
RAISE NOTICE 'attached organizations_slug_key to organizations';
ELSE
RAISE NOTICE 'organizations_slug_key already present; leaving it alone';
END IF;
END
$$;
COMMENT ON TABLE public.organizations IS
'Tenants that own projects and accounts. Phase 1 inserts no rows; the control plane may already contain some.';

139
umbratrans_web/migrations/README.md
View File

@@ -33,48 +33,101 @@ Exactly four things, matching `docs/umbratrans-backend-design.md`:
It creates no credential rows, backfills no username, inserts no organization,
and changes no project, role, permission, invitation, or audit table.
## Review questions the file cannot answer itself
## Review questions, and which are now answered
These are BLOCKER-level unknowns. This workspace has no access to the
production schema, so the file states its assumptions instead of guessing.
The questions below were BLOCKER-level unknowns while this workspace had no
access to a real schema. They have since been checked by applying this file, and
the `PostgresAccountRepository` adapter, to a PostgreSQL 16 instance carrying an
exact copy of the control-plane schema. **Answered** items below are verified
facts now. **Open** items still need a human with production access.
- **Membership model divergence (the important one).** The control plane
derives organization membership from `user_roles` with
`scope = 'organization'`, and supports a project-scoped membership the
application slice does not model. The application currently stores
membership as a column on the user record. One of these has to change
before the PostgreSQL adapter is written:
- keep the column and accept that a user has at most one organization and no
project membership, or
- move the service onto `user_roles`, which means reading and writing
role assignments and reconciling `RoleName` with the `roles` table.
### Answered by running against the real schema
This migration deliberately does not decide it. It creates the
`organizations` table and adds no membership column, so the choice stays
open and reviewable.
- Does `users` have a `uuid` primary key named `id`, and no existing column
named `username`?
- Does `organizations` already exist in the control plane, and if so does its
`status` check constraint already allow `archived`? The rollback keys on
that constraint, and a mismatch would make the guarded drop fall through to
the `RAISE NOTICE` branch and leave the table in place.
- What is the name of the "current user" RLS helper (`app.current_user_id()`,
a request-scoped GUC, or something else), and can a client influence it?
- Are `users` and `sessions` already under RLS, and with `FORCE`?
- 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:
- **`users` has a `uuid` primary key named `id` and no `username` column.**
Confirmed. The column and its unique constraint are created by this file.
- **`organizations` already exists in the control plane**, with
`UNIQUE (slug)` already attached as `organizations_slug_key`, and a
`status` check constraint allowing `active`, `suspended`, and `archived`.
This is why sections 4 uses `CREATE TABLE IF NOT EXISTS` and guards the
constraint attachment: an unguarded `ADD CONSTRAINT ... UNIQUE USING INDEX`
fails on an index that is already attached, and because the file runs in one
transaction that failure rolls back every other change. The same guard is now
on `users_username_key` so the file can be applied twice.
- **The "current user" RLS helper is `app.current_user_id()`, and it reads the
GUC `app.user_id`** — not `app.current_user_id`. Every policy helper resolves
the caller through it (`app.is_platform_admin`, `app.can_view_user`,
`app.is_organization_member`, `app.is_organization_admin`,
`app.is_organization_admin_for_project`), so this single setting is the whole
identity contract. The adapter sets it with
`set_config('app.user_id', $1, true)`, and
`PostgresAccountRepository.verify_rls_contract()` runs at startup to prove the
two agree. It is not a formality: the adapter originally set a similarly named
GUC, which left every helper resolving to `NULL`, every policy denying, and
every query returning zero rows — a silent total lockout that presents as an
empty database. `SET` itself takes no bind parameters at all, so `set_config`
is also what makes the value parameterisable.
- **RLS is enabled but not forced** on `audit_events`, `bootstrap_tokens`,
`invitations`, `mfa_methods`, `organizations`, `permissions`, `projects`,
`role_permissions`, `roles`, `sessions`, `user_roles`, and `users`
(`relrowsecurity = true`, `relforcerowsecurity = false`). A table owner
therefore bypasses the policies, and the adapter's own `WHERE` clauses are what
currently enforce tenant scope. The policies are defence in depth for a
non-owner application role, not the primary control. That is a **decision to
confirm**: if the application is ever to run as a non-owner role, the policies
become load-bearing and the isolation tests below must pass first.
- **A generic `set_updated_at()` trigger helper exists** (`app.set_updated_at`),
so `updated_at` is maintained by the database on the tables that use it.
- **The `roles` rows the application needs are present with the right scopes**:
`platform_admin` at `platform`, and `organization_admin` and `member` at
`organization`. This matters because `app.validate_user_role_scope()` runs
`BEFORE INSERT OR UPDATE` on `user_roles` and enforces the pairing, so a
missing or wrongly scoped role fails account creation. Verified, but re-check
before provisioning rather than trusting the copy:
```sql
SELECT name, scope FROM public.roles
WHERE name IN ('platform_admin', 'organization_admin', 'member')
ORDER BY name;
```
- **Every non-platform role is organization-scoped**, and
`user_roles_check` enforces it:
`scope = 'organization'` requires `organization_id IS NOT NULL`. So a member
with no organization is *not representable* in the authoritative schema. The
service now refuses that combination itself, rather than letting it succeed
against the development JSON store and fail in production.
- **`bootstrap_tokens` is shaped for verification tokens** — `(id, user_id,
token_hash, expires_at, consumed_at, created_at)` with a unique
`token_hash` — and is unused, which is why email validation needs no
migration. It has **no `purpose` column**, so the adapter reports every row as
an email-verification token. Phase 1 issues no other kind; a second purpose
would need a column added with it.
- **`citext` is available**, so `users.email` is case-insensitive in the
database and the adapter does not need to normalise the login key itself.
### Still open
- **Membership model divergence (the important one).** The control plane
derives organization membership from `user_roles` with
`scope = 'organization'`, and supports a project-scoped membership the
application slice does not model. The application stores membership as a
column on the user record. One of these has to change:
- keep the column and accept that a user has at most one organization and no
project membership, or
- move the service onto `user_roles`, which means reading and writing role
assignments and reconciling `RoleName` with the `roles` table.
The adapter resolves the role and organization from `user_roles` at read time
and writes the assignment on create, so it works against the real schema
today, but the divergence is unresolved and it is why the adapter **refuses**
a user holding more than one active assignment rather than picking one. The
organization on that assignment is the scope of every administrative decision,
so guessing would hand out the wrong tenant's authority. Project membership
remains unmodelled.
- **Can a client influence `app.user_id`?** The policies read it as an ordinary
session GUC. Whether an unprivileged client can `SET` it directly depends on
the role the application connects as, which is still unknown. If it can, the
policies are not a security boundary at all and the adapter's own scoping is
the only control. This needs an escalation test against the real
application role.
- **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
@@ -90,14 +143,15 @@ production schema, so the file states its assumptions instead of guessing.
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`?
- **What are the migration role and application role names?** The verification
run used a role owning the scratch database, which bypasses RLS. A production
application role should be a non-owner so the policies apply.
- Is `sessions` large enough that `CREATE INDEX` must be replaced with
`CREATE INDEX CONCURRENTLY` outside the transaction?
- Which existing `sessions` column should backfill `last_seen_at`, if any?
- Does the production `organizations` table match the copy used for
verification, in particular the `status` check constraint the rollback keys
on?
## Gates that must pass before approval
@@ -105,8 +159,13 @@ production schema, so the file states its assumptions instead of guessing.
- Migration role and application role privileges verified.
- RLS tests covering two organizations and two projects, including
`FORCE ROW LEVEL SECURITY` and a client-controlled-GUC escalation attempt.
**Not yet done.** The verification run used a table owner, which bypasses RLS,
so tenant isolation has been exercised only through the adapter's own `WHERE`
clauses. This is the most important remaining gap.
- Account-management tests passing against an isolated PostgreSQL database
with the PostgreSQL adapter implemented.
with the PostgreSQL adapter implemented. **Done**: the file applies cleanly to
an exact copy of the control-plane schema and is idempotent on re-apply, and
the suite passes against it (263 tests) via `UMBRATRANS_TEST_DSN`.
- Rollback rehearsed on the scratch database.
Until then, production PostgreSQL stays untouched and the application runs

226
umbratrans_web/postgres_repository.py
View File

@@ -46,14 +46,15 @@ import logging
from contextlib import contextmanager
from datetime import datetime
from typing import Any, Iterator, Mapping, Sequence
from uuid import UUID
from uuid import UUID, uuid4
import psycopg
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from .errors import ConflictError, NotFoundError, RepositoryError
from .errors import ConflictError, InputError, NotFoundError, RepositoryError
from .models import (
VERIFICATION_PURPOSE_EMAIL,
AccountStatus,
AuditEvent,
CredentialRecord,
@@ -63,6 +64,8 @@ from .models import (
RoleName,
SessionRecord,
User,
VerificationToken,
normalize_email,
username_key,
)
from .repository import AccountRepository
@@ -75,8 +78,20 @@ LOGGER = logging.getLogger("umbratrans_web.postgres")
#: 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"
#: The GUC that the row-level-security policy helpers read.
#:
#: This name is not a guess and it is not cosmetic. Every policy helper in the
#: control plane resolves the caller through ``app.current_user_id()``, and that
#: function is defined as ``current_setting('app.user_id', true)``. Setting a
#: similarly named GUC instead leaves every helper resolving to NULL, which makes
#: every policy return false and every query return zero rows: a silent total
#: lockout that looks like an empty database rather than an error.
#: :meth:`PostgresAccountRepository.verify_rls_contract` proves the two agree at
#: startup, for exactly that reason.
RLS_USER_SETTING = "app.user_id"
#: The function the policies call to resolve the caller.
RLS_USER_FUNCTION = "app.current_user_id"
#: Statement timeout applied to every transaction, so one slow query cannot pin
#: a pooled connection indefinitely.
@@ -147,6 +162,34 @@ class PostgresAccountRepository(AccountRepository):
LOGGER.warning("database health check failed", exc_info=True)
return False
def verify_rls_contract(self) -> None:
"""Prove the adapter and the policy helpers agree on the caller's identity.
The policies resolve the caller by calling ``app.current_user_id()``.
The adapter publishes that identity as a session GUC. Nothing in the SQL
connects the two, so a mismatch is possible and the failure mode is
silent: every policy evaluates false, every query returns zero rows, and
the service reports an empty database rather than an error.
So this runs a real probe: set the GUC, ask the policy function what it
sees, and refuse to continue unless the two agree. Called at startup,
because discovering this in production as a total lockout is far worse
than refusing to start.
"""
probe = uuid4()
with self._connection() as connection:
connection.execute("SELECT set_config(%s, %s, true)", (RLS_USER_SETTING, str(probe)))
row = connection.execute(f"SELECT {RLS_USER_FUNCTION}() AS resolved").fetchone()
resolved = row["resolved"] if row is not None else None
if resolved != probe:
raise RepositoryError(
"rls_identity_mismatch",
"The row-level-security helper does not read the session setting this "
"adapter publishes, so every policy would deny access. Refusing to start.",
)
LOGGER.info("row-level-security identity contract verified setting=%s", RLS_USER_SETTING)
# -- transaction plumbing ----------------------------------------------
@contextmanager
@@ -158,12 +201,27 @@ class PostgresAccountRepository(AccountRepository):
request's identity.
"""
# ``SET`` is a utility statement and takes no bind parameters at all -
# psycopg renders them as ``$1`` and the grammar rejects that. The first
# statement therefore carries a literal. It is coerced to ``int`` here so
# the interpolation cannot carry SQL, and the value is a module constant
# anyway. Verified against a real server, not assumed.
timeout_ms = int(STATEMENT_TIMEOUT_MS)
with self._pool.connection() as connection:
with connection.transaction():
connection.execute("SET LOCAL statement_timeout = %s", (STATEMENT_TIMEOUT_MS,))
connection.execute(f"SET LOCAL statement_timeout = {timeout_ms}")
connection.execute("SET LOCAL idle_in_transaction_session_timeout = '10s'")
if user_id is not None:
connection.execute(f"SET LOCAL {RLS_USER_SETTING} = %s", (str(user_id),))
# ``set_config`` is the parameterisable form and the right way
# to set a namespaced GUC, so the identity stays a bound value.
# The third argument is ``true`` for local scope, which is what
# makes pooling safe: the identity is discarded when the
# transaction ends, so a returned connection cannot carry the
# previous request's caller.
connection.execute(
"SELECT set_config(%s, %s, true)",
(RLS_USER_SETTING, str(user_id)),
)
yield connection
@contextmanager
@@ -237,6 +295,25 @@ class PostgresAccountRepository(AccountRepository):
def get_user(self, user_id: UUID) -> User | None:
return self._fetch_user("u.id = %s", (user_id,))
def get_user_by_email(self, key: str) -> User | None:
"""Look up an account by its email address. This is the login path.
The column comparison is left to the database. ``users.email`` is
``citext``, so equality is already case-insensitive there, which means
case-insensitivity does not depend on this process normalising
correctly. The value is still normalised, because a malformed address
cannot match any row and rejecting it here avoids a pointless query.
Validation failure is "no such account", not an error: the login form
must not tell an attacker which addresses exist.
"""
try:
candidate = normalize_email(key)
except InputError:
return None
return self._fetch_user("u.email = %s", (candidate,))
def get_user_by_username(self, key: str) -> User | None:
try:
normalized = username_key(key)
@@ -721,6 +798,96 @@ class PostgresAccountRepository(AccountRepository):
).fetchone()
return _session_from_row(row) if row else None
# -- verification tokens ------------------------------------------------
def create_verification_token(self, token: VerificationToken) -> VerificationToken:
"""Store a single-use token in ``bootstrap_tokens``.
That table predates this application and was unused, which is why email
validation needs no migration. It has no ``purpose`` column, so every row
in it is treated as an email-verification token. Phase 1 issues no other
kind, and if a second purpose is ever added the column has to be added
with it: without a purpose in the row, a token issued for one purpose
would be honoured for another. :func:`_verification_from_row` therefore
always reports the email-verification purpose rather than reading one.
"""
with self._connection() as connection:
connection.execute(
"""
INSERT INTO public.bootstrap_tokens (id, user_id, token_hash, expires_at, consumed_at, created_at)
VALUES (%s, %s, %s, %s, %s, %s)
ON CONFLICT (id) DO NOTHING
""",
(
token.id,
token.user_id,
token.token_hash,
token.expires_at,
token.consumed_at,
token.created_at,
),
)
return token
def get_live_verification_token(self, token_hash: str) -> VerificationToken | None:
"""Return an unconsumed, unexpired token matching this hash.
Expiry is filtered in SQL rather than in Python so that a token that
expired while the process was down is still refused, and so the decision
is made by the same clock the row was written with.
"""
with self._connection() as connection:
row = connection.execute(
"""
SELECT id, user_id, token_hash, expires_at, consumed_at, created_at
FROM public.bootstrap_tokens
WHERE token_hash = %s
AND consumed_at IS NULL
AND expires_at > now()
""",
(token_hash,),
).fetchone()
return _verification_from_row(row) if row else None
def mark_verification_token_consumed(self, token_id: UUID, *, now: datetime) -> bool:
"""Consume a token, returning whether this call was the one that did it.
The update carries ``consumed_at IS NULL`` in its predicate, so the
database decides the winner. Two simultaneous requests cannot both
observe success, which is what makes the link single-use; deciding this in
Python after a read would not.
"""
with self._connection() as connection:
row = connection.execute(
"""
UPDATE public.bootstrap_tokens
SET consumed_at = %s
WHERE id = %s
AND consumed_at IS NULL
RETURNING id
""",
(now, token_id),
).fetchone()
return row is not None
def delete_expired_verification_tokens(self, *, now: datetime) -> int:
"""Delete tokens that are spent or long expired, returning how many."""
with self._connection() as connection:
row = connection.execute(
"""
DELETE FROM public.bootstrap_tokens
WHERE consumed_at IS NOT NULL
OR expires_at <= %s
RETURNING id
""",
(now,),
).fetchall()
return len(row)
# -- audit --------------------------------------------------------------
def append_audit_event(self, event: AuditEvent) -> AuditEvent:
@@ -734,13 +901,19 @@ class PostgresAccountRepository(AccountRepository):
with self._connection() as connection:
row = connection.execute(
# The whole row comes back, not just the identifier: the store
# defaults and normalises these columns, so the returned row is
# what was actually written and is what a later read would see.
# Returning only ``id`` left the mapper reading keys the row did
# not have.
"""
INSERT INTO public.audit_events (
actor_user_id, organization_id, action, target_type, target_id,
request_id, metadata, created_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s::jsonb, %s)
RETURNING id
RETURNING id, actor_user_id, organization_id, action, target_type,
target_id, request_id, metadata, created_at
""",
(
event.actor_user_id,
@@ -964,19 +1137,49 @@ def _audit_from_row(row: Mapping[str, Any], original: AuditEvent | None) -> Audi
metadata = json.loads(metadata)
metadata = dict(metadata)
detail = {key: str(value) for key, value in metadata.items() if key != "application_event_id"}
# ``target_id`` is a text column that also holds non-user targets, so it is
# narrowed on ``target_type`` and then parsed. A ``user`` row whose text is not
# a UUID is a store inconsistency; it becomes no target rather than an
# exception raised while mapping an audit record, which would fail the write
# that was already committed.
target_user_id: UUID | None = None
if row["target_type"] == "user" and row["target_id"]:
try:
target_user_id = UUID(str(row["target_id"]))
except ValueError:
target_user_id = None
return AuditEvent(
id=str(row["id"]),
occurred_at=row["created_at"],
action=row["action"],
outcome=(original.outcome if original is not None else str(metadata.get("outcome", "success"))),
actor_user_id=row["actor_user_id"],
target_user_id=(row["target_id"] if row["target_type"] == "user" else None),
target_user_id=target_user_id,
organization_id=row["organization_id"],
request_id=row["request_id"],
detail=detail,
)
def _verification_from_row(row: Mapping[str, Any]) -> VerificationToken:
"""Map a ``bootstrap_tokens`` row.
``purpose`` is fixed rather than read: the table has no such column, and
phase 1 issues only email-verification tokens. See
:meth:`PostgresAccountRepository.create_verification_token`.
"""
return VerificationToken(
id=row["id"],
user_id=row["user_id"],
token_hash=row["token_hash"],
expires_at=row["expires_at"],
created_at=row["created_at"],
consumed_at=row["consumed_at"],
purpose=VERIFICATION_PURPOSE_EMAIL,
)
def _unique_conflict(error: Exception, username: str) -> ConflictError:
"""Map a unique violation onto the right domain conflict.
@@ -992,4 +1195,9 @@ def _unique_conflict(error: Exception, username: str) -> ConflictError:
return ConflictError("username_taken", "That username is already in use.")
__all__ = ["PostgresAccountRepository", "RLS_USER_SETTING", "STATEMENT_TIMEOUT_MS"]
__all__ = [
"PostgresAccountRepository",
"RLS_USER_FUNCTION",
"RLS_USER_SETTING",
"STATEMENT_TIMEOUT_MS",
]

73
umbratrans_web/repository.py
View File

@@ -39,10 +39,15 @@ class AccountRepository(Protocol):
def create_user(self, user: User) -> User:
"""Insert a new account, or raise :class:`ConflictError` on duplicates."""
raise NotImplementedError(f"{type(self).__name__} does not implement create_user().")
def save_user(self, user: User) -> User:
"""Persist a modified account, or raise :class:`NotFoundError`."""
def get_user(self, user_id: UUID) -> User | None: ...
raise NotImplementedError(f"{type(self).__name__} does not implement save_user().")
def get_user(self, user_id: UUID) -> User | None:
raise NotImplementedError(f"{type(self).__name__} does not implement get_user().")
def get_user_by_email(self, key: str) -> User | None:
"""Look up an account by its case-insensitive email key.
@@ -52,12 +57,18 @@ class AccountRepository(Protocol):
unique identifier and no separate username lookup is used to sign in.
"""
raise NotImplementedError(f"{type(self).__name__} does not implement get_user_by_email().")
def get_user_by_username(self, key: str) -> User | None:
"""Look up an account by its optional display handle. Not a login path."""
raise NotImplementedError(f"{type(self).__name__} does not implement get_user_by_username().")
def count_users(self, organization_id: UUID | None = None) -> int:
"""Count accounts, optionally within one organization."""
raise NotImplementedError(f"{type(self).__name__} does not implement count_users().")
def find_organizations(
self,
*,
@@ -68,14 +79,21 @@ class AccountRepository(Protocol):
) -> tuple[list[Organization], int]:
"""Return a page of organizations and the total match count."""
def get_organization(self, organization_id: UUID) -> Organization | None: ...
raise NotImplementedError(f"{type(self).__name__} does not implement find_organizations().")
def get_organization(self, organization_id: UUID) -> Organization | None:
raise NotImplementedError(f"{type(self).__name__} does not implement get_organization().")
def create_organization(self, organization: Organization) -> Organization:
"""Insert a new organization, or raise :class:`ConflictError` on a duplicate slug."""
raise NotImplementedError(f"{type(self).__name__} does not implement create_organization().")
def save_organization(self, organization: Organization) -> Organization:
"""Persist a modified organization, or raise :class:`NotFoundError`."""
raise NotImplementedError(f"{type(self).__name__} does not implement save_organization().")
def find_users(
self,
*,
@@ -87,23 +105,34 @@ class AccountRepository(Protocol):
) -> tuple[list[User], int]:
"""Return a page of accounts and the total match count."""
def get_credential(self, user_id: UUID) -> CredentialRecord | None: ...
raise NotImplementedError(f"{type(self).__name__} does not implement find_users().")
def save_credential(self, credential: CredentialRecord) -> CredentialRecord: ...
def get_credential(self, user_id: UUID) -> CredentialRecord | None:
raise NotImplementedError(f"{type(self).__name__} does not implement get_credential().")
def create_session(self, session: SessionRecord) -> SessionRecord: ...
def save_credential(self, credential: CredentialRecord) -> CredentialRecord:
raise NotImplementedError(f"{type(self).__name__} does not implement save_credential().")
def get_session(self, session_id: UUID) -> SessionRecord | None: ...
def create_session(self, session: SessionRecord) -> SessionRecord:
raise NotImplementedError(f"{type(self).__name__} does not implement create_session().")
def get_session_by_token_hash(self, token_hash: str) -> SessionRecord | None: ...
def get_session(self, session_id: UUID) -> SessionRecord | None:
raise NotImplementedError(f"{type(self).__name__} does not implement get_session().")
def save_session(self, session: SessionRecord) -> SessionRecord: ...
def get_session_by_token_hash(self, token_hash: str) -> SessionRecord | None:
raise NotImplementedError(f"{type(self).__name__} does not implement get_session_by_token_hash().")
def list_sessions(self, user_id: UUID, *, include_revoked: bool = False) -> list[SessionRecord]: ...
def save_session(self, session: SessionRecord) -> SessionRecord:
raise NotImplementedError(f"{type(self).__name__} does not implement save_session().")
def list_sessions(self, user_id: UUID, *, include_revoked: bool = False) -> list[SessionRecord]:
raise NotImplementedError(f"{type(self).__name__} does not implement list_sessions().")
def count_sessions(self, user_id: UUID) -> int:
"""Count currently valid (not revoked, not expired) sessions."""
raise NotImplementedError(f"{type(self).__name__} does not implement count_sessions().")
def revoke_sessions(
self,
user_id: UUID,
@@ -114,9 +143,13 @@ class AccountRepository(Protocol):
) -> int:
"""Revoke live sessions for one account and return how many were revoked."""
raise NotImplementedError(f"{type(self).__name__} does not implement revoke_sessions().")
def create_verification_token(self, token: VerificationToken) -> VerificationToken:
"""Store a single-use token. Only the hash is written."""
raise NotImplementedError(f"{type(self).__name__} does not implement create_verification_token().")
def get_live_verification_token(self, token_hash: str) -> VerificationToken | None:
"""Return an unconsumed, unexpired token for this purpose, if any.
@@ -124,6 +157,8 @@ class AccountRepository(Protocol):
the first use succeeded.
"""
raise NotImplementedError(f"{type(self).__name__} does not implement get_live_verification_token().")
def mark_verification_token_consumed(self, token_id: UUID, *, now: datetime) -> bool:
"""Consume a token. Returns False if it was already consumed.
@@ -131,9 +166,13 @@ class AccountRepository(Protocol):
concurrency: two simultaneous requests cannot both observe success.
"""
def delete_expired_verification_tokens(self, *, now: datetime) -> int: ...
raise NotImplementedError(f"{type(self).__name__} does not implement mark_verification_token_consumed().")
def append_audit_event(self, event: AuditEvent) -> AuditEvent: ...
def delete_expired_verification_tokens(self, *, now: datetime) -> int:
raise NotImplementedError(f"{type(self).__name__} does not implement delete_expired_verification_tokens().")
def append_audit_event(self, event: AuditEvent) -> AuditEvent:
raise NotImplementedError(f"{type(self).__name__} does not implement append_audit_event().")
def list_audit_events(
self,
@@ -143,7 +182,8 @@ class AccountRepository(Protocol):
actor_user_id: UUID | None = None,
limit: int = 50,
offset: int = 0,
) -> tuple[list[AuditEvent], int]: ...
) -> tuple[list[AuditEvent], int]:
raise NotImplementedError(f"{type(self).__name__} does not implement list_audit_events().")
class InMemoryAccountRepository:
@@ -444,11 +484,18 @@ def build_repository(settings: object, *, repository: AccountRepository | None =
"database_dsn_missing",
"The PostgreSQL repository requires a connection string.",
)
return PostgresAccountRepository.from_dsn(
repository = PostgresAccountRepository.from_dsn(
dsn,
min_size=int(getattr(settings, "database_pool_min", 1)),
max_size=int(getattr(settings, "database_pool_max", 10)),
)
# Checked here, at startup, so a mismatch between the setting this adapter
# publishes and the one the policy helpers read is a refusal to start
# rather than a running service in which every query quietly returns zero
# rows. The failure is silent by nature: policies evaluate false, the
# database looks empty, and nothing raises.
repository.verify_rls_contract()
return repository
raise RepositoryError("unknown_repository", "Unknown repository kind.")

56
umbratrans_web/service.py
View File

@@ -226,6 +226,20 @@ class AccountService:
mode = _enrollment_mode(enrollment_mode)
if mode is EnrollmentMode.TEMPORARY_PASSWORD:
if require_email_verification:
# Refused rather than ignored. This path is the operator's, and the
# operator attests to the address: the first account on a new
# deployment has to exist before any mail is configured, so this
# path cannot depend on a delivered message. Honouring the flag
# here would deadlock that bootstrap; ignoring it would hand back
# an unverified account to a caller that asked for a verified one.
# Administrator-driven enrollment is the path that sends
# verification mail: see provision_admin_temporary_password.
raise InputError(
"operator_path_skips_verification",
"The operator enrollment path attests to the email address and does not send a "
"verification message. Use the administrator path to require verification.",
)
return self._provision_local_account(
username=username,
display_name=display_name,
@@ -258,9 +272,14 @@ class AccountService:
password: str | None,
operator: OperatorContext | None,
request_id: str | None,
require_email_verification: bool = False,
) -> AccountView:
"""Operator-only enrollment with a one-time local password."""
"""Operator-only enrollment with a one-time local password.
Always creates an active, already-attested account. There is no
verification flag here on purpose: the caller-facing ``create_account``
refuses ``require_email_verification`` on this path rather than quietly
dropping it, so reaching this method with one is not possible.
"""
if operator is None:
raise EnrollmentRefusedError(
@@ -269,7 +288,7 @@ class AccountService:
)
if not isinstance(password, str) or not password:
raise InputError("password_required", "A one-time password is required for this enrollment mode.")
self._validate_organization_for_assignment(organization_id)
self._validate_organization_for_assignment(organization_id, role=role)
self._passwords.policy.validate(password, username=username)
password_hash = self._passwords.hash_password(password)
user, credential = self._insert_account(
@@ -281,11 +300,8 @@ class AccountService:
credential_type=CredentialType.LOCAL_PASSWORD,
password_hash=password_hash,
must_change_password=True,
initial_status=(AccountStatus.INVITED if require_email_verification else AccountStatus.ACTIVE),
initial_status=AccountStatus.ACTIVE,
)
if require_email_verification:
link = self.issue_email_verification(user.id, request_id=request_id)
self._deliver(user.email, link)
self._audit(
action=ACTION_ACCOUNT_PROVISIONED_LOCAL,
outcome="success",
@@ -336,7 +352,7 @@ class AccountService:
if actor is None: # pragma: no cover - _require_admin raises first
raise ForbiddenError("administrator_required")
self._authorize_creation(actor, role=role, organization_id=organization_id)
self._validate_organization_for_assignment(organization_id)
self._validate_organization_for_assignment(organization_id, role=role)
temporary_password = self._passwords.generate_temporary_password(username=normalize_username(username))
user, credential = self._insert_account(
username=username,
@@ -396,7 +412,7 @@ class AccountService:
if actor is None: # pragma: no cover - _require_admin raises first
raise ForbiddenError("administrator_required")
self._authorize_creation(actor, role=role, organization_id=organization_id)
self._validate_organization_for_assignment(organization_id)
self._validate_organization_for_assignment(organization_id, role=role)
user, credential = self._insert_account(
username=username,
display_name=display_name,
@@ -1351,14 +1367,32 @@ class AccountService:
return True
return organization.status is not OrganizationStatus.ACTIVE
def _validate_organization_for_assignment(self, organization_id: UUID | None) -> None:
"""Reject an account assignment to an organization that cannot hold it.
def _validate_organization_for_assignment(
self,
organization_id: UUID | None,
*,
role: str | None = None,
) -> None:
"""Reject an account assignment that the tenancy model cannot hold.
Without this, any caller-supplied UUID becomes a silent tenant
reference, because nothing else in phase 1 backs
``user.organization_id``.
``role`` is checked here rather than left to the store because the two
repositories must agree. ``user_roles`` requires a non-platform role to
carry an organization, so an organization-less member is storable in the
development JSON store and rejected by PostgreSQL. The same call would
then succeed in development and fail in production, which is the failure
mode this layer exists to prevent, so the rule is enforced here where it
behaves identically against every repository.
"""
if role is not None and role != str(RoleName.PLATFORM_ADMIN) and organization_id is None:
raise InputError(
"organization_required",
"Only a platform administrator can be created without an organization.",
)
if organization_id is None:
return
organization = self._require_organization(organization_id)

81
umbratrans_web/tests/conftest.py
View File

@@ -34,7 +34,11 @@ from umbratrans_web.models import (
)
from umbratrans_web.repository import InMemoryAccountRepository
from umbratrans_web.security import utc_now
from umbratrans_web.service import AccountService, OperatorContext
from umbratrans_web.service import (
AccountService,
IssuedTemporaryPassword,
OperatorContext,
)
ALPHABET = string.ascii_letters + string.digits + "-_.@"
@@ -171,6 +175,18 @@ def mailer() -> CapturingMailer:
return CapturingMailer()
@pytest.fixture
def organization(service: AccountService) -> Organization:
"""An active tenant, for tests that need to name an organization over the API.
Every role except ``platform_admin`` is organization-scoped, so an account
created for a member must say which tenant it belongs to. Tests that go
through HTTP cannot use the provisioning helpers and need this.
"""
return ensure_organization(service)
@pytest.fixture
def service(repository: InMemoryAccountRepository, settings: Settings, mailer: CapturingMailer) -> AccountService:
return AccountService(repository, settings, mailer=mailer)
@@ -225,6 +241,33 @@ def random_email(seed: str) -> str:
return f"{local}.{secrets.token_hex(4)}@example.test"
def _scope_organization(
service: AccountService,
role: str,
organization_id: UUID | None,
) -> UUID | None:
"""Supply the organization a non-platform role needs, if the caller omitted one.
Every role except ``platform_admin`` is organization-scoped, and
``user_roles_check`` in the control plane makes that structural: an
organization-scoped assignment with no organization is rejected by the
database. The service enforces the same rule, so a member created without
one is refused.
These helpers create that organization instead, so a test that only cares
about some other behaviour does not have to restate the tenancy setup. Tests
that assert tenancy pass ``organization_id`` explicitly and still see the
value they asked for.
"""
if organization_id is not None or role == str(RoleName.PLATFORM_ADMIN):
return organization_id
# A fresh slug each time. ``ensure_organization`` is deterministic, so two
# members created in one test would otherwise collide on the unique slug.
unique = uuid4().hex[:8]
return ensure_organization(service, name=f"Test Organization {unique}", slug=f"test-org-{unique}").id
def provision_local(
service: AccountService,
*,
@@ -248,13 +291,43 @@ def provision_local(
display_name=display_name,
email=email if email is not None else random_email(username),
role=role,
organization_id=organization_id,
organization_id=_scope_organization(service, role, organization_id),
password=candidate,
operator=OperatorContext(source="operator-cli"),
)
return SeededAccount(view=view, password=candidate, user_id=view.id, username=view.username, email=view.email)
def provision_invite(
service: AccountService,
*,
actor: Principal,
username: str = "Invited.User",
display_name: str = "Invited User",
role: str = str(RoleName.MEMBER),
organization_id: UUID | None = None,
email: str | None = None,
request_id: str | None = None,
) -> IssuedTemporaryPassword:
"""Enroll an account through the administrator one-time-password path.
``provision_admin_temporary_password`` defaults to ``member``, and a member
must belong to an organization, so this helper supplies one the same way
:func:`provision_local` and :func:`provision_oidc` do. It exists so the
invitation flow can be tested without each test restating tenancy setup.
"""
return service.provision_admin_temporary_password(
username=username,
display_name=display_name,
email=email or random_email(username),
role=role,
organization_id=_scope_organization(service, role, organization_id),
actor=actor,
request_id=request_id,
)
def provision_oidc(
service: AccountService,
actor: Principal,
@@ -272,7 +345,7 @@ def provision_oidc(
display_name=display_name,
email=email or random_email(username),
role=role,
organization_id=organization_id,
organization_id=_scope_organization(service, role, organization_id),
actor=actor,
)
return SeededAccount(view=view, password="", user_id=view.id, username=view.username, email=view.email)
@@ -341,10 +414,12 @@ __all__ = [
"admin_principal",
"client",
"ensure_organization",
"organization",
"mailer",
"verification_token",
"verify_account_email",
"member_principal",
"provision_invite",
"provision_local",
"provision_oidc",
"random_email",

20
umbratrans_web/tests/test_api_uadmin.py
View File

@@ -9,6 +9,7 @@ import pytest
from fastapi.testclient import TestClient
from umbratrans_web.mailer import CapturingMailer
from umbratrans_web.models import Organization
from umbratrans_web.service import AccountService
from .conftest import (
@@ -111,7 +112,9 @@ def test_temporary_password_accounts_cannot_reach_admin_routes(api: ApiHarness,
# ----------------------------------------------------------------------
def test_create_oidc_account(api: ApiHarness, admin: object, service: AccountService) -> None:
def test_create_oidc_account(
api: ApiHarness, admin: object, service: AccountService, organization: Organization
) -> None:
response = api.post(
"/api/uadmin/accounts",
json={
@@ -119,6 +122,7 @@ def test_create_oidc_account(api: ApiHarness, admin: object, service: AccountSer
"display_name": "OIDC User",
"enrollment_mode": "oidc",
"role": "member",
"organization_id": str(organization.id),
"email": "Oidc.User@Example.COM",
},
)
@@ -169,6 +173,7 @@ def test_admin_provisions_a_temporary_password_account(
api: ApiHarness,
admin: object,
service: AccountService,
organization: Organization,
) -> None:
response = api.post(
"/api/uadmin/accounts",
@@ -177,6 +182,7 @@ def test_admin_provisions_a_temporary_password_account(
"display_name": "Temp User",
"enrollment_mode": "temporary_password",
"role": "member",
"organization_id": str(organization.id),
"email": "Temp.User@Example.com",
},
)
@@ -218,6 +224,7 @@ def test_generated_temporary_password_must_be_replaced_on_first_login(
admin: object,
service: AccountService,
mailer: CapturingMailer,
organization: Organization,
) -> None:
email = random_email("first-login")
created = api.post(
@@ -227,6 +234,7 @@ def test_generated_temporary_password_must_be_replaced_on_first_login(
"display_name": "First Login",
"email": email,
"enrollment_mode": "temporary_password",
"organization_id": str(organization.id),
},
)
assert created.status_code == 201
@@ -264,6 +272,7 @@ def test_generated_temporary_password_must_be_replaced_on_first_login(
def test_generated_temporary_passwords_are_unique_per_account(
api: ApiHarness,
admin: object,
organization: Organization,
) -> None:
issued = {
api.post(
@@ -273,6 +282,7 @@ def test_generated_temporary_passwords_are_unique_per_account(
"display_name": f"Bulk User {index}",
"email": random_email(f"bulk-{index}"),
"enrollment_mode": "temporary_password",
"organization_id": str(organization.id),
},
).json()["temporary_password"]
for index in range(5)
@@ -314,7 +324,7 @@ def test_organization_admin_cannot_provision_outside_its_scope(api: ApiHarness,
assert api.get("/api/uadmin/accounts").json()["total"] == 1
def test_create_account_validates_input(api: ApiHarness, admin: object) -> None:
def test_create_account_validates_input(api: ApiHarness, admin: object, organization: Organization) -> None:
duplicate = api.post(
"/api/uadmin/accounts",
json={
@@ -322,6 +332,7 @@ def test_create_account_validates_input(api: ApiHarness, admin: object) -> None:
"display_name": "Duplicate",
"email": random_email("dup"),
"enrollment_mode": "oidc",
"organization_id": str(organization.id),
},
)
invalid = api.post(
@@ -561,7 +572,9 @@ def test_audit_is_scoped_to_the_callers_organization(api: ApiHarness, service: A
assert filtered.status_code == 403
def test_audit_events_reference_the_request_id(api: ApiHarness, admin: object, service: AccountService) -> None:
def test_audit_events_reference_the_request_id(
api: ApiHarness, admin: object, service: AccountService, organization: Organization
) -> None:
response = api.post(
"/api/uadmin/accounts",
json={
@@ -569,6 +582,7 @@ def test_audit_events_reference_the_request_id(api: ApiHarness, admin: object, s
"display_name": "Traced",
"email": random_email("traced"),
"enrollment_mode": "oidc",
"organization_id": str(organization.id),
},
)
assert response.status_code == 201

39
umbratrans_web/tests/test_cli.py
View File

@@ -75,8 +75,14 @@ def _run(
entries: list[str] | None = None,
stdin: TextIO | None = None,
env: dict[str, str] | None = None,
prompt: Callable[[str], str] | None = None,
) -> tuple[int, str, str]:
"""Run the CLI with an injected prompt. Values are generated per run."""
"""Run the CLI with an injected prompt. Values are generated per run.
``prompt`` replaces the default so a test can assert that the password
prompt was never reached, which is the only way to prove a refusal happens
before it.
"""
# 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.
@@ -99,7 +105,7 @@ def _run(
stdin=stdin if stdin is not None else _FakeTty(),
stdout=out, # type: ignore[arg-type]
stderr=err, # type: ignore[arg-type]
prompt=fake_prompt,
prompt=prompt if prompt is not None else fake_prompt,
)
return code, out.text, err.text
@@ -149,13 +155,38 @@ def test_bootstrap_creates_a_must_change_account_in_a_private_file(tmp_path: Pat
def test_bootstrap_uses_a_configurable_non_secret_username(tmp_path: Path) -> None:
path = tmp_path / "dev-accounts.json"
code, out, err = _run(["--repository-path", str(path), "--username", "Ops.Admin", "--role", "organization_admin"])
code, out, err = _run(["--repository-path", str(path), "--username", "Ops.Admin", "--role", "platform_admin"])
assert code == EXIT_OK, err
assert "Ops.Admin" in out
assert "organization_admin" in out
assert "platform_admin" in out
assert path.is_file()
def test_bootstrap_refuses_an_organization_role_with_no_organization(tmp_path: Path) -> None:
"""Refused before the password is read.
Every role except platform_admin is organization-scoped, so this request can
never be satisfied. Saying so up front matters: otherwise the operator types
a one-time password and only then learns it was never going to be used.
"""
prompted: list[str] = []
def _prompt(label: str) -> str:
prompted.append(label)
return ""
path = tmp_path / "dev-accounts.json"
code, out, err = _run(
["--repository-path", str(path), "--role", "organization_admin"],
prompt=_prompt,
)
assert code == EXIT_ERROR
assert "organization" in err.lower()
assert not prompted, "the password prompt must not be reached"
assert not path.exists()
def test_bootstrap_refuses_production_mode(tmp_path: Path) -> None:
code, out, err = _run(
["--repository-path", str(tmp_path / "dev-accounts.json")],

58
umbratrans_web/tests/test_email_verification.py
View File

@@ -26,13 +26,15 @@ from umbratrans_web.errors import (
VerificationFailedError,
)
from umbratrans_web.mailer import CapturingMailer, LoggingMailer, MailerError, build_mailer
from umbratrans_web.models import VERIFICATION_PURPOSE_EMAIL, AccountStatus
from umbratrans_web.models import VERIFICATION_PURPOSE_EMAIL, AccountStatus, RoleName
from umbratrans_web.security import hash_session_token, utc_now
from umbratrans_web.service import AccountService
from umbratrans_web.service import AccountService, OperatorContext
from .conftest import (
ApiHarness,
admin_principal,
ensure_organization,
provision_invite,
provision_local,
random_email,
random_password,
@@ -89,10 +91,10 @@ def test_the_handle_is_optional(service: AccountService) -> None:
username=None,
display_name="No Handle",
email=random_email("no-handle"),
# Platform scope so this test is about the absent handle, not tenancy.
role=str(RoleName.PLATFORM_ADMIN),
password=random_password(28),
operator=__import__("umbratrans_web.service", fromlist=["OperatorContext"]).OperatorContext(
source="operator-cli"
),
operator=OperatorContext(source="operator-cli"),
)
assert account.username is None
# The audit handle falls back to the address rather than becoming blank.
@@ -109,7 +111,8 @@ def test_the_handle_is_optional(service: AccountService) -> None:
def test_an_unverified_account_is_refused_with_a_distinct_error(
service: AccountService, admin_actor: object, mailer: CapturingMailer
) -> None:
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="Pending.User",
display_name="Pending User",
email=random_email("pending"),
@@ -133,7 +136,8 @@ def test_the_unverified_message_comes_only_after_the_password_is_right(
authentication error exists to prevent.
"""
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="Pending.Two",
display_name="Pending Two",
email=random_email("pending-two"),
@@ -153,7 +157,8 @@ def test_the_unverified_message_comes_only_after_the_password_is_right(
def test_an_unverified_account_has_no_session_at_all(
service: AccountService, admin_actor: object, mailer: CapturingMailer
) -> None:
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="No.Session",
display_name="No Session",
email=random_email("no-session"),
@@ -172,7 +177,8 @@ def test_an_unverified_account_has_no_session_at_all(
def test_verification_activates_the_account(
service: AccountService, admin_actor: object, mailer: CapturingMailer
) -> None:
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="Verify.Me",
display_name="Verify Me",
email=random_email("verify-me"),
@@ -186,7 +192,8 @@ def test_verification_activates_the_account(
def test_the_link_is_single_use(service: AccountService, admin_actor: object, mailer: CapturingMailer) -> None:
service.provision_admin_temporary_password(
provision_invite(
service,
username="Single.Use",
display_name="Single Use",
email=random_email("single-use"),
@@ -203,7 +210,8 @@ def test_every_failure_mode_looks_identical(
) -> None:
"""Unknown, already used and expired must be indistinguishable."""
service.provision_admin_temporary_password(
provision_invite(
service,
username="Same.Error",
display_name="Same Error",
email=random_email("same-error"),
@@ -221,7 +229,8 @@ def test_every_failure_mode_looks_identical(
def test_an_expired_link_is_refused(
service: AccountService, admin_actor: object, mailer: CapturingMailer, monkeypatch: pytest.MonkeyPatch
) -> None:
service.provision_admin_temporary_password(
provision_invite(
service,
username="Expired.Link",
display_name="Expired Link",
email=random_email("expired"),
@@ -236,7 +245,8 @@ def test_an_expired_link_is_refused(
def test_only_the_hash_is_stored(service: AccountService, admin_actor: object, mailer: CapturingMailer) -> None:
"""A leaked database must not yield a usable link."""
service.provision_admin_temporary_password(
provision_invite(
service,
username="Hashed.Only",
display_name="Hashed Only",
email=random_email("hashed"),
@@ -257,7 +267,8 @@ def test_issuing_a_second_link_invalidates_nothing_but_keeps_both_usable(
) -> None:
"""Re-mailing must not let a stale link be replayed to activate twice."""
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="Re.Mailed",
display_name="Re Mailed",
email=random_email("re-mailed"),
@@ -290,7 +301,8 @@ def test_verification_cannot_be_repeated_on_an_active_account(
def test_resend_says_nothing_about_whether_an_account_exists(
service: AccountService, admin_actor: object, mailer: CapturingMailer
) -> None:
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="Resend.Me",
display_name="Resend Me",
email=random_email("resend"),
@@ -330,6 +342,7 @@ def test_a_delivery_failure_is_reported_not_swallowed(
username="Undeliverable",
display_name="Undeliverable",
email=random_email("undeliverable"),
organization_id=ensure_organization(service).id,
actor=admin_actor, # type: ignore[arg-type]
)
# The message names no address and carries no token.
@@ -373,7 +386,8 @@ def test_login_with_a_username_field_is_rejected(api: ApiHarness, service: Accou
def test_an_unverified_login_returns_the_actionable_error_over_http(
api: ApiHarness, service: AccountService, admin_actor: object, mailer: CapturingMailer
) -> None:
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="Http.Pending",
display_name="Http Pending",
email=random_email("http-pending"),
@@ -389,7 +403,8 @@ def test_an_unverified_login_returns_the_actionable_error_over_http(
def test_verify_email_over_http(
api: ApiHarness, service: AccountService, admin_actor: object, mailer: CapturingMailer
) -> None:
service.provision_admin_temporary_password(
provision_invite(
service,
username="Http.Verify",
display_name="Http Verify",
email=random_email("http-verify"),
@@ -408,7 +423,8 @@ def test_verify_email_over_http(
def test_the_link_can_be_followed_in_a_browser(
api: ApiHarness, service: AccountService, admin_actor: object, mailer: CapturingMailer
) -> None:
service.provision_admin_temporary_password(
provision_invite(
service,
username="Http.Link",
display_name="Http Link",
email=random_email("http-link"),
@@ -422,7 +438,8 @@ def test_the_link_can_be_followed_in_a_browser(
def test_resend_answers_the_same_way_for_both_outcomes(
api: ApiHarness, service: AccountService, admin_actor: object, mailer: CapturingMailer
) -> None:
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="Http.Resend",
display_name="Http Resend",
email=random_email("http-resend"),
@@ -548,7 +565,8 @@ def test_the_development_store_keeps_a_token_across_a_reload(tmp_path) -> None:
mailer = CapturingMailer()
service = AccountService(JsonDevelopmentRepository(settings), settings, mailer=mailer)
actor = admin_principal(provision_local(service, username="Umbraadmin"))
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="Persisted",
display_name="Persisted",
email=random_email("persisted"),

373
umbratrans_web/tests/test_postgres_repository.py
View File

@@ -2,17 +2,23 @@
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.
**Mapping and contract tests run everywhere.** The row mappers, the
role-assignment rules, the unique-violation mapping, and the row-level-security
identity contract are pure logic, so they are asserted directly. 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.
selected by ``UMBRATRANS_TEST_DSN`` and must point at a *scratch* database, never
at ``umbracontrol``. The fixture refuses a DSN whose database name contains
``umbracontrol``.
The integration layer has already earned its keep: running the forward migration
against the real control-plane schema surfaced an ``ADD CONSTRAINT`` that failed
because the constraint already existed, and the adapter was setting a
row-level-security GUC whose name did not match the one the policy helpers read.
Neither is visible without a real database.
"""
from __future__ import annotations
@@ -21,12 +27,22 @@ import os
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any
from uuid import uuid4
import pytest
from umbratrans_web.errors import ConflictError, RepositoryError
from umbratrans_web.models import AccountStatus, OrganizationStatus, RoleName, User
from umbratrans_web.config import Environment, Settings
from umbratrans_web.dev_repository import InMemoryAccountRepository, JsonDevelopmentRepository
from umbratrans_web.errors import (
ConflictError,
EmailNotVerifiedError,
InvalidCredentialsError,
RepositoryError,
)
from umbratrans_web.mailer import CapturingMailer
from umbratrans_web.models import AccountStatus, OrganizationStatus, Principal, RoleName, User
from umbratrans_web.postgres_repository import (
RLS_USER_FUNCTION,
RLS_USER_SETTING,
STATEMENT_TIMEOUT_MS,
PostgresAccountRepository,
@@ -38,6 +54,9 @@ from umbratrans_web.postgres_repository import (
_user_from_row,
)
from umbratrans_web.repository import AccountRepository
from umbratrans_web.service import AccountService, OperatorContext
from .conftest import provision_invite, random_password, verification_token
TEST_DSN_ENV = "UMBRATRANS_TEST_DSN"
@@ -64,9 +83,64 @@ def _row(**overrides: Any) -> dict[str, Any]:
# ----------------------------------------------------------------------
def _unimplemented_protocol_methods(implementation: type) -> list[str]:
"""Return the protocol methods ``implementation`` does not actually provide.
Walking ``vars()`` rather than using ``hasattr`` is the whole point. Every
implementation inherits from the protocol, so ``hasattr`` is true for all of
them no matter what is written: it passed while five methods were missing and
email verification and login were both non-functional against a real
database. A method counts as provided only if some class below the protocol
defines it.
"""
provided: set[str] = set()
for klass in implementation.__mro__:
if klass is AccountRepository:
break
provided.update(name for name, value in vars(klass).items() if callable(value))
declared = {name for name, value in vars(AccountRepository).items() if callable(value)}
return sorted(declared - provided)
def test_the_adapter_satisfies_the_repository_protocol() -> None:
for name in AccountRepository.__protocol_attrs__: # type: ignore[attr-defined]
assert hasattr(PostgresAccountRepository, name), name
"""Every declared method must be implemented, not merely inherited.
The protocol raises :class:`NotImplementedError` for anything it does not
implement, so an inherited stub is a loud failure rather than a silent
``None``. This test is what keeps the two consistent: a method added to the
protocol without an implementation fails here rather than in production.
"""
missing = _unimplemented_protocol_methods(PostgresAccountRepository)
assert missing == [], f"PostgresAccountRepository does not implement: {missing}"
def test_every_repository_implements_every_protocol_method() -> None:
"""The same check for the development repositories.
The JSON repository inherits from the in-memory one, so this also proves the
inheritance actually carries the methods rather than the protocol's stubs.
"""
for implementation in (InMemoryAccountRepository, JsonDevelopmentRepository):
missing = _unimplemented_protocol_methods(implementation)
assert missing == [], f"{implementation.__name__} does not implement: {missing}"
def test_an_inherited_stub_raises_rather_than_returning_none() -> None:
"""A repository missing a method must fail loudly.
The failure this guards against is subtle: a stub returning ``None`` is
indistinguishable from "no such row", so a missing method presents as an
empty database instead of an error.
"""
class Incomplete(AccountRepository): # type: ignore[misc]
pass
with pytest.raises(NotImplementedError):
Incomplete().get_user_by_email("someone@example.test")
def test_the_adapter_imports_without_opening_a_connection() -> None:
@@ -77,10 +151,23 @@ def test_the_adapter_imports_without_opening_a_connection() -> None:
"""
assert PostgresAccountRepository.__doc__ is not None
assert RLS_USER_SETTING == "app.current_user_id"
assert STATEMENT_TIMEOUT_MS > 0
def test_the_identity_setting_matches_what_the_policies_read() -> None:
"""The adapter and the policy helpers must agree, by construction.
``app.current_user_id()`` is defined in the control plane as
``current_setting('app.user_id', true)``. A similarly named GUC would leave
every helper resolving to NULL, so every policy would deny and every query
would return zero rows - silently. This was wrong once and the integration
run is what found it, so it is pinned here where it is cheap to check.
"""
assert RLS_USER_SETTING == "app.user_id"
assert RLS_USER_FUNCTION == "app.current_user_id"
# ----------------------------------------------------------------------
# Role and organization resolution
# ----------------------------------------------------------------------
@@ -413,16 +500,25 @@ def test_the_adapter_never_interpolates_a_value_into_sql() -> None:
def test_the_adapter_issues_no_destructive_statement() -> None:
"""Insert and update are the adapter's job; nothing that destroys data is.
"""Insert and update are the adapter's job; nothing that destroys live 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.
One delete is permitted, and only one: purging verification tokens that are
already spent or expired. That is required by the storage contract, and it
cannot reach a row that is still live. It is excluded by name here and its
predicate is pinned separately by
:func:`test_the_token_purge_cannot_reach_a_live_token`, so widening this
method's ``WHERE`` clause cannot quietly become data loss.
"""
import inspect
source = inspect.getsource(PostgresAccountRepository)
purge = inspect.getsource(PostgresAccountRepository.delete_expired_verification_tokens)
source = source.replace(purge, "")
for statement in (
"DROP ",
"TRUNCATE",
@@ -435,6 +531,28 @@ def test_the_adapter_issues_no_destructive_statement() -> None:
assert statement not in source, statement
def test_the_token_purge_cannot_reach_a_live_token() -> None:
"""The one permitted delete must stay bounded to rows that are already dead.
``bootstrap_tokens`` holds a live row for exactly one reason: an unconsumed,
unexpired link that someone may still click. The purge is allowed to remove
spent and expired rows and nothing else, so both conditions are pinned here
rather than trusted to review.
"""
import inspect
purge = inspect.getsource(PostgresAccountRepository.delete_expired_verification_tokens)
assert "DELETE FROM public.bootstrap_tokens" in purge
assert "consumed_at IS NOT NULL" in purge
assert "expires_at <= %s" in purge
# An unconsumed, unexpired row must not match: that is the whole point.
assert "consumed_at IS NULL" not in purge
assert "expires_at > " not in purge
# And it must not be able to reach any other table.
assert purge.count("DELETE FROM") == 1
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.
@@ -488,45 +606,200 @@ def postgres() -> Any:
repository.close()
@integration
def test_a_user_round_trips_through_postgres(postgres: PostgresAccountRepository) -> None:
from umbratrans_web.security import utc_now
# -- fixtures over the real database --------------------------------------
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(),
@pytest.fixture
def pg_settings() -> Settings:
return Settings(
environment=Environment.TEST,
repository_kind="postgres",
database_dsn=_integration_dsn(),
argon2_memory_cost_kib=8 * 1024,
argon2_time_cost=1,
argon2_parallelism=1,
password_min_length=14,
)
# 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"
@pytest.fixture
def pg_service(postgres: PostgresAccountRepository, pg_settings: Settings) -> AccountService:
return AccountService(postgres, pg_settings, mailer=CapturingMailer())
@pytest.fixture
def platform(pg_service: AccountService) -> Principal:
"""A real platform administrator, bootstrapped through the operator path.
A platform admin needs no organization, which is what makes it the one role
that can be created before anything else exists.
"""
password = random_password(30)
view = pg_service.create_account(
enrollment_mode="temporary_password",
username=f"Platform.Admin.{uuid4().hex[:8]}",
display_name="Platform Admin",
email=f"platform-{uuid4().hex[:8]}@example.test",
role="platform_admin",
password=password,
operator=OperatorContext(source="operator-cli"),
)
return Principal(
user_id=view.id,
username=view.username or "",
role=RoleName.PLATFORM_ADMIN,
session_id=uuid4(),
)
# -- the contract that only a real database can prove ----------------------
@integration
def test_the_rls_identity_contract_holds_against_the_real_schema(
postgres: PostgresAccountRepository,
) -> None:
"""The startup self-check must pass against the control plane's own function.
This is the test that would have caught the wrong GUC name. It is cheap, it
runs before anything else, and a failure here means every policy would deny
every request.
"""
postgres.verify_rls_contract()
@integration
def test_organization_rows_round_trip(
postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
) -> None:
created = pg_service.create_organization(platform, name="Round Trip Org", slug=f"round-trip-{uuid4().hex[:8]}")
stored = postgres.get_organization(created.id)
assert stored is not None
assert stored.name == "Round Trip Org"
by_slug = postgres.get_organization_by_slug(stored.slug.upper())
assert by_slug is not None
assert by_slug.id == created.id
organizations, total = postgres.find_organizations(limit=10)
assert total >= 1
assert any(item.id == created.id for item in organizations)
@integration
def test_a_role_resolves_through_user_roles(
pg_service: AccountService, postgres: PostgresAccountRepository, platform: Principal
) -> None:
"""The adapter has no role column to read, so this exercises the join.
Creating the account also has to write the user_roles row, and reading it
back has to resolve the role and organization from that table alone.
"""
organization = pg_service.create_organization(platform, name="Scoped Org", slug=f"scoped-{uuid4().hex[:8]}")
email = f"scoped-{uuid4().hex[:8]}@example.test"
created = pg_service.create_oidc_account(
username=f"Scoped.Member.{uuid4().hex[:6]}",
display_name="Scoped Member",
email=email,
organization_id=organization.id,
actor=platform,
)
assert created.organization_id == organization.id
fetched = postgres.get_user_by_email(email)
assert fetched is not None
assert fetched.role is RoleName.MEMBER
assert fetched.organization_id == organization.id
assert fetched.status is AccountStatus.INVITED
assert fetched.is_email_verified is False
@integration
def test_the_database_stores_only_an_argon2id_hash(
pg_service: AccountService, postgres: PostgresAccountRepository
) -> None:
# Platform scope, because the operator path still assigns a role and every
# other role needs an organization. The tenant rules have their own tests;
# this one is only about what lands in user_credentials.
pg_service.create_account(
enrollment_mode="temporary_password",
username=f"Hash.Check.{uuid4().hex[:6]}",
display_name="Hash Check",
email=f"hash-{uuid4().hex[:8]}@example.test",
role="platform_admin",
password=random_password(30),
operator=OperatorContext(source="operator-cli"),
)
with postgres._pool.connection() as connection: # noqa: SLF001 - test introspection
rows = connection.execute(
"SELECT password_hash, must_change_password FROM public.user_credentials WHERE password_hash IS NOT NULL"
).fetchall()
assert rows, "expected at least one credential"
for row in rows:
assert row["password_hash"].startswith("$argon2id$")
assert row["must_change_password"] is True
@integration
def test_email_verification_survives_a_real_round_trip(pg_service: AccountService, platform: Principal) -> None:
"""The full flow, through real tables, using the real policy helpers."""
mailer = CapturingMailer()
service = AccountService(pg_service.repository, pg_service.settings, mailer=mailer)
organization = service.create_organization(platform, name="Verify Org", slug=f"verify-{uuid4().hex[:8]}")
created = provision_invite(
service,
actor=platform,
username=f"Verify.Live.{uuid4().hex[:6]}",
display_name="Verify Live",
email=f"verify-{uuid4().hex[:8]}@example.test",
organization_id=organization.id,
)
assert created.account.status is AccountStatus.INVITED
with pytest.raises(EmailNotVerifiedError):
service.authenticate(email=created.account.email, password=created.temporary_password)
verified = service.verify_email(verification_token(mailer))
assert verified.status is AccountStatus.ACTIVE
# And the password now works.
assert service.authenticate(email=verified.email, password=created.temporary_password)
@integration
def test_suspending_an_organization_blocks_its_member(pg_service: AccountService, platform: Principal) -> None:
organization = pg_service.create_organization(platform, name="Lockout Org", slug=f"lockout-{uuid4().hex[:8]}")
password = random_password(30)
email = f"lockout-{uuid4().hex[:8]}@example.test"
pg_service.create_account(
enrollment_mode="temporary_password",
username=f"Lockout.Member.{uuid4().hex[:6]}",
display_name="Lockout Member",
email=email,
organization_id=organization.id,
password=password,
operator=OperatorContext(source="operator-cli"),
)
assert pg_service.authenticate(email=email, password=password) is not None
pg_service.set_organization_status(platform, organization.id, "suspended")
with pytest.raises(InvalidCredentialsError):
pg_service.authenticate(email=email, password=password)
@integration
def test_audit_events_round_trip_with_the_store_identifier(pg_service: AccountService, platform: Principal) -> None:
pg_service.create_organization(platform, name="Audited Org", slug=f"audited-{uuid4().hex[:8]}")
events, total = pg_service.repository.list_audit_events(limit=50)
assert total >= 1
assert events
# audit_events.id is a bigint identity, so the domain id is its text form.
assert all(event.id.isdigit() for event in events)
assert any(event.action == "organization.created" for event in events)
@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

44
umbratrans_web/tests/test_service.py
View File

@@ -32,6 +32,7 @@ from .conftest import (
admin_principal,
ensure_organization,
member_principal,
provision_invite,
provision_local,
provision_oidc,
random_email,
@@ -120,7 +121,8 @@ def test_generic_create_account_path_requires_the_operator_context(service: Acco
def test_admin_enrollment_generates_a_one_time_password(service: AccountService, mailer: CapturingMailer) -> None:
actor = admin_principal(provision_local(service, username="Umbraadmin"))
issued = service.provision_admin_temporary_password(
issued = provision_invite(
service,
username="Temp.User",
display_name="Temp User",
email=random_email("temp-user"),
@@ -162,7 +164,8 @@ def test_admin_enrollment_generates_a_one_time_password(service: AccountService,
def test_admin_enrollment_requires_an_administrator(service: AccountService) -> None:
member = provision_local(service, username="Member.User", role="member")
with pytest.raises(ForbiddenError):
service.provision_admin_temporary_password(
provision_invite(
service,
username="Escalate.User",
display_name="Escalate",
email=random_email("escalate-user"),
@@ -191,11 +194,16 @@ def test_account_creation_requires_an_email(service: AccountService) -> None:
"""
actor = admin_principal(provision_local(service, username="Umbraadmin"))
# A tenant is named so the missing address is the only thing wrong. Both
# roles are organization-scoped, so without this the assignment is refused
# first and the email rule is never reached.
organization_id = ensure_organization(service).id
with pytest.raises(InputError) as error:
service.create_oidc_account(
username="No.Email",
display_name="No Email",
email=None, # type: ignore[arg-type]
organization_id=organization_id,
actor=actor,
)
assert error.value.code == "email_required"
@@ -206,6 +214,7 @@ def test_account_creation_requires_an_email(service: AccountService) -> None:
username="Also.No.Email",
display_name="Also No Email",
email=None, # type: ignore[arg-type]
organization_id=organization_id,
password=random_password(28),
operator=OperatorContext(source="operator-cli"),
)
@@ -230,6 +239,37 @@ def test_oidc_enrollment_creates_no_password_material(repository: object, servic
assert account.view.credential.credential_type is CredentialType.OIDC
def test_the_operator_path_refuses_to_pretend_it_verifies_email(service: AccountService) -> None:
"""``require_email_verification`` must be refused, never dropped.
The operator path attests to the address itself, so it cannot send a
verification message: the first account on a new deployment has to exist
before mail is configured. The flag used to be accepted here and silently
ignored, which returned an unverified account to a caller that had asked for
a verified one. Refusing says which path to use instead.
"""
with pytest.raises(InputError) as error:
service.create_account(
enrollment_mode="temporary_password",
username="Wants.Verification",
display_name="Wants Verification",
email=random_email("wants-verification"),
password=random_password(28),
operator=OperatorContext(source="operator-cli"),
require_email_verification=True,
)
assert error.value.code == "operator_path_skips_verification"
def test_the_operator_path_creates_an_already_attested_account(service: AccountService) -> None:
"""The operator path's own default, stated so it cannot change unnoticed."""
account = provision_local(service, username="Attested.User")
assert account.view.status is AccountStatus.ACTIVE
assert account.view.is_email_verified is True
def test_oidc_account_cannot_authenticate_with_a_local_password(service: AccountService) -> None:
actor = admin_principal(provision_local(service, username="Umbraadmin"))
account = provision_oidc(service, actor, username="Oidc.User")