feat(auth): connected apps - sign in to external applications with Bambuddy

Minimal OAuth 2.0 authorization-code flow with PKCE (S256): admins register
    an app with one exact callback URL (Settings > API Keys > Connected Apps);
    /connect/authorize asks for consent once and returns a single-use, 60 s code
    bound to app, callback and challenge; POST /api/v1/connect/token swaps it,
    with the client secret, for the user's identity and permissions. Codes and
    secrets stored hashed, exchanges rate-limited per client and IP, no redirect
    before the callback is validated, API keys cannot authorize, refused while
    auth is disabled. i18n for all 15 locales.

    -----

    fix(db): upgrading from 0.2.4.0 or older no longer crashes at startup

    The #2974 failure-reason conversion ran before the #1378 migration that adds
    print_log_entries.failure_reason, so older databases stopped with "no such
    column: failure_reason". It now skips a table without the column, only runs
    where a legacy label exists, and on SQLite rebuilds archive_fts first, since
    archives created before that index existed trip "database disk image is
    malformed" when updated.

backend/app/api/routes/connected_apps.py

@@ -0,0 +1,405 @@
+"""Connected apps: sign users in to external applications with Bambuddy.
+
+A minimal OAuth 2.0 authorization-code flow with PKCE (S256 only):
+
+1. The app sends the user's browser to the SPA page ``/connect/authorize``.
+   The page, running with the user's normal Bambuddy session, calls
+   ``GET /connect/authorize/info`` to show who is asking and, after consent,
+   ``POST /connect/authorize`` for a code. It then sends the browser back to
+   the app's registered callback URL with that code.
+2. The app's server calls ``POST /connect/token`` with the code, its client
+   secret and the PKCE verifier, and receives the user's identity and
+   permissions.
+
+What the app never gets is the user's Bambuddy login token. What a code is
+worth is deliberately small: single use, 60 seconds, bound to one app, its
+exact callback URL and the PKCE challenge, and only redeemable with the app's
+secret.
+
+Sign-in through an app only exists while Bambuddy authentication is enabled.
+With it disabled there are no users to sign in, and both authorize and token
+answer ``auth_disabled`` so the app can fall back to its own login rather
+than let everyone in.
+"""
+
+import base64
+import functools
+import hashlib
+import hmac
+import logging
+import secrets
+from datetime import datetime, timedelta, timezone
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi.security import HTTPAuthorizationCredentials
+from sqlalchemy import delete, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.core.auth import (
+    RequirePermissionIfAuthEnabled,
+    get_current_user,
+    get_password_hash,
+    get_user_by_username,
+    is_auth_enabled,
+    security,
+    verify_password,
+)
+from backend.app.core.database import get_db
+from backend.app.core.permissions import Permission
+from backend.app.models.auth_ephemeral import AuthEphemeralToken, EventType, TokenType
+from backend.app.models.connected_app import ConnectedApp, ConnectedAppGrant
+from backend.app.models.user import User
+from backend.app.schemas.connected_app import (
+    ConnectAuthorizeInfo,
+    ConnectAuthorizeRequest,
+    ConnectAuthorizeResponse,
+    ConnectedAppCreate,
+    ConnectedAppResponse,
+    ConnectedAppSecretResponse,
+    ConnectedAppUpdate,
+    ConnectedUser,
+    ConnectTokenRequest,
+    ConnectTokenResponse,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/connect", tags=["connected-apps"])
+
+CODE_TTL = timedelta(seconds=60)
+# Failed token exchanges tolerated per client and per IP in the rate-limit
+# window (mfa.LOCKOUT_WINDOW). A working app fails almost never; this only has
+# to stop someone guessing secrets or codes.
+MAX_FAILED_TOKEN_EXCHANGES = 20
+
+AUTH_DISABLED = "auth_disabled"
+
+
+@functools.cache
+def _dummy_secret_hash() -> str:
+    """Verified against when the client_id is unknown, so an unknown client
+    costs the same bcrypt round as a wrong secret and the two can't be told
+    apart by timing. Built on first use to keep bcrypt out of import time."""
+    return get_password_hash(secrets.token_urlsafe(32))
+
+
+def _hash_code(code: str) -> str:
+    return hashlib.sha256(code.encode()).hexdigest()
+
+
+def _s256(verifier: str) -> str:
+    digest = hashlib.sha256(verifier.encode("ascii")).digest()
+    return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
+
+
+def _new_client_credentials() -> tuple[str, str]:
+    return f"bba_{secrets.token_hex(12)}", f"bbs_{secrets.token_urlsafe(32)}"
+
+
+def _with_secret(app: ConnectedApp, client_secret: str) -> ConnectedAppSecretResponse:
+    return ConnectedAppSecretResponse(
+        **ConnectedAppResponse.model_validate(app).model_dump(), client_secret=client_secret
+    )
+
+
+async def _get_app_or_404(db: AsyncSession, app_id: int) -> ConnectedApp:
+    app = await db.get(ConnectedApp, app_id)
+    if app is None:
+        raise HTTPException(404, "Connected app not found")
+    return app
+
+
+# --------------------------------------------------------------------------
+# Admin: register and manage apps
+# --------------------------------------------------------------------------
+
+
+@router.get("/apps", response_model=list[ConnectedAppResponse])
+async def list_connected_apps(
+    db: AsyncSession = Depends(get_db),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
+):
+    result = await db.execute(select(ConnectedApp).order_by(ConnectedApp.created_at.desc()))
+    return list(result.scalars().all())
+
+
+@router.post("/apps", response_model=ConnectedAppSecretResponse)
+async def create_connected_app(
+    data: ConnectedAppCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
+):
+    """Register an app. The client secret is in this response and nowhere else."""
+    if not await is_auth_enabled(db):
+        # Nobody could sign in through it, and an admin who set one up would
+        # reasonably assume the app is now protected by Bambuddy's login.
+        raise HTTPException(400, "Connected apps require authentication to be enabled")
+    client_id, client_secret = _new_client_credentials()
+    app = ConnectedApp(
+        name=data.name.strip(),
+        client_id=client_id,
+        client_secret_hash=get_password_hash(client_secret),
+        redirect_uri=data.redirect_uri,
+        enabled=True,
+        created_by_id=current_user.id if current_user else None,
+    )
+    db.add(app)
+    await db.commit()
+    await db.refresh(app)
+    logger.info("Connected app %s '%s' registered", app.id, app.name)
+    return _with_secret(app, client_secret)
+
+
+@router.patch("/apps/{app_id}", response_model=ConnectedAppResponse)
+async def update_connected_app(
+    app_id: int,
+    data: ConnectedAppUpdate,
+    db: AsyncSession = Depends(get_db),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
+):
+    app = await _get_app_or_404(db, app_id)
+    if data.name is not None:
+        app.name = data.name.strip()
+    if data.redirect_uri is not None:
+        app.redirect_uri = data.redirect_uri
+    if data.enabled is not None:
+        app.enabled = data.enabled
+    await db.commit()
+    await db.refresh(app)
+    return app
+
+
+@router.post("/apps/{app_id}/rotate-secret", response_model=ConnectedAppSecretResponse)
+async def rotate_connected_app_secret(
+    app_id: int,
+    db: AsyncSession = Depends(get_db),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
+):
+    """Replace the client secret. The old one stops working immediately."""
+    app = await _get_app_or_404(db, app_id)
+    _, client_secret = _new_client_credentials()
+    app.client_secret_hash = get_password_hash(client_secret)
+    await db.commit()
+    await db.refresh(app)
+    logger.info("Connected app %s secret rotated", app.id)
+    return _with_secret(app, client_secret)
+
+
+@router.delete("/apps/{app_id}")
+async def delete_connected_app(
+    app_id: int,
+    db: AsyncSession = Depends(get_db),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
+):
+    app = await _get_app_or_404(db, app_id)
+    await db.execute(delete(ConnectedAppGrant).where(ConnectedAppGrant.app_id == app.id))
+    # Outstanding codes would fail anyway (their app is gone); drop them now.
+    await db.execute(
+        delete(AuthEphemeralToken).where(
+            AuthEphemeralToken.token_type == TokenType.CONNECT_CODE,
+            AuthEphemeralToken.provider_id == app.id,
+        )
+    )
+    await db.delete(app)
+    await db.commit()
+    logger.info("Connected app %s deleted", app_id)
+    return {"message": "Connected app deleted"}
+
+
+# --------------------------------------------------------------------------
+# Authorize: called by the SPA with the signed-in user's session
+# --------------------------------------------------------------------------
+
+
+async def require_signed_in_user(
+    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(security)] = None,
+    db: AsyncSession = Depends(get_db),
+) -> User:
+    """The user behind a Bambuddy login token. API keys are refused.
+
+    An API key would let a script sign its owner in to an app without the
+    owner ever seeing a consent screen; only a person's own session may do
+    that. ``get_current_user`` accepts JWTs only, so a ``bb_`` key fails there.
+    """
+    if not await is_auth_enabled(db):
+        raise HTTPException(409, AUTH_DISABLED)
+    return await get_current_user(credentials)
+
+
+async def _app_for_authorize(db: AsyncSession, client_id: str, redirect_uri: str) -> ConnectedApp:
+    result = await db.execute(select(ConnectedApp).where(ConnectedApp.client_id == client_id))
+    app = result.scalar_one_or_none()
+    # One message for every case: the page must not redirect anywhere unless
+    # all of these hold, and there is nothing useful to tell the user apart.
+    if app is None or not app.enabled or not hmac.compare_digest(app.redirect_uri, redirect_uri):
+        raise HTTPException(400, "Unknown app, app disabled, or callback URL does not match its registration")
+    return app
+
+
+@router.get("/authorize/info", response_model=ConnectAuthorizeInfo)
+async def authorize_info(
+    client_id: str = Query(..., max_length=64),
+    redirect_uri: str = Query(..., max_length=500),
+    db: AsyncSession = Depends(get_db),
+    user: User = Depends(require_signed_in_user),
+):
+    """What the consent screen shows. Validates the request before the page acts on it."""
+    app = await _app_for_authorize(db, client_id, redirect_uri)
+    granted = await db.execute(
+        select(ConnectedAppGrant.id).where(ConnectedAppGrant.app_id == app.id, ConnectedAppGrant.user_id == user.id)
+    )
+    return ConnectAuthorizeInfo(
+        app_name=app.name, username=user.username, already_granted=granted.scalar_one_or_none() is not None
+    )
+
+
+@router.post("/authorize", response_model=ConnectAuthorizeResponse)
+async def authorize(
+    data: ConnectAuthorizeRequest,
+    db: AsyncSession = Depends(get_db),
+    user: User = Depends(require_signed_in_user),
+):
+    """Issue a code for the signed-in user and record their consent."""
+    app = await _app_for_authorize(db, data.client_id, data.redirect_uri)
+    now = datetime.now(timezone.utc)
+
+    # Keep the table small: codes live for a minute, so anything expired is junk.
+    await db.execute(
+        delete(AuthEphemeralToken).where(
+            AuthEphemeralToken.token_type == TokenType.CONNECT_CODE,
+            AuthEphemeralToken.expires_at < now,
+        )
+    )
+
+    granted = await db.execute(
+        select(ConnectedAppGrant.id).where(ConnectedAppGrant.app_id == app.id, ConnectedAppGrant.user_id == user.id)
+    )
+    if granted.scalar_one_or_none() is None:
+        db.add(ConnectedAppGrant(app_id=app.id, user_id=user.id))
+
+    code = secrets.token_urlsafe(32)
+    db.add(
+        AuthEphemeralToken.new_connect_code(
+            code_hash=_hash_code(code),
+            username=user.username,
+            app_id=app.id,
+            code_challenge=data.code_challenge,
+            expires_at=now + CODE_TTL,
+        )
+    )
+    try:
+        await db.commit()
+    except IntegrityError:
+        # Two tabs consenting at once both inserted the grant; the other one won.
+        await db.rollback()
+        db.add(
+            AuthEphemeralToken.new_connect_code(
+                code_hash=_hash_code(code),
+                username=user.username,
+                app_id=app.id,
+                code_challenge=data.code_challenge,
+                expires_at=now + CODE_TTL,
+            )
+        )
+        await db.commit()
+
+    logger.info("Connected app %s: code issued for user %s", app.id, user.id)
+    return ConnectAuthorizeResponse(code=code, redirect_uri=app.redirect_uri)
+
+
+# --------------------------------------------------------------------------
+# Token: called by the app's server, authenticated by its client secret
+# --------------------------------------------------------------------------
+
+
+def _token_error(status_code: int, error: str) -> HTTPException:
+    return HTTPException(status_code, {"error": error})
+
+
+@router.post("/token", response_model=ConnectTokenResponse)
+async def exchange_code(
+    request: Request,
+    data: ConnectTokenRequest,
+    db: AsyncSession = Depends(get_db),
+):
+    """Swap a code for the user's identity and permissions.
+
+    Every failure after the rate-limit check counts against both the client and
+    the caller's IP. The response says only ``invalid_client`` (who is asking
+    is wrong) or ``invalid_grant`` (the code is), never which check failed.
+    """
+    from backend.app.api.routes.auth import _get_client_ip
+    from backend.app.api.routes.mfa import check_rate_limit, clear_failed_attempts, record_failed_attempt
+
+    client_ip = _get_client_ip(request)
+    await check_rate_limit(
+        db, data.client_id, event_type=EventType.CONNECT_TOKEN_CLIENT, max_attempts=MAX_FAILED_TOKEN_EXCHANGES
+    )
+    await check_rate_limit(
+        db, client_ip, event_type=EventType.CONNECT_TOKEN_IP, max_attempts=MAX_FAILED_TOKEN_EXCHANGES
+    )
+
+    async def fail(status_code: int, error: str, reason: str) -> HTTPException:
+        await record_failed_attempt(db, data.client_id, event_type=EventType.CONNECT_TOKEN_CLIENT)
+        await record_failed_attempt(db, client_ip, event_type=EventType.CONNECT_TOKEN_IP)
+        logger.warning("Connected app token exchange refused for client %s: %s", data.client_id, reason)
+        return _token_error(status_code, error)
+
+    if not await is_auth_enabled(db):
+        raise _token_error(400, AUTH_DISABLED)
+
+    result = await db.execute(select(ConnectedApp).where(ConnectedApp.client_id == data.client_id))
+    app = result.scalar_one_or_none()
+    secret_ok = verify_password(data.client_secret, app.client_secret_hash if app else _dummy_secret_hash())
+    if app is None or not secret_ok:
+        raise await fail(401, "invalid_client", "unknown client or wrong secret")
+    if not app.enabled:
+        raise await fail(401, "invalid_client", "app disabled")
+
+    # Consume first, then check: DELETE ... RETURNING makes the code single-use
+    # even under concurrent exchanges, and a code that fails a later check is
+    # spent all the same.
+    now = datetime.now(timezone.utc)
+    consumed = await db.execute(
+        delete(AuthEphemeralToken)
+        .where(
+            AuthEphemeralToken.token == _hash_code(data.code),
+            AuthEphemeralToken.token_type == TokenType.CONNECT_CODE,
+            AuthEphemeralToken.provider_id == app.id,
+            AuthEphemeralToken.expires_at > now,
+        )
+        .returning(AuthEphemeralToken.username, AuthEphemeralToken.nonce)
+    )
+    row = consumed.one_or_none()
+    await db.commit()
+    if row is None:
+        raise await fail(400, "invalid_grant", "unknown, expired, reused or foreign code")
+    username, code_challenge = row
+
+    if not hmac.compare_digest(app.redirect_uri, data.redirect_uri):
+        raise await fail(400, "invalid_grant", "callback URL mismatch")
+    if not code_challenge or not hmac.compare_digest(_s256(data.code_verifier), code_challenge):
+        raise await fail(400, "invalid_grant", "PKCE verifier mismatch")
+
+    user = await get_user_by_username(db, username)
+    if user is None or not user.is_active:
+        raise await fail(400, "invalid_grant", "user gone or disabled")
+
+    app.last_used_at = now.replace(tzinfo=None)
+    await clear_failed_attempts(db, data.client_id, event_type=EventType.CONNECT_TOKEN_CLIENT)
+    await db.commit()
+
+    logger.info("Connected app %s: user %s signed in", app.id, user.id)
+    return ConnectTokenResponse(
+        user=ConnectedUser(
+            id=user.id,
+            username=user.username,
+            email=user.email,
+            is_admin=user.is_admin,
+            groups=sorted(g.name for g in user.groups),
+            permissions=sorted(user.get_permissions()),
+        ),
+        issued_at=now,
+    )

