Tech-debt batch 9: typed app.state accessors; demo read-only on the resolved principal (#609 #611)

CHANGELOG.md

@@ -53,6 +53,7 @@ All notable changes to this project are documented in this file. The format is b
   tracked for removal in 0.1.0 (#594).
 
 ### Fixed
+- Users: read-only demo mode now follows the resolved identity, not only the session cookie. Writes made with a bearer access token or personal API token of a demo account are refused (403 `DEMO_READ_ONLY`, or the Inertia 409 redirect) through the new `AuthState.principal_guards` hook, and demo accounts can no longer mint tokens via `POST /api/users/auth/token` or `/token/refresh` while `demo_read_only` is on (#611).
 - Users: the default user role is created at startup when missing, so an app scaffolded with
   `smpy new` (schema baseline, no data migrations) can register users (#600).
 - Tenants: deleting a tenant now tears down its data. `TenantDeleted` (now defined in

framework/core/simple_module_core/app_state.py

@@ -0,0 +1,199 @@
+"""Typed reads of optional app services, preserving absent-state fallbacks.
+
+Concrete hosting types are imported only for type checking, as in Services:
+core stays usable without hosting or any plugin installed.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Protocol
+
+if TYPE_CHECKING:
+    from asyncio import Lock
+    from collections.abc import Callable
+
+    from simple_module_hosting._host_services import _HostServices
+    from simple_module_hosting._tenant import TenantResolver
+    from simple_module_hosting.middleware import PrincipalSerializer
+    from simple_module_hosting.shared_props import SharedPropsProvider
+    from simple_module_inertia import Inertia
+    from starlette.requests import Request
+
+    from simple_module_core.body_limits import BodyLimitRegistry
+    from simple_module_core.design_packs import DesignPackRegistry
+    from simple_module_core.module import ModuleBase
+    from simple_module_core.public_routes import PublicRouteRegistry
+    from simple_module_core.rate_limit import RateLimitRegistry
+    from simple_module_core.services import Services
+    from simple_module_core.tenancy import TenantExists, TenantMembers
+
+
+#: Attribute ``register_inertia_shared_provider`` writes and the renderer reads.
+INERTIA_SHARED_PROVIDERS_STATE_ATTR = "inertia_shared_providers"
+
+
+class AuthProvider(Protocol):
+    """Framework-facing auth extension; installed by users or keycloak."""
+
+    async def resolve_user(self, request: Request) -> Any: ...
+
+    def get_login_url(self, request: Request, next_url: str | None = None) -> str: ...
+
+
+class BrandingState(Protocol):
+    """Optional branding services consumed by the document renderer."""
+
+    @property
+    def settings(self) -> object: ...
+
+    @property
+    def favicon_url(self) -> str | None: ...
+
+
+def get_auth_provider(app: Any) -> AuthProvider | None:
+    """Auth.register_settings installs auth; users/keycloak install its provider."""
+    return getattr(getattr(app.state, "auth", None), "auth_provider", None)
+
+
+def get_services(app: Any) -> Services | None:
+    """create_app installs the framework service container.
+
+    Returns None when absent.
+    """
+    return getattr(getattr(app, "state", None), "sm", None)
+
+
+def get_host_services(app: Any) -> _HostServices | None:
+    """Host settings registration installs the host service container.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "host", None)
+
+
+def get_public_routes(app: Any) -> PublicRouteRegistry | None:
+    """create_app installs the public-route registry.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "public_routes", None)
+
+
+def get_body_limits(app: Any) -> BodyLimitRegistry | None:
+    """create_app installs the request-body limit registry.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "body_limits", None)
+
+
+def get_rate_limit_rules(app: Any) -> RateLimitRegistry | None:
+    """create_app installs the per-route rate-limit registry.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "rate_limit_rules", None)
+
+
+def get_design_packs(app: Any) -> DesignPackRegistry | None:
+    """create_app installs the design-pack registry.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "design_packs", None)
+
+
+def get_migration_status(app: Any) -> dict[str, Any] | None:
+    """The hosting lifespan and setup wizard install the migration snapshot.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "migration", None)
+
+
+def get_openapi_url(app: Any) -> str | None:
+    """Hosting boot records the served OpenAPI URL, or None when disabled.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "openapi_url", None)
+
+
+def get_spa_prefixes(app: Any) -> tuple[str, ...]:
+    """SPA mounting records the mounted prefixes; absent means no mounts.
+
+    Returns () when absent.
+    """
+    return getattr(app.state, "spa_prefixes", ())
+
+
+def get_deferred_startup(app: Any) -> list[ModuleBase] | None:
+    """The hosting lifespan records modules awaiting migrations.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "deferred_startup", None)
+
+
+def get_deferred_startup_lock(app: Any) -> Lock | None:
+    """Deferred startup replay lazily installs its per-app lock.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "deferred_startup_lock", None)
+
+
+def get_principal_serializer(app: Any) -> PrincipalSerializer | None:
+    """The auth module installs the principal-to-Inertia serializer.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "principal_serializer", None)
+
+
+def get_tenant_resolver(app: Any) -> TenantResolver | None:
+    """The tenants module installs request tenant resolution.
+
+    Returns None when absent.
+    """
+    return getattr(getattr(app, "state", None), "tenant_resolver", None)
+
+
+def get_tenant_exists(app: Any) -> TenantExists | None:
+    """The tenants module installs the tenant-existence callback.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "tenant_exists", None)
+
+
+def get_tenant_members(app: Any) -> TenantMembers | None:
+    """The tenants module installs the membership lookup callback.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "tenant_members", None)
+
+
+def get_inertia_dependency(app: Any) -> Callable[..., Inertia] | None:
+    """Hosting Inertia setup installs the configured dependency factory.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "inertia_dependency", None)
+
+
+def get_inertia_shared_providers(app: Any) -> list[SharedPropsProvider] | None:
+    """register_inertia_shared_provider lazily installs the provider list.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, INERTIA_SHARED_PROVIDERS_STATE_ATTR, None)
+
+
+def get_branding(app: Any) -> BrandingState | None:
+    """The branding module installs the optional application theme services.
+
+    Returns None when absent.
+    """
+    return getattr(app.state, "branding", None)

framework/core/simple_module_core/tenancy.py

@@ -17,6 +17,8 @@
 from enum import StrEnum
 from typing import Any
 
+from simple_module_core.app_state import get_tenant_exists, get_tenant_members
+
 TENANT_ROLE_PREFIX = "tenant:"
 
 
@@ -72,7 +74,7 @@ async def tenant_exists(app: Any, tenant_id: str) -> bool | None:
     URL ask through this, so they can refuse a typo without importing that
     module; with none installed the answer is ``None`` and they accept the id.
     """
