diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index b67ef96..b2d8930 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -66,6 +66,7 @@ enlace/ ├── util.py # Pure helpers: derive_display_name, derive_route_prefix, is_skippable ├── discover.py # ConventionDiscoverer: walks apps/, reads app.toml, detects types ├── compose.py # build_backend(): mounts ASGI sub-apps, proxy routes, static files +├── access.py # can_see_app / is_user_allowed: the ONE visibility predicate (launcher + auth gate) ├── proxy.py # Lightweight ASGI reverse proxy for process/external backends (httpx) ├── supervise.py # Dev-mode asyncio process supervisor (health checks, restart, logs) ├── diagnose.py # diagnose_app(): scan an app dir for enlace compatibility issues diff --git a/enlace/access.py b/enlace/access.py new file mode 100644 index 0000000..2b3dee2 --- /dev/null +++ b/enlace/access.py @@ -0,0 +1,105 @@ +"""Who may reach an app: the one predicate the gate and the launcher both call. + +Two code paths answer "may this user reach this app": the request-time gate +(``enlace_auth``'s ``PlatformAuthMiddleware``) and the ``/_apps`` launcher (which +hides what the caller could not open). When each carried its own copy of the +answer they drifted — the gate honoured runtime grants and the launcher did not, +so a granted user could open an app they could never find +(https://github.com/i2mint/enlace/issues/35). This module is the single copy; +both sides call it, so they cannot disagree. + +enlace core still enforces nothing — enforcement is ``enlace_auth``'s job. The +launcher uses this module to *filter*, the gate to *deny*. + +Runtime grants reach enlace core through one dependency-injection slot on the +root app's state, :data:`GRANTS_STATE_ATTR`: a callable ``app_id -> iterable of +emails`` of currently active grants, installed by the auth plugin (the same +callable its gate consults). Absent it, only the static ``allowed_users`` count. +""" + +import logging +from typing import Callable, Iterable, Optional + +_logger = logging.getLogger("enlace.access") + +#: The root app's ``state`` attribute an auth plugin sets to its grants resolver. +GRANTS_STATE_ATTR = "dynamic_allowed_users" + +GrantsResolver = Callable[[str], Iterable[str]] + + +def granted_users(resolver: Optional[GrantsResolver], app_id: str) -> frozenset[str]: + """The lowercased emails holding an active runtime grant for ``app_id``. + + Called per request, never cached: expiry is evaluated by the resolver at call + time, so the gate and the launcher expire a grant at the same instant. + + A failing resolver yields no grants (fail closed for grant-based access, + static ``allowed_users`` still work): a grants-store hiccup must never break + auth or the listing. It is logged, not raised. + """ + if resolver is None: + return frozenset() + try: + return frozenset(e.lower() for e in (resolver(app_id) or ())) + except Exception as exc: # noqa: BLE001 - see the fail-closed note above + # One line per failure, traceback at DEBUG: the launcher resolves every + # protected app per /_apps call, so a broken store would otherwise emit + # a traceback per app per page load. + _logger.warning( + "dynamic grants lookup failed for app_id=%r (%s: %s); " + "falling back to config allowed_users only", + app_id, + type(exc).__name__, + exc, + ) + _logger.debug("grants lookup traceback", exc_info=True) + return frozenset() + + +def is_user_allowed( + user_id: Optional[str], + user_email: Optional[str], + *, + allowed_users: Iterable[str] = (), + granted: Iterable[str] = (), +) -> bool: + """Whether an authenticated user passes a ``protected:user`` app's allowlist. + + The allowlist is the static ``allowed_users`` ∪ the active runtime + ``granted`` emails, compared case-insensitively. An empty union means the + app is open to any authenticated user. An unauthenticated caller + (``user_id is None``) never passes. + """ + if user_id is None: + return False + allowed = {e.lower() for e in allowed_users} | {e.lower() for e in granted} + if not allowed: + return True + return (user_email or user_id).lower() in allowed + + +def can_see_app( + access: str, + user_id: Optional[str], + user_email: Optional[str], + *, + allowed_users: Iterable[str] = (), + granted: Iterable[str] = (), +) -> bool: + """Whether the ``/_apps`` launcher shows an app of this access level. + + - ``public`` / ``local`` → always. + - ``protected:shared`` → always: it is gated when opened, not when listed, + so users know the app exists and can ask for the password. + - ``protected:user`` → exactly when :func:`is_user_allowed` — the gate's own + predicate, so a user sees precisely the apps they can open. + - anything else → never (deny by default). + """ + if access in ("public", "local", "protected:shared"): + return True + if access == "protected:user": + return is_user_allowed( + user_id, user_email, allowed_users=allowed_users, granted=granted + ) + return False diff --git a/enlace/base.py b/enlace/base.py index b8efdc4..4e21c44 100644 --- a/enlace/base.py +++ b/enlace/base.py @@ -6,7 +6,7 @@ don't see entries they couldn't open anyway. That makes the access string vocabulary part of enlace's contract — the values ``"public" | "local" | "protected:shared" | "protected:user"`` are the -ones ``compose._can_access`` understands; anything else is treated as +ones ``enlace.access.can_see_app`` understands; anything else is treated as deny-by-default. The fields are otherwise opaque to enlace: enforcement, session lookup, diff --git a/enlace/compose.py b/enlace/compose.py index 3d4269c..021e5a2 100644 --- a/enlace/compose.py +++ b/enlace/compose.py @@ -508,31 +508,30 @@ def _stamp_updated_at( app.updated_at = manifest.app_source.committed_at or manifest.deployed_at -def _can_access( - access: str, - user_id: Optional[str], - user_email: Optional[str], - allowed_users: list[str], -) -> bool: - """Whether a request can see an app of this access level in /_apps. - - `public` / `local` → always. - `protected:user` → only if authenticated AND (no allowed_users list, or - the user's email is in it). - `protected:shared` → visible either way (gated at open-time, not - discovery-time — users should know the app exists - so they can ask for the password). +def _can_access(app: AppConfig, request: Request) -> bool: + """Whether the caller of ``request`` may see ``app`` in /_apps. + + Delegates to :func:`enlace.access.can_see_app` — the same allowlist + predicate the auth gate calls — with the runtime grants the auth plugin + injected (see :data:`enlace.access.GRANTS_STATE_ATTR`), resolved per + request so a grant appears, and expires, in the launcher exactly when it + does at the gate. """ - if access in ("public", "local", "protected:shared"): - return True - if access == "protected:user": - if user_id is None: - return False - if allowed_users: - who = user_email or user_id - return who in allowed_users - return True - return False + from enlace import access + + resolver = getattr(request.app.state, access.GRANTS_STATE_ATTR, None) + granted = ( + access.granted_users(resolver, app.name) + if app.access == "protected:user" + else () + ) + return access.can_see_app( + app.access, + getattr(request.state, "user_id", None), + getattr(request.state, "user_email", None), + allowed_users=app.allowed_users, + granted=granted, + ) def _app_launch(app: AppConfig) -> tuple[bool, Optional[str]]: @@ -656,7 +655,7 @@ async def apps_listing(request: Request) -> dict: for app in apps: if app.name == landing_name: continue - if not _can_access(app.access, user_id, user_email, app.allowed_users): + if not _can_access(app, request): continue items.append( build_launcher_item(app, config, _overlay_entry(request, app.name)) @@ -676,9 +675,7 @@ def _visible_app(name: str, request: Request) -> Optional[AppConfig]: app = apps_by_name.get(name) if app is None or app.name == landing_name: return None - user_id = getattr(request.state, "user_id", None) - user_email = getattr(request.state, "user_email", None) - if not _can_access(app.access, user_id, user_email, app.allowed_users): + if not _can_access(app, request): return None return app @@ -767,11 +764,15 @@ def _icon_response(body: bytes, content_type: str, token: str, cache: str) -> Re def _add_index_route(parent: FastAPI, config: PlatformConfig) -> None: - """Add a GET / route that lists all discovered apps as a simple HTML page.""" - apps = config.apps + """Add a GET / route listing, as a simple HTML page, the apps the caller may see. + + Filtered by the same predicate as ``/_apps`` (:func:`_can_access`), so the + built-in index never names an app the caller could not open. + """ @parent.get("/", response_class=HTMLResponse) - async def index(): + async def index(request: Request): + apps = [a for a in config.apps if _can_access(a, request)] items = [] for app in apps: has_frontend = app.frontend_dir and app.frontend_dir.is_dir() diff --git a/enlace/tests/test_access.py b/enlace/tests/test_access.py new file mode 100644 index 0000000..9125b36 --- /dev/null +++ b/enlace/tests/test_access.py @@ -0,0 +1,182 @@ +"""The launcher and the auth gate share one visibility predicate (issue #35). + +A user holding a runtime grant for a ``protected:user`` app could open it (the +gate unions static ``allowed_users`` with live grants) but never see it in +``/_apps`` (the launcher did not). These tests pin the property that matters — +the launcher's verdict equals the gate's predicate for every combination of +static-only, grant-only, both and neither — not merely "a granted user sees it", +which a hard-coded ``True`` would pass. +""" + +import itertools + +import pytest +from starlette.testclient import TestClient + +from enlace import access +from enlace.base import AppConfig, PlatformConfig +from enlace.compose import build_backend + +STATIC = "static@x.com" +GRANTED = "granted@x.com" +BOTH = "both@x.com" +NEITHER = "neither@x.com" + + +class _FakeSession: + """Stand-in for the auth plugin's session middleware: identity from a header.""" + + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if scope["type"] == "http": + email = dict(scope["headers"]).get(b"x-test-user") + state = scope.setdefault("state", {}) + state["user_id"] = email.decode() if email else None + state["user_email"] = email.decode() if email else None + await self.app(scope, receive, send) + + +def _gated_app(allowed_users=(STATIC, BOTH)): + return AppConfig( + name="gated", + route_prefix="/api/gated", + app_type="frontend_only", + access="protected:user", + allowed_users=list(allowed_users), + ) + + +def _client(apps, grants): + """A backend whose injected grants resolver reads the mutable ``grants`` dict.""" + backend = build_backend(PlatformConfig(apps=apps)) + backend.add_middleware(_FakeSession) + setattr( + backend.state, access.GRANTS_STATE_ATTR, lambda app_id: grants.get(app_id, ()) + ) + return TestClient(backend) + + +def _listed(client, user): + headers = {"x-test-user": user} if user else {} + return [a["name"] for a in client.get("/_apps", headers=headers).json()["apps"]] + + +@pytest.mark.parametrize("user", [STATIC, GRANTED, BOTH, NEITHER, None]) +def test_launcher_verdict_equals_the_gate_predicate(user): + """For every kind of user, listed in /_apps ⟺ allowed by the gate's predicate.""" + grants = {"gated": {GRANTED, BOTH}} + app = _gated_app() + client = _client([app], grants) + gate_says = access.is_user_allowed( + user, + user, + allowed_users=app.allowed_users, + granted=access.granted_users(lambda a: grants.get(a, ()), "gated"), + ) + assert ("gated" in _listed(client, user)) is gate_says + # And the concrete expectations, so a predicate that is wrong on both + # sides cannot pass by agreeing with itself. + assert gate_says is (user in (STATIC, GRANTED, BOTH)) + + +def test_grant_appears_and_expires_without_restart(): + """The resolver is consulted per request: no startup snapshot to go stale.""" + grants: dict = {} + client = _client([_gated_app()], grants) + assert _listed(client, GRANTED) == [] + grants["gated"] = {GRANTED} + assert _listed(client, GRANTED) == ["gated"] + del grants["gated"] + assert _listed(client, GRANTED) == [] + + +def test_granted_user_gets_the_icon_too(): + """The icon route shares the verdict — a listed app never has a broken icon.""" + client = _client([_gated_app()], {"gated": {GRANTED}}) + assert client.get("/_apps/gated/icon", headers={"x-test-user": GRANTED}).is_success + assert ( + client.get("/_apps/gated/icon", headers={"x-test-user": NEITHER}).status_code + == 404 + ) + + +def test_matching_is_case_insensitive_on_both_sources(): + """The gate lowercases both sides; the launcher must as well.""" + app = _gated_app(allowed_users=["Static@X.com"]) + client = _client([app], {"gated": {"Granted@X.COM"}}) + assert _listed(client, "static@x.com") == ["gated"] + assert _listed(client, "GRANTED@x.com") == ["gated"] + + +def test_failing_resolver_falls_back_to_static_allowlist(): + """A grants-store hiccup hides grant-only access but never breaks the listing.""" + + def broken(app_id): + raise OSError("grants store unavailable") + + backend = build_backend(PlatformConfig(apps=[_gated_app()])) + backend.add_middleware(_FakeSession) + setattr(backend.state, access.GRANTS_STATE_ATTR, broken) + client = TestClient(backend) + assert _listed(client, STATIC) == ["gated"] + assert _listed(client, GRANTED) == [] + + +def test_no_resolver_means_static_allowlist_only(): + """Without an auth plugin there are no grants; behaviour is unchanged.""" + backend = build_backend(PlatformConfig(apps=[_gated_app()])) + backend.add_middleware(_FakeSession) + client = TestClient(backend) + assert _listed(client, STATIC) == ["gated"] + assert _listed(client, GRANTED) == [] + + +@pytest.mark.parametrize( + "static, granted", + list(itertools.product([(), ("a@x.com",)], [(), ("a@x.com",), ("b@x.com",)])), +) +def test_is_user_allowed_truth_table(static, granted): + """Empty union ⇒ open to any signed-in user; otherwise membership in the union.""" + union = set(static) | set(granted) + expected = (not union) or ("a@x.com" in union) + assert ( + access.is_user_allowed( + "a@x.com", "a@x.com", allowed_users=static, granted=granted + ) + is expected + ) + assert ( + access.is_user_allowed(None, None, allowed_users=static, granted=granted) + is False + ) + + +@pytest.mark.parametrize( + "level, expected", + [ + ("public", True), + ("local", True), + ("protected:shared", True), + ("protected:user", False), + ("something-else", False), + ], +) +def test_can_see_app_by_access_level_for_anonymous(level, expected): + """Anonymous callers see open and shared-password apps, never user-gated ones.""" + assert access.can_see_app(level, None, None) is expected + + +def test_builtin_index_lists_only_what_the_caller_may_open(): + """The HTML index (no landing app) uses the launcher's predicate too.""" + public = AppConfig( + name="open_one", + route_prefix="/api/open_one", + app_type="frontend_only", + access="public", + ) + client = _client([public, _gated_app()], {"gated": {GRANTED}}) + assert "Gated" not in client.get("/").text + assert "Open One" in client.get("/").text + assert "Gated" in client.get("/", headers={"x-test-user": GRANTED}).text