8054c1e1e4
- Environment-split settings: base/local/test/prod with django-environ - Postgres + Redis + Celery wiring (broker, beat, result backend) - All 9 domain app stubs: accounts, courses, cms, tracking, quizzes, training, certificates, reports, notifications - api app: /healthz/ endpoint, custom DRF exception handler, SecurityAuditMiddleware, permissions/throttle/upload-validation stubs - DRF global baseline: JWT+session auth, closed-by-default permissions, cursor/page pagination, drf-spectacular schema generation - Dockerfile (multi-env build arg), docker-compose.yml (local), docker-compose.test.yml (CI-friendly tmpfs Postgres) - pytest.ini with smoke + settings marker definitions - tests/test_smoke.py: startup, URL resolution, healthcheck shape - tests/test_settings_matrix.py: per-profile security assertions - .github/workflows/ci.yml: test, lint, schema CI jobs - .env.example with all required vars documented - .gitignore Co-Authored-By: Paperclip <noreply@paperclip.ing>
71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
import logging
|
|
import time
|
|
import uuid
|
|
|
|
access_logger = logging.getLogger(__name__)
|
|
security_logger = logging.getLogger("security")
|
|
|
|
# Paths where any 4xx triggers a security-level log entry
|
|
_SENSITIVE_PATH_PREFIXES = (
|
|
"/api/v1/auth/",
|
|
"/api/v1/token/",
|
|
"/api/v1/upload/",
|
|
"/api/v1/password/",
|
|
)
|
|
|
|
# Status codes always logged as security events regardless of path
|
|
_SECURITY_STATUSES = {401, 403, 429}
|
|
|
|
|
|
class SecurityAuditMiddleware:
|
|
"""Attaches a request-scoped ID, emits access logs, and flags security events."""
|
|
|
|
def __init__(self, get_response):
|
|
self.get_response = get_response
|
|
|
|
def __call__(self, request):
|
|
request.request_id = request.META.get("HTTP_X_REQUEST_ID") or str(uuid.uuid4())
|
|
start = time.monotonic()
|
|
response = self.get_response(request)
|
|
duration_ms = round((time.monotonic() - start) * 1000)
|
|
|
|
extra = {
|
|
"request_id": request.request_id,
|
|
"method": request.method,
|
|
"path": request.path,
|
|
"status": response.status_code,
|
|
"user": getattr(getattr(request, "user", None), "pk", None),
|
|
"ip": self._client_ip(request),
|
|
"duration_ms": duration_ms,
|
|
}
|
|
|
|
access_logger.info("request", extra=extra)
|
|
|
|
if self._is_security_event(request, response):
|
|
security_logger.warning(
|
|
"security_event path=%s method=%s status=%s user=%s ip=%s duration_ms=%s",
|
|
request.path,
|
|
request.method,
|
|
response.status_code,
|
|
extra["user"] or "anonymous",
|
|
extra["ip"],
|
|
duration_ms,
|
|
)
|
|
|
|
return response
|
|
|
|
@staticmethod
|
|
def _is_security_event(request, response) -> bool:
|
|
if response.status_code in _SECURITY_STATUSES:
|
|
return True
|
|
return response.status_code >= 400 and any(
|
|
request.path.startswith(p) for p in _SENSITIVE_PATH_PREFIXES
|
|
)
|
|
|
|
@staticmethod
|
|
def _client_ip(request) -> str:
|
|
forwarded = request.META.get("HTTP_X_FORWARDED_FOR")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
return request.META.get("REMOTE_ADDR", "unknown")
|