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 fn hang_up(&mut self) {
225 if let Some(drive) = self.drive.take() {
226 drive.abort();
227 }
228 self.client = None;
229 }
230}
231
232impl Drop for Session {
233 fn drop(&mut self) {
234 self.hang_up();
235 }
236}
237
238pub fn db_backoff_secs(attempt: u32, retry_delay: u64) -> u64 {
239 retry_delay
240 .saturating_mul(1u64 << (attempt + 1).min(6))
241 .min(60)
242}
243
244const CHAIN_SQL_TEMPLATE: &str = "
245WITH RECURSIVE cert_chain AS (
246 SELECT
247 c.id AS certificate_id,
248 c.issuer_ca_id,
249 c.certificate,
250 c.id AS leaf_id,
251 0 AS depth,
252 ARRAY[c.id]::bigint[] AS path
253 FROM certificate c
254 WHERE c.id = ANY(ARRAY[{ids}]::bigint[])
255 AND x509_notBefore(c.certificate) <= now()
256 AND x509_notAfter(c.certificate) >= now()
257
258 UNION ALL
259
260 SELECT
261 parent.id,
262 parent.issuer_ca_id,
263 parent.certificate,
264 cc.leaf_id,
265 cc.depth + 1,
266 cc.path || parent.id
267 FROM cert_chain cc
268
269 JOIN LATERAL (
270 SELECT c2.*
271 FROM ca_certificate cac
272 JOIN certificate c2
273 ON c2.id = cac.certificate_id
274
275 WHERE cac.ca_id = cc.issuer_ca_id
276 AND x509_notBefore(c2.certificate) <= now()
277 AND x509_notAfter(c2.certificate) >= now()
278 AND NOT c2.id = ANY(cc.path)
279
280 AND (
281 x509_authorityKeyId(cc.certificate) IS NULL
282 OR x509_subjectKeyIdentifier(c2.certificate)
283 = x509_authorityKeyId(cc.certificate)
284 )
285
286 ORDER BY
287 CASE
288 WHEN EXISTS (
289 SELECT 1
290 FROM ca_certificate self_ca
291 WHERE self_ca.certificate_id = c2.id
292 AND self_ca.ca_id = c2.issuer_ca_id
293 )
294 THEN 0
295 ELSE 1
296 END,
297 x509_notBefore(c2.certificate) DESC,
298 x509_notAfter(c2.certificate) DESC,
299 c2.id DESC
300
301 LIMIT 1
302 ) parent ON true
303
304 WHERE cc.depth < 6
305
306 AND NOT EXISTS (
307 SELECT 1
308 FROM ca_certificate self_ca
309 WHERE self_ca.certificate_id = cc.certificate_id
310 AND self_ca.ca_id = cc.issuer_ca_id
311 )
312)
313
314SELECT
315 cc.leaf_id,
316 cc.depth,
317 cc.certificate_id,
318 cc.issuer_ca_id,
319
320 x509_commonName(cc.certificate) AS common_name,
321
322 encode(
323 x509_subjectKeyIdentifier(cc.certificate),
324 'hex'
325 ) AS ski,
326
327 encode(
328 x509_authorityKeyId(cc.certificate),
329 'hex'
330 ) AS aki,
331
332 x509_notBefore(cc.certificate)::text AS not_before,
333 x509_notAfter(cc.certificate)::text AS not_after,
334
335 x509_keyAlgorithm(cc.certificate) AS key_algorithm,
336 x509_keySize(cc.certificate) AS key_size,
337
338 x509_signatureKeyAlgorithm(cc.certificate)
339 AS signature_key_algorithm,
340
341 x509_signatureHashAlgorithm(cc.certificate)
342 AS signature_hash_algorithm,
343
344 issuer_ca.name AS issuer_ca_name
345
346FROM cert_chain cc
347
348LEFT JOIN ca issuer_ca
349 ON issuer_ca.id = cc.issuer_ca_id
350
351ORDER BY
352 cc.leaf_id,
353 cc.depth";
354
355fn chain_sql(ct_ids: &[u64]) -> String {
356 let ids = ct_ids
357 .iter()
358 .map(|id| id.to_string())
359 .collect::<Vec<_>>()
360 .join(",");
361 CHAIN_SQL_TEMPLATE.replace("{ids}", &ids)
362}
363
364fn parse_chain_row(r: &tokio_postgres::SimpleQueryRow) -> Result<(u64, ChainEntry), String> {
365 let leaf_id: u64 = r
366 .get(0)
367 .ok_or("missing leaf id")?
368 .parse()
369 .map_err(|_| "bad leaf id")?;
370 let entry = ChainEntry {
371 depth: r
372 .get(1)
373 .ok_or("missing depth")?
374 .parse()
375 .map_err(|_| "bad depth")?,
376 certificate_id: r
377 .get(2)
378 .ok_or("missing cert id")?
379 .parse()
380 .map_err(|_| "bad cert id")?,
381 common_name: r.get(4).map(str::to_string),
382 ski: r.get(5).map(str::to_string),
383 aki: r.get(6).map(str::to_string),
384 not_before: r.get(7).and_then(crate::discover::ct::ctdate),
385 not_after: r.get(8).and_then(crate::discover::ct::ctdate),
386 key_algorithm: r.get(9).map(str::to_string),
387 key_size: r.get(10).and_then(|v| v.parse().ok()),
388 signature_key_algorithm: r.get(11).map(str::to_string),
389 signature_hash_algorithm: r.get(12).map(str::to_string),
390 issuer_ca_name: r.get(13).map(str::to_string),
391 };
392 Ok((leaf_id, entry))
393}
394
395fn parse_db_rows(messages: &[SimpleQueryMessage]) -> Result<Vec<DbRow>, String> {
396 messages
397 .iter()
398 .filter_map(|m| match m {
399 SimpleQueryMessage::Row(r) => Some(r),
400 _ => None,
401 })
402 .map(|r| {
403 Ok(DbRow {
404 cert_id: r
405 .get(0)
406 .ok_or("missing cert id")?
407 .parse()
408 .map_err(|_| "bad cert id")?,
409 serial: r.get(1).ok_or("missing serial")?.to_string(),
410 dns_name: r.get(2).map(str::to_string),
411 common_name: r.get(3).map(str::to_string),
412 not_before: r.get(4).map(str::to_string),
413 not_after: r.get(5).map(str::to_string),
414 key_algorithm: r.get(6).map(str::to_string),
415 key_size: r.get(7).and_then(|v| v.parse().ok()),
416 sig_key_algorithm: r.get(8).map(str::to_string),
417 sig_hash_algorithm: r.get(9).map(str::to_string),
418 issuer_ca_name: r.get(10).map(str::to_string),
419 })
420 })
421 .collect()
422}
423
424fn parse_chain_rows(
425 messages: &[SimpleQueryMessage],
426) -> Result<BTreeMap<u64, Vec<ChainEntry>>, String> {
427 let mut chains: BTreeMap<u64, Vec<ChainEntry>> = BTreeMap::new();
428 for m in messages {
429 if let SimpleQueryMessage::Row(r) = m {
430 let (leaf_id, entry) = parse_chain_row(r)?;
431 chains.entry(leaf_id).or_default().push(entry);
432 }
433 }
434 Ok(chains)
435}
436
437fn db_err(e: tokio_postgres::Error) -> String {
438 if let Some(db) = e.as_db_error() {
439 let mut s = format!("db error {:?}: {}", db.code(), db.message());
440 if let Some(detail) = db.detail() {
441 s.push_str(&format!(" ({detail})"));
442 }
443 return s;
444 }
445 format!("db error: {e}")
446}
447
448#[cfg(test)]
449mod tests {
450 use super::*;
451
452 #[test]
453 fn db_backoff_starts_at_2_cubed_and_caps() {
454 assert_eq!(db_backoff_secs(1, 2), 8);
455 assert_eq!(db_backoff_secs(2, 2), 16);
456 assert_eq!(db_backoff_secs(3, 2), 32);
457 assert_eq!(db_backoff_secs(4, 2), 60, "capped at 60s");
458 assert_eq!(db_backoff_secs(5, 2), 60);
459 assert_eq!(db_backoff_secs(1, 120), 60);
460 }
461}