pq_map/discover/
fetch.rs

1use std::collections::BTreeMap;
2use std::io::Read;
3use std::thread;
4use std::time::Duration;
5
6use chrono::Utc;
7
8use crate::config::Config;
9use crate::model::{Certificate, Provenance};
10
11use super::cache;
12use super::chain_cache;
13use super::ct;
14use super::ctdb;
15
16#[derive(Debug, thiserror::Error)]
17pub enum Error {
18    #[error("unable to retrieve CT data and no usable cache exists")]
19    NoCtData,
20    #[error("invalid CT data: {0}")]
21    CtParse(String),
22    #[error("task join failed")]
23    Join,
24}
25
26pub struct FetchOutcome {
27    pub certs: BTreeMap<String, Certificate>,
28    pub provenance: Provenance,
29    /// "db" or "http" for a fresh snapshot; None when served from cache.
30    pub source: Option<&'static str>,
31}
32
33pub async fn obtain(domain: &str, cfg: &Config, refresh: bool) -> Result<FetchOutcome, Error> {
34    let file = cache::cache_file(domain);
35
36    if !refresh {
37        if let Some(snap) = cache::load_snapshot(&file) {
38            if let Some(age) = cache::age(&file).filter(|a| *a < cfg.cache_ttl) {
39                eprintln!(
40                    "CT: using cached snapshot for {domain} ({} min old, source {})",
41                    age / 60,
42                    snap.source
43                );
44                return Ok(FetchOutcome {
45                    certs: ct::revive_certs(snap.certs, Utc::now()),
46                    provenance: Provenance::Cached { age_secs: age },
47                    source: None,
48                });
49            }
50        }
51    }
52
53    match ctdb::fetch_rows(domain, cfg).await {
54        Ok(rows) if !rows.is_empty() => {
55            let certs = ct::certificates_from_db(&rows, Utc::now());
56            if !certs.is_empty() {
57                let certs = attach_chains(domain, refresh, certs, cfg).await;
58                store(&file, "db", &certs);
59                return Ok(FetchOutcome {
60                    certs,
61                    provenance: Provenance::Fresh,
62                    source: Some("db"),
63                });
64            }
65            eprintln!("CT: certwatch DB returned no usable certificates");
66        }
67        Ok(_) => eprintln!("CT: certwatch DB returned an empty result; not caching"),
68        Err(e) => eprintln!("CT: falling back to crt.sh HTTP: {e}"),
69    }
70
71    let owned = domain.to_string();
72    let cfg_http = cfg.clone();
73    let http = tokio::task::spawn_blocking(move || http_obtain(&owned, &cfg_http))
74        .await
75        .map_err(|_| Error::Join)?;
76    match http {
77        Ok(body) => {
78            let rows = ct::parse_rows(&body).map_err(|e| Error::CtParse(e.to_string()))?;
79            let certs = ct::certificates(&rows, Utc::now());
80            if !certs.is_empty() {
81                store(&file, "http", &certs);
82                return Ok(FetchOutcome {
83                    certs,
84                    provenance: Provenance::Fresh,
85                    source: Some("http"),
86                });
87            }
88            eprintln!("CT: crt.sh returned no usable certificates");
89        }
90        Err(e) => eprintln!("CT: {e}"),
91    }
92
93    if let Some(snap) = cache::load_snapshot(&file) {
94        eprintln!("WARNING: CT sources unavailable; using stale cached snapshot");
95        return Ok(FetchOutcome {
96            certs: ct::revive_certs(snap.certs, Utc::now()),
97            provenance: Provenance::Stale {
98                age_secs: cache::age(&file),
99            },
100            source: None,
101        });
102    }
103    Err(Error::NoCtData)
104}
105
106async fn attach_chains(
107    domain: &str,
108    refresh: bool,
109    mut certs: BTreeMap<String, Certificate>,
110    cfg: &Config,
111) -> BTreeMap<String, Certificate> {
112    let leaf_ids: Vec<u64> = {
113        let mut ids: Vec<u64> = certs
114            .values()
115            .filter_map(|c| c.ct_ids.first().copied())
116            .collect();
117        ids.sort_unstable();
118        ids.dedup();
119        ids
120    };
121    if leaf_ids.is_empty() {
122        return certs;
123    }
124
125    if !refresh {
126        if let Some(snap) = chain_cache::load(domain) {
127            if let Some(age) = chain_cache::age(domain).filter(|a| *a < cfg.cache_ttl) {
128                eprintln!("CT: using cached chains ({domain}, {} min old)", age / 60);
129                for cert in certs.values_mut() {
130                    if let Some(id) = cert.ct_ids.first() {
131                        cert.chain = snap.chains.get(id).cloned().unwrap_or_default();
132                    }
133                }
134                return certs;
135            }
136        }
137    }
138
139    match ctdb::chain_rows(&leaf_ids, cfg).await {
140        Ok(chains) => {
141            if !chains.is_empty() {
142                chain_cache::store(domain, &chains);
143            }
144            for cert in certs.values_mut() {
145                if let Some(id) = cert.ct_ids.first() {
146                    cert.chain = chains.get(id).cloned().unwrap_or_default();
147                }
148            }
149        }
150        Err(e) => {
151            eprintln!("CT: chain query failed, proceeding without chains: {e}");
152            if let Some(snap) = chain_cache::load(domain) {
153                eprintln!("CT: using stale chains ({domain})");
154                for cert in certs.values_mut() {
155                    if let Some(id) = cert.ct_ids.first() {
156                        cert.chain = snap.chains.get(id).cloned().unwrap_or_default();
157                    }
158                }
159            }
160        }
161    }
162    certs
163}
164
165/// Failure of a single crt.sh request.
166enum FetchFailure {
167    /// HTTP status response, with Retry-After seconds when the server sent one.
168    Status(u16, Option<u64>),
169    /// Transport or read failure.
170    Other(String),
171}
172
173/// Exponential backoff: retry_delay * 2^(attempt-1), capped at 60s.
174pub fn backoff_secs(attempt: u32, retry_delay: u64) -> u64 {
175    retry_delay
176        .saturating_mul(1u64 << (attempt - 1).min(6))
177        .min(60)
178}
179
180fn http_obtain(domain: &str, cfg: &Config) -> Result<String, Error> {
181    for attempt in 1..=cfg.retries {
182        eprintln!(
183            "CT: querying crt.sh for *.{domain} (attempt {attempt}/{})",
184            cfg.retries
185        );
186        match fetch_once(domain, cfg) {
187            Ok(body) => match ct::parse_rows(&body) {
188                Ok(rows) if !rows.is_empty() => return Ok(body),
189                Ok(_) => eprintln!("CT: crt.sh returned an empty result; not caching"),
190                Err(_) => eprintln!("CT: crt.sh returned invalid JSON"),
191            },
192            Err(FetchFailure::Status(code, retry_after)) => {
193                let base = backoff_secs(attempt, cfg.retry_delay);
194                let (why, delay) = match code {
195                    // Rate limited: honor Retry-After when sane, otherwise back
196                    // off twice as hard as for plain server errors.
197                    429 => ("rate limited", retry_after.unwrap_or(base * 2).min(300)),
198                    502 | 503 => ("server error", base),
199                    _ => ("HTTP error", base),
200                };
201                if attempt < cfg.retries {
202                    eprintln!(
203                        "CT: {why} (HTTP {code}); retrying in {delay}s (attempt {attempt}/{} failed)",
204                        cfg.retries
205                    );
206                    thread::sleep(Duration::from_secs(delay));
207                } else {
208                    eprintln!(
209                        "CT: {why} (HTTP {code}); giving up after {} attempts",
210                        cfg.retries
211                    );
212                }
213            }
214            Err(FetchFailure::Other(e)) => {
215                eprintln!("CT: {e}");
216                if attempt < cfg.retries {
217                    let delay = backoff_secs(attempt, cfg.retry_delay);
218                    eprintln!(
219                        "CT: retrying in {delay}s (attempt {}/{} failed)",
220                        attempt, cfg.retries
221                    );
222                    thread::sleep(Duration::from_secs(delay));
223                }
224            }
225        }
226    }
227    Err(Error::NoCtData)
228}
229
230fn fetch_once(domain: &str, cfg: &Config) -> Result<String, FetchFailure> {
231    let agent = ureq::AgentBuilder::new()
232        .timeout_connect(Duration::from_secs(cfg.connect_timeout))
233        .timeout(Duration::from_secs(cfg.max_time))
234        .build();
235    let url = format!("https://crt.sh/?q=%25.{domain}&exclude=expired&output=json");
236    let response = match agent.get(&url).call() {
237        Ok(r) => r,
238        Err(ureq::Error::Status(code, resp)) => {
239            let retry_after = resp
240                .header("retry-after")
241                .and_then(|v| v.trim().parse::<u64>().ok());
242            return Err(FetchFailure::Status(code, retry_after));
243        }
244        Err(ureq::Error::Transport(t)) => {
245            return Err(FetchFailure::Other(format!("crt.sh request failed: {t}")));
246        }
247    };
248    let mut reader = response.into_reader().take(256 * 1024 * 1024);
249    let mut body = String::new();
250    reader
251        .read_to_string(&mut body)
252        .map_err(|e| FetchFailure::Other(format!("crt.sh read failed: {e}")))?;
253    Ok(body)
254}
255
256fn store(file: &std::path::Path, source: &str, certs: &BTreeMap<String, Certificate>) {
257    let snapshot = cache::Snapshot {
258        version: cache::SNAPSHOT_VERSION,
259        source: source.to_string(),
260        fetched_at: Utc::now().to_rfc3339(),
261        certs: certs.values().cloned().collect(),
262    };
263    match cache::store_snapshot(file, &snapshot) {
264        Ok(()) => eprintln!("CT: snapshot cached (source {source}): {}", file.display()),
265        Err(e) => eprintln!("CT: caching failed: {e}"),
266    }
267}
268
269#[cfg(test)]
270mod tests {
271    use super::*;
272
273    #[test]
274    fn backoff_is_exponential_and_capped() {
275        assert_eq!(backoff_secs(1, 2), 2);
276        assert_eq!(backoff_secs(2, 2), 4);
277        assert_eq!(backoff_secs(3, 2), 8);
278        assert_eq!(backoff_secs(4, 2), 16);
279        assert_eq!(backoff_secs(6, 2), 60, "capped at 60s");
280        assert_eq!(backoff_secs(20, 2), 60);
281        assert_eq!(backoff_secs(1, 120), 60, "capped at 60s");
282    }
283}
pq_map/discover/fetch.rs

pq_map/discover/
fetch.rs
