jesse12-21/threat-intel-enricher

## 📋 Project Overview A modern Security Operations Center doesn't just generate alerts — it **enriches** them. Raw IDS alerts tell you *what* happened, but analysts need context: Is this IP known-malicious? Has this domain been seen in other campaigns? How does this alert relate to others in the same time window? **Threat intelligence enrichment** answers these questions, and automation is the only way to do it at scale. This project is the automation layer that sits on top of the [Suricata IDS Rules project](https://github.com/jesse12-21/suricata-ids-rules) — taking raw EVE JSON alerts, extracting IOCs, querying multiple threat intelligence sources concurrently, scoring risk, and producing the kind of enriched incident reports that analysts actually use. It demonstrates production Python patterns employers look for: async concurrency, Pydantic data validation, abstract base classes for extensibility, comprehensive CLI design with Rich terminal output, and test coverage. ### What This Project Covers | Section | Skill Demonstrated | Technologies | |---|---|---| | **Setup & Project Structure** | Modern Python packaging, virtual environments, secrets management | `pyproject.toml`, `venv`, `python-dotenv` | | **Architecture & Design** | Object-oriented design, abstract base classes, separation of concerns | ABCs, Pydantic models, strategy pattern | | **Alert Ingestion** | Parsing JSON streams, data normalization, IOC extraction | `json`, regex, Pydantic | | **Concurrent Enrichment** | Async I/O, rate limiting, API integration, error handling | `asyncio`, `aiohttp`, retry logic | | **Risk Scoring** | Weighted scoring algorithms, confidence calculation | NumPy-style logic, custom algorithms | | **Report Generation** | Template-driven output, multi-format reporting | Jinja2, Markdown, HTML | ### Sample Run Results Running the full pipeline against the included `examples/sample_suricata_eve.json`: | Metric | Value | |---|---| | Alerts ingested | **10** | | IOCs extracted → deduplicated | **18 → 12 unique** | | Active threat intel sources | **2** (AbuseIPDB, OTX) | | Enrichment results collected | **16** across 12 IOCs | | Actionable (score ≥ 80) findings | **1** — `185.220.101.45` scored **CRITICAL / 100** (Tor exit node, flagged by multiple sources) | | Attack campaigns correlated | **2** — `40.80.148.42` (5 alerts, 115 s, 5 distinct signatures) and `10.0.2.15` (3 alerts, 149 s) | | Total wall-clock time | **~37 s** (concurrent async enrichment) | | Test suite | **29 pytest tests, 0.09 s** | ## 🏗️ Architecture The toolkit implements a classic ETL pipeline pattern: **Extract → Enrich → Load** — specialized for security operations. ### Data Flow +--------------------------------------------------------------------+ | Threat Intel Enrichment Pipeline | | | | +-------------------+ +----------------------+ | | | Suricata Alerts | | IOC Extractor | | | | (EVE JSON) | ---> | (IPs, domains, | | | | | | hashes, URLs) | | | +-------------------+ +----------+-----------+ | | | | | v | | +----------------------------------+ | | | Concurrent Enricher (async) | | | | | | | | +------------+ +-------------+ | | | | | AbuseIPDB | | URLhaus | | | | | +------------+ +-------------+ | | | | +------------+ | | | | | AlienVault | ... more | | | | | OTX | | | | | +------------+ | | | +-----------------+----------------+ | | | | | v | | +----------------------------------+ | | | Risk Scoring Engine | | | | (weighted multi-source) | | | +-----------------+----------------+ | | | | | v | | +----------------------------------+ | | | Alert Correlator | | | | (group by campaign/attacker) | | | +-----------------+----------------+ | | | | | v | | +----------------------------------+ | | | Report Generator | | | | +--------+ +-------+ +------+ | | | | | JSON | | MD | | HTML | | | | | +--------+ +-------+ +------+ | | | +----------------------------------+ | +--------------------------------------------------------------------+ ### 📂 Repository Structure

Text version (click to expand)

threat-intel-enricher/ ├── src/enricher/ │ ├── cli.py # Click CLI with Rich output │ ├── config.py # Settings and secrets │ ├── models.py # Pydantic models (IOC, Alert, RiskScore) │ ├── pipeline.py # Async orchestration, dedup, fan-out │ ├── scoring.py # Weighted multi-source risk scoring │ ├── correlator.py # Attack campaign clustering │ ├── enrichers/ │ │ ├── base.py # BaseEnricher ABC │ │ ├── abuseipdb.py # IP reputation │ │ ├── otx.py # Community pulses │ │ ├── urlhaus.py # Malicious URLs (Auth-Key required) │ │ ├── cisa_kev.py # NEW — confirmed exploited CVEs │ │ └── epss.py # NEW — exploitation probability │ ├── ingesters/ │ │ ├── base.py # BaseIngester ABC │ │ ├── suricata.py # Suricata EVE JSON │ │ └── nmap_cim.py # NEW — Nmap CIM JSON scan output │ └── reporters/ │ ├── base.py # BaseReporter ABC │ ├── markdown_reporter.py # Analyst-readable Markdown │ ├── html_reporter.py # Styled HTML report │ └── stix_reporter.py # NEW — STIX 2.1 bundles ├── tests/ # 102 tests, 89% coverage │ ├── conftest.py # NEW — FakeSession aiohttp mock │ ├── test_enrichers.py # NEW — incl. both bug regressions │ ├── test_pipeline_ingest_report.py # NEW — pipeline, ingesters, STIX │ ├── test_cli_reporters.py # NEW — CLI and reporters │ ├── test_models.py │ ├── test_scoring.py │ └── test_correlator.py ├── .github/workflows/validate.yml # NEW — CI with 85% coverage gate ├── examples/sample_suricata_eve.json ├── docs/known-limitations.md # NEW — tested findings and gaps ├── .env.example └── pyproject.toml

