EventLens-Public/backend/app/main.py

import asyncio
import json
import logging
import os
import re
from pathlib import Path
from typing import Any

from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from starlette.requests import ClientDisconnect

from .correlate import correlate
from .demo import load_demo
from .models import AnalysisResponse, ParseIssue, ParseStats
from .parser import (
    MAX_UPLOAD_BYTES,
    EvtxParseError,
    UnexpectedParserError,
    parse_evtx_bytes,
)
from .version import VERSION, read_revision

logger = logging.getLogger(__name__)

app = FastAPI(title="EventLens API", version=VERSION)
cors_origins = [
    origin.strip()
    for origin in os.environ.get(
        "EVENTLENS_CORS_ORIGINS",
        "http://localhost:5173,http://127.0.0.1:5173,http://127.0.0.1:4173",
    ).split(",")
    if origin.strip()
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=cors_origins,
    allow_credentials=False,
    allow_methods=["GET", "POST"],
    allow_headers=["Content-Type"],
)

MAX_MULTIPART_OVERHEAD_BYTES = 1024 * 1024
MAX_REQUEST_BYTES = MAX_UPLOAD_BYTES + MAX_MULTIPART_OVERHEAD_BYTES
CONTENT_SECURITY_POLICY = (
    "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; "
    "img-src 'self' data:; connect-src 'self' http://localhost:8000 http://127.0.0.1:8000 "
    "http://localhost:5173 http://127.0.0.1:5173 http://127.0.0.1:4173; "
    "font-src 'self'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'"
)
SECURITY_HEADERS = {
    "X-Content-Type-Options": "nosniff",
    "Referrer-Policy": "no-referrer",
    "Permissions-Policy": "camera=(), microphone=(), geolocation=()",
    "Content-Security-Policy": CONTENT_SECURITY_POLICY,
    "X-Frame-Options": "DENY",
}


def _bounded_env_int(name: str, default: int, minimum: int, maximum: int) -> int:
    try:
        value = int(os.environ.get(name, str(default)))
    except (TypeError, ValueError):
        value = default
    return min(max(value, minimum), maximum)


PARSE_CONCURRENCY = _bounded_env_int("EVENTLENS_PARSE_CONCURRENCY", 1, 1, 2)
PARSE_QUEUE_LIMIT = _bounded_env_int("EVENTLENS_PARSE_QUEUE", 1, 0, 4)
PARSE_QUEUE_TIMEOUT_SECONDS = 0.25
_parse_slots = asyncio.Semaphore(PARSE_CONCURRENCY)
_parse_admission_lock = asyncio.Lock()
_parse_inflight = 0


class RequestBodyLimitMiddleware:
    """Bound request bytes before Starlette's multipart parser expands the upload."""

    def __init__(self, app: Any, max_bytes: int) -> None:
        self.app = app
        self.max_bytes = max_bytes

    async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None:
        if scope.get("type") != "http" or scope.get("path") != "/api/analyze" or scope.get("method") != "POST":
            await self.app(scope, receive, send)
            return

        headers = {key.lower(): value for key, value in scope.get("headers", [])}
        content_length = headers.get(b"content-length")
        if content_length is not None:
            try:
                declared = int(content_length)
            except (TypeError, ValueError):
                declared = 0
            if declared > self.max_bytes:
                await self._send_too_large(send)
                return

        consumed = 0
        too_large = False
        response_started = False

        async def limited_receive() -> dict[str, Any]:
            nonlocal consumed, too_large
            message = await receive()
            if message.get("type") == "http.request":
                consumed += len(message.get("body", b""))
                if consumed > self.max_bytes:
                    too_large = True
                    return {"type": "http.disconnect"}
            return message

        async def guarded_send(message: dict[str, Any]) -> None:
            nonlocal response_started
            if too_large:
                if message.get("type") == "http.response.start" and not response_started:
                    response_started = True
                    await self._send_too_large(send)
                return
            await send(message)

        try:
            await self.app(scope, limited_receive, guarded_send)
        except ClientDisconnect:
            if not too_large:
                raise
        if too_large and not response_started:
            await self._send_too_large(send)

    async def _send_too_large(self, send: Any) -> None:
        body = json.dumps(
            {"detail": {"code": "request_body_too_large", "message": "Upload request exceeds the bounded request limit."}}
        ).encode("utf-8")
        headers = [
            (b"content-type", b"application/json"),
            (b"content-length", str(len(body)).encode()),
            (b"cache-control", b"no-store"),
            (b"pragma", b"no-cache"),
        ]
        headers.extend((name.lower().encode("ascii"), value.encode("ascii")) for name, value in SECURITY_HEADERS.items())
        await send(
            {
                "type": "http.response.start",
                "status": 413,
                "headers": headers,
            }
        )
        await send({"type": "http.response.body", "body": body})


app.add_middleware(RequestBodyLimitMiddleware, max_bytes=MAX_REQUEST_BYTES)


@app.middleware("http")
async def response_security_headers(request, call_next):
    response = await call_next(request)
    for name, value in SECURITY_HEADERS.items():
        response.headers[name] = value
    if request.url.path.startswith("/api/") or request.url.path == "/health":
        response.headers["Cache-Control"] = "no-store"
        response.headers["Pragma"] = "no-cache"
    return response


