Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .claude/CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
105 changes: 105 additions & 0 deletions enlace/access.py
Original file line number Diff line number Diff line change
@@ -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
2 changes: 1 addition & 1 deletion enlace/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
63 changes: 32 additions & 31 deletions enlace/compose.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]]:
Expand Down Expand Up @@ -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))
Expand All @@ -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

Expand Down Expand Up @@ -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()
Expand Down
182 changes: 182 additions & 0 deletions enlace/tests/test_access.py
Original file line number Diff line number Diff line change
@@ -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
Loading