1use std::collections::BTreeMap;
2use std::time::Duration;
3
4use tokio_postgres::{Client, NoTls, SimpleQueryMessage};
5
6use crate::config::Config;
7use crate::model::ChainEntry;
8
9#[derive(Debug, Clone)]
10pub struct DbRow {
11 pub cert_id: i64,
12 pub serial: String,
13 pub dns_name: Option<String>,
14 pub common_name: Option<String>,
15 pub not_before: Option<String>,
16 pub not_after: Option<String>,
17 pub key_algorithm: Option<String>,
18 pub key_size: Option<i64>,
19 pub sig_key_algorithm: Option<String>,
20 pub sig_hash_algorithm: Option<String>,
21 pub issuer_ca_name: Option<String>,
22}
23
24const CT_SQL_TEMPLATE: &str = "
25WITH matching AS (
26 SELECT c.id, c.issuer_ca_id, c.certificate
27 FROM certificate c
28 WHERE identities(c.certificate) @@ plainto_tsquery('certwatch', '{domain}')
29 AND x509_notAfter(c.certificate) >= now()
30)
31SELECT
32 m.id,
33 encode(x509_serialnumber(m.certificate), 'hex'),
34 n.name_value,
35 x509_commonName(m.certificate),
36 x509_notBefore(m.certificate)::text,
37 x509_notAfter(m.certificate)::text,
38 x509_keyAlgorithm(m.certificate),
39 x509_keySize(m.certificate),
40 x509_signatureKeyAlgorithm(m.certificate),
41 x509_signatureHashAlgorithm(m.certificate),
42 ca.name
43FROM matching m
44LEFT JOIN ca ON ca.id = m.issuer_ca_id
45CROSS JOIN LATERAL (
46 SELECT encode(raw_value, 'escape') AS name_value, 2 AS src
47 FROM x509_altNames_raw(m.certificate)
48 WHERE type_num = 2
49 UNION ALL
50 SELECT x509_commonName(m.certificate), 1
51) n
52WHERE lower(n.name_value) = lower('{domain}')
53 OR lower(n.name_value) LIKE '%.{domain}'";
54
55pub fn ct_sql(domain: &str) -> Result<String, String> {
56 let domain = domain.to_ascii_lowercase();
57 if domain.is_empty() || domain.len() > 253 {
58 return Err(format!("invalid domain: {domain}"));
59 }
60 if !domain
61 .chars()
62 .all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-')
63 {
64 return Err(format!("invalid characters in domain: {domain}"));
65 }
66 Ok(CT_SQL_TEMPLATE.replace("{domain}", &domain))
67}
68
69pub async fn fetch_rows(domain: &str, cfg: &Config) -> Result<Vec<DbRow>, String> {
70 let mut last_error = String::from("not attempted");
71 for attempt in 1..=cfg.retries {
72 eprintln!(
73 "CT: querying certwatch DB at {}:{} (attempt {attempt}/{})",
74 cfg.db_host, cfg.db_port, cfg.retries
75 );
76 match attempt_once(domain, cfg).await {
77 Ok(rows) => return Ok(rows),
78 Err(e) => {
79 last_error = e;
80 if attempt < cfg.retries {
81 let delay = db_backoff_secs(attempt, cfg.retry_delay);
82 eprintln!(
83 "CT: {last_error}; retrying in {delay}s (attempt {attempt}/{} failed)",
84 cfg.retries
85 );
86 tokio::time::sleep(Duration::from_secs(delay)).await;
87 }
88 }
89 }
90 }
91 eprintln!(
92 "CT: certwatch DB unreachable after {} attempts",
93 cfg.retries
94 );
95 Err(last_error)
96}
97
98pub fn db_backoff_secs(attempt: u32, retry_delay: u64) -> u64 {
99 retry_delay
100 .saturating_mul(1u64 << (attempt + 1).min(6))
101 .min(60)
102}
103
104async fn attempt_once(domain: &str, cfg: &Config) -> Result<Vec<DbRow>, String> {
105 let conninfo = format!(
106 "host={} port={} user={} dbname={}",
107 cfg.db_host, cfg.db_port, cfg.db_user, cfg.db_name
108 );
109 let connect = tokio_postgres::connect(&conninfo, NoTls);
110 let (client, connection) =
111 tokio::time::timeout(Duration::from_secs(cfg.db_connect_timeout), connect)
112 .await
113 .map_err(|_| "connect timeout".to_string())?
114 .map_err(db_err)?;
115 let drive = tokio::spawn(async move {
116 if let Err(e) = connection.await {
117 eprintln!("CT: db connection dropped: {}", db_err(e));
118 }
119 });
120
121 let rows = query_once(&client, domain, cfg).await;
122
123 drop(client);
124 let _ = drive.await;
125 rows
126}
127
128const CHAIN_SQL_TEMPLATE: &str = "
129WITH RECURSIVE cert_chain AS (
130 SELECT
131 c.id AS certificate_id,
132 c.issuer_ca_id,
133 c.certificate,
134 c.id AS leaf_id,
135 0 AS depth,
136 ARRAY[c.id]::bigint[] AS path
137 FROM certificate c
138 WHERE c.id = ANY(ARRAY[{ids}]::bigint[])
139 AND x509_notBefore(c.certificate) <= now()
140 AND x509_notAfter(c.certificate) >= now()
141
142 UNION ALL
143
144 SELECT
145 parent.id,
146 parent.issuer_ca_id,
147 parent.certificate,
148 cc.leaf_id,
149 cc.depth + 1,
150 cc.path || parent.id
151 FROM cert_chain cc
152
153 JOIN LATERAL (
154 SELECT c2.*
155 FROM ca_certificate cac
156 JOIN certificate c2
157 ON c2.id = cac.certificate_id
158
159 WHERE cac.ca_id = cc.issuer_ca_id
160 AND x509_notBefore(c2.certificate) <= now()
161 AND x509_notAfter(c2.certificate) >= now()
162 AND NOT c2.id = ANY(cc.path)
163
164 AND (
165 x509_authorityKeyId(cc.certificate) IS NULL
166 OR x509_subjectKeyIdentifier(c2.certificate)
167 = x509_authorityKeyId(cc.certificate)
168 )
169
170 ORDER BY
171 CASE
172 WHEN EXISTS (
173 SELECT 1
174 FROM ca_certificate self_ca
175 WHERE self_ca.certificate_id = c2.id
176 AND self_ca.ca_id = c2.issuer_ca_id
177 )
178 THEN 0
179 ELSE 1
180 END,
181 x509_notBefore(c2.certificate) DESC,
182 x509_notAfter(c2.certificate) DESC,
183 c2.id DESC
184
185 LIMIT 1
186 ) parent ON true
187
188 WHERE cc.depth < 6
189
190 AND NOT EXISTS (
191 SELECT 1
192 FROM ca_certificate self_ca
193 WHERE self_ca.certificate_id = cc.certificate_id
194 AND self_ca.ca_id = cc.issuer_ca_id
195 )
196)
197
198SELECT
199 cc.leaf_id,
200 cc.depth,
201 cc.certificate_id,
202 cc.issuer_ca_id,
203
204 x509_commonName(cc.certificate) AS common_name,
205
206 encode(
207 x509_subjectKeyIdentifier(cc.certificate),
208 'hex'
209 ) AS ski,
210
211 encode(
212 x509_authorityKeyId(cc.certificate),
213 'hex'
214 ) AS aki,
215
216 x509_notBefore(cc.certificate)::text AS not_before,
217 x509_notAfter(cc.certificate)::text AS not_after,
218
219 x509_keyAlgorithm(cc.certificate) AS key_algorithm,
220 x509_keySize(cc.certificate) AS key_size,
221
222 x509_signatureKeyAlgorithm(cc.certificate)
223 AS signature_key_algorithm,
224
225 x509_signatureHashAlgorithm(cc.certificate)
226 AS signature_hash_algorithm,
227
228 issuer_ca.name AS issuer_ca_name
229
230FROM cert_chain cc
231
232LEFT JOIN ca issuer_ca
233 ON issuer_ca.id = cc.issuer_ca_id
234
235ORDER BY
236 cc.leaf_id,
237 cc.depth";
238
239fn chain_sql(ct_ids: &[u64]) -> String {
240 let ids = ct_ids
241 .iter()
242 .map(|id| id.to_string())
243 .collect::<Vec<_>>()
244 .join(",");
245 CHAIN_SQL_TEMPLATE.replace("{ids}", &ids)
246}
247
248pub async fn chain_rows(
249 ct_ids: &[u64],
250 cfg: &Config,
251) -> Result<BTreeMap<u64, Vec<ChainEntry>>, String> {
252 if ct_ids.is_empty() {
253 return Ok(BTreeMap::new());
254 }
255 let mut last_error = String::from("not attempted");
256 for attempt in 1..=cfg.retries {
257 eprintln!(
258 "CT: querying certificate chains at {}:{} (attempt {attempt}/{})",
259 cfg.db_host, cfg.db_port, cfg.retries
260 );
261 match chain_attempt_once(ct_ids, cfg).await {
262 Ok(chains) => return Ok(chains),
263 Err(e) => {
264 last_error = e;
265 if attempt < cfg.retries {
266 let delay = db_backoff_secs(attempt, cfg.retry_delay);
267 eprintln!(
268 "CT: {last_error}; retrying in {delay}s (attempt {attempt}/{} failed)",
269 cfg.retries
270 );
271 tokio::time::sleep(Duration::from_secs(delay)).await;
272 }
273 }
274 }
275 }
276 eprintln!("CT: chain query failed after {} attempts", cfg.retries);
277 Err(last_error)
278}
279
280async fn chain_attempt_once(
281 ct_ids: &[u64],
282 cfg: &Config,
283) -> Result<BTreeMap<u64, Vec<ChainEntry>>, String> {
284 let conninfo = format!(
285 "host={} port={} user={} dbname={}",
286 cfg.db_host, cfg.db_port, cfg.db_user, cfg.db_name
287 );
288 let connect = tokio_postgres::connect(&conninfo, NoTls);
289 let (client, connection) =
290 tokio::time::timeout(Duration::from_secs(cfg.db_connect_timeout), connect)
291 .await
292 .map_err(|_| "connect timeout".to_string())?
293 .map_err(db_err)?;
294 let drive = tokio::spawn(async move {
295 if let Err(e) = connection.await {
296 eprintln!("CT: db connection dropped: {}", db_err(e));
297 }
298 });
299
300 let sql = chain_sql(ct_ids);
301 let query = client.simple_query(&sql);
302 let messages = tokio::time::timeout(Duration::from_secs(cfg.db_query_timeout), query)
303 .await
304 .map_err(|_| "query timeout".to_string())?
305 .map_err(db_err)?;
306
307 let mut chains: BTreeMap<u64, Vec<ChainEntry>> = BTreeMap::new();
308 for m in &messages {
309 if let SimpleQueryMessage::Row(r) = m {
310 let (leaf_id, entry) = parse_chain_row(r)?;
311 chains.entry(leaf_id).or_default().push(entry);
312 }
313 }
314
315 drop(client);
316 let _ = drive.await;
317 Ok(chains)
318}
319
320fn parse_chain_row(r: &tokio_postgres::SimpleQueryRow) -> Result<(u64, ChainEntry), String> {
321 let leaf_id: u64 = r
322 .get(0)
323 .ok_or("missing leaf id")?
324 .parse()
325 .map_err(|_| "bad leaf id")?;
326 let entry = ChainEntry {
327 depth: r
328 .get(1)
329 .ok_or("missing depth")?
330 .parse()
331 .map_err(|_| "bad depth")?,
332 certificate_id: r
333 .get(2)
334 .ok_or("missing cert id")?
335 .parse()
336 .map_err(|_| "bad cert id")?,
337 common_name: r.get(4).map(str::to_string),
338 ski: r.get(5).map(str::to_string),
339 aki: r.get(6).map(str::to_string),
340 not_before: r.get(7).and_then(crate::discover::ct::ctdate),
341 not_after: r.get(8).and_then(crate::discover::ct::ctdate),
342 key_algorithm: r.get(9).map(str::to_string),
343 key_size: r.get(10).and_then(|v| v.parse().ok()),
344 signature_key_algorithm: r.get(11).map(str::to_string),
345 signature_hash_algorithm: r.get(12).map(str::to_string),
346 issuer_ca_name: r.get(13).map(str::to_string),
347 };
348 Ok((leaf_id, entry))
349}
350
351async fn query_once(client: &Client, domain: &str, cfg: &Config) -> Result<Vec<DbRow>, String> {
352 let sql = ct_sql(domain)?;
353 let query = client.simple_query(&sql);
354 let rows = tokio::time::timeout(Duration::from_secs(cfg.db_query_timeout), query)
355 .await
356 .map_err(|_| "query timeout".to_string())?
357 .map_err(db_err)?;
358
359 rows.iter()
360 .filter_map(|m| match m {
361 SimpleQueryMessage::Row(r) => Some(r),
362 _ => None,
363 })
364 .map(|r| {
365 Ok(DbRow {
366 cert_id: r
367 .get(0)
368 .ok_or("missing cert id")?
369 .parse()
370 .map_err(|_| "bad cert id")?,
371 serial: r.get(1).ok_or("missing serial")?.to_string(),
372 dns_name: r.get(2).map(str::to_string),
373 common_name: r.get(3).map(str::to_string),
374 not_before: r.get(4).map(str::to_string),
375 not_after: r.get(5).map(str::to_string),
376 key_algorithm: r.get(6).map(str::to_string),
377 key_size: r.get(7).and_then(|v| v.parse().ok()),
378 sig_key_algorithm: r.get(8).map(str::to_string),
379 sig_hash_algorithm: r.get(9).map(str::to_string),
380 issuer_ca_name: r.get(10).map(str::to_string),
381 })
382 })
383 .collect()
384}
385
386fn db_err(e: tokio_postgres::Error) -> String {
387 if let Some(db) = e.as_db_error() {
388 let mut s = format!("db error {:?}: {}", db.code(), db.message());
389 if let Some(detail) = db.detail() {
390 s.push_str(&format!(" ({detail})"));
391 }
392 return s;
393 }
394 format!("db error: {e}")
395}
396
397#[cfg(test)]
398mod tests {
399 use super::*;
400
401 #[test]
402 fn db_backoff_starts_at_2_cubed_and_caps() {
403 assert_eq!(db_backoff_secs(1, 2), 8);
404 assert_eq!(db_backoff_secs(2, 2), 16);
405 assert_eq!(db_backoff_secs(3, 2), 32);
406 assert_eq!(db_backoff_secs(4, 2), 60, "capped at 60s");
407 assert_eq!(db_backoff_secs(5, 2), 60);
408 assert_eq!(db_backoff_secs(1, 120), 60);
409 }
410}