sentinel_aggregator.py

#!/usr/bin/env python3
"""
SENTINEL Aggregator
Reads Cowrie JSON log + HTTP honeypot log + Canary log
Exposes /api/sentinel as a local HTTP API for lab-dashboard
Port 8282
"""
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from aiohttp import web

COWRIE_LOG   = Path("/home/wizardg/cowrie/var/log/cowrie/cowrie.json")
HTTP_LOG     = Path("/home/wizardg/sentinel/logs/http_honeypot.json")
CANARY_LOG   = Path("/home/wizardg/sentinel/logs/canary.json")


def _read_ndjson(path: Path, limit: int = 200) -> list:
    if not path.exists():
        return []
    lines = path.read_text().strip().splitlines()[-limit:]
    out = []
    for line in lines:
        try:
            out.append(json.loads(line))
        except Exception:
            pass
    return out


def _cowrie_stats(events: list) -> dict:
    sessions = {}
    creds = []
    commands = []
    for e in events:
        sid = e.get("session", "")
        etype = e.get("eventid", "")
        src = e.get("src_ip", "")

        if etype == "cowrie.session.connect":
            sessions[sid] = {
                "session": sid,
                "src_ip": src,
                "timestamp": e.get("timestamp", ""),
                "protocol": e.get("protocol", "ssh"),
                "sensor": e.get("sensor", ""),
            }
        elif etype in ("cowrie.login.failed", "cowrie.login.success"):
            creds.append({
                "session": sid,
                "src_ip": src,
                "username": e.get("username", ""),
                "password": e.get("password", ""),
                "success": etype == "cowrie.login.success",
                "timestamp": e.get("timestamp", ""),
            })
        elif etype == "cowrie.command.input":
            commands.append({
                "session": sid,
                "src_ip": src,
                "command": e.get("input", ""),
                "timestamp": e.get("timestamp", ""),
            })

    return {
        "active_sessions": len(sessions),
        "sessions": list(sessions.values())[-20:],
        "credential_attempts": len(creds),
        "credentials": creds[-50:],
        "commands": commands[-30:],
    }


def _http_stats(events: list) -> dict:
    probes = [e for e in events if e.get("event_type") == "http_probe"]
    creds  = [e for e in events if e.get("event_type") == "credential_attempt"]
    unique_ips = len({e.get("ip") for e in events if e.get("ip")})
    top_paths = {}
    for e in probes:
        p = e.get("path", "")
        top_paths[p] = top_paths.get(p, 0) + 1
    top_paths_sorted = sorted(top_paths.items(), key=lambda x: x[1], reverse=True)[:10]
    return {
        "total_probes": len(probes),
        "credential_attempts": len(creds),
        "unique_ips": unique_ips,
        "top_paths": [{"path": p, "count": c} for p, c in top_paths_sorted],
        "recent_creds": creds[-20:],
        "recent_probes": probes[-20:],
    }


def _canary_stats(events: list) -> dict:
    triggers = [e for e in events if e.get("event_type") == "canary_trigger"]
    return {
        "total_triggers": len(triggers),
        "recent_triggers": triggers[-20:],
    }


async def sentinel_api(request: web.Request):
    cowrie_events = _read_ndjson(COWRIE_LOG)
    http_events   = _read_ndjson(HTTP_LOG)
    canary_events = _read_ndjson(CANARY_LOG)

    cowrie = _cowrie_stats(cowrie_events)
    http   = _http_stats(http_events)
    canary = _canary_stats(canary_events)

    # Combined timeline — last 30 events across all sources
    timeline = []
    for e in cowrie_events[-100:]:
        etype = e.get("eventid", "")
        if etype in ("cowrie.login.failed", "cowrie.login.success", "cowrie.session.connect", "cowrie.command.input"):
            timeline.append({
                "source": "cowrie",
                "type": etype.split(".")[-1],
                "ip": e.get("src_ip", ""),
                "detail": e.get("username") or e.get("input") or "",
                "timestamp": e.get("timestamp", ""),
                "severity": "high" if etype == "cowrie.login.success" else "medium",
            })
    for e in http_events[-100:]:
        if e.get("event_type") == "credential_attempt":
            timeline.append({
                "source": "http_honeypot",
                "type": "credential",
                "ip": e.get("ip", ""),
                "detail": f"{e.get('username')} / {e.get('path','')}",
                "timestamp": e.get("@timestamp", ""),
                "severity": "high",
            })
    for e in canary_events[-50:]:
        if e.get("event_type") == "canary_trigger":
            timeline.append({
                "source": "canary",
                "type": "trigger",
                "ip": e.get("ip", ""),
                "detail": e.get("label", ""),
                "timestamp": e.get("@timestamp", ""),
                "severity": "critical",
            })

    timeline.sort(key=lambda x: x.get("timestamp", ""), reverse=True)

    total_alerts = (
        cowrie["credential_attempts"] +
        http["credential_attempts"] +
        canary["total_triggers"]
    )

    return web.json_response({
        "status": "ok",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "summary": {
            "total_alerts": total_alerts,
            "ssh_sessions": cowrie["active_sessions"],
            "http_probes": http["total_probes"],
            "canary_triggers": canary["total_triggers"],
            "unique_attackers": http["unique_ips"],
        },
        "cowrie": cowrie,
        "http": http,
        "canary": canary,
        "timeline": timeline[:30],
    })


async def health(request: web.Request):
    return web.json_response({"ok": True})


def create_app():
    app = web.Application()
    app.router.add_get("/api/sentinel", sentinel_api)
    app.router.add_get("/health", health)
    return app


if __name__ == "__main__":
    print("[SENTINEL] Aggregator API — port 8282")
    web.run_app(create_app(), host="127.0.0.1", port=8282, access_log=None)