Cybersecurity-Projects/PROJECTS/beginner/linux-ebpf-security-tracer/src/renderer.py

"""
©AngelaMos | 2026
renderer.py
"""
from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import IO, Any, TextIO

from rich.console import Console
from rich.table import Table
from rich.text import Text

from .config import (
    SEVERITY_COLORS,
    OutputFormat,
)
from .processor import TracerEvent


def _event_to_dict(event: TracerEvent) -> dict[str, Any]:
    """
    Serialize a TracerEvent to a JSON-compatible dict
    """
    d: dict[str, Any] = {
        "timestamp": event.timestamp.isoformat(),
        "event_type": event.event_type,
        "category": event.category,
        "pid": event.pid,
        "ppid": event.ppid,
        "uid": event.uid,
        "username": event.username,
        "comm": event.comm,
        "severity": event.severity,
    }

    if event.filename:
        d["filename"] = event.filename

    if event.addr_v4:
        d["dest_ip"] = event.addr_v4
        d["dest_port"] = event.port

    if event.event_type in ("setuid", "setgid"):
        d["target_uid"] = event.target_uid
        d["target_gid"] = event.target_gid

    if event.event_type == "ptrace":
        d["ptrace_request"] = event.ptrace_request
        d["target_pid"] = event.target_pid

    if event.detection:
        d["detection"] = event.detection
        d["detection_id"] = event.detection_id
        d["mitre_id"] = event.mitre_id

    parent_comm = event.extra.get("parent_comm")
    if parent_comm:
        d["parent_comm"] = parent_comm

    return d


class JsonRenderer:
    """
    Outputs one JSON object per line
    """

    def __init__(
        self,
        stream: TextIO = sys.stdout,
    ) -> None:
        """
        Initialize with output stream
        """
        self._stream = stream

    def render(self, event: TracerEvent) -> None:
        """
        Write a single event as JSON
        """
        d = _event_to_dict(event)
        self._stream.write(json.dumps(d) + "\n")
        self._stream.flush()


class LiveRenderer:
    """
    Color-coded streaming output using Rich
    """

    def __init__(
        self,
        console: Console | None = None,
    ) -> None:
        """
        Initialize with Rich console
        """
        self._console = console or Console()

    def render(self, event: TracerEvent) -> None:
        """
        Print a color-coded event line
        """
        ts = event.timestamp.strftime("%H:%M:%S")
        color = SEVERITY_COLORS.get(event.severity, "white")

        sev = Text(f"{event.severity:8s}", style=color)

        detail = self._format_detail(event)

        line = Text()
        line.append(f"[{ts}] ")
        line.append_text(sev)
        line.append(f" {event.event_type:14s} ")
        line.append(f"pid={event.pid} "
                    f"comm={event.comm} ")
        if detail:
            line.append(detail)

        if event.detection:
            det_text = Text(
                f" [{event.detection}]",
                style="bold magenta",
            )
            line.append_text(det_text)

        self._console.print(line)

    def _format_detail(
        self,
        event: TracerEvent,
    ) -> str:
        """
        Build detail string based on event type
        """
        if event.filename:
            return event.filename
        if event.addr_v4:
            return f"{event.addr_v4}:{event.port}"
        if event.target_uid:
            return f"uid->{event.target_uid}"
        if event.ptrace_request:
            return (f"req={event.ptrace_request} "
                    f"target={event.target_pid}")
        return ""


class TableRenderer:
    """
    Periodic table summaries using Rich
    """

    def __init__(
        self,
        console: Console | None = None,
    ) -> None:
        """
        Initialize with Rich console and event buffer
        """
        self._console = console or Console()
        self._buffer: list[TracerEvent] = []
        self._flush_count = 20

    def render(self, event: TracerEvent) -> None:
        """
        Buffer events and flush as table periodically
        """
        self._buffer.append(event)
        if len(self._buffer) >= self._flush_count:
            self._flush()

    def _flush(self) -> None:
        """
        Render buffered events as a Rich table
        """
        if not self._buffer:
            return

        table = Table(
            title="eBPF Security Events",
            show_lines=False,
        )
        table.add_column("Time", width=8)
        table.add_column("Severity", width=8)
        table.add_column("Type", width=12)
        table.add_column("PID", width=7)
        table.add_column("Comm", width=15)
        table.add_column("Detail", min_width=20)
        table.add_column("Detection", width=18)

        for ev in self._buffer:
            color = SEVERITY_COLORS.get(ev.severity, "white")
            ts = ev.timestamp.strftime("%H:%M:%S")
            detail = ""
            if ev.filename:
                detail = ev.filename
            elif ev.addr_v4:
                detail = f"{ev.addr_v4}:{ev.port}"

            table.add_row(
                ts,
                Text(ev.severity, style=color),
                ev.event_type,
                str(ev.pid),
                ev.comm,
                detail,
                ev.detection or "",
            )

        self._console.print(table)
        self._buffer.clear()

    def finalize(self) -> None:
        """
        Flush remaining events on shutdown
        """
        self._flush()


class FileRenderer:
    """
    Append JSON events to a file
    """

    def __init__(self, path: Path) -> None:
        """
        Initialize with output file path
        """
        self._path = path
        self._fh: IO[str] | None = None

    def render(self, event: TracerEvent) -> None:
        """
        Append a single event as JSON to the file
        """
        if self._fh is None:
            self._fh = open(self._path, "a")

        d = _event_to_dict(event)
        self._fh.write(json.dumps(d) + "\n")
        self._fh.flush()

    def close(self) -> None:
        """
        Close the output file handle
        """
        if self._fh is not None:
            self._fh.close()
            self._fh = None


def create_renderer(
    fmt: OutputFormat, ) -> JsonRenderer | LiveRenderer | TableRenderer:
    """
    Factory for the appropriate renderer
    """
    if fmt == "json":
        return JsonRenderer()
    if fmt == "table":
        return TableRenderer()
    return LiveRenderer()