1
//! Access to the database schema.
2
//!
3
//! This module is not intended to provide a high-level ORM, instead it serves
4
//! the purpose of initializing and upgrading the database, if necessary.
5
//!
6
//! # Synchronous or Asynchronous?
7
//!
8
//! The question on whether the database and access to it shall be synchronous
9
//! or asynchronous has been fairly long debate that eventually got settled
10
//! after realizing that an asynchronous approach does not work.  This comment
11
//! should serve as a reminder for future devs, wondering why we use certain
12
//! synchronous primitives in an otherwise asynchronous codebase.
13
//!
14
//! Early on, it was clear that we would need some sort of connection pool,
15
//! primarily for two reasons:
16
//! 1. Performing frequent open and close calls in every task would be costly.
17
//! 2. Sharing a single connection object with a Mutex would be a waste
18
//!
19
//! Because the application itself is primarily asynchronous, we decided to go
20
//! with an asynchronous connection pool as well, leading to the choose of
21
//! `deadpool` initially.
22
//!
23
//! However, soon thereafter, problems with `deadpool` became evident.  Those
24
//! problems mostly stemmed from the synchronous nature of SQLite itself.  In our
25
//! case, this problem was initially triggered by figuring out a way to solve
26
//! `SQLITE_BUSY` handling.  In the end, we decided to settle upon the following
27
//! approach: Set `PRAGMA busy_timeout` to a certain value and create write
28
//! transactions with `BEGIN EXCLUSIVE`.  This way, SQLite would try to obtain
29
//! a write transaction for `busy_timeout` milliseconds by blocking the current
30
//! thread.  Due to this blocking, async no longer made any sense and was in
31
//! fact quite counter-productive because those potential sleep could screw a
32
//! lot of things up, which became very evident while trying to test this.
33
//!
34
//! Besides, throughout refactoring the code base, we realized that, even while
35
//! still using `deadpool`, the actual "asynchronous" calls interfacing with the
36
//! database became smaller and smaller.  In the end, the asynchronous code just
37
//! involved parts of obtaining a connection and creating a transaction,
38
//! eventually resulting in a calling a synchronous function taking the
39
//! transaction handle to perform the lion's share of the operation.
40

            
41
// TODO DIRMIRROR: This could benefit from methods by wrapping the pool into a
42
// custom type.
43

            
44
use std::{
45
    collections::HashSet,
46
    fmt::Display,
47
    io::{Cursor, Write},
48
    num::NonZero,
49
    ops::{Add, Sub},
50
    path::Path,
51
    time::{Duration, SystemTime},
52
};
53

            
54
use digest::Digest;
55
use flate2::write::{DeflateEncoder, GzEncoder};
56
use r2d2::Pool;
57
use r2d2_sqlite::SqliteConnectionManager;
58
use rand::Rng;
59
use rusqlite::{
60
    OptionalExtension, ToSql, Transaction, TransactionBehavior, named_params, params,
61
    types::{FromSql, FromSqlError, FromSqlResult, ToSqlOutput, ValueRef},
62
};
63
use saturating_time::SaturatingTime;
64
use tor_basic_utils::RngExt;
65
use tor_dircommon::config::DirTolerance;
66
use tor_error::into_internal;
67
use tor_netdoc::doc::{
68
    authcert::{AuthCert, AuthCertKeyIds},
69
    netstatus::ConsensusFlavor,
70
};
71

            
72
use crate::err::DatabaseError;
73

            
74
/// Version 1 of the database schema.
75
///
76
/// TODO DIRMIRROR: Before the release, figure out where to use rowid and where
77
/// to use docid.
78
const V1_SCHEMA: &str = include_str!("schema_v1.sql");
79

            
80
/// Global options set in every connection.
81
const GLOBAL_OPTIONS: &str = sql!(
82
    "
83
PRAGMA journal_mode=WAL;
84
PRAGMA foreign_keys=ON;
85
PRAGMA busy_timeout=1000;
86
"
87
);
88

            
89
/// Convenience macro for implementing a hash type in a rusqlite compatible fashion.
90
///
91
/// This macro accepts the following parameters:
92
/// 1. `name` for specifying an identifier of the type, such as [`Sha256`].
93
/// 2. `algo` for specifying the type from the rust-crypto [`digest`] ecosystem,
94
///    such as [`tor_llcrypto::d::Sha256`].
95
/// 3. The size in bytes of the hash output, such as `32` for [`Sha256`].
96
///     * Unfortunately, we cannot use something like [`Digest::output_size()`]
97
///       because it is not a constant.
98
///
99
/// It generates a struct with `name` as the identifier, which implements the
100
/// following methods:
101
/// * `digest` for wrapping around [`Digest::digest()`].
102
///
103
/// It also implements the following traits:
104
/// * [`Display`]
105
/// * [`FromSql`]
106
/// * [`ToSql`]
107
/// * [`PartialEq<&str>`] for base16 comparisons
108
/// * [`From<u8; $size>`] but only in tests
109
macro_rules! impl_hash_wrapper {
110
    ($name:ident, $algo:ty, $size:literal) => {
111
        /// Database wrapper type.
112
        #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
113
        pub(crate) struct $name([u8; $size]);
114

            
115
        impl $name {
116
            /// Computes the hash from arbitrary data.
117
388
            pub(crate) fn digest(data: &[u8]) -> Self {
118
388
                Self(<$algo>::digest(data).into())
119
388
            }
120
        }
121

            
122
        impl Display for $name {
123
            /// Formats the hash in uppercase hexadecimal.
124
764
            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
125
764
                write!(f, "{}", hex::encode_upper(self.0))
126
764
            }
127
        }
128

            
129
        impl FromSql for $name {
130
176
            fn column_result(value: ValueRef<'_>) -> FromSqlResult<Self> {
131
                // We read the hash as a hexadecimal string from the database.
132
                // Convert it to binary data and check length afterwards.
133
176
                let data: [u8; $size] = value
134
176
                    .as_str()
135
176
                    .map(hex::decode)?
136
176
                    .map_err(|e| {
137
                        FromSqlError::Other(Box::new(tor_error::internal!(
138
                            "non hex data in database? {e}"
139
                        )))
140
                    })?
141
176
                    .try_into()
142
176
                    .map_err(|_| {
143
                        FromSqlError::Other(Box::new(tor_error::internal!(
144
                            "$name with invalid length in database?"
145
                        )))
146
                    })?;
147

            
148
176
                Ok(Self(data))
149
176
            }
150
        }
151

            
152
        impl ToSql for $name {
153
762
            fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
154
                // Because Self is only constructed with FromSql and digest
155
                // data, it is safe to assume it is valid.
156
762
                Ok(ToSqlOutput::from(self.to_string()))
157
762
            }
158
        }
159

            
160
        impl PartialEq<&str> for $name {
161
2
            fn eq(&self, other: &&str) -> bool {
162
2
                self.to_string() == other.to_uppercase()
163
2
            }
164
        }
165

            
166
        #[cfg(test)]
167
        impl From<[u8; $size]> for $name {
168
40
            fn from(value: [u8; $size]) -> Self {
169
40
                Self(value)
170
40
            }
171
        }
172
    };
