diff --git a/.env.demo b/.env.demo index 20c6b926..9412e754 100644 --- a/.env.demo +++ b/.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. diff --git a/app/auth.py b/app/auth.py index 83701b1b..a1a40702 100644 --- a/app/auth.py +++ b/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 diff --git a/app/config.py b/app/config.py index d3a2080c..30b434d1 100644 --- a/app/config.py +++ b/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, diff --git a/app/main.py b/app/main.py index 82c4fd59..97bb1d57 100644 --- a/app/main.py +++ b/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 diff --git a/docs/ConfigurationGuide.md b/docs/ConfigurationGuide.md index 643ac930..37acf9c2 100644 --- a/docs/ConfigurationGuide.md +++ b/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. diff --git a/tests/test_auth.py b/tests/test_auth.py index 052c1f6e..85e047e8 100644 --- a/tests/test_auth.py +++ b/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: diff --git a/tests/test_logging_config.py b/tests/test_logging_config.py new file mode 100644 index 00000000..9c24c13c --- /dev/null +++ b/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"