A small, test-driven Python 3 tool that parses a security log, detects suspicious activity, and prints a clear report. The code is performance-oriented (streaming, O(1) per event updates) and designed to work in real time. A live “tail-like” pipeline is work in progress (see Real-time mode below).
-
Robust parsing of the provided sample log format:
[Timestamp] [Severity] [Source IP] [Event Type] [Message/Details] -
Detectors (incremental / streaming-safe):
- BruteForce — repeated
FAILED_LOGINfrom the same IP within a time window. - PortScan — multiple unique destination ports from one IP within a time window.
- SQLi — direct
SQL_INJECTION_ATTEMPTor message matching common SQL injection payloads. - UnusualAccess — direct
UNUSUAL_ACCESSor access to sensitive paths (e.g.,/etc/passwd,/var/log/auth.log).
- BruteForce — repeated
-
CLI with sane defaults, optional JSON output, file export, and CI-friendly exit code.
- Streaming design: the file is read line-by-line; entries are parsed lazily (no full in-memory load).
- O(1) per event: detectors use
dequesliding windows anddefaultdict/Counterto update state in constant time. - Minimal allocations: compiled regexes, normalized datatypes, and short-lived temporaries.
- Separation of concerns: parsing/IO in
analyzer.py, detection logic indetectors.py, UX/CLI incli.py.
.
├── analyzer.py # parsing, iterators, run(); (live run_streaming skeleton)
├── detectors.py # Detector base + BruteForce / PortScan / SQLi / UnusualAccess
├── cli.py # CLI entry point (argparse)
├── sample_security.log
└── tests/ # pytest suite (unit + integration/e2e)
- Python 3.11+ (tested on 3.12)
- Dev deps for tests:
pytest,pytest-mock
Install dev deps:
pip install -U pytest pytest-mock
Analyze the provided sample log:
python cli.py sample_security.log --window 60
Example output:
[OK] Opened: .../sample_security.log | lines: 48 | window: 60s
[INFO] Incidents: 11
- BruteForce:RepeatedFailedLogins | src=203.0.113.5 | 2025-07-03 10:00:05 — 2025-07-03 10:00:07 | attempts=3
- BruteForce:RepeatedFailedLogins | src=203.0.113.8 | 2025-07-03 10:00:33 — 2025-07-03 10:00:35 | attempts=3
- BruteForce:RepeatedFailedLogins | src=203.0.113.9 | 2025-07-03 10:00:46 — 2025-07-03 10:00:48 | attempts=3
- PORT_SCAN_ATTEMPT | src=172.16.0.10 | 2025-07-03 10:00:18 — 2025-07-03 10:00:20 | ports=[22, 80, 443]
- PORT_SCAN_ATTEMPT | src=172.16.0.13 | 2025-07-03 10:00:43 — 2025-07-03 10:00:45 | ports=[20, 21, 25]
- Injection:SQLi | src=10.0.0.22 | at=2025-07-03 10:00:10 | user_input=' OR 1=1--
- Injection:SQLi | src=10.0.0.24 | at=2025-07-03 10:00:27 | user_input=' UNION SELECT NULL,NULL--
- Injection:SQLi | src=10.0.0.26 | at=2025-07-03 10:00:41 | user_input=' SELECT * FROM users WHERE 1=1;--
- Unusual:SensitivePath | src=10.0.0.23 | at=2025-07-03 10:00:16 | /etc/passwd
- Unusual:SensitivePath | src=10.0.0.25 | at=2025-07-03 10:00:32 | /var/log/auth.log
- Unusual:SensitivePath | src=10.0.0.27 | at=2025-07-03 10:00:42 | /root/.bashrc
python cli.py <logfile> [--window 60] [--format text|json] [--out report.json] [--fail-on-find]
--window— sliding window (seconds) for windowed detectors; default 60.--format json— print JSON report (otherwise a pretty text report).--out <file>— write report to a file (works with both text/JSON).--fail-on-find— exit code 3 if any incidents were detected (useful for CI).
Tip: the text output can be filtered with standard tools (findstr/grep).
The detectors and iterators are real-time ready (they process each event as it arrives).
An optional run_streaming(...) pipeline (tail-like) is scaffolded in analyzer.py and will:
- follow a growing log file,
- feed new lines to detectors immediately,
- print/emit incidents as soon as they occur.
This live mode is intentionally not wired into the CLI yet (to keep the interface minimal for the assignment). The codebase already contains the necessary building blocks (lazy iterators, O(1) window maintenance, debouncing).
Run the full test suite:
pytest -q
The suite covers:
- parsing and line iterators,
- each detector (unit tests),
- CLI behavior,
- an end-to-end test over the entire sample log (expects 11 incidents grouped as: 3 brute-force, 2 port-scans, 3 SQLi, 3 unusual access).
-
Incremental detection: each detector exposes
feed(LogEntry)and keeps a small per-IP state:- BruteForce →
deque[datetime], - PortScan →
deque[(datetime, port)]+Counter[port], - SQLi/Unusual → pattern checks with compiled regexes.
- BruteForce →
-
Debounce: after emitting an incident for a burst/cluster, per-IP state is cleared to avoid duplicates.
-
Type-safety: dataclasses for parsed entries; forward-ref typing avoids cyclic imports.
-
Extensibility: add new detectors by subclassing
Detectorand implementingfeed().