")
+ cols_per_row = 1
+ else:
+ cols_per_row = min(len(items), 3)
+ if cols_per_row == 0:
+ cols_per_row = 1
+ parts.append(f"
")
+
+ for label, value in items:
+ parts.append(f"
")
+ parts.append(
+ f"
{html.escape(str(value))}
"
+ )
+ if not hide_labels:
+ parts.append(
+ f"
"
+ f"{html.escape(label)}
"
+ )
+ parts.append("
")
+ parts.append("
")
+
+ body = "".join(parts)
+
+ return f"""
+
+
+
+
+
+
+
+{body}
+
+"""
+
+
+@router.get("/stats", response_class=HTMLResponse)
+def get_embed_stats(
+ request: Request,
+ token: str = Query(..., description="Embed token"),
+ theme: str = Query("light", description="Theme: light|dark"),
+ accent: str = Query("#3b82f6", description="Accent hex color"),
+ radius: str = Query("md", description="Border radius: none|sm|md|lg|xl"),
+ density: str = Query("normal", description="Density: compact|normal|comfortable"),
+ hide_labels: bool = Query(False, description="Hide text labels"),
+ show: Optional[str] = Query(None, description="Comma-separated stat keys to display"),
+ lang: str = Query("en", description="HTML lang attribute"),
+ font_scale: float = Query(1.0, ge=0.5, le=3.0, description="Font size multiplier"),
+ layout: str = Query("grid", description="Layout: grid|list"),
+ session: Session = Depends(get_session),
+):
+ user = _verify_embed_token(token, session, request)
+
+ books = list(
+ session.exec(
+ select(Book).where(Book.user_id == user.id)
+ ).all()
+ )
+
+ stats = _compute_stats(books)
+
+ show_set: Optional[set[str]] = None
+ if show:
+ keys = {k.strip() for k in show.split(",") if k.strip()}
+ invalid = keys - VALID_STAT_KEYS
+ if invalid:
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ detail=f"Invalid stat keys: {', '.join(sorted(invalid))}. Valid: {', '.join(sorted(VALID_STAT_KEYS))}",
+ )
+ show_set = keys if keys else None
+
+ if layout not in LAYOUT_MODES:
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ detail=f"Invalid layout '{layout}'. Valid: {', '.join(sorted(LAYOUT_MODES))}",
+ )
+
+ html_content = _render_stats_html(
+ stats, theme, accent, radius, density, hide_labels, lang, font_scale, layout, show_set,
+ )
+
+ return Response(
+ content=html_content,
+ media_type="text/html; charset=utf-8",
+ headers={
+ "Cache-Control": "private, max-age=60",
+ "X-Content-Type-Options": "nosniff",
+ "Referrer-Policy": "no-referrer",
+ "Content-Security-Policy": "default-src 'none'; style-src 'unsafe-inline'; frame-ancestors *",
+ },
+ )
diff --git a/backend/app/routers/profile.py b/backend/app/routers/profile.py
index 256a66aa..fa6c169b 100644
--- a/backend/app/routers/profile.py
+++ b/backend/app/routers/profile.py
@@ -9,14 +9,17 @@
clear_browser_session,
ensure_password_complexity,
generate_api_key,
+ generate_embed_token,
get_api_key_prefix,
+ get_embed_token_prefix,
get_password_hash,
hash_api_key,
+ hash_embed_token,
require_user,
)
from app.config import settings as app_settings
from app.database import get_session
-from app.models import ApiKey, User, UserSettings
+from app.models import ApiKey, EmbedToken, User, UserSettings
from app.schemas import (
ApiKeyCreate,
ApiKeyCreateResponse,
@@ -24,6 +27,10 @@
ConfirmationPhrase,
DataResetDeleted,
DataResetResponse,
+ EmbedTokenCreate,
+ EmbedTokenCreateResponse,
+ EmbedTokenRead,
+ EmbedTokenUpdate,
ProfileUpdate,
UserRead,
UserSettingsRead,
@@ -94,7 +101,6 @@ def get_settings(
user_id=settings.user_id,
language=settings.language,
timezone=settings.timezone,
- quote_service_enabled=app_settings.dashboard_quote_enabled,
theme=settings.theme,
custom_theme=settings.custom_theme,
)
@@ -123,7 +129,6 @@ def update_settings(
user_id=settings.user_id,
language=settings.language,
timezone=settings.timezone,
- quote_service_enabled=app_settings.dashboard_quote_enabled,
theme=settings.theme,
custom_theme=settings.custom_theme,
)
@@ -200,7 +205,7 @@ def list_api_keys(
select(ApiKey).where(
ApiKey.user_id == current_user.id,
ApiKey.revoked_at.is_(None),
- )
+ ).order_by(ApiKey.created_at.desc())
).all()
return [ApiKeyRead.model_validate(k) for k in keys]
@@ -239,3 +244,119 @@ def delete_api_key(
key.revoked_at = utcnow()
session.add(key)
session.commit()
+
+
+@router.get("/embed-tokens", response_model=list[EmbedTokenRead])
+def list_embed_tokens(
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> list[EmbedTokenRead]:
+ """List non-revoked embed tokens for the current user."""
+ tokens = session.exec(
+ select(EmbedToken).where(
+ EmbedToken.user_id == current_user.id,
+ EmbedToken.revoked_at.is_(None),
+ ).order_by(EmbedToken.created_at.desc())
+ ).all()
+ return [EmbedTokenRead.model_validate(t) for t in tokens]
+
+
+@router.post("/embed-tokens", response_model=EmbedTokenCreateResponse, status_code=201)
+def create_embed_token(
+ body: EmbedTokenCreate,
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> EmbedTokenCreateResponse:
+ """Create a new embed token for the current user."""
+ plain_token = generate_embed_token()
+ token = EmbedToken(
+ user_id=current_user.id,
+ name=body.name,
+ token_prefix=get_embed_token_prefix(plain_token),
+ token_hash=hash_embed_token(plain_token),
+ allowed_origins=body.allowed_origins or None,
+ expires_at=body.expires_at,
+ )
+ session.add(token)
+ session.commit()
+ session.refresh(token)
+ return EmbedTokenCreateResponse(
+ token=plain_token,
+ embed_token=EmbedTokenRead.model_validate(token),
+ )
+
+
+@router.patch("/embed-tokens/{token_id}", response_model=EmbedTokenRead)
+def update_embed_token(
+ token_id: int,
+ body: EmbedTokenUpdate,
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> EmbedTokenRead:
+ """Update name, allowed_origins, or expires_at for an embed token."""
+ token = session.get(EmbedToken, token_id)
+ if not token or token.user_id != current_user.id or token.revoked_at is not None:
+ raise HTTPException(status_code=404, detail="Embed token not found")
+
+ update_data = body.model_dump(exclude_unset=True)
+ token.sqlmodel_update(update_data)
+ session.add(token)
+ session.commit()
+ session.refresh(token)
+ return EmbedTokenRead.model_validate(token)
+
+
+@router.post("/embed-tokens/{token_id}/rotate", response_model=EmbedTokenCreateResponse)
+def rotate_embed_token(
+ token_id: int,
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> EmbedTokenCreateResponse:
+ """Revoke the current embed token and create a new one with the same settings."""
+ token = session.get(EmbedToken, token_id)
+ if not token or token.user_id != current_user.id or token.revoked_at is not None:
+ raise HTTPException(status_code=404, detail="Embed token not found")
+
+ now = utcnow()
+ if token.expires_at is not None and token.expires_at < now:
+ raise HTTPException(
+ status_code=409,
+ detail="Expired embed tokens cannot be rotated",
+ )
+
+ token.revoked_at = now
+ session.add(token)
+
+ plain_token = generate_embed_token()
+ new_token = EmbedToken(
+ user_id=current_user.id,
+ name=token.name,
+ token_prefix=get_embed_token_prefix(plain_token),
+ token_hash=hash_embed_token(plain_token),
+ scopes=token.scopes,
+ allowed_origins=token.allowed_origins,
+ expires_at=token.expires_at,
+ )
+ session.add(new_token)
+ session.commit()
+ session.refresh(new_token)
+ return EmbedTokenCreateResponse(
+ token=plain_token,
+ embed_token=EmbedTokenRead.model_validate(new_token),
+ )
+
+
+@router.delete("/embed-tokens/{token_id}", status_code=204)
+def delete_embed_token(
+ token_id: int,
+ current_user: User = Depends(require_user),
+ session: Session = Depends(get_session),
+) -> None:
+ """Revoke an embed token by ID."""
+ token = session.get(EmbedToken, token_id)
+ if not token or token.user_id != current_user.id:
+ raise HTTPException(status_code=404, detail="Embed token not found")
+
+ token.revoked_at = utcnow()
+ session.add(token)
+ session.commit()
diff --git a/backend/app/schemas.py b/backend/app/schemas.py
index 80810a62..96c3b018 100644
--- a/backend/app/schemas.py
+++ b/backend/app/schemas.py
@@ -324,7 +324,6 @@ class UserSettingsRead(SQLModel):
user_id: int
language: str
timezone: str
- quote_service_enabled: bool
theme: str
custom_theme: Optional[str] = None
@@ -419,6 +418,13 @@ class OidcConfigRead(SQLModel):
provider_name: Optional[str] = None
+class AppConfigRead(SQLModel):
+ """Application-level feature flags."""
+ embed_enabled: bool
+ dashboard_quote_enabled: bool
+ thalia_cover_search_enabled: bool
+
+
class OidcLinkRead(SQLModel):
"""OIDC link status."""
linked: bool
@@ -599,6 +605,38 @@ class DataImportPreviewResponse(SQLModel):
errors: list[str] = []
+class EmbedTokenCreate(SQLModel):
+ """Embed token creation request."""
+ name: str = Field(max_length=255)
+ allowed_origins: Optional[str] = None
+ expires_at: Optional[datetime] = None
+
+
+class EmbedTokenUpdate(SQLModel):
+ """Embed token update request."""
+ name: Optional[str] = None
+ allowed_origins: Optional[str] = None
+ expires_at: Optional[datetime] = None
+
+
+class EmbedTokenRead(SQLModel):
+ """Embed token read response (without the raw token value)."""
+ id: int
+ name: str
+ token_prefix: str
+ scopes: str
+ allowed_origins: Optional[str]
+ expires_at: Optional[datetime]
+ last_used_at: Optional[datetime]
+ created_at: datetime
+
+
+class EmbedTokenCreateResponse(SQLModel):
+ """Embed token creation response containing the raw token (shown once)."""
+ token: str
+ embed_token: EmbedTokenRead
+
+
class DataImportExecuteResult(SQLModel):
"""Import execution result summary."""
imported: int
diff --git a/backend/tests/test_embed.py b/backend/tests/test_embed.py
new file mode 100644
index 00000000..2b0e45b5
--- /dev/null
+++ b/backend/tests/test_embed.py
@@ -0,0 +1,359 @@
+"""Tests for embed token lifecycle and the embed HTML widget endpoint."""
+
+import re
+from collections.abc import Callable
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlmodel import Session, select
+
+from app.auth import generate_embed_token, hash_embed_token
+from app.models import Book, EmbedToken, UserRole
+from app.time_utils import utcnow
+
+# ── Helpers ──────────────────────────────────────────────────────────────
+
+
+def _create_token(
+ session: Session,
+ user_id: int,
+ *,
+ name: str = "Test Token",
+ allowed_origins: str | None = None,
+ expires_at: datetime | None = None,
+) -> str:
+ plain = generate_embed_token()
+ token = EmbedToken(
+ user_id=user_id,
+ name=name,
+ token_prefix=plain[:12],
+ token_hash=hash_embed_token(plain),
+ allowed_origins=allowed_origins,
+ expires_at=expires_at,
+ )
+ session.add(token)
+ session.commit()
+ return plain
+
+
+# ── Profile CRUD Tests ───────────────────────────────────────────────────
+
+
+class TestListEmbedTokens:
+ def test_lists_own_tokens(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ resp = client.get("/api/profile/embed-tokens")
+ assert resp.status_code == 200
+ data = resp.json()
+ assert len(data) >= 1
+ assert data[0]["token_prefix"] == plain[:12]
+
+ def test_does_not_list_revoked_tokens(
+ self, client: TestClient, session: Session, create_user_with_key: Callable[..., Any]
+ ) -> None:
+ user, key = create_user_with_key(email="embed_list@example.com")
+ client.headers["X-API-Key"] = key
+ plain = _create_token(session, user.id)
+ # Revoke it
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ token.revoked_at = utcnow()
+ session.add(token)
+ session.commit()
+ resp = client.get("/api/profile/embed-tokens")
+ assert resp.status_code == 200
+ assert all(t["token_prefix"] != plain[:12] for t in resp.json())
+
+ def test_other_user_cannot_see_tokens(
+ self, client: TestClient, session: Session, create_user_with_key: Callable[..., Any]
+ ) -> None:
+ user, key = create_user_with_key(email="other_embed@example.com")
+ client.headers["X-API-Key"] = key
+ _create_token(session, user_id=user.id, name="Secret")
+ resp = client.get("/api/profile/embed-tokens")
+ assert resp.status_code == 200
+ assert len(resp.json()) == 1 # only own tokens
+
+
+class TestCreateEmbedToken:
+ def test_creates_token(self, client: TestClient) -> None:
+ resp = client.post(
+ "/api/profile/embed-tokens",
+ json={"name": "Homarr Widget"},
+ )
+ assert resp.status_code == 201
+ data = resp.json()
+ assert data["token"].startswith("le_")
+ assert data["embed_token"]["name"] == "Homarr Widget"
+ assert data["embed_token"]["token_prefix"] == data["token"][:12]
+
+ def test_with_allowed_origins(self, client: TestClient) -> None:
+ resp = client.post(
+ "/api/profile/embed-tokens",
+ json={
+ "name": "With Origins",
+ "allowed_origins": "https://homarr.local,https://dashy.local",
+ },
+ )
+ assert resp.status_code == 201
+ data = resp.json()
+ assert data["embed_token"]["allowed_origins"] == "https://homarr.local,https://dashy.local"
+
+ def test_with_expiry(self, client: TestClient) -> None:
+ expires = (datetime.now(timezone.utc) + timedelta(days=90)).isoformat()
+ resp = client.post(
+ "/api/profile/embed-tokens",
+ json={"name": "Expiring", "expires_at": expires},
+ )
+ assert resp.status_code == 201
+
+
+class TestUpdateEmbedToken:
+ def test_updates_name_and_origins(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ resp = client.patch(
+ f"/api/profile/embed-tokens/{token.id}",
+ json={"name": "Renamed", "allowed_origins": "https://example.com"},
+ )
+ assert resp.status_code == 200
+ assert resp.json()["name"] == "Renamed"
+ assert resp.json()["allowed_origins"] == "https://example.com"
+
+ def test_not_found_for_other_user(
+ self, client: TestClient, session: Session, create_user_with_key: Callable[..., Any]
+ ) -> None:
+ user, key = create_user_with_key(email="other_update@example.com")
+ client.headers["X-API-Key"] = key
+ resp = client.patch("/api/profile/embed-tokens/99999", json={"name": "Nope"})
+ assert resp.status_code == 404
+
+
+class TestRotateEmbedToken:
+ def test_rotate_returns_new_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1, name="Rotate Me")
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ old_id = token.id
+ resp = client.post(f"/api/profile/embed-tokens/{old_id}/rotate")
+ assert resp.status_code == 200
+ data = resp.json()
+ assert data["token"].startswith("le_")
+ assert data["token"] != plain
+ assert data["embed_token"]["name"] == "Rotate Me"
+ assert data["embed_token"]["id"] != old_id
+ # Old token should be revoked
+ session.expire_all()
+ old = session.get(EmbedToken, old_id)
+ assert old is not None
+ assert old.revoked_at is not None
+
+ def test_cannot_rotate_expired_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(
+ session,
+ user_id=1,
+ name="Expired",
+ expires_at=datetime.now(timezone.utc) - timedelta(hours=1),
+ )
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+
+ resp = client.post(f"/api/profile/embed-tokens/{token.id}/rotate")
+ assert resp.status_code == 409
+ assert resp.json()["detail"] == "Expired embed tokens cannot be rotated"
+
+
+class TestDeleteEmbedToken:
+ def test_revokes_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ resp = client.delete(f"/api/profile/embed-tokens/{token.id}")
+ assert resp.status_code == 204
+ session.expire_all()
+ deleted = session.get(EmbedToken, token.id)
+ assert deleted is not None
+ assert deleted.revoked_at is not None
+
+ def test_not_found(self, client: TestClient) -> None:
+ resp = client.delete("/api/profile/embed-tokens/99999")
+ assert resp.status_code == 404
+
+
+# ── Embed Widget Endpoint Tests ──────────────────────────────────────────
+
+
+class TestEmbedStatsEndpoint:
+ def test_returns_html_with_valid_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 200
+ assert resp.headers["content-type"].startswith("text/html")
+ assert "" in resp.text
+ assert "Books" in resp.text
+
+ def test_401_for_invalid_token(self, client: TestClient) -> None:
+ resp = client.get("/embed/v1/stats?token=le_invalid")
+ assert resp.status_code == 401
+
+ def test_401_for_revoked_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(session, user_id=1)
+ token = session.exec(
+ select(EmbedToken).where(EmbedToken.token_hash == hash_embed_token(plain))
+ ).first()
+ assert token is not None
+ token.revoked_at = utcnow()
+ session.add(token)
+ session.commit()
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 401
+
+ def test_401_for_expired_token(self, client: TestClient, session: Session) -> None:
+ plain = _create_token(
+ session, user_id=1,
+ expires_at=datetime.now(timezone.utc) - timedelta(hours=1),
+ )
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 401
+
+ def test_shows_stats_for_user(self, client: TestClient, session: Session) -> None:
+ for i in range(3):
+ session.add(Book(title=f"Book {i}", author="Author", page_count=200, user_id=1, reading_status="read"))
+ session.commit()
+
+ plain = _create_token(session, user_id=1)
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 200
+ assert "3" in resp.text
+
+ def test_user_isolation(self, client: TestClient, session: Session) -> None:
+ for i in range(5):
+ session.add(Book(title=f"U1 Book {i}", author="Author", page_count=100, user_id=1, reading_status="want_to_read"))
+ session.commit()
+
+ from app.auth import generate_api_key, get_api_key_prefix, get_password_hash, hash_api_key, encrypt_api_key
+ from app.models import ApiKey, User, UserSettings
+
+ user2 = User(firstname="User", lastname="Two", email="u2@example.com",
+ role=UserRole.user, hashed_password=get_password_hash("secret"))
+ session.add(user2)
+ session.commit()
+ session.refresh(user2)
+ session.add(UserSettings(user_id=user2.id, language="en"))
+ key2 = generate_api_key()
+ session.add(ApiKey(user_id=user2.id, key_prefix=get_api_key_prefix(key2),
+ key_hash=hash_api_key(key2), key_encrypted=encrypt_api_key(key2),
+ description="Key"))
+ session.commit()
+ plain = _create_token(session, user2.id)
+ resp = client.get(f"/embed/v1/stats?token={plain}")
+ assert resp.status_code == 200
+ text = resp.text
+ books_match = re.search(r'
]*>\s*0\s*