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

pq_map/discover/
fetch.rs
