diff --git a/Cargo.lock b/Cargo.lock index 07261db1..43fa3eef 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 c26e22e1..754357d8 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, diff --git a/benches/enrichment.rs b/benches/enrichment.rs index 8cba0b51..822fcaa0 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 8da3dc50..6c9bf5d1 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 2e343e96..cf9f561e 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 93f1cd28..c4a4c05b 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 15d20e6f..f729649f 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 3e149078..ab0b2060 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 9882e5bf..0ccaacc5 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 26711309..e8666795 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 59018401..653fc073 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_eq!(risk.countries().unwrap().medium, [] as [String; 0]); + 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 3f58653f..307c3ba7 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 716cbb2a..5d7740a7 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 5631562c..a3eb083c 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 9d3e701f..e05f642c 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 2ffd1dbc..29feab68 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 f76bca4d..2a97ffff 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 88aab11b..71f22854 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)