backend/app/api/routes/users.py

@@ -25,6 +25,8 @@ from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
 from backend.app.models.api_key import APIKey
 from backend.app.models.archive import PrintArchive
+from backend.app.models.auth_ephemeral import AuthEphemeralToken, TokenType
+from backend.app.models.connected_app import ConnectedApp, ConnectedAppGrant
 from backend.app.models.group import Group
 from backend.app.models.library import LibraryFile
 from backend.app.models.long_lived_token import LongLivedToken
@@ -499,6 +501,20 @@ async def delete_user(
     await db.execute(delete(UserOTPCode).where(UserOTPCode.user_id == user_id))
     await db.execute(delete(LongLivedToken).where(LongLivedToken.user_id == user_id))
 
+    # Connected apps, same SQLite/FK pattern. A leftover consent row would
+    # skip the consent screen for whoever is next given this user id, and a
+    # code issued in the last minute is keyed by username, not id.
+    from sqlalchemy import update as _update
+
+    await db.execute(delete(ConnectedAppGrant).where(ConnectedAppGrant.user_id == user_id))
+    await db.execute(
+        delete(AuthEphemeralToken).where(
+            AuthEphemeralToken.token_type == TokenType.CONNECT_CODE,
+            func.lower(AuthEphemeralToken.username) == user.username.lower(),
+        )
+    )
+    await db.execute(_update(ConnectedApp).where(ConnectedApp.created_by_id == user_id).values(created_by_id=None))
+
     await db.delete(user)
     await db.commit()
 

backend/app/core/database.py

@@ -293,6 +293,7 @@ async def init_db():
         auth_ephemeral,
         bug_report,
         color_catalog,
+        connected_app,
         external_link,
         filament,
         filament_sku_settings,
@@ -1702,6 +1703,31 @@ _LEGACY_FAILURE_REASON_LABELS: dict[str, str] = {
 }
 
 
+async def _table_has_column(conn, table: str, column: str) -> bool:
+    """Whether ``table`` has ``column``, on either dialect.
+
+    ``table`` is interpolated into the SQLite PRAGMA (it cannot be bound), so
+    callers pass literals only.
+    """
+    from sqlalchemy import text
+
+    if is_sqlite():
+        result = await conn.execute(text(f"PRAGMA table_info({table})"))
+        return any(row[1] == column for row in result)
+    result = await conn.execute(
+        text("SELECT 1 FROM information_schema.columns WHERE table_name = :table AND column_name = :col"),
+        {"table": table, "col": column},
+    )
+    return result.first() is not None
+
+
+async def _sqlite_table_exists(conn, name: str) -> bool:
+    from sqlalchemy import text
+
+    result = await conn.execute(text("SELECT 1 FROM sqlite_master WHERE name = :name"), {"name": name})
+    return result.first() is not None
+
+
 async def _migrate_failure_reason_vocabulary(conn):
     """Fold historical failure-reason labels onto the canonical keys (#2974).
 
@@ -1734,6 +1760,8 @@ async def _migrate_failure_reason_vocabulary(conn):
         if label != key:
             by_key[key].append(label)
 
+    all_labels = [label for labels in by_key.values() for label in labels]
+
     total = 0
     async with conn.begin_nested():
         # nosec B608 — the only interpolated fragment is `table`, which the loop
@@ -1741,6 +1769,30 @@ async def _migrate_failure_reason_vocabulary(conn):
         # Both the key and the label list are bound parameters. A table name
         # cannot be expressed as one, which is why it is interpolated at all.
         for table in ("print_archives", "print_log_entries"):
+            # A database older than #1378 has no print_log_entries.failure_reason
+            # yet (a later ALTER in run_migrations adds it). Such a table cannot
+            # hold a legacy label, and querying it crashed startup with "no such
+            # column: failure_reason".
+            if not await _table_has_column(conn, table, "failure_reason"):
+                continue
+            has_work = (
+                await conn.execute(
+                    text(
+                        f"SELECT 1 FROM {table} WHERE failure_reason IN :labels LIMIT 1"  # noqa: S608  # nosec B608
+                    ).bindparams(bindparam("labels", expanding=True)),
+                    {"labels": all_labels},
+                )
+            ).first() is not None
+            if not has_work:
+                continue
+            if table == "print_archives" and is_sqlite() and await _sqlite_table_exists(conn, "archive_fts"):
+                # Same trap as the plate_id backfill further down: archives
+                # created before the external-content FTS index existed were
+                # never indexed, and the AFTER UPDATE trigger's FTS 'delete' on
+                # such a row fails with "database disk image is malformed".
+                # Rebuild first so every row is present. Only when there is
+                # work, since a rebuild re-reads every archive.
+                await conn.execute(text("INSERT INTO archive_fts(archive_fts) VALUES('rebuild')"))
             for key, labels in by_key.items():
                 result = await conn.execute(
                     text(

backend/app/main.py

@@ -28,6 +28,7 @@ from backend.app.api.routes import (
     camera,
     camwall,
     cloud,
+    connected_apps,
     discovery,
     external_links,
     filaments,
@@ -9472,6 +9473,9 @@ PUBLIC_API_ROUTES = {
     "/api/v1/auth/oidc/providers",  # Public list of enabled providers
     "/api/v1/auth/oidc/callback",  # Redirect target from OIDC provider
     "/api/v1/auth/oidc/exchange",  # Exchange short-lived OIDC token for JWT
+    # Connected apps: the app's server swaps a code for the user's identity,
+    # authenticated by its client secret rather than a login token.
+    "/api/v1/connect/token",
     # Version check for updates (no sensitive data)
     "/api/v1/updates/version",
     # Metrics endpoint handles its own prometheus_token authentication
@@ -9939,6 +9943,7 @@ app.include_router(slicer_presets.router, prefix=app_settings.api_prefix)
 app.include_router(archive_purge.router, prefix=app_settings.api_prefix)
 app.include_router(makerworld.router, prefix=app_settings.api_prefix)
 app.include_router(api_keys.router, prefix=app_settings.api_prefix)
+app.include_router(connected_apps.router, prefix=app_settings.api_prefix)
 app.include_router(webhook.router, prefix=app_settings.api_prefix)
 app.include_router(ams_history.router, prefix=app_settings.api_prefix)
 app.include_router(printer_sensor_history.router, prefix=app_settings.api_prefix)

backend/app/models/__init__.py

@@ -4,6 +4,7 @@ from backend.app.models.api_key import APIKey
 from backend.app.models.archive import PrintArchive
 from backend.app.models.auth_ephemeral import AuthEphemeralToken, AuthRateLimitEvent
 from backend.app.models.color_catalog import ColorCatalogEntry
+from backend.app.models.connected_app import ConnectedApp, ConnectedAppGrant
 from backend.app.models.filament import Filament
 from backend.app.models.github_backup import GitHubBackupConfig, GitHubBackupLog
 from backend.app.models.group import Group, user_groups
@@ -98,4 +99,6 @@ __all__ = [
     "AuthEphemeralToken",
     "AuthRateLimitEvent",
     "LongLivedToken",
+    "ConnectedApp",
+    "ConnectedAppGrant",
 ]

backend/app/models/auth_ephemeral.py

@@ -42,6 +42,7 @@ class TokenType(str, Enum):
     PASSWORD_RESET = "password_reset"
     EMAIL_OTP_SETUP = "email_otp_setup"
     SLICER_DOWNLOAD = "slicer_download"
+    CONNECT_CODE = "connect_code"
 
 
 class EventType(str, Enum):
@@ -57,6 +58,8 @@ class EventType(str, Enum):
     LOGIN_IP = "login_ip"
     PASSWORD_RESET_SEND = "password_reset_send"
     PASSWORD_RESET_IP = "password_reset_ip"
+    CONNECT_TOKEN_CLIENT = "connect_client"
+    CONNECT_TOKEN_IP = "connect_ip"
 
 
 class AuthEphemeralToken(Base):
@@ -183,6 +186,30 @@ class AuthEphemeralToken(Base):
             expires_at=expires_at,
         )
 
+    @classmethod
+    def new_connect_code(
+        cls,
+        code_hash: str,
+        username: str,
+        app_id: int,
+        code_challenge: str,
+        expires_at: datetime,
+    ) -> AuthEphemeralToken:
+        """Create a connected-app authorization code.
+
+        Only the SHA-256 of the code is stored (``token``), so a database read
+        does not yield a usable code. Field reuse: ``provider_id`` holds the
+        connected app's id and ``nonce`` the PKCE S256 challenge.
+        """
+        return cls(
+            token=code_hash,
+            token_type=TokenType.CONNECT_CODE,
+            username=username,
+            provider_id=app_id,
+            nonce=code_challenge,
+            expires_at=expires_at,
+        )
+
 
 class AuthRateLimitEvent(Base):
     """Timestamped events used for sliding-window rate limiting."""

backend/app/models/connected_app.py

@@ -0,0 +1,48 @@
+"""Connected apps: external applications that sign users in with Bambuddy.
+
+A connected app is registered by an admin with one exact callback URL. It
+signs a user in through a minimal OAuth 2.0 authorization-code flow with PKCE
+(see ``api/routes/connected_apps.py``): Bambuddy hands the app a single-use,
+60-second code, and the app's server swaps it for the user's identity and
+permissions. The app never sees the user's Bambuddy login token.
+"""
+
+from datetime import datetime
+
+from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.app.core.database import Base
+
+
+class ConnectedApp(Base):
+    __tablename__ = "connected_apps"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String(100))
+    # Public identifier the app sends on every request.
+    client_id: Mapped[str] = mapped_column(String(64), unique=True, index=True)
+    # bcrypt hash; the secret itself is shown once, at creation or rotation.
+    client_secret_hash: Mapped[str] = mapped_column(String(255))
+    # Codes are only ever delivered here. Compared exactly, never by prefix.
+    redirect_uri: Mapped[str] = mapped_column(String(500))
+    enabled: Mapped[bool] = mapped_column(Boolean, default=True)
+    created_by_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+    last_used_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+
+
+class ConnectedAppGrant(Base):
+    """A user's consent for one app, so the consent screen is shown only once.
+
+    Deleting the app or the user removes the grant, and the next sign-in asks
+    again.
+    """
+
+    __tablename__ = "connected_app_grants"
+    __table_args__ = (Index("uq_connected_app_grants_app_user", "app_id", "user_id", unique=True),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    app_id: Mapped[int] = mapped_column(ForeignKey("connected_apps.id", ondelete="CASCADE"), index=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
+    granted_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

backend/app/schemas/connected_app.py

@@ -0,0 +1,109 @@
+from datetime import datetime
+from typing import Literal
+from urllib.parse import urlsplit
+
+from pydantic import BaseModel, Field, field_validator
+
+from backend.app.schemas.print_queue import UTCDatetime
+
+# RFC 7636: 43-128 chars from the unreserved set. The challenge is the
+# base64url SHA-256 of the verifier, so it is always exactly 43 chars.
+_PKCE_VERIFIER_PATTERN = r"^[A-Za-z0-9\-._~]{43,128}$"
+_PKCE_CHALLENGE_PATTERN = r"^[A-Za-z0-9\-_]{43}$"
+
+
+def _validate_redirect_uri(value: str) -> str:
+    """Accept only an absolute http(s) URL without a fragment.
+
+    Plain http is allowed: connected apps typically run on the same LAN as
+    Bambuddy, often without TLS. What matters is that codes can only ever be
+    delivered to the one URL an admin registered, and that is enforced by an
+    exact match at authorize and token time.
+    """
+    value = value.strip()
+    parts = urlsplit(value)
+    if parts.scheme not in ("http", "https") or not parts.netloc:
+        raise ValueError("Callback URL must be an absolute http:// or https:// URL")
+    if parts.fragment:
+        raise ValueError("Callback URL must not contain a #fragment")
+    if parts.username or parts.password:
+        raise ValueError("Callback URL must not contain credentials")
+    return value
+
+
+class ConnectedAppCreate(BaseModel):
+    name: str = Field(min_length=1, max_length=100)
+    redirect_uri: str = Field(min_length=1, max_length=500)
+
+    _check_redirect = field_validator("redirect_uri")(_validate_redirect_uri)
+
+
+class ConnectedAppUpdate(BaseModel):
+    name: str | None = Field(default=None, min_length=1, max_length=100)
+    redirect_uri: str | None = Field(default=None, min_length=1, max_length=500)
+    enabled: bool | None = None
+
+    @field_validator("redirect_uri")
+    @classmethod
+    def _check_redirect(cls, value: str | None) -> str | None:
+        return None if value is None else _validate_redirect_uri(value)
+
+
+class ConnectedAppResponse(BaseModel):
+    id: int
+    name: str
+    client_id: str
+    redirect_uri: str
+    enabled: bool
+    created_at: UTCDatetime
+    last_used_at: UTCDatetime | None = None
+
+    class Config:
+        from_attributes = True
+
+
+class ConnectedAppSecretResponse(ConnectedAppResponse):
+    """Returned by create and rotate only: the one time the secret is shown."""
+
+    client_secret: str
+
+
+class ConnectAuthorizeInfo(BaseModel):
+    app_name: str
+    username: str
+    already_granted: bool
+
+
+class ConnectAuthorizeRequest(BaseModel):
+    client_id: str = Field(min_length=1, max_length=64)
+    redirect_uri: str = Field(min_length=1, max_length=500)
+    code_challenge: str = Field(pattern=_PKCE_CHALLENGE_PATTERN)
+    code_challenge_method: Literal["S256"]
+
+
+class ConnectAuthorizeResponse(BaseModel):
+    code: str
+    redirect_uri: str
+
+
+class ConnectTokenRequest(BaseModel):
+    grant_type: Literal["authorization_code"]
+    code: str = Field(min_length=1, max_length=128)
+    redirect_uri: str = Field(min_length=1, max_length=500)
+    client_id: str = Field(min_length=1, max_length=64)
+    client_secret: str = Field(min_length=1, max_length=128)
+    code_verifier: str = Field(pattern=_PKCE_VERIFIER_PATTERN)
+
+
+class ConnectedUser(BaseModel):
+    id: int
+    username: str
+    email: str | None = None
+    is_admin: bool
+    groups: list[str]
+    permissions: list[str]
+
+
+class ConnectTokenResponse(BaseModel):
+    user: ConnectedUser
+    issued_at: datetime

backend/tests/integration/test_connected_apps.py

@@ -0,0 +1,363 @@
+"""Connected apps: sign-in to external applications with a Bambuddy account.
+
+These lock down what a code is worth: single use, 60 seconds, one app, its
+exact callback URL, the PKCE challenge, and only with the app's secret. And
+that none of it works while Bambuddy authentication is disabled.
+"""
+
+import base64
+import hashlib
+import secrets
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from httpx import AsyncClient
+
+CALLBACK = "http://orders.local:8090/auth/callback"
+
+
+def _pkce() -> tuple[str, str]:
+    verifier = secrets.token_urlsafe(48)
+    challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
+    return verifier, challenge
+
+
+def _auth(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
+
+
+@pytest.fixture
+async def admin_token(async_client: AsyncClient) -> str:
+    await async_client.post(
+        "/api/v1/auth/setup",
+        json={"auth_enabled": True, "admin_username": "connectadmin", "admin_password": "AdminPass1!"},
+    )
+    login = await async_client.post("/api/v1/auth/login", json={"username": "connectadmin", "password": "AdminPass1!"})
+    return login.json()["access_token"]
+
+
+@pytest.fixture
+async def operator(async_client: AsyncClient, admin_token: str) -> dict:
+    groups = (await async_client.get("/api/v1/groups/", headers=_auth(admin_token))).json()
+    operators = next(g for g in groups if g["name"] == "Operators")
+    user = (
+        await async_client.post(
+            "/api/v1/users/",
+            headers=_auth(admin_token),
+            json={"username": "shopworker", "password": "Operatorpass1!", "group_ids": [operators["id"]]},
+        )
+    ).json()
+    login = await async_client.post("/api/v1/auth/login", json={"username": "shopworker", "password": "Operatorpass1!"})
+    return {"id": user["id"], "token": login.json()["access_token"]}
+
+
+async def _register(async_client: AsyncClient, admin_token: str, **overrides) -> dict:
+    payload = {"name": "Bambuddy Orders", "redirect_uri": CALLBACK, **overrides}
+    response = await async_client.post("/api/v1/connect/apps", headers=_auth(admin_token), json=payload)
+    assert response.status_code == 200, response.text
+    return response.json()
+
+
+async def _authorize(async_client: AsyncClient, user_token: str, app: dict, challenge: str, **overrides):
+    payload = {
+        "client_id": app["client_id"],
+        "redirect_uri": app["redirect_uri"],
+        "code_challenge": challenge,
+        "code_challenge_method": "S256",
+        **overrides,
+    }
+    return await async_client.post("/api/v1/connect/authorize", headers=_auth(user_token), json=payload)
+
+
+async def _code(async_client: AsyncClient, user_token: str, app: dict) -> tuple[str, str]:
+    verifier, challenge = _pkce()
+    response = await _authorize(async_client, user_token, app, challenge)
+    assert response.status_code == 200, response.text
+    return response.json()["code"], verifier
+
+
+async def _exchange(async_client: AsyncClient, app: dict, code: str, verifier: str, **overrides):
+    payload = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": app["redirect_uri"],
+        "client_id": app["client_id"],
+        "client_secret": app["client_secret"],
+        "code_verifier": verifier,
+        **overrides,
+    }
+    # No Authorization header: the app authenticates with its secret alone.
+    return await async_client.post("/api/v1/connect/token", json=payload)
+
+
+class TestRegistration:
+    async def test_secret_is_shown_once(self, async_client, admin_token):
+        app = await _register(async_client, admin_token)
+        assert app["client_id"].startswith("bba_")
+        assert app["client_secret"].startswith("bbs_")
+        listed = (await async_client.get("/api/v1/connect/apps", headers=_auth(admin_token))).json()
+        assert [a["client_id"] for a in listed] == [app["client_id"]]
+        assert "client_secret" not in listed[0]
+
+    async def test_refused_while_authentication_is_disabled(self, async_client):
+        response = await async_client.post("/api/v1/connect/apps", json={"name": "Orders", "redirect_uri": CALLBACK})
+        assert response.status_code == 400
+
+    async def test_non_admin_cannot_register(self, async_client, operator):
+        response = await async_client.post(
+            "/api/v1/connect/apps",
+            headers=_auth(operator["token"]),
+            json={"name": "Orders", "redirect_uri": CALLBACK},
+        )
+        assert response.status_code == 403
+
+    @pytest.mark.parametrize(
+        "redirect_uri",
+        [
+            "javascript:alert(1)",
+            "/relative/callback",
+            "http://orders.local/cb#frag",
+            "http://user:pw@orders.local/cb",
+            "ftp://orders.local/cb",
+        ],
+    )
+    async def test_callback_must_be_a_plain_absolute_http_url(self, async_client, admin_token, redirect_uri):
+        response = await async_client.post(
+            "/api/v1/connect/apps",
+            headers=_auth(admin_token),
+            json={"name": "Orders", "redirect_uri": redirect_uri},
+        )
+        assert response.status_code == 422
+
+
+class TestSignIn:
+    async def test_full_flow_returns_identity_and_permissions(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+
+        info = await async_client.get(
+            "/api/v1/connect/authorize/info",
+            headers=_auth(operator["token"]),
+            params={"client_id": app["client_id"], "redirect_uri": CALLBACK},
+        )
+        assert info.status_code == 200
+        assert info.json() == {"app_name": "Bambuddy Orders", "username": "shopworker", "already_granted": False}
+
+        code, verifier = await _code(async_client, operator["token"], app)
+        response = await _exchange(async_client, app, code, verifier)
+        assert response.status_code == 200, response.text
+        user = response.json()["user"]
+        assert user["id"] == operator["id"]
+        assert user["username"] == "shopworker"
+        assert user["is_admin"] is False
+        assert user["groups"] == ["Operators"]
+        assert "queue:create" in user["permissions"]
+
+        info_after = await async_client.get(
+            "/api/v1/connect/authorize/info",
+            headers=_auth(operator["token"]),
+            params={"client_id": app["client_id"], "redirect_uri": CALLBACK},
+        )
+        assert info_after.json()["already_granted"] is True
+
+    async def test_authorize_needs_a_login_not_an_api_key(self, async_client, admin_token):
+        app = await _register(async_client, admin_token)
+        key = (
+            await async_client.post("/api/v1/api-keys/", headers=_auth(admin_token), json={"name": "script"})
+        ).json()["key"]
+        _, challenge = _pkce()
+        for headers in ({"X-API-Key": key}, _auth(key)):
+            response = await async_client.post(
+                "/api/v1/connect/authorize",
+                headers=headers,
+                json={
+                    "client_id": app["client_id"],
+                    "redirect_uri": CALLBACK,
+                    "code_challenge": challenge,
+                    "code_challenge_method": "S256",
+                },
+            )
+            assert response.status_code == 401
+
+    async def test_callback_must_match_exactly_at_authorize(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        _, challenge = _pkce()
+        for wrong in (CALLBACK + "/", CALLBACK + "?x=1", "http://evil.example/auth/callback"):
+            response = await _authorize(async_client, operator["token"], app, challenge, redirect_uri=wrong)
+            assert response.status_code == 400
+
+    async def test_only_s256_is_accepted(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        verifier, _ = _pkce()
+        response = await _authorize(async_client, operator["token"], app, verifier[:43], code_challenge_method="plain")
+        assert response.status_code == 422
+
+
+class TestCodeIsWorthLittle:
+    async def test_code_is_single_use(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        code, verifier = await _code(async_client, operator["token"], app)
+        assert (await _exchange(async_client, app, code, verifier)).status_code == 200
+        replay = await _exchange(async_client, app, code, verifier)
+        assert replay.status_code == 400
+        assert replay.json()["detail"] == {"error": "invalid_grant"}
+
+    async def test_expired_code_is_refused(self, async_client, admin_token, operator, db_session):
+        from sqlalchemy import update
+
+        from backend.app.models.auth_ephemeral import AuthEphemeralToken, TokenType
+
+        app = await _register(async_client, admin_token)
+        code, verifier = await _code(async_client, operator["token"], app)
+        await db_session.execute(
+            update(AuthEphemeralToken)
+            .where(AuthEphemeralToken.token_type == TokenType.CONNECT_CODE)
+            .values(expires_at=datetime.now(timezone.utc) - timedelta(seconds=1))
+        )
+        await db_session.commit()
+        assert (await _exchange(async_client, app, code, verifier)).status_code == 400
+
+    async def test_code_is_not_stored_in_plain(self, async_client, admin_token, operator, db_session):
+        from sqlalchemy import select
+
+        from backend.app.models.auth_ephemeral import AuthEphemeralToken, TokenType
+
+        app = await _register(async_client, admin_token)
+        code, _ = await _code(async_client, operator["token"], app)
+        stored = (
+            await db_session.execute(
+                select(AuthEphemeralToken.token).where(AuthEphemeralToken.token_type == TokenType.CONNECT_CODE)
+            )
+        ).scalar_one()
+        assert stored != code
+
+    async def test_wrong_verifier_spends_the_code(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        code, verifier = await _code(async_client, operator["token"], app)
+        other_verifier, _ = _pkce()
+        assert (await _exchange(async_client, app, code, other_verifier)).status_code == 400
+        # A failed attempt burns the code, so the right verifier can't be tried next.
+        assert (await _exchange(async_client, app, code, verifier)).status_code == 400
+
+    async def test_wrong_callback_at_exchange_is_refused(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        code, verifier = await _code(async_client, operator["token"], app)
+        response = await _exchange(async_client, app, code, verifier, redirect_uri=CALLBACK + "x")
+        assert response.status_code == 400
+
+    async def test_wrong_secret_is_invalid_client(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        code, verifier = await _code(async_client, operator["token"], app)
+        response = await _exchange(async_client, app, code, verifier, client_secret="bbs_wrong")
+        assert response.status_code == 401
+        assert response.json()["detail"] == {"error": "invalid_client"}
+        # The code survives a caller that can't prove who it is.
+        assert (await _exchange(async_client, app, code, verifier)).status_code == 200
+
+    async def test_unknown_client_is_invalid_client(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        code, verifier = await _code(async_client, operator["token"], app)
+        response = await _exchange(async_client, app, code, verifier, client_id="bba_unknown")
+        assert response.status_code == 401
+
+    async def test_one_apps_code_is_useless_to_another(self, async_client, admin_token, operator):
+        app_a = await _register(async_client, admin_token, name="A")
+        app_b = await _register(async_client, admin_token, name="B")
+        code, verifier = await _code(async_client, operator["token"], app_a)
+        assert (await _exchange(async_client, app_b, code, verifier)).status_code == 400
+
+    async def test_user_disabled_after_consent_is_refused(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        code, verifier = await _code(async_client, operator["token"], app)
+        await async_client.patch(
+            f"/api/v1/users/{operator['id']}", headers=_auth(admin_token), json={"is_active": False}
+        )
+        assert (await _exchange(async_client, app, code, verifier)).status_code == 400
+
+
+class TestAppLifecycle:
+    async def test_disabled_app_can_neither_authorize_nor_exchange(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        code, verifier = await _code(async_client, operator["token"], app)
+        await async_client.patch(
+            f"/api/v1/connect/apps/{app['id']}", headers=_auth(admin_token), json={"enabled": False}
+        )
+        _, challenge = _pkce()
+        assert (await _authorize(async_client, operator["token"], app, challenge)).status_code == 400
+        assert (await _exchange(async_client, app, code, verifier)).status_code == 401
+
+    async def test_rotated_secret_replaces_the_old_one(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        rotated = (
+            await async_client.post(f"/api/v1/connect/apps/{app['id']}/rotate-secret", headers=_auth(admin_token))
+        ).json()
+        assert rotated["client_secret"] != app["client_secret"]
+        code, verifier = await _code(async_client, operator["token"], app)
+        assert (await _exchange(async_client, app, code, verifier)).status_code == 401
+        assert (
+            await _exchange(async_client, app, code, verifier, client_secret=rotated["client_secret"])
+        ).status_code == 200
+
+    async def test_changed_callback_invalidates_codes_for_the_old_one(self, async_client, admin_token, operator):
+        app = await _register(async_client, admin_token)
+        code, verifier = await _code(async_client, operator["token"], app)
+        await async_client.patch(
+            f"/api/v1/connect/apps/{app['id']}",
+            headers=_auth(admin_token),
+            json={"redirect_uri": "http://orders.local:9000/cb"},
+        )
+        assert (await _exchange(async_client, app, code, verifier)).status_code == 400
+
+    async def test_deleting_the_app_removes_grants_and_codes(self, async_client, admin_token, operator, db_session):
+        from sqlalchemy import func, select
+
+        from backend.app.models.auth_ephemeral import AuthEphemeralToken, TokenType
+        from backend.app.models.connected_app import ConnectedAppGrant
+
+        app = await _register(async_client, admin_token)
+        await _code(async_client, operator["token"], app)
+        response = await async_client.delete(f"/api/v1/connect/apps/{app['id']}", headers=_auth(admin_token))
+        assert response.status_code == 200
+        grants = await db_session.execute(select(func.count()).select_from(ConnectedAppGrant))
+        codes = await db_session.execute(
+            select(func.count())
+            .select_from(AuthEphemeralToken)
+            .where(AuthEphemeralToken.token_type == TokenType.CONNECT_CODE)
+        )
+        assert grants.scalar_one() == 0
+        assert codes.scalar_one() == 0
+
+
+class TestAuthenticationDisabled:
+    async def test_authorize_and_token_say_auth_disabled(self, async_client):
+        info = await async_client.get(
+            "/api/v1/connect/authorize/info", params={"client_id": "bba_x", "redirect_uri": CALLBACK}
+        )
+        assert info.status_code == 409
+        assert info.json()["detail"] == "auth_disabled"
+        verifier, _ = _pkce()
+        token = await async_client.post(
+            "/api/v1/connect/token",
+            json={
+                "grant_type": "authorization_code",
+                "code": "x",
+                "redirect_uri": CALLBACK,
+                "client_id": "bba_x",
+                "client_secret": "bbs_x",
+                "code_verifier": verifier,
+            },
+        )
+        assert token.status_code == 400
+        assert token.json()["detail"] == {"error": "auth_disabled"}
+
+
+class TestRateLimit:
+    async def test_repeated_failures_lock_the_client_out(self, async_client, admin_token, operator):
+        from backend.app.api.routes.connected_apps import MAX_FAILED_TOKEN_EXCHANGES
+
+        app = await _register(async_client, admin_token)
+        verifier, _ = _pkce()
+        for _ in range(MAX_FAILED_TOKEN_EXCHANGES):
+            response = await _exchange(async_client, app, "not-a-code", verifier)
+            assert response.status_code == 400
+        code, good_verifier = await _code(async_client, operator["token"], app)
+        assert (await _exchange(async_client, app, code, good_verifier)).status_code == 429

backend/tests/integration/test_failure_reason_vocabulary_migration.py

@@ -168,3 +168,84 @@ async def test_running_twice_changes_nothing(db_session: AsyncSession, model) ->
     first = await _read(db_session, model, ids)
     await _run(db_session)
     assert await _read(db_session, model, ids) == first == ["layerShift", "Custom legacy reason"]
+
+
+# ---------------------------------------------------------------------------
+# Position within run_migrations
+# ---------------------------------------------------------------------------
+
+
+async def test_upgrade_from_before_print_log_failure_reason_boots(tmp_path) -> None:
+    """A database older than #1378 has no print_log_entries.failure_reason.
+
+    The conversion used to run at the top of run_migrations, before the ALTER
+    that adds that column, so upgrading such a database crashed startup with
+    "no such column: failure_reason". Run the real migration sequence over a
+    database shaped like that and require it to finish, add the column, and
+    still convert the labels that were already there.
+    """
+    from sqlalchemy import text
+    from sqlalchemy.ext.asyncio import create_async_engine
+
+    from backend.app.core.database import Base, run_migrations
+
+    # Register every model on Base.metadata, as init_db does before create_all;
+    # otherwise tables the migrations ALTER would be missing.
+    from backend.app.main import app  # noqa: F401
+
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'old.db'}")
+    try:
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+            await conn.execute(text("ALTER TABLE print_log_entries DROP COLUMN failure_reason"))
+        async with AsyncSession(engine) as session:
+            session.add(PrintArchive(status="failed", failure_reason="Layer shift", **_REQUIRED["PrintArchive"]))
+            await session.commit()
+
+        async with engine.begin() as conn:
+            await run_migrations(conn)
+
+        async with engine.connect() as conn:
+            columns = {row[1] for row in (await conn.execute(text("PRAGMA table_info(print_log_entries)"))).all()}
+            reason = (await conn.execute(text("SELECT failure_reason FROM print_archives"))).scalar_one()
+        assert "failure_reason" in columns
+        assert reason == "layerShift"
+    finally:
+        await engine.dispose()
+
+
+async def test_archive_missing_from_the_fts_index_is_converted(tmp_path) -> None:
+    """Archives created before archive_fts existed were never indexed.
+
+    Updating such a row fires the FTS 'delete' trigger for a row the index
+    doesn't hold, which SQLite reports as "database disk image is malformed" --
+    fatal at startup. The conversion must rebuild the index first.
+    """
+    from sqlalchemy import text
+    from sqlalchemy.ext.asyncio import create_async_engine
+
+    from backend.app.core.database import Base, run_migrations
+
+    # Register every model on Base.metadata, as init_db does before create_all.
+    from backend.app.main import app  # noqa: F401
+
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'fts.db'}")
+    try:
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+            await run_migrations(conn)  # creates archive_fts and its triggers
+        async with AsyncSession(engine) as session:
+            session.add(PrintArchive(status="failed", failure_reason="Layer shift", **_REQUIRED["PrintArchive"]))
+            await session.commit()
+        async with engine.begin() as conn:
+            # Empty the index: the archive is now what a pre-FTS row looks like.
+            await conn.execute(text("INSERT INTO archive_fts(archive_fts) VALUES('delete-all')"))
+
+        async with engine.begin() as conn:
+            await _migrate_failure_reason_vocabulary(conn)
+
+        async with engine.connect() as conn:
+            reason = (await conn.execute(text("SELECT failure_reason FROM print_archives"))).scalar_one()
+        assert reason == "layerShift"
+    finally:
+        await engine.dispose()

backend/tests/integration/test_users_auth_cleanup.py

@@ -178,6 +178,51 @@ class TestDeleteUserCleansAuthRows:
             "LongLivedToken orphan — camera-stream secret still in DB after user delete"
         )
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_delete_user_removes_connected_app_consent_and_codes(
+        self,
+        async_client: AsyncClient,
+        db_session: AsyncSession,
+        auth_token: str,
+    ):
+        """A leftover consent row would skip the consent screen for the next
+        user given the same id; a live code is keyed by username."""
+        from backend.app.models.auth_ephemeral import AuthEphemeralToken, TokenType
+        from backend.app.models.connected_app import ConnectedApp, ConnectedAppGrant
+
+        user_id = await self._create_user(async_client, auth_token, "appclean")
+        app = ConnectedApp(
+            name="Orders", client_id="bba_cleanup", client_secret_hash="x", redirect_uri="http://o.local/cb"
+        )
+        db_session.add(app)
+        await db_session.flush()
+        db_session.add(ConnectedAppGrant(app_id=app.id, user_id=user_id))
+        db_session.add(
+            AuthEphemeralToken.new_connect_code(
+                code_hash="h" * 64,
+                username="AppClean",
+                app_id=app.id,
+                code_challenge="c" * 43,
+                expires_at=datetime.now(timezone.utc) + timedelta(seconds=60),
+            )
+        )
+        await db_session.commit()
+
+        resp = await async_client.delete(
+            f"/api/v1/users/{user_id}",
+            headers={"Authorization": f"Bearer {auth_token}"},
+        )
+        assert resp.status_code == 204
+
+        await db_session.commit()
+        grants = await db_session.execute(select(ConnectedAppGrant).where(ConnectedAppGrant.user_id == user_id))
+        assert grants.scalar_one_or_none() is None
+        codes = await db_session.execute(
+            select(AuthEphemeralToken).where(AuthEphemeralToken.token_type == TokenType.CONNECT_CODE)
+        )
+        assert codes.scalar_one_or_none() is None
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_delete_user_removes_user_otp_codes(

backend/tests/unit/test_route_auth_coverage.py

@@ -70,6 +70,10 @@ _PUBLIC_ROUTES: frozenset[tuple[str, str]] = frozenset(
         ("GET", "/api/v1/auth/oidc/authorize/{provider_id}"),
         ("GET", "/api/v1/auth/oidc/callback"),
         ("POST", "/api/v1/auth/oidc/exchange"),
+        # Connected-app token exchange — called by the app's server, not a browser. The client
+        # secret (bcrypt-verified) plus a single-use, 60 s, PKCE-bound code in the body are the auth;
+        # rate-limited per client and per IP. Refuses outright while Bambuddy auth is disabled.
+        ("POST", "/api/v1/connect/token"),
         # 2FA send + verify — issued after password check; pre-auth token in cookie is the auth.
         ("POST", "/api/v1/auth/2fa/email/send"),
         ("POST", "/api/v1/auth/2fa/verify"),