nas-tools/dashboard/backend/routers/security.py

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 = docker.DockerClient(base_url=DOCKER_HOST, timeout=10)

_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:
        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:
        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,
    }