+
+
+
+
+
+ Author HTML or plain text with Jinja2 placeholders. The banner renders inside the /setup page —
+ HTML tags are allowed. Leave empty and click Remove banner to go back to no banner.
+ {{ "{{ user }}" }} may be null for anonymous visitors — guard with {% if user %}.
+
+
+
+ Available placeholders
+
+ {{ "{{ instance.name }}" }} — instance display name
+{{ "{{ instance.subtitle }}" }} — operator name
+{{ "{{ server.url }}" }} — full server URL
+{{ "{{ server.hostname }}" }} — host part only
+{{ "{{ user.email }}" }}, {{ "{{ user.name }}" }}, {{ "{{ user.is_admin }}" }}
+ — null for anonymous visitors; guard with {% if user %}
+{{ "{{ now }}" }}, {{ "{{ today }}" }}
+ Copy
+
+
+
+
+
+
+ Remove banner
+ Save banner
+
+
+
+ {% if banner_html %}
+ {{ banner_html | safe }}
+ {% endif %}
+
Getting started
diff --git a/src/db.py b/src/db.py
index adb0cf0..0a521ed 100644
--- a/src/db.py
+++ b/src/db.py
@@ -39,7 +39,7 @@ def _maybe_instrument(con, db_tag: str):
_SAFE_IDENTIFIER = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]{0,63}$")
-SCHEMA_VERSION = 21
+SCHEMA_VERSION = 22
_SYSTEM_SCHEMA = """
CREATE TABLE IF NOT EXISTS schema_version (
@@ -417,6 +417,16 @@ CREATE TABLE IF NOT EXISTS welcome_template (
updated_by VARCHAR,
CONSTRAINT singleton CHECK (id = 1)
);
+
+-- v22: customizable banner shown above setup commands on /setup page.
+-- Singleton row (id=1). NULL content means "no banner".
+CREATE TABLE IF NOT EXISTS setup_banner (
+ id INTEGER PRIMARY KEY DEFAULT 1,
+ content TEXT,
+ updated_at TIMESTAMP,
+ updated_by VARCHAR,
+ CONSTRAINT singleton CHECK (id = 1)
+);
"""
@@ -1637,6 +1647,17 @@ _V20_TO_V21_MIGRATIONS = [
"INSERT INTO welcome_template (id, content) VALUES (1, NULL) ON CONFLICT (id) DO NOTHING",
]
+_V21_TO_V22_MIGRATIONS = [
+ """CREATE TABLE IF NOT EXISTS setup_banner (
+ id INTEGER PRIMARY KEY DEFAULT 1,
+ content TEXT,
+ updated_at TIMESTAMP,
+ updated_by VARCHAR,
+ CONSTRAINT singleton CHECK (id = 1)
+ )""",
+ "INSERT INTO setup_banner (id, content) VALUES (1, NULL) ON CONFLICT (id) DO NOTHING",
+]
+
def _ensure_schema(conn: duckdb.DuckDBPyConnection) -> None:
"""Create tables if they don't exist. Apply migrations if schema version changed.
@@ -1699,6 +1720,10 @@ def _ensure_schema(conn: duckdb.DuckDBPyConnection) -> None:
"INSERT INTO welcome_template (id, content) VALUES (1, NULL) "
"ON CONFLICT (id) DO NOTHING"
)
+ conn.execute(
+ "INSERT INTO setup_banner (id, content) VALUES (1, NULL) "
+ "ON CONFLICT (id) DO NOTHING"
+ )
# Fresh-install seed is handled by the unconditional
# _seed_core_roles call at the bottom of _ensure_schema —
# left as a no-op branch here so the migration ladder still
@@ -1779,6 +1804,9 @@ def _ensure_schema(conn: duckdb.DuckDBPyConnection) -> None:
if current < 21:
for sql in _V20_TO_V21_MIGRATIONS:
conn.execute(sql)
+ if current < 22:
+ for sql in _V21_TO_V22_MIGRATIONS:
+ conn.execute(sql)
conn.execute(
"UPDATE schema_version SET version = ?, applied_at = current_timestamp",
[SCHEMA_VERSION],
diff --git a/src/repositories/setup_banner.py b/src/repositories/setup_banner.py
new file mode 100644
index 0000000..5443754
--- /dev/null
+++ b/src/repositories/setup_banner.py
@@ -0,0 +1,53 @@
+"""Repository for the per-instance setup-page banner override (singleton row)."""
+
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+import duckdb
+
+
+class SetupBannerRepository:
+ def __init__(self, conn: duckdb.DuckDBPyConnection):
+ self.conn = conn
+
+ def get(self) -> dict[str, Any]:
+ """Return the singleton row. Always exists post-migration; content
+ is None when no banner is set."""
+ row = self.conn.execute(
+ "SELECT id, content, updated_at, updated_by FROM setup_banner WHERE id = 1"
+ ).fetchone()
+ if row is None:
+ # Defensive: re-seed if a previous admin manually deleted it.
+ self.conn.execute(
+ "INSERT INTO setup_banner (id, content) VALUES (1, NULL) "
+ "ON CONFLICT (id) DO NOTHING"
+ )
+ return {"id": 1, "content": None, "updated_at": None, "updated_by": None}
+ return {
+ "id": row[0],
+ "content": row[1],
+ "updated_at": row[2],
+ "updated_by": row[3],
+ }
+
+ def set(self, content: str, *, updated_by: str) -> None:
+ now = datetime.now(timezone.utc)
+ self.conn.execute(
+ """INSERT INTO setup_banner (id, content, updated_at, updated_by)
+ VALUES (1, ?, ?, ?)
+ ON CONFLICT (id) DO UPDATE SET
+ content = excluded.content,
+ updated_at = excluded.updated_at,
+ updated_by = excluded.updated_by""",
+ [content, now, updated_by],
+ )
+
+ def reset(self, *, updated_by: str) -> None:
+ """Clear the banner; /setup will show no banner."""
+ now = datetime.now(timezone.utc)
+ self.conn.execute(
+ """UPDATE setup_banner
+ SET content = NULL, updated_at = ?, updated_by = ?
+ WHERE id = 1""",
+ [now, updated_by],
+ )
diff --git a/src/setup_banner.py b/src/setup_banner.py
new file mode 100644
index 0000000..e4d11a1
--- /dev/null
+++ b/src/setup_banner.py
@@ -0,0 +1,85 @@
+"""Render the admin-editable setup-page banner.
+
+Smaller surface than welcome_template — only instance/server/user context.
+Setup banner is for organization-specific operational notes (VPN, support,
+data classification), not for analyst-side content.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime, timezone
+from typing import Any, Optional
+from urllib.parse import urlparse
+
+import duckdb
+from jinja2 import Environment, StrictUndefined, TemplateError
+
+from app.instance_config import get_instance_name, get_instance_subtitle
+from src.repositories.setup_banner import SetupBannerRepository
+
+_logger = logging.getLogger(__name__)
+
+
+def build_setup_banner_context(
+ *,
+ user: Optional[dict],
+ server_url: str,
+) -> dict[str, Any]:
+ """Compose the Jinja2 render context for the setup banner.
+
+ ``user`` may be None on the anonymous path of /setup (the page is partly
+ public — anonymous visitors get the curl-install one-liner). Templates
+ must guard for that with ``{% if user %}``.
+ """
+ parsed = urlparse(server_url)
+ return {
+ "instance": {
+ "name": get_instance_name(),
+ "subtitle": get_instance_subtitle(),
+ },
+ "server": {
+ "url": server_url,
+ "hostname": parsed.hostname or "",
+ },
+ "user": (
+ {
+ "id": user.get("id", ""),
+ "email": user.get("email", ""),
+ "name": user.get("name") or "",
+ "is_admin": bool(user.get("is_admin")),
+ }
+ if user
+ else None
+ ),
+ "now": datetime.now(timezone.utc),
+ "today": date.today().isoformat(),
+ }
+
+
+def render_setup_banner(
+ conn: duckdb.DuckDBPyConnection,
+ *,
+ user: Optional[dict],
+ server_url: str,
+) -> str:
+ """Render the banner. Returns "" when no override is set or render fails.
+
+ Render failures are swallowed (logged) — a broken admin banner must NOT
+ break /setup for analysts. The /admin/setup-banner editor catches Jinja
+ errors at PUT time anyway, so this is defense-in-depth.
+ """
+ row = SetupBannerRepository(conn).get()
+ source = row.get("content")
+ if not source:
+ return ""
+ env = Environment(undefined=StrictUndefined, autoescape=True)
+ try:
+ template = env.from_string(source)
+ return template.render(**build_setup_banner_context(user=user, server_url=server_url))
+ except TemplateError:
+ _logger.warning(
+ "setup_banner render failed; returning empty banner. "
+ "Admin can fix at /admin/setup-banner."
+ )
+ return ""
diff --git a/tests/test_setup_banner_api.py b/tests/test_setup_banner_api.py
new file mode 100644
index 0000000..9e1c0aa
--- /dev/null
+++ b/tests/test_setup_banner_api.py
@@ -0,0 +1,104 @@
+"""End-to-end tests for /api/admin/setup-banner endpoints."""
+
+
+def _auth(token: str) -> dict[str, str]:
+ return {"Authorization": f"Bearer {token}"}
+
+
+def test_admin_can_set_and_clear_banner(seeded_app):
+ c = seeded_app["client"]
+ admin = _auth(seeded_app["admin_token"])
+
+ # GET initial state
+ r = c.get("/api/admin/setup-banner", headers=admin)
+ assert r.status_code == 200
+ body = r.json()
+ assert body["content"] is None
+
+ # PUT banner
+ r = c.put(
+ "/api/admin/setup-banner",
+ json={"content": "VPN required before install.
"},
+ headers=admin,
+ )
+ assert r.status_code == 200
+
+ # GET shows new content
+ r = c.get("/api/admin/setup-banner", headers=admin)
+ assert r.json()["content"] == "VPN required before install.
"
+ assert r.json()["updated_by"] is not None
+
+ # DELETE = clear
+ r = c.delete("/api/admin/setup-banner", headers=admin)
+ assert r.status_code == 204
+
+ r = c.get("/api/admin/setup-banner", headers=admin)
+ assert r.json()["content"] is None
+
+
+def test_non_admin_cannot_edit_banner(seeded_app):
+ c = seeded_app["client"]
+ analyst = _auth(seeded_app["analyst_token"])
+ r = c.put("/api/admin/setup-banner", json={"content": "x
"}, headers=analyst)
+ assert r.status_code == 403
+
+
+def test_put_rejects_invalid_jinja2(seeded_app):
+ c = seeded_app["client"]
+ admin = _auth(seeded_app["admin_token"])
+ r = c.put(
+ "/api/admin/setup-banner",
+ json={"content": "{% for x in y %}"}, # unclosed loop
+ headers=admin,
+ )
+ assert r.status_code == 400
+ assert "invalid" in r.json()["detail"].lower()
+
+
+def test_put_rejects_undefined_placeholder(seeded_app):
+ """Templates that reference unknown placeholders must be rejected at PUT
+ time so the admin sees the error immediately."""
+ c = seeded_app["client"]
+ admin = _auth(seeded_app["admin_token"])
+ r = c.put(
+ "/api/admin/setup-banner",
+ json={"content": "Hello {{ user.emial }}"}, # typo
+ headers=admin,
+ )
+ assert r.status_code == 400
+ assert "emial" in r.json()["detail"] or "undefined" in r.json()["detail"].lower()
+
+
+def test_preview_renders_arbitrary_content(seeded_app):
+ c = seeded_app["client"]
+ admin = _auth(seeded_app["admin_token"])
+ r = c.post(
+ "/api/admin/setup-banner/preview",
+ json={"content": "Hello {{ user.email }} "},
+ headers=admin,
+ )
+ assert r.status_code == 200
+ # autoescape=True: rendered content must contain the escaped or literal email
+ assert "admin@test.com" in r.json()["content"]
+
+
+def test_preview_requires_admin(seeded_app):
+ c = seeded_app["client"]
+ analyst = _auth(seeded_app["analyst_token"])
+ r = c.post(
+ "/api/admin/setup-banner/preview",
+ json={"content": "x
"},
+ headers=analyst,
+ )
+ assert r.status_code == 403
+
+
+def test_preview_rejects_invalid_template(seeded_app):
+ c = seeded_app["client"]
+ admin = _auth(seeded_app["admin_token"])
+ r = c.post(
+ "/api/admin/setup-banner/preview",
+ json={"content": "{% for x in y %}"},
+ headers=admin,
+ )
+ assert r.status_code == 400
diff --git a/tests/test_setup_banner_migration.py b/tests/test_setup_banner_migration.py
new file mode 100644
index 0000000..8b8b3d0
--- /dev/null
+++ b/tests/test_setup_banner_migration.py
@@ -0,0 +1,54 @@
+"""v21 → v22 migration: adds setup_banner singleton table."""
+
+from pathlib import Path
+
+import duckdb
+
+from src.db import SCHEMA_VERSION, _ensure_schema, get_schema_version
+
+
+def _open(path: Path) -> duckdb.DuckDBPyConnection:
+ return duckdb.connect(str(path))
+
+
+def test_v22_creates_setup_banner_table(tmp_path):
+ db_path = tmp_path / "system.duckdb"
+ conn = _open(db_path)
+ # Pretend we're on v21: run schema then roll version back.
+ _ensure_schema(conn)
+ conn.execute("UPDATE schema_version SET version = 21")
+ conn.execute("DROP TABLE IF EXISTS setup_banner")
+ conn.close()
+
+ # Re-open: migration ladder runs.
+ conn = _open(db_path)
+ _ensure_schema(conn)
+ assert get_schema_version(conn) == SCHEMA_VERSION
+ # Singleton row must exist with NULL content (= no banner).
+ rows = conn.execute(
+ "SELECT id, content, updated_at, updated_by FROM setup_banner"
+ ).fetchall()
+ assert len(rows) == 1
+ assert rows[0][0] == 1 # singleton id
+ assert rows[0][1] is None # NULL = no banner
+
+
+def test_fresh_install_seeds_setup_banner(tmp_path):
+ db_path = tmp_path / "system.duckdb"
+ conn = _open(db_path)
+ _ensure_schema(conn)
+ assert get_schema_version(conn) == SCHEMA_VERSION
+ rows = conn.execute("SELECT id, content FROM setup_banner").fetchall()
+ assert len(rows) == 1
+ assert rows[0][0] == 1
+ assert rows[0][1] is None
+
+
+def test_welcome_template_unaffected_by_v22(tmp_path):
+ """welcome_template table must still coexist after v22 migration."""
+ db_path = tmp_path / "system.duckdb"
+ conn = _open(db_path)
+ _ensure_schema(conn)
+ rows = conn.execute("SELECT id, content FROM welcome_template").fetchall()
+ assert len(rows) == 1
+ assert rows[0][0] == 1
diff --git a/tests/test_setup_banner_render.py b/tests/test_setup_banner_render.py
new file mode 100644
index 0000000..fb7c0ed
--- /dev/null
+++ b/tests/test_setup_banner_render.py
@@ -0,0 +1,80 @@
+"""Unit tests for the setup-banner renderer module."""
+
+import duckdb
+import pytest
+
+from src.db import _ensure_schema
+from src.repositories.setup_banner import SetupBannerRepository
+from src.setup_banner import build_setup_banner_context, render_setup_banner
+
+
+@pytest.fixture
+def conn(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATA_DIR", str(tmp_path))
+ db_path = tmp_path / "system.duckdb"
+ c = duckdb.connect(str(db_path))
+ _ensure_schema(c)
+ yield c
+ c.close()
+
+
+def _user(email="alice@example.com"):
+ return {"id": "u1", "email": email, "name": "Alice", "is_admin": False}
+
+
+def test_render_returns_empty_when_no_override(conn):
+ out = render_setup_banner(conn, user=_user(), server_url="https://example.com")
+ assert out == ""
+
+
+def test_render_uses_override(conn):
+ SetupBannerRepository(conn).set(
+ "VPN: {{ server.hostname }}
", updated_by="admin@example.com"
+ )
+ out = render_setup_banner(conn, user=_user(), server_url="https://example.com")
+ # autoescape=True — rendered as HTML
+ assert "example.com" in out
+ assert "" in out
+
+
+def test_render_returns_empty_on_invalid_template_does_not_raise(conn):
+ """A broken admin banner must not raise; it must return "" (defense-in-depth)."""
+ SetupBannerRepository(conn).set(
+ "{{ does_not_exist }}", updated_by="admin@example.com"
+ )
+ out = render_setup_banner(conn, user=_user(), server_url="https://example.com")
+ assert out == "" # swallowed, not raised
+
+
+def test_render_with_anonymous_user(conn):
+ SetupBannerRepository(conn).set(
+ "{% if user %}{{ user.email }}{% else %}anonymous{% endif %}",
+ updated_by="admin@example.com",
+ )
+ out = render_setup_banner(conn, user=None, server_url="https://example.com")
+ assert "anonymous" in out
+
+
+def test_context_exposes_documented_keys(conn):
+ ctx = build_setup_banner_context(user=_user(), server_url="https://example.com")
+ for top in ("instance", "server", "user", "now", "today"):
+ assert top in ctx, f"missing top-level key: {top}"
+ assert ctx["server"]["hostname"] == "example.com"
+ assert ctx["user"]["email"] == "alice@example.com"
+
+
+def test_context_with_anonymous_user_returns_none(conn):
+ ctx = build_setup_banner_context(user=None, server_url="https://example.com")
+ assert ctx["user"] is None
+
+
+def test_autoescape_escapes_html_entities(conn):
+ """autoescape=True must escape < > & in template variable output."""
+ SetupBannerRepository(conn).set(
+ "{{ server.hostname }}", updated_by="admin@example.com"
+ )
+ out = render_setup_banner(
+ conn, user=_user(), server_url="https://example.com/"
+ )
+ # hostname won't contain < > but the render must succeed without injection
+ assert out != ""
diff --git a/tests/test_setup_banner_repo.py b/tests/test_setup_banner_repo.py
new file mode 100644
index 0000000..fb421f6
--- /dev/null
+++ b/tests/test_setup_banner_repo.py
@@ -0,0 +1,49 @@
+"""Unit tests for SetupBannerRepository."""
+
+import duckdb
+import pytest
+
+from src.db import _ensure_schema
+from src.repositories.setup_banner import SetupBannerRepository
+
+
+@pytest.fixture
+def conn(tmp_path):
+ db_path = tmp_path / "system.duckdb"
+ c = duckdb.connect(str(db_path))
+ _ensure_schema(c)
+ yield c
+ c.close()
+
+
+def test_get_returns_none_on_fresh_install(conn):
+ repo = SetupBannerRepository(conn)
+ row = repo.get()
+ assert row is not None
+ assert row["content"] is None # no banner by default
+
+
+def test_set_stores_content(conn):
+ repo = SetupBannerRepository(conn)
+ repo.set("VPN required
", updated_by="admin@example.com")
+ row = repo.get()
+ assert row["content"] == "VPN required
"
+ assert row["updated_by"] == "admin@example.com"
+ assert row["updated_at"] is not None
+
+
+def test_reset_clears_content(conn):
+ repo = SetupBannerRepository(conn)
+ repo.set("Note
", updated_by="admin@example.com")
+ repo.reset(updated_by="admin@example.com")
+ row = repo.get()
+ assert row["content"] is None
+
+
+def test_set_overwrites_existing(conn):
+ repo = SetupBannerRepository(conn)
+ repo.set("first", updated_by="a@example.com")
+ repo.set("second", updated_by="b@example.com")
+ row = repo.get()
+ assert row["content"] == "second"
+ assert row["updated_by"] == "b@example.com"