diff --git a/alembic/versions/0045_repair_contact_migration.py b/alembic/versions/0045_repair_contact_migration.py new file mode 100644 index 0000000..97857bd --- /dev/null +++ b/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)") diff --git a/app/config.py b/app/config.py index dfc571a..5b6487b 100644 --- a/app/config.py +++ b/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 diff --git a/app/core/auth.py b/app/core/auth.py index c2195dd..aadd093 100644 --- a/app/core/auth.py +++ b/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, diff --git a/app/core/middleware.py b/app/core/middleware.py index fc0d836..1b525ee 100644 --- a/app/core/middleware.py +++ b/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,30 +107,27 @@ 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: - raw = await redis.get(f"session:{session_id}") - if raw is None: - return JSONResponse( - status_code=status.HTTP_403_FORBIDDEN, - content={"detail": "Session expired for CSRF validation", "code": "csrf_session_expired"}, - ) + redis = get_redis() + raw = await redis.get(f"session:{session_id}") + if raw is None: + return JSONResponse( + status_code=status.HTTP_403_FORBIDDEN, + content={"detail": "Session expired for CSRF validation", "code": "csrf_session_expired"}, + ) - session_data = json.loads(raw) - stored_token = session_data.get("csrf_token") + session_data = json.loads(raw) + stored_token = session_data.get("csrf_token") - if not stored_token or stored_token != csrf_header: - return JSONResponse( - status_code=status.HTTP_403_FORBIDDEN, - content={"detail": "CSRF token mismatch", "code": "csrf_token_mismatch"}, - ) + if not stored_token or stored_token != csrf_header: + return JSONResponse( + status_code=status.HTTP_403_FORBIDDEN, + content={"detail": "CSRF token mismatch", "code": "csrf_token_mismatch"}, + ) - # 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() + # Sliding session: also extend TTL on CSRF-validated unsafe requests + await redis.expire(f"session:{session_id}", settings.session_ttl_seconds) return await call_next(request) diff --git a/app/core/rate_limit.py b/app/core/rate_limit.py index 3e9e95d..bd98b47 100644 --- a/app/core/rate_limit.py +++ b/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.""" - forwarded = request.headers.get("x-forwarded-for") - if forwarded: - return forwarded.split(",")[0].strip() - return request.client.host if request.client else "unknown" + """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() + # 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 diff --git a/app/core/storage.py b/app/core/storage.py index edc3188..759f087 100644 --- a/app/core/storage.py +++ b/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) diff --git a/app/main.py b/app/main.py index 1172b8d..f414fb8 100644 --- a/app/main.py +++ b/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 ── diff --git a/app/plugins/builtins/ai_ui_control/routes.py b/app/plugins/builtins/ai_ui_control/routes.py index 66b4c16..94ee491 100644 --- a/app/plugins/builtins/ai_ui_control/routes.py +++ b/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") diff --git a/app/plugins/builtins/kommunikation/routes.py b/app/plugins/builtins/kommunikation/routes.py index 8f9e859..8d1cd95 100644 --- a/app/plugins/builtins/kommunikation/routes.py +++ b/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") diff --git a/app/routes/errors.py b/app/routes/errors.py index b04c793..c35497f 100644 --- a/app/routes/errors.py +++ b/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: