import hashlib import io import json import logging import time from dataclasses import dataclass from evtx import PyEvtxParser from .models import NormalizedEvent, ParseIssue, ParseStatus from .normalize import normalize_record logger = logging.getLogger(__name__) MAX_EVENTS = 100_000 MAX_PARSE_SECONDS = 8.0 MAX_UPLOAD_BYTES = 50 * 1024 * 1024 MAX_RECORD_JSON_BYTES = 2 * 1024 * 1024 MAX_NORMALIZED_CHARS = 16 * 1024 * 1024 class EvtxParseError(ValueError): pass class UnexpectedParserError(RuntimeError): pass @dataclass class ParseResult: events: list[NormalizedEvent] issues: list[ParseIssue] truncated: bool duration_ms: int status: ParseStatus = "complete" truncation_reason: str | None = None records_seen: int = 0 skipped_records: int = 0 def _safe_record_id(value: object) -> int | None: try: parsed = int(value) if not isinstance(value, bool) else None except (TypeError, ValueError, OverflowError): return None return parsed if parsed is not None and 0 <= parsed <= 9_223_372_036_854_775_807 else None def _fingerprint(stage: str, exc: BaseException) -> str: return hashlib.sha256(f"{stage}:{type(exc).__name__}".encode("ascii", errors="replace")).hexdigest()[:12] def parse_evtx_bytes( data: bytes, max_events: int = MAX_EVENTS, max_seconds: float = MAX_PARSE_SECONDS, ) -> ParseResult: if len(data) > MAX_UPLOAD_BYTES: raise EvtxParseError("EVTX file exceeds the 50 MiB MVP limit") max_events = min(max(1, int(max_events)), MAX_EVENTS) max_seconds = min(max(0.1, float(max_seconds)), MAX_PARSE_SECONDS) started = time.monotonic() events: list[NormalizedEvent] = [] issues: list[ParseIssue] = [] truncated = False truncation_reason: str | None = None records_seen = 0 skipped_records = 0 normalized_chars = 0 seen_ids: set[str] = set() def add_issue( code: str, message: str, record_id: int | None = None, skipped_record: bool = False, ) -> None: nonlocal skipped_records issues.append(ParseIssue(code=code, record_id=record_id, message=message, skipped_record=skipped_record)) if skipped_record: skipped_records += 1 try: parser = PyEvtxParser(io.BytesIO(data), number_of_threads=1) for ordinal, record in enumerate(parser.records_json()): records_seen += 1 if len(events) >= max_events: truncated = True truncation_reason = "event_limit_reached" add_issue( "event_limit_reached", f"Event limit reached ({max_events}); this record and later records were not analyzed.", _safe_record_id(record.get("event_record_id")) if isinstance(record, dict) else None, skipped_record=True, ) break if time.monotonic() - started > max_seconds: truncated = True truncation_reason = "parse_time_budget_reached" add_issue( "parse_time_budget_reached", f"Parse time budget reached ({max_seconds:.1f}s); later records were not analyzed.", _safe_record_id(record.get("event_record_id")) if isinstance(record, dict) else None, skipped_record=True, ) break if not isinstance(record, dict): add_issue("malformed_record_skipped", "Skipped a malformed parser record.", skipped_record=True) continue record_id = _safe_record_id(record.get("event_record_id")) try: raw_data = record["data"] if not isinstance(raw_data, str): raise TypeError("record data was not textual") if ( len(raw_data) > MAX_RECORD_JSON_BYTES or len(raw_data.encode("utf-8")) > MAX_RECORD_JSON_BYTES ): add_issue( "record_json_limit_exceeded", "Skipped a parser record that exceeded the record JSON limit.", record_id, skipped_record=True, ) continue payload = json.loads(raw_data) normalized = normalize_record( payload, record_id=record_id, record_timestamp=record.get("timestamp"), record_ordinal=ordinal, ) normalized_size = len(normalized.provider) + len(normalized.channel) + len(normalized.computer) normalized_size += len(normalized.summary) normalized_size += sum(len(key) + len(value) for key, value in normalized.fields.items()) if normalized_chars + normalized_size > MAX_NORMALIZED_CHARS: truncated = True truncation_reason = "normalized_output_budget_reached" add_issue( "normalized_output_budget_reached", "Normalized output budget reached; this record and later records were not analyzed.", record_id, skipped_record=True, ) break if normalized.id in seen_ids: normalized.id = f"{normalized.id}-duplicate-{ordinal}"[:128] seen_ids.add(normalized.id) normalized_chars += normalized_size events.append(normalized) if normalized.timestamp_source in {"invalid", "missing"}: add_issue( "invalid_timestamp" if normalized.timestamp_source == "invalid" else "missing_timestamp", "Event timestamp is not reliable; the event remains evidence but is excluded from temporal correlation.", record_id, ) except (KeyError, TypeError, ValueError, UnicodeError, json.JSONDecodeError, RecursionError): add_issue("malformed_record_skipped", "Skipped a malformed record.", record_id, skipped_record=True) except (OSError, ValueError, TypeError, UnicodeError, RuntimeError, EOFError, RecursionError) as exc: if not events: logger.warning( "EVTX parser rejected input stage=%s exception_type=%s fingerprint=%s", "open_or_iterate", type(exc).__name__, _fingerprint("open_or_iterate", exc), ) raise EvtxParseError("The EVTX parser rejected the file or could not read its records.") from exc logger.warning( "EVTX parser recovered partially stage=%s exception_type=%s fingerprint=%s", "iterate", type(exc).__name__, _fingerprint("iterate", exc), ) add_issue("parser_partial_recovery", "Parser stopped after partial recovery.") truncated = True truncation_reason = "parser_partial_recovery" except Exception as exc: logger.error( "Unexpected EVTX parser failure stage=%s exception_type=%s fingerprint=%s", "iterate", type(exc).__name__, _fingerprint("iterate", exc), ) raise UnexpectedParserError("The EVTX parser encountered an unexpected internal failure.") from exc duration_ms = int((time.monotonic() - started) * 1000) # A warning about an unreliable timestamp does not mean that the record was # omitted. Only truncation/recovery or an actually skipped record makes # the analysis partial; the issue remains visible in the parse metadata. incomplete = truncated or skipped_records > 0 or any( issue.code == "parser_partial_recovery" for issue in issues ) status: ParseStatus = "partial" if incomplete else "complete" return ParseResult( events=events, issues=issues, truncated=truncated, duration_ms=duration_ms, status=status, truncation_reason=truncation_reason, records_seen=records_seen, skipped_records=skipped_records, )