copilot-swe-agent[bot]
653c137222
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
332 lines
11 KiB
Python
332 lines
11 KiB
Python
"""
|
||
Comprehensive audit-event service for DocuElevate.
|
||
|
||
Provides helpers to **record** audit events (append-only database writes)
|
||
and to optionally **forward** them to external SIEM systems.
|
||
|
||
Supported SIEM transports:
|
||
* **Syslog** – RFC 5424 structured-data messages over UDP or TCP.
|
||
* **HTTP** – JSON POST payloads compatible with Splunk HEC, Logstash
|
||
HTTP input, Grafana Loki push API, and any generic webhook endpoint.
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
import socket
|
||
import threading
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from fastapi import Request
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.config import settings
|
||
from app.middleware.audit_log import get_client_ip, get_username
|
||
from app.models import AuditLog
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Public helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def record_event(
|
||
db: Session,
|
||
*,
|
||
action: str,
|
||
user: str = "system",
|
||
resource_type: str | None = None,
|
||
resource_id: str | None = None,
|
||
ip_address: str | None = None,
|
||
details: dict[str, Any] | None = None,
|
||
severity: str = "info",
|
||
) -> AuditLog:
|
||
"""Persist an audit event and optionally forward it to SIEM.
|
||
|
||
Args:
|
||
db: Active SQLAlchemy session.
|
||
action: Short action identifier (e.g. ``"login"``, ``"document.create"``).
|
||
user: Username performing the action.
|
||
resource_type: Category of the affected resource (``"document"``, ``"user"`` …).
|
||
resource_id: Identifier of the affected resource.
|
||
ip_address: Client IP address (``None`` when not applicable).
|
||
details: Arbitrary key/value context serialised as JSON.
|
||
severity: One of ``info``, ``warning``, ``error``, ``critical``.
|
||
|
||
Returns:
|
||
The newly created :class:`AuditLog` row.
|
||
"""
|
||
details_json = json.dumps(details, default=str) if details else None
|
||
|
||
entry = AuditLog(
|
||
user=user,
|
||
action=action,
|
||
resource_type=resource_type,
|
||
resource_id=str(resource_id) if resource_id is not None else None,
|
||
ip_address=ip_address,
|
||
details=details_json,
|
||
severity=severity,
|
||
)
|
||
db.add(entry)
|
||
db.commit()
|
||
db.refresh(entry)
|
||
|
||
# Fire-and-forget SIEM forwarding in a background thread so we never
|
||
# block the request path.
|
||
if settings.audit_siem_enabled:
|
||
payload = _build_siem_payload(entry)
|
||
thread = threading.Thread(target=_forward_to_siem, args=(payload,), daemon=True)
|
||
thread.start()
|
||
|
||
return entry
|
||
|
||
|
||
def record_event_from_request(
|
||
db: Session,
|
||
request: Request,
|
||
*,
|
||
action: str,
|
||
resource_type: str | None = None,
|
||
resource_id: str | None = None,
|
||
details: dict[str, Any] | None = None,
|
||
severity: str = "info",
|
||
) -> AuditLog:
|
||
"""Convenience wrapper that extracts user and IP from a :class:`Request`.
|
||
|
||
Args:
|
||
db: Active SQLAlchemy session.
|
||
request: The current HTTP request.
|
||
action: Short action identifier.
|
||
resource_type: Category of the affected resource.
|
||
resource_id: Identifier of the affected resource.
|
||
details: Arbitrary key/value context serialised as JSON.
|
||
severity: One of ``info``, ``warning``, ``error``, ``critical``.
|
||
|
||
Returns:
|
||
The newly created :class:`AuditLog` row.
|
||
"""
|
||
return record_event(
|
||
db,
|
||
action=action,
|
||
user=get_username(request),
|
||
resource_type=resource_type,
|
||
resource_id=resource_id,
|
||
ip_address=get_client_ip(request),
|
||
details=details,
|
||
severity=severity,
|
||
)
|
||
|
||
|
||
def query_events(
|
||
db: Session,
|
||
*,
|
||
action: str | None = None,
|
||
user: str | None = None,
|
||
resource_type: str | None = None,
|
||
severity: str | None = None,
|
||
since: datetime | None = None,
|
||
until: datetime | None = None,
|
||
limit: int = 200,
|
||
offset: int = 0,
|
||
) -> list[AuditLog]:
|
||
"""Query audit log entries with optional filtering.
|
||
|
||
Args:
|
||
db: Active SQLAlchemy session.
|
||
action: Filter by action string (exact match).
|
||
user: Filter by username (exact match).
|
||
resource_type: Filter by resource type (exact match).
|
||
severity: Filter by severity level (exact match).
|
||
since: Only events at or after this timestamp.
|
||
until: Only events at or before this timestamp.
|
||
limit: Maximum number of rows to return.
|
||
offset: Number of rows to skip (for pagination).
|
||
|
||
Returns:
|
||
List of :class:`AuditLog` rows ordered by *timestamp descending*.
|
||
"""
|
||
q = db.query(AuditLog)
|
||
if action:
|
||
q = q.filter(AuditLog.action == action)
|
||
if user:
|
||
q = q.filter(AuditLog.user == user)
|
||
if resource_type:
|
||
q = q.filter(AuditLog.resource_type == resource_type)
|
||
if severity:
|
||
q = q.filter(AuditLog.severity == severity)
|
||
if since:
|
||
q = q.filter(AuditLog.timestamp >= since)
|
||
if until:
|
||
q = q.filter(AuditLog.timestamp <= until)
|
||
return q.order_by(AuditLog.timestamp.desc()).offset(offset).limit(limit).all()
|
||
|
||
|
||
def count_events(
|
||
db: Session,
|
||
*,
|
||
action: str | None = None,
|
||
user: str | None = None,
|
||
resource_type: str | None = None,
|
||
severity: str | None = None,
|
||
since: datetime | None = None,
|
||
until: datetime | None = None,
|
||
) -> int:
|
||
"""Return the total count of events matching the given filters.
|
||
|
||
Args:
|
||
db: Active SQLAlchemy session.
|
||
action: Filter by action string.
|
||
user: Filter by username.
|
||
resource_type: Filter by resource type.
|
||
severity: Filter by severity level.
|
||
since: Only events at or after this timestamp.
|
||
until: Only events at or before this timestamp.
|
||
|
||
Returns:
|
||
Integer count.
|
||
"""
|
||
q = db.query(AuditLog)
|
||
if action:
|
||
q = q.filter(AuditLog.action == action)
|
||
if user:
|
||
q = q.filter(AuditLog.user == user)
|
||
if resource_type:
|
||
q = q.filter(AuditLog.resource_type == resource_type)
|
||
if severity:
|
||
q = q.filter(AuditLog.severity == severity)
|
||
if since:
|
||
q = q.filter(AuditLog.timestamp >= since)
|
||
if until:
|
||
q = q.filter(AuditLog.timestamp <= until)
|
||
return q.count()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SIEM forwarding internals
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_SYSLOG_FACILITY_LOCAL0 = 16
|
||
_SYSLOG_SEVERITY_MAP = {
|
||
"info": 6,
|
||
"warning": 4,
|
||
"error": 3,
|
||
"critical": 2,
|
||
}
|
||
|
||
|
||
def _build_siem_payload(entry: AuditLog) -> dict[str, Any]:
|
||
"""Convert an :class:`AuditLog` row into a plain dict for SIEM delivery."""
|
||
ts = entry.timestamp if entry.timestamp else datetime.now(timezone.utc)
|
||
return {
|
||
"id": entry.id,
|
||
"timestamp": ts.isoformat(),
|
||
"user": entry.user,
|
||
"action": entry.action,
|
||
"resource_type": entry.resource_type,
|
||
"resource_id": entry.resource_id,
|
||
"ip_address": entry.ip_address,
|
||
"details": entry.details,
|
||
"severity": entry.severity,
|
||
"source": "docuelevate",
|
||
}
|
||
|
||
|
||
def _forward_to_siem(payload: dict[str, Any]) -> None:
|
||
"""Route a SIEM payload to the configured transport."""
|
||
transport = settings.audit_siem_transport.lower()
|
||
try:
|
||
if transport == "syslog":
|
||
_send_syslog(payload)
|
||
elif transport == "http":
|
||
_send_http(payload)
|
||
else:
|
||
logger.warning("Unknown SIEM transport %r; skipping forwarding", transport)
|
||
except Exception:
|
||
logger.exception("Failed to forward audit event to SIEM (%s)", transport)
|
||
|
||
|
||
def _send_syslog(payload: dict[str, Any]) -> None:
|
||
"""Send a RFC 5424 syslog message to the configured receiver."""
|
||
severity_num = _SYSLOG_SEVERITY_MAP.get(payload.get("severity", "info"), 6)
|
||
priority = _SYSLOG_FACILITY_LOCAL0 * 8 + severity_num
|
||
ts = payload.get("timestamp", datetime.now(timezone.utc).isoformat())
|
||
hostname = socket.gethostname()
|
||
app_name = "docuelevate"
|
||
msg_id = payload.get("action", "-")
|
||
|
||
# Structured data (SD) element with key event fields.
|
||
sd = (
|
||
f'[docuelevate@0 user="{payload.get("user", "-")}" '
|
||
f'action="{payload.get("action", "-")}" '
|
||
f'resource_type="{payload.get("resource_type", "-")}" '
|
||
f'resource_id="{payload.get("resource_id", "-")}" '
|
||
f'ip="{payload.get("ip_address", "-")}"]'
|
||
)
|
||
message = json.dumps(payload, default=str)
|
||
syslog_msg = f"<{priority}>1 {ts} {hostname} {app_name} - {msg_id} {sd} {message}"
|
||
|
||
proto = settings.audit_siem_syslog_protocol.lower()
|
||
host = settings.audit_siem_syslog_host
|
||
port = settings.audit_siem_syslog_port
|
||
|
||
if proto == "tcp":
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||
sock.settimeout(5)
|
||
sock.connect((host, port))
|
||
sock.sendall(syslog_msg.encode("utf-8"))
|
||
else:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
||
sock.settimeout(5)
|
||
sock.sendto(syslog_msg.encode("utf-8"), (host, port))
|
||
|
||
logger.debug("Syslog audit event sent to %s:%s (%s)", host, port, proto)
|
||
|
||
|
||
def _send_http(payload: dict[str, Any]) -> None:
|
||
"""POST a JSON audit event to the configured HTTP endpoint."""
|
||
url = settings.audit_siem_http_url
|
||
if not url:
|
||
logger.warning("SIEM HTTP URL not configured; skipping HTTP forwarding")
|
||
return
|
||
|
||
headers: dict[str, str] = {"Content-Type": "application/json"}
|
||
token = settings.audit_siem_http_token
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
|
||
# Parse custom headers (comma-separated "Key:Value" pairs).
|
||
# Reject headers that could override security-critical ones already set,
|
||
# and validate that header names contain only RFC 7230 token characters.
|
||
_PROTECTED_HEADERS = {"authorization", "content-type", "host"}
|
||
_VALID_HEADER_NAME = re.compile(r"^[A-Za-z0-9!#$%&'*+\-.^_`|~]+$")
|
||
raw_custom = settings.audit_siem_http_custom_headers
|
||
if raw_custom:
|
||
for raw_pair in raw_custom.split(","):
|
||
pair = raw_pair.strip()
|
||
if ":" in pair:
|
||
k, _, v = pair.partition(":")
|
||
name = k.strip()
|
||
if not name or not _VALID_HEADER_NAME.match(name):
|
||
logger.warning("Skipping invalid SIEM custom header name: %r", name)
|
||
continue
|
||
if name.lower() in _PROTECTED_HEADERS:
|
||
logger.warning("Skipping protected SIEM custom header: %r", name)
|
||
continue
|
||
headers[name] = v.strip()
|
||
|
||
# Wrap in Splunk HEC-style envelope when URL contains ``/services/collector``.
|
||
body: dict[str, Any]
|
||
if "/services/collector" in url:
|
||
body = {"event": payload, "sourcetype": "docuelevate:audit", "source": "docuelevate"}
|
||
else:
|
||
body = payload
|
||
|
||
with httpx.Client(timeout=10) as client:
|
||
resp = client.post(url, json=body, headers=headers)
|
||
resp.raise_for_status()
|
||
|
||
logger.debug("HTTP audit event forwarded to %s (status %s)", url, resp.status_code)
|