diff --git a/SECURITY_AUDIT.md b/SECURITY_AUDIT.md index 6d1956cb..508e9453 100644 --- a/SECURITY_AUDIT.md +++ b/SECURITY_AUDIT.md @@ -289,7 +289,12 @@ ftp = ftplib.FTP() # nosec B321 - Plaintext FTP intentional when configured - Individual header control and customization - Documented in DeploymentGuide.md and ConfigurationGuide.md - ⏳ **TODO:** Implement proper CORS configuration (currently not configured) ([#175](https://github.com/christianlouis/DocuElevate/issues/175)) -- ⏳ **TODO:** Add request logging with sensitive data masking ([#170](https://github.com/christianlouis/DocuElevate/issues/170)) +- ✅ **Request logging with sensitive data masking implemented** ([#170](https://github.com/christianlouis/DocuElevate/issues/170)) + - `AuditLogMiddleware` in `app/middleware/audit_log.py` logs every HTTP request + - Logs: method, path, status code, response time, client IP (configurable), username + - Sensitive query-parameter values (password, token, key, secret, etc.) are automatically replaced with ``[REDACTED]`` + - Security events (401, 403, login attempts, 5xx errors) receive elevated ``[SECURITY]`` log entries + - Configurable via `AUDIT_LOGGING_ENABLED` and `AUDIT_LOG_INCLUDE_CLIENT_IP` environment variables ## Recommendations @@ -303,7 +308,7 @@ ftp = ftplib.FTP() # nosec B321 - Plaintext FTP intentional when configured ### Medium Priority 1. ~~**Add security headers**~~ ✅ Implemented - Configurable HSTS, CSP, X-Frame-Options, X-Content-Type-Options middleware 2. **Configure CORS properly** - Currently no CORS middleware configured ([#175](https://github.com/christianlouis/DocuElevate/issues/175)) -3. **Implement audit logging** - Track security-relevant events ([#170](https://github.com/christianlouis/DocuElevate/issues/170)) +3. ~~**Implement audit logging**~~ ✅ Implemented - Request/audit logging with sensitive data masking ([#170](https://github.com/christianlouis/DocuElevate/issues/170)) 4. ~~**Add file upload size limits**~~ ✅ Implemented - Configurable limits with 1GB default, optional file splitting 5. **Document security architecture** - Security design decisions diff --git a/app/auth.py b/app/auth.py index 0706df1e..2d5c8c53 100644 --- a/app/auth.py +++ b/app/auth.py @@ -1,5 +1,6 @@ import hashlib import inspect +import logging import pathlib from functools import wraps @@ -12,6 +13,8 @@ from app.config import settings oauth = OAuth() +logger = logging.getLogger(__name__) + AUTH_ENABLED = settings.auth_enabled # Set up templates for authentication @@ -123,13 +126,13 @@ async def oauth_callback(request: Request): request.session["user"] = user_data # Log the successful authentication - print(f"User authenticated via OAuth: {user_data.get('email', 'No email')} (admin: {is_admin})") + logger.info(f"[SECURITY] OAUTH_LOGIN_SUCCESS user={user_data.get('email', 'unknown')} admin={is_admin}") # Redirect to original destination or default redirect_url = request.session.pop("redirect_after_login", "/upload") return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND) except Exception as e: - print(f"OAuth authentication error: {str(e)}") + logger.warning(f"[SECURITY] OAUTH_LOGIN_FAILURE error={type(e).__name__}") return RedirectResponse(url=f"/login?error=Authentication+failed:+{str(e)}", status_code=status.HTTP_302_FOUND) @@ -149,15 +152,22 @@ async def auth(request: Request): "picture": "/static/images/default-avatar.svg", "is_admin": True, } + logger.info(f"[SECURITY] LOCAL_LOGIN_SUCCESS user={username}") # Redirect to original destination or default redirect_url = request.session.pop("redirect_after_login", "/upload") return RedirectResponse(url=redirect_url, status_code=302) else: + logger.warning(f"[SECURITY] LOCAL_LOGIN_FAILURE user={username}") return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302) async def logout(request: Request): """Handle user logout""" + user = request.session.get("user") + username = "unknown" + if isinstance(user, dict): + username = user.get("preferred_username") or user.get("email") or "unknown" + logger.info(f"[SECURITY] LOGOUT user={username}") request.session.pop("user", None) return RedirectResponse(url="/login?message=You+have+been+logged+out+successfully", status_code=302) diff --git a/app/config.py b/app/config.py index 2c0ff620..67fbb6f4 100644 --- a/app/config.py +++ b/app/config.py @@ -248,6 +248,25 @@ class Settings(BaseSettings): default=True, description="Enable X-Content-Type-Options header (always set to 'nosniff')." ) + # Audit Logging Configuration (see SECURITY_AUDIT.md – Infrastructure Security) + # Logs every HTTP request and security-relevant events (auth failures, 5xx errors). + # Sensitive query-parameter values (passwords, tokens, keys) are always masked. + audit_logging_enabled: bool = Field( + default=True, + description=( + "Enable audit/request logging middleware. Logs every HTTP request with " + "method, path, status code, response time, and username. " + "Sensitive query-parameter values are automatically masked." + ), + ) + audit_log_include_client_ip: bool = Field( + default=True, + description=( + "Include the client IP address in audit log entries. " + "Disable for privacy-sensitive deployments where IP logging is restricted." + ), + ) + # Rate Limiting Configuration (see SECURITY_AUDIT.md and docs/API.md) # Protects against DoS attacks and API abuse rate_limiting_enabled: bool = Field( diff --git a/app/main.py b/app/main.py index f2e1af2d..438ae3fa 100644 --- a/app/main.py +++ b/app/main.py @@ -18,6 +18,7 @@ from app.api import router as api_router from app.auth import router as auth_router from app.config import settings from app.database import init_db +from app.middleware.audit_log import AuditLogMiddleware from app.middleware.rate_limit import create_limiter, get_rate_limit_exceeded_handler from app.middleware.security_headers import SecurityHeadersMiddleware from app.utils.config_validator import check_all_configs @@ -115,13 +116,18 @@ app.add_exception_handler(RateLimitExceeded, get_rate_limit_exceeded_handler()) # Set to False if reverse proxy (Traefik, Nginx) handles security headers app.add_middleware(SecurityHeadersMiddleware, config=settings) -# 2) Session Middleware (for request.session to work) +# 2) Audit Logging Middleware - logs all requests with sensitive data masking +# Configure via AUDIT_LOGGING_ENABLED environment variable +# See SECURITY_AUDIT.md – Infrastructure Security section +app.add_middleware(AuditLogMiddleware, config=settings) + +# 3) Session Middleware (for request.session to work) app.add_middleware(SessionMiddleware, secret_key=SESSION_SECRET) -# 3) Respect the X-Forwarded-* headers from reverse proxy (Traefik, Nginx) +# 4) Respect the X-Forwarded-* headers from reverse proxy (Traefik, Nginx) app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*") -# 4) Restrict valid hosts to prevent Host header attacks +# 5) Restrict valid hosts to prevent Host header attacks app.add_middleware(TrustedHostMiddleware, allowed_hosts=[settings.external_hostname, "localhost", "127.0.0.1"]) # Mount the static files directory diff --git a/app/middleware/audit_log.py b/app/middleware/audit_log.py new file mode 100644 index 00000000..de8a092d --- /dev/null +++ b/app/middleware/audit_log.py @@ -0,0 +1,250 @@ +#!/usr/bin/env python3 + +""" +Audit Logging Middleware for DocuElevate. + +This middleware logs all HTTP requests and security-relevant events. Sensitive +data (passwords, tokens, secrets, API keys) is masked before logging so that +credentials are never recorded in application logs. + +Security-relevant events that receive elevated ``[SECURITY]`` log entries: +- Authentication failures (401 Unauthorized) +- Authorisation denials (403 Forbidden) +- Login / logout endpoint access +- Server errors (5xx responses) + +Logged per request: +- HTTP method +- Request path (query-param values for known sensitive keys are replaced with ``[REDACTED]``) +- Response status code +- Response time in milliseconds +- Client IP address (configurable) +- Authenticated username when available + +See SECURITY_AUDIT.md – Infrastructure Security section for background. +""" + +import logging +import re +import time +from typing import Callable + +from fastapi import Request, Response +from starlette.middleware.base import BaseHTTPMiddleware + +logger = logging.getLogger(__name__) + +# Query-parameter / form-field names whose *values* must never appear in logs. +# Matching is case-insensitive. +_SENSITIVE_PARAM_PATTERN = re.compile( + r"^(password|passwd|pwd|secret|token|access_token|refresh_token|" + r"api_key|apikey|key|credential|credentials|auth|authorization|" + r"client_secret|private_key|session)$", + re.IGNORECASE, +) + +# HTTP headers whose values must never appear in logs. +_SENSITIVE_HEADERS = frozenset( + { + "authorization", + "cookie", + "set-cookie", + "x-api-key", + "x-auth-token", + } +) + +# Endpoints considered security-sensitive for elevated logging. +_AUTH_PATHS = frozenset({"/auth", "/login", "/logout", "/oauth-login", "/oauth-callback"}) + + +def mask_query_string(query_string: str) -> str: + """ + Replace values of sensitive query parameters with ``[REDACTED]``. + + Args: + query_string: Raw URL query string (e.g. ``"user=alice&password=secret"``). + + Returns: + Query string with sensitive values replaced. + """ + if not query_string: + return query_string + + parts = [] + for pair in query_string.split("&"): + if "=" in pair: + name, _, value = pair.partition("=") + if _SENSITIVE_PARAM_PATTERN.match(name): + parts.append(f"{name}=[REDACTED]") + else: + parts.append(pair) + else: + parts.append(pair) + return "&".join(parts) + + +def get_client_ip(request: Request) -> str: + """ + Extract the real client IP, honouring X-Forwarded-For when present. + + Args: + request: Incoming HTTP request. + + Returns: + Client IP address string. + """ + forwarded_for = request.headers.get("x-forwarded-for") + if forwarded_for: + # Take only the first (leftmost) address – that is the original client. + return forwarded_for.split(",")[0].strip() + if request.client: + return request.client.host + return "unknown" + + +def get_username(request: Request) -> str: + """ + Extract the authenticated username from the session, if available. + + Args: + request: Incoming HTTP request. + + Returns: + Username string, or ``"anonymous"`` when not authenticated. + """ + try: + user = request.session.get("user") if hasattr(request, "session") else None + except Exception: + user = None + + if not user: + return "anonymous" + + if isinstance(user, dict): + return ( + user.get("preferred_username") + or user.get("username") + or user.get("email") + or user.get("id") + or "authenticated" + ) + return str(user) + + +class AuditLogMiddleware(BaseHTTPMiddleware): + """ + Middleware to log HTTP requests and security-relevant events. + + Each request produces a single ``INFO``-level audit log line. + Requests that result in 401/403 responses, or that target + authentication endpoints, additionally produce a ``WARNING``-level + security event line. Server errors (5xx) produce an ``ERROR``-level + security event line. + + Configuration is read from the application settings object passed at + construction time via the ``config`` keyword argument. + """ + + def __init__(self, app, config) -> None: + """ + Initialise the audit-log middleware. + + Args: + app: FastAPI / ASGI application instance. + config: Application settings object (must expose + ``audit_logging_enabled`` and + ``audit_log_include_client_ip`` boolean attributes). + """ + super().__init__(app) + self.enabled = config.audit_logging_enabled + self.include_ip = config.audit_log_include_client_ip + + if self.enabled: + logger.info(f"Audit logging middleware enabled (include_client_ip={self.include_ip})") + else: + logger.info("Audit logging middleware disabled") + + async def dispatch(self, request: Request, call_next: Callable) -> Response: + """ + Process the request, call the next handler, then emit audit log entries. + + Args: + request: Incoming HTTP request. + call_next: Next middleware or route handler in the chain. + + Returns: + HTTP response (unmodified). + """ + if not self.enabled: + return await call_next(request) + + start_time = time.monotonic() + response = await call_next(request) + duration_ms = int((time.monotonic() - start_time) * 1000) + + self._log_request(request, response.status_code, duration_ms) + return response + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _build_path_with_masked_query(self, request: Request) -> str: + """Return the request path with sensitive query-param values masked.""" + path = request.url.path + raw_query = request.url.query + if raw_query: + masked = mask_query_string(raw_query) + return f"{path}?{masked}" + return path + + def _log_request(self, request: Request, status_code: int, duration_ms: int) -> None: + """ + Emit audit log entries for a completed request. + + Args: + request: The HTTP request object. + status_code: HTTP response status code. + duration_ms: Total request processing time in milliseconds. + """ + method = request.method + path = self._build_path_with_masked_query(request) + username = get_username(request) + ip_part = f" - {get_client_ip(request)}" if self.include_ip else "" + + # Core request log line (always INFO). + logger.info(f"[AUDIT] {method} {path} {status_code} {duration_ms}ms{ip_part} - {username}") + + # Security-event log lines for noteworthy conditions. + self._log_security_event(method, path, status_code, username, ip_part) + + def _log_security_event( + self, + method: str, + path: str, + status_code: int, + username: str, + ip_part: str, + ) -> None: + """ + Emit an additional security-event log line when warranted. + + Args: + method: HTTP method (GET, POST, …). + path: Sanitised request path (with masked query params). + status_code: HTTP response status code. + username: Authenticated username or ``"anonymous"``. + ip_part: Pre-formatted IP string (may be empty string). + """ + base_path = path.split("?", maxsplit=1)[0] + + if status_code == 401: + logger.warning(f"[SECURITY] AUTH_FAILURE {method} {path} 401{ip_part} - {username}") + elif status_code == 403: + logger.warning(f"[SECURITY] ACCESS_DENIED {method} {path} 403{ip_part} - {username}") + elif base_path in _AUTH_PATHS and method == "POST": + # Login attempts (successful or not) are always noted. + logger.info(f"[SECURITY] AUTH_ATTEMPT {method} {path} {status_code}{ip_part} - {username}") + elif status_code >= 500: + logger.error(f"[SECURITY] SERVER_ERROR {method} {path} {status_code}{ip_part} - {username}") diff --git a/tests/test_audit_logging.py b/tests/test_audit_logging.py new file mode 100644 index 00000000..70fc5542 --- /dev/null +++ b/tests/test_audit_logging.py @@ -0,0 +1,400 @@ +#!/usr/bin/env python3 + +""" +Tests for AuditLogMiddleware and associated helper functions. + +Validates: +- Sensitive query-parameter masking +- Client-IP extraction +- Username extraction from session +- Middleware initialisation (enabled / disabled) +- Audit log entries for normal requests +- Elevated security-event log entries for 401 / 403 / 5xx responses + and for authentication-endpoint POST requests +""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import Response + +from app.middleware.audit_log import ( + AuditLogMiddleware, + get_client_ip, + get_username, + mask_query_string, +) + +# --------------------------------------------------------------------------- +# mask_query_string +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestMaskQueryString: + """Tests for the mask_query_string helper.""" + + def test_empty_string_returns_empty(self): + assert mask_query_string("") == "" + + def test_non_sensitive_param_unchanged(self): + assert mask_query_string("page=2&limit=10") == "page=2&limit=10" + + def test_password_param_masked(self): + result = mask_query_string("user=alice&password=secret123") + assert "secret123" not in result + assert "password=[REDACTED]" in result + assert "user=alice" in result + + def test_token_param_masked(self): + result = mask_query_string("access_token=abc123&foo=bar") + assert "abc123" not in result + assert "access_token=[REDACTED]" in result + + def test_multiple_sensitive_params_all_masked(self): + result = mask_query_string("key=mykey&secret=mysecret&name=test") + assert "mykey" not in result + assert "mysecret" not in result + assert "key=[REDACTED]" in result + assert "secret=[REDACTED]" in result + assert "name=test" in result + + def test_case_insensitive_masking(self): + result = mask_query_string("PASSWORD=topsecret") + assert "topsecret" not in result + assert "PASSWORD=[REDACTED]" in result + + def test_param_without_value(self): + """A bare param name (no '=') should be left as-is.""" + result = mask_query_string("flag") + assert result == "flag" + + def test_api_key_masked(self): + result = mask_query_string("api_key=supersecret&query=docs") + assert "supersecret" not in result + assert "api_key=[REDACTED]" in result + + def test_refresh_token_masked(self): + result = mask_query_string("refresh_token=r3fr3sh") + assert "r3fr3sh" not in result + assert "refresh_token=[REDACTED]" in result + + def test_non_sensitive_value_not_masked(self): + result = mask_query_string("username=alice&page=1") + assert result == "username=alice&page=1" + + +# --------------------------------------------------------------------------- +# get_client_ip +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestGetClientIp: + """Tests for the get_client_ip helper.""" + + def _make_request(self, headers=None, client_host=None): + req = MagicMock() + req.headers = headers or {} + if client_host: + req.client = MagicMock() + req.client.host = client_host + else: + req.client = None + return req + + def test_returns_forwarded_for_first_ip(self): + req = self._make_request( + headers={"x-forwarded-for": "203.0.113.1, 10.0.0.1"}, + client_host="10.0.0.1", + ) + assert get_client_ip(req) == "203.0.113.1" + + def test_falls_back_to_client_host(self): + req = self._make_request(client_host="192.168.1.42") + assert get_client_ip(req) == "192.168.1.42" + + def test_returns_unknown_when_no_client(self): + req = self._make_request() + assert get_client_ip(req) == "unknown" + + def test_single_forwarded_for_value(self): + req = self._make_request(headers={"x-forwarded-for": "1.2.3.4"}) + assert get_client_ip(req) == "1.2.3.4" + + +# --------------------------------------------------------------------------- +# get_username +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestGetUsername: + """Tests for the get_username helper.""" + + def _make_request(self, session_user=None, has_session=True): + req = MagicMock() + if has_session: + req.session = {"user": session_user} if session_user is not None else {} + else: + del req.session + return req + + def test_returns_anonymous_when_no_session(self): + req = self._make_request(has_session=False) + assert get_username(req) == "anonymous" + + def test_returns_anonymous_when_user_not_in_session(self): + req = self._make_request(session_user=None) + assert get_username(req) == "anonymous" + + def test_returns_preferred_username(self): + req = self._make_request(session_user={"preferred_username": "alice", "email": "alice@example.com"}) + assert get_username(req) == "alice" + + def test_falls_back_to_email(self): + req = self._make_request(session_user={"email": "bob@example.com"}) + assert get_username(req) == "bob@example.com" + + def test_falls_back_to_id(self): + req = self._make_request(session_user={"id": "admin"}) + assert get_username(req) == "admin" + + def test_returns_anonymous_for_empty_dict(self): + req = self._make_request(session_user={}) + assert get_username(req) == "anonymous" + + def test_returns_string_for_non_dict_user(self): + req = self._make_request(session_user="some_user") + assert get_username(req) == "some_user" + + +# --------------------------------------------------------------------------- +# AuditLogMiddleware initialisation +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestAuditLogMiddlewareInit: + """Tests for AuditLogMiddleware.__init__.""" + + def _make_config(self, enabled=True, include_ip=True): + cfg = MagicMock() + cfg.audit_logging_enabled = enabled + cfg.audit_log_include_client_ip = include_ip + return cfg + + def test_middleware_enabled_flag(self): + mw = AuditLogMiddleware(app=None, config=self._make_config(enabled=True)) + assert mw.enabled is True + + def test_middleware_disabled_flag(self): + mw = AuditLogMiddleware(app=None, config=self._make_config(enabled=False)) + assert mw.enabled is False + + def test_include_ip_flag(self): + mw = AuditLogMiddleware(app=None, config=self._make_config(include_ip=True)) + assert mw.include_ip is True + + def test_exclude_ip_flag(self): + mw = AuditLogMiddleware(app=None, config=self._make_config(include_ip=False)) + assert mw.include_ip is False + + +# --------------------------------------------------------------------------- +# AuditLogMiddleware.dispatch +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestAuditLogMiddlewareDispatch: + """Tests for AuditLogMiddleware.dispatch.""" + + def _make_middleware(self, enabled=True, include_ip=True): + cfg = MagicMock() + cfg.audit_logging_enabled = enabled + cfg.audit_log_include_client_ip = include_ip + return AuditLogMiddleware(app=None, config=cfg) + + def _make_request(self, path="/test", query="", method="GET", session_user=None): + req = MagicMock() + req.method = method + req.url.path = path + req.url.query = query + req.headers = {} + req.client = MagicMock() + req.client.host = "127.0.0.1" + req.session = {"user": session_user} if session_user else {} + return req + + @pytest.mark.asyncio + async def test_disabled_middleware_passes_through(self): + mw = self._make_middleware(enabled=False) + mock_response = Response(content="ok", status_code=200) + call_next = AsyncMock(return_value=mock_response) + + req = self._make_request() + result = await mw.dispatch(req, call_next) + + assert result is mock_response + call_next.assert_awaited_once_with(req) + + @pytest.mark.asyncio + async def test_enabled_middleware_logs_request(self): + mw = self._make_middleware(enabled=True) + mock_response = Response(content="ok", status_code=200) + call_next = AsyncMock(return_value=mock_response) + req = self._make_request(path="/api/test", method="GET") + + with patch("app.middleware.audit_log.logger") as mock_logger: + await mw.dispatch(req, call_next) + + # At least one info call should contain [AUDIT] + info_calls = [str(c) for c in mock_logger.info.call_args_list] + assert any("[AUDIT]" in c for c in info_calls) + + @pytest.mark.asyncio + async def test_sensitive_query_param_masked_in_log(self): + mw = self._make_middleware(enabled=True) + mock_response = Response(content="ok", status_code=200) + call_next = AsyncMock(return_value=mock_response) + req = self._make_request(path="/search", query="q=hello&password=supersecret") + + with patch("app.middleware.audit_log.logger") as mock_logger: + await mw.dispatch(req, call_next) + + all_calls = " ".join(str(c) for c in mock_logger.info.call_args_list) + assert "supersecret" not in all_calls + assert "[REDACTED]" in all_calls + + @pytest.mark.asyncio + async def test_401_triggers_security_warning(self): + mw = self._make_middleware(enabled=True) + mock_response = Response(content="unauth", status_code=401) + call_next = AsyncMock(return_value=mock_response) + req = self._make_request(path="/api/protected") + + with patch("app.middleware.audit_log.logger") as mock_logger: + await mw.dispatch(req, call_next) + + warning_calls = [str(c) for c in mock_logger.warning.call_args_list] + assert any("AUTH_FAILURE" in c for c in warning_calls) + + @pytest.mark.asyncio + async def test_403_triggers_security_warning(self): + mw = self._make_middleware(enabled=True) + mock_response = Response(content="forbidden", status_code=403) + call_next = AsyncMock(return_value=mock_response) + req = self._make_request(path="/admin") + + with patch("app.middleware.audit_log.logger") as mock_logger: + await mw.dispatch(req, call_next) + + warning_calls = [str(c) for c in mock_logger.warning.call_args_list] + assert any("ACCESS_DENIED" in c for c in warning_calls) + + @pytest.mark.asyncio + async def test_5xx_triggers_security_error(self): + mw = self._make_middleware(enabled=True) + mock_response = Response(content="error", status_code=500) + call_next = AsyncMock(return_value=mock_response) + req = self._make_request(path="/api/crash") + + with patch("app.middleware.audit_log.logger") as mock_logger: + await mw.dispatch(req, call_next) + + error_calls = [str(c) for c in mock_logger.error.call_args_list] + assert any("SERVER_ERROR" in c for c in error_calls) + + @pytest.mark.asyncio + async def test_login_post_triggers_auth_attempt_log(self): + mw = self._make_middleware(enabled=True) + mock_response = Response(content="ok", status_code=302) + call_next = AsyncMock(return_value=mock_response) + req = self._make_request(path="/auth", method="POST") + + with patch("app.middleware.audit_log.logger") as mock_logger: + await mw.dispatch(req, call_next) + + info_calls = [str(c) for c in mock_logger.info.call_args_list] + assert any("AUTH_ATTEMPT" in c for c in info_calls) + + @pytest.mark.asyncio + async def test_ip_included_in_log_when_enabled(self): + mw = self._make_middleware(enabled=True, include_ip=True) + mock_response = Response(content="ok", status_code=200) + call_next = AsyncMock(return_value=mock_response) + req = self._make_request() + + with patch("app.middleware.audit_log.logger") as mock_logger: + await mw.dispatch(req, call_next) + + info_calls = [str(c) for c in mock_logger.info.call_args_list] + # 127.0.0.1 should appear somewhere in the log + assert any("127.0.0.1" in c for c in info_calls) + + @pytest.mark.asyncio + async def test_ip_excluded_from_log_when_disabled(self): + mw = self._make_middleware(enabled=True, include_ip=False) + mock_response = Response(content="ok", status_code=200) + call_next = AsyncMock(return_value=mock_response) + req = self._make_request() + + with patch("app.middleware.audit_log.logger") as mock_logger: + await mw.dispatch(req, call_next) + + info_calls = [str(c) for c in mock_logger.info.call_args_list] + assert not any("127.0.0.1" in c for c in info_calls) + + +# --------------------------------------------------------------------------- +# Configuration settings +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +class TestAuditLoggingConfiguration: + """Tests that the audit logging settings are present in the app config.""" + + def test_audit_logging_enabled_setting_exists(self): + from app.config import settings + + assert hasattr(settings, "audit_logging_enabled") + assert isinstance(settings.audit_logging_enabled, bool) + + def test_audit_log_include_client_ip_setting_exists(self): + from app.config import settings + + assert hasattr(settings, "audit_log_include_client_ip") + assert isinstance(settings.audit_log_include_client_ip, bool) + + def test_audit_logging_enabled_by_default(self): + from app.config import Settings + + # Instantiate with only the required minimal fields + s = Settings( + database_url="sqlite:///:memory:", + redis_url="redis://localhost:6379/0", + openai_api_key="test", + azure_ai_key="test", + azure_region="test", + azure_endpoint="https://test.cognitiveservices.azure.com/", + gotenberg_url="http://localhost:3000", + workdir="/tmp", + ) + assert s.audit_logging_enabled is True + + def test_audit_log_include_client_ip_enabled_by_default(self): + from app.config import Settings + + s = Settings( + database_url="sqlite:///:memory:", + redis_url="redis://localhost:6379/0", + openai_api_key="test", + azure_ai_key="test", + azure_region="test", + azure_endpoint="https://test.cognitiveservices.azure.com/", + gotenberg_url="http://localhost:3000", + workdir="/tmp", + ) + assert s.audit_log_include_client_ip is True