## Part 1 - Installation & Project Setup ### Prerequisites - **Python 3.12** (for modern async syntax and type hints — tested on 3.12.3) - **Git** - API keys (free tiers work): - [AbuseIPDB](https://www.abuseipdb.com/api) — IP reputation - [AlienVault OTX](https://otx.alienvault.com/api) — Community threat intel URLhaus does not require an API key. ### Setup # Clone the repository git clone https://github.com/YOUR_USERNAME/threat-intel-enricher.git cd threat-intel-enricher # Create and activate a virtual environment python3 -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # Install dependencies pip install -r requirements.txt # Install the package in editable mode pip install -e . ### Configuration The toolkit uses a `.env` file for secrets. Copy the template and add your API keys: cp .env.example .env nano .env `.env` contents: # AbuseIPDB API key (https://www.abuseipdb.com/account/api) ABUSEIPDB_API_KEY=your_key_here # AlienVault OTX API key (https://otx.alienvault.com/api) OTX_API_KEY=your_key_here # Optional: rate limiting ENRICHER_MAX_CONCURRENT_REQUESTS=10 ENRICHER_REQUEST_TIMEOUT=30
### Verifying the Installation enricher --help
## Part 2 - Architecture & Design Patterns ### Abstract Base Classes for Extensibility The toolkit uses Python's **Abstract Base Class (ABC)** pattern to make every pipeline stage pluggable. Want to add a new threat intel source? Implement `BaseEnricher`. Want to support a new input format? Implement `BaseIngester`. This is the same pattern used by production SOAR platforms. # src/enricher/enrichers/base.py from abc import ABC, abstractmethod from typing import Optional from enricher.models import IOC, EnrichmentResult class BaseEnricher(ABC): """Abstract base class for threat intelligence enrichers. All enrichers must implement enrich() to query their backing service and return a standardized EnrichmentResult. """ name: str supported_ioc_types: list[str] @abstractmethod async def enrich(self, ioc: IOC) -> Optional[EnrichmentResult]: """Query this source for information about the given IOC. Returns None if the IOC type is unsupported or no data found. Raises EnrichmentError on unrecoverable API failures. """ ... def supports(self, ioc: IOC) -> bool: """Check if this enricher can process the given IOC type.""" return ioc.type in self.supported_ioc_types ### Pydantic Data Models All data flowing through the pipeline is validated with Pydantic v2 — catching schema errors at the boundary rather than deep in processing logic: # src/enricher/models.py """Pydantic data models for the threat intelligence enrichment pipeline. All data flowing through the pipeline is validated with Pydantic, catching schema errors at boundaries rather than deep in processing logic. """ from __future__ import annotations from datetime import datetime from enum import Enum from typing import Any from pydantic import BaseModel, ConfigDict, Field class IOCType(str, Enum): """Types of Indicators of Compromise the toolkit processes.""" IP = "ip" DOMAIN = "domain" URL = "url" HASH_MD5 = "md5" HASH_SHA256 = "sha256" class Severity(str, Enum): """Risk severity bands for scored IOCs.""" CRITICAL = "critical" HIGH = "high" MEDIUM = "medium" LOW = "low" INFO = "info" class IOC(BaseModel): """An Indicator of Compromise extracted from an alert.""" model_config = ConfigDict(frozen=True) value: str = Field(..., min_length=1, description="The IOC value (IP, domain, hash, etc.)") type: IOCType first_seen: datetime source_alert_id: str class EnrichmentResult(BaseModel): """Standardized output from any threat intel source.""" ioc: IOC source: str malicious: bool confidence: float = Field(ge=0.0, le=1.0) categories: list[str] = [] last_seen: datetime | None = None reference_url: str | None = None raw_response: dict[str, Any] = {} class RiskScore(BaseModel): """Aggregated risk score for an IOC across all sources.""" ioc: IOC score: int = Field(ge=0, le=100) severity: Severity sources_reporting: int enrichments: list[EnrichmentResult] ### Async Concurrency for API Calls Querying 5 threat intel APIs sequentially takes 5+ seconds per IOC. Concurrent async execution reduces this to the time of the slowest single API — typically under a second: # Simplified from src/enricher/pipeline.py async def enrich_all(ioc: IOC, enrichers: list[BaseEnricher]): """Query all applicable enrichers concurrently.""" tasks = [ enricher.enrich(ioc) for enricher in enrichers if enricher.supports(ioc) ] results = await asyncio.gather(*tasks, return_exceptions=True) return [r for r in results if isinstance(r, EnrichmentResult)]
## Part 3 - Alert Ingestion & IOC Extraction ### Parsing Suricata EVE JSON Suricata's EVE JSON format is line-delimited JSON — each line is a separate event. The ingester streams the file, filters for alert events, and extracts indicators: # src/enricher/ingesters/suricata.py from pathlib import Path from typing import Iterator from enricher.models import Alert, IOC, IOCType class SuricataIngester(BaseIngester): name = "suricata_eve" def ingest(self, source: Path) -> Iterator[Alert]: """Stream alerts from a Suricata EVE JSON file.""" with source.open() as f: for line_num, line in enumerate(f, 1): try: event = json.loads(line) if event.get("event_type") != "alert": continue yield self._parse_alert(event) except json.JSONDecodeError as e: logger.warning(f"Skipping malformed line {line_num}: {e}") def extract_iocs(self, alert: Alert) -> list[IOC]: """Extract IP, domain, and URL IOCs from an alert.""" iocs = [] # Source IP is always present iocs.append(IOC( value=alert.src_ip, type=IOCType.IP, first_seen=alert.timestamp, source_alert_id=alert.id )) # DNS queries contain domain IOCs if alert.dns and alert.dns.rrname: iocs.append(IOC( value=alert.dns.rrname, type=IOCType.DOMAIN, first_seen=alert.timestamp, source_alert_id=alert.id )) # HTTP traffic contains URL IOCs if alert.http and alert.http.hostname: iocs.append(IOC( value=f"http://{alert.http.hostname}{alert.http.url}", type=IOCType.URL, first_seen=alert.timestamp, source_alert_id=alert.id )) return iocs ### Running the Ingester enricher ingest --source examples/sample_suricata_eve.json --output output/iocs.json
## Part 4 - Multi-Source Threat Intelligence Enrichment ### Concurrent API Integration Each threat intel source is implemented as a `BaseEnricher` subclass. The pipeline orchestrates them concurrently with `asyncio.gather()`, handling rate limits and retries transparently. **AbuseIPDB — IP reputation:** # src/enricher/enrichers/abuseipdb.py class AbuseIPDBEnricher(BaseEnricher): name = "abuseipdb" supported_ioc_types = [IOCType.IP] def __init__(self, api_key: str, session: aiohttp.ClientSession): self.api_key = api_key self.session = session async def enrich(self, ioc: IOC) -> EnrichmentResult | None: async with self.session.get( "https://api.abuseipdb.com/api/v2/check", params={"ipAddress": ioc.value, "maxAgeInDays": 90}, headers={"Key": self.api_key, "Accept": "application/json"} ) as response: if response.status == 429: raise RateLimitError("AbuseIPDB rate limit hit") data = await response.json() confidence_score = data["data"]["abuseConfidenceScore"] return EnrichmentResult( ioc=ioc, source=self.name, malicious=confidence_score >= 25, confidence=confidence_score / 100.0, categories=data["data"].get("categoriesDescription", []), reference_url=f"https://www.abuseipdb.com/check/{ioc.value}", raw_response=data["data"] ) **URLhaus — malicious URL database (no API key required):** # src/enricher/enrichers/urlhaus.py class URLhausEnricher(BaseEnricher): name = "urlhaus" supported_ioc_types = [IOCType.URL, IOCType.DOMAIN] async def enrich(self, ioc: IOC) -> EnrichmentResult | None: endpoint = "host" if ioc.type == IOCType.DOMAIN else "url" async with self.session.post( f"https://urlhaus-api.abuse.ch/v1/{endpoint}/", data={endpoint: ioc.value} ) as response: data = await response.json() if data["query_status"] != "ok": return None return EnrichmentResult( ioc=ioc, source=self.name, malicious=True, # URLhaus only lists confirmed malicious confidence=1.0, categories=data.get("tags", []), last_seen=parse_datetime(data.get("date_added")), reference_url=data.get("urlhaus_reference"), raw_response=data ) ### Running Enrichment The full pipeline handles ingestion, enrichment, scoring, correlation, and reporting in a single command: enricher pipeline \ --source examples/sample_suricata_eve.json \ --format both \ --output output/incident
### Supported Threat Intel Sources | Source | IOC Types | API Key Required | Rate Limit (Free) | |---|---|---|---| | **AbuseIPDB** | IPv4 | Yes | 1,000/day | | **URLhaus** (abuse.ch) | URLs, domains | **Yes** — mandatory since 2025-06-30 | Fair use | | **AlienVault OTX** | IPs, domains, URLs, hashes | Yes | 10,000/hour | | **CISA KEV** | CVEs | No | None (public JSON catalog) | | **EPSS** (FIRST) | CVEs | No | None documented | | **Extensible** | Add your own | — | Implement `BaseEnricher` | ## Part 5 - Risk Scoring & Alert Correlation ### Weighted Multi-Source Risk Scoring A single source saying "this IP is malicious" is useful. **Three sources** saying so is far more compelling. The scoring engine aggregates enrichment results into a single risk score using weighted confidence: # src/enricher/scoring.py SOURCE_WEIGHTS = { "abuseipdb": 1.0, # Community-reported, high volume "urlhaus": 1.2, # Curated, high precision "otx": 1.0, # Community pulses "virustotal": 1.3, # Aggregator of aggregators } def score_ioc(enrichments: list[EnrichmentResult]) -> RiskScore: """Calculate weighted risk score across all enrichment sources. Scores range 0-100. Severity bands: - critical: 80-100 (multiple sources, high confidence) - high: 60-79 (strong indicators) - medium: 40-59 (some indicators) - low: 20-39 (weak signals) - info: 0-19 (clean or minimal data) """ if not enrichments: return RiskScore(score=0, severity="info", ...) weighted_sum = 0.0 total_weight = 0.0 malicious_sources = 0 for e in enrichments: weight = SOURCE_WEIGHTS.get(e.source, 1.0) weighted_sum += e.confidence * weight * 100 total_weight += weight if e.malicious: malicious_sources += 1 base_score = weighted_sum / total_weight if total_weight > 0 else 0 # Corroboration bonus: multiple sources agreeing boosts confidence corroboration_multiplier = 1.0 + (0.15 * (malicious_sources - 1)) final_score = min(100, int(base_score * corroboration_multiplier)) return RiskScore( score=final_score, severity=_score_to_severity(final_score), sources_reporting=len(enrichments), enrichments=enrichments, ) ### Alert Correlation Individual alerts are noise. Correlated alerts tell a story. The correlator groups alerts by source IP within configurable time windows to surface **attack campaigns**: # src/enricher/correlator.py def correlate_by_attacker( alerts: list[Alert], time_window_minutes: int = 15 ) -> list[AttackCampaign]: """Group alerts into attack campaigns based on source IP and time proximity. An attack campaign is a cluster of alerts from the same source IP occurring within a time window — typical of automated scanning or staged attacks. """ by_src_ip = defaultdict(list) for alert in sorted(alerts, key=lambda a: a.timestamp): by_src_ip[alert.src_ip].append(alert) campaigns = [] for src_ip, ip_alerts in by_src_ip.items(): clusters = _time_cluster(ip_alerts, time_window_minutes) for cluster in clusters: if len(cluster) < 2: # Skip isolated alerts continue campaigns.append(AttackCampaign( attacker_ip=src_ip, start_time=cluster[0].timestamp, end_time=cluster[-1].timestamp, alert_count=len(cluster), signatures=list({a.signature for a in cluster}), alerts=cluster, )) return sorted(campaigns, key=lambda c: c.alert_count, reverse=True)
## Part 6 - Report Generation & SIEM Integration ### Jinja2-Powered Report Templates Reports are generated from Jinja2 templates, keeping presentation logic separate from data processing. This makes it trivial to customize report formats or add new output types: # src/enricher/reporters/markdown_reporter.py class MarkdownReporter(BaseReporter): name = "markdown" def __init__(self): self.env = Environment( loader=PackageLoader("enricher.reporters", "templates"), autoescape=False, ) def render(self, campaigns: list[AttackCampaign], iocs: list[RiskScore]) -> str: template = self.env.get_template("incident_report.md.j2") return template.render( generated_at=datetime.utcnow(), campaigns=campaigns, top_iocs=sorted(iocs, key=lambda i: i.score, reverse=True)[:20], critical_count=sum(1 for i in iocs if i.severity == "critical"), stats=_calculate_stats(campaigns, iocs), ) ### Generating Reports Reports are emitted automatically at the end of the `pipeline` run. The `--format` flag accepts `markdown`, `html`, or `both`, and `--output` takes a path stem that both extensions are appended to: # Both formats in one run (produces output/incident.md and output/incident.html) enricher pipeline --source examples/sample_suricata_eve.json --format both --output output/incident # HTML only enricher pipeline --source examples/sample_suricata_eve.json --format html --output output/incident

