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")