173
}
174

            
175
impl_hash_wrapper!(Sha1, tor_llcrypto::d::Sha1, 20);
176
impl_hash_wrapper!(Sha256, tor_llcrypto::d::Sha256, 32);
177
impl_hash_wrapper!(Sha3_256, tor_llcrypto::d::Sha3_256, 32);
178

            
179
/// The identifier for documents in the content-addressable cache.
180
///
181
/// Right now, this is a [`Sha256`] hash, but this may change in future.
182
pub(crate) type DocumentId = Sha256;
183

            
184
/// The supported content encodings.
185
#[derive(Debug, Clone, Copy, PartialEq, strum::EnumString, strum::Display, strum::EnumIter)]
186
#[strum(serialize_all = "kebab-case", ascii_case_insensitive)]
187
pub(crate) enum ContentEncoding {
188
    /// RFC2616 section 3.5.
189
    Identity,
190
    /// RFC2616 section 3.5.
191
    Deflate,
192
    /// RFC2616 section 3.5.
193
    Gzip,
194
    /// The zstandard compression algorithm (www.zstd.net).
195
    XZstd,
196
    /// The lzma compression algorithm with a "present" value no higher than 6.
197
    XTorLzma,
198
}
199

            
200
/// A wrapper around [`SystemTime`] with convenient features.
201
///
202
/// Please use this type throughout the crate internally, instead of
203
/// [`SystemTime`].
204
///
205
/// # Conversion
206
///
207
/// This type can be safely converted from and into a [`SystemTime`], because
208
/// it is just a wrapper type.
209
///
210
/// # Saturating Arithmetic
211
///
212
/// This type implements [`Add`] and [`Sub`] for [`Duration`] and [`Timestamp`]
213
/// ([`Sub`] only) using saturating arithmetic from the [`saturating_time`]
214
/// crate.  It means that addition and subtraction can be safely performed
215
/// without the potential risk of an unexpected panic, instead wrapping to
216
/// a local maximum/minimum or [`Duration::ZERO`] depending on the type.
217
///
218
/// Note that we don't provide a saturating version of [`Duration`], so addition
219
/// or subtraction of two [`Duration`]s still needs care to avoid panics.
220
///
221
/// # SQLite Interaction
222
///
223
/// This type implements [`FromSql`] and [`ToSql`], making it convenient to
224
/// integrate into SQL statements, as the database schema represents timestamps
225
/// internally using a non-negative [`i64`] storing the seconds since the epoch.
226
#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)]
227
pub(crate) struct Timestamp(SystemTime);
228

            
229
impl From<SystemTime> for Timestamp {
230
60
    fn from(value: SystemTime) -> Self {
231
60
        Self(value)
232
60
    }
233
}
234

            
235
impl From<Timestamp> for SystemTime {
236
18
    fn from(value: Timestamp) -> Self {
237
18
        value.0
238
18
    }
239
}
240

            
241
impl Add<Duration> for Timestamp {
242
    type Output = Self;
243

            
244
    /// Performs a saturating addition wrapping to [`SystemTime::max_value()`].
245
20020
    fn add(self, rhs: Duration) -> Self::Output {
246
20020
        Self(self.0.saturating_add(rhs))
247
20020
    }
248
}
249

            
250
impl Sub<Duration> for Timestamp {
251
    type Output = Self;
252

            
253
    /// Performs a saturating subtraction wrapping to [`SystemTime::min_value()`].
254
4
    fn sub(self, rhs: Duration) -> Self::Output {
255
4
        Self(self.0.saturating_sub(rhs))
256
4
    }
257
}
258

            
259
impl Sub<Timestamp> for Timestamp {
260
    type Output = Duration;
261

            
262
    /// Performs a saturating duration_since wrapping to [`Duration::ZERO`].
263
20002
    fn sub(self, rhs: Timestamp) -> Self::Output {
264
        #[allow(unstable_name_collisions)]
265
20002
        self.0.saturating_duration_since(rhs.0)
266
20002
    }
267
}
268

            
269
impl FromSql for Timestamp {
270
130
    fn column_result(value: ValueRef<'_>) -> FromSqlResult<Self> {
271
130
        let mut res = SystemTime::UNIX_EPOCH;
272
130
        res = res.saturating_add(Duration::from_secs(value.as_i64()?.try_into().unwrap_or(0)));
273
130
        Ok(Self(res))
274
130
    }
275
}
276

            
277
impl ToSql for Timestamp {
278
202
    fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
279
        #[allow(unstable_name_collisions)]
280
202
        Ok(ToSqlOutput::from(
281
202
            self.0
282
202
                .saturating_duration_since(SystemTime::UNIX_EPOCH)
283
202
                .as_secs()
284
202
                .try_into()
285
202
                .unwrap_or(i64::MAX),
286
202
        ))
287
202
    }