### Full Pipeline in One Command Everything composed end-to-end. `--format both` emits Markdown and HTML side by side so analysts can triage in the browser while SOAR platforms ingest the Markdown: enricher pipeline \ --source /var/log/suricata/eve.json \ --format both \ --output "incident_$(date +%Y%m%d)" ### SIEM Integration The JSON output format is designed for SIEM ingestion. The same pipeline that produces analyst reports can push enriched events back into Splunk for correlation with other logs: # Enriched JSON is compatible with Splunk HTTP Event Collector curl -k https://splunk.local:8088/services/collector/event \ -H "Authorization: Splunk YOUR_HEC_TOKEN" \ -d @enriched.json This closes the loop with the [Splunk SIEM Analysis project](https://github.com/jesse12-21/splunk-siem-analysis) — Suricata detects, this toolkit enriches, Splunk correlates. ## Part 7 - Vulnerability Intelligence: KEV and EPSS Parts 4 and 5 enrich network indicators — IPs, domains, URLs, hashes. This section adds a different kind of indicator: the CVE. The distinction matters because reputation and vulnerability intelligence answer different questions. Reputation tells you whether an address has behaved badly. Vulnerability intelligence tells you whether the software you are running is being attacked. ### Three questions, three sources | Source | Question | Nature | |---|---|---| | **CVSS** | How bad would it be? | Severity | | **CISA KEV** | Has it actually happened? | Confirmed fact | | **EPSS** | Will it happen soon? | Forecast | Ranking by CVSS alone treats a 9.8 with an EPSS score of 0.0004 the same as a 7.5 with an EPSS of 0.87. The second is far more likely to be used against you this month. Neither is visible from severity. ### `cisa_kev.py` The Known Exploited Vulnerabilities catalog is CISA's authoritative list of vulnerabilities with confirmed in-the-wild exploitation. Public JSON, no key, no rate limit. name = "cisa_kev" supported_ioc_types: ClassVar[list[IOCType]] = [IOCType.CVE] Two implementation points worth explaining: **The catalog is fetched once, not per indicator.** It is roughly 1,500 entries, and the pipeline enriches concurrently — so without a guard, a batch of CVEs would each trigger their own download of the same document. An `asyncio.Lock` guards the cached load. `test_kev_catalog_fetched_once_across_many_cves` asserts a single HTTP call across five indicators. **Confidence is 1.0, not scaled.** KEV membership is binary and authoritative: a government body has confirmed exploitation. There is no partial credit to express. It carries the highest source weight in scoring (1.5) for the same reason. Absence from KEV is explicitly *not* treated as evidence of safety — the catalog records confirmed exploitation, not all exploitable vulnerabilities. ### `epss.py` The Exploit Prediction Scoring System, maintained by FIRST, estimates the probability a vulnerability will be exploited within 30 days. Public API, no key. EPSS probabilities are heavily skewed — the median CVE scores well under 0.01. The 0.10 threshold used here places a vulnerability in roughly the top few percent by predicted exploitation, which is where it stops being a backlog item. The score maps directly onto `confidence`, since both are 0–1 estimates of the same underlying question. EPSS is weighted at 1.1 — above pure community reporting, below confirmed-fact sources, because it is a forecast. A miss is treated as unknown rather than safe: an unscored CVE is usually too new to have been scored. ## Part 8 - STIX 2.1 Export Markdown and HTML reports are for humans. Without a machine-readable format, enrichment output is a dead end — an analyst reads the report and retypes indicators somewhere else. **STIX 2.1** is the OASIS standard for threat intelligence interchange and what MISP, OpenCTI, Anomali, and commercial TIPs consume. [`reporters/stix_reporter.py`](src/enricher/reporters/stix_reporter.py) emits bundles containing: | Object | Purpose | |---|---| | `identity` | The producing tool, with a stable ID across runs | | `indicator` | One per scored IOC, with a STIX pattern and confidence | | `note` | The enrichment evidence behind each score | { "type": "indicator", "spec_version": "2.1", "pattern": "[ipv4-addr:value = '45.227.255.206']", "pattern_type": "stix", "confidence": 95, "labels": ["botnet", "ssh-bruteforce"] } ### Design decisions worth stating **Only actionable indicators are exported by default.** A bundle full of INFO-severity indicators dilutes a TIP feed and trains analysts to ignore it. `actionable_only=True` filters to medium and above. **The identity ID is deterministic.** STIX consumers deduplicate on ID, so a fresh UUID each run would create a new "producer" in the TIP on every execution. A UUID5 over a fixed namespace keeps it stable — asserted by `test_stix_identity_id_is_stable_across_runs`. **Evidence travels with the indicator.** A bare indicator carrying a score is unauditable. The `note` object records which source said what, so a downstream analyst can judge the verdict rather than inherit it. **Campaigns are deliberately not serialised.** STIX models a campaign as an SDO carrying attribution semantics this pipeline does not establish. Emitting under-evidenced campaign objects into a shared TIP is worse than omitting them. Scope, stated plainly: this emits STIX and does not consume it, and does not implement TAXII. Bundles can be pushed to an existing TAXII endpoint or imported directly. See [`docs/known-limitations.md`](docs/known-limitations.md). ## Part 9 - Pipeline Integration This is the enrichment layer of a five-repository pipeline. It now ingests from two sources. suricata-ids-rules nmap-network-recon EVE JSON alerts CIM JSON scan output │ │ ▼ ▼ threat-intel-enricher enrich → score → STIX 2.1 bundle │ ▼ splunk-siem-analysis ### Suricata EVE JSON The original ingester. Alerts carry source and destination IPs, DNS queries, and HTTP hostnames — all enrichable against reputation sources. ### Nmap CIM JSON — new [`ingesters/nmap_cim.py`](src/enricher/ingesters/nmap_cim.py) consumes the output of the recon project's parser: python3 parsers/nmap_to_siem.py scan.xml --format json > scan.json enricher pipeline --source scan.json **A port scan is not an alert.** Nothing has attacked anything; what a scan produces is *attack surface*. Feeding that surface through the same pipeline lets one question be asked across both inputs: which of the things I can see are things an attacker is currently exploiting? That question is answered by the CVE enrichers rather than the reputation ones, which is why this ingester extracts two indicator types — the discovered host's IP, and any CVE identifiers embedded in service version strings by Nmap's vulners script. Scan findings are mapped to `severity=3` deliberately. Raising it would let attack surface outrank actual IDS alerts in scoring. **One subtlety worth knowing:** Python's `ipaddress` module classifies RFC 5737 documentation ranges as private, so `198.51.100.0/24` and friends are correctly skipped as non-routable. Sample data using those ranges produces no IP indicators, which is surprising until you know it. Asserted in `test_documentation_ranges_are_not_enrichable`. ## Part 10 - Correctness and CI The toolchain was configured from the start — ruff, mypy strict, pytest with coverage — and nothing enforced it. That gap is what this section closes, and what it found is instructive. ### What running the toolchain revealed | Check | Before | After | |---|---|---| | `pytest` | 29 passed | **102 passed** | | `ruff check` | **47 errors** | clean | | `mypy --strict` | clean | clean (23 files) | | Coverage | **23%** | **89%** | mypy strict passing across the whole package was already true and is genuinely uncommon. The rest was aspiration. The coverage *distribution* mattered more than the number: models.py 99% correlator.py 97% scoring.py 91% everything else 0% The three tested modules were the three pure-logic ones. Every module touching I/O — all enrichers, the ingester, the pipeline, both reporters, the CLI, config — had no tests. That is the ordinary shape of a suite written without HTTP mocking, and it is exactly where the bugs were. ### Two bugs the tests found **The retry decorator never retried.** Every enricher carried `@retry(stop_after_attempt(3))` on `enrich()` while `enrich()` caught `aiohttp.ClientError` internally. Tenacity only retries on exceptions that *escape* the decorated function, so the exception was swallowed before it could be observed. Measured: before: HTTP attempts made : 1 after: HTTP attempts made : 3 A transient blip produced a permanent failure result, and the backoff never ran. Fixed by splitting the request into a `_fetch()` that carries the decorator and lets errors propagate. **The URLhaus enricher had been returning 401 for over a year.** abuse.ch made the `Auth-Key` header mandatory on 30 June 2025. The enricher sent no headers at all — and `.env.example`, `config.py`, and this README's source table all asserted no key was needed. One of three sources was silently dead while every artifact in the repository said otherwise. The general lesson: **a third-party API is a dependency that changes without a version bump.** ### What CI enforces [`.github/workflows/validate.yml`](.github/workflows/validate.yml): | Job | Checks | |---|---| | **test** | pytest across Python 3.11 and 3.12, with an 85% coverage gate | | **lint** | `ruff check`, `mypy --strict`, and a placeholder scan | | **contract** | Every enricher, ingester, and reporter implements its base class; CLI is invocable | The contract job exists because a plugin architecture only holds if components actually conform. During this work my first STIX reporter used `generate(scores, path)` while `BaseReporter` requires `render(scores, campaigns) -> str` — it type-checked in isolation and would have broken the CLI's format selection. ### The gap that remains **Every enricher test mocks the HTTP layer.** The suite proves the code handles a given response shape, not that the API still returns it. The URLhaus break is precisely the case where mocks kept passing while production was broken. Closing it needs contract tests against live APIs on a schedule, which means credentials in CI — a deliberate omission for a public repository rather than an oversight, but a real gap. It is recorded in [`docs/known-limitations.md`](docs/known-limitations.md) alongside the rest. ## 🧪 Testing The project ships with a **29-test pytest suite** covering correlation logic, scoring algorithms, severity band boundaries, and Pydantic model validation: pytest -v
## 🔑 CLI Reference | Command | Purpose | |---|---| | `enricher ingest --source FILE --output FILE` | Parse Suricata EVE JSON and extract deduplicated IOCs to JSON | | `enricher correlate --source FILE` | Group alerts into attack campaigns by source IP and time window (no enrichment) | | `enricher pipeline --source FILE --format {markdown,html,both} --output STEM` | Full end-to-end run: ingest → enrich → score → correlate → report | | `enricher --help` | Show all commands and options | | `enricher --version` | Print the installed toolkit version | ## 🧰 Tools & Environment | Component | Purpose | |---|---| | **Python 3.12** | Modern async syntax, PEP 695 type parameters, improved error messages | | **Pydantic v2** | Data validation and settings management | | **aiohttp** | Async HTTP client for concurrent API calls | | **Click** | CLI framework with rich subcommand support | | **Rich** | Beautiful terminal output, progress bars, tables | | **Jinja2** | Template engine for report generation | | **pytest** | Test framework with async support | | **python-dotenv** | Environment-based secrets management | ## 📚 Summary This project demonstrates production Python skills through ten progressive sections: 1. **Setup & Project Structure** — Modern Python packaging with `pyproject.toml`, virtual environments, and `.env`-based secrets management 2. **Architecture & Design** — Clean separation of concerns using abstract base classes, Pydantic data models for boundary validation, and the strategy pattern for pluggable components 3. **Alert Ingestion** — Streaming JSON parser for Suricata EVE format with IOC extraction across IPs, domains, URLs, and hashes 4. **Concurrent Enrichment** — Async/await integration with AbuseIPDB, URLhaus, and AlienVault OTX APIs, with proper rate limiting, retry logic, and error handling 5. **Risk Scoring & Correlation** — Weighted multi-source scoring algorithm with corroboration bonuses, plus time-window-based alert correlation to surface attack campaigns 6. **Report Generation** — Jinja2-templated Markdown and HTML reports for analyst consumption, plus JSON output for SIEM integration 7. **Vulnerability Intelligence** — Added CISA KEV and FIRST EPSS enrichers covering CVE indicators, answering "is this confirmed exploited" and "how likely is it to be exploited soon" — questions CVSS severity cannot answer. The KEV catalog is fetched once behind an asyncio lock rather than per indicator 8. **STIX 2.1 Export** — Machine-readable bundles with stable identity IDs, escaped patterns, and enrichment evidence attached as notes, so output is importable into MISP, OpenCTI, or any TAXII-compatible platform instead of being a dead end 9. **Pipeline Integration** — Added an ingester for the recon project's CIM-normalised scan output, extracting both host IPs and CVE identifiers embedded in service version strings, so attack surface and IDS alerts flow through one enrichment path 10. **Correctness & CI** — Raised coverage from 23% to 89% across 102 tests, and enforced the previously-unenforced toolchain in CI. Writing the missing tests surfaced two shipped bugs: a retry decorator that never retried, and an enricher that had been returning 401 for over a year after abuse.ch made authentication mandatory ### Skills Demonstrated `Python 3.11/3.12` · `Async Programming` · `API Integration` · `Pydantic v2` · `Abstract Base Classes` · `mypy strict` · `pytest & Async Mocking` · `Test Coverage Engineering` · `CI/CD` · `STIX 2.1` · `Threat Intelligence` · `Vulnerability Management (KEV/EPSS)` · `SOC Automation` · `CLI Design` · `Secrets Management` · `Jinja2 Templating` ### Integration With Other Projects This toolkit is the automation layer connecting the detection and analysis projects: - **Input:** Suricata EVE JSON from the [Suricata IDS Rules](https://github.com/jesse12-21/suricata-ids-rules) project - **Output:** Enriched JSON compatible with the [Splunk SIEM Analysis](https://github.com/jesse12-21/splunk-siem-analysis) project - **Context:** Uses packet-level insights from the [Wireshark Threat Detection](https://github.com/jesse12-21/wireshark-threat-detection) project to inform IOC extraction - **Input:** CIM-normalised scan output from the [Nmap Network Recon](https://github.com/jesse12-21/nmap-network-recon) project, via `enricher pipeline --source scan.json` - **Output:** STIX 2.1 bundles for import into MISP, OpenCTI, or any TAXII-compatible threat intelligence platform

标签：IDS告警分诊, Pydantic, Python, SOC自动化, STIX, 威胁情报, 安全规则引擎, 开发者工具, 异步IO, 无后门, 计算机取证, 逆向工具