diff --git a/enlace_auth/appmeta.py b/enlace_auth/appmeta.py new file mode 100644 index 0000000..c845a88 --- /dev/null +++ b/enlace_auth/appmeta.py @@ -0,0 +1,215 @@ +"""Editable app-metadata overlay: the write surface for the launcher. + +enlace core owns the *read* model for launcher metadata (title / description / +keywords / icon) — it harvests app-declared values and serves ``/_apps`` and the +icon endpoint. But the *editable* layer — an owner-curated overlay of added +keywords and icon/title/description overrides, mutated live from the launcher UI +— needs authentication, CSRF, an admin allowlist, and durable per-app storage. +Those are exactly the concerns enlace core delegates to this plugin, so the +overlay lives here, mirroring the runtime-grants pattern +(:mod:`enlace_auth.auth.grants`). + +The contract with core is two dependency-injection slots on the parent app's +``state`` (read by core, written here): + +- ``app_meta_overlay`` — a ``MutableMapping[str, dict]`` of per-app overlay + records; core reads it when resolving ``/_apps`` and the icon. +- ``app_meta_can_edit`` — ``Callable[[Optional[str]], bool]`` deciding whether + an email may edit; core surfaces the result as ``can_edit_meta``. + +Without this plugin (or on an older enlace_auth), those slots default in core to +an empty overlay and "nobody can edit", so the launcher still works read-only. + +Overlay record schema (one JSON file per app under ``/app_meta/{name}``):: + + { + "keywords": ["owner", "added", "tags"], # additive; unioned by core + "icon": "emoji:🎸" | "assets/x.png" | null, + "display_name": "Override Title" | null, + "description": "Override blurb" | null + } + +Empty records are deleted, so "no overlay" and "empty overlay" coincide. +""" + +from __future__ import annotations + +from collections.abc import MutableMapping +from typing import Callable, Optional + +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel + +# Bounds so an editor can't wedge the store with absurd input. +_MAX_KEYWORDS = 40 +_MAX_KEYWORD_LEN = 60 +_MAX_SPEC_LEN = 4096 + + +class _MetaPatchBody(BaseModel): + """PATCH body — every field optional. ``null`` on a scalar *clears* it.""" + + add_keywords: Optional[list[str]] = None + remove_keywords: Optional[list[str]] = None + icon: Optional[str] = None + display_name: Optional[str] = None + description: Optional[str] = None + + # Distinguish "field omitted" from "field set to null" so a caller can + # explicitly clear an override. Pydantic v2: model_fields_set. + model_config = {"extra": "forbid"} + + +def make_appmeta_can_edit(editors: tuple[str, ...]) -> Callable[[Optional[str]], bool]: + """Build the ``app_meta_can_edit`` closure over a normalized editor set. + + Case-insensitive; an empty editor set means *nobody* can edit (safe default, + so a misconfigured deploy fails closed rather than open). + """ + editor_set = frozenset(e.strip().lower() for e in editors if e and e.strip()) + + def can_edit(email: Optional[str]) -> bool: + if not editor_set or not email: + return False + return email.strip().lower() in editor_set + + return can_edit + + +def _norm_keywords(values) -> list[str]: + """Strip, drop blanks/oversize/non-strings, casefold-dedupe, keep first form.""" + out: list[str] = [] + seen: set[str] = set() + for v in values or (): + if not isinstance(v, str): + continue + s = v.strip() + if not s or len(s) > _MAX_KEYWORD_LEN: + continue + key = s.casefold() + if key in seen: + continue + seen.add(key) + out.append(s) + return out + + +def _clean_record(record: dict) -> dict: + """Drop empty/None fields so an emptied overlay round-trips to deletion.""" + out: dict = {} + kw = _norm_keywords(record.get("keywords")) + if kw: + out["keywords"] = kw + for key in ("icon", "display_name", "description"): + val = record.get(key) + if isinstance(val, str) and val.strip(): + out[key] = val.strip() + return out + + +def _apply_patch(current: dict, body: _MetaPatchBody) -> dict: + """Apply a PATCH to an overlay record; return the cleaned new record. + + Keywords are read-modify-write (add ∪, then remove); scalar fields set when + present (``null`` clears, a string sets). ``model_fields_set`` distinguishes + "omitted" (leave as-is) from "explicitly null" (clear). + """ + record = dict(current) + keywords = _norm_keywords(record.get("keywords")) + + if body.add_keywords: + keywords = _norm_keywords([*keywords, *body.add_keywords]) + if body.remove_keywords: + drop = {k.casefold() for k in _norm_keywords(body.remove_keywords)} + keywords = [k for k in keywords if k.casefold() not in drop] + if len(keywords) > _MAX_KEYWORDS: + raise HTTPException(400, f"Too many overlay keywords (max {_MAX_KEYWORDS}).") + record["keywords"] = keywords + + fields_set = body.model_fields_set + for key in ("icon", "display_name", "description"): + if key not in fields_set: + continue + val = getattr(body, key) + if val is None: + record.pop(key, None) # explicit clear + else: + if len(val) > _MAX_SPEC_LEN: + raise HTTPException(400, f"{key} too long.") + record[key] = val + + return _clean_record(record) + + +def make_appmeta_router( + *, + apps: list, + config, + overlay_store: MutableMapping, + can_edit: Callable[[Optional[str]], bool], +) -> APIRouter: + """Build the ``PATCH/DELETE /_apps/{name}/meta`` router (CSRF-gated by the + plugin's middleware, since ``/_apps/*`` is not in the CSRF-exempt list). + + Editor authz is enforced per request via ``can_edit``. The app name is + validated against the known-apps set (unknown ⇒ 404), which also keeps the + store key confined to a real app name — defense-in-depth over the ``[^/]+`` + path convertor that already blocks traversal. + """ + from enlace.compose import build_launcher_item + + app_by_name = {a.name: a for a in apps} + router = APIRouter() + + def _require_editor(request: Request) -> str: + email = (getattr(request.state, "user_email", None) or "").strip() + if not can_edit(email): + # Uniform 403 for anonymous / non-editor — never 500 when auth is off. + raise HTTPException(403, "Not permitted to edit app metadata.") + return email + + def _require_app(name: str): + app = app_by_name.get(name) + if app is None or name != app.name or any(c in name for c in "/\\"): + raise HTTPException(404, "Unknown app.") + return app + + def _item(app) -> dict: + try: + record = overlay_store.get(app.name, {}) + except Exception: + record = {} + return build_launcher_item( + app, config, record if isinstance(record, dict) else {} + ) + + @router.patch("/_apps/{name}/meta") + async def patch_meta(name: str, body: _MetaPatchBody, request: Request) -> dict: + _require_editor(request) + app = _require_app(name) + try: + current = overlay_store.get(name, {}) + except Exception: + current = {} + new_record = _apply_patch(current if isinstance(current, dict) else {}, body) + if new_record: + overlay_store[name] = new_record + else: + # An emptied overlay is a deletion — keep "absent" and "empty" one state. + try: + del overlay_store[name] + except KeyError: + pass + return _item(app) + + @router.delete("/_apps/{name}/meta") + async def delete_meta(name: str, request: Request) -> dict: + _require_editor(request) + app = _require_app(name) + try: + del overlay_store[name] + except KeyError: + pass + return _item(app) + + return router diff --git a/enlace_auth/plugin.py b/enlace_auth/plugin.py index b7665a8..cf888b3 100644 --- a/enlace_auth/plugin.py +++ b/enlace_auth/plugin.py @@ -316,6 +316,34 @@ def wire(parent: "FastAPI", config) -> None: ) parent.include_router(store_router) + # App-metadata overlay: the editable launcher-metadata layer (owner-added + # keywords + icon/title overrides). Core reads the overlay + can-edit closure + # via parent.state; the write surface (PATCH/DELETE /_apps/{name}/meta) is + # here so it inherits CSRF + auth. Editors default to the admin allowlist. + from enlace_auth.appmeta import make_appmeta_can_edit, make_appmeta_router + + app_meta_cfg = getattr(config, "app_meta", None) + editors = tuple(getattr(app_meta_cfg, "editors", ()) or ()) or admin_emails + if getattr(app_meta_cfg, "store_path", None): + overlay_factory = make_file_store_factory(str(app_meta_cfg.store_path)) + else: + overlay_factory = platform_factory # beside sessions/users/grants + overlay_store = overlay_factory("app_meta") + can_edit_meta = make_appmeta_can_edit(editors) + + # DI slots read by enlace core (compose._overlay_entry / apps_listing). + parent.state.app_meta_overlay = overlay_store + parent.state.app_meta_can_edit = can_edit_meta + + parent.include_router( + make_appmeta_router( + apps=list(getattr(config, "apps", [])), + config=config, + overlay_store=overlay_store, + can_edit=can_edit_meta, + ) + ) + # Admin router (API + UI). UI is only mounted when there is at least one # admin email; otherwise the dashboard would be unreachable anyway. admin_router = make_admin_router( diff --git a/pyproject.toml b/pyproject.toml index 25e99ea..8d0a004 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ requires-python = ">=3.10" keywords = ["enlace", "auth", "fastapi", "platform", "argon2", "session", "admin"] authors = [{ name = "Thor Whalen" }] dependencies = [ - "enlace>=0.1.0", + "enlace>=0.1.25", # needs build_launcher_item + app_meta DI slots "fastapi>=0.100.0", "pydantic>=2.0.0", "itsdangerous>=2.1", diff --git a/tests/test_appmeta.py b/tests/test_appmeta.py new file mode 100644 index 0000000..4be4654 --- /dev/null +++ b/tests/test_appmeta.py @@ -0,0 +1,209 @@ +"""Tests for the editable app-metadata overlay (PATCH/DELETE /_apps/{name}/meta). + +Sets up a full enlace gateway with the auth plugin and an editor allowlist, then +exercises the overlay endpoints end-to-end: editor authz, CSRF, additive keyword +edits, scalar overrides + clears, empty-record deletion, and that the resolved +``/_apps`` listing reflects the overlay. Also unit-tests the pure patch logic. +""" + +from __future__ import annotations + +import pytest +from enlace.base import PlatformConfig +from enlace.compose import build_backend +from enlace.discover import discover_apps +from starlette.testclient import TestClient + +from enlace_auth import plugin as auth_plugin +from enlace_auth.appmeta import _apply_patch, _MetaPatchBody, make_appmeta_can_edit + +_SIGNING_KEY = "appmeta-test-key-thirtytwobyteslong!" + + +# --------------------------------------------------------------------------- +# Pure unit tests: can-edit + patch logic +# --------------------------------------------------------------------------- + + +def test_can_edit_normalizes_and_fails_closed(): + can = make_appmeta_can_edit(("Boss@Example.com",)) + assert can("boss@example.com") is True + assert can("BOSS@example.com ".strip()) is True + assert can("other@example.com") is False + assert can(None) is False + # empty editor set ⇒ nobody can edit + assert make_appmeta_can_edit(())("boss@example.com") is False + + +def test_apply_patch_adds_and_removes_keywords(): + body = _MetaPatchBody(add_keywords=["Music", "music", "chords"]) + rec = _apply_patch({}, body) + assert rec["keywords"] == ["Music", "chords"] # casefold-deduped + rec2 = _apply_patch(rec, _MetaPatchBody(remove_keywords=["chords"])) + assert rec2["keywords"] == ["Music"] + + +def test_apply_patch_scalar_set_and_clear(): + rec = _apply_patch({}, _MetaPatchBody(icon="emoji:🎸", display_name="Cool")) + assert rec["icon"] == "emoji:🎸" + assert rec["display_name"] == "Cool" + # explicit null clears; omitted leaves as-is + rec2 = _apply_patch(rec, _MetaPatchBody(icon=None)) + assert "icon" not in rec2 + assert rec2["display_name"] == "Cool" + + +def test_apply_patch_empty_record_is_falsy(): + assert _apply_patch({}, _MetaPatchBody()) == {} + # removing the last keyword empties the record + rec = _apply_patch({}, _MetaPatchBody(add_keywords=["only"])) + assert _apply_patch(rec, _MetaPatchBody(remove_keywords=["only"])) == {} + + +# --------------------------------------------------------------------------- +# Integration: full gateway with the plugin +# --------------------------------------------------------------------------- + + +def _write_public_app(apps_dir, name="widget"): + app = apps_dir / name + app.mkdir() + (app / "server.py").write_text( + "from fastapi import FastAPI\napp = FastAPI()\n" + "@app.get('/ping')\ndef ping():\n return {'ok': True}\n" + ) + (app / "app.toml").write_text( + 'access = "public"\ndisplay_name = "Widget"\nkeywords = ["base"]\n' + ) + + +@pytest.fixture +def gw(tmp_path, monkeypatch): + apps_dir = tmp_path / "apps" + apps_dir.mkdir() + _write_public_app(apps_dir) + + monkeypatch.setenv("ENLACE_SIGNING_KEY", _SIGNING_KEY) + monkeypatch.setenv("ENLACE_ADMIN_EMAILS", "boss@example.com") + + config = PlatformConfig( + apps_dir=apps_dir, + auth={ + "enabled": True, + "secure_cookies": False, + "registration_open": True, + "stores": {"backend": "file", "path": str(tmp_path / "platform")}, + }, + # editors falls back to ENLACE_ADMIN_EMAILS (no [app_meta] editors set) + ) + config = discover_apps(config) + app = build_backend(config, plugins=[auth_plugin]) + return TestClient(app) + + +def _csrf(client): + client.get("/api/widget/ping") + from enlace_auth.auth.cookies import verify_cookie + + raw = verify_cookie(client.cookies.get("enlace_csrf"), _SIGNING_KEY, salt="csrf") + return {"X-CSRF-Token": raw} + + +def _register(client, email, password, csrf): + return client.post( + "/auth/register", json={"email": email, "password": password}, headers=csrf + ) + + +def test_anonymous_cannot_edit(gw): + csrf = _csrf(gw) + r = gw.patch("/_apps/widget/meta", json={"add_keywords": ["x"]}, headers=csrf) + assert r.status_code == 403 + + +def test_non_editor_cannot_edit(gw): + csrf = _csrf(gw) + _register(gw, "alice@example.com", "secretpw1", csrf) # logs alice in + r = gw.patch("/_apps/widget/meta", json={"add_keywords": ["x"]}, headers=csrf) + assert r.status_code == 403 + + +def test_patch_requires_csrf(gw): + csrf = _csrf(gw) + _register(gw, "boss@example.com", "bosspw1!", csrf) # admin -> editor + # missing X-CSRF-Token header ⇒ blocked by CSRF middleware (not 200) + r = gw.patch("/_apps/widget/meta", json={"add_keywords": ["x"]}) + assert r.status_code == 403 + + +def test_editor_adds_keywords_and_listing_reflects_it(gw): + csrf = _csrf(gw) + _register(gw, "boss@example.com", "bosspw1!", csrf) + r = gw.patch( + "/_apps/widget/meta", json={"add_keywords": ["Extra", "tag"]}, headers=csrf + ) + assert r.status_code == 200, r.text + item = r.json() + assert item["keyword_sources"]["app"] == ["base"] + assert item["keyword_sources"]["overlay"] == ["Extra", "tag"] + assert item["keywords"] == ["base", "Extra", "tag"] + + # the /_apps listing reflects the overlay too + listing = gw.get("/_apps").json() + widget = next(a for a in listing["apps"] if a["name"] == "widget") + assert widget["keywords"] == ["base", "Extra", "tag"] + assert listing["can_edit_meta"] is True + + +def test_overlay_icon_override_serves_via_icon_endpoint(gw): + csrf = _csrf(gw) + _register(gw, "boss@example.com", "bosspw1!", csrf) + gw.patch("/_apps/widget/meta", json={"icon": "emoji:⭐"}, headers=csrf) + r = gw.get("/_apps/widget/icon") + assert r.status_code == 200 + assert "⭐" in r.content.decode("utf-8") + + +def test_delete_clears_overlay(gw): + csrf = _csrf(gw) + _register(gw, "boss@example.com", "bosspw1!", csrf) + gw.patch("/_apps/widget/meta", json={"add_keywords": ["temp"]}, headers=csrf) + r = gw.delete("/_apps/widget/meta", headers=csrf) + assert r.status_code == 200 + assert r.json()["keyword_sources"]["overlay"] == [] + widget = next(a for a in gw.get("/_apps").json()["apps"] if a["name"] == "widget") + assert widget["keywords"] == ["base"] + + +def test_unknown_app_404(gw): + csrf = _csrf(gw) + _register(gw, "boss@example.com", "bosspw1!", csrf) + r = gw.patch("/_apps/nope/meta", json={"add_keywords": ["x"]}, headers=csrf) + assert r.status_code == 404 + + +def test_remove_keyword_persists_across_requests(gw): + csrf = _csrf(gw) + _register(gw, "boss@example.com", "bosspw1!", csrf) + gw.patch("/_apps/widget/meta", json={"add_keywords": ["a", "b"]}, headers=csrf) + gw.patch("/_apps/widget/meta", json={"remove_keywords": ["a"]}, headers=csrf) + widget = next(a for a in gw.get("/_apps").json()["apps"] if a["name"] == "widget") + assert widget["keyword_sources"]["overlay"] == ["b"] + + +def test_patch_item_shape_matches_listing_item(gw): + """The PATCH/DELETE response item has the IDENTICAL key set as a /_apps item. + + Both are built by enlace.compose.build_launcher_item — this pins that SSOT so + the frontend can drop a PATCH response straight into the store, and App.parse + (which validates it with the same zod schema as the listing) can't drift. + """ + csrf = _csrf(gw) + _register(gw, "boss@example.com", "bosspw1!", csrf) + listing_item = next( + a for a in gw.get("/_apps").json()["apps"] if a["name"] == "widget" + ) + patched = gw.patch( + "/_apps/widget/meta", json={"add_keywords": ["z"]}, headers=csrf + ).json() + assert set(patched.keys()) == set(listing_item.keys()) diff --git a/tests/test_recovery.py b/tests/test_recovery.py index 8599cbf..753a4a6 100644 --- a/tests/test_recovery.py +++ b/tests/test_recovery.py @@ -162,7 +162,7 @@ def test_shared_login_page_redirects_when_already_authed(): follow_redirects=False, ) assert r.status_code == 303 - assert r.headers["location"] == "/xa/" + assert r.headers["location"].endswith("/xa/") # newer httpx returns absolute def test_shared_login_post_still_returns_json():