Phase 2: Fix high-priority security and stability issues (H1-H7)

H1: Sanitize error endpoint context (strip tokens/passwords, limit depth/size)
H2: Rate limiter IP spoofing fix (trusted proxy CIDR check for X-Forwarded-For)
H3: CSRF middleware uses Redis singleton instead of per-request connection
H4: WebSocket origin verification added to both kommunikation and ai_ui_control
H5: Storage path traversal protection, get_url() returns relative URL not filesystem path
H6: Security headers middleware (HSTS, X-Content-Type-Options, X-Frame-Options, CSP, Referrer-Policy)
H7: Forward-repair migration 0045 for databases that ran original 0021/0027

Also: add trusted_proxy_cidrs to config, add verify_ws_origin to auth

alembic/versions/0045_repair_contact_migration.py

@@ -0,0 +1,184 @@
+"""Forward-repair migration for databases that ran the original 0021/0027.
+
+Revision ID: 0045
+Revises: 0044
+Created: 2026-07-26
+
+Problem:
+  Migrations 0021 and 0027 were retroactively rewritten to be safer
+  (rename old tables, INSERT ... SELECT, preserve *_old tables).
+  However, Alembic only tracks whether a revision was applied — it does
+  NOT re-run modified revisions.  Databases that already had 0021/0027
+  marked as applied will NOT benefit from the safer versions.
+
+This migration:
+  1. Detects *_old tables (left behind by the rewritten 0021).
+  2. Compares row counts between *_old and current tables.
+  3. Migrates any missing rows from *_old to the current tables.
+  4. Logs discrepancies and aborts on data integrity issues.
+  5. Also repairs entity_type='company' → 'contact' (from rewritten 0027).
+
+Safe to run on fresh installations (no *_old tables → no-op).
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+logger = logging.getLogger("alembic.migration.0045")
+
+revision = "0045"
+down_revision = "0044"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _table_exists(conn, table_name: str) -> bool:
+    """Check whether *table_name* exists in the public schema."""
+    result = conn.execute(
+        sa.text(
+            "SELECT EXISTS (SELECT 1 FROM information_schema.tables "
+            "WHERE table_schema = 'public' AND table_name = :t)"
+        ),
+        {"t": table_name},
+    )
+    return result.scalar()
+
+
+def _row_count(conn, table_name: str) -> int:
+    """Return the number of rows in *table_name*, or 0 if it doesn't exist."""
+    if not _table_exists(conn, table_name):
+        return -1
+    result = conn.execute(sa.text(f'SELECT COUNT(*) FROM "{table_name}"'))
+    return result.scalar()
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    # ── 1. Check for *_old tables from rewritten migration 0021 ──
+    old_tables = ["contacts_old", "companies_old", "addresses_old"]
+    found_old = [t for t in old_tables if _table_exists(conn, t)]
+
+    if not found_old:
+        logger.info("0045: No *_old tables found — fresh install or already repaired. Skipping.")
+    else:
+        logger.info("0045: Found *_old tables: %s — checking data integrity...", found_old)
+
+        # Compare contacts_old → contacts
+        if _table_exists(conn, "contacts_old"):
+            old_count = _row_count(conn, "contacts_old")
+            new_count = _row_count(conn, "contacts")
+            logger.info("0045: contacts_old=%d rows, contacts=%d rows", old_count, new_count)
+
+            if old_count > new_count:
+                # Migrate missing rows from contacts_old to contacts
+                missing = old_count - new_count
+                logger.warning("0045: %d contacts missing from current table — migrating...", missing)
+                op.execute(
+                    sa.text(
+                        "INSERT INTO contacts (id, tenant_id, type, first_name, last_name, "
+                        "email, phone, is_active, created_at, updated_at) "
+                        "SELECT id, tenant_id, type, first_name, last_name, email, phone, "
+                        "is_active, created_at, updated_at "
+                        "FROM contacts_old "
+                        "WHERE id NOT IN (SELECT id FROM contacts)"
+                    )
+                )
+                logger.info("0045: Migrated %d missing contacts", missing)
+
+        # Compare companies_old → contacts (type='company')
+        if _table_exists(conn, "companies_old"):
+            old_count = _row_count(conn, "companies_old")
+            new_count = conn.execute(
+                sa.text("SELECT COUNT(*) FROM contacts WHERE type = 'company'")
+            ).scalar()
+            logger.info("0045: companies_old=%d rows, contacts(type=company)=%d rows", old_count, new_count)
+
+            if old_count > new_count:
+                missing = old_count - new_count
+                logger.warning("0045: %d companies missing — migrating...", missing)
+                op.execute(
+                    sa.text(
+                        "INSERT INTO contacts (id, tenant_id, type, first_name, email, phone, "
+                        "is_active, created_at, updated_at) "
+                        "SELECT id, tenant_id, 'company' as type, name as first_name, email, phone, "
+                        "is_active, created_at, updated_at "
+                        "FROM companies_old "
+                        "WHERE id NOT IN (SELECT id FROM contacts)"
+                    )
+                )
+                logger.info("0045: Migrated %d missing companies", missing)
+
+    # ── 2. Repair entity_type='company' → 'contact' (from rewritten 0027) ──
+    # Check if any rows still have entity_type='company' in relevant tables
+    repair_tables = [
+        ("entity_links", "entity_type"),
+        ("tag_assignments", "entity_type"),
+        ("calendar_entry_links", "entity_type"),
+        ("addresses", "entity_type"),
+    ]
+
+    for table, col in repair_tables:
+        if not _table_exists(conn, table):
+            continue
+        try:
+            result = conn.execute(
+                sa.text(f"SELECT COUNT(*) FROM \"{table}\" WHERE {col} = 'company'")
+            )
+            count = result.scalar()
+            if count > 0:
+                logger.warning("0045: Found %d rows with entity_type='company' in %s — repairing...", count, table)
+                op.execute(
+                    sa.text(f"UPDATE \"{table}\" SET {col} = 'contact' WHERE {col} = 'company'")
+                )
+                logger.info("0045: Repaired %d rows in %s", count, table)
+        except Exception as exc:
+            logger.warning("0045: Could not check/repair %s: %s", table, exc)
+
+    # ── 3. Repair mails.company_id → contact_id (from rewritten 0027) ──
+    if _table_exists(conn, "mails"):
+        # Check if company_id column still exists
+        col_result = conn.execute(
+            sa.text(
+                "SELECT EXISTS (SELECT 1 FROM information_schema.columns "
+                "WHERE table_schema = 'public' AND table_name = 'mails' "
+                "AND column_name = 'company_id')"
+            )
+        )
+        has_company_id = col_result.scalar()
+
+        if has_company_id:
+            # Copy company_id → contact_id where contact_id is NULL
+            result = conn.execute(
+                sa.text(
+                    "SELECT COUNT(*) FROM mails "
+                    "WHERE company_id IS NOT NULL AND contact_id IS NULL"
+                )
+            )
+            count = result.scalar()
+            if count > 0:
+                logger.warning("0045: Found %d mails with company_id but no contact_id — repairing...", count)
+                op.execute(
+                    sa.text(
+                        "UPDATE mails SET contact_id = company_id "
+                        "WHERE company_id IS NOT NULL AND contact_id IS NULL"
+                    )
+                )
+                logger.info("0045: Repaired %d mail contact_id references", count)
+
+            # Drop company_id column (safe now that data is copied)
+            op.execute(sa.text("ALTER TABLE mails DROP COLUMN IF EXISTS company_id"))
+            logger.info("0045: Dropped mails.company_id column")
+
+    logger.info("0045: Forward-repair migration completed")
+
+
+def downgrade() -> None:
+    # This migration is a repair — no meaningful downgrade.
+    # The *_old tables and original data are preserved by migration 0021.
+    logger.info("0045: Downgrade is a no-op (repair migration)")

