fix: add security log investigation filters
Deploy Dashboard (Dev) / deploy-dev (push) Successful in 3m22s
Deploy Dashboard (Dev) / deploy-dev (push) Successful in 3m22s
Add server-side IP and date range filtering for security logs so investigations can narrow events without relying on client-side filtering alone. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
import docker
|
||||
import re
|
||||
import json
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from datetime import datetime, timezone, timedelta, time
|
||||
from collections import Counter
|
||||
from fastapi import APIRouter, Query
|
||||
from config import DOCKER_HOST
|
||||
@@ -130,10 +130,60 @@ def _parse_caddy_logs(lines: str, limit: int) -> list[dict]:
|
||||
return entries[-limit:]
|
||||
|
||||
|
||||
def _parse_timestamp(value: str) -> datetime | None:
|
||||
if not value:
|
||||
return None
|
||||
try:
|
||||
return datetime.strptime(value, "%Y-%m-%d %H:%M:%S").replace(tzinfo=_tz_cst)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _filter_security_entries(
|
||||
entries: list[dict],
|
||||
event_type: str | None,
|
||||
ip: str | None,
|
||||
start_date: str | None,
|
||||
end_date: str | None,
|
||||
) -> list[dict]:
|
||||
normalized_ip = ip.strip() if ip else None
|
||||
start_dt = None
|
||||
end_dt = None
|
||||
|
||||
if start_date:
|
||||
start_dt = datetime.combine(datetime.strptime(start_date, "%Y-%m-%d").date(), time.min, tzinfo=_tz_cst)
|
||||
if end_date:
|
||||
end_dt = datetime.combine(datetime.strptime(end_date, "%Y-%m-%d").date(), time.max, tzinfo=_tz_cst)
|
||||
|
||||
filtered = []
|
||||
for entry in entries:
|
||||
if event_type and entry.get("type") != event_type:
|
||||
continue
|
||||
if normalized_ip and entry.get("ip", "").strip() != normalized_ip:
|
||||
continue
|
||||
|
||||
if start_dt or end_dt:
|
||||
entry_ts = _parse_timestamp(entry.get("timestamp", ""))
|
||||
if entry_ts is None:
|
||||
continue
|
||||
if start_dt and entry_ts < start_dt:
|
||||
continue
|
||||
if end_dt and entry_ts > end_dt:
|
||||
continue
|
||||
|
||||
filtered.append(entry)
|
||||
|
||||
return filtered
|
||||
|
||||
|
||||
@router.get("/logs")
|
||||
def security_logs(
|
||||
tail: int = Query(200, le=2000),
|
||||
limit: int = Query(100, le=500),
|
||||
type: str | None = Query(None),
|
||||
ip: str | None = Query(None),
|
||||
start_date: str | None = Query(None),
|
||||
end_date: str | None = Query(None),
|
||||
):
|
||||
"""Fetch and parse security-relevant logs from Authelia and Caddy."""
|
||||
results = []
|
||||
@@ -153,7 +203,8 @@ def security_logs(
|
||||
|
||||
# Sort by timestamp descending
|
||||
results.sort(key=lambda x: x.get("timestamp", ""), reverse=True)
|
||||
return {"logs": results[:limit]}
|
||||
filtered = _filter_security_entries(results, type, ip, start_date, end_date)
|
||||
return {"logs": filtered[:limit]}
|
||||
|
||||
|
||||
@router.get("/stats")
|
||||
|
||||
Reference in New Issue
Block a user