From 1a93d0085c08bca22d6cf80c4461355255d320e8 Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 16:30:05 +1100 Subject: [PATCH 1/3] fix(deps): take dfe-clickhouse 0.1.11 0.1.11 sets input_format_try_infer_dates=0 and input_format_try_infer_datetimes=0 on every HTTP insert into JSON columns. A date-like string at a _json path now stays String instead of mixing Date and String rows at that path. --- Cargo.lock | 6 +++--- Cargo.toml | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 07261db..43fa3ee 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1172,9 +1172,9 @@ dependencies = [ [[package]] name = "dfe-clickhouse" -version = "0.1.10" +version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f45c9c5d373c402f8f39a805b3c859b7dfc0add77b5e61cdfd53c383b822ab4c" +checksum = "459bcf7d97b2395e1894ef43a3024bbf0e89e97bb07776fd3ff626485c76c40d" dependencies = [ "backon", "bytes", @@ -4463,7 +4463,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d43fb82d550b07e6f929a58d7d0935182602f48ddead875f5ee313c41c8f07fe" dependencies = [ "annotate-snippets", - "base64 0.23.1", + "base64 0.22.1", "encoding_rs_io", "granit-parser", "nohash-hasher", diff --git a/Cargo.toml b/Cargo.toml index c26e22e..754357d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -82,7 +82,7 @@ maxminddb = { version = ">=0.32.0, <0.33", features = ["mmap"] } clickhouse = { version = ">=0.15.3, <0.16", features = ["lz4", "time", "rustls-tls", "rustls-tls-native-roots", "inserter"] } # `ext` reads the server's exception code out of an HTTP rejection; `tls` # (implies `tcp`) is the native-protocol client, rustls only. -dfe-clickhouse = { version = ">=0.1.10, <0.2", default-features = false, features = ["ext", "tls", "dynamic"] } +dfe-clickhouse = { version = ">=0.1.11, <0.2", default-features = false, features = ["ext", "tls", "dynamic"] } time = ">=0.3.55, <0.4" # The HTTP client handed to clickhouse::Client::with_http_client: connect timeout, From 9fd427bec01ca90accc09323df74334bf90241b4 Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 16:31:28 +1100 Subject: [PATCH 2/3] fix(config): drive risk scoring and discovery from config Risk scoring kept its country lists, weights, signal scores and level bands in match arms. Unknown presets silently fell back to global. Those settings now live in enrichment.risk_scoring. The five shipped presets form the default presets map. Deployments can override any value. Each preset entry replaces only the shipped preset of that name. Startup and reload reject unknown presets and name the valid ones. They also reject scores over 100 and level bands out of order. Auto-discovery subscribed to _(land|load)$ regardless of routing.topic_suffixes. The discovery include now uses the escaped suffixes. Loading fails for an empty suffix, or no suffix while discovery is on. Remove the coercion section and three ClickHouse type helpers. Only their own tests used them. The loader defaults in dfe-engine still write a coercion section, which the loader now ignores. --- benches/enrichment.rs | 16 +- config.example.yaml | 44 +- docs/config-schema.json | 716 +++++++++++++++++++---- docs/config-schema.yaml | 606 +++++++++++++++---- docs/pipeline/enrichment.md | 12 + docs/pipeline/routing.md | 2 +- src/clickhouse/mod.rs | 4 +- src/clickhouse/types.rs | 151 ----- src/config/loader.rs | 172 +++--- src/config/mod.rs | 10 +- src/config/pipeline.rs | 369 ++++++++---- src/enrich/risk.rs | 597 +++++++------------ src/kafka/transport.rs | 96 ++- src/pipeline/enrichment.rs | 33 +- src/pipeline/orchestrator.rs | 2 +- tests/integration/kafka_transport_e2e.rs | 31 +- tests/integration/shared_services.rs | 6 +- 17 files changed, 1841 insertions(+), 1026 deletions(-) diff --git a/benches/enrichment.rs b/benches/enrichment.rs index 8cba0b5..822fcaa 100644 --- a/benches/enrichment.rs +++ b/benches/enrichment.rs @@ -6,9 +6,10 @@ //! Measures CPU cost per operation for `GeoIP`, Reputation, and Risk scoring. use criterion::{Criterion, Throughput, criterion_group, criterion_main}; +use dfe_loader::config::RiskScoringConfig; use dfe_loader::enrich::geoip::{GeoIpEnricher, GeoIpResult}; use dfe_loader::enrich::reputation::{ReputationEnricher, ThreatSource, ThreatType}; -use dfe_loader::enrich::risk::{RiskInput, RiskPreset, RiskScorer}; +use dfe_loader::enrich::risk::{RiskInput, RiskScorer}; use std::hint::black_box; use std::net::IpAddr; @@ -160,7 +161,7 @@ fn bench_reputation_prefix_match(c: &mut Criterion) { } fn bench_risk_minimal(c: &mut Criterion) { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); c.bench_function("risk/minimal_input", |b| { b.iter(|| { @@ -171,7 +172,7 @@ fn bench_risk_minimal(c: &mut Criterion) { } fn bench_risk_full_input(c: &mut Criterion) { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); c.bench_function("risk/full_input", |b| { b.iter(|| { @@ -190,7 +191,10 @@ fn bench_risk_full_input(c: &mut Criterion) { } fn bench_risk_with_preset(c: &mut Criterion) { - let scorer = RiskScorer::from_preset(RiskPreset::HighSecurity); + let scorer = RiskScorer::new(&RiskScoringConfig { + preset: "high_security".to_string(), + ..RiskScoringConfig::default() + }); c.bench_function("risk/high_security_preset", |b| { b.iter(|| { @@ -205,7 +209,7 @@ fn bench_risk_with_preset(c: &mut Criterion) { } fn bench_risk_from_enrichment(c: &mut Criterion) { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); // Simulate results from GeoIP and Reputation let geo_result = GeoIpResult { @@ -235,7 +239,7 @@ fn bench_risk_from_enrichment(c: &mut Criterion) { fn bench_full_enrichment_pipeline(c: &mut Criterion) { let geoip = GeoIpEnricher::new(); let reputation = ReputationEnricher::new(); - let risk = RiskScorer::new(); + let risk = RiskScorer::default(); // Add some blocklist entries reputation.add_ip( diff --git a/config.example.yaml b/config.example.yaml index 8da3dc5..6c9bf5d 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -159,6 +159,8 @@ routing: # Topic suffixes stripped when deriving _source from Kafka topic name # Example: topic "auth_land" with suffix "_land" → _source = "auth" + # With kafka.topics and kafka.topic_regex unset, auto-discovery subscribes to + # the topics ending in one of these. topic_suffixes: - _land - _load @@ -304,10 +306,44 @@ metadata: raw_source_fields: [] raw_output: "_raw" -# Type coercion settings -coercion: - unknown_type_fallback: String - array_to_json: true +# IP enrichment. Risk scoring weighs the GeoIP and reputation results, so it +# needs at least one of them on. +enrichment: + ip_fields: [src_ip, client_ip, ip, source_ip] + risk_scoring: + enabled: false + # One of the presets below. A name not listed is refused at startup. + preset: global + # Shipped: global, us_enterprise, eu_enterprise, apac_enterprise, + # high_security. An entry here replaces the shipped one of that name, or + # adds a preset, and the rest stay as shipped. + # presets: + # global: + # high: [KP, IR, SY] + # medium: [RU, CN, VE, CU] + weights: {geo: 15, reputation: 25, privacy: 25, threat: 35} + # Score each signal gives its component, 0-100. Reputation and threat + # signals add up, capped at 100. + scores: + unknown_country: 35 + high_risk_country: 90 + medium_risk_country: 60 + other_country: 15 + vpn: 50 + proxy: 60 + tor: 80 + datacenter: 40 + relay: 45 + one_anonymiser: 40 + two_anonymisers: 70 + three_anonymisers: 95 + botnet: 95 + malicious: 90 + spam: 70 + scanner: 60 + high_abuse_score: 80 # abuse score that adds the high_abuse_score factor + # Lowest composite score of each level. Below low is minimal. + levels: {low: 20, medium: 40, high: 60, critical: 80} # Field mapping (rename/copy source fields to destination schema names) # Maps incoming field names to standardised schemas (ECS, CIM, etc.) diff --git a/docs/config-schema.json b/docs/config-schema.json index 2e343e9..cf9f561 100644 --- a/docs/config-schema.json +++ b/docs/config-schema.json @@ -297,35 +297,6 @@ "remove_routing_fields": true } }, - "coercion": { - "description": "Type coercion config. Refused unless every field is at its default: the\ndynamic insert applies fixed coercion rules.", - "$ref": "#/$defs/CoercionConfig", - "default": { - "type_mappings": {}, - "unknown_type_fallback": "String", - "null_handling": "default", - "null_strings": [ - "null", - "NULL", - "Null", - "None", - "nil", - "undefined", - "\\N", - "", - "NA", - "N/A", - "n/a", - "NaN" - ], - "default_timezone": "UTC", - "timezone_fields": [ - "tags_collector_timezone" - ], - "array_to_json": true, - "strict": false - } - }, "field_mapping": { "description": "Field mapping overrides. **Restart required.**", "$ref": "#/$defs/FieldMappingConfig", @@ -354,7 +325,117 @@ }, "risk_scoring": { "enabled": false, - "preset": "global" + "preset": "global", + "presets": { + "apac_enterprise": { + "high": [ + "KP", + "AF", + "MM" + ], + "medium": [ + "RU", + "IR" + ] + }, + "eu_enterprise": { + "high": [ + "RU", + "KP", + "IR", + "SY", + "BY" + ], + "medium": [ + "CN", + "UA", + "PK" + ] + }, + "global": { + "high": [ + "KP", + "IR", + "SY" + ], + "medium": [ + "RU", + "CN", + "VE", + "CU" + ] + }, + "high_security": { + "high": [ + "CN", + "RU", + "KP", + "IR", + "SY", + "CU", + "VE", + "BY", + "MM" + ], + "medium": [ + "UA", + "PK", + "NG", + "VN", + "BD", + "NP" + ] + }, + "us_enterprise": { + "high": [ + "CN", + "RU", + "KP", + "IR", + "SY", + "CU", + "VE" + ], + "medium": [ + "UA", + "BY", + "PK", + "NG", + "VN" + ] + } + }, + "weights": { + "geo": 15, + "reputation": 25, + "privacy": 25, + "threat": 35 + }, + "scores": { + "unknown_country": 35, + "high_risk_country": 90, + "medium_risk_country": 60, + "other_country": 15, + "vpn": 50, + "proxy": 60, + "tor": 80, + "datacenter": 40, + "relay": 45, + "one_anonymiser": 40, + "two_anonymisers": 70, + "three_anonymisers": 95, + "botnet": 95, + "malicious": 90, + "spam": 70, + "scanner": 60, + "high_abuse_score": 80 + }, + "levels": { + "low": 20, + "medium": 40, + "high": 60, + "critical": 80 + } } } } @@ -1732,96 +1813,6 @@ } ] }, - "CoercionConfig": { - "description": "Type coercion configuration. No encoder reads it, so `Config::validate`\nrefuses any field set away from its default.", - "type": "object", - "properties": { - "type_mappings": { - "description": "Map custom type names to base coercer categories\ne.g., {\"MyCustomInt\": \"Int\", \"SpecialString\": \"String\"}", - "type": "object", - "additionalProperties": { - "type": "string" - }, - "default": {} - }, - "unknown_type_fallback": { - "description": "Fallback coercer for unknown types (default: \"String\")", - "type": "string", - "default": "String" - }, - "null_handling": { - "description": "How to handle null values in non-nullable columns", - "$ref": "#/$defs/NullHandling", - "default": "default" - }, - "null_strings": { - "description": "Strings to recognise as null (case-sensitive)", - "type": "array", - "items": { - "type": "string" - }, - "default": [ - "null", - "NULL", - "Null", - "None", - "nil", - "undefined", - "\\N", - "", - "NA", - "N/A", - "n/a", - "NaN" - ] - }, - "default_timezone": { - "description": "Default timezone for naive timestamps (IANA name or +HH:MM)", - "type": "string", - "default": "UTC" - }, - "timezone_fields": { - "description": "Event fields to check for timezone info", - "type": "array", - "items": { - "type": "string" - }, - "default": [ - "tags_collector_timezone" - ] - }, - "array_to_json": { - "description": "Convert arrays to JSON strings if true", - "type": "boolean", - "default": true - }, - "strict": { - "description": "Strict mode: fail on any coercion error", - "type": "boolean", - "default": false - } - } - }, - "NullHandling": { - "description": "Null handling strategy", - "oneOf": [ - { - "description": "Substitute type-appropriate default value (recommended)", - "type": "string", - "const": "default" - }, - { - "description": "Return error for null in non-nullable column", - "type": "string", - "const": "error" - }, - { - "description": "Pass null through (may cause `ClickHouse` errors)", - "type": "string", - "const": "passthrough" - } - ] - }, "FieldMappingConfig": { "description": "Field mapping configuration for normalising source field names.\n\nSupports renaming or copying fields from source to destination names.\nRules come from two sources with clear precedence:\n1. `ClickHouse` column comments (`@renamed` directives) — highest priority\n2. External remap files (CSV/YAML/JSON) and built-in presets — lower priority\n\nCSV files are compatible with the elastic/ecs-mapper format:\n`source_field,destination_field,copy_action`", "type": "object", @@ -1903,7 +1894,117 @@ "$ref": "#/$defs/RiskScoringConfig", "default": { "enabled": false, - "preset": "global" + "preset": "global", + "presets": { + "apac_enterprise": { + "high": [ + "KP", + "AF", + "MM" + ], + "medium": [ + "RU", + "IR" + ] + }, + "eu_enterprise": { + "high": [ + "RU", + "KP", + "IR", + "SY", + "BY" + ], + "medium": [ + "CN", + "UA", + "PK" + ] + }, + "global": { + "high": [ + "KP", + "IR", + "SY" + ], + "medium": [ + "RU", + "CN", + "VE", + "CU" + ] + }, + "high_security": { + "high": [ + "CN", + "RU", + "KP", + "IR", + "SY", + "CU", + "VE", + "BY", + "MM" + ], + "medium": [ + "UA", + "PK", + "NG", + "VN", + "BD", + "NP" + ] + }, + "us_enterprise": { + "high": [ + "CN", + "RU", + "KP", + "IR", + "SY", + "CU", + "VE" + ], + "medium": [ + "UA", + "BY", + "PK", + "NG", + "VN" + ] + } + }, + "weights": { + "geo": 15, + "reputation": 25, + "privacy": 25, + "threat": 35 + }, + "scores": { + "unknown_country": 35, + "high_risk_country": 90, + "medium_risk_country": 60, + "other_country": 15, + "vpn": 50, + "proxy": 60, + "tor": 80, + "datacenter": 40, + "relay": 45, + "one_anonymiser": 40, + "two_anonymisers": 70, + "three_anonymisers": 95, + "botnet": 95, + "malicious": 90, + "spam": 70, + "scanner": 60, + "high_abuse_score": 80 + }, + "levels": { + "low": 20, + "medium": 40, + "high": 60, + "critical": 80 + } } } } @@ -1935,7 +2036,7 @@ } }, "RiskScoringConfig": { - "description": "Risk scoring configuration", + "description": "Risk scoring configuration. Every country list, weight, score and level band\nbelow is a default a deployment can override.", "type": "object", "properties": { "enabled": { @@ -1944,9 +2045,378 @@ "default": false }, "preset": { - "description": "Risk preset to use for country risk tables\n\nOptions: \"global\" (default), \"`us_enterprise`\", \"`eu_enterprise`\",\n \"`apac_enterprise`\", \"`high_security`\"", + "description": "The entry in `presets` whose country lists apply. A name `presets` does\nnot hold is refused at load.", "type": "string", "default": "global" + }, + "presets": { + "description": "Country lists by preset name. The loader ships `global`,\n`us_enterprise`, `eu_enterprise`, `apac_enterprise` and `high_security`.\nAn entry set in config replaces the shipped entry of that name, or adds\na new preset, and leaves the others as shipped.", + "type": "object", + "additionalProperties": { + "$ref": "#/$defs/RiskCountries" + }, + "default": { + "apac_enterprise": { + "high": [ + "KP", + "AF", + "MM" + ], + "medium": [ + "RU", + "IR" + ] + }, + "eu_enterprise": { + "high": [ + "RU", + "KP", + "IR", + "SY", + "BY" + ], + "medium": [ + "CN", + "UA", + "PK" + ] + }, + "global": { + "high": [ + "KP", + "IR", + "SY" + ], + "medium": [ + "RU", + "CN", + "VE", + "CU" + ] + }, + "high_security": { + "high": [ + "CN", + "RU", + "KP", + "IR", + "SY", + "CU", + "VE", + "BY", + "MM" + ], + "medium": [ + "UA", + "PK", + "NG", + "VN", + "BD", + "NP" + ] + }, + "us_enterprise": { + "high": [ + "CN", + "RU", + "KP", + "IR", + "SY", + "CU", + "VE" + ], + "medium": [ + "UA", + "BY", + "PK", + "NG", + "VN" + ] + } + } + }, + "weights": { + "description": "Weight of each component in the composite score.", + "$ref": "#/$defs/RiskWeights", + "default": { + "geo": 15, + "reputation": 25, + "privacy": 25, + "threat": 35 + } + }, + "scores": { + "description": "Score each signal gives its component, 0-100.", + "$ref": "#/$defs/RiskScores", + "default": { + "unknown_country": 35, + "high_risk_country": 90, + "medium_risk_country": 60, + "other_country": 15, + "vpn": 50, + "proxy": 60, + "tor": 80, + "datacenter": 40, + "relay": 45, + "one_anonymiser": 40, + "two_anonymisers": 70, + "three_anonymisers": 95, + "botnet": 95, + "malicious": 90, + "spam": 70, + "scanner": 60, + "high_abuse_score": 80 + } + }, + "levels": { + "description": "Lowest composite score of each level above `minimal`.", + "$ref": "#/$defs/RiskLevelBands", + "default": { + "low": 20, + "medium": 40, + "high": 60, + "critical": 80 + } + } + } + }, + "RiskCountries": { + "description": "The high- and medium-risk country codes of one preset (ISO 3166-1 alpha-2).", + "type": "object", + "properties": { + "high": { + "description": "Countries scored `scores.high_risk_country`.", + "type": "array", + "items": { + "type": "string" + }, + "default": [] + }, + "medium": { + "description": "Countries scored `scores.medium_risk_country`.", + "type": "array", + "items": { + "type": "string" + }, + "default": [] + } + } + }, + "RiskWeights": { + "description": "Weight of each component in the composite score. All zero scores 0.", + "type": "object", + "properties": { + "geo": { + "description": "Geographic component.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 15 + }, + "reputation": { + "description": "Reputation component.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 25 + }, + "privacy": { + "description": "Privacy (anonymiser) component.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 25 + }, + "threat": { + "description": "Threat component.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 35 + } + } + }, + "RiskScores": { + "description": "Score each signal gives its component. Reputation and threat signals add\nup, capped at 100.", + "type": "object", + "properties": { + "unknown_country": { + "description": "Geo: a public IP with no country.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 35 + }, + "high_risk_country": { + "description": "Geo: a country in the preset's `high` list.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 90 + }, + "medium_risk_country": { + "description": "Geo: a country in the preset's `medium` list.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 60 + }, + "other_country": { + "description": "Geo: any other country.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 15 + }, + "vpn": { + "description": "Reputation: VPN exit.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 50 + }, + "proxy": { + "description": "Reputation: open proxy.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 60 + }, + "tor": { + "description": "Reputation: Tor exit.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 80 + }, + "datacenter": { + "description": "Reputation: datacenter range.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 40 + }, + "relay": { + "description": "Reputation: private relay.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 45 + }, + "one_anonymiser": { + "description": "Privacy: one anonymiser among VPN, proxy, Tor and relay.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 40 + }, + "two_anonymisers": { + "description": "Privacy: two anonymisers.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 70 + }, + "three_anonymisers": { + "description": "Privacy: three or more anonymisers.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 95 + }, + "botnet": { + "description": "Threat: botnet member.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 95 + }, + "malicious": { + "description": "Threat: known malicious.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 90 + }, + "spam": { + "description": "Threat: spam source.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 70 + }, + "scanner": { + "description": "Threat: scanner.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 60 + }, + "high_abuse_score": { + "description": "Threat: the external abuse score at or above which the row carries the\n`high_abuse_score` factor. The abuse score itself is added as given.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 80 + } + } + }, + "RiskLevelBands": { + "description": "Lowest composite score of each risk level. Below `low` is `minimal`.", + "type": "object", + "properties": { + "low": { + "description": "Lowest `low` score.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 20 + }, + "medium": { + "description": "Lowest `medium` score.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 40 + }, + "high": { + "description": "Lowest `high` score.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 60 + }, + "critical": { + "description": "Lowest `critical` score.", + "type": "integer", + "format": "uint8", + "minimum": 0, + "maximum": 255, + "default": 80 } } } diff --git a/docs/config-schema.yaml b/docs/config-schema.yaml index 93f1cd2..c4a4c05 100644 --- a/docs/config-schema.yaml +++ b/docs/config-schema.yaml @@ -275,33 +275,6 @@ properties: disable_json_tables: [] disable_raw_tables: [] remove_routing_fields: true - coercion: - description: |- - Type coercion config. Refused unless every field is at its default: the - dynamic insert applies fixed coercion rules. - $ref: "#/$defs/CoercionConfig" - default: - type_mappings: {} - unknown_type_fallback: String - null_handling: default - null_strings: - - "null" - - "NULL" - - "Null" - - None - - nil - - undefined - - \N - - - - NA - - N/A - - n/a - - "NaN" - default_timezone: UTC - timezone_fields: - - tags_collector_timezone - array_to_json: true - strict: false field_mapping: description: Field mapping overrides. **Restart required.** $ref: "#/$defs/FieldMappingConfig" @@ -327,6 +300,97 @@ properties: risk_scoring: enabled: false preset: global + presets: + apac_enterprise: + high: + - KP + - AF + - MM + medium: + - RU + - IR + eu_enterprise: + high: + - RU + - KP + - IR + - SY + - BY + medium: + - CN + - UA + - PK + global: + high: + - KP + - IR + - SY + medium: + - RU + - CN + - VE + - CU + high_security: + high: + - CN + - RU + - KP + - IR + - SY + - CU + - VE + - BY + - MM + medium: + - UA + - PK + - NG + - VN + - BD + - NP + us_enterprise: + high: + - CN + - RU + - KP + - IR + - SY + - CU + - VE + medium: + - UA + - BY + - PK + - NG + - VN + weights: + geo: 15 + reputation: 25 + privacy: 25 + threat: 35 + scores: + unknown_country: 35 + high_risk_country: 90 + medium_risk_country: 60 + other_country: 15 + vpn: 50 + proxy: 60 + tor: 80 + datacenter: 40 + relay: 45 + one_anonymiser: 40 + two_anonymisers: 70 + three_anonymisers: 95 + botnet: 95 + malicious: 90 + spam: 70 + scanner: 60 + high_abuse_score: 80 + levels: + low: 20 + medium: 40 + high: 60 + critical: 80 $defs: KafkaConfig: type: object @@ -1568,77 +1632,6 @@ $defs: - description: "`_json` = NULL, `_raw` = NULL -- only promoted schema fields" type: string const: extracted_only - CoercionConfig: - description: |- - Type coercion configuration. No encoder reads it, so `Config::validate` - refuses any field set away from its default. - type: object - properties: - type_mappings: - description: |- - Map custom type names to base coercer categories - e.g., {"MyCustomInt": "Int", "SpecialString": "String"} - type: object - additionalProperties: - type: string - default: {} - unknown_type_fallback: - description: "Fallback coercer for unknown types (default: \"String\")" - type: string - default: String - null_handling: - description: How to handle null values in non-nullable columns - $ref: "#/$defs/NullHandling" - default: default - null_strings: - description: Strings to recognise as null (case-sensitive) - type: array - items: - type: string - default: - - "null" - - "NULL" - - "Null" - - None - - nil - - undefined - - \N - - - - NA - - N/A - - n/a - - "NaN" - default_timezone: - description: Default timezone for naive timestamps (IANA name or +HH:MM) - type: string - default: UTC - timezone_fields: - description: Event fields to check for timezone info - type: array - items: - type: string - default: - - tags_collector_timezone - array_to_json: - description: Convert arrays to JSON strings if true - type: boolean - default: true - strict: - description: "Strict mode: fail on any coercion error" - type: boolean - default: false - NullHandling: - description: Null handling strategy - oneOf: - - description: Substitute type-appropriate default value (recommended) - type: string - const: default - - description: Return error for null in non-nullable column - type: string - const: error - - description: Pass null through (may cause `ClickHouse` errors) - type: string - const: passthrough FieldMappingConfig: description: |- Field mapping configuration for normalising source field names. @@ -1720,6 +1713,97 @@ $defs: default: enabled: false preset: global + presets: + apac_enterprise: + high: + - KP + - AF + - MM + medium: + - RU + - IR + eu_enterprise: + high: + - RU + - KP + - IR + - SY + - BY + medium: + - CN + - UA + - PK + global: + high: + - KP + - IR + - SY + medium: + - RU + - CN + - VE + - CU + high_security: + high: + - CN + - RU + - KP + - IR + - SY + - CU + - VE + - BY + - MM + medium: + - UA + - PK + - NG + - VN + - BD + - NP + us_enterprise: + high: + - CN + - RU + - KP + - IR + - SY + - CU + - VE + medium: + - UA + - BY + - PK + - NG + - VN + weights: + geo: 15 + reputation: 25 + privacy: 25 + threat: 35 + scores: + unknown_country: 35 + high_risk_country: 90 + medium_risk_country: 60 + other_country: 15 + vpn: 50 + proxy: 60 + tor: 80 + datacenter: 40 + relay: 45 + one_anonymiser: 40 + two_anonymisers: 70 + three_anonymisers: 95 + botnet: 95 + malicious: 90 + spam: 70 + scanner: 60 + high_abuse_score: 80 + levels: + low: 20 + medium: 40 + high: 60 + critical: 80 ReputationEnrichmentConfig: description: Reputation enrichment configuration type: object @@ -1741,7 +1825,9 @@ $defs: type: string default: [] RiskScoringConfig: - description: Risk scoring configuration + description: |- + Risk scoring configuration. Every country list, weight, score and level band + below is a default a deployment can override. type: object properties: enabled: @@ -1750,9 +1836,323 @@ $defs: default: false preset: description: |- - Risk preset to use for country risk tables - - Options: "global" (default), "`us_enterprise`", "`eu_enterprise`", - "`apac_enterprise`", "`high_security`" + The entry in `presets` whose country lists apply. A name `presets` does + not hold is refused at load. type: string default: global + presets: + description: |- + Country lists by preset name. The loader ships `global`, + `us_enterprise`, `eu_enterprise`, `apac_enterprise` and `high_security`. + An entry set in config replaces the shipped entry of that name, or adds + a new preset, and leaves the others as shipped. + type: object + additionalProperties: + $ref: "#/$defs/RiskCountries" + default: + apac_enterprise: + high: + - KP + - AF + - MM + medium: + - RU + - IR + eu_enterprise: + high: + - RU + - KP + - IR + - SY + - BY + medium: + - CN + - UA + - PK + global: + high: + - KP + - IR + - SY + medium: + - RU + - CN + - VE + - CU + high_security: + high: + - CN + - RU + - KP + - IR + - SY + - CU + - VE + - BY + - MM + medium: + - UA + - PK + - NG + - VN + - BD + - NP + us_enterprise: + high: + - CN + - RU + - KP + - IR + - SY + - CU + - VE + medium: + - UA + - BY + - PK + - NG + - VN + weights: + description: Weight of each component in the composite score. + $ref: "#/$defs/RiskWeights" + default: + geo: 15 + reputation: 25 + privacy: 25 + threat: 35 + scores: + description: Score each signal gives its component, 0-100. + $ref: "#/$defs/RiskScores" + default: + unknown_country: 35 + high_risk_country: 90 + medium_risk_country: 60 + other_country: 15 + vpn: 50 + proxy: 60 + tor: 80 + datacenter: 40 + relay: 45 + one_anonymiser: 40 + two_anonymisers: 70 + three_anonymisers: 95 + botnet: 95 + malicious: 90 + spam: 70 + scanner: 60 + high_abuse_score: 80 + levels: + description: Lowest composite score of each level above `minimal`. + $ref: "#/$defs/RiskLevelBands" + default: + low: 20 + medium: 40 + high: 60 + critical: 80 + RiskCountries: + description: The high- and medium-risk country codes of one preset (ISO 3166-1 alpha-2). + type: object + properties: + high: + description: Countries scored `scores.high_risk_country`. + type: array + items: + type: string + default: [] + medium: + description: Countries scored `scores.medium_risk_country`. + type: array + items: + type: string + default: [] + RiskWeights: + description: Weight of each component in the composite score. All zero scores 0. + type: object + properties: + geo: + description: Geographic component. + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 15 + reputation: + description: Reputation component. + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 25 + privacy: + description: Privacy (anonymiser) component. + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 25 + threat: + description: Threat component. + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 35 + RiskScores: + description: |- + Score each signal gives its component. Reputation and threat signals add + up, capped at 100. + type: object + properties: + unknown_country: + description: "Geo: a public IP with no country." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 35 + high_risk_country: + description: "Geo: a country in the preset's `high` list." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 90 + medium_risk_country: + description: "Geo: a country in the preset's `medium` list." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 60 + other_country: + description: "Geo: any other country." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 15 + vpn: + description: "Reputation: VPN exit." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 50 + proxy: + description: "Reputation: open proxy." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 60 + tor: + description: "Reputation: Tor exit." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 80 + datacenter: + description: "Reputation: datacenter range." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 40 + relay: + description: "Reputation: private relay." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 45 + one_anonymiser: + description: "Privacy: one anonymiser among VPN, proxy, Tor and relay." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 40 + two_anonymisers: + description: "Privacy: two anonymisers." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 70 + three_anonymisers: + description: "Privacy: three or more anonymisers." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 95 + botnet: + description: "Threat: botnet member." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 95 + malicious: + description: "Threat: known malicious." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 90 + spam: + description: "Threat: spam source." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 70 + scanner: + description: "Threat: scanner." + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 60 + high_abuse_score: + description: |- + Threat: the external abuse score at or above which the row carries the + `high_abuse_score` factor. The abuse score itself is added as given. + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 80 + RiskLevelBands: + description: Lowest composite score of each risk level. Below `low` is `minimal`. + type: object + properties: + low: + description: Lowest `low` score. + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 20 + medium: + description: Lowest `medium` score. + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 40 + high: + description: Lowest `high` score. + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 60 + critical: + description: Lowest `critical` score. + type: integer + format: uint8 + minimum: 0 + maximum: 255 + default: 80 diff --git a/docs/pipeline/enrichment.md b/docs/pipeline/enrichment.md index 15d20e6..f729649 100644 --- a/docs/pipeline/enrichment.md +++ b/docs/pipeline/enrichment.md @@ -41,6 +41,18 @@ Risk scoring is a composite: it needs at least GeoIP or reputation enabled to have anything to weigh. Enabling it alone does nothing useful, which the config documents at the field. +Every value the scorer uses is config under `enrichment.risk_scoring`, with the shipped values as defaults. `config.example.yaml` lists them all. + +| Key | What it sets | Default | +|-----|--------------|---------| +| `preset` | the entry in `presets` whose country lists apply | `global` | +| `presets` | `high` and `medium` country lists by name | `global`, `us_enterprise`, `eu_enterprise`, `apac_enterprise`, `high_security` | +| `weights` | `geo`, `reputation`, `privacy`, `threat` share of the composite | 15, 25, 25, 35 | +| `scores` | what each signal adds to its component: country class, VPN, proxy, Tor, datacenter, relay, anonymiser count, botnet, malicious, spam, scanner, and the `high_abuse_score` threshold | `config.example.yaml` | +| `levels` | lowest composite score of `low`, `medium`, `high`, `critical` | 20, 40, 60, 80 | + +A `presets` entry in config replaces the shipped preset of that name, or adds a new one, and the rest stay as shipped. Startup and a reload refuse a `preset` that `presets` does not hold, naming the valid ones, as well as a score over 100 and level bands that do not rise strictly within 1-100. + ## Where the GeoIP databases come from With `geoip.enabled` on, startup provisions the MMDB files through scalo's `geoip_download`, then opens them with `maxminddb`. An explicit `geoip.city_db_path` or `asn_db_path` is used as given. Otherwise a file in `geoip.auto_download.data_dir` (default `/var/lib/dfe/geoip`, the volume the deployment contract mounts) younger than `max_age_days` is kept, a missing or older one is downloaded, and a failed download falls back to the old file. A download larger than `max_database_bytes` (default 512 MiB), or one that does not read as an MMDB, is refused. A missing database leaves enrichment inactive and never stops the loader. diff --git a/docs/pipeline/routing.md b/docs/pipeline/routing.md index 3e14907..ab0b206 100644 --- a/docs/pipeline/routing.md +++ b/docs/pipeline/routing.md @@ -45,7 +45,7 @@ flowchart TB | `routing.table_fields` | fields whose value selects the table, first one carrying a value wins. Dot-notation reaches nested fields (`event.category`). Default `["_source"]` | | `routing.default_db` | database for every org `org_routes` does not list -- default `dfe` | | `routing.default_table` | table used when neither a `table_fields` value nor the topic names one -- default `main`, so an unrouted message lands in `dfe.main` | -| `routing.topic_suffixes` | suffixes that mark a topic as one source's own -- default `_land` / `_load`, so `cisco-ios_load` names the source `cisco-ios` | +| `routing.topic_suffixes` | suffixes that mark a topic as one source's own -- default `_land` / `_load`, so `cisco-ios_load` names the source `cisco-ios`. With `kafka.topics` and `kafka.topic_regex` unset, auto-discovery subscribes to the topics ending in one, so an empty list or an empty suffix is refused | | `routing.org_id_field` | field the tenant is read from -- default `_org_id`, the common-header field the first hop stamps | | `routing.default_org_id` | tenant for a record that carries none -- default `default`, the deployment's default org | | `routing.org_routes` | orgs that get a database of their own, keyed on the tenant: `[{org_id: acme, database: acme_dfe}]` | diff --git a/src/clickhouse/mod.rs b/src/clickhouse/mod.rs index 9882e5b..0ccaacc 100644 --- a/src/clickhouse/mod.rs +++ b/src/clickhouse/mod.rs @@ -41,9 +41,7 @@ pub mod schema; pub use client_http::{ChClient, ClickHouseQueryClient, PoolStats, SharedQueryClient}; pub use config::{ClickHouseConfig, InsertFormat, Timeouts, Transport, host_implies_tls}; pub use error::{ClickHouseError, ErrorCategory}; -pub use types::{ - ColumnInfo, NULL_STRINGS, ParsedType, TableSchema, default_value_for_category, is_null_string, -}; +pub use types::{ColumnInfo, ParsedType, TableSchema}; // Export resilience modules pub use inserter::{BatchDisposition, FailedRow, InsertResult, Inserter, InserterConfig}; diff --git a/src/clickhouse/types.rs b/src/clickhouse/types.rs index 2671130..e866679 100644 --- a/src/clickhouse/types.rs +++ b/src/clickhouse/types.rs @@ -80,54 +80,6 @@ impl TableSchema { } } -/// Default values for `ClickHouse` types (used when handling nulls). -/// -/// These are safe zero values that avoid NULL in `ClickHouse` -/// (following `ClickHouse` best practices to avoid Nullable overhead). -#[must_use] -pub fn default_value_for_category(category: &str) -> &'static str { - match category { - "String" => "", - "Int" | "UInt" => "0", - "Float" | "Decimal" => "0.0", - "Bool" => "false", - "Date" => "1970-01-01", - "DateTime" | "DateTime64" => "1970-01-01T00:00:00Z", - "UUID" => "00000000-0000-0000-0000-000000000000", - "IPv4" => "0.0.0.0", - "IPv6" => "::", - "Array" => "[]", - "Map" => "{}", - "JSON" => "{}", - "Enum" => "", - "Geo" => "(0, 0)", - _ => "", - } -} - -/// Common null string representations to recognise. -pub const NULL_STRINGS: &[&str] = &[ - "null", - "NULL", - "Null", - "None", - "nil", - "undefined", - "\\N", - "", - "NA", - "N/A", - "n/a", - "NaN", - "", -]; - -/// Check if a string value represents null. -#[must_use] -pub fn is_null_string(value: &str) -> bool { - NULL_STRINGS.contains(&value) -} - #[cfg(test)] mod tests { use super::*; @@ -157,16 +109,6 @@ mod tests { } } - #[test] - fn test_null_strings() { - assert!(is_null_string("null")); - assert!(is_null_string("NULL")); - assert!(is_null_string("None")); - assert!(is_null_string("")); - assert!(!is_null_string("hello")); - assert!(!is_null_string("0")); - } - // ---- NEW: TableSchema tests ---- #[test] @@ -276,97 +218,4 @@ mod tests { assert!(col.is_in_primary_key); assert!(col.is_in_sorting_key); } - - // ---- NEW: default_value_for_category tests ---- - - #[test] - fn default_values_all_known_categories() { - assert_eq!(default_value_for_category("String"), ""); - assert_eq!(default_value_for_category("Int"), "0"); - assert_eq!(default_value_for_category("UInt"), "0"); - assert_eq!(default_value_for_category("Float"), "0.0"); - assert_eq!(default_value_for_category("Decimal"), "0.0"); - assert_eq!(default_value_for_category("Bool"), "false"); - assert_eq!(default_value_for_category("Date"), "1970-01-01"); - assert_eq!( - default_value_for_category("DateTime"), - "1970-01-01T00:00:00Z" - ); - assert_eq!( - default_value_for_category("DateTime64"), - "1970-01-01T00:00:00Z" - ); - assert_eq!( - default_value_for_category("UUID"), - "00000000-0000-0000-0000-000000000000" - ); - assert_eq!(default_value_for_category("IPv4"), "0.0.0.0"); - assert_eq!(default_value_for_category("IPv6"), "::"); - assert_eq!(default_value_for_category("Array"), "[]"); - assert_eq!(default_value_for_category("Map"), "{}"); - assert_eq!(default_value_for_category("JSON"), "{}"); - assert_eq!(default_value_for_category("Enum"), ""); - assert_eq!(default_value_for_category("Geo"), "(0, 0)"); - } - - #[test] - fn default_value_unknown_category_returns_empty() { - assert_eq!(default_value_for_category(""), ""); - assert_eq!(default_value_for_category("SomeFutureType"), ""); - assert_eq!(default_value_for_category("Tuple"), ""); - } - - // ---- NEW: is_null_string edge cases ---- - - #[test] - fn null_string_all_known_representations() { - for &s in NULL_STRINGS { - assert!(is_null_string(s), "Expected '{s}' to be recognised as null"); - } - } - - #[test] - fn null_string_case_sensitivity() { - // "null", "NULL", "Null" are in the list, but not "nULL" or "nUll" - assert!(is_null_string("null")); - assert!(is_null_string("NULL")); - assert!(is_null_string("Null")); - assert!(!is_null_string("nULL")); - assert!(!is_null_string("nUll")); - } - - #[test] - fn null_string_whitespace_not_recognised() { - assert!(!is_null_string(" ")); - assert!(!is_null_string(" null ")); - assert!(!is_null_string("\tnull")); - assert!(!is_null_string("null\n")); - } - - #[test] - fn null_string_similar_but_not_null() { - assert!(!is_null_string("0")); - assert!(!is_null_string("false")); - assert!(!is_null_string("no")); - assert!(!is_null_string("nulls")); - assert!(!is_null_string("NONE")); // "None" is in list, "NONE" is not - assert!(!is_null_string("Nil")); // "nil" is in list, "Nil" is not - } - - #[test] - fn null_string_backslash_n() { - // \\N is a ClickHouse-specific null representation - assert!(is_null_string("\\N")); - assert!(!is_null_string("\\n")); // lowercase — not in list - } - - #[test] - fn null_string_na_variants() { - assert!(is_null_string("NA")); - assert!(is_null_string("N/A")); - assert!(is_null_string("n/a")); - assert!(is_null_string("NaN")); - assert!(!is_null_string("na")); // not in list - assert!(!is_null_string("nan")); // not in list - } } diff --git a/src/config/loader.rs b/src/config/loader.rs index 5901840..5ae0584 100644 --- a/src/config/loader.rs +++ b/src/config/loader.rs @@ -103,9 +103,6 @@ pub struct Config { pub field_sanitization: FieldSanitizationConfig, /// Common header injection, tags, _raw handling. **Hot-reloaded.** pub metadata: MetadataConfig, - /// Type coercion config. Refused unless every field is at its default: the - /// dynamic insert applies fixed coercion rules. - pub coercion: CoercionConfig, /// Field mapping overrides. **Restart required.** pub field_mapping: FieldMappingConfig, /// IP enrichment field selection. **Restart required.** @@ -150,7 +147,6 @@ impl Default for Config { timestamp_dq: TimestampDqConfig::default(), field_sanitization: FieldSanitizationConfig::default(), metadata: MetadataConfig::default(), - coercion: CoercionConfig::default(), field_mapping: FieldMappingConfig::default(), enrichment: EnrichmentConfig::default(), } @@ -626,7 +622,6 @@ impl Config { registry::register("metrics", &self.metrics); registry::register("logging", &self.logging); registry::register("payload", &self.payload); - registry::register("coercion", &self.coercion); registry::register("enrichment", &self.enrichment); } @@ -697,6 +692,10 @@ impl Config { } self.refuse_unread_settings()?; + self.enrichment + .risk_scoring + .validate() + .map_err(crate::Error::Config)?; // An empty default would land every unstamped record in no org at all. if self.routing.default_org_id.is_empty() { @@ -705,6 +704,24 @@ impl Config { )); } + // An empty suffix strips nothing, and with no suffix auto-discovery has + // no topic to subscribe to. + if self.routing.topic_suffixes.iter().any(String::is_empty) { + return Err(crate::Error::Config( + "routing.topic_suffixes must not hold an empty value".into(), + )); + } + let discovers = self.transport != TRANSPORT_GRPC + && self.kafka.topics.is_empty() + && self.kafka.topic_regex.as_deref().is_none_or(str::is_empty); + if discovers && self.routing.topic_suffixes.is_empty() { + return Err(crate::Error::Config( + "routing.topic_suffixes must name a suffix while kafka.topics and \ + kafka.topic_regex are unset: auto-discovery subscribes to topics ending in one" + .into(), + )); + } + self.validate_external_topics()?; // An empty name would subscribe to no topic, or route a dead letter to no table. @@ -808,35 +825,10 @@ impl Config { /// Refuse a setting no code reads, rather than accept it and ignore it. /// /// The dynamic insert applies fixed coercion rules - /// (`docs/clickhouse/clickhouse-ext.md`), so the `coercion` section and the - /// `coerce` column directive stay at their defaults, and an invalid - /// timestamp is always replaced with the ingest time. + /// (`docs/clickhouse/clickhouse-ext.md`), so the `coerce` column directive + /// stays unset, and an invalid timestamp is always replaced with the ingest + /// time. fn refuse_unread_settings(&self) -> Result<()> { - let c = &self.coercion; - let d = CoercionConfig::default(); - let set: Vec<&str> = [ - ("type_mappings", c.type_mappings != d.type_mappings), - ( - "unknown_type_fallback", - c.unknown_type_fallback != d.unknown_type_fallback, - ), - ("null_handling", c.null_handling != d.null_handling), - ("null_strings", c.null_strings != d.null_strings), - ("default_timezone", c.default_timezone != d.default_timezone), - ("timezone_fields", c.timezone_fields != d.timezone_fields), - ("array_to_json", c.array_to_json != d.array_to_json), - ("strict", c.strict != d.strict), - ] - .into_iter() - .filter_map(|(name, changed)| changed.then_some(name)) - .collect(); - if !set.is_empty() { - return Err(crate::Error::Config(format!( - "coercion.{} reaches no encoder -- the dynamic insert applies fixed \ - coercion rules. Leave the coercion section at its defaults.", - set.join(", coercion.") - ))); - } let directives = &self.column_directives; if let Some(column) = directives .global @@ -1279,17 +1271,31 @@ mod tests { assert!(config.validate().is_ok()); } - /// A setting no code reads is refused rather than accepted and ignored: - /// the `coercion` section, a `coerce` column directive, and any + /// A setting no code reads, or a value the loader cannot use, is refused + /// rather than accepted and ignored: an empty or missing topic suffix under + /// discovery, an unknown risk preset, a risk score past 100, risk level + /// bands out of order, a `coerce` column directive, and any /// `timestamp_dq.invalid_action` but the one the loader takes. #[test] fn a_setting_nothing_honours_is_refused() { - let cases: [(&str, fn(&mut Config)); 5] = [ - ("coercion.strict", |c| c.coercion.strict = true), - ("coercion.default_timezone", |c| { - c.coercion.default_timezone = "Australia/Sydney".into(); + let cases: [(&str, fn(&mut Config)); 7] = [ + ("routing.topic_suffixes must not hold an empty value", |c| { + c.routing.topic_suffixes.push(String::new()); + }), + ("routing.topic_suffixes must name a suffix", |c| { + c.routing.topic_suffixes.clear(); + }), + ( + "preset 'us' is not a preset -- valid: apac_enterprise, eu_enterprise, \ + global, high_security, us_enterprise", + |c| c.enrichment.risk_scoring.preset = "us".into(), + ), + ("enrichment.risk_scoring.scores.tor", |c| { + c.enrichment.risk_scoring.scores.tor = 101; + }), + ("enrichment.risk_scoring.levels", |c| { + c.enrichment.risk_scoring.levels.high = 30; }), - ("coercion.null_strings", |c| c.coercion.null_strings.clear()), ("timestamp_dq.invalid_action", |c| { c.timestamp_dq.invalid_action = "drop".into(); }), @@ -1462,55 +1468,47 @@ mod tests { assert!(config.parse_mechanism().is_none()); } - // ======================================================================== - // Coercion Config Tests - // ======================================================================== - + /// A config file overrides any risk list, weight, score or band, and a + /// preset it sets replaces only the shipped preset of that name. #[test] - fn test_coercion_config_is_null_string() { - let config = CoercionConfig::default(); - - // Empty string is null - assert!(config.is_null_string("")); - - // Standard null strings - assert!(config.is_null_string("null")); - assert!(config.is_null_string("NULL")); - assert!(config.is_null_string("None")); - assert!(config.is_null_string("nil")); - assert!(config.is_null_string("undefined")); - assert!(config.is_null_string("\\N")); - assert!(config.is_null_string("N/A")); - assert!(config.is_null_string("NaN")); - - // Not null strings - assert!(!config.is_null_string("hello")); - assert!(!config.is_null_string("0")); - assert!(!config.is_null_string("false")); - } - - #[test] - fn test_coercion_config_get_coercer_category() { - let mut config = CoercionConfig::default(); - config - .type_mappings - .insert("MyInt".to_string(), "Int".to_string()); - config - .type_mappings - .insert("SpecialString".to_string(), "String".to_string()); - - // Custom mappings - assert_eq!(config.get_coercer_category("MyInt"), "Int"); - assert_eq!(config.get_coercer_category("SpecialString"), "String"); - - // Unknown type falls back to default - assert_eq!(config.get_coercer_category("UnknownType"), "String"); - } + fn a_config_file_overrides_the_risk_scoring_defaults() { + let yaml = r" +enrichment: + risk_scoring: + enabled: true + preset: global + presets: + global: + high: [XX] + mine: + medium: [ZZ] + weights: + threat: 50 + scores: + tor: 70 + levels: + critical: 90 +"; + let config: Config = + scalo::yaml::from_str(yaml, "config.yaml", scalo::yaml::Profile::Config).unwrap(); + config.validate().unwrap(); + let risk = &config.enrichment.risk_scoring; + let shipped = crate::config::RiskScoringConfig::default(); - #[test] - fn test_null_handling_default() { - let handling = NullHandling::default(); - assert_eq!(handling, NullHandling::Default); + assert_eq!(risk.countries().unwrap().high, ["XX"]); + assert!(risk.countries().unwrap().medium.is_empty()); + assert_eq!(risk.presets["mine"].medium, ["ZZ"]); + assert_eq!( + risk.presets["high_security"], + shipped.presets["high_security"] + ); + assert_eq!(risk.presets.len(), shipped.presets.len() + 1); + assert_eq!(risk.weights.threat, 50); + assert_eq!(risk.weights.geo, shipped.weights.geo); + assert_eq!(risk.scores.tor, 70); + assert_eq!(risk.scores.vpn, shipped.scores.vpn); + assert_eq!(risk.levels.critical, 90); + assert_eq!(risk.levels.low, shipped.levels.low); } // ======================================================================== diff --git a/src/config/mod.rs b/src/config/mod.rs index 3f58653..307c3ba 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -38,11 +38,11 @@ pub use loader::{ MemoryConfig, MetricsConfig, ScalingConfig, }; pub use pipeline::{ - AutoDownloadConfig, CaptureMode, CoercionConfig, ComputedColumnsConfig, DEFAULT_ORG_ID, - DlqConfig, DlqLoadConfig, EnrichmentConfig, ExternalTopic, FieldMappingConfig, - FieldMappingOverride, FieldSanitizationConfig, GeoIpConfig, GeoIpProvider, MetadataConfig, - NullHandling, OrgRoute, PayloadConfig, ReputationEnrichmentConfig, RiskScoringConfig, - RoutingConfig, SchemaConfig, TableCaptureConfig, TimestampDqConfig, + AutoDownloadConfig, CaptureMode, ComputedColumnsConfig, DEFAULT_ORG_ID, DlqConfig, + DlqLoadConfig, EnrichmentConfig, ExternalTopic, FieldMappingConfig, FieldMappingOverride, + FieldSanitizationConfig, GeoIpConfig, GeoIpProvider, MetadataConfig, OrgRoute, PayloadConfig, + ReputationEnrichmentConfig, RiskCountries, RiskLevelBands, RiskScores, RiskScoringConfig, + RiskWeights, RoutingConfig, SchemaConfig, TableCaptureConfig, TimestampDqConfig, }; pub use shared::SharedConfig; pub use watcher::{ConfigWatcher, WatcherConfig}; diff --git a/src/config/pipeline.rs b/src/config/pipeline.rs index 716cbb2..5d7740a 100644 --- a/src/config/pipeline.rs +++ b/src/config/pipeline.rs @@ -1,9 +1,9 @@ // SPDX-License-Identifier: BUSL-1.1 // Copyright (c) 2026 HYPERI PTY LIMITED -//! Pipeline configuration: routing, enrichment, coercion, metadata, field mapping. +//! Pipeline configuration: routing, enrichment, metadata, field mapping. -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::path::PathBuf; use scalo::config::sensitive::SensitiveString; @@ -364,18 +364,33 @@ impl Default for ReputationEnrichmentConfig { } } -/// Risk scoring configuration +/// Risk scoring configuration. Every country list, weight, score and level band +/// below is a default a deployment can override. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(default)] pub struct RiskScoringConfig { /// Enable risk scoring (requires at least `GeoIP` or reputation to be useful) pub enabled: bool, - /// Risk preset to use for country risk tables - /// - /// Options: "global" (default), "`us_enterprise`", "`eu_enterprise`", - /// "`apac_enterprise`", "`high_security`" + /// The entry in `presets` whose country lists apply. A name `presets` does + /// not hold is refused at load. pub preset: String, + + /// Country lists by preset name. The loader ships `global`, + /// `us_enterprise`, `eu_enterprise`, `apac_enterprise` and `high_security`. + /// An entry set in config replaces the shipped entry of that name, or adds + /// a new preset, and leaves the others as shipped. + #[serde(deserialize_with = "over_shipped_presets")] + pub presets: BTreeMap, + + /// Weight of each component in the composite score. + pub weights: RiskWeights, + + /// Score each signal gives its component, 0-100. + pub scores: RiskScores, + + /// Lowest composite score of each level above `minimal`. + pub levels: RiskLevelBands, } impl Default for RiskScoringConfig { @@ -383,6 +398,254 @@ impl Default for RiskScoringConfig { Self { enabled: false, preset: "global".into(), + presets: shipped_risk_presets(), + weights: RiskWeights::default(), + scores: RiskScores::default(), + levels: RiskLevelBands::default(), + } + } +} + +impl RiskScoringConfig { + /// The country lists `preset` names. + #[must_use] + pub fn countries(&self) -> Option<&RiskCountries> { + self.presets.get(&self.preset) + } + + /// Refuse a preset `presets` does not hold, a score past 100, and level + /// bands that do not rise strictly within 1-100. + /// + /// # Errors + /// + /// The offending key under `enrichment.risk_scoring`, and the valid preset + /// names for an unknown preset. + pub fn validate(&self) -> std::result::Result<(), String> { + if self.countries().is_none() { + let valid: Vec<&str> = self.presets.keys().map(String::as_str).collect(); + return Err(format!( + "enrichment.risk_scoring.preset '{}' is not a preset -- valid: {}", + self.preset, + valid.join(", ") + )); + } + if let Some((name, score)) = self.scores.named().into_iter().find(|(_, s)| *s > 100) { + return Err(format!( + "enrichment.risk_scoring.scores.{name} is {score} -- a score is 0-100" + )); + } + let l = &self.levels; + if !(0 < l.low && l.low < l.medium && l.medium < l.high && l.high < l.critical) + || l.critical > 100 + { + return Err(format!( + "enrichment.risk_scoring.levels must rise strictly within 1-100 \ + (low {}, medium {}, high {}, critical {})", + l.low, l.medium, l.high, l.critical + )); + } + Ok(()) + } +} + +/// The high- and medium-risk country codes of one preset (ISO 3166-1 alpha-2). +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] +#[serde(default)] +pub struct RiskCountries { + /// Countries scored `scores.high_risk_country`. + pub high: Vec, + /// Countries scored `scores.medium_risk_country`. + pub medium: Vec, +} + +/// The presets the loader ships, as `(name, high-risk, medium-risk)`. +const SHIPPED_RISK_PRESETS: [(&str, &[&str], &[&str]); 5] = [ + // OFAC sanctions + ("global", &["KP", "IR", "SY"], &["RU", "CN", "VE", "CU"]), + ( + "us_enterprise", + &["CN", "RU", "KP", "IR", "SY", "CU", "VE"], + &["UA", "BY", "PK", "NG", "VN"], + ), + ( + "eu_enterprise", + &["RU", "KP", "IR", "SY", "BY"], + &["CN", "UA", "PK"], + ), + ("apac_enterprise", &["KP", "AF", "MM"], &["RU", "IR"]), + ( + "high_security", + &["CN", "RU", "KP", "IR", "SY", "CU", "VE", "BY", "MM"], + &["UA", "PK", "NG", "VN", "BD", "NP"], + ), +]; + +fn shipped_risk_presets() -> BTreeMap { + let owned = + |codes: &[&str]| -> Vec { codes.iter().map(|c| (*c).to_string()).collect() }; + SHIPPED_RISK_PRESETS + .iter() + .map(|(name, high, medium)| { + let countries = RiskCountries { + high: owned(high), + medium: owned(medium), + }; + ((*name).to_string(), countries) + }) + .collect() +} + +/// Lay the configured presets over the shipped ones. +fn over_shipped_presets<'de, D>( + deserializer: D, +) -> std::result::Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + let mut presets = shipped_risk_presets(); + presets.extend(BTreeMap::::deserialize( + deserializer, + )?); + Ok(presets) +} + +/// Weight of each component in the composite score. All zero scores 0. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] +#[serde(default)] +pub struct RiskWeights { + /// Geographic component. + pub geo: u8, + /// Reputation component. + pub reputation: u8, + /// Privacy (anonymiser) component. + pub privacy: u8, + /// Threat component. + pub threat: u8, +} + +impl Default for RiskWeights { + fn default() -> Self { + Self { + geo: 15, + reputation: 25, + privacy: 25, + threat: 35, + } + } +} + +/// Score each signal gives its component. Reputation and threat signals add +/// up, capped at 100. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] +#[serde(default)] +pub struct RiskScores { + /// Geo: a public IP with no country. + pub unknown_country: u8, + /// Geo: a country in the preset's `high` list. + pub high_risk_country: u8, + /// Geo: a country in the preset's `medium` list. + pub medium_risk_country: u8, + /// Geo: any other country. + pub other_country: u8, + /// Reputation: VPN exit. + pub vpn: u8, + /// Reputation: open proxy. + pub proxy: u8, + /// Reputation: Tor exit. + pub tor: u8, + /// Reputation: datacenter range. + pub datacenter: u8, + /// Reputation: private relay. + pub relay: u8, + /// Privacy: one anonymiser among VPN, proxy, Tor and relay. + pub one_anonymiser: u8, + /// Privacy: two anonymisers. + pub two_anonymisers: u8, + /// Privacy: three or more anonymisers. + pub three_anonymisers: u8, + /// Threat: botnet member. + pub botnet: u8, + /// Threat: known malicious. + pub malicious: u8, + /// Threat: spam source. + pub spam: u8, + /// Threat: scanner. + pub scanner: u8, + /// Threat: the external abuse score at or above which the row carries the + /// `high_abuse_score` factor. The abuse score itself is added as given. + pub high_abuse_score: u8, +} + +impl Default for RiskScores { + fn default() -> Self { + Self { + unknown_country: 35, + high_risk_country: 90, + medium_risk_country: 60, + other_country: 15, + vpn: 50, + proxy: 60, + tor: 80, + datacenter: 40, + relay: 45, + one_anonymiser: 40, + two_anonymisers: 70, + three_anonymisers: 95, + botnet: 95, + malicious: 90, + spam: 70, + scanner: 60, + high_abuse_score: 80, + } + } +} + +impl RiskScores { + /// Every score with its config key. + fn named(&self) -> [(&'static str, u8); 17] { + [ + ("unknown_country", self.unknown_country), + ("high_risk_country", self.high_risk_country), + ("medium_risk_country", self.medium_risk_country), + ("other_country", self.other_country), + ("vpn", self.vpn), + ("proxy", self.proxy), + ("tor", self.tor), + ("datacenter", self.datacenter), + ("relay", self.relay), + ("one_anonymiser", self.one_anonymiser), + ("two_anonymisers", self.two_anonymisers), + ("three_anonymisers", self.three_anonymisers), + ("botnet", self.botnet), + ("malicious", self.malicious), + ("spam", self.spam), + ("scanner", self.scanner), + ("high_abuse_score", self.high_abuse_score), + ] + } +} + +/// Lowest composite score of each risk level. Below `low` is `minimal`. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] +#[serde(default)] +pub struct RiskLevelBands { + /// Lowest `low` score. + pub low: u8, + /// Lowest `medium` score. + pub medium: u8, + /// Lowest `high` score. + pub high: u8, + /// Lowest `critical` score. + pub critical: u8, +} + +impl Default for RiskLevelBands { + fn default() -> Self { + Self { + low: 20, + medium: 40, + high: 60, + critical: 80, } } } @@ -1047,98 +1310,6 @@ pub struct TableCaptureConfig { pub mode: CaptureMode, } -// ============================================================================ -// Type Coercion Configuration -// ============================================================================ - -/// Null handling strategy -#[derive( - Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default, schemars::JsonSchema, -)] -#[serde(rename_all = "lowercase")] -pub enum NullHandling { - /// Substitute type-appropriate default value (recommended) - #[default] - Default, - /// Return error for null in non-nullable column - Error, - /// Pass null through (may cause `ClickHouse` errors) - Passthrough, -} - -/// Type coercion configuration. No encoder reads it, so `Config::validate` -/// refuses any field set away from its default. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] -#[serde(default)] -pub struct CoercionConfig { - /// Map custom type names to base coercer categories - /// e.g., {"MyCustomInt": "Int", "SpecialString": "String"} - pub type_mappings: std::collections::HashMap, - - /// Fallback coercer for unknown types (default: "String") - pub unknown_type_fallback: String, - - /// How to handle null values in non-nullable columns - pub null_handling: NullHandling, - - /// Strings to recognise as null (case-sensitive) - pub null_strings: Vec, - - /// Default timezone for naive timestamps (IANA name or +HH:MM) - pub default_timezone: String, - - /// Event fields to check for timezone info - pub timezone_fields: Vec, - - /// Convert arrays to JSON strings if true - pub array_to_json: bool, - - /// Strict mode: fail on any coercion error - pub strict: bool, -} - -impl Default for CoercionConfig { - fn default() -> Self { - Self { - type_mappings: std::collections::HashMap::new(), - unknown_type_fallback: "String".to_string(), - null_handling: NullHandling::Default, - null_strings: vec![ - "null".to_string(), - "NULL".to_string(), - "Null".to_string(), - "None".to_string(), - "nil".to_string(), - "undefined".to_string(), - "\\N".to_string(), - "".to_string(), - "NA".to_string(), - "N/A".to_string(), - "n/a".to_string(), - "NaN".to_string(), - ], - default_timezone: "UTC".to_string(), - timezone_fields: vec!["tags_collector_timezone".to_string()], - array_to_json: true, - strict: false, - } - } -} - -impl CoercionConfig { - /// Check if a string value should be treated as null - pub fn is_null_string(&self, value: &str) -> bool { - value.is_empty() || self.null_strings.iter().any(|s| s == value) - } - - /// Get the coercer category for a type, checking custom mappings first - pub fn get_coercer_category(&self, type_name: &str) -> &str { - self.type_mappings - .get(type_name) - .map_or(&self.unknown_type_fallback, std::string::String::as_str) - } -} - // ============================================================================ // Schema Configuration // ============================================================================ diff --git a/src/enrich/risk.rs b/src/enrich/risk.rs index 5631562..a3eb083 100644 --- a/src/enrich/risk.rs +++ b/src/enrich/risk.rs @@ -12,24 +12,27 @@ //! ## Hot Path Optimizations //! //! 1. **Integer math only**: All scores are u8 (0-100), no floating point -//! 2. **Preset configurations**: Pre-computed country/region risk tables -//! 3. **Short-circuit evaluation**: Early exit if risk is already critical -//! 4. **Schema-aware output**: Only compute needed risk components +//! 2. **Pre-computed country sets**: the configured preset's lists, built once +//! 3. **Schema-aware output**: Only compute needed risk components +//! +//! Every country list, weight, score and level band comes from +//! [`RiskScoringConfig`]. use std::collections::HashMap; use super::geoip::GeoIpResult; use super::reputation::ReputationResult; +use crate::config::{RiskLevelBands, RiskScores, RiskScoringConfig, RiskWeights}; /// Risk level category (low cardinality for `ClickHouse`) #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum RiskLevel { #[default] - Minimal, // 0-19 - Low, // 20-39 - Medium, // 40-59 - High, // 60-79 - Critical, // 80-100 + Minimal, + Low, + Medium, + High, + Critical, } impl RiskLevel { @@ -43,16 +46,20 @@ impl RiskLevel { } } - /// Derive risk level from score (0-100) + /// The level `score` falls in under `bands`. #[inline] - pub fn from_score(score: u8) -> Self { - match score { - 0..=19 => RiskLevel::Minimal, - 20..=39 => RiskLevel::Low, - 40..=59 => RiskLevel::Medium, - 60..=79 => RiskLevel::High, - 80..=100 => RiskLevel::Critical, - _ => RiskLevel::Critical, // Safety for > 100 + #[must_use] + pub fn from_score(score: u8, bands: &RiskLevelBands) -> Self { + if score >= bands.critical { + RiskLevel::Critical + } else if score >= bands.high { + RiskLevel::High + } else if score >= bands.medium { + RiskLevel::Medium + } else if score >= bands.low { + RiskLevel::Low + } else { + RiskLevel::Minimal } } } @@ -184,151 +191,36 @@ impl RiskOutput { } } -/// Risk weight configuration -#[derive(Debug, Clone)] -pub struct RiskWeights { - /// Geographic risk weight (0-100) - pub geo_weight: u8, - /// Reputation risk weight (0-100) - pub reputation_weight: u8, - /// Privacy risk weight (0-100) - pub privacy_weight: u8, - /// Threat risk weight (0-100) - pub threat_weight: u8, -} - -impl Default for RiskWeights { - fn default() -> Self { - Self { - geo_weight: 15, - reputation_weight: 25, - privacy_weight: 25, - threat_weight: 35, - } - } -} - -/// Preset risk configurations for different use cases -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum RiskPreset { - /// US enterprise - high risk: CN, RU, KP, IR - UsEnterprise, - /// EU enterprise - GDPR-aware - EuEnterprise, - /// APAC enterprise - regional focus - ApacEnterprise, - /// Global - balanced, sanctions-focused - Global, - /// High security - financial/government - HighSecurity, -} - -impl RiskPreset { - /// Get high-risk country codes for this preset - pub fn high_risk_countries(&self) -> &'static [&'static str] { - match self { - RiskPreset::UsEnterprise => &["CN", "RU", "KP", "IR", "SY", "CU", "VE"], - RiskPreset::EuEnterprise => &["RU", "KP", "IR", "SY", "BY"], - RiskPreset::ApacEnterprise => &["KP", "AF", "MM"], - RiskPreset::Global => &["KP", "IR", "SY"], // OFAC sanctions - RiskPreset::HighSecurity => &["CN", "RU", "KP", "IR", "SY", "CU", "VE", "BY", "MM"], - } - } - - /// Get medium-risk country codes for this preset - pub fn medium_risk_countries(&self) -> &'static [&'static str] { - match self { - RiskPreset::UsEnterprise => &["UA", "BY", "PK", "NG", "VN"], - RiskPreset::EuEnterprise => &["CN", "UA", "PK"], - RiskPreset::ApacEnterprise => &["RU", "IR"], - RiskPreset::Global => &["RU", "CN", "VE", "CU"], - RiskPreset::HighSecurity => &["UA", "PK", "NG", "VN", "BD", "NP"], - } - } -} - -/// Risk scorer configuration -#[derive(Debug, Clone)] -pub struct RiskConfig { - pub weights: RiskWeights, - pub preset: Option, - pub high_risk_countries: Vec, - pub medium_risk_countries: Vec, -} - -impl Default for RiskConfig { - fn default() -> Self { - Self { - weights: RiskWeights::default(), - preset: Some(RiskPreset::Global), - high_risk_countries: Vec::new(), - medium_risk_countries: Vec::new(), - } - } -} - -impl RiskConfig { - /// Create config from a preset - pub fn from_preset(preset: RiskPreset) -> Self { - Self { - weights: RiskWeights::default(), - preset: Some(preset), - high_risk_countries: Vec::new(), - medium_risk_countries: Vec::new(), - } - } -} - /// Risk scorer pub struct RiskScorer { - config: RiskConfig, - /// Pre-computed high-risk country set for O(1) lookup + weights: RiskWeights, + scores: RiskScores, + levels: RiskLevelBands, + /// The preset's high-risk countries, for O(1) lookup high_risk_set: rustc_hash::FxHashSet, - /// Pre-computed medium-risk country set for O(1) lookup + /// The preset's medium-risk countries, for O(1) lookup medium_risk_set: rustc_hash::FxHashSet, } impl RiskScorer { - /// Create a new risk scorer with default config - pub fn new() -> Self { - Self::with_config(RiskConfig::default()) - } - - /// Create with specific config - pub fn with_config(config: RiskConfig) -> Self { - let mut high_risk_set = rustc_hash::FxHashSet::default(); - let mut medium_risk_set = rustc_hash::FxHashSet::default(); - - // Add preset countries - if let Some(preset) = config.preset { - for &code in preset.high_risk_countries() { - high_risk_set.insert(code.to_string()); - } - for &code in preset.medium_risk_countries() { - medium_risk_set.insert(code.to_string()); - } - } - - // Add custom countries - for code in &config.high_risk_countries { - high_risk_set.insert(code.clone()); - } - for code in &config.medium_risk_countries { - medium_risk_set.insert(code.clone()); - } - + /// A scorer for `config`. A preset `config.presets` does not hold scores + /// every country as `other_country`. [`RiskScoringConfig::validate`] + /// refuses one at load. + #[must_use] + pub fn new(config: &RiskScoringConfig) -> Self { + let countries = config.countries(); + let set = |codes: Option<&Vec>| -> rustc_hash::FxHashSet { + codes.into_iter().flatten().cloned().collect() + }; Self { - config, - high_risk_set, - medium_risk_set, + weights: config.weights.clone(), + scores: config.scores.clone(), + levels: config.levels.clone(), + high_risk_set: set(countries.map(|c| &c.high)), + medium_risk_set: set(countries.map(|c| &c.medium)), } } - /// Create from a preset - pub fn from_preset(preset: RiskPreset) -> Self { - Self::with_config(RiskConfig::from_preset(preset)) - } - /// Calculate risk score from input pub fn score(&self, input: &RiskInput) -> RiskOutput { let mut output = RiskOutput::default(); @@ -341,22 +233,22 @@ impl RiskScorer { output.threat_risk_score = self.calculate_threat_risk(input, &mut output.risk_factors); // Weighted composite score - let weights = &self.config.weights; - let total_weight = weights.geo_weight as u32 - + weights.reputation_weight as u32 - + weights.privacy_weight as u32 - + weights.threat_weight as u32; + let weights = &self.weights; + let total_weight = weights.geo as u32 + + weights.reputation as u32 + + weights.privacy as u32 + + weights.threat as u32; - let weighted_sum = (output.geo_risk_score as u32 * weights.geo_weight as u32) - + (output.reputation_risk_score as u32 * weights.reputation_weight as u32) - + (output.privacy_risk_score as u32 * weights.privacy_weight as u32) - + (output.threat_risk_score as u32 * weights.threat_weight as u32); + let weighted_sum = (output.geo_risk_score as u32 * weights.geo as u32) + + (output.reputation_risk_score as u32 * weights.reputation as u32) + + (output.privacy_risk_score as u32 * weights.privacy as u32) + + (output.threat_risk_score as u32 * weights.threat as u32); if let Some(divided) = weighted_sum.checked_div(total_weight) { output.risk_score = divided.min(100) as u8; } - output.risk_level = RiskLevel::from_score(output.risk_score); + output.risk_level = RiskLevel::from_score(output.risk_score, &self.levels); output } @@ -368,58 +260,35 @@ impl RiskScorer { return 0; } - // Unknown country = moderate risk let Some(country) = input.country_code else { factors.push("unknown_country"); - return 35; + return self.scores.unknown_country; }; - // High-risk country if self.high_risk_set.contains(country) { factors.push("high_risk_country"); - return 90; + return self.scores.high_risk_country; } - // Medium-risk country if self.medium_risk_set.contains(country) { factors.push("medium_risk_country"); - return 60; + return self.scores.medium_risk_country; } - // Normal country - low base risk - 15 + self.scores.other_country } /// Calculate reputation risk score fn calculate_reputation_risk(&self, input: &RiskInput, factors: &mut Vec<&'static str>) -> u8 { - let mut score: u8 = 0; - - if input.is_vpn { - score = score.saturating_add(50); - factors.push("vpn_detected"); - } - - if input.is_proxy { - score = score.saturating_add(60); - factors.push("proxy_detected"); - } - - if input.is_tor { - score = score.saturating_add(80); - factors.push("tor_detected"); - } - - if input.is_datacenter { - score = score.saturating_add(40); - factors.push("datacenter_ip"); - } - - if input.is_relay { - score = score.saturating_add(45); - factors.push("private_relay"); - } - - score.min(100) + let s = &self.scores; + let signals = [ + (input.is_vpn, s.vpn, "vpn_detected"), + (input.is_proxy, s.proxy, "proxy_detected"), + (input.is_tor, s.tor, "tor_detected"), + (input.is_datacenter, s.datacenter, "datacenter_ip"), + (input.is_relay, s.relay, "private_relay"), + ]; + add_signals(&signals, factors) } /// Calculate privacy risk score (anonymization attempts) @@ -441,57 +310,54 @@ impl RiskScorer { match anonymizer_count { 0 => 0, - 1 => 40, + 1 => self.scores.one_anonymiser, 2 => { factors.push("multi_layer_anonymization"); - 70 + self.scores.two_anonymisers } _ => { factors.push("extensive_anonymization"); - 95 + self.scores.three_anonymisers } } } /// Calculate threat risk score fn calculate_threat_risk(&self, input: &RiskInput, factors: &mut Vec<&'static str>) -> u8 { - let mut score: u8 = 0; - - if input.is_botnet { - score = score.saturating_add(95); - factors.push("botnet_ip"); - } - - if input.is_malicious { - score = score.saturating_add(90); - factors.push("malicious_ip"); - } - - if input.is_spam { - score = score.saturating_add(70); - factors.push("spam_source"); - } - - if input.is_scanner { - score = score.saturating_add(60); - factors.push("scanner_ip"); - } + let s = &self.scores; + let signals = [ + (input.is_botnet, s.botnet, "botnet_ip"), + (input.is_malicious, s.malicious, "malicious_ip"), + (input.is_spam, s.spam, "spam_source"), + (input.is_scanner, s.scanner, "scanner_ip"), + ]; + let score = add_signals(&signals, factors); // Include abuse score from external sources - if input.abuse_score > 0 { - score = score.saturating_add(input.abuse_score); - if input.abuse_score >= 80 { - factors.push("high_abuse_score"); - } + if input.abuse_score > 0 && input.abuse_score >= s.high_abuse_score { + factors.push("high_abuse_score"); } + score.saturating_add(input.abuse_score).min(100) + } +} - score.min(100) +/// The sum of the scores of the signals that are set, capped at 100, with each +/// one's factor pushed. +fn add_signals(signals: &[(bool, u8, &'static str)], factors: &mut Vec<&'static str>) -> u8 { + let mut score: u8 = 0; + for &(set, points, factor) in signals { + if set { + score = score.saturating_add(points); + factors.push(factor); + } } + score.min(100) } impl Default for RiskScorer { + /// A scorer for the shipped default config. fn default() -> Self { - Self::new() + Self::new(&RiskScoringConfig::default()) } } @@ -499,23 +365,30 @@ impl Default for RiskScorer { mod tests { use super::*; - #[test] - fn test_risk_level_from_score() { - assert_eq!(RiskLevel::from_score(0), RiskLevel::Minimal); - assert_eq!(RiskLevel::from_score(19), RiskLevel::Minimal); - assert_eq!(RiskLevel::from_score(20), RiskLevel::Low); - assert_eq!(RiskLevel::from_score(39), RiskLevel::Low); - assert_eq!(RiskLevel::from_score(40), RiskLevel::Medium); - assert_eq!(RiskLevel::from_score(59), RiskLevel::Medium); - assert_eq!(RiskLevel::from_score(60), RiskLevel::High); - assert_eq!(RiskLevel::from_score(79), RiskLevel::High); - assert_eq!(RiskLevel::from_score(80), RiskLevel::Critical); - assert_eq!(RiskLevel::from_score(100), RiskLevel::Critical); + /// A scorer for the shipped preset `name`. + fn scorer_for(name: &str) -> RiskScorer { + RiskScorer::new(&RiskScoringConfig { + preset: name.to_string(), + ..RiskScoringConfig::default() + }) + } + + /// A scorer for the default config with `weights`. + fn weighted(geo: u8, reputation: u8, privacy: u8, threat: u8) -> RiskScorer { + RiskScorer::new(&RiskScoringConfig { + weights: RiskWeights { + geo, + reputation, + privacy, + threat, + }, + ..RiskScoringConfig::default() + }) } #[test] fn test_clean_input() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput::default(); let output = scorer.score(&input); @@ -527,7 +400,7 @@ mod tests { #[test] fn test_high_risk_country() { - let scorer = RiskScorer::from_preset(RiskPreset::UsEnterprise); + let scorer = scorer_for("us_enterprise"); let input = RiskInput { country_code: Some("CN"), ..Default::default() @@ -540,7 +413,7 @@ mod tests { #[test] fn test_tor_detection() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { is_tor: true, ..Default::default() @@ -554,7 +427,7 @@ mod tests { #[test] fn test_botnet_critical() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { is_botnet: true, is_malicious: true, @@ -571,7 +444,7 @@ mod tests { #[test] fn test_multi_layer_anonymization() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { is_vpn: true, is_tor: true, @@ -585,7 +458,7 @@ mod tests { #[test] fn test_private_ip_no_geo_risk() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { is_private: true, ..Default::default() @@ -621,7 +494,7 @@ mod tests { #[test] fn test_preset_us_enterprise_high_risk_countries() { - let scorer = RiskScorer::from_preset(RiskPreset::UsEnterprise); + let scorer = scorer_for("us_enterprise"); // CN is high-risk for UsEnterprise let input = RiskInput { @@ -652,7 +525,7 @@ mod tests { #[test] fn test_preset_eu_enterprise() { - let scorer = RiskScorer::from_preset(RiskPreset::EuEnterprise); + let scorer = scorer_for("eu_enterprise"); // RU is high-risk for EuEnterprise let input = RiskInput { @@ -673,7 +546,7 @@ mod tests { #[test] fn test_preset_apac_enterprise() { - let scorer = RiskScorer::from_preset(RiskPreset::ApacEnterprise); + let scorer = scorer_for("apac_enterprise"); // KP is high-risk for APAC let input = RiskInput { @@ -702,7 +575,7 @@ mod tests { #[test] fn test_preset_global() { - let scorer = RiskScorer::from_preset(RiskPreset::Global); + let scorer = scorer_for("global"); // KP, IR, SY are high-risk (OFAC sanctions) for code in &["KP", "IR", "SY"] { @@ -728,7 +601,7 @@ mod tests { #[test] fn test_preset_high_security() { - let scorer = RiskScorer::from_preset(RiskPreset::HighSecurity); + let scorer = scorer_for("high_security"); // HighSecurity has the largest high-risk set for code in &["CN", "RU", "KP", "IR", "SY", "CU", "VE", "BY", "MM"] { @@ -763,7 +636,7 @@ mod tests { #[test] fn test_reputation_risk_saturates_at_100() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); // VPN (50) + Proxy (60) + Tor (80) + Datacenter (40) + Relay (45) = 275, capped at 100 let input = RiskInput { is_vpn: true, @@ -782,7 +655,7 @@ mod tests { #[test] fn test_threat_risk_saturates_at_100() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); // Botnet (95) + Malicious (90) + Spam (70) + Scanner (60) = 315, capped at 100 let input = RiskInput { is_botnet: true, @@ -800,7 +673,7 @@ mod tests { #[test] fn test_threat_risk_with_high_abuse_score() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { abuse_score: 95, is_botnet: true, // 95 + 95 + 90 (malicious implied by botnet? no) = 95 + 95 = 190 -> cap 100 @@ -818,7 +691,7 @@ mod tests { #[test] fn test_completely_clean_private_ip() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { country_code: Some("AU"), is_private: true, @@ -840,7 +713,7 @@ mod tests { #[test] fn test_clean_known_country_minimal_risk() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { country_code: Some("US"), ..Default::default() @@ -862,7 +735,7 @@ mod tests { #[test] fn test_risk_level_minimal() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { country_code: Some("AU"), is_private: true, @@ -875,7 +748,7 @@ mod tests { #[test] fn test_risk_level_low() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); // Unknown country (35 geo) + single VPN (50 rep, 40 priv) // Weighted: (35*15 + 50*25 + 40*25 + 0*35) / 100 = (525 + 1250 + 1000) / 100 = 27.75 -> 27 let input = RiskInput { @@ -893,7 +766,7 @@ mod tests { #[test] fn test_risk_level_medium() { - let scorer = RiskScorer::from_preset(RiskPreset::UsEnterprise); + let scorer = scorer_for("us_enterprise"); // Weighted: (60 geo * 15 + 90 rep * 25 + 40 priv * 25 + 0 threat * 35) / 100 // = (900 + 2250 + 1000) / 100 = 41 -> Medium let input = RiskInput { @@ -917,7 +790,7 @@ mod tests { #[test] fn test_risk_level_high() { - let scorer = RiskScorer::from_preset(RiskPreset::UsEnterprise); + let scorer = scorer_for("us_enterprise"); // High risk country (90 geo) + Tor (80 rep, 40 priv) + botnet (95 threat) // Weighted: (90*15 + 80*25 + 40*25 + 100*35) / 100 = (1350+2000+1000+3500)/100 = 78.5 -> 78 let input = RiskInput { @@ -938,7 +811,7 @@ mod tests { #[test] fn test_risk_level_critical() { - let scorer = RiskScorer::from_preset(RiskPreset::HighSecurity); + let scorer = scorer_for("high_security"); // All signals maxed out let input = RiskInput { country_code: Some("KP"), // 90 geo @@ -967,19 +840,7 @@ mod tests { #[test] fn test_custom_weights_threat_only() { - // 100% threat weight, 0 everything else - let config = RiskConfig { - weights: RiskWeights { - geo_weight: 0, - reputation_weight: 0, - privacy_weight: 0, - threat_weight: 100, - }, - preset: Some(RiskPreset::Global), - high_risk_countries: Vec::new(), - medium_risk_countries: Vec::new(), - }; - let scorer = RiskScorer::with_config(config); + let scorer = weighted(0, 0, 0, 100); let input = RiskInput { country_code: Some("KP"), // Would be 90 geo, but weight=0 @@ -999,18 +860,7 @@ mod tests { #[test] fn test_custom_weights_geo_only() { - let config = RiskConfig { - weights: RiskWeights { - geo_weight: 100, - reputation_weight: 0, - privacy_weight: 0, - threat_weight: 0, - }, - preset: Some(RiskPreset::Global), - high_risk_countries: Vec::new(), - medium_risk_countries: Vec::new(), - }; - let scorer = RiskScorer::with_config(config); + let scorer = weighted(100, 0, 0, 0); let input = RiskInput { country_code: Some("KP"), @@ -1024,19 +874,7 @@ mod tests { #[test] fn test_zero_total_weight() { - // Edge case: all weights zero - let config = RiskConfig { - weights: RiskWeights { - geo_weight: 0, - reputation_weight: 0, - privacy_weight: 0, - threat_weight: 0, - }, - preset: None, - high_risk_countries: Vec::new(), - medium_risk_countries: Vec::new(), - }; - let scorer = RiskScorer::with_config(config); + let scorer = weighted(0, 0, 0, 0); let input = RiskInput { country_code: Some("KP"), @@ -1055,7 +893,7 @@ mod tests { #[test] fn test_unknown_country_code() { - let scorer = RiskScorer::from_preset(RiskPreset::UsEnterprise); + let scorer = scorer_for("us_enterprise"); let input = RiskInput { country_code: Some("ZZ"), // Not a real country ..Default::default() @@ -1067,7 +905,7 @@ mod tests { #[test] fn test_empty_string_country_code() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { country_code: Some(""), ..Default::default() @@ -1079,7 +917,7 @@ mod tests { #[test] fn test_no_country_code() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { country_code: None, ..Default::default() @@ -1095,7 +933,7 @@ mod tests { #[test] fn test_tor_vpn_botnet_high_risk_country_critical() { - let scorer = RiskScorer::from_preset(RiskPreset::UsEnterprise); + let scorer = scorer_for("us_enterprise"); let input = RiskInput { country_code: Some("IR"), is_tor: true, @@ -1141,7 +979,7 @@ mod tests { #[test] fn test_privacy_risk_zero_anonymizers() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput::default(); let output = scorer.score(&input); assert_eq!(output.privacy_risk_score, 0); @@ -1149,7 +987,7 @@ mod tests { #[test] fn test_privacy_risk_single_anonymizer() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { is_vpn: true, ..Default::default() @@ -1160,7 +998,7 @@ mod tests { #[test] fn test_privacy_risk_two_anonymizers() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { is_vpn: true, is_proxy: true, @@ -1173,7 +1011,7 @@ mod tests { #[test] fn test_privacy_risk_three_plus_anonymizers() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { is_vpn: true, is_proxy: true, @@ -1187,7 +1025,7 @@ mod tests { #[test] fn test_privacy_risk_all_four_anonymizers() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { is_vpn: true, is_proxy: true, @@ -1299,7 +1137,7 @@ mod tests { #[test] fn test_abuse_score_below_threshold_no_factor() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { abuse_score: 79, // Below 80 threshold ..Default::default() @@ -1314,7 +1152,7 @@ mod tests { #[test] fn test_abuse_score_at_threshold() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { abuse_score: 80, ..Default::default() @@ -1325,7 +1163,7 @@ mod tests { #[test] fn test_abuse_score_zero_no_contribution() { - let scorer = RiskScorer::new(); + let scorer = RiskScorer::default(); let input = RiskInput { abuse_score: 0, ..Default::default() @@ -1335,84 +1173,81 @@ mod tests { } // ======================================================================== - // Custom countries in config + // Configured lists, scores and bands // ======================================================================== + /// Every value the scorer uses comes from config: a deployment's own + /// preset, scores and bands change the output. #[test] - fn test_custom_high_risk_countries() { - let config = RiskConfig { - weights: RiskWeights::default(), - preset: None, // No preset - high_risk_countries: vec!["XX".to_string(), "YY".to_string()], - medium_risk_countries: vec!["ZZ".to_string()], - }; - let scorer = RiskScorer::with_config(config); - - let input = RiskInput { - country_code: Some("XX"), - ..Default::default() - }; - let output = scorer.score(&input); - assert_eq!(output.geo_risk_score, 90); - - let input = RiskInput { - country_code: Some("ZZ"), - ..Default::default() - }; - let output = scorer.score(&input); - assert_eq!(output.geo_risk_score, 60); - } - - #[test] - fn test_custom_countries_merged_with_preset() { - let config = RiskConfig { - weights: RiskWeights::default(), - preset: Some(RiskPreset::Global), - high_risk_countries: vec!["XX".to_string()], - medium_risk_countries: Vec::new(), - }; - let scorer = RiskScorer::with_config(config); - - // XX from custom - let input = RiskInput { - country_code: Some("XX"), - ..Default::default() - }; - let output = scorer.score(&input); - assert_eq!(output.geo_risk_score, 90); - - // KP from preset - let input = RiskInput { - country_code: Some("KP"), - ..Default::default() + fn configured_presets_scores_and_bands_drive_the_score() { + let mut config = RiskScoringConfig { + preset: "mine".to_string(), + ..RiskScoringConfig::default() }; - let output = scorer.score(&input); - assert_eq!(output.geo_risk_score, 90); + config.presets.insert( + "mine".to_string(), + crate::config::RiskCountries { + high: vec!["XX".to_string()], + medium: vec!["ZZ".to_string()], + }, + ); + config.scores.high_risk_country = 70; + config.scores.medium_risk_country = 30; + config.scores.vpn = 10; + config.levels.low = 5; + let scorer = RiskScorer::new(&config); + + // (country, vpn, geo score, reputation score, level) + let cases = [ + (Some("XX"), false, 70, 0, RiskLevel::Low), + (Some("ZZ"), false, 30, 0, RiskLevel::Minimal), + // KP is high risk in the shipped presets, not in this one. + (Some("KP"), false, 15, 0, RiskLevel::Minimal), + (Some("AU"), true, 15, 10, RiskLevel::Low), + ]; + for (country, is_vpn, geo, reputation, level) in cases { + let output = scorer.score(&RiskInput { + country_code: country, + is_vpn, + ..Default::default() + }); + assert_eq!(output.geo_risk_score, geo, "{country:?}"); + assert_eq!(output.reputation_risk_score, reputation, "{country:?}"); + assert_eq!(output.risk_level, level, "{country:?}"); + } } - // ======================================================================== - // Risk level boundary values (exhaustive) - // ======================================================================== - #[test] fn test_risk_level_boundary_values_exhaustive() { - // Test every boundary value - assert_eq!(RiskLevel::from_score(0), RiskLevel::Minimal); - assert_eq!(RiskLevel::from_score(1), RiskLevel::Minimal); - assert_eq!(RiskLevel::from_score(19), RiskLevel::Minimal); - assert_eq!(RiskLevel::from_score(20), RiskLevel::Low); - assert_eq!(RiskLevel::from_score(39), RiskLevel::Low); - assert_eq!(RiskLevel::from_score(40), RiskLevel::Medium); - assert_eq!(RiskLevel::from_score(59), RiskLevel::Medium); - assert_eq!(RiskLevel::from_score(60), RiskLevel::High); - assert_eq!(RiskLevel::from_score(79), RiskLevel::High); - assert_eq!(RiskLevel::from_score(80), RiskLevel::Critical); - assert_eq!(RiskLevel::from_score(99), RiskLevel::Critical); - assert_eq!(RiskLevel::from_score(100), RiskLevel::Critical); - - // Safety for values > 100 - assert_eq!(RiskLevel::from_score(101), RiskLevel::Critical); - assert_eq!(RiskLevel::from_score(255), RiskLevel::Critical); + let shipped = RiskLevelBands::default(); + let custom = RiskLevelBands { + low: 10, + medium: 50, + high: 70, + critical: 95, + }; + // (bands, score, level) + let cases = [ + (&shipped, 0, RiskLevel::Minimal), + (&shipped, 19, RiskLevel::Minimal), + (&shipped, 20, RiskLevel::Low), + (&shipped, 39, RiskLevel::Low), + (&shipped, 40, RiskLevel::Medium), + (&shipped, 59, RiskLevel::Medium), + (&shipped, 60, RiskLevel::High), + (&shipped, 79, RiskLevel::High), + (&shipped, 80, RiskLevel::Critical), + (&shipped, 100, RiskLevel::Critical), + (&shipped, 255, RiskLevel::Critical), + (&custom, 9, RiskLevel::Minimal), + (&custom, 10, RiskLevel::Low), + (&custom, 69, RiskLevel::Medium), + (&custom, 94, RiskLevel::High), + (&custom, 95, RiskLevel::Critical), + ]; + for (bands, score, level) in cases { + assert_eq!(RiskLevel::from_score(score, bands), level, "{score}"); + } } #[test] diff --git a/src/kafka/transport.rs b/src/kafka/transport.rs index 9d3e701..e05f642 100644 --- a/src/kafka/transport.rs +++ b/src/kafka/transport.rs @@ -23,7 +23,7 @@ use std::time::Duration; use crate::Result; use crate::buffer::KafkaOffset; -use crate::config::{DlqConfig, KafkaConfig}; +use crate::config::{KafkaConfig, RoutingConfig}; use scalo::SelfRegulationGovernor; use scalo::memory::MemoryGuard; use scalo::transport::filter::FilteredDlqEntry; @@ -38,8 +38,13 @@ use tracing::{debug, trace}; use super::KafkaMessage; /// What auto-discovery subscribes to when neither `topics` nor `topic_regex` -/// is set. scalo reads an empty include list as every topic on the broker. -const LANDING_TOPIC_INCLUDE: &str = "_(land|load)$"; +/// is set: a topic ending in one of `suffixes`. `None` for no suffix, which +/// [`Config::validate`](crate::config::Config::validate) refuses because scalo +/// reads an empty include list as every topic on the broker. +fn suffix_include(suffixes: &[String]) -> Option { + let alternatives: Vec = suffixes.iter().map(|s| regex::escape(s)).collect(); + (!alternatives.is_empty()).then(|| format!("(?:{})$", alternatives.join("|"))) +} /// A transport that failed to start: a config scalo refused is a /// [`Config`](crate::Error::Config) error, which no retry fixes. @@ -206,9 +211,10 @@ impl TransportAdapter { /// Create a new transport adapter from local `KafkaConfig`. /// /// scalo's `KafkaTransport::new()` handles auto-discovery when - /// `config.topics` is empty -- no app-side resolver needed. `dlq` names the - /// topics auto-discovery must never subscribe to, and `dlq_load` the DLQ - /// topics it always does (see [`Self::convert_config`]). + /// `config.topics` is empty -- no app-side resolver needed. `routing` + /// names the topic suffixes auto-discovery subscribes to and the DLQ + /// topics it must never subscribe to, and `dlq_load` the DLQ topics it + /// always does (see [`Self::convert_config`]). /// /// When `governor` is `Some`, the self-regulation inbound brake is attached /// to the Kafka receiver: under memory pressure the consumer's ASSIGNED @@ -219,11 +225,11 @@ impl TransportAdapter { /// `None` (self-regulation disabled) construction is byte-identical to before. pub async fn new( config: &KafkaConfig, - dlq: &DlqConfig, + routing: &RoutingConfig, dlq_load: &[String], governor: Option<&SelfRegulationGovernor>, ) -> Result { - let transport_config = Self::convert_config(config, dlq, dlq_load); + let transport_config = Self::convert_config(config, routing, dlq_load); let transport = KafkaTransport::receiver(&transport_config) .await @@ -243,18 +249,20 @@ impl TransportAdapter { /// Convert local `KafkaConfig` to scalo `TransportKafkaConfig`. /// /// The topic filters only take effect when `topics` is empty and scalo - /// auto-discovers. The include is `topic_regex` when set, else - /// `*_land` / `*_load`. The loader's own DLQ topics are appended to - /// scalo's default excludes (`^__`, `_dlq$`), never replacing them. + /// auto-discovers. The include is `topic_regex` when set, else a topic + /// ending in one of `routing.topic_suffixes`. The loader's own DLQ topics + /// are appended to scalo's default excludes (`^__`, `_dlq$`), never + /// replacing them. /// /// Each `dlq_load` topic is subscribed too: appended to an explicit list, /// or included by name under discovery, with every exclude narrowed to let /// those names through and nothing else (see `exclude_except`). pub fn convert_config( config: &KafkaConfig, - dlq: &DlqConfig, + routing: &RoutingConfig, dlq_load: &[String], ) -> TransportKafkaConfig { + let dlq = &routing.dlq; let mut topics = config.topics.clone(); if !topics.is_empty() { for topic in dlq_load { @@ -282,12 +290,13 @@ impl TransportAdapter { }; // An empty regex would match every topic, the same failure as no include. - let include = config + transport_config.topic_include = config .topic_regex - .as_deref() + .clone() .filter(|r| !r.is_empty()) - .unwrap_or(LANDING_TOPIC_INCLUDE); - transport_config.topic_include = vec![include.to_string()]; + .or_else(|| suffix_include(&routing.topic_suffixes)) + .into_iter() + .collect(); transport_config .topic_exclude .extend(dlq.topic_exclude_patterns()); @@ -857,7 +866,7 @@ impl TransportBackend { } else { let adapter = TransportAdapter::new( &config.kafka, - &config.routing.dlq, + &config.routing, &config.dlq_load.topics, governor, ) @@ -1286,6 +1295,7 @@ mod discovery_tests { use scalo::transport::kafka::topic_resolver::{apply_suppression_rules, passes_filters}; use super::*; + use crate::config::DlqConfig; /// A Strimzi broker: two loader topics, two a `topic_regex` can pick, plus /// everything else that lives on it. @@ -1326,12 +1336,20 @@ mod discovery_tests { resolved } + /// The default routing with `dlq`. + fn with_dlq(dlq: DlqConfig) -> RoutingConfig { + RoutingConfig { + dlq, + ..RoutingConfig::default() + } + } + /// dfe-engine's default DLQ topic, which a `_land` include would match. - fn dlq_land() -> DlqConfig { - DlqConfig { + fn dlq_land() -> RoutingConfig { + with_dlq(DlqConfig { topic: "dlq_land".to_string(), ..DlqConfig::default() - } + }) } /// scalo's own defaults, which the loader's excludes are appended to. @@ -1339,13 +1357,33 @@ mod discovery_tests { TransportKafkaConfig::default().topic_exclude } + /// Discovery subscribes to the topics ending in a configured suffix, each + /// matched as a literal. #[test] - fn empty_topics_discover_only_land_and_load_topics() { - let cfg = TransportAdapter::convert_config(&KafkaConfig::default(), &dlq_land(), &[]); - - assert!(cfg.auto_discover, "an empty topic list auto-discovers"); - assert_eq!(cfg.topic_include, [LANDING_TOPIC_INCLUDE]); - assert_eq!(resolve(&cfg, BROKER_TOPICS), ["main_land", "x_load"]); + fn empty_topics_discover_the_configured_topic_suffixes() { + let broker: Vec<&str> = BROKER_TOPICS + .iter() + .copied() + .chain(["a_raw", "b.in", "bxin"]) + .collect(); + // (topic_suffixes, include, discovered) + let cases: [(&[&str], &str, &[&str]); 2] = [ + ( + &["_land", "_load"], + "(?:_land|_load)$", + &["main_land", "x_load"], + ), + (&["_raw", ".in"], r"(?:_raw|\.in)$", &["a_raw", "b.in"]), + ]; + for (suffixes, include, discovered) in cases { + let mut routing = dlq_land(); + routing.topic_suffixes = suffixes.iter().map(|s| (*s).to_string()).collect(); + let cfg = TransportAdapter::convert_config(&KafkaConfig::default(), &routing, &[]); + + assert!(cfg.auto_discover, "an empty topic list auto-discovers"); + assert_eq!(cfg.topic_include, [include]); + assert_eq!(resolve(&cfg, &broker), discovered, "{suffixes:?}"); + } } /// The fleet's five DLQ topics, plus a sixth `_dlq` topic nobody loads. @@ -1387,7 +1425,7 @@ mod discovery_tests { topic: own.to_string(), ..DlqConfig::default() }; - let cfg = TransportAdapter::convert_config(&kafka, &dlq, &loaded); + let cfg = TransportAdapter::convert_config(&kafka, &with_dlq(dlq), &loaded); let mut want: Vec<&str> = data.iter().chain(&DLQ_TOPICS[..5]).copied().collect(); want.sort_unstable(); @@ -1471,7 +1509,7 @@ mod discovery_tests { ), ]; for (dlq_load, want) in cases { - let cfg = TransportAdapter::convert_config(&kafka, &DlqConfig::default(), dlq_load); + let cfg = TransportAdapter::convert_config(&kafka, &RoutingConfig::default(), dlq_load); assert!(!cfg.auto_discover); assert_eq!(cfg.topics, want); } @@ -1525,7 +1563,7 @@ mod discovery_tests { topic_suffix: ".dlq".to_string(), ..DlqConfig::default() }; - let cfg = TransportAdapter::convert_config(&kafka, &dlq, &[]); + let cfg = TransportAdapter::convert_config(&kafka, &with_dlq(dlq), &[]); let resolved = resolve( &cfg, diff --git a/src/pipeline/enrichment.rs b/src/pipeline/enrichment.rs index 2ffd1db..29feab6 100644 --- a/src/pipeline/enrichment.rs +++ b/src/pipeline/enrichment.rs @@ -15,7 +15,7 @@ use tracing::{debug, info, warn}; use crate::config::Config; use crate::enrich::geoip::{GeoIpEnricher, GeoIpResult}; use crate::enrich::reputation::{ReputationEnricher, ReputationResult, ThreatSource, ThreatType}; -use crate::enrich::risk::{RiskInput, RiskOutput, RiskPreset, RiskScorer}; +use crate::enrich::risk::{RiskInput, RiskOutput, RiskScorer}; /// Active enrichment pipeline (GeoIP + reputation + risk scoring). /// @@ -88,15 +88,8 @@ impl EnrichmentPipeline { }; let risk = if config.enrichment.risk_scoring.enabled { - let preset = match config.enrichment.risk_scoring.preset.as_str() { - "us_enterprise" => RiskPreset::UsEnterprise, - "eu_enterprise" => RiskPreset::EuEnterprise, - "apac_enterprise" => RiskPreset::ApacEnterprise, - "high_security" => RiskPreset::HighSecurity, - _ => RiskPreset::Global, - }; info!(preset = %config.enrichment.risk_scoring.preset, "Risk scoring enabled"); - Some(RiskScorer::from_preset(preset)) + Some(RiskScorer::new(&config.enrichment.risk_scoring)) } else { None }; @@ -265,6 +258,14 @@ mod tests { use super::*; use serde_json::{Map, json}; + /// A scorer for the shipped preset `name`. + fn scorer_for(name: &str) -> RiskScorer { + RiskScorer::new(&crate::config::RiskScoringConfig { + preset: name.to_string(), + ..crate::config::RiskScoringConfig::default() + }) + } + // ======================================================================== // extract_enrich_ip // ======================================================================== @@ -585,7 +586,7 @@ mod tests { #[test] fn pipeline_with_risk_only_is_active() { - let scorer = RiskScorer::from_preset(RiskPreset::Global); + let scorer = scorer_for("global"); let pipeline = EnrichmentPipeline { ip_fields: vec!["src_ip".into()], geoip: None, @@ -618,7 +619,7 @@ mod tests { #[test] fn enrich_noop_when_no_ip_found() { - let scorer = RiskScorer::from_preset(RiskPreset::Global); + let scorer = scorer_for("global"); let pipeline = EnrichmentPipeline { ip_fields: vec!["src_ip".into()], geoip: None, @@ -670,7 +671,7 @@ mod tests { let addr: IpAddr = "198.51.100.99".parse().expect("valid IP"); enricher.add_ip(addr, ThreatType::Tor, ThreatSource::TorProject); - let scorer = RiskScorer::from_preset(RiskPreset::HighSecurity); + let scorer = scorer_for("high_security"); let pipeline = EnrichmentPipeline { ip_fields: vec!["client_ip".into()], @@ -960,7 +961,7 @@ mod tests { let addr: IpAddr = "203.0.113.100".parse().expect("valid"); enricher.add_ip(addr, ThreatType::Botnet, ThreatSource::AbuseCh); - let scorer = RiskScorer::from_preset(RiskPreset::HighSecurity); + let scorer = scorer_for("high_security"); let pipeline = EnrichmentPipeline { ip_fields: vec!["src_ip".into(), "client_ip".into()], @@ -1006,7 +1007,7 @@ mod tests { let addr: IpAddr = "203.0.113.200".parse().expect("valid"); enricher.add_ip(addr, ThreatType::Malware, ThreatSource::AbuseCh); - let scorer = RiskScorer::from_preset(RiskPreset::Global); + let scorer = scorer_for("global"); let pipeline = EnrichmentPipeline { ip_fields: vec!["ip".into()], @@ -1084,7 +1085,7 @@ mod tests { fn enrich_risk_only_without_geo_or_rep_skips_scoring() { // Risk scorer alone can't produce meaningful output without GeoIP or reputation. // Implementation: risk only runs when geo_result.is_some() || rep_result.is_some(). - let scorer = RiskScorer::from_preset(RiskPreset::Global); + let scorer = scorer_for("global"); let pipeline = EnrichmentPipeline { ip_fields: vec!["ip".into()], geoip: None, @@ -1277,7 +1278,7 @@ mod tests { #[test] fn enrich_empty_data_map_is_noop() { - let scorer = RiskScorer::from_preset(RiskPreset::Global); + let scorer = scorer_for("global"); let pipeline = EnrichmentPipeline { ip_fields: vec!["src_ip".into()], geoip: None, diff --git a/src/pipeline/orchestrator.rs b/src/pipeline/orchestrator.rs index f76bca4..2a97fff 100644 --- a/src/pipeline/orchestrator.rs +++ b/src/pipeline/orchestrator.rs @@ -2864,7 +2864,7 @@ fn dlq_backends(config: &Config) -> (scalo::dlq::DlqConfig, Option { // Construction succeeded — recv should fail or return empty within // a reasonable window. We tolerate either, but it must NOT panic. @@ -702,9 +702,12 @@ async fn test_auto_discovery_subscribes_only_land_and_load_topics() { produce(&producer, topic, &[br#"{"event_category":"x"}"#.to_vec()]).await; } - let dlq = DlqConfig { - topic: "dlq_land".to_string(), - ..DlqConfig::default() + let dlq = RoutingConfig { + dlq: DlqConfig { + topic: "dlq_land".to_string(), + ..DlqConfig::default() + }, + ..RoutingConfig::default() }; // The same resolver KafkaTransport::new runs when topics is empty. @@ -754,7 +757,7 @@ async fn test_kafka_transport_convert_config_auto_discover() { ..Default::default() }; - let out = TransportAdapter::convert_config(&config, &DlqConfig::default(), &[]); + let out = TransportAdapter::convert_config(&config, &RoutingConfig::default(), &[]); assert_eq!(out.brokers.len(), 2); assert_eq!(out.group, "cfg-group"); assert!( diff --git a/tests/integration/shared_services.rs b/tests/integration/shared_services.rs index 88aab11..71f2285 100644 --- a/tests/integration/shared_services.rs +++ b/tests/integration/shared_services.rs @@ -24,7 +24,7 @@ use dfe_loader::buffer::KafkaOffset; use dfe_loader::clickhouse::client_http::build_client; use dfe_loader::clickhouse::config::{ClickHouseConfig, InsertFormat, Transport}; use dfe_loader::clickhouse::{ClickHouseQueryClient, Inserter, InserterConfig}; -use dfe_loader::config::{DlqConfig, KafkaConfig, SaslConfig, TlsConfig}; +use dfe_loader::config::{KafkaConfig, RoutingConfig, SaslConfig, TlsConfig}; use dfe_loader::kafka::TransportAdapter; use crate::common::containers::{ @@ -704,7 +704,7 @@ async fn kafka_cases(target: &KafkaTarget, topic: &str, group: &str) -> Result<( } let config = target.loader_config(group, topic); - let first = TransportAdapter::new(&config, &DlqConfig::default(), &[], None) + let first = TransportAdapter::new(&config, &RoutingConfig::default(), &[], None) .await .map_err(|e| format!("first adapter: {e}"))?; let got = receive(&first, 6).await; @@ -735,7 +735,7 @@ async fn kafka_cases(target: &KafkaTarget, topic: &str, group: &str) -> Result<( // in the group holds the partition from the second one. drop(first); - let second = TransportAdapter::new(&config, &DlqConfig::default(), &[], None) + let second = TransportAdapter::new(&config, &RoutingConfig::default(), &[], None) .await .map_err(|e| format!("second adapter: {e}"))?; let rest: Vec = receive(&second, 3) From 25451e01eda09969847db23faeca952b6efd6a63 Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 17:00:46 +1100 Subject: [PATCH 3/3] test(risk): compare the empty medium-risk list with assert_eq Clippy refuses assert! on is_empty in the risk config override test, which failed the Quality job. --- src/config/loader.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/config/loader.rs b/src/config/loader.rs index 5ae0584..653fc07 100644 --- a/src/config/loader.rs +++ b/src/config/loader.rs @@ -1496,7 +1496,7 @@ enrichment: let shipped = crate::config::RiskScoringConfig::default(); assert_eq!(risk.countries().unwrap().high, ["XX"]); - assert!(risk.countries().unwrap().medium.is_empty()); + assert_eq!(risk.countries().unwrap().medium, [] as [String; 0]); assert_eq!(risk.presets["mine"].medium, ["ZZ"]); assert_eq!( risk.presets["high_security"],