89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from app import parser
|
|
from app.correlate import correlate
|
|
from app.models import NormalizedEvent
|
|
from app.normalize import normalize_record
|
|
from app.parser import EvtxParseError, parse_evtx_bytes
|
|
|
|
|
|
def test_corrupt_evtx_has_a_generic_error_without_parser_details():
|
|
with pytest.raises(EvtxParseError, match="rejected the file"):
|
|
parse_evtx_bytes(b"not-an-evtx")
|
|
|
|
|
|
def test_oversized_parser_record_is_skipped_before_json_expansion(monkeypatch):
|
|
class FakeParser:
|
|
def __init__(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
def records_json(self):
|
|
yield {"event_record_id": "9" * 5000, "data": "x" * (parser.MAX_RECORD_JSON_BYTES + 1)}
|
|
|
|
monkeypatch.setattr(parser, "PyEvtxParser", FakeParser)
|
|
result = parse_evtx_bytes(b"ElfFile\x00synthetic")
|
|
assert result.events == []
|
|
assert result.issues[0].code == "record_json_limit_exceeded"
|
|
assert result.issues[0].record_id is None
|
|
|
|
|
|
def test_normalization_bounds_nested_and_textual_values():
|
|
payload = {
|
|
"Event": {
|
|
"System": {
|
|
"Provider": {"#attributes": {"Name": "P" * 500}},
|
|
"EventID": 1,
|
|
"Channel": "C" * 500,
|
|
"Computer": "H" * 500,
|
|
"TimeCreated": {"#attributes": {"SystemTime": "2026-09-09T08:00:00Z"}},
|
|
},
|
|
"EventData": {"value": "V" * 5000},
|
|
}
|
|
}
|
|
event = normalize_record(payload, record_ordinal=1)
|
|
assert len(event.provider) == 256
|
|
assert len(event.channel) == 256
|
|
assert len(event.computer) == 256
|
|
assert len(event.fields["EventData.value"]) == 1000
|
|
|
|
|
|
def test_records_without_event_record_ids_get_distinct_deterministic_ids():
|
|
payload = {
|
|
"Event": {
|
|
"System": {
|
|
"Provider": "Synthetic",
|
|
"EventID": 1,
|
|
"TimeCreated": {"#attributes": {"SystemTime": "2026-09-09T08:00:00Z"}},
|
|
}
|
|
}
|
|
}
|
|
first = normalize_record(payload, record_timestamp="2026-09-09T08:00:00Z", record_ordinal=0)
|
|
second = normalize_record(payload, record_timestamp="2026-09-09T08:00:00Z", record_ordinal=1)
|
|
repeat = normalize_record(payload, record_timestamp="2026-09-09T08:00:00Z", record_ordinal=0)
|
|
assert first.id != second.id
|
|
assert first.id == repeat.id
|
|
|
|
|
|
def test_correlation_does_not_bridge_beyond_rule_window():
|
|
start = datetime(2026, 9, 9, 8, 0, tzinfo=UTC)
|
|
events = [
|
|
NormalizedEvent(
|
|
id=str(index),
|
|
timestamp=start + timedelta(seconds=offset),
|
|
provider="Microsoft-Windows-DNS-Client",
|
|
event_id=1014,
|
|
level="warning",
|
|
channel="System",
|
|
computer="LAB-PC",
|
|
category="network",
|
|
summary="DNS timeout",
|
|
)
|
|
for index, offset in enumerate((0, 119, 238))
|
|
]
|
|
incidents = correlate(events)
|
|
assert len(incidents) == 2
|
|
assert [incident.original_count for incident in incidents] == [2, 1]
|
|
assert all(incident.end - incident.start <= timedelta(seconds=120) for incident in incidents)
|