Merge pull request #707 from christianlouis/copilot/fix-debug-logging-issue

feat: fix DEBUG logging and add LOG_LEVEL/LOG_FORMAT/LOG_SYSLOG for standard log management

.env.demo

@@ -7,6 +7,24 @@ GOTENBERG_URL=http://gotenberg:3000
 ALLOW_FILE_DELETE=true  # Allow deletion of file records
 COMPLIANCE_ENABLED=true  # Enable compliance templates dashboard (GDPR, HIPAA, SOC 2)
 
+# **Logging**
+# LOG_LEVEL controls the Python root-logger level.
+# Accepted values: DEBUG, INFO, WARNING, ERROR, CRITICAL (default: INFO).
+# When DEBUG=true and LOG_LEVEL is not set, the level is automatically lowered to DEBUG.
+# LOG_LEVEL=INFO
+# DEBUG=false
+
+# Log output format: "text" (human-readable, default) or "json" (structured JSON lines).
+# Use "json" when shipping logs to Grafana Loki, Splunk, ELK, Datadog, or any SIEM.
+# LOG_FORMAT=text
+
+# Forward application logs to a syslog receiver (in addition to stdout).
+# Useful for traditional (non-container) deployments and centralised SIEM ingestion.
+# LOG_SYSLOG_ENABLED=false
+# LOG_SYSLOG_HOST=localhost
+# LOG_SYSLOG_PORT=514
+# LOG_SYSLOG_PROTOCOL=udp          # udp | tcp
+
 # **UI / Appearance**
 # Default colour scheme: system (follow OS), light, or dark
 # Individual users can always override with the navbar dark-mode toggle.

app/auth.py

@@ -125,8 +125,17 @@ def get_current_user(request: Request):
     # Check for Bearer token auth first (API tokens)
     api_user = getattr(request.state, "api_token_user", None)
     if isinstance(api_user, dict):
+        logger.debug("[AUTH] get_current_user: resolved from API token (user_id=%s)", api_user.get("id"))
         return api_user
-    return request.session.get("user")
+    session_user = request.session.get("user")
+    if session_user:
+        logger.debug(
+            "[AUTH] get_current_user: resolved from session (user=%s)",
+            session_user.get("preferred_username") or session_user.get("email") or session_user.get("id"),
+        )
+    else:
+        logger.debug("[AUTH] get_current_user: no user in session or API token")
+    return session_user
 
 
 def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
@@ -141,10 +150,12 @@ def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
     """
     auth_header = request.headers.get("authorization", "")
     if not isinstance(auth_header, str) or not auth_header.startswith("Bearer "):
+        logger.debug("[AUTH] _resolve_bearer_user: no Bearer token in Authorization header")
         return None
 
     raw_token = auth_header[7:]
     if not raw_token or not isinstance(raw_token, str):
+        logger.debug("[AUTH] _resolve_bearer_user: empty or invalid token after 'Bearer ' prefix")
         return None
 
     from app.api.api_tokens import hash_token
@@ -153,8 +164,15 @@ def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
     token_hash = hash_token(raw_token)
     db_token = db.query(ApiToken).filter(ApiToken.token_hash == token_hash, ApiToken.is_active.is_(True)).first()
     if db_token is None:
+        logger.debug("[AUTH] _resolve_bearer_user: no active API token matched the provided hash")
         return None
 
+    logger.debug(
+        "[AUTH] _resolve_bearer_user: matched API token id=%s owner=%s",
+        db_token.id,
+        db_token.owner_id,
+    )
+
     # Update usage tracking
     try:
         db_token.last_used_at = datetime.now(timezone.utc)
@@ -205,16 +223,18 @@ def require_login(func):
 
     @wraps(func)
     async def wrapper(request: Request, *args, **kwargs):
+        url_path = urlparse(str(request.url)).path
         # Check session auth first
         if request.session.get("user"):
+            logger.debug("[AUTH] require_login: session auth OK for %s", url_path)
             if inspect.iscoroutinefunction(func):
                 return await func(*args, request=request, **kwargs)
             else:
                 return func(*args, request=request, **kwargs)
 
         # Fall back to Bearer token auth for API endpoints
-        url_path = urlparse(str(request.url)).path
         if url_path.startswith("/api/"):
+            logger.debug("[AUTH] require_login: no session, trying Bearer token for %s", url_path)
             try:
                 from app.database import SessionLocal
 
@@ -228,17 +248,22 @@ def require_login(func):
 
             if api_user:
                 request.state.api_token_user = api_user
+                logger.debug(
+                    "[AUTH] require_login: Bearer token auth OK for %s (user=%s)", url_path, api_user.get("id")
+                )
                 if inspect.iscoroutinefunction(func):
                     return await func(*args, request=request, **kwargs)
                 else:
                     return func(*args, request=request, **kwargs)
 
+            logger.debug("[AUTH] require_login: no valid auth for API endpoint %s — returning 401", url_path)
             return JSONResponse(
                 status_code=status.HTTP_401_UNAUTHORIZED,
                 content={"error": "Not authenticated"},
             )
 
         # Non-API endpoint with no session — redirect to login
+        logger.debug("[AUTH] require_login: no session for %s — redirecting to /login", url_path)
         request.session["redirect_after_login"] = str(request.url)
         return RedirectResponse(url="/login", status_code=status.HTTP_302_FOUND)
 
@@ -307,9 +332,15 @@ async def login(request: Request):
 async def oauth_login(request: Request):
     """Handle OAuth login flow"""
     if not OAUTH_CONFIGURED:
+        logger.debug("[AUTH] oauth_login: OAuth not configured — redirecting to /login")
         return RedirectResponse(url="/login?error=OAuth+not+configured", status_code=status.HTTP_302_FOUND)
 
     redirect_uri = request.url_for("oauth_callback")
+    logger.debug(
+        "[AUTH] oauth_login: initiating Authentik OAuth redirect_uri=%s session_keys=%s",
+        redirect_uri,
+        list(request.session.keys()),
+    )
     return await oauth.authentik.authorize_redirect(request, redirect_uri)
 
 
@@ -324,13 +355,23 @@ async def social_login(request: Request, provider: str):
         A redirect to the provider's authorization page, or back to /login on error.
     """
     if provider not in SOCIAL_PROVIDERS:
