diff --git a/Cargo.lock b/Cargo.lock index 7235b17..b7b2dfb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -361,7 +361,7 @@ dependencies = [ [[package]] name = "certstream-server-rust" -version = "1.5.3" +version = "1.5.4" dependencies = [ "ahash", "arc-swap", diff --git a/Cargo.toml b/Cargo.toml index 33bbcd8..bd77d4e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "certstream-server-rust" -version = "1.5.3" +version = "1.5.4" edition = "2024" repository = "https://github.com/reloading01/certstream-server-rust" license = "MIT" diff --git a/README.md b/README.md index 60712fb..e210b0c 100644 --- a/README.md +++ b/README.md @@ -139,6 +139,20 @@ rate_limit: | `CERTSTREAM_CT_LOG_REQUEST_TIMEOUT_SECS` | 30 | Request timeout | | `CERTSTREAM_CT_LOG_BATCH_SIZE` | 1024 | Entries requested per get-entries call (servers clamp to their own max) | | `CERTSTREAM_CT_LOG_FETCH_CONCURRENCY` | 4 | Concurrent range/tile fetches per watcher during catch-up (1-16) | +| `CERTSTREAM_USER_AGENT` | certstream-server-rust/{VERSION} | User-Agent for all outbound HTTP (CT log fetches and catalog fetches). Some operators (e.g. Geomys) apply a more generous rate limit tier to clients that include a contact email. Blank falls back to the default. | +| `CERTSTREAM_CT_LOG_FORCE_HTTP1_OPERATORS` | - | Comma-separated operator names whose watchers fetch over HTTP/1.1 instead of HTTP/2 (see below) | + +**Forcing HTTP/1.1 per operator** + +Some operators (DigiCert) throttle per TCP connection rather than per IP. Under HTTP/2 reqwest multiplexes every request for a host onto a single connection, so that per-connection quota ends up capping the whole process — DigiCert serves several logs from one host (`wyvern.ct.digicert.com`), so their watchers all share it. Listing an operator gives its watchers a dedicated HTTP/1.1 client, where each in-flight fetch needs its own connection and therefore carries its own quota: + +```yaml +ct_log: + force_http1_operators: + - DigiCert +``` + +This does not raise the request rate — the per-operator token bucket (`default_operator_rate_limit_ms`, 500 ms) still gates every fetch. It only spreads the same requests across more connections; `fetch_concurrency` sets how many of those stay warm in the pool between polls. Operator names are matched case-insensitively and ignoring punctuation, the same as `operator_rate_limits`; a name that matches no discovered log is logged as a warning at startup. **Hot Reload** diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 5f25680..11dce36 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,3 +1,65 @@ +# Release Notes — v1.5.4 + +**Release date:** August 22, 2026 + +Two outbound-HTTP controls for deployments that hit CT log operator rate limits. No wire-format or API changes. + +## Configurable User-Agent + +`ct_log.user_agent` (env `CERTSTREAM_USER_AGENT`) sets the User-Agent on every outbound request — CT log fetches and catalog fetches alike. Some operators (Geomys) apply a more generous rate limit tier to clients that carry a contact address: + +```yaml +ct_log: + user_agent: "certstream-server-rust (security@example.com)" +``` + +Unset or blank falls back to `certstream-server-rust/{VERSION}`, so `CERTSTREAM_USER_AGENT=` in a compose file or `.env` cannot silently strip the header; a blank value is logged at startup. A value that is not legal HTTP header content fails config validation before the server binds. + +Contributed by Effy Elden (@ineffyble) in #12. + +## Per-operator HTTP/1.1 transport + +DigiCert throttles per TCP connection rather than per IP. Under HTTP/2 reqwest multiplexes every request for a host onto a single connection, so a per-connection quota ends up capping the whole process — and DigiCert serves several logs from one host (`wyvern.ct.digicert.com` negotiates h2), so all of their watchers share it. Listing an operator gives its watchers a dedicated HTTP/1.1 client, where each in-flight fetch needs its own connection and therefore carries its own quota: + +```yaml +ct_log: + force_http1_operators: + - DigiCert +``` + +Env: `CERTSTREAM_CT_LOG_FORCE_HTTP1_OPERATORS=DigiCert,Geomys`. + +This does not raise the outbound request rate. The per-operator token bucket (`default_operator_rate_limit_ms`, 500 ms) still gates every fetch — the same requests are spread across more connections, and `fetch_concurrency` only decides how many of those stay warm in the pool. Operators not listed keep the shared HTTP/2 client. + +The observation comes from certspotter's `digicerthack` branch ([SSLMate/certspotter#126](https://github.com/SSLMate/certspotter/issues/126)), which drops keep-alives outright instead. + +## Operator name matching + +`operator_rate_limits` and `force_http1_operators` are both looked up by canonicalized operator name (case, whitespace and punctuation collapsed). A key matching no discovered log used to be silently inert, which is indistinguishable from a working config until you measure. Startup now names them: + +``` +WARN configured operator names match no discovered CT log operator and have no effect field="ct_log.operator_rate_limits" unmatched=["digicert inc"] +``` + +## Configuration + +| Setting | Old | New | +| ------- | --: | --: | +| `ct_log.user_agent` | — | `null` (new; env `CERTSTREAM_USER_AGENT`) | +| `ct_log.force_http1_operators` | — | `[]` (new; env `CERTSTREAM_CT_LOG_FORCE_HTTP1_OPERATORS`) | + +## Tests + +262 unit tests (was 251) plus the integration and snapshot suites, unchanged. + +## Upgrade + +Drop-in — both settings default to the v1.5.3 behaviour. + +```bash +docker pull ghcr.io/reloading01/certstream-server-rust:1.5.4 +``` + # Release Notes — v1.5.3 **Release date:** July 18, 2026 diff --git a/config.example.yaml b/config.example.yaml index f3eac41..4799d9c 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -54,6 +54,19 @@ ct_log: # any operator absent from operator_rate_limits. default_operator_rate_limit_ms: 500 operator_rate_limits: {} + # User-Agent for all outbound HTTP. Some operators (e.g. Geomys) apply a + # more generous rate limit tier to clients that include a contact email. + # Unset or blank defaults to certstream-server-rust/{VERSION}. + # (env: CERTSTREAM_USER_AGENT) + user_agent: null + # Operators whose watchers fetch over HTTP/1.1 instead of HTTP/2. DigiCert + # throttles per TCP connection, and HTTP/2 puts every request for a host on + # one connection; under HTTP/1.1 each in-flight fetch needs its own + # connection and so carries its own quota. The per-operator rate limiter + # still gates every fetch, so this redistributes the request rate rather + # than raising it. Names are matched like operator_rate_limits keys. + # (env: CERTSTREAM_CT_LOG_FORCE_HTTP1_OPERATORS, comma-separated) + force_http1_operators: [] # Per-catalog-source runtime-authority overrides. Keys are # google_v3_usable, google_v3_all, and apple. An override can only grant # authority to a source that currently verifies; it cannot promote an diff --git a/docs/docs.html b/docs/docs.html index 1a88135..41f1999 100644 --- a/docs/docs.html +++ b/docs/docs.html @@ -495,6 +495,8 @@

CT log settings

CERTSTREAM_CT_LOG_REQUEST_TIMEOUT_SECS30Request timeout CERTSTREAM_CT_LOG_BATCH_SIZE1024Entries requested per get-entries call (servers clamp to their own max) CERTSTREAM_CT_LOG_FETCH_CONCURRENCY4Concurrent range/tile fetches per watcher during catch-up (1-16) + CERTSTREAM_USER_AGENTcertstream-server-rust/{VERSION}User-Agent for all outbound HTTP. Some operators (e.g. Geomys) apply a more generous rate limit tier to clients that include a contact email. Blank falls back to the default. + CERTSTREAM_CT_LOG_FORCE_HTTP1_OPERATORS-Comma-separated operators whose watchers fetch over HTTP/1.1 instead of HTTP/2, so each in-flight fetch carries its own per-connection quota (DigiCert) CERTSTREAM_STATIC_CT_CHECKPOINT_SIGNATUREwarnCheckpoint signature policy: warn or enforce CERTSTREAM_DEDUP_CAPACITY200000Cross-log dedup capacity CERTSTREAM_DEDUP_TTL_SECS900Dedup window (seconds) @@ -549,6 +551,8 @@

YAML config file

poll_interval_ms: 500 retry_max_attempts: 3 request_timeout_secs: 30 + user_agent: "certstream-server-rust (security@example.com)" + force_http1_operators: ["DigiCert"] dedup: capacity: 200000 diff --git a/src/cli.rs b/src/cli.rs index b6d3604..4f23852 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -2,6 +2,11 @@ use std::env; pub const VERSION: &str = env!("CARGO_PKG_VERSION"); +/// User-Agent sent on every outbound HTTP request unless `ct_log.user_agent` +/// overrides it. Built from the package version at compile time so the CT log +/// fetch client and the TLS-pinned Apple catalog client can never drift apart. +pub const DEFAULT_USER_AGENT: &str = concat!("certstream-server-rust/", env!("CARGO_PKG_VERSION")); + #[derive(Debug, Clone)] pub struct CliArgs { pub validate_config: bool, @@ -45,6 +50,7 @@ impl CliArgs { println!(" CERTSTREAM_PORT Server port (default: 8080)"); println!(" CERTSTREAM_LOG_LEVEL Log level (default: info)"); println!(" CERTSTREAM_BUFFER_SIZE Broadcast buffer size (default: 1000)"); + println!(" CERTSTREAM_USER_AGENT Override the outbound HTTP User-Agent"); println!(); println!("For more information, see: https://github.com/reloading01/certstream-server-rust"); } diff --git a/src/config.rs b/src/config.rs index fda8e53..5a82288 100644 --- a/src/config.rs +++ b/src/config.rs @@ -165,6 +165,22 @@ pub struct CtLogConfig { /// whitespace, or punctuation. Empty map means every operator uses the default. #[serde(default)] pub operator_rate_limits: std::collections::HashMap, + /// HTTP User-Agent for outbound requests. Some CT log operators (e.g. + /// Geomys) apply a more generous rate limit tier to clients that include + /// a contact email. Unset or blank falls back to the compiled-in + /// `certstream-server-rust/{VERSION}`; read this through + /// [`CtLogConfig::user_agent_override`] rather than directly. + #[serde(default)] + pub user_agent: Option, + /// Operators whose watchers fetch over a dedicated HTTP/1.1-only client. + /// DigiCert throttles per TCP connection rather than per IP; under HTTP/2 + /// reqwest multiplexes every request for a host onto one connection, so + /// that per-connection quota becomes a whole-process quota. HTTP/1.1 + /// spreads the same request rate — still capped by the per-operator + /// limiter — over one connection per in-flight fetch. Names are + /// canonicalized with the same rules as `operator_rate_limits`. + #[serde(default)] + pub force_http1_operators: Vec, /// Per-catalog-source runtime-authority overrides. Keys are the catalog /// registry source names (`google_v3_usable`, `google_v3_all`, `apple`). /// An override can only grant authority to a source that currently verifies; @@ -199,6 +215,20 @@ impl std::str::FromStr for CheckpointSignatureMode { } } +impl CtLogConfig { + /// The configured User-Agent with surrounding whitespace trimmed, or + /// `None` when unset or blank. A blank value is treated as unset because + /// `CERTSTREAM_USER_AGENT=` in a compose file or `.env` reads back as an + /// empty string, and an empty `User-Agent:` header is exactly the opposite + /// of what an operator setting this is asking for. + pub fn user_agent_override(&self) -> Option<&str> { + self.user_agent + .as_deref() + .map(str::trim) + .filter(|ua| !ua.is_empty()) + } +} + impl Default for CtLogConfig { fn default() -> Self { Self { @@ -219,6 +249,8 @@ impl Default for CtLogConfig { static_ct_enabled: true, default_operator_rate_limit_ms: default_operator_rate_limit_ms(), operator_rate_limits: std::collections::HashMap::new(), + user_agent: None, + force_http1_operators: Vec::new(), catalog_authority_overrides: std::collections::HashMap::new(), } } @@ -226,6 +258,16 @@ impl Default for CtLogConfig { pub const MAX_START_OVERLAP_LEAVES: u64 = 100_000; +/// Split a comma-separated env value into operator names, dropping blanks so +/// `"digicert,"` and `"digicert, ,geomys"` behave like the obvious YAML list. +fn parse_operator_list(raw: &str) -> Vec { + raw.split(',') + .map(str::trim) + .filter(|name| !name.is_empty()) + .map(str::to_string) + .collect() +} + fn default_operator_rate_limit_ms() -> u64 { 500 } @@ -586,6 +628,10 @@ impl Config { ct_log.checkpoint_signature_mode, "CERTSTREAM_STATIC_CT_CHECKPOINT_SIGNATURE" ); + env_override!(ct_log.user_agent, "CERTSTREAM_USER_AGENT", some_str); + if let Ok(v) = env::var("CERTSTREAM_CT_LOG_FORCE_HTTP1_OPERATORS") { + ct_log.force_http1_operators = parse_operator_list(&v); + } let mut connection_limit = yaml_config.connection_limit.unwrap_or_default(); env_override!(connection_limit.enabled, "CERTSTREAM_CONNECTION_LIMIT_ENABLED"); @@ -706,6 +752,14 @@ impl Config { message: "Fetch concurrency must be between 1 and 16".to_string(), }); } + if let Some(ua) = self.ct_log.user_agent_override() + && reqwest::header::HeaderValue::try_from(ua).is_err() + { + errors.push(ConfigValidationError { + field: "ct_log.user_agent".to_string(), + message: "User-Agent must be a valid HTTP header value".to_string(), + }); + } if errors.is_empty() { Ok(()) @@ -801,6 +855,96 @@ mod tests { assert_eq!(config.start_overlap_leaves, 256); assert!(config.rfc6962_enabled); assert!(config.static_ct_enabled); + assert!(config.user_agent.is_none()); + } + + #[test] + fn test_ct_log_config_deserialize_user_agent() { + let yaml = r#" +user_agent: "certstream-server-rust/1.5.3 (contact@example.com)" +"#; + let config: CtLogConfig = serde_yaml::from_str(yaml).unwrap(); + assert_eq!( + config.user_agent.as_deref(), + Some("certstream-server-rust/1.5.3 (contact@example.com)") + ); + } + + #[test] + fn test_blank_user_agent_falls_back_to_default() { + // `CERTSTREAM_USER_AGENT=` in a compose file reads back as Ok(""), and + // an empty User-Agent header is worse than the default one. + for blank in ["", " ", "\t"] { + let config = CtLogConfig { + user_agent: Some(blank.to_string()), + ..CtLogConfig::default() + }; + assert_eq!(config.user_agent_override(), None, "blank: {blank:?}"); + } + } + + #[test] + fn test_user_agent_override_is_trimmed() { + let config = CtLogConfig { + user_agent: Some(" certstream/1.0 (me@example.com) ".to_string()), + ..CtLogConfig::default() + }; + assert_eq!( + config.user_agent_override(), + Some("certstream/1.0 (me@example.com)") + ); + } + + #[test] + fn test_validate_blank_user_agent_is_not_an_error() { + let config = Config { + ct_log: CtLogConfig { + user_agent: Some(" ".to_string()), + ..CtLogConfig::default() + }, + ..test_config() + }; + assert!(config.validate().is_ok()); + } + + #[test] + fn test_parse_operator_list_drops_blanks_and_trims() { + assert_eq!( + parse_operator_list("DigiCert, Geomys"), + vec!["DigiCert".to_string(), "Geomys".to_string()] + ); + assert_eq!( + parse_operator_list("digicert,, ,geomys,"), + vec!["digicert".to_string(), "geomys".to_string()] + ); + assert!(parse_operator_list("").is_empty()); + assert!(parse_operator_list(" , ").is_empty()); + } + + #[test] + fn test_ct_log_config_deserialize_force_http1_operators() { + let yaml = r#" +force_http1_operators: + - DigiCert + - Geomys +"#; + let config: CtLogConfig = serde_yaml::from_str(yaml).unwrap(); + assert_eq!(config.force_http1_operators, vec!["DigiCert", "Geomys"]); + } + + #[test] + fn test_validate_user_agent_invalid_header() { + let config = Config { + ct_log: CtLogConfig { + user_agent: Some("bad\nuser-agent".to_string()), + ..CtLogConfig::default() + }, + ..test_config() + }; + let result = config.validate(); + assert!(result.is_err()); + let errors = result.unwrap_err(); + assert!(errors.iter().any(|e| e.field == "ct_log.user_agent")); } #[test] diff --git a/src/ct/catalog/tls_pin.rs b/src/ct/catalog/tls_pin.rs index c127745..3c6eaa4 100644 --- a/src/ct/catalog/tls_pin.rs +++ b/src/ct/catalog/tls_pin.rs @@ -118,7 +118,10 @@ impl ServerCertVerifier for PinnedIssuerVerifier { /// Build a dedicated reqwest client that pins the Apple issuer-CA SPKI on top of /// normal WebPKI validation. Used ONLY for the `apple` catalog fetch. `timeout` /// The timeout bounds the fetch so a hung Apple endpoint cannot wedge startup. -pub fn build_apple_pinned_client(timeout: std::time::Duration) -> Result { +pub fn build_apple_pinned_client( + timeout: std::time::Duration, + user_agent: &str, +) -> Result { let provider = Arc::new(aws_lc_rs::default_provider()); // Trust roots from the OS store (same source rustls-native-certs feeds the @@ -152,10 +155,7 @@ pub fn build_apple_pinned_client(timeout: std::time::Duration) -> Result, custom_logs: Vec, request_timeout: Duration, + user_agent: &str, ) -> Result, LogListError> { // Apple has no detached signature, so it is fetched through a dedicated // client that pins the issuer-CA SPKI on top of WebPKI validation. If that // client cannot be built, Apple is skipped this cycle rather than fetched // unpinned. Apple is non-authoritative, so skipping has no spawn impact. - let apple_client = match catalog::build_apple_pinned_client(request_timeout) { + let apple_client = match catalog::build_apple_pinned_client(request_timeout, user_agent) { Ok(c) => Some(c), Err(e) => { warn!(error = %e, "failed to build TLS-pinned Apple client; skipping the Apple catalog this cycle"); diff --git a/src/main.rs b/src/main.rs index 999d589..33ec7c2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -21,12 +21,12 @@ use std::time::Duration; use tokio::sync::broadcast; use tokio_util::sync::CancellationToken; use tower_http::cors::{Any as CorsAny, CorsLayer}; -use tracing::{error, info}; +use tracing::{error, info, warn}; use tracing_subscriber::EnvFilter; use api::{ApiState, CertificateCache, LogTracker, ServerStats}; -use cli::{CliArgs, VERSION}; -use config::Config; +use cli::{CliArgs, DEFAULT_USER_AGENT, VERSION}; +use config::{Config, CtLogConfig}; use ct::{fetch_log_list, WatcherContext}; use dedup::DedupFilter; use health::{deep_health, example_json, health, HealthState}; @@ -140,21 +140,22 @@ async fn main() { let tx: broadcast::Sender> = broadcast::channel(config.buffer_size).0; - let client = Client::builder() - .user_agent(format!("certstream-server-rust/{}", VERSION)) - // Pre-1.5.0 kept 20 idle connections per host × 55 hosts = 1100 - // hot TCP sockets, ~40-55 MiB of kernel + TLS state per process. - // Watchers now pipeline up to `fetch_concurrency` range/tile fetches - // during catch-up, so keep that many idle connections per host (with - // HTTP/2 they multiplex over fewer; this matters for HTTP/1.1 hosts). - .pool_max_idle_per_host((config.ct_log.fetch_concurrency as usize).max(2)) - .pool_idle_timeout(Duration::from_secs(30)) - // Global timeout for catalog list/signature fetches. Watcher fetches - // also set per-request bounds; this backstops shared-client requests. - .timeout(Duration::from_secs(config.ct_log.request_timeout_secs)) - .tcp_nodelay(true) - .build() + let user_agent = match config.ct_log.user_agent_override() { + Some(ua) => { + info!(user_agent = ua, "using configured User-Agent"); + ua.to_string() + } + None => { + if config.ct_log.user_agent.is_some() { + warn!("configured User-Agent is blank; falling back to the default"); + } + DEFAULT_USER_AGENT.to_string() + } + }; + + let client = build_ct_client(&config.ct_log, &user_agent, false) .expect("failed to build http client"); + let transport = OperatorTransport::new(&config.ct_log, &user_agent); let state_manager = StateManager::new(config.ct_log.state_file.clone()); if config.ct_log.state_file.is_some() { @@ -251,7 +252,7 @@ async fn main() { }; let (rfc_count, static_count) = - discover_and_spawn(&config, &log_tracker, &watcher_ctx).await; + discover_and_spawn(&config, &log_tracker, &watcher_ctx, &transport).await; if rfc_count == 0 && static_count == 0 { error!("no CT log watchers were started — refusing to run with zero sources"); @@ -384,11 +385,118 @@ fn spawn_signal_handler(shutdown_token: CancellationToken) { }); } +/// Build a CT-fetch HTTP client. `http1_only` yields the transport used by the +/// operators listed in `ct_log.force_http1_operators`. +fn build_ct_client( + ct_log: &CtLogConfig, + user_agent: &str, + http1_only: bool, +) -> reqwest::Result { + let builder = Client::builder() + .user_agent(user_agent) + // Pre-1.5.0 kept 20 idle connections per host × 55 hosts = 1100 + // hot TCP sockets, ~40-55 MiB of kernel + TLS state per process. + // Watchers now pipeline up to `fetch_concurrency` range/tile fetches + // during catch-up, so keep that many idle connections per host (with + // HTTP/2 they multiplex over fewer; this matters for HTTP/1.1 hosts). + .pool_max_idle_per_host((ct_log.fetch_concurrency as usize).max(2)) + .pool_idle_timeout(Duration::from_secs(30)) + // Global timeout for catalog list/signature fetches. Watcher fetches + // also set per-request bounds; this backstops shared-client requests. + .timeout(Duration::from_secs(ct_log.request_timeout_secs)) + .tcp_nodelay(true); + if http1_only { + builder.http1_only().build() + } else { + builder.build() + } +} + +/// Per-operator transport selection. +/// +/// DigiCert throttles per TCP connection rather than per IP. Under HTTP/2 +/// reqwest multiplexes every watcher's request for a host onto one connection, +/// so that per-connection quota caps the whole process — and DigiCert serves +/// several logs from one host, so their watchers all land on it. HTTP/1.1 needs +/// a connection per in-flight request, so each concurrent fetch carries its own +/// quota (`pool_max_idle_per_host` only decides how many stay warm between +/// polls). The per-operator token bucket still gates every fetch, so this +/// redistributes our request rate across connections rather than raising it. +/// Same observation as certspotter's `digicerthack` branch +/// (), which drops +/// keep-alives outright instead. +struct OperatorTransport { + /// `None` when no operator is listed, or when the client failed to build. + client: Option, + /// Canonicalized operator names that should use `client`. + operators: std::collections::HashSet, +} + +impl OperatorTransport { + fn new(ct_log: &CtLogConfig, user_agent: &str) -> Self { + let operators: std::collections::HashSet = ct_log + .force_http1_operators + .iter() + .map(|op| ct::normalize_operator(op)) + .collect(); + if operators.is_empty() { + return Self { + client: None, + operators, + }; + } + let client = match build_ct_client(ct_log, user_agent, true) { + Ok(c) => { + let mut names: Vec<&str> = operators.iter().map(String::as_str).collect(); + names.sort_unstable(); + info!(operators = ?names, "forcing HTTP/1.1 transport"); + Some(c) + } + Err(e) => { + error!(error = %e, "failed to build the HTTP/1.1 client; listed operators stay on the shared client"); + None + } + }; + Self { client, operators } + } + + fn uses_http1(&self, operator_key: &str) -> bool { + self.client.is_some() && self.operators.contains(operator_key) + } + + /// Resolve the client for an already-canonicalized operator name. + fn client_for<'a>(&'a self, operator_key: &str, shared: &'a Client) -> &'a Client { + match &self.client { + Some(client) if self.uses_http1(operator_key) => client, + _ => shared, + } + } +} + +/// Configured operator names that match none of the discovered logs. Both +/// `operator_rate_limits` and `force_http1_operators` are looked up by +/// canonicalized name, so a key that never matches is silently inert — which +/// from the outside looks exactly like a working config. +fn unmatched_operator_keys<'a>( + configured: impl IntoIterator, + known: &std::collections::HashSet, +) -> Vec { + let mut missing: Vec = configured + .into_iter() + .filter(|key| !known.contains(&ct::normalize_operator(key))) + .cloned() + .collect(); + missing.sort(); + missing.dedup(); + missing +} + /// Discovery + spawn pipeline. Returns `(rfc6962_count, static_ct_count)`. async fn discover_and_spawn( config: &Config, log_tracker: &Arc, ctx: &WatcherContext, + transport: &OperatorTransport, ) -> (usize, usize) { use std::collections::HashMap; use ct::{LogType, OperatorRateLimiter}; @@ -401,6 +509,7 @@ async fn discover_and_spawn( &config.ct_log.catalog_authority_overrides, config.custom_logs.clone(), Duration::from_secs(config.ct_log.request_timeout_secs), + ctx.config.user_agent_override().unwrap_or(DEFAULT_USER_AGENT), ) .await; @@ -460,6 +569,29 @@ async fn discover_and_spawn( } } + let known_operators: std::collections::HashSet = all_logs + .iter() + .map(|log| ct::normalize_operator(&log.operator)) + .collect(); + for (field, unmatched) in [ + ( + "ct_log.operator_rate_limits", + unmatched_operator_keys(config.ct_log.operator_rate_limits.keys(), &known_operators), + ), + ( + "ct_log.force_http1_operators", + unmatched_operator_keys(&config.ct_log.force_http1_operators, &known_operators), + ), + ] { + if !unmatched.is_empty() { + warn!( + field, + unmatched = ?unmatched, + "configured operator names match no discovered CT log operator and have no effect" + ); + } + } + // Partition by type — the two watcher pools differ in protocol. let (rfc_logs, static_logs): (Vec<_>, Vec<_>) = all_logs.into_iter().partition(|l| l.log_type == LogType::Rfc6962); @@ -472,6 +604,7 @@ async fn discover_and_spawn( rfc_logs, log_tracker, ctx, + transport, &mut operator_limiters, "certstream_ct_logs_count", 50, @@ -484,6 +617,7 @@ async fn discover_and_spawn( static_logs, log_tracker, ctx, + transport, &mut operator_limiters, "certstream_static_ct_logs_count", 100, @@ -518,6 +652,7 @@ fn spawn_pool( logs: Vec, log_tracker: &Arc, ctx: &WatcherContext, + transport: &OperatorTransport, operator_limiters: &mut std::collections::HashMap, count_gauge: &'static str, startup_stagger_ms: u64, @@ -560,9 +695,9 @@ fn spawn_pool( let count = logs.len(); for (index, log) in logs.into_iter().enumerate() { let mut wctx = ctx.clone(); - wctx.rate_limiter = operator_limiters - .get(&ct::normalize_operator(&log.operator)) - .cloned(); + let operator_key = ct::normalize_operator(&log.operator); + wctx.rate_limiter = operator_limiters.get(&operator_key).cloned(); + wctx.client = transport.client_for(&operator_key, &ctx.client).clone(); // Catalog-discovered logs share the global config. Local custom/static // entries with per-log overrides get their own resolved config. if log.batch_size.is_some() || log.poll_interval_ms.is_some() { @@ -923,3 +1058,61 @@ fn print_config_validation(config: &Config) { } } } + +#[cfg(test)] +mod tests { + use super::*; + + fn transport_with(operators: &[&str]) -> OperatorTransport { + let ct_log = CtLogConfig { + force_http1_operators: operators.iter().map(|s| s.to_string()).collect(), + ..CtLogConfig::default() + }; + OperatorTransport::new(&ct_log, DEFAULT_USER_AGENT) + } + + #[test] + fn transport_matches_operators_by_canonical_name() { + let transport = transport_with(&["DigiCert, Inc."]); + assert!(transport.uses_http1(&ct::normalize_operator("digicert inc"))); + assert!(transport.uses_http1(&ct::normalize_operator(" DigiCert Inc "))); + // The catalog-emitted "DigiCert" is a different canonical name than + // "DigiCert, Inc." — punctuation collapses, the suffix does not. + assert!(!transport.uses_http1(&ct::normalize_operator("DigiCert"))); + assert!(!transport.uses_http1(&ct::normalize_operator("Google"))); + } + + #[test] + fn transport_without_listed_operators_builds_no_client() { + let transport = transport_with(&[]); + assert!(transport.client.is_none()); + assert!(!transport.uses_http1("digicert")); + } + + #[test] + fn transport_returns_shared_client_for_unlisted_operators() { + let transport = transport_with(&["DigiCert"]); + let shared = Client::new(); + assert!(!std::ptr::eq( + transport.client_for("digicert", &shared), + &shared + )); + assert!(std::ptr::eq(transport.client_for("google", &shared), &shared)); + } + + #[test] + fn unmatched_operator_keys_reports_only_misses() { + let known: std::collections::HashSet = + ["digicert", "google"].iter().map(|s| s.to_string()).collect(); + let configured = vec![ + "DigiCert".to_string(), + "Geomys".to_string(), + "sectigo".to_string(), + ]; + assert_eq!( + unmatched_operator_keys(&configured, &known), + vec!["Geomys".to_string(), "sectigo".to_string()] + ); + assert!(unmatched_operator_keys(std::iter::empty(), &known).is_empty()); + } +}