Port ip-enrichment-worker and correlator-worker to the Rust tier (#1610) - #1615
Merged
Merged
Conversation
Moves the portbridge via_port -> real attacker IP join from dashboard read-time to ingest time. Unlike every other #1610 worker, this one never touches Elasticsearch or the network at all: it tails ~19 raw sensor JSON log files, rewrites any line whose src_ip is still the WireGuard tunnel peer address (10.8.0.1) to the real attacker IP (joined against portbridge's own connection log by source port), and promotes cross-sensor canonical_* fields (creds/command/fingerprint/ payload hash/ATT&CK technique). The rewritten stream is written to OUT_DIR for Filebeat to tail instead of the raw file -- this port doesn't touch Filebeat's own config, only keeps producing the same enriched-file format at the same path. New src/ip_enrichment/ module: mod.rs (source discovery/orchestration), tail.rs (offset-persisted incremental reads), rotate.rs (self-rotating output writer), pending.rs (retry queue for via_port misses), viamap.rs (the portbridge via_port map, including the incremental-refresh ViaMapBuilder a prior production incident (#1206) made necessary -- re-parsing both portbridge generations from scratch every 2s throttled the container's own CPU limit badly enough to push cowrie's join success rate near zero after ~2 days uptime), tftp.rs (the separate tftp-relay session join for dionaea's TFTP events), canonical.rs (all 13 per-persona field-promotion cases), attck.rs (the ATT&CK technique subset derivable from promoted canonical fields), sensors.rs (the generic src_ip join plus five bespoke per-sensor enrichers for beelzebub/hellpot/galah/sentrypeer/wordpot, each adapting a raw log shape the generic path can't parse). New backend-worker-enrichment compose service, network_mode: none (carried over unchanged from the Go worker's own compose block -- a deliberate security posture, not an oversight) -- distinct from backend-worker (whose other loops all need to reach Elasticsearch). 23 new unit tests, all passing (117/117 crate-wide). The old Go worker stays running unchanged -- parity port, not cutover. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
campaigns-v1 (CIDR-grouped) and attacker-clusters-v1 (fingerprint/payload/ ASN/provider-grouped) correlation, entirely via Elasticsearch-native aggregations -- no raw-document paging or PIT needed at all, the simplest of the six #1610 workers ported this pass. Two sibling ip_prefix aggregations (IPv4 /24, IPv6 /64) with a scripted cred_pairs terms sub-aggregation, plus four cluster terms aggregations gated on a real (not just min_doc_count) >=2-unique-IP threshold. Both indices are fully resynced each cycle: upsert this cycle's fresh docs, then delete everything else via a new Es::delete_by_query_except primitive (elasticsearch::DeleteByQueryParts + Conflicts::Proceed, verified against the vendored crate source). Runs as the `correlator` WORKER_LOOPS entry on backend-worker -- pure ES, no host mounts, no local state, same shape as its attacker-identity/agent-intrusion siblings. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.Scanned FilesNone |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Ports the two additional workers surfaced while surveying the codebase beyond #1610's originally-named inventory (
ip-enrichment-worker,correlator-worker) — the last two of six workers total, on top ofport-foundation(post-#1613/#1614).ip-enrichment-worker(Go, 2,174 lines — the largest single worker in this whole migration) — architecturally distinct from every other worker ported so far: it never touches Elasticsearch at all. It's a pure host-log-file rewriter that tails ~20 sensors' raw JSON logs, resolves the real attacker IP for connections that only ever see the WireGuard tunnel-peer address (a portbridgevia_portjoin, done at ingest time so every dashboard replica sees an already-correctsrc_ip), and promotes cross-sensorcanonical_*fields. Writes to a parallel output file Filebeat tails instead of the raw one. New, separate compose servicebackend-worker-enrichmentwithnetwork_mode: none— carried over unchanged from the original's own compose block, a deliberate security posture (this loop needs zero network access, incompatible withbackend-worker's other ES-dependent loops).correlator-worker(Go, 953 lines) — recomputescampaigns-v1(CIDR-grouped) andattacker-clusters-v1(fingerprint/payload/ASN/provider-grouped) from scratch every cycle, entirely via Elasticsearch-native aggregations (including a server-side Painless script for credential-pair grouping) — no raw-document paging needed. Added adelete_by_query_exceptprimitive toes.rsfor the stale-document sweep this worker's full-resync pattern needs. Added to the existingbackend-worker(pure ES, no host mounts).Both were caught by re-surveying every
honeypot-*worker/importer directory in the tree rather than assuming #1610's example list was exhaustive —ip-enrichment-workerin particular is foundational (it's what produces thecanonical_*fieldsattacker-identity-worker/agent-intrusion-worker/correlator-workerall read).This completes every worker discovered in this migration sweep. Not done (per #1610's own remaining text, out of scope for all of #1610's PRs): the BFF scalability hard requirements (Node cluster mode, backpressure, streaming, horizontal-readiness, load-test gate).
Test plan
cargo buildclean on every commit (no new warnings beyond one pre-existing unrelatedcanarytokens.rswarning)cargo test— 125/125 passing (31 new across these two workers' ported test suites)docker compose --profile next configresolves all new/changed services cleanlyelasticsearchcrate source and the existing Go source throughout🤖 Generated with Claude Code