288
}
289

            
290
/// Representation of consensus metadata from the database.
291
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
292
pub(crate) struct ConsensusMeta {
293
    /// The document id uniquely identifying the consensus.
294
    pub docid: DocumentId,
295

            
296
    /// The SHA3 of the unsigned part of the consensus.
297
    pub unsigned_sha3_256: Sha3_256,
298

            
299
    /// The flavor of the consensus.
300
    pub flavor: ConsensusFlavor,
301

            
302
    /// The time after which this consensus is valid.
303
    pub valid_after: Timestamp,
304

            
305
    /// The time after which this consensus stops being fresh.
306
    pub fresh_until: Timestamp,
307

            
308
    /// The time after which this consensus stops being valid.
309
    pub valid_until: Timestamp,
310
}
311

            
312
impl ConsensusMeta {
313
    /// Obtains the most recent valid consensus from the database.
314
    ///
315
    /// This function queries the database using a [`Transaction`] in order to
316
    /// have a consistent view upon it.  It will return an [`Option`] containing
317
    /// a consensus.  In order to obtain a *valid* consensus, a [`Timestamp`]
318
    /// plus a [`DirTolerance`] are supplied, which will be used for querying
319
    /// the database in a time-constrained fashion.
320
    ///
321
    /// The [`None`] case implies that no valid consensus has been found, that
322
    /// is, no consensus at all or no consensus whose `valid-before` or
323
    /// `valid-after` lies within the range composed by `now` and `tolerance`.
324
28
    pub(crate) fn query_recent(
325
28
        tx: &Transaction,
326
28
        flavor: ConsensusFlavor,
327
28
        tolerance: &DirTolerance,
328
28
        now: Timestamp,
329
28
    ) -> Result<Option<Self>, DatabaseError> {
330
        // Select the most recent flavored consensus document from the database.
331
        //
332
        // The `valid_after` and `valid_until` cells must be a member of the range:
333
        // `[valid_after - pre_valid_tolerance; valid_after + post_valid_tolerance]`
334
        // (inclusively).
335
28
        let mut meta_stmt = tx.prepare_cached(sql!(
336
28
            "
337
28
            SELECT docid, unsigned_sha3_256, valid_after, fresh_until, valid_until
338
28
            FROM consensus
339
28
            WHERE
340
28
              flavor = :flavor
341
28
              AND :now >= valid_after - :pre_valid
342
28
              AND :now <= valid_until + :post_valid
343
28
            ORDER BY valid_after DESC
344
28
            LIMIT 1
345
28
            "
346
28
        ))?;
347

            
348
        // Actually execute the query; a None is totally valid and considered as
349
        // no consensus being present in the current database.
350
28
        let res = meta_stmt.query_one(named_params! {
351
28
            ":flavor": flavor.name(),
352
28
            ":now": now,
353
28
            ":pre_valid": tolerance.pre_valid_tolerance().as_secs().try_into().unwrap_or(i64::MAX),
354
28
            ":post_valid": tolerance.post_valid_tolerance().as_secs().try_into().unwrap_or(i64::MAX),
355
18
        }, |row| {
356
            Ok(Self {
357
18
                docid: row.get(0)?,
358
18
                unsigned_sha3_256: row.get(1)?,
359
18
                flavor,
360
18
                valid_after: row.get(2)?,
361
18
                fresh_until: row.get(3)?,
362
18
                valid_until: row.get(4)?,
363
            })
364
37
        }).optional()?;
365

            
366
28
        Ok(res)
367
28
    }
368

            
369
    /// Queries the raw data of a [`ConsensusMeta`].
370
2
    pub(crate) fn data(&self, tx: &Transaction<'_>) -> Result<String, DatabaseError> {
371
2
        let mut stmt = tx.prepare_cached(sql!(
372
2
            "
373
2
            SELECT content
374
2
            FROM store
375
2
            WHERE docid = :docid
376
2
            "
377
2
        ))?;
378

            
379
3
        let raw = stmt.query_one(named_params! {":docid": self.docid}, |row| {
380
2
            row.get::<_, Vec<u8>>(0)
381
2
        })?;
382
2
        let raw = String::from_utf8(raw).map_err(into_internal!("utf-8 constraint violated?"))?;
383
2
        Ok(raw)
384
2
    }
385

            
386
    /// Calculates the [`Timestamp`] at which the authorities will be queried again.
387
    ///
388
    /// # Specifications
389
    ///
390
    /// * <https://spec.torproject.org/dir-spec/directory-cache-operation.html#download-ns-from-auth>
391
20002
    pub(crate) fn lifetime<R: Rng>(&self, rng: &mut R) -> Timestamp {
392
20002
        assert!(self.fresh_until < self.valid_until);
393

            
394
20002
        let offset = rng
395
20002
            .gen_range_checked(0..=((self.valid_until - self.fresh_until).as_secs() / 2))
396
20002
            .expect("invalid range?");
397

            
398
20002
        self.fresh_until + Duration::from_secs(offset)
399
20002
    }
400

            
401
    /// Returns the missing server descriptors for this consensus.
402
6
    pub(crate) fn missing_servers(
403
6
        &self,
404
6
        tx: &Transaction<'_>,
405
6
    ) -> Result<HashSet<Sha1>, DatabaseError> {
406
6
        if self.flavor != ConsensusFlavor::Plain {
407
            return Ok(HashSet::new());
408
6
        }
409

            
410
        // Select the missing router descriptors.
411
        //
412
        // A router descriptor is considered missing if it exists in
413
        // `consensus_router_descriptor_member` but not in `router_descriptor`
414
        // because the first entry is added once the consensus got parsed,
415
        // whereas the second entry is added once we have actually retrieved it.
416
        //
417
        // It works by doing a left join on router_descriptor and filtering for
418
        // all entries where the join is NULL, as that implies we are aware of
419
        // the descriptor but not have it stored.
420
        //
421
        // Parameters:
422
        // :docid - The docid of the consensus.
423
6
        let mut stmt = tx.prepare_cached(sql!(
424
6
            "
425
6
            SELECT cr.unsigned_sha1
426
6
            FROM consensus_router_descriptor_member AS cr
427
6
              LEFT JOIN router_descriptor AS server ON cr.unsigned_sha1 = server.unsigned_sha1
428
6
            WHERE
429
6
              cr.consensus_docid = :docid
430
6
              AND cr.unsigned_sha1 IS NOT NULL
431
6
              AND server.unsigned_sha1 IS NULL
432
6
            "
433
6
        ))?;
434

            
435
6
        let missing = stmt
436
11
            .query_map(named_params! {":docid": self.docid}, |row| row.get(0))?
437
6
            .collect::<Result<HashSet<_>, _>>()?;
438
6
        Ok(missing)
439
6
    }
440

            
441
    /// Returns the missing extra infos for this consensus to the best of our abilities.
442
    ///
443
    /// Keep in mind that this does not return **all** missing extra infos but
444
    /// only the missing extra infos of server descriptors we have.
445
6
    pub(crate) fn missing_extras(
446
6
        &self,
447
6
        tx: &Transaction<'_>,
448
6
    ) -> Result<HashSet<Sha1>, DatabaseError> {
449
6
        if self.flavor != ConsensusFlavor::Plain {
450
            return Ok(HashSet::new());
451
6
        }
452

            
453
        // Select the missing extra infos for this consensus.
454
        //
455
        // This return value is not complete because we only know the missing
456
        // extra-infos to the best of our abilities.  In other words: We are
457
        // only aware of a missing extra-info if we have parsed the respective
458
        // server descriptor.
459
        //
460
        // It works by doing an inner join from
461
        // `consensus_router_descriptor_member` to `router_descriptor` because
462
        // we can only know about the extra-infos of which we have the server
463
        // descriptors from.  Afterwards, we do a left join with the
464
        // `router_extra_info` table and filter for all results where the left
465
        // join result is null, hence where we have a server descriptor but not
466
        // the respective extra-info.
467
        //
468
        // Parameters:
469
        // :docid - The docid of the consensus.
470
6
        let mut stmt = tx.prepare_cached(sql!(
471
6
            "
472
6
            SELECT server.extra_unsigned_sha1
473
6
            FROM consensus_router_descriptor_member AS cr
474
6
              INNER JOIN router_descriptor AS server ON cr.unsigned_sha1 = server.unsigned_sha1
475
6
              LEFT JOIN router_extra_info AS extra ON server.extra_unsigned_sha1 = extra.unsigned_sha1
476
6
            WHERE
477
6
              cr.consensus_docid = :docid
478
6
              AND server.extra_unsigned_sha1 IS NOT NULL
479
6
              AND extra.unsigned_sha1 IS NULL
480
6
            "
481
6
        ))?;
482

            
483
6
        let missing = stmt
484
7
            .query_map(named_params! {":docid": self.docid}, |row| row.get(0))?
485
6
            .collect::<Result<HashSet<_>, _>>()?;
486
6
        Ok(missing)
487
6
    }
488

            
489
    /// Returns the missing micro descriptors for this consensus.
490
6
    pub(crate) fn missing_micros(
491
6
        &self,
492
6
        tx: &Transaction<'_>,
493
6
    ) -> Result<HashSet<Sha256>, DatabaseError> {
494
6
        if self.flavor != ConsensusFlavor::Microdesc {
495
2
            return Ok(HashSet::new());
496
4
        }
497

            
498
        // Select the missing micro descriptors.
499
        //
500
        // A micro descriptor is considered missing if it exists in
501
        // `consensus_router_descriptor_member` but not in `router_descriptor`
502
        // because the first entry is added once the consensus got parsed,
503
        // whereas the second entry is added once we have actually retrieved it.
504
        //
505
        // It works by doing a left join on router_descriptor and filtering for
506
        // all entries where the join is NULL, as that implies we are aware of
507
        // the descriptor but not have it stored.
508
        //
509
        // Parameters:
510
        // :docid - The docid of the consensus.
511
4
        let mut stmt = tx.prepare_cached(sql!(
512
4
            "
513
4
            SELECT cr.unsigned_sha2
514
4
            FROM consensus_router_descriptor_member AS cr
515
4
              LEFT JOIN router_descriptor AS micro ON cr.unsigned_sha2 = micro.unsigned_sha2
516
4
            WHERE
517
4
              cr.consensus_docid = :docid
518
4
              AND cr.unsigned_sha2 IS NOT NULL
519
4
              AND micro.unsigned_sha2 IS NULL
520
4
            "
521
4
        ))?;
522

            
523
4
        let missing = stmt
524
8
            .query_map(named_params! {":docid": self.docid}, |row| row.get(0))?
525
4
            .collect::<Result<HashSet<_>, _>>()?;
526
4
        Ok(missing)
527
6
    }
528
}
529

            
530
/// Representation of authority certificate metadata from the database.
531
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
532
pub(crate) struct AuthCertMeta {
533
    /// The document id uniquely identifying the consensus.
534
    pub docid: DocumentId,
535

            
536
    /// The SHA-1 fingerprint of the identity key.
537
    // TODO DIRMIRROR: Change this to RsaIdentity.
538
    pub kp_auth_id_rsa_sha1: Sha1,
539

            
540
    /// The SHA-1 fingerprint of the signign key.
541
    // TODO DIRMIRROR: Change this to RsaIdentity.
542
    pub kp_auth_sign_rsa_sha1: Sha1,
543

            
544
    /// The timestamp after which this certificate will be valid.
545
    pub dir_key_published: Timestamp,
546

            
547
    /// The timestamp until this certificate will be valid.
548
    pub dir_key_expires: Timestamp,
549
}
550

            
551
impl AuthCertMeta {
552
    /// Obtain the most recently published and valid certificate for each authority.
553
    ///
554
    /// Returns the found [`AuthCertMeta`] items as well as the missing
555
    /// [`AuthCertKeyIds`].
556
    ///
557
    /// # Performance
558
    ///
559
    /// This function has a performance between `O(n * log n)` and `O(n^2)`
560
    /// because it performs `signatories.len()` database queries, with each
561
    /// database query potentially taking something between `O(log n)` to
562
    /// `O(n)` to execute.  However, given that this respective value is
563
    /// oftentimes fairly small, it should not be much of a big concern.
564
12
    pub(crate) fn query_recent(
565
12
        tx: &Transaction,
566
12
        signatories: &[AuthCertKeyIds],
567
12
        tolerance: &DirTolerance,
568
12
        now: Timestamp,
569
12
    ) -> Result<(Vec<Self>, Vec<AuthCertKeyIds>), DatabaseError> {
570
        // For every key pair in `signatories`, get the most recent valid cert.
571
        //
572
        // This query selects the most recent timestamp valid certificate from
573
        // the database for a single given key pair.  It means that this query
574
        // has to be executed as many times as there are entries in
575
        // `signatories`.
576
        //
577
        // Unfortunately, there is no neater way to do this, because the
578
        // alternative would involve using a nested set which SQLite does not
579
        // support, even with the carray extension.  An alternative might be to
580
        // precompute that string and then insert it here using `format!` but
581
        // that feels hacky, error- and injection-prone.
582
        //
583
        // Parameters:
584
        // :id_rsa: The RSA identity key fingerprint in uppercase hexadecimal.
585
        // :sk_rsa: The RSA signing key fingerprint in uppercase hexadecimal.
586
        // :now: The current system timestamp.
587
        // :pre_tolerance: The tolerance for not-yet-valid certificates.
588
        // :post_tolerance: The tolerance for expired certificates.
589
12
        let mut stmt = tx.prepare_cached(sql!(
590
12
            "
591
12
            SELECT docid, kp_auth_id_rsa_sha1, kp_auth_sign_rsa_sha1,
592
12
              dir_key_published, dir_key_expires
593
12
            FROM authority_key_certificate
594
12
            WHERE
595
12
              (:id_rsa, :sk_rsa) = (kp_auth_id_rsa_sha1, kp_auth_sign_rsa_sha1)
596
12
              AND :now >= dir_key_published - :pre_tolerance
597
12
              AND :now <= dir_key_expires + :post_tolerance
598
12
            ORDER BY dir_key_published DESC
599
12
            LIMIT 1
600
12
            "
601
12
        ))?;
602

            
603
        // Keep track of the found (and parsed) certificates and the missing ones.
604
12
        let mut found = Vec::new();
605
12
        let mut missing = Vec::new();
606

            
607
        // Iterate over every key pair and query it, adding it to found if it exists
608
        // and was parsed successfully or to missing if it does not exist within the
609
        // database.
610
78
        for kp in signatories {
611
            // Query the certificate from the database.
612
78
            let res = stmt
613
78
            .query_one(
614
78
                named_params! {
615
78
                    ":id_rsa": kp.id_fingerprint.as_hex_upper(),
616
78
                    ":sk_rsa": kp.sk_fingerprint.as_hex_upper(),
617
78
                    ":now": now,
618
78
                    ":pre_tolerance": tolerance.pre_valid_tolerance().as_secs().try_into().unwrap_or(i64::MAX),
619
78
                    ":post_tolerance": tolerance.post_valid_tolerance().as_secs().try_into().unwrap_or(i64::MAX),
620
                },
621
                |row| Ok(Self {
622
38
                    docid: row.get(0)?,
623
38
                    kp_auth_id_rsa_sha1: row.get(1)?,
624
38
                    kp_auth_sign_rsa_sha1: row.get(2)?,
625
38
                    dir_key_published: row.get(3)?,
626
38
                    dir_key_expires: row.get(4)?,
627
                })
628
            )
629
78
            .optional()?;
630

            
631
78
            match res {
632
38
                Some(cert) => found.push(cert),
633
40
                None => missing.push(*kp),
634
            }
635
        }
636

            
637
12
        Ok((found, missing))
638
12
    }
639

            
640
    /// Queries the raw data of an [`AuthCertMeta`].
641
    pub(crate) fn data(&self, tx: &Transaction<'_>) -> Result<String, DatabaseError> {
642
        let mut stmt = tx.prepare_cached(sql!(
643
            "
644
            SELECT content
645
            FROM store
646
            WHERE docid = :docid
647
            "
648
        ))?;
649

            
650
        let raw = stmt.query_one(named_params! {":docid": self.docid}, |row| {
651
            row.get::<_, Vec<u8>>(0)
652
        })?;
653
        let raw = String::from_utf8(raw).map_err(into_internal!("utf-8 constraint violated?"))?;
654
        Ok(raw)
655
    }
656

            
657
    /// Inserts a new authority certificate into the database.
658
    ///
659
    /// Keep in mind that the data in the [`AuthCert`] should correspond to the
660
    /// data found in `data`, as this method performs no parsing.
661
18
    pub(crate) fn insert<I: Iterator<Item = ContentEncoding>>(
662
18
        tx: &Transaction<'_>,
663
18
        encodings: I,
664
18
        cert: &AuthCert,
665
18
        data: &str,
666
18
    ) -> Result<(), DatabaseError> {
667
        // Inserts a new certificate into the meta table.
668
        //
669
        // Parameters:
670
        // :docid - The document id.
671
        // :id_rsa - The identity key fingerprint.
672
        // :sign_rsa - The signing key fingerprint
673
        // :published - The published timestamp.
674
        // :expires - The expires timestamp.
675
18
        let mut stmt = tx.prepare_cached(sql!(
676
18
            "
677
18
            INSERT INTO authority_key_certificate
678
18
            (docid, kp_auth_id_rsa_sha1, kp_auth_sign_rsa_sha1, dir_key_published, dir_key_expires)
679
18
            VALUES
680
18
            (:docid, :id_rsa, :sign_rsa, :published, :expires)
681
18
            "
682
18
        ))?;
683

            
684
18
        let docid = store_insert(tx, data.as_bytes(), encodings)?;
685
18
        stmt.execute(named_params! {
686
18
            ":docid": docid,
687
18
            ":id_rsa": cert.dir_identity_key.to_rsa_identity().as_hex_upper(),
688
18
            ":sign_rsa": cert.dir_signing_key.to_rsa_identity().as_hex_upper(),
689
18
            ":published": Timestamp::from(cert.dir_key_published.0),
690
18
            ":expires": Timestamp::from(cert.dir_key_expires.0),
691
18
        })?;
692

            
693
18
        Ok(())
694
18
    }
695
}
696

            
697
/// A no-op macro just returning the supplied.
698
///
699
/// The purpose of this macro is to semantically mark [`str`] literals to be
700
/// SQL statement.
701
///
702
/// Keep in mind that the compiler will not notice if you forget this macro.
703
/// Unfortunately, you have to ensure it yourself.
704
macro_rules! sql {
705
    ($s:literal) => {
706
        $s
707
    };
708
}
709

            
710
pub(crate) use sql;
711

            
712
/// Opens a database from disk, creating a [`Pool`] for it.
713
///
714
/// This function should be the entry point for all things requiring a database
715
/// handle, as this function prepares all necessary steps required for operating
716
/// on the database correctly, such as:
717
/// * Schema initialization.
718
/// * Schema upgrade.
719
/// * Setting connection specific settings.
720
///
721
/// # `SQLITE_BUSY` Caveat
722
///
723
/// There is a problem with the handling of `SQLITE_BUSY` when opening an
724
/// SQLite database.  In WAL, opening a database might acquire an exclusive lock
725
/// for a very short amount of time, in order to perform clean-up from previous
726
/// connections alongside other tasks for maintaining database integrity?  This
727
/// means, that opening multiple SQLite databases simultaneously will result in
728
/// a busy error regardless of a busy handler, as setting a busy handler will
729
/// require an existing connection, something we are unable to obtain in the
730
/// first place.
731
///
732
/// In order to mitigate this issue, the recommended way in the SQLite community
733
/// is to simply ensure that database connections are opened sequentially,
734
/// by urging calling applications to just use a single [`Pool`] instance.
735
///
736
/// Testing this is hard unfortunately.
737
36
pub(crate) fn open<P: AsRef<Path>>(
738
36
    path: P,
739
36
) -> Result<Pool<SqliteConnectionManager>, DatabaseError> {
740
36
    let num_cores = std::thread::available_parallelism()
741
36
        .unwrap_or(NonZero::new(8).expect("8 == 0?"))
742
36
        .get() as u32;
743

            
744
36
    let manager = r2d2_sqlite::SqliteConnectionManager::file(&path);
745
36
    let pool = Pool::builder().max_size(num_cores).build(manager)?;
746

            
747
36
    rw_tx(&pool, |tx| {
748
        // Prepare the database, doing the following steps:
749
        // 1. Checking the database schema.
750
        // 2. Upgrading (in future) or initializing the database schema (if empty).
751

            
752
36
        let has_arti_dirserver_schema_version = match tx.query_one(
753
            sql!(
754
36
                "
755
36
                SELECT name
756
36
                FROM sqlite_master
757
36
                  WHERE type = 'table'
758
36
                    AND name = 'arti_dirserver_schema_version'
759
36
                "
760
            ),
761
36
            params![],
762
2
            |_| Ok(()),
763
        ) {
764
2
            Ok(()) => true,
765
34
            Err(rusqlite::Error::QueryReturnedNoRows) => false,
766
            Err(e) => return Err(DatabaseError::LowLevel(e)),
767
        };
768

            
769
36
        if has_arti_dirserver_schema_version {
770
2
            let version = tx.query_one(
771
2
                sql!("SELECT version FROM arti_dirserver_schema_version WHERE rowid = 1"),
772
2
                params![],
773
2
                |row| row.get::<_, String>(0),
774
            )?;
775

            
776
2
            match version.as_ref() {
777
2
                "1" => {}
778
2
                unknown => {
779
2
                    return Err(DatabaseError::IncompatibleSchema {
780
2
                        version: unknown.into(),
781
2
                    });
782
                }
783
            }
784
        } else {
785
34
            tx.execute_batch(V1_SCHEMA)?;
786
        }
787

            
788
34
        Ok::<_, DatabaseError>(())
789
36
    })??;
790

            
791
34
    Ok(pool)
792
36
}
793

            
794
/// Executes a closure `op` with a given read-only [`Transaction`].
795
///
796
/// The [`Transaction`] always gets rolled back the moment `op` returns.
797
///
798
/// The [`Transaction`] gets initialized with the global pragma options set.
799
///
800
/// **The closure shall not perform write operations!**
801
/// Not only do they get rolled back anyways, but upgrading the [`Transaction`]
802
/// from a read to a write transaction will lead to other simultaneous write upgrades
803
/// to fail.  Unfortunately, there is no real programmatic way to ensure this.
804
50
pub(crate) fn read_tx<U, F>(pool: &Pool<SqliteConnectionManager>, op: F) -> Result<U, DatabaseError>
805
50
where
806
50
    F: FnOnce(&Transaction<'_>) -> U,
807
{
808
50
    let mut conn = pool.get()?;
809
50
    conn.execute_batch(GLOBAL_OPTIONS)?;
810
50
    let tx = conn.transaction_with_behavior(TransactionBehavior::Deferred)?;
811
50
    let res = op(&tx);
812
50
    tx.rollback()?;
813
50
    Ok(res)
814
50
}
815

            
816
/// Executes a closure `op` with a given read-write [`Transaction`].
817
///
818
/// The [`Transaction`] always gets committed the moment `op` returns.
819
///
820
/// The [`Transaction`] gets initialized with the global pragma options set.
821
///
822
/// The [`Transaction`] gets created with [`TransactionBehavior::Immediate`],
823
/// meaning it will immediately exist as a write connection, retrying in the
824
/// case of a [`rusqlite::ErrorCode::DatabaseBusy`] until it failed after 1s.
825
68
pub(crate) fn rw_tx<U, F>(pool: &Pool<SqliteConnectionManager>, op: F) -> Result<U, DatabaseError>
826
68
where
827
68
    F: FnOnce(&Transaction<'_>) -> U,
828
{
829
68
    let mut conn = pool.get()?;
830
68
    conn.execute_batch(GLOBAL_OPTIONS)?;
831
68
    let tx = conn.transaction_with_behavior(TransactionBehavior::Exclusive)?;
832
66
    let res = op(&tx);
833
66
    tx.commit()?;
834
66
    Ok(res)
835
68
}
836

            
837
/// Inserts `data` into store while also compressing it with given encodings.
838
///
839
/// Returns the [`DocumentId`] of `data`.
840
///
841
/// This function inserts `data` into store and also compresses it into all
842
/// given compression formats.
843
///
844
/// Duplicates get re-encoded and replaced in the database, including
845
/// [`ContentEncoding::Identity`].
846
42
pub(crate) fn store_insert<I: Iterator<Item = ContentEncoding>>(
847
42
    tx: &Transaction,
848
42
    data: &[u8],
849
42
    encodings: I,
850
42
) -> Result<DocumentId, DatabaseError> {
851
    // The statement to insert some data into the store.
852
    //
853
    // Parameters:
854
    // :docid - The docid.
855
    // :content - The binary data.
856
42
    let mut store_stmt = tx.prepare_cached(sql!(
857
42
        "
858
42
        INSERT OR REPLACE INTO store (docid, content)
859
42
        VALUES
860
42
        (:docid, :content)
861
42
        "
862
42
    ))?;
863

            
864
    // The statement to insert a compressed document into the metatable.
865
    //
866
    // Parameters:
867
    // :algorithm - The name of the encoding algorithm.
868
    // :identity_docid - The docid of the plain-text document in the store.
869
    // :compressed_docid - The docid of the encoded document in the store.
870
42
    let mut compressed_stmt = tx.prepare_cached(sql!(
871
42
        "
872
42
        INSERT OR REPLACE INTO compressed_document (algorithm, identity_docid, compressed_docid)
873
42
        VALUES
874
42
        (:algorithm, :identity_docid, :compressed_docid)
875
42
        "
876
42
    ))?;
877

            
878
    // Insert the plain document into the store.
879
42
    let identity_docid = DocumentId::digest(data);
880
42
    store_stmt.execute(named_params! {
881
42
        ":docid": identity_docid,
882
42
        ":content": data
883
42
    })?;
884

            
885
    // Compress it into all formats and insert it into store and compressed.
886
120
    for encoding in encodings {
887
120
        if encoding == ContentEncoding::Identity {
888
            // Ignore identity because we inserted that above.
889
24
            continue;
890
96
        }
891

            
892
        // We map a compression error to a bug because there is no good reason
893
        // on why it should fail, given that we compress from memory data to
894
        // memory data.  Probably because it uses the std::io::Writer interface
895
        // which itself demands use of std::io::Result.
896
96
        let compressed = compress(data, encoding).map_err(into_internal!("{encoding} failed?"))?;
897
96
        let compressed_docid = DocumentId::digest(&compressed);
898
96
        store_stmt.execute(named_params! {
899
96
            ":docid": compressed_docid,
900
96
            ":content": compressed,
901
96
        })?;
902
96
        compressed_stmt.execute(named_params! {
903
96
            ":algorithm": encoding.to_string(),
904
96
            ":identity_docid": identity_docid,
905
96
            ":compressed_docid": compressed_docid,
906
96
        })?;
907
    }
908

            
909
42
    Ok(identity_docid)
910
42
}
911

            
912
/// Compresses `data` into a specified [`ContentEncoding`].
913
///
914
/// Returns a [`Vec`] containing the encoded data.
915
106
fn compress(data: &[u8], encoding: ContentEncoding) -> Result<Vec<u8>, std::io::Error> {
916
106
    match encoding {
917
2
        ContentEncoding::Identity => Ok(data.to_vec()),
918
        ContentEncoding::Deflate => {
919
26
            let mut w = DeflateEncoder::new(Vec::new(), Default::default());
920
26
            w.write_all(data)?;
921
26
            w.finish()
922
        }
923
        ContentEncoding::Gzip => {
924
26
            let mut w = GzEncoder::new(Vec::new(), Default::default());
925
26
            w.write_all(data)?;
926
26
            w.finish()
927
        }
928
26
        ContentEncoding::XZstd => zstd::encode_all(data, Default::default()),
929
        ContentEncoding::XTorLzma => {
930
26
            let mut res = Vec::new();
931
26
            lzma_rs::lzma_compress(&mut Cursor::new(data), &mut res)?;
932
26
            Ok(res)
933
        }
934
    }
935
106
}
936

            
937
#[cfg(test)]
938
mod test {
939
    // @@ begin test lint list maintained by maint/add_warning @@
940
    #![allow(clippy::bool_assert_comparison)]
941
    #![allow(clippy::clone_on_copy)]
942
    #![allow(clippy::dbg_macro)]
943
    #![allow(clippy::mixed_attributes_style)]
944
    #![allow(clippy::print_stderr)]
945
    #![allow(clippy::print_stdout)]
946
    #![allow(clippy::single_char_pattern)]
947
    #![allow(clippy::unwrap_used)]
948
    #![allow(clippy::unchecked_time_subtraction)]
949
    #![allow(clippy::useless_vec)]
950
    #![allow(clippy::needless_pass_by_value)]
951
    #![allow(clippy::string_slice)] // See arti#2571
952
    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
953
    use std::{
954
        collections::HashSet,
955
        io::Read,
956
        sync::{Arc, Once},
957
    };
958

            
959
    use flate2::read::{DeflateDecoder, GzDecoder};
960
    use lazy_static::lazy_static;
961
    use rusqlite::Connection;
962
    use strum::IntoEnumIterator;
963
    use tempfile::tempdir;
964
    use tor_basic_utils::test_rng::testing_rng;
965
    use tor_dircommon::config::DirToleranceBuilder;
966
    use tor_llcrypto::pk::rsa::RsaIdentity;
967

            
968
    use super::*;
969

            
970
    lazy_static! {
971
    /// Wed Jan 01 2020 00:00:00 GMT+0000
972
    static ref VALID_AFTER: Timestamp =
973
        (SystemTime::UNIX_EPOCH + Duration::from_secs(1577836800)).into();
974

            
975
    /// Wed Jan 01 2020 01:00:00 GMT+0000
976
    static ref FRESH_UNTIL: Timestamp =
977
        *VALID_AFTER + Duration::from_secs(60 * 60);
978

            
979
    /// Wed Jan 01 2020 02:00:00 GMT+0000
980
    static ref FRESH_UNTIL_HALF: Timestamp =
981
        *FRESH_UNTIL + Duration::from_secs(60 * 60);
982

            
983
    /// Wed Jan 01 2020 03:00:00 GMT+0000
984
    static ref VALID_UNTIL: Timestamp =
985
        *FRESH_UNTIL + Duration::from_secs(60 * 60 * 2);
986
    }
987

            
988
    const CONSENSUS_CONTENT: &str = "Lorem ipsum dolor sit amet.";
989
    const CONSENSUS_MD_CONTENT: &str = "Lorem ipsum dolor sit amet!";
990
    const CERT_CONTENT: &[u8] = include_bytes!("../testdata/authcert-longclaw");
991

            
992
    lazy_static! {
993
        static ref CONSENSUS_DOCID: DocumentId = DocumentId::digest(CONSENSUS_CONTENT.as_bytes());
994
        static ref CONSENSUS_MD_DOCID: DocumentId =
995
            DocumentId::digest(CONSENSUS_MD_CONTENT.as_bytes());
996
        static ref CERT_DOCID: DocumentId = DocumentId::digest(CERT_CONTENT);
997
    }
998

            
999
    fn create_dummy_db() -> Pool<SqliteConnectionManager> {
        let pool = open("").unwrap();
        rw_tx(&pool, |tx| {
            tx.execute(
                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),
                params![*CONSENSUS_DOCID, CONSENSUS_CONTENT.as_bytes()],
            )
            .unwrap();
            tx.execute(
                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),
                params![*CONSENSUS_MD_DOCID, CONSENSUS_MD_CONTENT.as_bytes()],
            )
            .unwrap();
            tx.execute(
                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),
                params![*CERT_DOCID, CERT_CONTENT],
            )
            .unwrap();
            tx.execute(
                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),
                params![
                    DocumentId::digest(include_bytes!("../testdata/descriptor1-ns")),
                    include_bytes!("../testdata/descriptor1-ns")
                ]
            ).unwrap();
            tx.execute(
                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),
                params![
                    DocumentId::digest(include_bytes!("../testdata/descriptor1-extra-info")),
                    include_bytes!("../testdata/descriptor1-extra-info")
                ]
            ).unwrap();
            tx.execute(
                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),
                params![
                    DocumentId::digest(include_bytes!("../testdata/descriptor1-md")),
                    include_bytes!("../testdata/descriptor1-md"),
            ]).unwrap();
            // Insert descriptor into router_extra_info.
            tx.execute(sql!(
                "
                INSERT INTO router_extra_info
                (docid, unsigned_sha1, kp_relay_id_rsa_sha1)
                VALUES (?1, ?2, ?3)
                "
            ), params![
                Sha256::digest(include_bytes!("../testdata/descriptor1-extra-info")),
                Sha1::digest(include_bytes!("../testdata/descriptor1-extra-info-unsigned")),
                "000004ACBB9D29BCBA17256BB35928DDBFC8ABA9",
            ]).unwrap();
            // We only insert descriptor1 here.
            tx.execute(sql!(
                "
                INSERT INTO router_descriptor
                (docid, unsigned_sha1, unsigned_sha2, kp_relay_id_rsa_sha1, flavor, extra_unsigned_sha1)
                VALUES
                (?1, ?2, ?3, ?4, 'ns', ?5)
                "
            ), params![
                DocumentId::digest(include_bytes!("../testdata/descriptor1-ns")),
                Sha1::digest(include_bytes!("../testdata/descriptor1-ns-unsigned")),
                Sha256::digest(include_bytes!("../testdata/descriptor1-ns-unsigned")),
                Sha1::from([0, 0, 4, 172, 187, 157, 41, 188, 186, 23, 37, 107, 179, 89, 40, 221, 191, 200, 171, 169]),
                Sha1::digest(include_bytes!("../testdata/descriptor1-extra-info-unsigned")),
            ]).unwrap();
            // Only insert descriptor1's md
            tx.execute(sql!(
                "
                INSERT INTO router_descriptor
                (docid, unsigned_sha1, unsigned_sha2, kp_relay_id_rsa_sha1, flavor)
                VALUES (?1, ?2, ?3, ?4, 'microdesc')
                "
            ), params![
                DocumentId::digest(include_bytes!("../testdata/descriptor1-md")),
                Sha1::digest(include_bytes!("../testdata/descriptor1-md")),
                Sha256::digest(include_bytes!("../testdata/descriptor1-md")),
                Sha1::from([0, 0, 4, 172, 187, 157, 41, 188, 186, 23, 37, 107, 179, 89, 40, 221, 191, 200, 171, 169]),
            ]).unwrap();
            tx.execute(
                sql!(
                    "
                    INSERT INTO consensus
                    (docid, unsigned_sha3_256, flavor, valid_after, fresh_until, valid_until)
                    VALUES
                    (?1, ?2, ?3, ?4, ?5, ?6)
                    "
                ),
                params![
                    *CONSENSUS_DOCID,
                    "0000000000000000000000000000000000000000000000000000000000000000", // not the correct hash
                    ConsensusFlavor::Plain.name(),
                    *VALID_AFTER,
                    *FRESH_UNTIL,
                    *VALID_UNTIL,
                ],
            )
            .unwrap();
            tx.execute(
                sql!(
                    "
                    INSERT INTO consensus
                    (docid, unsigned_sha3_256, flavor, valid_after, fresh_until, valid_until)
                    VALUES
                    (?1, ?2, ?3, ?4, ?5, ?6)
                    "
                ),
                params![
                    *CONSENSUS_MD_DOCID,
                    "0000000000000000000000000000000000000000000000000000000000000001", // not the correct hash
                    ConsensusFlavor::Microdesc.name(),
                    *VALID_AFTER,
                    *FRESH_UNTIL,
                    *VALID_UNTIL,
                ],
            )
            .unwrap();
            tx.execute(sql!(
                "
                INSERT INTO consensus_router_descriptor_member
                (consensus_docid, unsigned_sha1, unsigned_sha2)
                VALUES
                (?1, ?2, NULL),
                (?1, ?3, NULL)
                "
            ), params![
                *CONSENSUS_DOCID,
                Sha1::digest(include_bytes!("../testdata/descriptor1-ns-unsigned")),
                Sha1::digest(include_bytes!("../testdata/descriptor2-ns-unsigned")),
            ]).unwrap();
            tx.execute(sql!(
                "
                INSERT INTO consensus_router_descriptor_member
                (consensus_docid, unsigned_sha1, unsigned_sha2)
                VALUES
                (?1, NULL, ?2),
                (?1, NULL, ?3)
                "
            ), params![
                *CONSENSUS_MD_DOCID,
                Sha256::digest(include_bytes!("../testdata/descriptor1-md")),
                Sha256::digest(include_bytes!("../testdata/descriptor2-md")),
            ]).unwrap();
            tx.execute(sql!(
                "
                INSERT INTO authority_key_certificate
                  (docid, kp_auth_id_rsa_sha1, kp_auth_sign_rsa_sha1, dir_key_published, dir_key_expires)
                VALUES
                  (:docid, :id_rsa, :sk_rsa, :published, :expires)
                "
                ),
                named_params! {
                ":docid": *CERT_DOCID,
                ":id_rsa": "49015F787433103580E3B66A1707A00E60F2D15B",
                ":sk_rsa": "C5D153A6F0DA7CC22277D229DCBBF929D0589FE0",
                ":published": 1764543578,
                ":expires": 1772492378,
            }).unwrap();
        })
        .unwrap();
        pool
    }
    #[test]
    fn open_test() {
        let db_dir = tempdir().unwrap();
        let db_path = db_dir.path().join("db");
        open(&db_path).unwrap();
        let conn = Connection::open(&db_path).unwrap();
        // Check if the version was initialized properly.
        let version = conn
            .query_one(
                "SELECT version FROM arti_dirserver_schema_version WHERE rowid = 1",
                params![],
                |row| row.get::<_, String>(0),
            )
            .unwrap();
        assert_eq!(version, "1");
        // Set the version to something unknown.
        conn.execute(
            "UPDATE arti_dirserver_schema_version SET version = 42",
            params![],
        )
        .unwrap();
        drop(conn);
        assert_eq!(
            open(&db_path).unwrap_err().to_string(),
            "incompatible schema version: 42"
        );
    }
    #[test]
    fn read_tx_test() {
        let db_dir = tempdir().unwrap();
        let db_path = db_dir.path().join("db");
        let pool = open(&db_path).unwrap();
        // Do a write transaction despite forbidden.
        read_tx(&pool, |tx| {
            tx.execute_batch("DELETE FROM arti_dirserver_schema_version")
                .unwrap();
            let e = tx
                .query_one(
                    sql!("SELECT version FROM arti_dirserver_schema_version"),
                    params![],
                    |row| row.get::<_, String>(0),
                )
                .unwrap_err();
            assert_eq!(e, rusqlite::Error::QueryReturnedNoRows);
        })
        .unwrap();
        // Normal check.
        let version: String = read_tx(&pool, |tx| {
            tx.query_one(
                sql!("SELECT version FROM arti_dirserver_schema_version"),
                params![],
                |row| row.get(0),
            )
            .unwrap()
        })
        .unwrap();
        assert_eq!(version, "1");
    }
    #[test]
    fn rw_tx_test() {
        let db_dir = tempdir().unwrap();
        let db_path = db_dir.path().join("db");
        let pool = open(&db_path).unwrap();
        // Do a write transaction.
        rw_tx(&pool, |tx| {
            tx.execute_batch("DELETE FROM arti_dirserver_schema_version")
                .unwrap();
        })
        .unwrap();
        // Check that it was deleted.
        read_tx(&pool, |tx| {
            let e = tx
                .query_one(
                    sql!("SELECT version FROM arti_dirserver_schema_version"),
                    params![],
                    |row| row.get::<_, String>(0),
                )
                .unwrap_err();
            assert_eq!(e, rusqlite::Error::QueryReturnedNoRows);
        })
        .unwrap();
    }
    /// Tests whether our SQLite busy error handling works in normal situations.
    ///
    /// A normal situations means a situation where a lock is never held for
    /// more than 1000ms.  In our case, we will work with two threads.
    /// t1 will acquire an exclusive lock and inform t2 about it.  t2 waits
    /// until t1 has acquired this lock and then immediately informs t1, that
    /// it will now wait for a lock too.  Now, t1 will immediately terminate,
    /// thereby releasing the lock and leading t2 to eventually acquire it.
    #[test]
    fn rw_tx_busy_timeout_working() {
        let db_dir = tempdir().unwrap();
        let db_path = db_dir.path().join("db");
        let pool = open(db_path).unwrap();
        // t2 will wait on this before it starts doing stuff.
        let t1_acquired_lock = Arc::new(Once::new());
        // t1 will wait on this in order to terminate properly.
        let t2_is_waiting = Arc::new(Once::new());
        let t1 = std::thread::spawn({
            let pool = pool.clone();
            let t1_acquired_lock = t1_acquired_lock.clone();
            let t2_is_waiting = t2_is_waiting.clone();
            move || {
                rw_tx(&pool, move |_tx| {
                    // Inform t2 we have write lock.
                    t1_acquired_lock.call_once(|| ());
                    println!("t1 acquired write lock");
                    // Wait for t2 to start waiting.
                    t2_is_waiting.wait();
                })
                .unwrap();
                println!("t2 released write lock");
            }
        });
        println!("t2 waits for t1 to acquire write lock");
        t1_acquired_lock.wait();
        t2_is_waiting.call_once(|| ());
        rw_tx(&pool, |_| ()).unwrap();
        println!("t2 acquired and released write lock");
        t1.join().unwrap();
    }
    /// Tests whether our SQLite busy error handlings fails as expected.
    ///
    /// We configure SQLite to fail after 1000ms.  This test works with two
    /// threads.  t1 will acquire an exclusive lock on the database and will
    /// inform t2 about it, which itself will wait until t1 has acquired the
    /// lock.  t2 will then immediately try to also obtain an exclusive lock,
    /// which should fail after about 1000ms.  After the failure, t2 informs
    /// t1 that it has failed, causing t1 to terminate.
    #[test]
    fn rw_tx_busy_timeout_busy() {
        let db_dir = tempdir().unwrap();
        let db_path = db_dir.path().join("db");
        let pool = open(db_path).unwrap();
        // t2 will wait on this before it starts doing stuff.
        let t1_acquired_lock = Arc::new(Once::new());
        // t1 will wait on this in order to terminate properly.
        let t2_gave_up = Arc::new(Once::new());
        let t1 = std::thread::spawn({
            let pool = pool.clone();
            let t1_acquired_lock = t1_acquired_lock.clone();
            let t2_gave_up = t2_gave_up.clone();
            move || {
                rw_tx(&pool, move |_tx| {
                    // Inform t2 we have the write lock.
                    t1_acquired_lock.call_once(|| ());
                    println!("t1 acquired write lock");
                    // Wait for t2 to give up before we release (how mean from us).
                    t2_gave_up.wait();
                })
                .unwrap();
                println!("t1 released write lock");
            }
        });
        println!("t2 waits for t1 to acquire write lock");
        t1_acquired_lock.wait();
        let e = rw_tx(&pool, |_| ()).unwrap_err();
        assert_eq!(
            e.to_string(),
            "low-level rusqlite error: database is locked"
        );
        println!("t2 gave up on acquiring write lock");
        t2_gave_up.call_once(|| ());
        t1.join().unwrap();
    }
    #[test]
    fn store_insert_test() {
        let db_dir = tempdir().unwrap();
        let db_path = db_dir.path().join("db");
        open(&db_path).unwrap();
        let mut conn = Connection::open(&db_path).unwrap();
        let tx = conn.transaction().unwrap();
        let docid = store_insert(&tx, "foobar".as_bytes(), ContentEncoding::iter()).unwrap();
        assert_eq!(
            docid,
            "C3AB8FF13720E8AD9047DD39466B3C8974E592C2FA383D4A3960714CAEF0C4F2"
        );
        let res = tx
            .query_one(
                sql!(
                    "
                    SELECT content
                    FROM store
                    WHERE docid = 'C3AB8FF13720E8AD9047DD39466B3C8974E592C2FA383D4A3960714CAEF0C4F2'
                    "
                ),
                params![],
                |row| row.get::<_, Vec<u8>>(0),
            )
            .unwrap();
        assert_eq!(res, "foobar".as_bytes());
        let mut stmt = tx.prepare_cached(sql!(
            "
            SELECT algorithm
            FROM compressed_document
            WHERE identity_docid = 'C3AB8FF13720E8AD9047DD39466B3C8974E592C2FA383D4A3960714CAEF0C4F2'
            "
        )).unwrap();
        let algorithms = stmt
            .query_map(params![], |row| row.get::<_, String>(0))
            .unwrap();
        let algorithms = algorithms.map(|x| x.unwrap()).collect::<HashSet<_>>();
        assert_eq!(
            algorithms,
            HashSet::from([
                "deflate".to_string(),
                "gzip".to_string(),
                "x-zstd".to_string(),
                "x-tor-lzma".to_string()
            ])
        );
        // Now insert the same thing a second time again and see whether the
        // ON CONFLICT magic works.
        let docid_second = store_insert(&tx, "foobar".as_bytes(), ContentEncoding::iter()).unwrap();
        assert_eq!(docid, docid_second);
        // Remove a few compressed entries and get them again.
        let n = tx
            .execute(
                sql!(
                    "
                    DELETE FROM
                    compressed_document
                    WHERE algorithm IN ('deflate', 'x-zstd')
                    "
                ),
                params![],
            )
            .unwrap();
        assert_eq!(n, 2);
        let docid_third = store_insert(&tx, "foobar".as_bytes(), ContentEncoding::iter()).unwrap();
        assert_eq!(docid, docid_third);
        let algorithms = stmt
            .query_map(params![], |row| row.get::<_, String>(0))
            .unwrap();
        let algorithms = algorithms.map(|x| x.unwrap()).collect::<HashSet<_>>();
        assert_eq!(
            algorithms,
            HashSet::from([
                "deflate".to_string(),
                "gzip".to_string(),
                "x-zstd".to_string(),
                "x-tor-lzma".to_string()
            ])
        );
    }
    #[test]
    fn compress_test() {
        /// Asserts that `res` contains `encoding`.
        fn contains(encoding: ContentEncoding, res: &[(ContentEncoding, Vec<u8>)]) {
            assert!(res.iter().any(|x| x.0 == encoding));
        }
        const INPUT: &[u8] = "foobar".as_bytes();
        // Check whether everything was encoded.
        let res = ContentEncoding::iter()
            .map(|encoding| (encoding, compress(INPUT, encoding).unwrap()))
            .collect::<Vec<_>>();
        assert_eq!(res.len(), 5);
        contains(ContentEncoding::Identity, &res);
        contains(ContentEncoding::Deflate, &res);
        contains(ContentEncoding::Gzip, &res);
        contains(ContentEncoding::XTorLzma, &res);
        contains(ContentEncoding::XZstd, &res);
        // Check if we can decode it.
        for (encoding, compressed) in res {
            let mut decompressed = Vec::new();
            match encoding {
                ContentEncoding::Identity => decompressed = compressed,
                ContentEncoding::Deflate => {
                    DeflateDecoder::new(Cursor::new(compressed))
                        .read_to_end(&mut decompressed)
                        .unwrap();
                }
                ContentEncoding::Gzip => {
                    GzDecoder::new(Cursor::new(compressed))
                        .read_to_end(&mut decompressed)
                        .unwrap();
                }
                ContentEncoding::XTorLzma => {
                    lzma_rs::lzma_decompress(&mut Cursor::new(compressed), &mut decompressed)
                        .unwrap();
                }
                ContentEncoding::XZstd => {
                    decompressed = zstd::decode_all(Cursor::new(compressed)).unwrap();
                }
            }
            assert_eq!(decompressed, INPUT);
        }
    }
    #[test]
    fn recent_consensus() {
        let pool = create_dummy_db();
        let no_tolerance = DirToleranceBuilder::default()
            .pre_valid_tolerance(Duration::ZERO)
            .post_valid_tolerance(Duration::ZERO)
            .build()
            .unwrap();
        let liberal_tolerance = DirToleranceBuilder::default()
            .pre_valid_tolerance(Duration::from_secs(60 * 60)) // 1h before
            .post_valid_tolerance(Duration::from_secs(60 * 60)) // 1h after
            .build()
            .unwrap();
        read_tx(&pool, move |tx| {
            // Get None by being way before valid-after.
            assert!(
                ConsensusMeta::query_recent(
                    tx,
                    ConsensusFlavor::Plain,
                    &no_tolerance,
                    SystemTime::UNIX_EPOCH.into(),
                )
                .unwrap()
                .is_none()
            );
            // Get None by being way behind valid-until.
            assert!(
                ConsensusMeta::query_recent(
                    tx,
                    ConsensusFlavor::Plain,
                    &no_tolerance,
                    *VALID_UNTIL + Duration::from_secs(60 * 60 * 24 * 365),
                )
                .unwrap()
                .is_none()
            );
            // Get None by being minimally before valid-after.
            assert!(
                ConsensusMeta::query_recent(
                    tx,
                    ConsensusFlavor::Plain,
                    &no_tolerance,
                    *VALID_AFTER - Duration::from_secs(1),
                )
                .unwrap()
                .is_none()
            );
            // Get None by being minimally behind valid-until.
            assert!(
                ConsensusMeta::query_recent(
                    tx,
                    ConsensusFlavor::Plain,
                    &no_tolerance,
                    *VALID_UNTIL + Duration::from_secs(1),
                )
                .unwrap()
                .is_none()
            );
            // Get a valid consensus by being in the interval.
            let res1 = ConsensusMeta::query_recent(
                tx,
                ConsensusFlavor::Plain,
                &no_tolerance,
                *VALID_AFTER,
            )
            .unwrap()
            .unwrap();
            let res2 = ConsensusMeta::query_recent(
                tx,
                ConsensusFlavor::Plain,
                &no_tolerance,
                *VALID_UNTIL,
            )
            .unwrap()
            .unwrap();
            let res3 = ConsensusMeta::query_recent(
                tx,
                ConsensusFlavor::Plain,
                &no_tolerance,
                *VALID_AFTER + Duration::from_secs(60 * 30),
            )
            .unwrap()
            .unwrap();
            assert_eq!(
                res1,
                ConsensusMeta {
                    docid: *CONSENSUS_DOCID,
                    unsigned_sha3_256: Sha3_256::from([0; 32]),
                    flavor: ConsensusFlavor::Plain,
                    valid_after: *VALID_AFTER,
                    fresh_until: *FRESH_UNTIL,
                    valid_until: *VALID_UNTIL,
                }
            );
            assert_eq!(res1, res2);
            assert_eq!(res2, res3);
            // Get a valid consensus using a liberal dir tolerance.
            let res1 = ConsensusMeta::query_recent(
                tx,
                ConsensusFlavor::Plain,
                &liberal_tolerance,
                *VALID_AFTER - Duration::from_secs(60 * 30),
            )
            .unwrap()
            .unwrap();
            let res2 = ConsensusMeta::query_recent(
                tx,
                ConsensusFlavor::Plain,
                &liberal_tolerance,
                *VALID_UNTIL + Duration::from_secs(60 * 30),
            )
            .unwrap()
            .unwrap();
            assert_eq!(
                res1,
                ConsensusMeta {
                    docid: *CONSENSUS_DOCID,
                    unsigned_sha3_256: Sha3_256::from([0; 32]),
                    flavor: ConsensusFlavor::Plain,
                    valid_after: *VALID_AFTER,
                    fresh_until: *FRESH_UNTIL,
                    valid_until: *VALID_UNTIL,
                }
            );
            assert_eq!(res1, res2);
        })
        .unwrap();
    }
    #[test]
    fn sync_timeout() {
        // We repeat the tests a few thousand times to go over many random values.
        let cons = ConsensusMeta {
            docid: *CONSENSUS_DOCID,
            unsigned_sha3_256: Sha3_256::from([0; 32]),
            flavor: ConsensusFlavor::Plain,
            valid_after: *VALID_AFTER,
            fresh_until: *FRESH_UNTIL,
            valid_until: *VALID_UNTIL,
        };
        for _ in 0..10000 {
            let when = cons.lifetime(&mut testing_rng());
            assert!(when >= *FRESH_UNTIL);
            assert!(when <= *FRESH_UNTIL_HALF);
        }
    }
    #[test]
    fn get_auth_cert() {
        let pool = create_dummy_db();
        // Empty.
        let (found, missing) = read_tx(&pool, |tx| {
            AuthCertMeta::query_recent(
                tx,
                &[],
                &DirTolerance::default(),
                (SystemTime::UNIX_EPOCH + Duration::from_secs(1765900013)).into(),
            )
        })
        .unwrap()
        .unwrap();
        assert!(found.is_empty());
        assert!(missing.is_empty());
        // Find one and two missing ones.
        let (found, missing) = read_tx(&pool, |tx| {
            AuthCertMeta::query_recent(
                tx,
                &[
                    // Found one.
                    AuthCertKeyIds {
                        id_fingerprint: RsaIdentity::from_hex(
                            "49015F787433103580E3B66A1707A00E60F2D15B",
                        )
                        .unwrap(),
                        sk_fingerprint: RsaIdentity::from_hex(
                            "C5D153A6F0DA7CC22277D229DCBBF929D0589FE0",
                        )
                        .unwrap(),
                    },
                    // Missing.
                    AuthCertKeyIds {
                        id_fingerprint: RsaIdentity::from_hex(
                            "0000000000000000000000000000000000000000",
                        )
                        .unwrap(),
                        sk_fingerprint: RsaIdentity::from_hex(
                            "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF",
                        )
                        .unwrap(),
                    },
                    // Missing.
                    AuthCertKeyIds {
                        id_fingerprint: RsaIdentity::from_hex(
                            "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF",
                        )
                        .unwrap(),
                        sk_fingerprint: RsaIdentity::from_hex(
                            "0000000000000000000000000000000000000000",
                        )
                        .unwrap(),
                    },
                ],
                &DirTolerance::default(),
                (SystemTime::UNIX_EPOCH + Duration::from_secs(1765900013)).into(),
            )
        })
        .unwrap()
        .unwrap();
        assert_eq!(
            found,
            vec![AuthCertMeta {
                docid: DocumentId::digest(CERT_CONTENT),
                kp_auth_id_rsa_sha1: Sha1::from([
                    73, 1, 95, 120, 116, 51, 16, 53, 128, 227, 182, 106, 23, 7, 160, 14, 96, 242,
                    209, 91
                ]),
                kp_auth_sign_rsa_sha1: Sha1::from([
                    197, 209, 83, 166, 240, 218, 124, 194, 34, 119, 210, 41, 220, 187, 249, 41,
                    208, 88, 159, 224
                ]),
                dir_key_published: (SystemTime::UNIX_EPOCH + Duration::from_secs(1764543578))
                    .into(),
                dir_key_expires: (SystemTime::UNIX_EPOCH + Duration::from_secs(1772492378)).into()
            }]
        );
        assert_eq!(
            missing,
            vec![
                AuthCertKeyIds {
                    id_fingerprint: RsaIdentity::from_hex(
                        "0000000000000000000000000000000000000000",
                    )
                    .unwrap(),
                    sk_fingerprint: RsaIdentity::from_hex(
                        "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF",
                    )
                    .unwrap(),
                },
                AuthCertKeyIds {
                    id_fingerprint: RsaIdentity::from_hex(
                        "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF",
                    )
                    .unwrap(),
                    sk_fingerprint: RsaIdentity::from_hex(
                        "0000000000000000000000000000000000000000",
                    )
                    .unwrap(),
                }
            ]
        );
    }
    #[test]
    fn missing_server_descriptors() {
        let pool = create_dummy_db();
        let meta = read_tx(&pool, |tx| {
            ConsensusMeta::query_recent(
                tx,
                ConsensusFlavor::Plain,
                &DirTolerance::default(),
                *VALID_AFTER,
            )
        })
        .unwrap()
        .unwrap()
        .unwrap();
        // Only one should be returned.
        let missing_servers = read_tx(&pool, |tx| meta.missing_servers(tx))
            .unwrap()
            .unwrap();
        assert_eq!(
            missing_servers,
            HashSet::from([Sha1::digest(include_bytes!(
                "../testdata/descriptor2-ns-unsigned"
            ))])
        );
        // If we delete all router descriptors we have, we should get both.
        rw_tx(&pool, |tx| {
            tx.execute(sql!("DELETE FROM router_descriptor"), params![])
        })
        .unwrap()
        .unwrap();
        // Now both should be returned
        let missing_servers = read_tx(&pool, |tx| meta.missing_servers(tx))
            .unwrap()
            .unwrap();
        assert_eq!(
            missing_servers,
            HashSet::from([
                Sha1::digest(include_bytes!("../testdata/descriptor1-ns-unsigned")),
                Sha1::digest(include_bytes!("../testdata/descriptor2-ns-unsigned"))
            ])
        );
    }
    #[test]
    fn missing_extra_infos() {
        let pool = create_dummy_db();
        let meta = read_tx(&pool, |tx| {
            ConsensusMeta::query_recent(
                tx,
                ConsensusFlavor::Plain,
                &DirTolerance::default(),
                *VALID_AFTER,
            )
        })
        .unwrap()
        .unwrap()
        .unwrap();
        // We should have no missing extra-infos.
        // Technically extra-info of the second relay is missing too, but we
        // cannot know that.
        let missing_extras = read_tx(&pool, |tx| meta.missing_extras(tx))
            .unwrap()
            .unwrap();
        assert!(missing_extras.is_empty());
        // Now delete the record of router_extra_info.
        pool.get()
            .unwrap()
            .execute(sql!("DELETE FROM router_extra_info"), params![])
            .unwrap();
        // Now we should get a single missing extra-info.
        let missing_extras = read_tx(&pool, |tx| meta.missing_extras(tx))
            .unwrap()
            .unwrap();
        assert_eq!(
            missing_extras,
            HashSet::from([Sha1::digest(include_bytes!(
                "../testdata/descriptor1-extra-info-unsigned"
            ))])
        );
    }
    #[test]
    fn missing_micro_descriptors() {
        let pool = create_dummy_db();
        let meta = read_tx(&pool, |tx| {
            ConsensusMeta::query_recent(
                tx,
                ConsensusFlavor::Microdesc,
                &DirTolerance::default(),
                *VALID_AFTER,
            )
        })
        .unwrap()
        .unwrap()
        .unwrap();
        // Only one should be returned.
        let missing_micros = read_tx(&pool, |tx| meta.missing_micros(tx))
            .unwrap()
            .unwrap();
        assert_eq!(
            missing_micros,
            HashSet::from([Sha256::digest(include_bytes!("../testdata/descriptor2-md"))])
        );
        // If we delete all router descriptors we have, we should get both.
        rw_tx(&pool, |tx| {
            tx.execute(sql!("DELETE FROM router_descriptor"), params![])
        })
        .unwrap()
        .unwrap();
        // Now both should be returned
        let missing_servers = read_tx(&pool, |tx| meta.missing_micros(tx))
            .unwrap()
            .unwrap();
        assert_eq!(
            missing_servers,
            HashSet::from([
                Sha256::digest(include_bytes!("../testdata/descriptor1-md")),
                Sha256::digest(include_bytes!("../testdata/descriptor2-md"))
            ])
        );
    }
}