Gan, Jimmy
262 lines
8.3 KiB
Python
262 lines
8.3 KiB
Python
import docker
|
|
import re
|
|
import json
|
|
from datetime import datetime, timezone, timedelta, time
|
|
from collections import Counter
|
|
from fastapi import APIRouter, Query
|
|
from config import DOCKER_HOST
|
|
|
|
router = APIRouter()
|
|
|
|
_client = None
|
|
|
|
def get_docker_client():
|
|
"""Get or create Docker client (lazy initialization)."""
|
|
global _client
|
|
if _client is None:
|
|
_client = docker.DockerClient(base_url=DOCKER_HOST, timeout=10)
|
|
return _client
|
|
|
|
_tz_cst = timezone(timedelta(hours=8))
|
|
|
|
# Suspicious path patterns (scanners, exploits, etc.)
|
|
_SUSPICIOUS_PATHS = re.compile(
|
|
r"(\.env|\.git/|wp-login|wp-admin|phpmy|/cgi-bin|/shell|/eval|"
|
|
r"\.asp|/etc/passwd|/proc/self|\.sql|/actuator|/debug|/config\.json|"
|
|
r"xmlrpc\.php|wp-cron\.php|wp-includes|/\.well-known/security)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
# Known internal API paths that should not be flagged
|
|
_INTERNAL_PATHS = re.compile(
|
|
r"^/(api/|rest/|backend/|_next/|static/|assets/|favicon)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
_LOGFMT_RE = re.compile(r'(\w+)="([^"]*)"|\b(\w+)=(\S+)')
|
|
|
|
|
|
def _parse_logfmt(line: str) -> dict:
|
|
"""Parse a logfmt line into a dict."""
|
|
result = {}
|
|
for m in _LOGFMT_RE.finditer(line):
|
|
if m.group(1):
|
|
result[m.group(1)] = m.group(2)
|
|
elif m.group(3):
|
|
result[m.group(3)] = m.group(4)
|
|
return result
|
|
|
|
|
|
def _parse_authelia_logs(lines: str, limit: int) -> list[dict]:
|
|
"""Extract failed auth attempts from Authelia logs (logfmt or JSON)."""
|
|
entries = []
|
|
for line in lines.strip().splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
# Try JSON first, fall back to logfmt
|
|
try:
|
|
obj = json.loads(line)
|
|
except (json.JSONDecodeError, ValueError):
|
|
obj = _parse_logfmt(line)
|
|
if not obj:
|
|
continue
|
|
|
|
level = obj.get("level", "")
|
|
msg = obj.get("msg", "")
|
|
|
|
is_failed = (
|
|
level in ("error", "warning")
|
|
and any(kw in msg.lower() for kw in ("unsuccessful", "banned", "failed", "invalid", "denied"))
|
|
)
|
|
# Also catch "not authorized" for anonymous access attempts
|
|
is_unauthorized = "is not authorized" in msg.lower()
|
|
|
|
if not is_failed and not is_unauthorized:
|
|
continue
|
|
|
|
ts_raw = obj.get("time", "")
|
|
try:
|
|
ts = datetime.fromisoformat(ts_raw.replace("Z", "+00:00")).astimezone(_tz_cst).strftime("%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
ts = ts_raw[:19] if ts_raw else ""
|
|
|
|
entries.append({
|
|
"type": "auth_failure" if is_failed else "unauthorized",
|
|
"timestamp": ts,
|
|
"message": msg,
|
|
"username": obj.get("username", obj.get("user", "")),
|
|
"ip": obj.get("remote_ip", obj.get("ip", "")),
|
|
"level": level,
|
|
})
|
|
|
|
return entries[-limit:]
|
|
|
|
|
|
def _parse_caddy_logs(lines: str, limit: int) -> list[dict]:
|
|
"""Extract suspicious requests from Caddy JSON access logs."""
|
|
entries = []
|
|
for line in lines.strip().splitlines():
|
|
try:
|
|
obj = json.loads(line.strip())
|
|
except (json.JSONDecodeError, ValueError):
|
|
continue
|
|
|
|
status = obj.get("status", 0)
|
|
uri = obj.get("uri", obj.get("request", {}).get("uri", ""))
|
|
remote_ip = obj.get("remote_ip", obj.get("request", {}).get("remote_ip", ""))
|
|
|
|
# Flag: suspicious path (skip known internal paths), or 4xx on unknown paths
|
|
is_suspicious = False
|
|
if _SUSPICIOUS_PATHS.search(uri):
|
|
is_suspicious = not _INTERNAL_PATHS.match(uri)
|
|
elif status >= 400 and not _INTERNAL_PATHS.match(uri):
|
|
is_suspicious = True
|
|
if not is_suspicious:
|
|
continue
|
|
|
|
ts_raw = obj.get("ts", obj.get("timestamp", ""))
|
|
try:
|
|
if isinstance(ts_raw, (int, float)):
|
|
ts = datetime.fromtimestamp(ts_raw, tz=_tz_cst).strftime("%Y-%m-%d %H:%M:%S")
|
|
else:
|
|
ts = datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")).astimezone(_tz_cst).strftime("%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
ts = str(ts_raw)[:19]
|
|
|
|
entries.append({
|
|
"type": "suspicious_request",
|
|
"timestamp": ts,
|
|
"message": f"{status} {obj.get('request', {}).get('method', 'GET')} {uri}",
|
|
"ip": remote_ip,
|
|
"status": status,
|
|
"path": uri,
|
|
})
|
|
|
|
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 = []
|
|
|
|
# Authelia logs
|
|
try:
|
|
client = get_docker_client()
|
|
authelia = client.containers.get("authelia")
|
|
raw = authelia.logs(tail=tail, timestamps=False).decode(errors="replace")
|
|
results.extend(_parse_authelia_logs(raw, limit))
|
|
except docker.errors.NotFound:
|
|
pass
|
|
except Exception as e:
|
|
results.append({"type": "error", "message": f"Authelia log error: {e}", "timestamp": "", "ip": ""})
|
|
|
|
# Caddy logs disabled — Caddy outputs massive logs that timeout the Docker API
|
|
# TODO: re-enable when Caddy is configured with structured access logging to a file
|
|
|
|
# Sort by timestamp descending
|
|
results.sort(key=lambda x: x.get("timestamp", ""), reverse=True)
|
|
filtered = _filter_security_entries(results, type, ip, start_date, end_date)
|
|
return {"logs": filtered[:limit]}
|
|
|
|
|
|
@router.get("/stats")
|
|
def security_stats(tail: int = Query(500, le=2000)):
|
|
"""Aggregated security stats: failed logins, top IPs, hourly timeline."""
|
|
auth_entries = []
|
|
caddy_entries = []
|
|
|
|
try:
|
|
client = get_docker_client()
|
|
authelia = client.containers.get("authelia")
|
|
raw = authelia.logs(tail=tail, timestamps=False).decode(errors="replace")
|
|
auth_entries = _parse_authelia_logs(raw, 1000)
|
|
except Exception:
|
|
pass
|
|
|
|
# Caddy logs disabled — too large, causes Docker API timeouts
|
|
|
|
all_entries = auth_entries + caddy_entries
|
|
|
|
# Count by IP
|
|
ip_counter = Counter(e.get("ip", "unknown") for e in all_entries if e.get("ip"))
|
|
top_ips = ip_counter.most_common(10)
|
|
|
|
# Hourly timeline (last 24 hours)
|
|
now = datetime.now(_tz_cst)
|
|
hourly = Counter()
|
|
for e in all_entries:
|
|
try:
|
|
ts = datetime.strptime(e["timestamp"], "%Y-%m-%d %H:%M:%S").replace(tzinfo=_tz_cst)
|
|
hours_ago = int((now - ts).total_seconds() / 3600)
|
|
if 0 <= hours_ago < 24:
|
|
hourly[hours_ago] += 1
|
|
except Exception:
|
|
continue
|
|
|
|
timeline = [{"hours_ago": h, "count": hourly.get(h, 0)} for h in range(24)]
|
|
|
|
return {
|
|
"failed_logins_24h": len(auth_entries),
|
|
"suspicious_requests": len(caddy_entries),
|
|
"unique_ips": len(ip_counter),
|
|
"top_ips": [{"ip": ip, "count": c} for ip, c in top_ips],
|
|
"timeline": timeline,
|
|
}
|