Add trusted header auth and publisher shell

README.md

@@ -42,7 +42,11 @@ In `--dev-mode`, requests under `/feeds/...` are served from `out/feeds/...`.
 
 In production, do not rely on Quart to serve published feeds. Configure the reverse proxy to serve `out/feeds/...` directly at `/feeds/...`.
 
-Important: the admin UI has no built-in authentication. Keep it bound to localhost or put it behind a trusted network layer such as Tailscale.
+By default the UI runs with `REPUBLISHER_AUTH_MODE=disabled` for local development.
+
+For production, set `REPUBLISHER_AUTH_MODE=trusted-headers`, keep the app bound to `127.0.0.1`, and put it behind nginx plus oauth2-proxy.
+
+In trusted-header mode, nginx must overwrite the `X-Republisher-*` identity headers before proxying to the app.
 
 Once the UI is running:
 
@@ -84,8 +88,8 @@ uv run repub cleanup-media --feeds-dir out/feeds --days 25 --dry-run
 ```
 
 - Remove `--dry-run` to delete matching files. The command protects media
-  referenced by the latest published feed and uses a lock to avoid racing with
+  referenced by the latest published feed, lists each matched file before the
-  active crawls.
+  aggregate summary, and uses a lock to avoid racing with active crawls.
 - For config-driven deployments, pass the runtime config so cleanup uses the
   configured `out_dir` and media directory names:
 

demo/README.md

@@ -42,7 +42,8 @@ uv run repub cleanup-media --config demo/repub.toml --dry-run
 With `--config`, cleanup scans `demo/out/feeds/` and honors any
 `REPUBLISHER_*_DIR` media directory overrides in the config. Remove `--dry-run`
 to delete old unreferenced media. The default retention window is 25 days; use
-`--days N` to override it.
+`--days N` to override it. Cleanup prints each matched path before the aggregate
+summary.
 
 ## Local File Feed
 

repub/auth_headers.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Literal, cast
+
+AUTH_MODE_DISABLED = "disabled"
+AUTH_MODE_TRUSTED_HEADERS = "trusted-headers"
+AUTH_MODE_ENV = "REPUBLISHER_AUTH_MODE"
+
+AuthMode = Literal["disabled", "trusted-headers"]
+AuthRole = Literal["admin", "publisher"]
+
+ROLE_HEADER = "X-Republisher-Auth-Role"
+PROVIDER_HEADER = "X-Republisher-Auth-Provider"
+USER_HEADER = "X-Republisher-Auth-User"
+EMAIL_HEADER = "X-Republisher-Auth-Email"
+PREFERRED_USERNAME_HEADER = "X-Republisher-Auth-Preferred-Username"
+GROUPS_HEADER = "X-Republisher-Auth-Groups"
+VALID_ROLES = frozenset({"admin", "publisher"})
+
+
+@dataclass(frozen=True)
+class TrustedIdentity:
+    role: AuthRole
+    provider: str
+    user: str
+    email: str
+    preferred_username: str
+    groups: tuple[str, ...]
+
+
+def load_auth_mode(environ: Mapping[str, str] | None = None) -> AuthMode:
+    raw_mode = (environ or os.environ).get(AUTH_MODE_ENV, AUTH_MODE_DISABLED).strip()
+    if raw_mode in {AUTH_MODE_DISABLED, AUTH_MODE_TRUSTED_HEADERS}:
+        return cast(AuthMode, raw_mode)
+    raise ValueError(
+        f"Unsupported {AUTH_MODE_ENV}: {raw_mode!r}. "
+        f"Expected {AUTH_MODE_DISABLED!r} or {AUTH_MODE_TRUSTED_HEADERS!r}."
+    )
+
+
+def load_trusted_identity(headers: Mapping[str, str]) -> TrustedIdentity | None:
+    role = _read_header(headers, ROLE_HEADER)
+    if role not in VALID_ROLES:
+        return None
+
+    provider = _read_header(headers, PROVIDER_HEADER)
+    user = _read_header(headers, USER_HEADER)
+    email = _read_header(headers, EMAIL_HEADER)
+    if provider is None or user is None or email is None:
+        return None
+
+    preferred_username = _read_header(headers, PREFERRED_USERNAME_HEADER) or user
+    return TrustedIdentity(
+        role=cast(AuthRole, role),
+        provider=provider,
+        user=user,
+        email=email,
+        preferred_username=preferred_username,
+        groups=_read_groups(headers.get(GROUPS_HEADER, "")),
+    )
+
+
+def _read_header(headers: Mapping[str, str], name: str) -> str | None:
+    value = headers.get(name)
+    if value is None:
+        return None
+    stripped = value.strip()
+    return stripped or None
+
+
+def _read_groups(value: str) -> tuple[str, ...]:
+    return tuple(
+        group for group in (part.strip() for part in value.split(",")) if group
+    )

repub/cleanup.py

@@ -160,6 +160,10 @@ def cleanup_media(
             if path.resolve() in protected:
                 continue
             result.matched_files += 1
+            print(
+                f"media cleanup: matched path={path.resolve()} bytes={stat.st_size}",
+                file=output,
+            )
             if dry_run:
                 continue
             try:

repub/jobs.py

@@ -166,7 +166,7 @@ class JobRuntime:
             worker = self._workers.pop(execution_id)
             if worker.process.poll() is None:
                 worker.process.kill()
-                worker.process.wait(timeout=2)
+                worker.process.wait(timeout=10)
             worker.log_handle.close()
 
         if self._started:

repub/pages/__init__.py

@@ -1,4 +1,5 @@
 from repub.pages.dashboard import dashboard_page, dashboard_page_with_data
+from repub.pages.publisher import publisher_page
 from repub.pages.runs import execution_logs_page, runs_page
 from repub.pages.settings import settings_page
 from repub.pages.shim import shim_page
@@ -10,6 +11,7 @@ __all__ = [
     "dashboard_page_with_data",
     "edit_source_page",
     "execution_logs_page",
+    "publisher_page",
     "runs_page",
     "settings_page",
     "shim_page",

repub/pages/publisher.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+import htpy as h
+from htpy import Renderable
+
+from repub.components import app_shell
+
+
+def publisher_page(*, current_path: str) -> Renderable:
+    return app_shell(
+        current_path=current_path,
+        content=(
+            h.section[
+                h.h1(class_="text-3xl font-semibold tracking-tight text-slate-950")[
+                    "Hello publishers"
+                ]
+            ],
+        ),
+    )

repub/web.py

@@ -2,11 +2,12 @@ from __future__ import annotations
 
 import asyncio
 import hashlib
-from collections.abc import AsyncGenerator, Awaitable, Callable
+from collections.abc import AsyncGenerator, Awaitable, Callable, Mapping, Sequence
 from contextlib import suppress
 from datetime import timedelta
+from functools import wraps
 from pathlib import Path
-from typing import TypedDict, cast
+from typing import Any, TypedDict, cast
 from urllib.parse import urlparse
 
 import htpy as h
@@ -17,6 +18,13 @@ from htpy import Renderable
 from peewee import IntegrityError
 from quart import Quart, Response, request, send_from_directory, url_for
 
+from repub.auth_headers import (
+    AUTH_MODE_DISABLED,
+    AUTH_MODE_TRUSTED_HEADERS,
+    AuthRole,
+    load_auth_mode,
+    load_trusted_identity,
+)
 from repub.datastar import RefreshBroker, TabStateStore, render_stream
 from repub.jobs import (
     COMPLETED_EXECUTION_PAGE_SIZE,
@@ -44,6 +52,7 @@ from repub.pages import (
     dashboard_page_with_data,
     edit_source_page,
     execution_logs_page,
+    publisher_page,
     runs_page,
     settings_page,
     shim_page,
@@ -143,11 +152,14 @@ def create_app(*, dev_mode: bool = False) -> Quart:
     app.config.setdefault("REPUB_LOG_DIR", DEFAULT_LOG_DIR)
     app.config.setdefault("REPUB_FEEDS_DIR", DEFAULT_FEEDS_DIR)
     app.config["REPUB_DEV_MODE"] = dev_mode
+    app.config["REPUB_AUTH_MODE"] = load_auth_mode()
     app.extensions[REFRESH_BROKER_KEY] = RefreshBroker()
     app.extensions[JOB_RUNTIME_KEY] = None
     app.extensions[TAB_STATE_STORE_KEY] = TabStateStore()
     app.extensions[TAB_STATE_CLEANER_TASK_KEY] = None
     app.extensions[SHUTDOWN_EVENT_KEY] = None
+    admin_required = _require_role(app, "admin")
+    publisher_required = _require_role(app, "publisher")
 
     @app.get("/feeds/<path:feed_path>")
     async def published_feed(feed_path: str) -> Response:
@@ -177,6 +189,16 @@ def create_app(*, dev_mode: bool = False) -> Quart:
         response = await send_from_directory(str(STATIC_DIR), requested_filename)
         return response
 
+    @app.get("/publisher")
+    @publisher_required
+    async def publisher_home() -> Response:
+        return _shim_page_response(current_path="/publisher")
+
+    @app.get("/admin/publisher")
+    @admin_required
+    async def admin_publisher_home() -> Response:
+        return _shim_page_response(current_path="/admin/publisher")
+
     @app.get("/")
     @app.get("/sources")
     @app.get("/sources/create")
@@ -184,51 +206,62 @@ def create_app(*, dev_mode: bool = False) -> Quart:
     @app.get("/runs")
     @app.get("/settings")
     @app.get("/job/<int:job_id>/execution/<int:execution_id>/logs")
+    @admin_required
     async def page_shim(
         slug: str | None = None,
         job_id: int | None = None,
         execution_id: int | None = None,
     ) -> Response:
         del slug, job_id, execution_id
-        body, etag = _render_shim_page(
+        return _shim_page_response(current_path=request.path)
-            stylesheet_href=versioned_static_asset_href("app.css"),
-            datastar_src=url_for("static", filename="datastar@1.0.0-RC.8.js"),
-            current_path=request.path,
-        )
-        if request.if_none_match.contains(etag):
-            response = Response(status=304)
-            response.set_etag(etag)
-            return response
-
-        response = Response(body, mimetype="text/html")
-        response.set_etag(etag)
-        return response
 
     @app.post("/")
+    @admin_required
     async def dashboard_patch() -> DatastarResponse:
         return await _page_patch_response(app, lambda _tab_id: render_dashboard(app))
 
+    @app.post("/publisher")
+    @publisher_required
+    async def publisher_patch() -> DatastarResponse:
+        return await _page_patch_response(
+            app,
+            lambda _tab_id: render_publisher(current_path="/publisher"),
+        )
+
+    @app.post("/admin/publisher")
+    @admin_required
+    async def admin_publisher_patch() -> DatastarResponse:
+        return await _page_patch_response(
+            app,
+            lambda _tab_id: render_publisher(current_path="/admin/publisher"),
+        )
+
     @app.post("/sources")
+    @admin_required
     async def sources_patch() -> DatastarResponse:
         return await _page_patch_response(app, lambda _tab_id: render_sources(app))
 
     @app.post("/sources/create")
+    @admin_required
     async def create_source_patch() -> DatastarResponse:
         return await _page_patch_response(
             app, lambda _tab_id: render_create_source(app)
         )
 
     @app.post("/sources/<string:slug>/edit")
+    @admin_required
     async def edit_source_patch(slug: str) -> DatastarResponse:
         return await _page_patch_response(
             app, lambda _tab_id: render_edit_source(slug, app)
         )
 
     @app.post("/settings")
+    @admin_required
     async def settings_patch() -> DatastarResponse:
         return await _page_patch_response(app, lambda _tab_id: render_settings(app))
 
     @app.post("/actions/sources/create")
+    @admin_required
     async def create_source_action() -> DatastarResponse:
         signals = cast(dict[str, object], await read_signals())
         source, error = validate_source_form(
@@ -254,6 +287,7 @@ def create_app(*, dev_mode: bool = False) -> Quart:
         return DatastarResponse(SSE.redirect("/sources"))
 
     @app.post("/actions/sources/<string:slug>/edit")
+    @admin_required
     async def edit_source_action(slug: str) -> DatastarResponse:
         signals = cast(dict[str, object], await read_signals())
         source, error = validate_source_form(
@@ -279,6 +313,7 @@ def create_app(*, dev_mode: bool = False) -> Quart:
         return DatastarResponse(SSE.redirect("/sources"))
 
     @app.post("/actions/sources/<string:slug>/delete")
+    @admin_required
     async def delete_source_action(slug: str) -> Response:
         delete_source(slug)
         get_job_runtime(app).sync_jobs()
@@ -286,6 +321,7 @@ def create_app(*, dev_mode: bool = False) -> Quart:
         return Response(status=204)
 
     @app.post("/actions/settings")
+    @admin_required
     async def update_settings_action() -> DatastarResponse:
         signals = cast(dict[str, object], await read_signals())
         settings, error = validate_settings_form(signals)
@@ -301,6 +337,7 @@ def create_app(*, dev_mode: bool = False) -> Quart:
         return DatastarResponse(SSE.redirect("/settings"))
 
     @app.post("/runs")
+    @admin_required
     async def runs_patch() -> DatastarResponse:
         return await _page_patch_response(
             app,
@@ -308,6 +345,7 @@ def create_app(*, dev_mode: bool = False) -> Quart:
         )
 
     @app.post("/actions/runs/completed-page/<int:page>")
+    @admin_required
     async def set_completed_runs_page_action(page: int) -> Response:
         signals = await _read_optional_signals()
         tab_id = _read_tab_id(signals)
@@ -322,12 +360,14 @@ def create_app(*, dev_mode: bool = False) -> Quart:
         return Response(status=204)
 
     @app.post("/actions/jobs/<int:job_id>/run-now")
+    @admin_required
     async def run_job_now_action(job_id: int) -> Response:
         get_job_runtime(app).run_job_now(job_id, reason="manual")
         trigger_refresh(app)
         return Response(status=204)
 
     @app.post("/actions/jobs/<int:job_id>/toggle-enabled")
+    @admin_required
     async def toggle_job_enabled_action(job_id: int) -> Response:
         enabled = load_job_enabled(job_id)
         if enabled is not None:
@@ -336,6 +376,7 @@ def create_app(*, dev_mode: bool = False) -> Quart:
         return Response(status=204)
 
     @app.post("/actions/jobs/<int:job_id>/delete")
+    @admin_required
     async def delete_job_action(job_id: int) -> Response:
         delete_job_source(job_id)
         get_job_runtime(app).sync_jobs()
@@ -343,34 +384,40 @@ def create_app(*, dev_mode: bool = False) -> Quart:
         return Response(status=204)
 
     @app.post("/actions/executions/<int:execution_id>/cancel")
+    @admin_required
     async def cancel_execution_action(execution_id: int) -> Response:
         get_job_runtime(app).request_execution_cancel(execution_id)
         trigger_refresh(app)
         return Response(status=204)
 
     @app.post("/actions/queued-executions/<int:execution_id>/cancel")
+    @admin_required
     async def cancel_queued_execution_action(execution_id: int) -> Response:
         get_job_runtime(app).cancel_queued_execution(execution_id)
         trigger_refresh(app)
         return Response(status=204)
 
     @app.post("/actions/queued-executions/<int:execution_id>/move-up")
+    @admin_required
     async def move_queued_execution_up_action(execution_id: int) -> Response:
         get_job_runtime(app).move_queued_execution(execution_id, direction="up")
         return Response(status=204)
 
     @app.post("/actions/queued-executions/<int:execution_id>/move-down")
+    @admin_required
     async def move_queued_execution_down_action(execution_id: int) -> Response:
         get_job_runtime(app).move_queued_execution(execution_id, direction="down")
         return Response(status=204)
 
     @app.post("/actions/completed-executions/clear")
+    @admin_required
     async def clear_completed_executions_action() -> Response:
         clear_completed_executions(log_dir=app.config["REPUB_LOG_DIR"])
         trigger_refresh(app)
         return Response(status=204)
 
     @app.post("/job/<int:job_id>/execution/<int:execution_id>/logs")
+    @admin_required
     async def logs_patch(job_id: int, execution_id: int) -> DatastarResponse:
         async def render() -> Renderable:
             return await render_execution_logs(
@@ -400,6 +447,55 @@ def create_app(*, dev_mode: bool = False) -> Quart:
     return app
 
 
+def _shim_page_response(*, current_path: str) -> Response:
+    body, etag = _render_shim_page(
+        stylesheet_href=versioned_static_asset_href("app.css"),
+        datastar_src=url_for("static", filename="datastar@1.0.0-RC.8.js"),
+        current_path=current_path,
+    )
+    if request.if_none_match.contains(etag):
+        response = Response(status=304)
+        response.set_etag(etag)
+        return response
+
+    response = Response(body, mimetype="text/html")
+    response.set_etag(etag)
+    return response
+
+
+def _require_role(
+    app: Quart, *roles: AuthRole
+) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
+    def decorate(
+        handler: Callable[..., Awaitable[Any]],
+    ) -> Callable[..., Awaitable[Any]]:
+        @wraps(handler)
+        async def wrapped(*args: object, **kwargs: object) -> Any:
+            failure = _authorization_failure(app, roles)
+            if failure is not None:
+                return failure
+            return await handler(*args, **kwargs)
+
+        return wrapped
+
+    return decorate
+
+
+def _authorization_failure(app: Quart, roles: Sequence[AuthRole]) -> Response | None:
+    auth_mode = cast(str, app.config["REPUB_AUTH_MODE"])
+    if auth_mode == AUTH_MODE_DISABLED:
+        return None
+    if auth_mode != AUTH_MODE_TRUSTED_HEADERS:
+        return Response(status=401)
+
+    identity = load_trusted_identity(cast(Mapping[str, str], request.headers))
+    if identity is None:
+        return Response(status=401)
+    if identity.role not in roles:
+        return Response(status=403)
+    return None
+
+
 def get_refresh_broker(app: Quart) -> RefreshBroker:
     return cast(RefreshBroker, app.extensions[REFRESH_BROKER_KEY])
 
@@ -442,6 +538,10 @@ async def render_dashboard(app: Quart | None = None) -> Renderable:
     )
 
 
+async def render_publisher(*, current_path: str) -> Renderable:
+    return publisher_page(current_path=current_path)
+
+
 async def render_sources(app: Quart | None = None) -> Renderable:
     if app is None:
         return sources_page()

tests/test_cleanup.py

@@ -102,6 +102,29 @@ def test_cleanup_media_dry_run_reports_matches_without_deleting(tmp_path: Path)
     assert result.failures == 0
 
 
+def test_cleanup_media_lists_matched_files_before_summary(tmp_path: Path) -> None:
+    feeds_dir = tmp_path / "feeds"
+    old_file = feeds_dir / "demo" / "audio" / "old.mp3"
+    write_media(old_file, b"audio", age_days=40)
+    output = io.StringIO()
+
+    result = cleanup_media(
+        feeds_dir=feeds_dir,
+        retention_days=25,
+        now=NOW,
+        dry_run=True,
+        output=output,
+    )
+
+    assert old_file.exists()
+    assert result.matched_files == 1
+    output_lines = output.getvalue().splitlines()
+    assert (
+        output_lines[0] == f"media cleanup: matched path={old_file.resolve()} bytes=5"
+    )
+    assert "matched_files=1" in output_lines[-1]
+
+
 def test_cleanup_media_uses_configured_media_dirs(tmp_path: Path) -> None:
     feeds_dir = tmp_path / "feeds"
     demo_dir = feeds_dir / "demo"

tests/test_header_auth.py

@@ -0,0 +1,344 @@
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from typing import Any, cast
+
+from repub.auth_headers import load_trusted_identity
+from repub.web import create_app
+
+
+def _trusted_headers(*, role: str, provider: str | None = None) -> dict[str, str]:
+    resolved_provider = provider or ("gp" if role == "admin" else "ocb")
+    return {
+        "X-Republisher-Auth-Role": role,
+        "X-Republisher-Auth-Provider": resolved_provider,
+        "X-Republisher-Auth-User": f"{role}-user",
+        "X-Republisher-Auth-Email": f"{role}@example.org",
+        "X-Republisher-Auth-Preferred-Username": f"{role}-user",
+        "X-Republisher-Auth-Groups": (
+            "/ocb-republisher-admins, /staff"
+            if role == "admin"
+            else "/ocb-republisher-publishers, /publishers"
+        ),
+    }
+
+
+def _configure_trusted_auth(monkeypatch, tmp_path: Path, name: str) -> None:
+    monkeypatch.setenv("REPUBLISHER_AUTH_MODE", "trusted-headers")
+    monkeypatch.setenv("REPUBLISHER_DB_PATH", str(tmp_path / f"{name}.db"))
+
+
+def _assert_datastar_shell(body: str) -> None:
+    assert body.startswith("<!doctype html>")
+    assert 'id="js"' in body
+    assert 'src="/static/datastar@1.0.0-RC.8.js"' in body
+    assert 'data-init="@post(window.location.pathname +' in body
+    assert '<main id="morph"' in body
+    assert "Connecting" in body
+
+
+def test_load_trusted_identity_parses_groups_and_defaults_preferred_username() -> None:
+    identity = load_trusted_identity(
+        {
+            "X-Republisher-Auth-Role": "admin",
+            "X-Republisher-Auth-Provider": "gp",
+            "X-Republisher-Auth-User": "abel",
+            "X-Republisher-Auth-Email": "abel@example.org",
+            "X-Republisher-Auth-Groups": " /staff, ,/ocb-republisher-admins ",
+        }
+    )
+
+    assert identity is not None
+    assert identity.preferred_username == "abel"
+    assert identity.groups == ("/staff", "/ocb-republisher-admins")
+
+
+def test_trusted_header_mode_rejects_admin_route_without_identity(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "missing-identity")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get("/")
+
+        assert response.status_code == 401
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_ignores_generic_forwarded_identity_headers(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "generic-forwarded")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get(
+            "/",
+            headers={
+                "X-Forwarded-User": "mallory",
+                "X-Forwarded-Email": "mallory@example.org",
+                "X-Forwarded-Groups": "/ocb-republisher-admins",
+            },
+        )
+
+        assert response.status_code == 401
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_rejects_malformed_trusted_identity_headers(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "malformed-identity")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get(
+            "/",
+            headers={
+                "X-Republisher-Auth-Role": "admin",
+                "X-Republisher-Auth-Provider": "gp",
+            },
+        )
+
+        assert response.status_code == 401
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_allows_admin_identity_on_admin_route(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "admin-allowed")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get("/", headers=_trusted_headers(role="admin"))
+
+        assert response.status_code == 200
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_rejects_publisher_identity_on_admin_route(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "publisher-admin-rejected")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get("/", headers=_trusted_headers(role="publisher"))
+
+        assert response.status_code == 403
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_rejects_admin_action_without_identity(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "action-missing-identity")
+
+    async def run() -> None:
+        app = create_app()
+        app.config["REPUB_LOG_DIR"] = tmp_path / "logs"
+        client = app.test_client()
+
+        response = await client.post("/actions/completed-executions/clear")
+
+        assert response.status_code == 401
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_rejects_publisher_identity_on_admin_action(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "action-publisher-rejected")
+
+    async def run() -> None:
+        app = create_app()
+        app.config["REPUB_LOG_DIR"] = tmp_path / "logs"
+        client = app.test_client()
+
+        response = await client.post(
+            "/actions/completed-executions/clear",
+            headers=_trusted_headers(role="publisher"),
+        )
+
+        assert response.status_code == 403
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_allows_publisher_identity_on_publisher_route(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "publisher-allowed")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get(
+            "/publisher",
+            headers=_trusted_headers(role="publisher"),
+        )
+        body = await response.get_data(as_text=True)
+
+        assert response.status_code == 200
+        _assert_datastar_shell(body)
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_publisher_post_serves_hello_publishers_morph(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "publisher-post")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        async with client.request(
+            "/publisher?u=shim",
+            method="POST",
+            headers=_trusted_headers(role="publisher"),
+        ) as connection:
+            await connection.send_complete()
+            chunk = await asyncio.wait_for(connection.receive(), timeout=1)
+            raw_connection = cast(Any, connection)
+
+            assert raw_connection.status_code == 200
+            assert raw_connection.headers["Content-Type"] == "text/event-stream"
+            assert b"event: datastar-patch-elements" in chunk
+            assert b'<main id="morph"' in chunk
+            assert b"Hello publishers" in chunk
+            await connection.disconnect()
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_rejects_admin_identity_on_publisher_route(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "admin-publisher-rejected")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get(
+            "/publisher", headers=_trusted_headers(role="admin")
+        )
+
+        assert response.status_code == 403
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_allows_admin_identity_on_admin_publisher_alias(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "admin-alias-allowed")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get(
+            "/admin/publisher",
+            headers=_trusted_headers(role="admin"),
+        )
+        body = await response.get_data(as_text=True)
+
+        assert response.status_code == 200
+        _assert_datastar_shell(body)
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_admin_publisher_post_serves_hello_publishers_morph(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "admin-alias-post")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        async with client.request(
+            "/admin/publisher?u=shim",
+            method="POST",
+            headers=_trusted_headers(role="admin"),
+        ) as connection:
+            await connection.send_complete()
+            chunk = await asyncio.wait_for(connection.receive(), timeout=1)
+            raw_connection = cast(Any, connection)
+
+            assert raw_connection.status_code == 200
+            assert raw_connection.headers["Content-Type"] == "text/event-stream"
+            assert b"event: datastar-patch-elements" in chunk
+            assert b'<main id="morph"' in chunk
+            assert b"Hello publishers" in chunk
+            await connection.disconnect()
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_rejects_publisher_identity_on_admin_publisher_alias(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "publisher-alias-rejected")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get(
+            "/admin/publisher",
+            headers=_trusted_headers(role="publisher"),
+        )
+
+        assert response.status_code == 403
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_keeps_static_assets_public(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "static-public")
+
+    async def run() -> None:
+        client = create_app().test_client()
+
+        response = await client.get("/static/datastar@1.0.0-RC.8.js")
+
+        assert response.status_code == 200
+
+    asyncio.run(run())
+
+
+def test_trusted_header_mode_keeps_dev_feeds_public(
+    monkeypatch, tmp_path: Path
+) -> None:
+    _configure_trusted_auth(monkeypatch, tmp_path, "feeds-public")
+
+    async def run() -> None:
+        feeds_dir = tmp_path / "out" / "feeds"
+        feed_path = feeds_dir / "demo-source" / "feed.rss"
+        feed_path.parent.mkdir(parents=True)
+        feed_path.write_text("<rss/>\n", encoding="utf-8")
+        app = create_app(dev_mode=True)
+        app.config["REPUB_FEEDS_DIR"] = feeds_dir
+        client = app.test_client()
+
+        response = await client.get("/feeds/demo-source/feed.rss")
+
+        assert response.status_code == 200
+
+    asyncio.run(run())

tests/test_scheduler_runtime.py

@@ -781,7 +781,9 @@ def test_job_runtime_start_reattaches_live_worker_after_app_restart(
             assert running_execution.running_status == JobExecutionStatus.RUNNING
             assert running_execution.ended_at is None
 
-            completed_execution = _wait_for_terminal_execution(int(execution.get_id()))
+            completed_execution = _wait_for_terminal_execution(
+                int(execution.get_id()), timeout_seconds=10.0
+            )
             assert completed_execution.running_status == JobExecutionStatus.SUCCEEDED
             assert "reattached" in artifacts.log_path.read_text(encoding="utf-8")
         finally:
@@ -861,7 +863,9 @@ def test_job_runtime_start_restores_live_worker_marked_failed_by_restart_bug(
             assert restored_execution.running_status == JobExecutionStatus.RUNNING
             assert restored_execution.ended_at is None
 
-            completed_execution = _wait_for_terminal_execution(int(execution.get_id()))
+            completed_execution = _wait_for_terminal_execution(
+                int(execution.get_id()), timeout_seconds=10.0
+            )
             assert completed_execution.running_status == JobExecutionStatus.SUCCEEDED
             assert "restored execution state" in artifacts.log_path.read_text(
                 encoding="utf-8"