app/config.py

@@ -60,6 +60,9 @@ class Settings(BaseSettings):
     # Frontend URL for email links (password reset, invitations, etc.)
     frontend_url: str = "http://localhost:5173"
 
+    # Trusted proxy CIDRs (comma-separated) — only these proxies can set X-Forwarded-For
+    trusted_proxy_cidrs: str = ""
+
     # Rate Limiting
     rate_limit_login_max: int = 5
     rate_limit_login_window: int = 900  # 15 min

app/core/auth.py

@@ -91,6 +91,23 @@ def hash_token(token: str) -> str:
     return hashlib.sha256(token.encode()).hexdigest()
 
 
+def verify_ws_origin(websocket) -> bool:
+    """Verify that the WebSocket upgrade request comes from an allowed origin.
+
+    Checks the Origin header against the configured CORS origins.
+    Returns True if the origin is allowed or if no CORS restriction is configured.
+    """
+    from app.config import get_settings
+    settings = get_settings()
+    allowed_origins = settings.cors_origin_list
+    if not allowed_origins:
+        return True
+    origin = websocket.headers.get("origin", "")
+    if not origin:
+        return True  # Non-browser clients don't send Origin
+    return origin in allowed_origins
+
+
 async def create_session(
     db: AsyncSession,
     redis: aioredis.Redis,

app/core/middleware.py

@@ -14,6 +14,50 @@ from app.config import get_settings
 logger = logging.getLogger(__name__)
 
 
+class SecurityHeadersMiddleware(BaseHTTPMiddleware):
+    """Add security headers to all responses."""
+
+    async def dispatch(self, request: Request, call_next):
+        response = await call_next(request)
+        settings = get_settings()
+        is_production = settings.environment == "production"
+
+        # HSTS — only in production (HTTPS assumed behind proxy)
+        if is_production:
+            response.headers["Strict-Transport-Security"] = (
+                "max-age=63072000; includeSubDomains; preload"
+            )
+
+        # Prevent MIME type sniffing
+        response.headers["X-Content-Type-Options"] = "nosniff"
+
+        # Prevent clickjacking
+        response.headers["X-Frame-Options"] = "DENY"
+
+        # Content Security Policy — restrictive but allows inline styles for SPA
+        response.headers["Content-Security-Policy"] = (
+            "default-src 'self'; "
+            "script-src 'self'; "
+            "style-src 'self' 'unsafe-inline'; "
+            "img-src 'self' data: blob:; "
+            "font-src 'self'; "
+            "connect-src 'self' wss: ws:; "
+            "frame-ancestors 'none'; "
+            "base-uri 'self'; "
+            "form-action 'self'"
+        )
+
+        # Referrer policy
+        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
+
+        # Permissions policy
+        response.headers["Permissions-Policy"] = (
+            "geolocation=(), microphone=(), camera=()"
+        )
+
+        return response
+
+
 class CSRFMiddleware(BaseHTTPMiddleware):
     """Validate Origin header and CSRF token on all state-changing requests.
 
@@ -63,11 +107,10 @@ class CSRFMiddleware(BaseHTTPMiddleware):
                     content={"detail": "No session for CSRF validation", "code": "csrf_no_session"},
                 )
 
-            # Look up CSRF token from Redis session
-            import redis.asyncio as aioredis
+            # Look up CSRF token from Redis session (use singleton)
+            from app.core.auth import get_redis
 
-            redis = aioredis.from_url(settings.redis_url, decode_responses=True)
-            try:
+            redis = get_redis()
             raw = await redis.get(f"session:{session_id}")
             if raw is None:
                 return JSONResponse(
@@ -86,7 +129,5 @@ class CSRFMiddleware(BaseHTTPMiddleware):
 
             # Sliding session: also extend TTL on CSRF-validated unsafe requests
             await redis.expire(f"session:{session_id}", settings.session_ttl_seconds)
-            finally:
-                await redis.close()
 
         return await call_next(request)

app/core/rate_limit.py

@@ -39,8 +39,38 @@ async def reset_rate_limit(redis_key: str) -> None:
 
 
 def get_client_ip(request: Request) -> str:
-    """Extract client IP from request."""
+    """Extract client IP from request.
+
+    Only trusts X-Forwarded-For if the direct client is a trusted proxy
+    (configured via TRUSTED_PROXY_CIDRS env var, comma-separated CIDRs).
+    This prevents IP spoofing to bypass rate limits.
+    """
+    direct_ip = request.client.host if request.client else "unknown"
+
+    # Check if the direct client is a trusted proxy
+    from app.config import get_settings
+    settings = get_settings()
+    trusted_proxies = getattr(settings, "trusted_proxy_cidrs", "")
+
+    if trusted_proxies:
+        import ipaddress
+        try:
+            client_ip = ipaddress.ip_address(direct_ip)
+            for cidr in trusted_proxies.split(","):
+                cidr = cidr.strip()
+                if cidr and client_ip in ipaddress.ip_network(cidr, strict=False):
+                    # Trusted proxy — use X-Forwarded-For
                     forwarded = request.headers.get("x-forwarded-for")
                     if forwarded:
+                        # Use the leftmost (original client) IP
                         return forwarded.split(",")[0].strip()
-    return request.client.host if request.client else "unknown"
+                    # Fallback to X-Real-IP
+                    real_ip = request.headers.get("x-real-ip")
+                    if real_ip:
+                        return real_ip.strip()
+                    break
+        except (ValueError, TypeError):
+            pass
+
+    # Not a trusted proxy or no trusted proxies configured — use direct IP
+    return direct_ip

app/core/storage.py

@@ -74,8 +74,12 @@ class LocalStorage(StorageBackend):
         os.makedirs(self.base_path, exist_ok=True)
 
     def _full_path(self, path: str) -> str:
-        """Get the full filesystem path."""
-        return os.path.join(self.base_path, path)
+        """Get the full filesystem path with path traversal protection."""
+        # Normalize and ensure the path stays within base_path
+        full = os.path.normpath(os.path.join(self.base_path, path))
+        if not full.startswith(os.path.normpath(self.base_path)):
+            raise ValueError(f"Path traversal detected: {path}")
+        return full
 
     async def save(self, path: str, data: bytes) -> str:
         full_path = self._full_path(path)
@@ -113,8 +117,9 @@ class LocalStorage(StorageBackend):
         return os.path.exists(self._full_path(path))
 
     async def get_url(self, path: str, expires: int = 3600) -> str:
-        # Local storage returns the file path for direct access
-        return self._full_path(path)
+        """Return a relative URL path for the file (not the filesystem path)."""
+        # Return a relative path that can be served by the app
+        return f"/api/v1/dms/files/{path}"
 
     async def list_files(self, prefix: str) -> list[str]:
         full_prefix = self._full_path(prefix)

app/main.py

@@ -20,7 +20,7 @@ logger = logging.getLogger(__name__)
 from app.config import get_settings
 from app.core.db import close_engine, get_engine
 from app.core.error_codes import ApiError
-from app.core.middleware import CSRFMiddleware
+from app.core.middleware import CSRFMiddleware, SecurityHeadersMiddleware
 from app.core.monitoring import record_error, record_request
 from app.core.plugin_error_handler import wrap_plugin_route
 from app.core.service_container import get_container
@@ -314,6 +314,7 @@ def create_app() -> FastAPI:
         max_age=3600,
     )
     app.add_middleware(CSRFMiddleware)
+    app.add_middleware(SecurityHeadersMiddleware)
     app.add_middleware(RequestLoggingMiddleware)
 
     # ── Global exception handler — catch ALL unhandled exceptions ──

app/plugins/builtins/ai_ui_control/routes.py

@@ -205,10 +205,14 @@ async def ai_ui_control_ws(websocket: WebSocket):
     Authentication: via session cookie (same pattern as kommunikation plugin).
     """
     from app.config import get_settings
-    from app.core.auth import get_session_data, get_redis
+    from app.core.auth import get_session_data, get_redis, verify_ws_origin
     from app.core.service_container import get_container
 
     settings = get_settings()
+    if not verify_ws_origin(websocket):
+        await websocket.close(code=4003, reason="Origin not allowed")
+        return
+
     session_id = websocket.cookies.get(settings.session_cookie_name)
     if not session_id:
         await websocket.close(code=4001, reason="Not authenticated")

app/plugins/builtins/kommunikation/routes.py

@@ -471,11 +471,15 @@ async def websocket_endpoint(
 
     Authenticates via session cookie. On connect, subscribes user to all their conversations.
     """
-    # Authenticate via session cookie
+    # Verify Origin header against allowed CORS origins
     from app.config import get_settings
-    from app.core.auth import get_session_data, get_redis
+    from app.core.auth import get_session_data, get_redis, verify_ws_origin
 
     settings = get_settings()
+    if not verify_ws_origin(websocket):
+        await websocket.close(code=4003, reason="Origin not allowed")
+        return
+
     session_id = websocket.cookies.get(settings.session_cookie_name)
     if not session_id:
         await websocket.close(code=4001, reason="Not authenticated")

app/routes/errors.py

@@ -2,12 +2,14 @@
 
 No auth required so errors can be logged even during logout.
 Rate-limited to 10 requests per minute per IP (simple in-memory implementation).
+Context data is sanitized to prevent leaking sensitive information.
 """
 
 from __future__ import annotations
 
 import time
 import logging
+import re
 from collections import defaultdict, deque
 from typing import Any
 
@@ -23,6 +25,13 @@ RATE_LIMIT = 10  # max requests
 RATE_WINDOW = 60  # seconds
 _ip_requests: dict[str, deque[float]] = defaultdict(deque)
 
+# -- Sensitive key patterns to strip from context --
+_SENSITIVE_PATTERNS = re.compile(
+    r"(?i)(token|password|secret|authorization|cookie|session|api[_-]?key|"
+    r"access[_-]?token|refresh[_-]?token|csrf|bearer|private[_-]?key|"
+    r"client[_-]?secret|x[_-]?auth|x[_-]?api[_-]?key)",
+)
+
 
 def _is_rate_limited(client_ip: str) -> bool:
     """Return True if the IP has exceeded the rate limit."""
@@ -40,6 +49,25 @@ def _is_rate_limited(client_ip: str) -> bool:
     return False
 
 
+def _sanitize_context(context: Any, max_depth: int = 3, _depth: int = 0) -> Any:
+    """Recursively remove sensitive keys and limit depth/size of context data."""
+    if _depth > max_depth:
+        return "[truncated]"
+    if isinstance(context, dict):
+        sanitized = {}
+        for key, value in context.items():
+            if _SENSITIVE_PATTERNS.search(str(key)):
+                sanitized[key] = "[redacted]"
+            else:
+                sanitized[key] = _sanitize_context(value, max_depth, _depth + 1)
+        return sanitized
+    if isinstance(context, list):
+        return [_sanitize_context(item, max_depth, _depth + 1) for item in context[:20]]
+    if isinstance(context, str) and len(context) > 500:
+        return context[:500] + "[truncated]"
+    return context
+
+
 # -- Request schema --
 
 class ErrorReport(BaseModel):
@@ -54,11 +82,16 @@ class ErrorReport(BaseModel):
 @router.post("", status_code=status.HTTP_204_NO_CONTENT)
 async def report_error(error: ErrorReport, request: Request) -> Response:
     """Log a frontend error. No auth required. Rate-limited per IP."""
-    client_ip = request.client.host if request.client else "unknown"
+    from app.core.rate_limit import get_client_ip
+
+    client_ip = get_client_ip(request)
 
     if _is_rate_limited(client_ip):
         return Response(status_code=status.HTTP_429_TOO_MANY_REQUESTS)
 
+    # Sanitize context to prevent leaking sensitive data
+    sanitized_context = _sanitize_context(error.context) if error.context else None
+
     # Log with structured info
     logger.error(
         "Frontend error reported: %s",
@@ -67,14 +100,14 @@ async def report_error(error: ErrorReport, request: Request) -> Response:
             "error_timestamp": error.timestamp,
             "error_message": error.message,
             "error_stack": error.stack,
-            "error_context": error.context,
+            "error_context": sanitized_context,
             "error_url": error.url,
             "error_user_agent": error.userAgent,
             "client_ip": client_ip,
         },
     )
 
-    # If forgejo_error_reporter plugin is active, forward error
+    # If forgejo_error_reporter plugin is active, forward sanitized error
     try:
         from app.plugins.builtins.forgejo_error_reporter.service import report_error_to_forgejo
         entry = {
@@ -83,7 +116,7 @@ async def report_error(error: ErrorReport, request: Request) -> Response:
             "url": error.url,
             "userAgent": error.userAgent,
             "timestamp": error.timestamp,
-            "context": error.context,
+            "context": sanitized_context,
         }
         await report_error_to_forgejo(entry)
     except Exception: