proofpoint-quarantine-portal/logs.py

"""Log tailing for the admin log viewer.

Reads the tail of a rotating log file backwards in blocks — never loads the whole file.
The log source is an ENUM KEY resolved to a path server-side; the client can never name
a file path (a `?path=` parameter here would be an arbitrary-file-read hole).

Level classification is done here, server-side, so the ops-log format stays owned by the
backend and the browser just colours by the returned `level`.
"""

from __future__ import annotations

import os
from pathlib import Path

MAX_LINES = 2000
MAX_BYTES = 2_000_000

# name -> function that returns the file path from the app config section.
LOG_SOURCES = {
    "ops": lambda app_cfg: app_cfg.get("worker_log", "worker.log"),
    "app": lambda app_cfg: app_cfg.get("app_log", "app.log"),
}


def resolve_path(source: str, app_cfg) -> Path:
    if source not in LOG_SOURCES:
        raise KeyError(source)
    return Path(LOG_SOURCES[source](app_cfg)).resolve()


def tail(path: Path, lines: int = 200, max_bytes: int = MAX_BYTES) -> tuple[list[str], bool]:
    """Return (last `lines` lines, truncated?). Reads backwards; bounded memory."""
    lines = max(1, min(lines, MAX_LINES))
    if not path.exists():
        return [], False
    with path.open("rb") as fh:
        fh.seek(0, os.SEEK_END)
        pos = fh.tell()
        data = b""
        while pos > 0 and data.count(b"\n") <= lines and len(data) < max_bytes:
            step = min(65536, pos)
            pos -= step
            fh.seek(pos)
            data = fh.read(step) + data
    text = data.decode("utf-8", errors="replace").splitlines()
    return text[-lines:], pos > 0


def classify(source: str, line: str) -> str:
    """Map a log line to a severity for colouring."""
    if source == "ops":
        return "error" if ("result=FAILED" in line or "error=" in line) else "info"
    if " ERROR " in line or " CRITICAL " in line:
        return "error"
    if " WARNING " in line:
        return "warn"
    return "info"


def read_log(source: str, app_cfg, lines: int = 200, query: str | None = None) -> dict:
    path = resolve_path(source, app_cfg)
    raw, truncated = tail(path, lines)
    if query:
        q = query.casefold()
        raw = [ln for ln in raw if q in ln.casefold()]
    return {
        "source": source,
        "truncated": truncated,
        "lines": [{"text": ln, "level": classify(source, ln)} for ln in raw],
    }