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 struct Session {
77 conninfo: String,
78 client: Option<Client>,
79 drive: Option<tokio::task::JoinHandle<()>>,
80}
81
82impl Session {
83 pub fn new(cfg: &Config) -> Self {
84 Self {
85 conninfo: format!(
86 "host={} port={} user={} dbname={}",
87 cfg.db_host, cfg.db_port, cfg.db_user, cfg.db_name
88 ),
89 client: None,
90 drive: None,
91 }
92 }
93
94 pub async fn fetch_rows(&mut self, domain: &str, cfg: &Config) -> Result<Vec<DbRow>, String> {
95 let mut last_error = String::from("not attempted");
96 for attempt in 1..=cfg.retries {
97 eprintln!(
98 "CT: querying certwatch DB at {}:{} (attempt {attempt}/{})",
99 cfg.db_host, cfg.db_port, cfg.retries
100 );
101 match self.fetch_once(domain, cfg).await {
102 Ok(rows) => return Ok(rows),
103 Err(e) => {
104 last_error = e;
105 if attempt < cfg.retries {
106 let delay = db_backoff_secs(attempt, cfg.retry_delay);
107 eprintln!(
108 "CT: {last_error}; re-running this query in {delay}s \
109 (attempt {attempt}/{} failed)",
110 cfg.retries
111 );
112 tokio::time::sleep(Duration::from_secs(delay)).await;
113 }
114 }
115 }
116 }
117 eprintln!(
118 "CT: certwatch DB unreachable after {} attempts",
119 cfg.retries
120 );
121 Err(last_error)
122 }
123
124 async fn fetch_once(&mut self, domain: &str, cfg: &Config) -> Result<Vec<DbRow>, String> {
125 let client = self.ensure_connected(cfg).await?;
126 let sql = ct_sql(domain)?;
127 let result = {
128 let query = client.simple_query(&sql);
129 tokio::time::timeout(Duration::from_secs(cfg.db_query_timeout), query).await
130 };
131 match result {
132 Ok(Ok(messages)) => parse_db_rows(&messages),
133 Ok(Err(e)) => {
134 self.hang_up();
135 Err(db_err(e))
136 }
137 Err(_) => {
138 self.hang_up();
139 Err("query timeout (no response)".to_string())
140 }
141 }
142 }
143
144 pub async fn chain_rows(
145 &mut self,
146 ct_ids: &[u64],
147 cfg: &Config,
148 ) -> Result<BTreeMap<u64, Vec<ChainEntry>>, String> {
149 if ct_ids.is_empty() {
150 return Ok(BTreeMap::new());
151 }
152 let mut last_error = String::from("not attempted");
153 for attempt in 1..=cfg.retries {
154 eprintln!(
155 "CT: querying certificate chains at {}:{} (attempt {attempt}/{})",
156 cfg.db_host, cfg.db_port, cfg.retries
157 );
158 match self.chain_once(ct_ids, cfg).await {
159 Ok(chains) => return Ok(chains),
160 Err(e) => {
161 last_error = e;
162 if attempt < cfg.retries {
163 let delay = db_backoff_secs(attempt, cfg.retry_delay);
164 eprintln!(
165 "CT: {last_error}; re-running this query in {delay}s \
166 (attempt {attempt}/{} failed)",
167 cfg.retries
168 );
169 tokio::time::sleep(Duration::from_secs(delay)).await;
170 }
171 }
172 }
173 }
174 eprintln!("CT: chain query failed after {} attempts", cfg.retries);
175 Err(last_error)
176 }
177
178 async fn chain_once(
179 &mut self,
180 ct_ids: &[u64],
181 cfg: &Config,
182 ) -> Result<BTreeMap<u64, Vec<ChainEntry>>, String> {
183 let client = self.ensure_connected(cfg).await?;
184 let sql = chain_sql(ct_ids);
185 let result = {
186 let query = client.simple_query(&sql);
187 tokio::time::timeout(Duration::from_secs(cfg.db_query_timeout), query).await
188 };
189 match result {
190 Ok(Ok(messages)) => parse_chain_rows(&messages),
191 Ok(Err(e)) => {
192 self.hang_up();
193 Err(db_err(e))
194 }
195 Err(_) => {
196 self.hang_up();
197 Err("query timeout (no response)".to_string())
198 }
199 }
200 }
201
202 async fn ensure_connected(&mut self, cfg: &Config) -> Result<&Client, String> {
203 if self.client.is_none() {
204 eprintln!(
205 "CT: connecting to certwatch DB at {}:{}",
206 cfg.db_host, cfg.db_port
207 );
208 let connect = tokio_postgres::connect(&self.conninfo, NoTls);
209 let (client, connection) =
210 tokio::time::timeout(Duration::from_secs(cfg.db_connect_timeout), connect)
211 .await
212 .map_err(|_| "connect timeout".to_string())?
213 .map_err(db_err)?;
214 self.drive = Some(tokio::spawn(async move {
215 if let Err(e) = connection.await {
216 eprintln!("CT: db connection dropped: {}", db_err(e));
217 }
218 }));
219 self.client = Some(client);
220 }
221 Ok(self.client.as_ref().unwrap())
222 }
223
224 pub fn close(&mut self) {
225 self.hang_up();
226 }
227
228 fn hang_up(&mut self) {
229 if let Some(drive) = self.drive.take() {
230 drive.abort();
231 }
232 self.client = None;
233 }
234}
235
236impl Drop for Session {
237 fn drop(&mut self) {
238 self.hang_up();
239 }
240}
241
242pub fn db_backoff_secs(attempt: u32, retry_delay: u64) -> u64 {
243 retry_delay
244 .saturating_mul(1u64 << (attempt + 1).min(6))
245 .min(60)
246}
247
248const CHAIN_SQL_TEMPLATE: &str = "
249WITH RECURSIVE cert_chain AS (
250 SELECT
251 c.id AS certificate_id,
252 c.issuer_ca_id,
253 c.certificate,
254 c.id AS leaf_id,
255 0 AS depth,
256 ARRAY[c.id]::bigint[] AS path
257 FROM certificate c
258 WHERE c.id = ANY(ARRAY[{ids}]::bigint[])
259 AND x509_notBefore(c.certificate) <= now()
260 AND x509_notAfter(c.certificate) >= now()
261
262 UNION ALL
263
264 SELECT
265 parent.id,
266 parent.issuer_ca_id,
267 parent.certificate,
268 cc.leaf_id,
269 cc.depth + 1,
270 cc.path || parent.id
271 FROM cert_chain cc
272
273 JOIN LATERAL (
274 SELECT c2.*
275 FROM ca_certificate cac
276 JOIN certificate c2
277 ON c2.id = cac.certificate_id
278
279 WHERE cac.ca_id = cc.issuer_ca_id
280 AND x509_notBefore(c2.certificate) <= now()
281 AND x509_notAfter(c2.certificate) >= now()
282 AND NOT c2.id = ANY(cc.path)
283
284 AND (
285 x509_authorityKeyId(cc.certificate) IS NULL
286 OR x509_subjectKeyIdentifier(c2.certificate)
287 = x509_authorityKeyId(cc.certificate)
288 )
289
290 ORDER BY
291 CASE
292 WHEN EXISTS (
293 SELECT 1
294 FROM ca_certificate self_ca
295 WHERE self_ca.certificate_id = c2.id
296 AND self_ca.ca_id = c2.issuer_ca_id
297 )
298 THEN 0
299 ELSE 1
300 END,
301 x509_notBefore(c2.certificate) DESC,
302 x509_notAfter(c2.certificate) DESC,
303 c2.id DESC
304
305 LIMIT 1
306 ) parent ON true
307
308 WHERE cc.depth < 6
309
310 AND NOT EXISTS (
311 SELECT 1
312 FROM ca_certificate self_ca
313 WHERE self_ca.certificate_id = cc.certificate_id
314 AND self_ca.ca_id = cc.issuer_ca_id
315 )
316)
317
318SELECT
319 cc.leaf_id,
320 cc.depth,
321 cc.certificate_id,
322 cc.issuer_ca_id,
323
324 x509_commonName(cc.certificate) AS common_name,
325
326 encode(
327 x509_subjectKeyIdentifier(cc.certificate),
328 'hex'
329 ) AS ski,
330
331 encode(
332 x509_authorityKeyId(cc.certificate),
333 'hex'
334 ) AS aki,
335
336 x509_notBefore(cc.certificate)::text AS not_before,
337 x509_notAfter(cc.certificate)::text AS not_after,
338
339 x509_keyAlgorithm(cc.certificate) AS key_algorithm,
340 x509_keySize(cc.certificate) AS key_size,
341
342 x509_signatureKeyAlgorithm(cc.certificate)
343 AS signature_key_algorithm,
344
345 x509_signatureHashAlgorithm(cc.certificate)
346 AS signature_hash_algorithm,
347
348 issuer_ca.name AS issuer_ca_name
349
350FROM cert_chain cc
351
352LEFT JOIN ca issuer_ca
353 ON issuer_ca.id = cc.issuer_ca_id
354
355ORDER BY
356 cc.leaf_id,
357 cc.depth";
358
359fn chain_sql(ct_ids: &[u64]) -> String {
360 let ids = ct_ids
361 .iter()
362 .map(|id| id.to_string())
363 .collect::<Vec<_>>()
364 .join(",");
365 CHAIN_SQL_TEMPLATE.replace("{ids}", &ids)
366}
367
368fn parse_chain_row(r: &tokio_postgres::SimpleQueryRow) -> Result<(u64, ChainEntry), String> {
369 let leaf_id: u64 = r
370 .get(0)
371 .ok_or("missing leaf id")?
372 .parse()
373 .map_err(|_| "bad leaf id")?;
374 let entry = ChainEntry {
375 depth: r
376 .get(1)
377 .ok_or("missing depth")?
378 .parse()
379 .map_err(|_| "bad depth")?,
380 certificate_id: r
381 .get(2)
382 .ok_or("missing cert id")?
383 .parse()
384 .map_err(|_| "bad cert id")?,
385 common_name: r.get(4).map(str::to_string),
386 ski: r.get(5).map(str::to_string),
387 aki: r.get(6).map(str::to_string),
388 not_before: r.get(7).and_then(crate::discover::ct::ctdate),
389 not_after: r.get(8).and_then(crate::discover::ct::ctdate),
390 key_algorithm: r.get(9).map(str::to_string),
391 key_size: r.get(10).and_then(|v| v.parse().ok()),
392 signature_key_algorithm: r.get(11).map(str::to_string),
393 signature_hash_algorithm: r.get(12).map(str::to_string),
394 issuer_ca_name: r.get(13).map(str::to_string),
395 };
396 Ok((leaf_id, entry))
397}
398
399fn parse_db_rows(messages: &[SimpleQueryMessage]) -> Result<Vec<DbRow>, String> {
400 messages
401 .iter()
402 .filter_map(|m| match m {
403 SimpleQueryMessage::Row(r) => Some(r),
404 _ => None,
405 })
406 .map(|r| {
407 Ok(DbRow {
408 cert_id: r
409 .get(0)
410 .ok_or("missing cert id")?
411 .parse()
412 .map_err(|_| "bad cert id")?,
413 serial: r.get(1).ok_or("missing serial")?.to_string(),
414 dns_name: r.get(2).map(str::to_string),
415 common_name: r.get(3).map(str::to_string),
416 not_before: r.get(4).map(str::to_string),
417 not_after: r.get(5).map(str::to_string),
418 key_algorithm: r.get(6).map(str::to_string),
419 key_size: r.get(7).and_then(|v| v.parse().ok()),
420 sig_key_algorithm: r.get(8).map(str::to_string),
421 sig_hash_algorithm: r.get(9).map(str::to_string),
422 issuer_ca_name: r.get(10).map(str::to_string),
423 })
424 })
425 .collect()
426}
427
428fn parse_chain_rows(
429 messages: &[SimpleQueryMessage],
430) -> Result<BTreeMap<u64, Vec<ChainEntry>>, String> {
431 let mut chains: BTreeMap<u64, Vec<ChainEntry>> = BTreeMap::new();
432 for m in messages {
433 if let SimpleQueryMessage::Row(r) = m {
434 let (leaf_id, entry) = parse_chain_row(r)?;
435 chains.entry(leaf_id).or_default().push(entry);
436 }
437 }
438 Ok(chains)
439}
440
441fn db_err(e: tokio_postgres::Error) -> String {
442 if let Some(db) = e.as_db_error() {
443 let mut s = format!("db error {:?}: {}", db.code(), db.message());
444 if let Some(detail) = db.detail() {
445 s.push_str(&format!(" ({detail})"));
446 }
447 return s;
448 }
449 format!("db error: {e}")
450}
451
452#[cfg(test)]
453mod tests {
454 use super::*;
455
456 #[test]
457 fn db_backoff_starts_at_2_cubed_and_caps() {
458 assert_eq!(db_backoff_secs(1, 2), 8);
459 assert_eq!(db_backoff_secs(2, 2), 16);
460 assert_eq!(db_backoff_secs(3, 2), 32);
461 assert_eq!(db_backoff_secs(4, 2), 60, "capped at 60s");
462 assert_eq!(db_backoff_secs(5, 2), 60);
463 assert_eq!(db_backoff_secs(1, 120), 60);
464 }
465}