"""The account system's front door to the vendored BW kit: the mapped store, the permissions, the seed levels, the owner, the resolvers `is_owner` and `can`, and the admin router. Sources of truth: this module, `app/vendor/bw_accounts.py` (the model and its guards, never edited), `app/vendor/bw_store_sqlalchemy.py` (the store), `app/vendor/bw_admin_api.py` (the router and the `/api/bw/me` contract), `docs/ACCOUNTS.md`. Design: `.logs/planning/accounts-2026-09.md` §4.1, §4.2, §4.6, §4.11. The kit is wired in `init()` at startup; every permission question the app asks goes through `can()` and `is_owner()` here, never a level name. The store maps the kit's four concepts onto DFP's own tables (`owns_tables=False`: Alembic owns every table, startup runs no DDL) and converts an `IntegrityError` on the kit's add paths into `AccountsError('EXISTS')`, the race-tolerance rule; `add_member` also refuses a disabled principal with `DISABLED`. The owner is code, not a row: `ACCOUNT_OWNER` names the super admin. When it is unset or a placeholder the kit receives the sentinel `NO_OWNER` (a truthy string that fails the username regex), because the kit falls back to a default owner on any falsy value and a silently defaulted owner is a footgun in a copied kit. The directory the router talks to is reached through a proxy so the test suite can wrap it and record the kit's calls. """ from __future__ import annotations import logging import re from sqlalchemy import select from sqlalchemy.exc import IntegrityError from app import db as appdb from app.config import settings from app.models import Account, AccountGrant, AccountLevel, AccountMember, Brand from app.services import audit_log, identity, sessions from app.vendor import bw_accounts as bwa from app.vendor import bw_store_sqlalchemy as bwstore from app.vendor import bw_view_as from app.vendor.bw_admin_api import build_router, me_payload log = logging.getLogger(__name__) AccountsError = bwa.AccountsError USERNAME_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{1,31}$") NO_OWNER = "!no-owner" KIT_PREFIX = "/api/bw" # DFP's own permissions (accounts plan §4.6). Registered with the kit as enforced so the # Levels editor labels them honestly; the kit's accounts.*, levels.*, instances.* ride along. PERM_CLIENT_VIEW = "client.view" PERM_CLIENT_PARTICIPATE = "client.participate" PERM_DISCUSSION_CURATE = "discussion.curate" PERM_PLAN_VIEW = "plan.view" PERM_ITEMS_ACT = "items.act" PERM_SOURCES_MANAGE = "sources.manage" #: What one address is: indexable, noindex, unlisted or hidden, shown on every page to the #: people who decide it (docs/SEO.md "Page status: what a signed-in admin sees"). PERM_PAGES_STATUS = "pages.status" DFP_PERMISSIONS = ( PERM_CLIENT_VIEW, PERM_CLIENT_PARTICIPATE, PERM_DISCUSSION_CURATE, PERM_PLAN_VIEW, PERM_ITEMS_ACT, PERM_SOURCES_MANAGE, PERM_PAGES_STATUS, ) # Seeded by `app.cli backfill levels` (insert if absent, never update). `admin` is rian's # word for Adam and Mark: the client pages and participation; the meaning is DFP's, not the # kit's template of the same name. `member` holds nothing: the level a consumer is promoted # onto if they ever need a membership row. SEED_LEVELS = { "admin": {"permissions": [PERM_CLIENT_VIEW, PERM_CLIENT_PARTICIPATE, PERM_PAGES_STATUS], "assignable": []}, "member": {"permissions": [], "assignable": []}, } #: What each level held before the permission above joined its seed. A host seeded earlier #: gains it from `app.cli backfill level_permissions`, and ONLY where the level still holds #: exactly this: a human value is never overwritten by a machine. PREVIOUS_SEED_PERMISSIONS = { "admin": [PERM_CLIENT_VIEW, PERM_CLIENT_PARTICIPATE], "member": [], } def owner_or_sentinel() -> str: name = settings.owner_username return name if name and USERNAME_RE.match(name) else NO_OWNER def valid_username(name: str) -> bool: return bool(USERNAME_RE.match(name or "")) # --- the store ------------------------------------------------------------------------------ class Store(bwstore.SqlAlchemyStore): """The kit's store over DFP's tables, with the two DFP rules on the add paths.""" def _status_of(self, username: str) -> str | None: with self._sf() as db: return db.scalar(select(Account.status).where(Account.username == username)) def add_level(self, name, permissions, assignable): try: super().add_level(name, permissions, assignable) except IntegrityError: raise AccountsError(f"Level '{name}' already exists.", "EXISTS") from None def add_member(self, username, level, all_instances, added_by): status = self._status_of(username) if status is None: raise AccountsError(f"No account '{username}' to add.", "NO_SUCH_MEMBER") if status == "disabled": raise AccountsError(f"'{username}' is disabled; enable it first.", "DISABLED") try: super().add_member(username, level, all_instances, added_by) except IntegrityError: raise AccountsError(f"'{username}' is already a member.", "EXISTS") from None def set_grant(self, username, instance_id, level): try: super().set_grant(username, instance_id, level) except IntegrityError: raise AccountsError(f"'{username}' already holds '{instance_id}'.", "EXISTS") from None def _session_factory(): """Looked up at call time, so the test suite's engine swap reaches the kit's store.""" return appdb.SessionLocal() STORE = Store( _session_factory, levels=bwstore.LevelsSpec(AccountLevel.__table__), members=bwstore.MembersSpec(AccountMember.__table__, active_col="active", soft_delete=True), instances=bwstore.InstancesSpec(Brand.__table__, id="slug", label="name", readonly=True), grants=bwstore.GrantsSpec(AccountGrant.__table__), ) def init() -> str: """Wire the kit: once at startup, and again by the test suite after it sets the owner.""" owner = owner_or_sentinel() bwa.init(store=STORE, owner=owner, has_instances=False, audit=audit_log.kit_sink) bwa.register_enforced(*DFP_PERMISSIONS) if owner == NO_OWNER: log.critical( "ACCOUNT_OWNER is unset, a placeholder or not a valid username: nobody is the " "super admin on this host. Set it in .app.env and recreate the container " "(docs/ACCOUNTS.md); `accounts owner-check` says the same." ) return owner # --- the resolvers ---------------------------------------------------------------------------- def _usable(username: str | None) -> bool: """A name that could belong to a principal: the sentinel and anything else that fails the username regex is nobody, so the kit's owner check can never match it.""" return bool(username) and valid_username((username or "").strip().lower()) def is_owner(username: str | None) -> bool: return _usable(username) and bwa.is_owner(username) def can(username: str | None, permission: str) -> bool: """THE permission check: the owner always; otherwise the app-wide level's permissions; a principal with no member row (a consumer) holds nothing.""" return _usable(username) and bwa.can(username, permission) def holder(username: str | None): """A `permission -> bool` closure for the access policy; the literal `owner` too.""" return lambda permission: is_owner(username) if permission == "owner" else can(username, permission) # --- the directory, behind a proxy the tests can wrap ----------------------------------------- DIRECTORY = None # set by `directory_instance()` on first use; the tests replace it def directory_instance(): global DIRECTORY if DIRECTORY is None: from app.services.directory import LocalDirectory DIRECTORY = LocalDirectory() return DIRECTORY class _DirectoryProxy: """What `build_router` receives as `bw_auth`: every attribute resolves against the current `DIRECTORY` at call time, so a recorder installed by the tests is what the kit calls, and `getattr(bw_auth, name, None)` probes still work.""" def __getattr__(self, name): return getattr(directory_instance(), name) # --- /api/bw/me with DFP's capabilities ------------------------------------------------------- def capabilities_for(request) -> dict: """The DFP block beside the kit's booleans: the DFP permissions of the EFFECTIVE user plus the must-change flag of the REAL account; all false for an anonymous caller.""" user = identity.optional_user(request) row = sessions.of_request(request) return { "client_view": can(user, PERM_CLIENT_VIEW), "client_participate": can(user, PERM_CLIENT_PARTICIPATE), "discussion_curate": can(user, PERM_DISCUSSION_CURATE), "plan_view": can(user, PERM_PLAN_VIEW), "items_act": can(user, PERM_ITEMS_ACT), "sources_manage": can(user, PERM_SOURCES_MANAGE), "pages_status": can(user, PERM_PAGES_STATUS), "must_change_password": bool(user and row is not None and row.must_change_password), } def me_payload_for(request) -> dict: """The kit's `/api/bw/me` contract (same arguments `build_router` receives) plus DFP's `capabilities` and the effective user's display name.""" from app.services import view_as user = identity.optional_user(request) real = identity.real_user(request) row = sessions.of_request(request) out = me_payload(bwa, user, real=real, bw_view_as=bw_view_as, session=row if user else None, view_as_policy=view_as.can_view_as, account_url=None) out["capabilities"] = capabilities_for(request) out["display_name"] = _display_name(user) if user else None return out def _display_name(username: str) -> str: with appdb.SessionLocal() as db: return db.scalar(select(Account.display_name).where(Account.username == username)) or username def kit_router(): """The vendored admin API at `/api/bw`, every hook wired; mounted after DFP's own `/api/bw/me` so the kit's `/me` is shadowed, never edited.""" from app.services import view_as return build_router( bwa=bwa, optional_user=identity.optional_user, bw_auth=_DirectoryProxy(), bw_view_as=bw_view_as, real_user=identity.real_user, instance_obj=None, session_of=sessions.of_request, view_as_policy=view_as.can_view_as, rank_of=view_as.rank_of, target_valid=view_as.target_valid, start_hook=view_as.start, stop_hook=view_as.stop, account_url=None, prefix=KIT_PREFIX, audit=audit_log.kit_sink, )