-    check: TenantExists | None = getattr(app.state, "tenant_exists", None)
+    check: TenantExists | None = get_tenant_exists(app)
     if check is None:
         return None
     return bool(await check(tenant_id))
@@ -97,7 +99,7 @@ async def tenant_member_ids(
     no membership store is installed; callers on a multi-tenant app treat it
     as "nobody" (fail closed).
     """
-    check: TenantMembers | None = getattr(app.state, "tenant_members", None)
+    check: TenantMembers | None = get_tenant_members(app)
     if check is None:
         return None
     if not user_ids:

framework/core/tests/test_core_app_state.py

@@ -0,0 +1,57 @@
+"""Optional app services retain their defaults and installed object identity."""
+
+from types import SimpleNamespace
+
+import pytest
+from fastapi import FastAPI
+from simple_module_core import app_state
+
+
+@pytest.mark.parametrize(
+    ("attribute", "accessor_name", "default"),
+    [
+        ("sm", "get_services", None),
+        ("host", "get_host_services", None),
+        ("public_routes", "get_public_routes", None),
+        ("body_limits", "get_body_limits", None),
+        ("rate_limit_rules", "get_rate_limit_rules", None),
+        ("design_packs", "get_design_packs", None),
+        ("migration", "get_migration_status", None),
+        ("openapi_url", "get_openapi_url", None),
+        ("spa_prefixes", "get_spa_prefixes", ()),
+        ("deferred_startup", "get_deferred_startup", None),
+        ("deferred_startup_lock", "get_deferred_startup_lock", None),
+        ("principal_serializer", "get_principal_serializer", None),
+        ("tenant_resolver", "get_tenant_resolver", None),
+        ("tenant_exists", "get_tenant_exists", None),
+        ("tenant_members", "get_tenant_members", None),
+        ("inertia_dependency", "get_inertia_dependency", None),
+        ("inertia_shared_providers", "get_inertia_shared_providers", None),
+        ("branding", "get_branding", None),
+        ("auth", "get_auth_provider", None),
+    ],
+)
+def test_core_optional_app_service(attribute, accessor_name, default):
+    app = FastAPI()
+    accessor = getattr(app_state, accessor_name)
+    assert accessor(app) == default
+
+    installed = object()
+    value = SimpleNamespace(auth_provider=installed) if attribute == "auth" else installed
+    setattr(app.state, attribute, value)
+    assert accessor(app) is installed
+
+    setattr(app.state, attribute, None)
+    assert accessor(app) is None
+
+
+def test_core_partial_auth_state_has_no_provider():
+    app = FastAPI()
+    app.state.auth = SimpleNamespace()
+    assert app_state.get_auth_provider(app) is None
+
+
+def test_core_optional_scope_app_can_lack_state():
+    assert app_state.get_services(None) is None
+    assert app_state.get_services(SimpleNamespace()) is None
+    assert app_state.get_tenant_resolver(None) is None

framework/hosting/simple_module_hosting/_boot_checks.py

@@ -13,6 +13,7 @@
 import logging
 from typing import TYPE_CHECKING, Any
 
+from simple_module_core.app_state import get_tenant_resolver
 from simple_module_core.diagnostics import Diagnostic, check_tenant_resolver, print_diagnostics
 
 from simple_module_hosting._route_access import check_unguarded_api_routes
@@ -27,7 +28,7 @@
 def _tenant_checks(app: FastAPI, settings: Any) -> list[Diagnostic]:
     return check_tenant_resolver(
         multi_tenant=bool(getattr(settings, "multi_tenant", False)),
-        resolver=getattr(app.state, "tenant_resolver", None),
+        resolver=get_tenant_resolver(app),
     )
 
 

framework/hosting/simple_module_hosting/_deferred_startup.py

@@ -6,6 +6,7 @@
 import logging
 
 from fastapi import FastAPI
+from simple_module_core.app_state import get_deferred_startup, get_deferred_startup_lock
 from simple_module_core.module_settings import get_module_settings_registry
 from simple_module_db.stored_settings import SessionOverrideSource, settings_table_exists
 
@@ -46,10 +47,10 @@ async def run_deferred_startup(app: FastAPI) -> None:
     rather than raised, because this runs after the migrations have already
     committed and a 500 there would help nobody.
     """
-    deferred = getattr(app.state, "deferred_startup", None)
+    deferred = get_deferred_startup(app)
     if not deferred:
         return
-    lock = getattr(app.state, "deferred_startup_lock", None)
+    lock = get_deferred_startup_lock(app)
     if lock is None:  # no await between check and set, so this cannot race
         lock = app.state.deferred_startup_lock = asyncio.Lock()
     async with lock:

framework/hosting/simple_module_hosting/_error_page.py

@@ -7,6 +7,7 @@
 from http import HTTPStatus
 
 from fastapi.responses import JSONResponse
+from simple_module_core.app_state import get_auth_provider, get_inertia_dependency
 from simple_module_inertia import (
     Inertia,
     InertiaConfig,
@@ -52,8 +53,7 @@ def _login_url(request: Request) -> str | None:
     concern and ``SM009`` forbids framework code importing ``modules/*``. An
     app with no auth provider installed simply gets no sign-in button.
     """
-    auth_state = getattr(request.app.state, "auth", None)
-    provider = getattr(auth_state, "auth_provider", None)
+    provider = get_auth_provider(request.app)
     if provider is None:
         return None
     try:
@@ -94,7 +94,7 @@ async def render_error_page(
         # root-relative page url: the adapter does that by construction now,
         # so the fallback no longer needs a separate patch to avoid the
         # cross-scheme `pushState` SecurityError behind a TLS proxy.
-        inertia_dep = getattr(request.app.state, "inertia_dependency", None)
+        inertia_dep = get_inertia_dependency(request.app)
         if inertia_dep is not None:
             inertia = inertia_dep(request, None)
         else:

framework/hosting/simple_module_hosting/_inertia_setup.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 
 from fastapi import FastAPI
+from simple_module_core.app_state import get_branding, get_host_services
 from simple_module_inertia import InertiaConfig, SiteInfo, inertia_dependency_factory
 from starlette.requests import Request
 
@@ -50,8 +51,8 @@ def branding_head(request: Request) -> dict:
     """
     # A per-request object (same duck shape) wins: branding resolves a tenant's
     # own theme into ``request.state.branding`` from its shared-props provider.
-    services = getattr(getattr(request, "state", None), "branding", None) or getattr(
-        request.app.state, "branding", None
+    services = getattr(getattr(request, "state", None), "branding", None) or get_branding(
+        request.app
     )
     settings = getattr(services, "settings", None)
     app_name = getattr(settings, "app_name", "") or _DEFAULT_APP_NAME
@@ -68,7 +69,7 @@ def branding_head(request: Request) -> dict:
 
 def site_info(request: Request) -> SiteInfo:
     """The public base URL (operator setting, never the request ``Host``) and app name."""
-    host = getattr(request.app.state, "host", None)
+    host = get_host_services(request.app)
     base_url = getattr(getattr(host, "settings", None), "public_base_url", "") or ""
     return SiteInfo(base_url=base_url, app_name=branding_head(request)["app_name"])