+        logger.debug(
+            "[AUTH] social_login: unknown provider=%r (registered=%s)", provider, list(SOCIAL_PROVIDERS.keys())
+        )
         return RedirectResponse(url="/login?error=Unknown+social+provider", status_code=status.HTTP_302_FOUND)
 
     redirect_uri = request.url_for("social_callback", provider=provider)
     oauth_client = getattr(oauth, provider, None)
     if oauth_client is None:
+        logger.debug("[AUTH] social_login: provider=%r registered but OAuth client not configured", provider)
         return RedirectResponse(url="/login?error=Provider+not+configured", status_code=status.HTTP_302_FOUND)
 
+    logger.debug(
+        "[AUTH] social_login: initiating %s OAuth, redirect_uri=%s session_keys=%s",
+        provider,
+        redirect_uri,
+        list(request.session.keys()),
+    )
     return await oauth_client.authorize_redirect(request, redirect_uri)
 
 
@@ -389,27 +430,39 @@ async def social_callback(request: Request, provider: str, db: Session = Depends
         A redirect to the user's original destination or the upload page.
     """
     if provider not in SOCIAL_PROVIDERS:
+        logger.debug("[AUTH] social_callback: unknown provider=%r", provider)
         return RedirectResponse(url="/login?error=Unknown+social+provider", status_code=status.HTTP_302_FOUND)
 
     oauth_client = getattr(oauth, provider, None)
     if oauth_client is None:
+        logger.debug("[AUTH] social_callback: provider=%r not configured", provider)
         return RedirectResponse(url="/login?error=Provider+not+configured", status_code=status.HTTP_302_FOUND)
 
     try:
+        logger.debug("[AUTH] social_callback: exchanging auth code for provider=%s", provider)
         token = await oauth_client.authorize_access_token(request)
 
         # Try standard OIDC userinfo first, fall back to token-embedded userinfo
         raw_userinfo = token.get("userinfo")
         if not raw_userinfo:
+            logger.debug("[AUTH] social_callback: no userinfo in token, fetching from userinfo endpoint")
             try:
                 resp = await oauth_client.userinfo(token=token)
                 raw_userinfo = resp if isinstance(resp, dict) else resp.json() if hasattr(resp, "json") else {}
             except Exception:
+                logger.debug("[AUTH] social_callback: userinfo endpoint failed, using empty dict", exc_info=True)
                 raw_userinfo = {}
 
         user_data = _normalize_social_userinfo(provider, token, raw_userinfo)
+        logger.debug(
+            "[AUTH] social_callback: normalized user_data email=%s sub=%s provider=%s",
+            user_data.get("email"),
+            user_data.get("sub"),
+            provider,
+        )
 
         if not user_data.get("email"):
+            logger.debug("[AUTH] social_callback: no email in user_data — aborting")
             return RedirectResponse(
                 url="/login?error=Could+not+retrieve+email+from+provider",
                 status_code=status.HTTP_302_FOUND,
@@ -442,21 +495,29 @@ async def social_callback(request: Request, provider: str, db: Session = Depends
         )
 
         # Mobile app flow: issue an inline API token and redirect back to the app.
+        logger.debug(
+            "[MOBILE] social_callback: checking for mobile redirect (session has mobile_redirect_uri=%s)",
+            "mobile_redirect_uri" in request.session,
+        )
         mobile_resp = _create_mobile_redirect(request, db)
         if mobile_resp:
+            logger.info("[MOBILE] social_callback: returning mobile redirect response for provider=%s", provider)
             return mobile_resp
 
         if user_id:
             profile = db.query(_UserProfile).filter(_UserProfile.user_id == user_id).first()
             if profile and not profile.onboarding_completed:
+                logger.debug("[AUTH] social_callback: user=%s needs onboarding, redirecting", user_id)
                 post_onboarding = request.session.pop("redirect_after_login", "/upload")
                 request.session["post_onboarding_redirect"] = post_onboarding
                 return RedirectResponse(url="/onboarding", status_code=status.HTTP_302_FOUND)
 
         redirect_url = request.session.pop("redirect_after_login", "/upload")
+        logger.debug("[AUTH] social_callback: login complete, redirecting to %s", redirect_url)
         return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)
     except Exception as e:
         logger.warning("[SECURITY] SOCIAL_LOGIN_FAILURE provider=%s error=%s", provider, type(e).__name__)
+        logger.debug("[AUTH] social_callback: full exception for provider=%s", provider, exc_info=True)
         return RedirectResponse(
             url="/login?error=Social+login+failed.+Please+try+again.", status_code=status.HTTP_302_FOUND
         )
@@ -561,15 +622,23 @@ def _ensure_user_profile(db: Session, user_data: dict, is_admin: bool = False) -
 async def oauth_callback(request: Request, db: Session = Depends(get_db)):
     """Handle OAuth callback from provider"""
     try:
+        logger.debug("[AUTH] oauth_callback: exchanging authorization code for token")
         token = await oauth.authentik.authorize_access_token(request)
         userinfo = token.get("userinfo")
         if not userinfo:
+            logger.debug("[AUTH] oauth_callback: no userinfo in token response — aborting")
             return RedirectResponse(
                 url="/login?error=Failed+to+retrieve+user+information", status_code=status.HTTP_302_FOUND
             )
 
         # Store user info in session
         user_data = dict(userinfo)
+        logger.debug(
+            "[AUTH] oauth_callback: received userinfo email=%s sub=%s groups=%s",
+            user_data.get("email"),
+            user_data.get("sub"),
+            user_data.get("groups", []),
+        )
 
         # Add Gravatar picture if no picture is provided
         if not user_data.get("picture") and user_data.get("email"):
@@ -584,6 +653,12 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)):
             groups = user_data.get("groups", [])
             admin_group = (settings.admin_group_name or "admin").strip().lower()
             is_admin = admin_group in [group.lower() for group in groups]
+            logger.debug(
+                "[AUTH] oauth_callback: admin group check — looking for %r in %s → is_admin=%s",
+                admin_group,
+                [g.lower() for g in groups],
+                is_admin,
+            )
 
         # Set is_admin flag (defaults to False for OAuth users unless they're in admin group)
         user_data["is_admin"] = is_admin
@@ -623,15 +698,18 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)):
         if user_id:
             profile = db.query(_UserProfile).filter(_UserProfile.user_id == user_id).first()
             if profile and not profile.onboarding_completed:
+                logger.debug("[AUTH] oauth_callback: user=%s needs onboarding, redirecting", user_id)
                 post_onboarding = request.session.pop("redirect_after_login", "/upload")
                 request.session["post_onboarding_redirect"] = post_onboarding
                 return RedirectResponse(url="/onboarding", status_code=status.HTTP_302_FOUND)
 
         # Redirect to original destination or default
         redirect_url = request.session.pop("redirect_after_login", "/upload")
+        logger.debug("[AUTH] oauth_callback: login complete, redirecting to %s", redirect_url)
         return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)
     except Exception as e:
         logger.warning(f"[SECURITY] OAUTH_LOGIN_FAILURE error={type(e).__name__}")
+        logger.debug("[AUTH] oauth_callback: full exception details", exc_info=True)
         return RedirectResponse(url=f"/login?error=Authentication+failed:+{str(e)}", status_code=status.HTTP_302_FOUND)
 
 
@@ -931,6 +1009,7 @@ async def logout(request: Request, db: Session = Depends(get_db)):
     username = "unknown"
     if isinstance(user, dict):
         username = user.get("preferred_username") or user.get("email") or "unknown"
+    logger.debug("[AUTH] logout: clearing session for user=%s client_ip=%s", username, get_client_ip(request))
     logger.info(f"[SECURITY] LOGOUT user={username}")
     try:
         from app.utils.audit_service import record_event

app/config.py

@@ -48,6 +48,51 @@ class Settings(BaseSettings):
     workdir: str
     debug: bool = False  # Default to False
 
+    # Logging level for the application.  Accepts standard Python level names:
+    # DEBUG, INFO, WARNING, ERROR, CRITICAL.  When *debug* is True and
+    # *log_level* has not been explicitly set, the effective level is forced to
+    # DEBUG so that all ``logger.debug()`` calls produce output.
+    log_level: str = Field(
+        default="INFO",
+        description=(
+            "Python logging level for the application root logger.  "
+            "Accepts: DEBUG, INFO, WARNING, ERROR, CRITICAL.  "
+            "When DEBUG=True and LOG_LEVEL is not explicitly set, "
+            "the effective level is automatically lowered to DEBUG."
+        ),
+    )
+
+    # Log output format.  ``text`` is the human-readable default.
+    # ``json`` emits one JSON object per line, ideal for log collectors
+    # (Promtail, Fluentd, Filebeat, Datadog agent) and SIEM ingestion.
+    log_format: str = Field(
+        default="text",
+        description=(
+            "Log output format: 'text' (human-readable, default) or "
+            "'json' (structured JSON lines for SIEM / log aggregation)."
+        ),
+    )
+
+    # Optional syslog forwarding for application logs (not just audit events).
+    # When enabled, a Python SysLogHandler is added to the root logger so that
+    # every log message is also sent to the configured syslog receiver.
+    log_syslog_enabled: bool = Field(
+        default=False,
+        description="Forward application logs to a syslog receiver in addition to stdout.",
+    )
+    log_syslog_host: str = Field(
+        default="localhost",
+        description="Hostname or IP of the syslog receiver for application logs.",
+    )
+    log_syslog_port: int = Field(
+        default=514,
+        description="Port of the syslog receiver for application logs.",
+    )
+    log_syslog_protocol: str = Field(
+        default="udp",
+        description="Protocol for syslog transport: 'udp' or 'tcp'.",
+    )
+
     # Making Dropbox optional
     dropbox_enabled: bool = Field(
         default=True,

app/main.py

@@ -1,8 +1,11 @@
 #!/usr/bin/env python3
+import json as _json_mod
 import logging
 import os
 import pathlib
 from contextlib import asynccontextmanager
+from datetime import datetime as _dt
+from datetime import timezone as _tz
 
 from fastapi import FastAPI, HTTPException, Request, status
 from fastapi.middleware.cors import CORSMiddleware
@@ -36,6 +39,114 @@ from app.views import router as frontend_router
 # Explicitly include the files router
 from app.views.files import router as files_router
 
+# ---------------------------------------------------------------------------
+# Configure Python root logging level early so that *all* loggers (including
+# those already created via ``logging.getLogger(__name__)`` in other modules)
+# respect the configured level.
+#
+# Standard behaviour (matches Django, Flask, 12-factor conventions):
+#   • ``LOG_LEVEL`` env var takes precedence when explicitly set.
+#   • When ``DEBUG=True`` and ``LOG_LEVEL`` is **not** set, the effective
+#     level is automatically lowered to ``DEBUG``.
+#   • Default (neither flag set): ``INFO``.
+#
+# ``LOG_FORMAT=json`` enables structured JSON lines on stdout, suitable for
+# Promtail, Fluentd, Filebeat, Datadog, Splunk UF, or any log collector.
+#
+# ``LOG_SYSLOG_ENABLED=true`` adds a Python SysLogHandler so that every log
+# message is also forwarded to the configured syslog receiver — useful for
+# traditional (non-container) deployments and centralised SIEM ingestion.
+#
+# Noisy third-party loggers (httpx, httpcore, authlib, etc.) are pinned to
+# WARNING when the app-level is DEBUG to keep output useful.
+# ---------------------------------------------------------------------------
+_explicit_log_level = os.environ.get("LOG_LEVEL")
+if settings.debug and _explicit_log_level is None:
+    _effective_level = "DEBUG"
+else:
+    _effective_level = settings.log_level.upper()
+
+_effective_level_int = getattr(logging, _effective_level, logging.INFO)
+
+
+class _JsonFormatter(logging.Formatter):
+    """Emit one JSON object per log line for machine consumption.
+
+    Fields emitted: ``timestamp``, ``level``, ``logger``, ``message``,
+    ``module``, ``funcName``, ``lineno``, and — when present — ``exc_info``.
+    Compatible with Grafana Loki, Splunk, ELK, Datadog, and most SIEM tools.
+    """
+
+    def format(self, record: logging.LogRecord) -> str:
+        log_entry: dict = {
+            "timestamp": _dt.fromtimestamp(record.created, tz=_tz.utc).isoformat(),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+            "module": record.module,
+            "funcName": record.funcName,
+            "lineno": record.lineno,
+        }
+        if record.exc_info and record.exc_info[1] is not None:
+            log_entry["exc_info"] = self.formatException(record.exc_info)
+        return _json_mod.dumps(log_entry, default=str)
+
+
+# Choose formatter based on LOG_FORMAT setting
+if settings.log_format.lower() == "json":
+    _handler = logging.StreamHandler()
+    _handler.setFormatter(_JsonFormatter())
+    logging.root.handlers = [_handler]
+    logging.root.setLevel(_effective_level_int)
+else:
+    logging.basicConfig(
+        level=_effective_level_int,
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+        force=True,
+    )
+
+# Optional: forward application logs to a syslog receiver
+if settings.log_syslog_enabled:
+    import logging.handlers as _lh
+    import socket as _socket
+
+    _proto = settings.log_syslog_protocol.lower()
+    _socktype = _socket.SOCK_STREAM if _proto == "tcp" else _socket.SOCK_DGRAM
+    _syslog_handler = _lh.SysLogHandler(
+        address=(settings.log_syslog_host, settings.log_syslog_port),
+        socktype=_socktype,
+    )
+    _syslog_handler.setLevel(_effective_level_int)
+    # Use the same formatter as stdout (text or JSON)
+    if settings.log_format.lower() == "json":
+        _syslog_handler.setFormatter(_JsonFormatter())
+    else:
+        _syslog_handler.setFormatter(logging.Formatter("%(name)s - %(levelname)s - %(message)s"))
+    logging.root.addHandler(_syslog_handler)
+
+# Keep noisy third-party loggers quiet at DEBUG level
+if _effective_level_int <= logging.DEBUG:
+    for _noisy in (
+        "httpx",
+        "httpcore",
+        "authlib",
+        "urllib3",
+        "hpack",
+        "multipart",
+        "watchfiles",
+    ):
+        logging.getLogger(_noisy).setLevel(logging.WARNING)
+
+_startup_logger = logging.getLogger(__name__)
+_startup_logger.info(
+    "Root logging level set to %s (debug=%s, format=%s, syslog=%s)",
+    _effective_level,
+    settings.debug,
+    settings.log_format,
+    settings.log_syslog_enabled,
+)
+
 # Load configuration from .env for the session key
 config = Config(".env")
 # Use settings.session_secret which has proper validation

docs/ConfigurationGuide.md

@@ -457,6 +457,87 @@ default overage buffer applied across all plans.
 
 DocuElevate supports HTTP security headers to improve browser-side security. **These headers are disabled by default** since most deployments use a reverse proxy (Traefik, Nginx, etc.) that already adds them. Enable only if deploying directly without a reverse proxy. See [Deployment Guide - Security Headers](DeploymentGuide.md#security-headers) for detailed configuration examples.
 
+### Application Logging
+
+DocuElevate uses Python's standard `logging` module. Two environment variables control log verbosity:
+
+| **Variable** | **Description** | **Default** |
+|-------------|----------------|-------------|
+| `LOG_LEVEL` | Root logger level. Accepts standard Python level names: `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL`. | `INFO` |
+| `DEBUG` | Enable debug mode. When `true` **and** `LOG_LEVEL` is **not** explicitly set, the effective log level is automatically lowered to `DEBUG`. | `false` |
+
+**Precedence rules (standard behaviour):**
+
+1. If `LOG_LEVEL` is explicitly set, it always wins — regardless of `DEBUG`.
+2. If only `DEBUG=true` is set (no `LOG_LEVEL`), the effective level becomes `DEBUG`.
+3. If neither is set, the default level is `INFO`.
+
+```bash
+# Typical production (default)
+# LOG_LEVEL=INFO
+
+# Quick debug mode — sets level to DEBUG automatically
+DEBUG=true
+
+# Explicit level override (DEBUG flag is ignored for level selection)
+LOG_LEVEL=WARNING
+```
+
+> **Tip:** At `DEBUG` level, noisy third-party libraries (httpx, authlib, urllib3, etc.) are automatically pinned to `WARNING` so that application debug output remains readable.
+
+#### Structured JSON Logging
+
+Set `LOG_FORMAT=json` to emit structured JSON lines on stdout — one JSON object per log message. This is the standard format for log collectors and SIEM tools:
+
+| **Variable** | **Description** | **Default** |
+|-------------|----------------|-------------|
+| `LOG_FORMAT` | Log output format: `text` (human-readable) or `json` (structured JSON lines). | `text` |
+
+Each JSON log line contains: `timestamp` (ISO 8601), `level`, `logger`, `message`, `module`, `funcName`, `lineno`, and `exc_info` (when an exception is logged).
+
+```bash
+# Enable JSON logging for SIEM / log aggregation
+LOG_FORMAT=json
+```
+
+**Example JSON output:**
+```json
+{"timestamp": "2025-03-16T09:18:05.192000+00:00", "level": "INFO", "logger": "app.auth", "message": "[SECURITY] OAUTH_LOGIN_SUCCESS user=alice@example.com admin=False", "module": "auth", "funcName": "oauth_callback", "lineno": 654}
+```
+
+**Compatible with:**
+- **Grafana Loki** — Promtail scrapes JSON from Docker stdout
+- **Splunk** — Universal Forwarder or HEC with JSON sourcetype
+- **ELK / OpenSearch** — Filebeat with JSON codec
+- **Datadog** — Agent auto-parses JSON logs
+- **Fluentd / Vector** — JSON input plugin
+- **Docker log drivers** — `--log-driver=json-file` (default) preserves structure
+
+#### Syslog Forwarding (Application Logs)
+
+For traditional (non-container) deployments, application logs can be forwarded directly to a syslog receiver. This is **separate** from audit-log SIEM forwarding (see below) — it sends _every_ Python log message, not just audit events.
+
+| **Variable** | **Description** | **Default** |
+|-------------|----------------|-------------|
+| `LOG_SYSLOG_ENABLED` | Forward application logs to a syslog receiver in addition to stdout. | `false` |
+| `LOG_SYSLOG_HOST` | Hostname or IP of the syslog receiver. | `localhost` |
+| `LOG_SYSLOG_PORT` | Port of the syslog receiver. | `514` |
+| `LOG_SYSLOG_PROTOCOL` | Protocol: `udp` or `tcp`. | `udp` |
+
+```bash
+# Forward all application logs to syslog
+LOG_SYSLOG_ENABLED=true
+LOG_SYSLOG_HOST=syslog.internal.example.com
+LOG_SYSLOG_PORT=514
+LOG_SYSLOG_PROTOCOL=udp
+
+# Combine with JSON format for structured syslog messages
+LOG_FORMAT=json
+LOG_SYSLOG_ENABLED=true
+```
+
+> **Note:** When `LOG_FORMAT=json`, syslog messages are also sent as JSON. When `LOG_FORMAT=text`, syslog messages use the standard `name - level - message` format.
+
 ### Audit Logging
 
 DocuElevate provides comprehensive audit logging that records significant actions (logins, document CRUD, settings changes) to an append-only database table. Every entry captures the timestamp, user, action, resource, client IP, and optional JSON details.

tests/test_auth.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import hashlib
+import logging
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -38,6 +39,40 @@ class TestGetCurrentUser:
         result = get_current_user(mock_request)
         assert result is None
 
+    def test_logs_debug_when_session_user_found(self, caplog):
+        """Test that get_current_user emits a DEBUG log when session user is found."""
+        mock_request = MagicMock(spec=Request)
+        mock_request.session = {"user": {"id": "u1", "preferred_username": "alice"}}
+        mock_request.state = MagicMock(spec=[])  # no api_token_user attribute
+
+        with caplog.at_level(logging.DEBUG, logger="app.auth"):
+            get_current_user(mock_request)
+
+        assert any("[AUTH] get_current_user: resolved from session" in m for m in caplog.messages)
+
+    def test_logs_debug_when_no_user(self, caplog):
+        """Test that get_current_user emits a DEBUG log when no user is present."""
+        mock_request = MagicMock(spec=Request)
+        mock_request.session = {}
+        mock_request.state = MagicMock(spec=[])
+
+        with caplog.at_level(logging.DEBUG, logger="app.auth"):
+            get_current_user(mock_request)
+
+        assert any("[AUTH] get_current_user: no user in session or API token" in m for m in caplog.messages)
+
+    def test_logs_debug_when_api_token_user(self, caplog):
+        """Test that get_current_user emits a DEBUG log when resolved from API token."""
+        mock_request = MagicMock(spec=Request)
+        mock_request.state.api_token_user = {"id": "tok_user"}
+        mock_request.session = {}
+
+        with caplog.at_level(logging.DEBUG, logger="app.auth"):
+            result = get_current_user(mock_request)
+
+        assert result == {"id": "tok_user"}
+        assert any("[AUTH] get_current_user: resolved from API token" in m for m in caplog.messages)
+
 
 @pytest.mark.unit
 class TestGetGravatarUrl:

tests/test_logging_config.py

@@ -0,0 +1,285 @@
+"""Tests for application logging configuration.
+
+Validates that the LOG_LEVEL and DEBUG settings correctly control the
+Python root-logger level and that the standard precedence rules are respected:
+    1. Explicit LOG_LEVEL always wins.
+    2. DEBUG=True without LOG_LEVEL → effective DEBUG.
+    3. Neither set → default INFO.
+"""
+
+import logging
+import os
+from unittest.mock import patch
+
+import pytest
+
+from app.config import Settings
+
+
+@pytest.mark.unit
+class TestLogLevelSetting:
+    """Tests for the log_level config field."""
+
+    _BASE_KWARGS = {
+        "database_url": "sqlite:///test.db",
+        "redis_url": "redis://localhost:6379",
+        "openai_api_key": "test",
+        "azure_ai_key": "test",
+        "azure_region": "test",
+        "azure_endpoint": "https://test.example.com",
+        "gotenberg_url": "http://localhost:3000",
+        "workdir": "/tmp",
+        "auth_enabled": False,
+        "session_secret": None,
+    }
+
+    def test_log_level_default_is_info(self):
+        """Test that log_level defaults to INFO."""
+        config = Settings(**self._BASE_KWARGS)
+        assert config.log_level.upper() == "INFO"
+
+    def test_log_level_accepts_debug(self):
+        """Test that log_level accepts DEBUG."""
+        config = Settings(**self._BASE_KWARGS, log_level="DEBUG")
+        assert config.log_level.upper() == "DEBUG"
+
+    def test_log_level_accepts_warning(self):
+        """Test that log_level accepts WARNING."""
+        config = Settings(**self._BASE_KWARGS, log_level="WARNING")
+        assert config.log_level.upper() == "WARNING"
+
+    def test_log_level_accepts_error(self):
+        """Test that log_level accepts ERROR."""
+        config = Settings(**self._BASE_KWARGS, log_level="ERROR")
+        assert config.log_level.upper() == "ERROR"
+
+    def test_log_level_case_insensitive(self):
+        """Test that log_level is case-insensitive in usage."""
+        config = Settings(**self._BASE_KWARGS, log_level="debug")
+        assert config.log_level.upper() == "DEBUG"
+
+    def test_debug_flag_defaults_to_false(self):
+        """Test that debug defaults to False."""
+        config = Settings(**self._BASE_KWARGS)
+        assert config.debug is False
+
+
+@pytest.mark.unit
+class TestEffectiveLogLevel:
+    """Tests for the effective log-level resolution logic in main.py."""
+
+    def test_debug_true_without_log_level_gives_debug(self):
+        """When DEBUG=True and LOG_LEVEL is not set, effective level is DEBUG."""
+        with patch.dict(os.environ, {"DEBUG": "true"}, clear=False):
+            # Remove LOG_LEVEL from env if present
+            env = os.environ.copy()
+            env.pop("LOG_LEVEL", None)
+            with patch.dict(os.environ, env, clear=True):
+                s = Settings(
+                    database_url="sqlite:///test.db",
+                    redis_url="redis://localhost:6379",
+                    openai_api_key="test",
+                    azure_ai_key="test",
+                    azure_region="test",
+                    azure_endpoint="https://test.example.com",
+                    gotenberg_url="http://localhost:3000",
+                    workdir="/tmp",
+                    auth_enabled=False,
+                    session_secret=None,
+                    debug=True,
+                )
+                explicit = os.environ.get("LOG_LEVEL")
+                if s.debug and explicit is None:
+                    effective = "DEBUG"
+                else:
+                    effective = s.log_level.upper()
+                assert effective == "DEBUG"
+
+    def test_explicit_log_level_overrides_debug(self):
+        """When LOG_LEVEL is explicitly set, it takes precedence over DEBUG=True."""
+        with patch.dict(os.environ, {"LOG_LEVEL": "WARNING", "DEBUG": "true"}, clear=False):
+            s = Settings(
+                database_url="sqlite:///test.db",
+                redis_url="redis://localhost:6379",
+                openai_api_key="test",
+                azure_ai_key="test",
+                azure_region="test",
+                azure_endpoint="https://test.example.com",
+                gotenberg_url="http://localhost:3000",
+                workdir="/tmp",
+                auth_enabled=False,
+                session_secret=None,
+                debug=True,
+                log_level="WARNING",
+            )
+            explicit = os.environ.get("LOG_LEVEL")
+            if s.debug and explicit is None:
+                effective = "DEBUG"
+            else:
+                effective = s.log_level.upper()
+            assert effective == "WARNING"
+
+    def test_default_no_flags_gives_info(self):
+        """When neither DEBUG nor LOG_LEVEL is set, effective level is INFO."""
+        env = os.environ.copy()
+        env.pop("LOG_LEVEL", None)
+        env.pop("DEBUG", None)
+        with patch.dict(os.environ, env, clear=True):
+            s = Settings(
+                database_url="sqlite:///test.db",
+                redis_url="redis://localhost:6379",
+                openai_api_key="test",
+                azure_ai_key="test",
+                azure_region="test",
+                azure_endpoint="https://test.example.com",
+                gotenberg_url="http://localhost:3000",
+                workdir="/tmp",
+                auth_enabled=False,
+                session_secret=None,
+            )
+            explicit = os.environ.get("LOG_LEVEL")
+            if s.debug and explicit is None:
+                effective = "DEBUG"
+            else:
+                effective = s.log_level.upper()
+            assert effective == "INFO"
+
+    def test_effective_level_maps_to_logging_constant(self):
+        """The effective level string maps to a valid logging constant."""
+        for level_name in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):
+            assert getattr(logging, level_name) is not None
+
+
+@pytest.mark.unit
+class TestLoggingConfiguredAtStartup:
+    """Tests that the main module configures the root logger on import."""
+
+    def test_root_logger_has_handler(self):
+        """Root logger should have at least one handler after app import."""
+        root = logging.getLogger()
+        assert len(root.handlers) > 0, "Root logger has no handlers after app startup"
+
+    def test_root_logger_level_is_not_warning_default(self):
+        """Root logger should not be at the unconfigured WARNING default.
+
+        Our basicConfig(force=True) should have set it to at least INFO.
+        """
+        root = logging.getLogger()
+        # The test env doesn't set DEBUG=True, so the level should be INFO (20)
+        assert root.level <= logging.INFO
+
+
+@pytest.mark.unit
+class TestJsonFormatter:
+    """Tests for the _JsonFormatter used when LOG_FORMAT=json."""
+
+    def _make_formatter(self):
+        """Lazily import the JSON formatter from main module."""
+        from app.main import _JsonFormatter
+
+        return _JsonFormatter()
+
+    def test_output_is_valid_json(self):
+        """JSON formatter output should be parseable JSON."""
+        import json
+
+        fmt = self._make_formatter()
+        record = logging.LogRecord(
+            name="test.logger",
+            level=logging.INFO,
+            pathname="test.py",
+            lineno=42,
+            msg="Hello %s",
+            args=("world",),
+            exc_info=None,
+        )
+        result = fmt.format(record)
+        parsed = json.loads(result)
+        assert parsed["level"] == "INFO"
+        assert parsed["logger"] == "test.logger"
+        assert parsed["message"] == "Hello world"
+        assert parsed["lineno"] == 42
+
+    def test_includes_timestamp_iso8601(self):
+        """JSON output should contain an ISO 8601 timestamp."""
+        import json
+
+        fmt = self._make_formatter()
+        record = logging.LogRecord(
+            name="x",
+            level=logging.DEBUG,
+            pathname="x.py",
+            lineno=1,
+            msg="test",
+            args=(),
+            exc_info=None,
+        )
+        parsed = json.loads(fmt.format(record))
+        assert "timestamp" in parsed
+        # ISO 8601 timestamps contain "T" and "+00:00" (UTC)
+        assert "T" in parsed["timestamp"]
+
+    def test_includes_exc_info_when_present(self):
+        """JSON output should include exc_info when an exception is logged."""
+        import json
+
+        fmt = self._make_formatter()
+        try:
+            raise ValueError("boom")  # noqa: TRY301
+        except ValueError:
+            import sys
+
+            record = logging.LogRecord(
+                name="x",
+                level=logging.ERROR,
+                pathname="x.py",
+                lineno=1,
+                msg="error",
+                args=(),
+                exc_info=sys.exc_info(),
+            )
+        parsed = json.loads(fmt.format(record))
+        assert "exc_info" in parsed
+        assert "ValueError" in parsed["exc_info"]
+
+
+@pytest.mark.unit
+class TestLogFormatSetting:
+    """Tests for the log_format and log_syslog_* config fields."""
+
+    _BASE_KWARGS = {
+        "database_url": "sqlite:///test.db",
+        "redis_url": "redis://localhost:6379",
+        "openai_api_key": "test",
+        "azure_ai_key": "test",
+        "azure_region": "test",
+        "azure_endpoint": "https://test.example.com",
+        "gotenberg_url": "http://localhost:3000",
+        "workdir": "/tmp",
+        "auth_enabled": False,
+        "session_secret": None,
+    }
+
+    def test_log_format_default_is_text(self):
+        """Test that log_format defaults to 'text'."""
+        config = Settings(**self._BASE_KWARGS)
+        assert config.log_format == "text"
+
+    def test_log_format_accepts_json(self):
+        """Test that log_format accepts 'json'."""
+        config = Settings(**self._BASE_KWARGS, log_format="json")
+        assert config.log_format == "json"
+
+    def test_log_syslog_defaults(self):
+        """Test syslog forwarding defaults."""
+        config = Settings(**self._BASE_KWARGS)
+        assert config.log_syslog_enabled is False
+        assert config.log_syslog_host == "localhost"
+        assert config.log_syslog_port == 514
+        assert config.log_syslog_protocol == "udp"
+
+    def test_log_syslog_can_be_enabled(self):
+        """Test that syslog forwarding can be enabled."""
+        config = Settings(**self._BASE_KWARGS, log_syslog_enabled=True, log_syslog_host="syslog.example.com")
+        assert config.log_syslog_enabled is True
+        assert config.log_syslog_host == "syslog.example.com"