@app.get("/health")
def health():
    return {
        "status": "ok",
        "evtx_ingestion": True,
        "version": VERSION,
        "revision": read_revision(),
    }


@app.get("/api/demo", response_model=AnalysisResponse)
def demo():
    events = load_demo()
    return AnalysisResponse(source="bundled-demo", events=events, incidents=correlate(events))


async def _parse_with_admission(data: bytes):
    global _parse_inflight
    async with _parse_admission_lock:
        if _parse_inflight >= PARSE_CONCURRENCY + PARSE_QUEUE_LIMIT:
            raise HTTPException(
                503,
                detail={
                    "code": "parse_capacity_exceeded",
                    "message": "EventLens is already analyzing its maximum concurrent uploads; retry shortly.",
                },
                headers={"Retry-After": "1"},
            )
        _parse_inflight += 1
    acquired = False
    try:
        try:
            await asyncio.wait_for(_parse_slots.acquire(), timeout=PARSE_QUEUE_TIMEOUT_SECONDS)
            acquired = True
        except TimeoutError as exc:
            raise HTTPException(
                503,
                detail={
                    "code": "parse_capacity_exceeded",
                    "message": "EventLens is busy analyzing another upload; retry shortly.",
                },
                headers={"Retry-After": "1"},
            ) from exc
    finally:
        if not acquired:
            async with _parse_admission_lock:
                _parse_inflight -= 1
    try:
        return await asyncio.to_thread(parse_evtx_bytes, data)
    finally:
        if acquired:
            _parse_slots.release()
        async with _parse_admission_lock:
            _parse_inflight -= 1


def _failed_parse_stats() -> ParseStats:
    return ParseStats(
        status="failed",
        parsed_events=0,
        records_seen=0,
        skipped_records=0,
        issue_count=1,
        truncated=False,
        truncation_reason=None,
        duration_ms=0,
        issues=[ParseIssue(code="parser_rejected_input", message="The EVTX parser rejected the input.")],
    )


@app.post("/api/analyze", response_model=AnalysisResponse)
async def analyze(file: UploadFile = File(...)):  # noqa: B008
    try:
        source = _safe_source_name(file.filename)
        if not source.lower().endswith(".evtx"):
            raise HTTPException(400, "Upload an .evtx file")
        data = bytearray()
        while True:
            chunk = await file.read(min(1024 * 1024, MAX_UPLOAD_BYTES + 1 - len(data)))
            if not chunk:
                break
            data.extend(chunk)
            if len(data) > MAX_UPLOAD_BYTES:
                raise HTTPException(413, "EVTX file exceeds the 50 MiB MVP limit")
        try:
            parsed = await _parse_with_admission(bytes(data))
        except EvtxParseError as exc:
            raise HTTPException(
                422,
                detail={
                    "code": "parser_rejected_input",
                    "message": str(exc),
                    "parse": _failed_parse_stats().model_dump(mode="json"),
                },
            ) from exc
        except UnexpectedParserError as exc:
            logger.error("Unexpected EVTX parser failure reached API exception_type=%s", type(exc).__name__)
            raise HTTPException(
                500,
                detail={"code": "parser_internal_error", "message": "EVTX analysis failed unexpectedly; retry the request."},
            ) from exc
        if not parsed.events:
            raise HTTPException(
                422,
                detail={
                    "code": "no_readable_events",
                    "message": "No readable Windows events were found in this file.",
                    "parse": ParseStats(
                        status="failed",
                        parsed_events=0,
                        records_seen=parsed.records_seen,
                        skipped_records=parsed.skipped_records,
                        issue_count=len(parsed.issues),
                        truncated=parsed.truncated,
                        truncation_reason=parsed.truncation_reason,
                        duration_ms=parsed.duration_ms,
                        issues=parsed.issues,
                    ).model_dump(mode="json"),
                },
            )
        return AnalysisResponse(
            source=source,
            events=parsed.events,
            incidents=correlate(parsed.events),
            parse=ParseStats(
                status=parsed.status,
                parsed_events=len(parsed.events),
                records_seen=parsed.records_seen,
                skipped_records=parsed.skipped_records,
                issue_count=len(parsed.issues),
                truncated=parsed.truncated,
                truncation_reason=parsed.truncation_reason,
                duration_ms=parsed.duration_ms,
                issues=parsed.issues,
            ),
        )
    finally:
        await file.close()


def _safe_source_name(filename: str | None) -> str:
    """Return a short display/export name without path or control characters."""
    candidate = re.split(r"[\\/]", filename or "")[-1]
    candidate = "".join(char for char in candidate if char.isprintable()).strip()
    candidate = re.sub(r"[^A-Za-z0-9._ -]", "_", candidate)
    candidate = candidate[:128]
    return candidate or "upload.evtx"


frontend_dir = os.environ.get("EVENTLENS_FRONTEND_DIR")
if frontend_dir:
    path = Path(frontend_dir)
    if path.is_dir():
        app.mount("/", StaticFiles(directory=str(path), html=True), name="frontend")