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

pq_map/discover/
fetch.rs
