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
    marker::PhantomData,
49
    num::NonZero,
50
    ops::{Add, Sub},
51
    path::Path,
52
    rc::Rc,
53
    time::{Duration, SystemTime},
54
};
55

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

            
71
use crate::{
72
    err::DatabaseError,
73
    types::{
74
        FlavoredConsensusBody, FlavoredConsensusSignatures, FlavoredConsensusUnverified,
75
        FlavoredDescriptor,
76
    },
77
};
78

            
79
/// Version 1 of the database schema.
80
///
81
/// TODO DIRMIRROR: Before the release, figure out where to use rowid and where
82
/// to use docid.
83
const V1_SCHEMA: &str = include_str!("schema_v1.sql");
84

            
85
/// Global options set in every connection.
86
const GLOBAL_OPTIONS: &str = sql!(
87
    "
88
PRAGMA journal_mode=WAL;
89
PRAGMA foreign_keys=ON;
90
PRAGMA busy_timeout=1000;
91
"
92
);
93

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

            
123
        impl $name {
124
            /// Computes the hash from arbitrary data.
125
646
            pub(crate) fn digest(data: &[u8]) -> Self {
126
646
                Self(<$algo>::digest(data).into())
127
646
            }
128
        }
129

            
130
        impl Display for $name {
131
            /// Formats the hash in uppercase hexadecimal.
132
2467
            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
133
2467
                write!(f, "{}", hex::encode_upper(self.0))
134
2467
            }
135
        }
136

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

            
156
231
                Ok(Self(data))
157
231
            }
158
        }
159

            
160
        impl ToSql for $name {
161
2444
            fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
162
                // Because Self is only constructed with FromSql and digest
163
                // data, it is safe to assume it is valid.
164
2444
                Ok(ToSqlOutput::from(self.to_string()))
165
2444
            }
166
        }
167

            
168
        impl From<$name> for Value {
169
22
            fn from(value: $name) -> Value {
170
22
                Value::Text(value.to_string())
171
22
            }
172
        }
173

            
174
        impl PartialEq<&str> for $name {
175
1
            fn eq(&self, other: &&str) -> bool {
176
1
                self.to_string() == other.to_uppercase()
177
1
            }
178
        }
179

            
180
        impl From<[u8; $size]> for $name {
181
857
            fn from(value: [u8; $size]) -> Self {
182
857
                Self(value)
183
857
            }
184
        }
185
    };
186
}
187

            
188
impl_hash_wrapper!(Sha1, tor_llcrypto::d::Sha1, 20);
189
impl_hash_wrapper!(Sha256, tor_llcrypto::d::Sha256, 32);
190
impl_hash_wrapper!(Sha3_256, tor_llcrypto::d::Sha3_256, 32);
191

            
192
/// The identifier for documents in the content-addressable cache.
193
///
194
/// Right now, this is a [`Sha256`] hash, but this may change in future.
195
pub(crate) type DocumentId = Sha256;
196

            
197
/// The supported content encodings.
198
#[derive(
199
    Debug, Clone, Copy, PartialEq, Eq, Hash, strum::EnumString, strum::Display, strum::EnumIter,
200
)]
201
#[strum(serialize_all = "kebab-case", ascii_case_insensitive)]
202
pub(crate) enum ContentEncoding {
203
    /// RFC2616 section 3.5.
204
    Identity,
205
    /// RFC2616 section 3.5.
206
    Deflate,
207
    /// RFC2616 section 3.5.
208
    Gzip,
209
    /// The zstandard compression algorithm (www.zstd.net).
210
    XZstd,
211
    /// The lzma compression algorithm with a "present" value no higher than 6.
212
    XTorLzma,
213
}
214

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

            
244
impl From<SystemTime> for Timestamp {
245
20188
    fn from(value: SystemTime) -> Self {
246
20188
        Self(value)
247
20188
    }
248
}
249

            
250
impl From<Timestamp> for SystemTime {
251
60
    fn from(value: Timestamp) -> Self {
252
60
        value.0
253
60
    }
254
}
255

            
256
impl Add<Duration> for Timestamp {
257
    type Output = Self;
258

            
259
    /// Performs a saturating addition wrapping to [`SystemTime::max_value()`].
260
10008
    fn add(self, rhs: Duration) -> Self::Output {
261
10008
        Self(self.0.saturating_add(rhs))
262
10008
    }
263
}
264

            
265
impl Sub<Duration> for Timestamp {
266
    type Output = Self;
267

            
268
    /// Performs a saturating subtraction wrapping to [`SystemTime::min_value()`].
269
    fn sub(self, rhs: Duration) -> Self::Output {
270
        Self(self.0.saturating_sub(rhs))
271
    }
272
}
273

            
274
impl Sub<Timestamp> for Timestamp {
275
    type Output = Duration;
276

            
277
    /// Performs a saturating duration_since wrapping to [`Duration::ZERO`].
278
10004
    fn sub(self, rhs: Timestamp) -> Self::Output {
279
        #[allow(unstable_name_collisions)]
280
10004
        self.0.saturating_duration_since(rhs.0)
281
10004
    }
282
}
283

            
284
impl FromSql for Timestamp {
285
129
    fn column_result(value: ValueRef<'_>) -> FromSqlResult<Self> {
286
129
        let mut res = SystemTime::UNIX_EPOCH;
287
129
        res = res.saturating_add(Duration::from_secs(value.as_i64()?.try_into().unwrap_or(0)));
288
129
        Ok(Self(res))
289
129
    }
290
}
291

            
292
impl ToSql for Timestamp {
293
168
    fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
294
        #[allow(unstable_name_collisions)]
295
168
        Ok(ToSqlOutput::from(
296
168
            self.0
297
168
                .saturating_duration_since(SystemTime::UNIX_EPOCH)
298
168
                .as_secs()
299
168
                .try_into()
300
168
                .unwrap_or(i64::MAX),
301
168
        ))
302
168
    }
303
}
304

            
305
/// Representation of consensus metadata from the database.
306
#[derive(Educe)]
307
#[educe(Debug, Clone, Copy, PartialEq, Eq)]
308
pub(crate) struct ConsensusMeta<T> {
309
    /// The document id uniquely identifying the consensus.
310
    pub docid: DocumentId,
311

            
312
    /// The SHA3 of the unsigned part of the consensus.
313
    pub sha3_256: Sha3_256,
314

            
315
    /// The time after which this consensus is valid.
316
    pub valid_after: Timestamp,
317

            
318
    /// The time after which this consensus stops being fresh.
319
    pub fresh_until: Timestamp,
320

            
321
    /// The time after which this consensus stops being valid.
322
    pub valid_until: Timestamp,
323

            
324
    /// The flavor of the consensus; determined at compile time.
325
    flavor: PhantomData<T>,
326
}
327

            
328
impl<T: FlavoredConsensusUnverified> ConsensusMeta<T> {
329
    /// Select the missing router descriptors.
330
    ///
331
    /// A router descriptor is considered missing if it exists in
332
    /// `consensus_descriptor_member` but not in `descriptor`
333
    /// because the first entry is added once the consensus got parsed,
334
    /// whereas the second entry is added once we have actually retrieved it.
335
    ///
336
    /// It works by doing a left join on descriptor and filtering for
337
    /// all entries where the join is NULL, as that implies we are aware of
338
    /// the descriptor but not have it stored.
339
    ///
340
    /// Parameters:
341
    /// :docid - The docid of the consensus.
342
    /// :exclude - The descriptors to skip.
343
    /// :limit - The maximum number of descriptors to return.
344
    //
345
    // TODO DIRMIRROR: Potentially constify more queries.
346
    const MISSING_ROUTERS_QUERY: &'static str = sql!(
347
        "
348
        SELECT cr.sha1, RANDOM() AS rand
349
        FROM consensus_descriptor_member AS cr
350
          LEFT JOIN descriptor AS router ON cr.sha1 = router.sha1
351
        WHERE
352
          cr.consensus_docid = :docid
353
          AND cr.sha1 IS NOT NULL
354
          AND cr.sha1 NOT IN rarray(:exclude)
355
          AND router.sha1 IS NULL
356
        ORDER BY rand
357
        LIMIT :limit
358
        "
359
    );
360

            
361
    /// Obtains all consensuses found in the database.
362
    ///
363
    /// This should be reasonable size-wise, given that a ConsensusMeta instance
364
    /// is very small and that the garbage collector removes old consensuses
365
    /// anyways.  If this becomes a problem, we may want to add an optional
366
    /// limit.
367
11
    pub(crate) fn query(tx: &Transaction) -> Result<Vec<Self>, DatabaseError> {
368
        // Select the most recent flavored consensus document from the database.
369
11
        let mut meta_stmt = tx.prepare_cached(sql!(
370
11
            "
371
11
            SELECT docid, sha3_256, valid_after, fresh_until, valid_until
372
11
            FROM consensus
373
11
            WHERE
374
11
              flavor = :flavor
375
11
            ORDER BY valid_after DESC
376
11
            "
377
11
        ))?;
378

            
379
        // Actually execute the query.
380
11
        let rows = meta_stmt.query_map(
381
11
            named_params! {
382
11
                ":flavor": T::flavor().name(),
383
            },
384
11
            |row| {
385
                Ok(Self {
386
11
                    docid: row.get(0)?,
387
11
                    sha3_256: row.get(1)?,
388
11
                    valid_after: row.get(2)?,
389
11
                    fresh_until: row.get(3)?,
390
11
                    valid_until: row.get(4)?,
391
11
                    flavor: Default::default(),
392
                })
393
11
            },
394
        )?;
395

            
396
11
        Ok(rows.collect::<Result<Vec<_>, _>>()?)
397
11
    }
398

            
399
    /// Queries the raw data of a [`ConsensusMeta`].
400
5
    pub(crate) fn data(&self, tx: &Transaction<'_>) -> Result<String, DatabaseError> {
401
5
        let mut stmt = tx.prepare_cached(sql!(
402
5
            "
403
5
            SELECT content
404
5
            FROM store
405
5
            WHERE docid = :docid
406
5
            "
407
5
        ))?;
408

            
409
5
        let raw = stmt.query_one(named_params! {":docid": self.docid}, |row| {
410
5
            row.get::<_, Vec<u8>>(0)
411
5
        })?;
412
5
        let raw = String::from_utf8(raw).map_err(into_internal!("utf-8 constraint violated?"))?;
413
5
        Ok(raw)
414
5
    }
415

            
416
    /// Calculates the [`Timestamp`] at which the authorities will be queried again.
417
    ///
418
    /// # Specifications
419
    ///
420
    /// * <https://spec.torproject.org/dir-spec/directory-cache-operation.html#download-ns-from-auth>
421
10003
    pub(crate) fn ttl<R: Rng>(&self, rng: &mut R) -> Timestamp {
422
10003
        assert!(self.fresh_until < self.valid_until);
423

            
424
10003
        let offset = rng
425
10003
            .gen_range_checked(0..=((self.valid_until - self.fresh_until).as_secs() / 2))
426
10003
            .expect("invalid range?");
427

            
428
10003
        self.fresh_until + Duration::from_secs(offset)
429
10003
    }
430

            
431
    /// Returns the missing router descriptors for this consensus.
432
    ///
433
    /// `limit` may be given to specify an optional upper limit, in which case
434
    /// the result will contain at most `limit` missing descriptors.
435
    ///
436
    /// # Performance
437
    ///
438
    /// The performance here is O(n * log n) for each 64-element batch, leading
439
    /// to an overall performance of O(n^2 * (log n)/64) per consensus period,
440
    /// which sould be acceptable.  See the link below for a further discussion
441
    /// on performance related matters:
442
    ///
443
    /// <https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/4378#note_3467300>
444
    // TODO DIRMIRROR: This has a large overlap with missing_extras and
445
    // missing_micros, we should genericise or macrofy this at one point.
446
14
    pub(crate) fn missing_routers(
447
14
        &self,
448
14
        tx: &Transaction<'_>,
449
14
        exclude: impl Iterator<Item = Sha1>,
450
14
        limit: Option<u64>,
451
14
    ) -> Result<HashSet<Sha1>, DatabaseError> {
452
14
        if T::flavor() != ConsensusFlavor::Plain {
453
            return Ok(HashSet::new());
454
14
        }
455

            
456
14
        let mut stmt = tx.prepare_cached(Self::MISSING_ROUTERS_QUERY)?;
457

            
458
14
        let exclude = exclude.map(Value::from).collect();
459
14
        let exclude: Rc<Vec<Value>> = Rc::new(exclude);
460
14
        let limit = limit.map_or(-1, |n| n.try_into().unwrap_or(i64::MAX));
461
14
        let missing = stmt
462
14
            .query_map(
463
14
                named_params! {":docid": self.docid, ":limit": limit, ":exclude": exclude},
464
52
                |row| row.get(0),
465
            )?
466
14
            .collect::<Result<HashSet<_>, _>>()?;
467
14
        Ok(missing)
468
14
    }
469

            
470
    /// Returns the missing extra infos for this consensus to the best of our abilities.
471
    ///
472
    /// Keep in mind that this does not return **all** missing extra infos but
473
    /// only the missing extra infos of router descriptors we have.
474
    ///
475
    /// `limit` may be given to specify an optional upper limit, in which case
476
    /// the result will contain at most `limit` missing extra-infos.
477
    ///
478
    /// See [`ConsensusMeta::missing_routers()`] for a discussion on
479
    /// performance.
480
3
    pub(crate) fn missing_extras(
481
3
        &self,
482
3
        tx: &Transaction<'_>,
483
3
        exclude: impl Iterator<Item = Sha1>,
484
3
        limit: Option<u64>,
485
3
    ) -> Result<HashSet<Sha1>, DatabaseError> {
486
3
        if T::flavor() != ConsensusFlavor::Plain {
487
            return Ok(HashSet::new());
488
3
        }
489

            
490
        // Select the missing extra infos for this consensus.
491
        //
492
        // This return value is not complete because we only know the missing
493
        // extra-infos to the best of our abilities.  In other words: We are
494
        // only aware of a missing extra-info if we have parsed the respective
495
        // router descriptor.
496
        //
497
        // It works by doing an inner join from
498
        // `consensus_descriptor_member` to `descriptor` because
499
        // we can only know about the extra-infos of which we have the router
500
        // descriptors from.  Afterwards, we do a left join with the
501
        // `extra_info` table and filter for all results where the left
502
        // join result is null, hence where we have a router descriptor but not
503
        // the respective extra-info.
504
        //
505
        // Parameters:
506
        // :docid - The docid of the consensus.
507
        // :exclude - The descriptors to skip.
508
        // :limit - The maximum number of descriptors to return.
509
3
        let mut stmt = tx.prepare_cached(sql!(
510
3
            "
511
3
            SELECT router.extra_sha1, RANDOM() AS rand
512
3
            FROM consensus_descriptor_member AS cr
513
3
              INNER JOIN descriptor AS router ON cr.sha1 = router.sha1
514
3
              LEFT JOIN extra_info AS extra ON router.extra_sha1 = extra.sha1
515
3
            WHERE
516
3
              cr.consensus_docid = :docid
517
3
              AND router.extra_sha1 IS NOT NULL
518
3
              AND extra.sha1 IS NULL
519
3
              AND extra.sha1 NOT IN rarray(:exclude)
520
3
            ORDER BY rand
521
3
            LIMIT :limit
522
3
            "
523
3
        ))?;
524

            
525
3
        let exclude = exclude.map(Value::from).collect();
526
3
        let exclude: Rc<Vec<Value>> = Rc::new(exclude);
527
3
        let limit = limit.map_or(-1, |n| n.try_into().unwrap_or(i64::MAX));
528
3
        let missing = stmt
529
3
            .query_map(
530
3
                named_params! {":docid": self.docid, ":limit": limit, ":exclude": exclude},
531
                |row| row.get(0),
532
            )?
533
3
            .collect::<Result<HashSet<_>, _>>()?;
534
3
        Ok(missing)
535
3
    }
536

            
537
    /// Returns the missing micro descriptors for this consensus.
538
    ///
539
    /// `limit` may be given to specify an optional upper limit, in which case
540
    /// the result will contain at most `limit` missing descriptors.
541
    ///
542
    /// See [`ConsensusMeta::missing_routers()`] for a discussion on
543
    /// performance.
544
4
    pub(crate) fn missing_micros(
545
4
        &self,
546
4
        tx: &Transaction<'_>,
547
4
        exclude: impl Iterator<Item = Sha256>,
548
4
        limit: Option<u64>,
549
4
    ) -> Result<HashSet<Sha256>, DatabaseError> {
550
4
        if T::flavor() != ConsensusFlavor::Microdesc {
551
2
            return Ok(HashSet::new());
552
2
        }
553

            
554
        // Select the missing micro descriptors.
555
        //
556
        // A micro descriptor is considered missing if it exists in
557
        // `consensus_descriptor_member` but not in `descriptor`
558
        // because the first entry is added once the consensus got parsed,
559
        // whereas the second entry is added once we have actually retrieved it.
560
        //
561
        // It works by doing a left join on descriptor and filtering for
562
        // all entries where the join is NULL, as that implies we are aware of
563
        // the descriptor but not have it stored.
564
        //
565
        // Parameters:
566
        // :docid - The docid of the consensus.
567
        // :limit - The maximum number of descriptors to return.
568
2
        let mut stmt = tx.prepare_cached(sql!(
569
2
            "
570
2
            SELECT cr.sha2, RANDOM() AS rand
571
2
            FROM consensus_descriptor_member AS cr
572
2
              LEFT JOIN descriptor AS micro ON cr.sha2 = micro.sha2
573
2
            WHERE
574
2
              cr.consensus_docid = :docid
575
2
              AND cr.sha2 IS NOT NULL
576
2
              AND micro.sha2 IS NULL
577
2
              AND micro.sha2 NOT IN rarray(:exclude)
578
2
            ORDER BY rand
579
2
            LIMIT :limit
580
2
            "
581
2
        ))?;
582

            
583
2
        let exclude = exclude.map(Value::from).collect();
584
2
        let exclude: Rc<Vec<Value>> = Rc::new(exclude);
585
2
        let limit = limit.map_or(-1, |n| n.try_into().unwrap_or(i64::MAX));
586
2
        let missing = stmt
587
2
            .query_map(
588
2
                named_params! {":docid": self.docid, ":limit": limit, ":exclude": exclude},
589
8
                |row| row.get(0),
590
            )?
591
2
            .collect::<Result<HashSet<_>, _>>()?;
592
2
        Ok(missing)
593
4
    }
594

            
595
    /// Inserts a verified consensus into the database.
596
    ///
597
    /// This method *DOES NOT* compute any consensus diffs but populates
598
    /// associated meta tables such as `consensus_descriptor_member`
599
    /// and `consensus_authority_voter` properly.
600
24
    pub(crate) fn insert<I>(
601
24
        tx: &Transaction<'_>,
602
24
        encodings: I,
603
24
        (body, sigs): (&T::Body, &T::Signatures),
604
24
        data: &str,
605
24
    ) -> Result<ConsensusMeta<T>, DatabaseError>
606
24
    where
607
24
        I: Iterator<Item = ContentEncoding>,
608
    {
609
        // Insert a consensus into the consensus meta table.
610
        //
611
        // This requires the consensus to already be present in the `store`
612
        // table.
613
        //
614
        // Parameters:
615
        // :docid - The docid of the consensus.
616
        // :sha3_256 - The SHA3-256 sum of the part of the consensus
617
        //   without signatures but a trailing `\ndirectory-signature<SPACE>`.
618
        // :flavor - The consensus flavor to use.
619
        // :valid_after, :fresh_until, :valid_until - The respective validities.
620
24
        let mut cons_stmt = tx.prepare_cached(sql!(
621
24
            "
622
24
            INSERT INTO consensus
623
24
            (docid, sha3_256, flavor, valid_after, fresh_until, valid_until)
624
24
            VALUES
625
24
            (:docid, :sha3_256, :flavor, :valid_after, :fresh_until, :valid_until)
626
24
            ON CONFLICT DO NOTHING
627
24
            "
628
24
        ))?;
629

            
630
        // Insert the relationship between a router status to a consensus.
631
        //
632
        // Parameters:
633
        // :docid - The consensus docid.
634
        // :sha1 - The router descriptor digest (plain consensus only, NULL otherwise).
635
        // :sha2 - The micro descriptor digest (microdesc consensus only, NULL otherwise).
636
24
        let mut cons_rs_member_stmt = tx.prepare_cached(sql!(
637
24
            "
638
24
            INSERT INTO consensus_descriptor_member
639
24
            (consensus_docid, sha1, sha2)
640
24
            VALUES
641
24
            (:docid, :sha1, :sha2)
642
24
            ON CONFLICT DO NOTHING
643
24
            "
644
24
        ))?;
645

            
646
        // Insert the relationship between authority certificate to consensus votes.
647
        //
648
        // Parameters:
649
        // :cons - The consensus docid.
650
        // :auth - The authority certificate docid.
651
24
        let mut cons_auth_voter = tx.prepare_cached(sql!(
652
24
            "
653
24
            INSERT INTO consensus_authority_voter
654
24
            (consensus_docid, kp_auth_id_rsa_sha1)
655
24
            VALUES
656
24
            (:consensus_docid, :id_rsa)
657
24
            ON CONFLICT DO NOTHING
658
24
            "
659
24
        ))?;
660

            
661
        // First, insert the consensus as is.
662
24
        let docid = store_insert(tx, data.as_bytes(), encodings)?;
663

            
664
        // TODO DIRMIRROR: We *need* this in tor-netdoc!!!
665
24
        let sha3_256 = Sha3_256::digest(
666
24
            data.split_inclusive("\ndirectory-signature ")
667
24
                .next()
668
24
                .ok_or_else(|| internal!("verified document without signatures?"))?
669
24
                .as_bytes(),
670
        );
671

            
672
24
        let lifetime = body.lifetime();
673
24
        let meta = ConsensusMeta {
674
24
            docid,
675
24
            sha3_256,
676
24
            valid_after: lifetime.valid_after.0.into(),
677
24
            fresh_until: lifetime.fresh_until.0.into(),
678
24
            valid_until: lifetime.valid_until.0.into(),
679
24
            flavor: Default::default(),
680
24
        };
681
24
        cons_stmt.execute(named_params! {
682
24
            ":docid": meta.docid,
683
24
            ":sha3_256": meta.sha3_256,
684
24
            ":valid_after": meta.valid_after,
685
24
            ":fresh_until": meta.fresh_until,
686
24
            ":valid_until": meta.valid_until,
687
24
            ":flavor": T::flavor().name(),
688
24
        })?;
689

            
690
        // Insert all router descriptor member relationships.
691
        //
692
        // Many small queries are fast in sqlite, so this shouldn't be a perf problem.
693
        // https://sqlite.org/np1queryprob.html
694
        //
695
        // Depending on the flavor, we map the document digests to an iterator
696
        // of a tuple of Option's, where one of the values is always None.
697
        // This represents the fact that router descriptors use SHA-1 and micro
698
        // descriptors SHA-256.
699
168
        for (sha1, sha2) in body.doc_digests() {
700
168
            cons_rs_member_stmt.execute(named_params! {
701
168
                ":docid": docid,
702
168
                ":sha1": sha1,
703
168
                ":sha2": sha2,
704
168
            })?;
705
        }
706

            
707
96
        for sig in sigs.signatories() {
708
96
            cons_auth_voter.execute(named_params! {
709
96
                ":consensus_docid": docid,
710
96
                ":id_rsa": Sha1(sig.id_fingerprint.to_bytes()),
711
96
            })?;
712
        }
713

            
714
24
        Ok(meta)
715
24
    }
716
}
717

            
718
/// Representation of authority certificate metadata from the database.
719
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
720
pub(crate) struct AuthCertMeta {
721
    /// The document id uniquely identifying the consensus.
722
    pub docid: DocumentId,
723

            
724
    /// The SHA-1 fingerprint of the identity key.
725
    // TODO DIRMIRROR: Change this to RsaIdentity.
726
    pub kp_auth_id_rsa_sha1: Sha1,
727

            
728
    /// The SHA-1 fingerprint of the signign key.
729
    // TODO DIRMIRROR: Change this to RsaIdentity.
730
    pub kp_auth_sign_rsa_sha1: Sha1,
731

            
732
    /// The timestamp after which this certificate will be valid.
733
    pub dir_key_published: Timestamp,
734

            
735
    /// The timestamp until this certificate will be valid.
736
    pub dir_key_expires: Timestamp,
737
}
738

            
739
impl AuthCertMeta {
740
    /// Obtains the authority certificates from the database.
741
14
    pub(crate) fn query(tx: &Transaction) -> Result<Vec<Self>, DatabaseError> {
742
        // Obtain all certificates from the database.
743
        //
744
        // This is okay because the set is not very big.
745
        //
746
        // In the unlikely edge case of on identity-signing key pair having
747
        // multiple certificates, the most recently published certificate is
748
        // going to be used.
749
        //
750
        // TODO DIRMIRROR: Perhaps we should modify the auth_certs table to
751
        // add a UNIQUE constraint on that combination, while modifying the
752
        // insertion logic to replace with the newer one in the case of a
753
        // conflict.
754
14
        let mut stmt = tx.prepare_cached(sql!(
755
14
            "
756
14
            SELECT docid, kp_auth_id_rsa_sha1, kp_auth_sign_rsa_sha1,
757
14
              dir_key_published, dir_key_expires
758
14
            FROM authority_key_certificate
759
14
            GROUP BY kp_auth_id_rsa_sha1, kp_auth_sign_rsa_sha1
760
14
            ORDER BY MAX(dir_key_published)
761
14
            "
762
14
        ))?;
763

            
764
14
        let certs = stmt
765
48
            .query_map(params![], |row| {
766
                Ok(Self {
767
48
                    docid: row.get(0)?,
768
48
                    kp_auth_id_rsa_sha1: row.get(1)?,
769
48
                    kp_auth_sign_rsa_sha1: row.get(2)?,
770
48
                    dir_key_published: row.get(3)?,
771
48
                    dir_key_expires: row.get(4)?,
772
                })
773
48
            })?
774
14
            .collect::<Result<_, _>>()?;
775
14
        Ok(certs)
776
14
    }
777

            
778
    /// Queries the raw data of an [`AuthCertMeta`].
779
48
    pub(crate) fn data(&self, tx: &Transaction<'_>) -> Result<String, DatabaseError> {
780
48
        let mut stmt = tx.prepare_cached(sql!(
781
48
            "
782
48
            SELECT content
783
48
            FROM store
784
48
            WHERE docid = :docid
785
48
            "
786
48
        ))?;
787

            
788
48
        let raw = stmt.query_one(named_params! {":docid": self.docid}, |row| {
789
48
            row.get::<_, Vec<u8>>(0)
790
48
        })?;
791
48
        let raw = String::from_utf8(raw).map_err(into_internal!("utf-8 constraint violated?"))?;
792
48
        Ok(raw)
793
48
    }
794

            
795
    /// Inserts a new authority certificate into the database.
796
    ///
797
    /// Keep in mind that the data in the [`AuthCert`] should correspond to the
798
    /// data found in `data`, as this method performs no parsing.
799
48
    pub(crate) fn insert<I: Iterator<Item = ContentEncoding>>(
800
48
        tx: &Transaction<'_>,
801
48
        encodings: I,
802
48
        cert: &AuthCert,
803
48
        data: &str,
804
48
    ) -> Result<(), DatabaseError> {
805
        // Inserts a new certificate into the meta table.
806
        //
807
        // Parameters:
808
        // :docid - The document id.
809
        // :id_rsa - The identity key fingerprint.
810
        // :sign_rsa - The signing key fingerprint
811
        // :published - The published timestamp.
812
        // :expires - The expires timestamp.
813
48
        let mut stmt = tx.prepare_cached(sql!(
814
48
            "
815
48
            INSERT INTO authority_key_certificate
816
48
            (docid, kp_auth_id_rsa_sha1, kp_auth_sign_rsa_sha1, dir_key_published, dir_key_expires)
817
48
            VALUES
818
48
            (:docid, :id_rsa, :sign_rsa, :published, :expires)
819
48
            ON CONFLICT DO NOTHING
820
48
            "
821
48
        ))?;
822

            
823
48
        let docid = store_insert(tx, data.as_bytes(), encodings)?;
824
48
        stmt.execute(named_params! {
825
48
            ":docid": docid,
826
48
            ":id_rsa": cert.dir_identity_key.to_rsa_identity().as_hex_upper(),
827
48
            ":sign_rsa": cert.dir_signing_key.to_rsa_identity().as_hex_upper(),
828
48
            ":published": Timestamp::from(cert.dir_key_published.0),
829
48
            ":expires": Timestamp::from(cert.dir_key_expires.0),
830
48
        })?;
831

            
832
48
        Ok(())
833
48
    }
834
}
835

            
836
/// Representation of a descriptor from the database.
837
///
838
/// This is different from [`FlavoredDescriptor`] in the sense that it represents
839
/// a row in the `descriptor` table of the database, whereas [`FlavoredDescriptor`]
840
/// represents an agnostic wrapper around the [`tor_netdoc`] equivalent.
841
///
842
/// Yet, it still accepts a [`FlavoredDescriptor`] as a type parameter in order
843
/// to obtain the flavor from it.
844
// TODO DIRMIRROR: Is it possible to change this generic type somehow?
845
// Potentially a const ConsensusFlavor?  After all, this is not a wrapper
846
// type but rather a unit marker which we only use to call ::flavor().
847
// Similar logic probably also applies to ConsensusMeta.
848
#[derive(Educe)]
849
#[educe(Debug, Clone, Copy, PartialEq, Eq, Hash)]
850
pub(crate) struct DescriptorMeta<Flavor> {
851
    /// The document id uniquely identifying the document.
852
    pub docid: DocumentId,
853

            
854
    /// The SHA-1 as found in the signature.
855
    pub sha1: Sha1,
856

            
857
    /// The SHA-256 as found in the signature.
858
    pub sha2: Sha256,
859

            
860
    /// The RSA identity fingerprint, if present.
861
    pub kp_relay_id_rsa_sha1: Option<Sha1>,
862

            
863
    /// Use of the generic for determining the flavor.
864
    flavor: PhantomData<Flavor>,
865

            
866
    /// The SHA-1 of the extra-info document, if present.
867
    pub extra_sha1: Option<Sha1>,
868
}
869

            
870
impl<Flavor: FlavoredDescriptor> DescriptorMeta<Flavor> {
871
    /// Queries a [`DescriptorMeta`] by it's [`DocumentId`], mostly for internal
872
    /// use.
873
    fn query_docid(
874
        tx: &Transaction<'_>,
875
        docid: DocumentId,
876
    ) -> Result<Option<DescriptorMeta<Flavor>>, DatabaseError> {
877
        let mut stmt = tx.prepare_cached(sql!(
878
            "
879
            SELECT docid, sha1, sha2, kp_relay_id_rsa_sha1, extra_sha1
880
            FROM descriptor
881
            WHERE docid = :docid AND flavor = :flavor
882
            "
883
        ))?;
884

            
885
        stmt.query_one(
886
            named_params! {
887
                ":docid": docid,
888
                ":flavor": Flavor::flavor().name()
889
            },
890
            |row| {
891
                Ok(DescriptorMeta::<Flavor> {
892
                    docid: row.get(0)?,
893
                    sha1: row.get(1)?,
894
                    sha2: row.get(2)?,
895
                    kp_relay_id_rsa_sha1: row.get(3)?,
896
                    extra_sha1: row.get(4)?,
897
                    flavor: Default::default(),
898
                })
899
            },
900
        )
901
        .optional()
902
        .map_err(|e| e.into())
903
    }
904

            
905
    /// Inserts a [`FlavoredDescriptor`] into the database.
906
7
    pub(crate) fn insert<I>(
907
7
        tx: &Transaction<'_>,
908
7
        encodings: I,
909
7
        descriptor: &Flavor,
910
7
        data: &str,
911
7
    ) -> Result<(), DatabaseError>
912
7
    where
913
7
        I: Iterator<Item = ContentEncoding>,
914
    {
915
        // Insertion statement of a single descriptor.
916
        //
917
        // Parameters:
918
        // :docid - The document id.
919
        // :sha1 - The SHA-1 as found in the signature.
920
        // :sha256 - The SHA-256 as found in the signature.
921
        // :id_rsa - The RSA identity key fingerprint, if present.
922
        // :flavor - The ConsensusFlavor associated with it.
923
        // :extra_info - The extra-info SHA-1, if present.
924
7
        let mut stmt = tx.prepare_cached(sql!(
925
7
            "
926
7
            INSERT INTO descriptor
927
7
            (docid, sha1, sha2, kp_relay_id_rsa_sha1, flavor, extra_sha1)
928
7
            VALUES
929
7
            (:docid, :sha1, :sha256, :id_rsa, :flavor, :extra_info)
930
7
            "
931
7
        ))?;
932

            
933
7
        let docid = store_insert(tx, data.as_bytes(), encodings)?;
934
7
        stmt.execute(named_params! {
935
            ":docid": docid,
936
7
            ":sha1": descriptor.sha1()?,
937
7
            ":sha256": descriptor.sha256()?,
938
7
            ":id_rsa": descriptor.id_rsa().map(|rsa| Sha1(rsa.to_bytes())),
939
7
            ":flavor": Flavor::flavor().name(),
940
7
            ":extra_info": descriptor.extra_info_sha1().map(|extra| Sha1(extra.sha1.0)),
941
        })?;
942

            
943
7
        Ok(())
944
7
    }
945

            
946
    /// Queries the raw data of a [`DescriptorMeta`].
947
    pub(crate) fn data(&self, tx: &Transaction<'_>) -> Result<String, DatabaseError> {
948
        let mut stmt = tx.prepare_cached(sql!("SELECT content FROM store WHERE docid = :docid"))?;
949

            
950
        let raw = stmt.query_one(named_params! {":docid": self.docid}, |row| {
951
            row.get::<_, Vec<u8>>(0)
952
        })?;
953
        let raw = String::from_utf8(raw).map_err(into_internal!("utf-8 constraint violated?"))?;
954
        Ok(raw)
955
    }
956
}
957

            
958
/// A no-op macro just returning the supplied.
959
///
960
/// The purpose of this macro is to semantically mark [`str`] literals to be
961
/// SQL statement.
962
///
963
/// Keep in mind that the compiler will not notice if you forget this macro.
964
/// Unfortunately, you have to ensure it yourself.
965
macro_rules! sql {
966
    ($s:literal) => {
967
        $s
968
    };
969
}
970

            
971
pub(crate) use sql;
972

            
973
/// Opens a database from disk, creating a [`Pool`] for it.
974
///
975
/// This function should be the entry point for all things requiring a database
976
/// handle, as this function prepares all necessary steps required for operating
977
/// on the database correctly, such as:
978
/// * Schema initialization.
979
/// * Schema upgrade.
980
/// * Setting connection specific settings.
981
///
982
/// # `SQLITE_BUSY` Caveat
983
///
984
/// There is a problem with the handling of `SQLITE_BUSY` when opening an
985
/// SQLite database.  In WAL, opening a database might acquire an exclusive lock
986
/// for a very short amount of time, in order to perform clean-up from previous
987
/// connections alongside other tasks for maintaining database integrity?  This
988
/// means, that opening multiple SQLite databases simultaneously will result in
989
/// a busy error regardless of a busy handler, as setting a busy handler will
990
/// require an existing connection, something we are unable to obtain in the
991
/// first place.
992
///
993
/// In order to mitigate this issue, the recommended way in the SQLite community
994
/// is to simply ensure that database connections are opened sequentially,
995
/// by urging calling applications to just use a single [`Pool`] instance.
996
///
997
/// Testing this is hard unfortunately.
998
22
pub(crate) fn open<P: AsRef<Path>>(
999
22
    path: P,
22
) -> Result<Pool<SqliteConnectionManager>, DatabaseError> {
22
    let num_cores = std::thread::available_parallelism()
22
        .unwrap_or(NonZero::new(8).expect("8 == 0?"))
22
        .get() as u32;
22
    let manager = r2d2_sqlite::SqliteConnectionManager::file(&path)
2112
        .with_init(|c| rusqlite::vtab::array::load_module(c));
22
    let pool = Pool::builder().max_size(num_cores).build(manager)?;
22
    rw_tx(&pool, |tx| {
        // Prepare the database, doing the following steps:
        // 1. Checking the database schema.
        // 2. Upgrading (in future) or initializing the database schema (if empty).
22
        let has_arti_dirserver_schema_version = match tx.query_one(
            sql!(
22
                "
22
                SELECT name
22
                FROM sqlite_master
22
                  WHERE type = 'table'
22
                    AND name = 'arti_dirserver_schema_version'
22
                "
            ),
22
            params![],
1
            |_| Ok(()),
        ) {
1
            Ok(()) => true,
21
            Err(rusqlite::Error::QueryReturnedNoRows) => false,
            Err(e) => return Err(DatabaseError::LowLevel(e)),
        };
22
        if has_arti_dirserver_schema_version {
1
            let version = tx.query_one(
1
                sql!("SELECT version FROM arti_dirserver_schema_version WHERE rowid = 1"),
1
                params![],
1
                |row| row.get::<_, String>(0),
            )?;
1
            match version.as_ref() {
1
                "1" => {}
1
                unknown => {
1
                    return Err(DatabaseError::IncompatibleSchema {
1
                        version: unknown.into(),
1
                    });
                }
            }
        } else {
21
            tx.execute_batch(V1_SCHEMA)?;
        }
21
        Ok::<_, DatabaseError>(())
22
    })??;
21
    Ok(pool)
22
}
/// Executes a closure `op` with a given read-only [`Transaction`].
///
/// The [`Transaction`] always gets rolled back the moment `op` returns.
///
/// The [`Transaction`] gets initialized with the global pragma options set.
///
/// **The closure shall not perform write operations!**
/// Not only do they get rolled back anyways, but upgrading the [`Transaction`]
/// from a read to a write transaction will lead to other simultaneous write upgrades
/// to fail.  Unfortunately, there is no real programmatic way to ensure this.
36
pub(crate) fn read_tx<U, F>(pool: &Pool<SqliteConnectionManager>, op: F) -> Result<U, DatabaseError>
36
where
36
    F: FnOnce(&Transaction<'_>) -> U,
{
36
    let mut conn = pool.get()?;
36
    conn.execute_batch(GLOBAL_OPTIONS)?;
36
    let tx = conn.transaction_with_behavior(TransactionBehavior::Deferred)?;
36
    let res = op(&tx);
36
    tx.rollback()?;
36
    Ok(res)
36
}
/// Executes a closure `op` with a given read-write [`Transaction`].
///
/// The [`Transaction`] always gets committed the moment `op` returns.
///
/// The [`Transaction`] gets initialized with the global pragma options set.
///
/// The [`Transaction`] gets created with [`TransactionBehavior::Immediate`],
/// meaning it will immediately exist as a write connection, retrying in the
/// case of a [`rusqlite::ErrorCode::DatabaseBusy`] until it failed after 1s.
40
pub(crate) fn rw_tx<U, F>(pool: &Pool<SqliteConnectionManager>, op: F) -> Result<U, DatabaseError>
40
where
40
    F: FnOnce(&Transaction<'_>) -> U,
{
40
    let mut conn = pool.get()?;
40
    conn.execute_batch(GLOBAL_OPTIONS)?;
40
    let tx = conn.transaction_with_behavior(TransactionBehavior::Exclusive)?;
39
    let res = op(&tx);
39
    tx.commit()?;
39
    Ok(res)
40
}
/// Inserts `data` into store while also compressing it with given encodings.
///
/// Returns the [`DocumentId`] of `data`.
///
/// This function inserts `data` into store and also compresses it into all
/// given compression formats.
///
/// Duplicates get re-encoded and replaced in the database, including
/// [`ContentEncoding::Identity`].
379
pub(crate) fn store_insert<I: Iterator<Item = ContentEncoding>>(
379
    tx: &Transaction,
379
    data: &[u8],
379
    encodings: I,
379
) -> Result<DocumentId, DatabaseError> {
    // The statement to insert some data into the store.
    //
    // Parameters:
    // :docid - The docid.
    // :content - The binary data.
379
    let mut store_stmt = tx.prepare_cached(sql!(
379
        "
379
        INSERT INTO store (docid, content)
379
        VALUES
379
        (:docid, :content)
379
        ON CONFLICT DO NOTHING
379
        "
379
    ))?;
    // The statement to insert a compressed document into the metatable.
    //
    // Parameters:
    // :algorithm - The name of the encoding algorithm.
    // :identity_docid - The docid of the plain-text document in the store.
    // :compressed_docid - The docid of the encoded document in the store.
379
    let mut compressed_stmt = tx.prepare_cached(sql!(
379
        "
379
        INSERT INTO compressed_document (algorithm, identity_docid, compressed_docid)
379
        VALUES
379
        (:algorithm, :identity_docid, :compressed_docid)
379
        ON CONFLICT DO NOTHING
379
        "
379
    ))?;
    // Insert the plain document into the store.
379
    let identity_docid = DocumentId::digest(data);
379
    store_stmt.execute(named_params! {
379
        ":docid": identity_docid,
379
        ":content": data
379
    })?;
    // Compress it into all formats and insert it into store and compressed.
439
    for encoding in encodings {
439
        if encoding == ContentEncoding::Identity {
            // Ignore identity because we inserted that above.
379
            continue;
60
        }
        // We map a compression error to a bug because there is no good reason
        // on why it should fail, given that we compress from memory data to
        // memory data.  Probably because it uses the std::io::Writer interface
        // which itself demands use of std::io::Result.
60
        let compressed = compress(data, encoding).map_err(into_internal!("{encoding} failed?"))?;
60
        let compressed_docid = DocumentId::digest(&compressed);
60
        store_stmt.execute(named_params! {
60
            ":docid": compressed_docid,
60
            ":content": compressed,
60
        })?;
60
        compressed_stmt.execute(named_params! {
60
            ":algorithm": encoding.to_string(),
60
            ":identity_docid": identity_docid,
60
            ":compressed_docid": compressed_docid,
60
        })?;
    }
379
    Ok(identity_docid)
379
}
/// Compresses `data` into a specified [`ContentEncoding`].
///
/// Returns a [`Vec`] containing the encoded data.
65
fn compress(data: &[u8], encoding: ContentEncoding) -> Result<Vec<u8>, std::io::Error> {
65
    match encoding {
1
        ContentEncoding::Identity => Ok(data.to_vec()),
        ContentEncoding::Deflate => {
16
            let mut w = DeflateEncoder::new(Vec::new(), Default::default());
16
            w.write_all(data)?;
16
            w.finish()
        }
        ContentEncoding::Gzip => {
16
            let mut w = GzEncoder::new(Vec::new(), Default::default());
16
            w.write_all(data)?;
16
            w.finish()
        }
16
        ContentEncoding::XZstd => zstd::encode_all(data, Default::default()),
        ContentEncoding::XTorLzma => {
16
            let mut res = Vec::new();
16
            lzma_rs::lzma_compress(&mut Cursor::new(data), &mut res)?;
16
            Ok(res)
        }
    }
65
}
#[cfg(test)]
mod test {
    // @@ begin test lint list maintained by maint/add_warning @@
    #![allow(clippy::bool_assert_comparison)]
    #![allow(clippy::clone_on_copy)]
    #![allow(clippy::dbg_macro)]
    #![allow(clippy::mixed_attributes_style)]
    #![allow(clippy::print_stderr)]
    #![allow(clippy::print_stdout)]
    #![allow(clippy::single_char_pattern)]
    #![allow(clippy::unwrap_used)]
    #![allow(clippy::unchecked_time_subtraction)]
    #![allow(clippy::useless_vec)]
    #![allow(clippy::needless_pass_by_value)]
    #![allow(clippy::string_slice)] // See arti#2571
    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
    use std::{
        collections::HashSet,
        io::Read,
        iter,
        sync::{Arc, Once},
    };
    use flate2::read::{DeflateDecoder, GzDecoder};
    use rusqlite::Connection;
    use strum::IntoEnumIterator;
    use tempfile::tempdir;
    use tor_basic_utils::test_rng::testing_rng;
    use tor_netdoc::doc::netstatus::{md, plain};
    use crate::testdata2::{self, current_consensus_ns};
    use super::*;
    type Plain = plain::NetworkStatusUnverified;
    type Md = md::NetworkStatusUnverified;
    #[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);
        }
    }
    /// Verify that the insertion of a consensus works.
    ///
    /// For this, we do not use the test database but rather a fresh one and
    /// call ConsensusMeta::insert().  After, we verify that the consensus
    /// can be queried as well as that the missing descriptors are properly
    /// calculated.
    #[test]
    fn consensus_insert() {
        let pool = open("").unwrap();
        let mut conn = pool.get().unwrap();
        let tx = conn.transaction().unwrap();
        let (body, sigs, data) = current_consensus_ns();
        let meta = ConsensusMeta::<Plain>::insert(
            &tx,
            iter::once(ContentEncoding::Identity),
            (&body, &sigs),
            data,
        )
        .unwrap();
        assert_eq!(
            meta,
            ConsensusMeta::<Plain> {
                docid: Sha256::digest(data.as_bytes()),
                sha3_256: testdata2::consensus_sha3(data),
                valid_after: body.preamble.lifetime.valid_after.0.into(),
                fresh_until: body.preamble.lifetime.fresh_until.0.into(),
                valid_until: body.preamble.lifetime.valid_until.0.into(),
                flavor: Default::default()
            }
        );
        let meta2 = ConsensusMeta::<Plain>::query(&tx).unwrap();
        assert_eq!(meta2, vec![meta]);
        let missing_descs = meta.missing_routers(&tx, iter::empty(), None).unwrap();
        let missing_descs2 = body.routers.iter().map(|r| Sha1(*r.doc_digest())).collect();
        assert_eq!(missing_descs, missing_descs2);
    }
    /// Tests whether the timeout computation lies within the proper interval.
    ///
    /// Because this involves randomness, it performs the test several thousand
    /// times.  This should be okay performance wise, as it takes about ~250ms
    /// with a debug build on my machine.
    #[test]
    fn sync_timeout() {
        // We repeat the tests a few thousand times to go over many random values.
        let docid = Sha256::digest(testdata2::current_consensus_ns().2.as_bytes());
        let lifetime = testdata2::current_consensus_ns().0.preamble.lifetime;
        let sha3_256 = testdata2::consensus_sha3(testdata2::current_consensus_ns().2);
        let cons = ConsensusMeta::<Plain> {
            docid,
            sha3_256,
            valid_after: lifetime.valid_after.0.into(),
            fresh_until: lifetime.fresh_until.0.into(),
            valid_until: lifetime.valid_until.0.into(),
            flavor: Default::default(),
        };
        for _ in 0..10000 {
            let when = cons.ttl(&mut testing_rng());
            assert!(when >= lifetime.fresh_until.0.into());
            // Computes the half between fresh_until and valid_until.
            assert!(
                when <= (lifetime.fresh_until.0
                    + (lifetime
                        .valid_until
                        .0
                        .duration_since(lifetime.fresh_until.0)
                        .unwrap()
                        / 2))
                    .into()
            );
        }
    }
    /// Tests whether the missing router descriptor queue is computed properly.
    ///
    /// For this, we remove existing router descriptors from the database and
    /// see whether they are determined as missing properly.
    #[test]
    fn missing_descriptors() {
        let pool = testdata2::test_db();
        let meta = read_tx(&pool, ConsensusMeta::<Plain>::query)
            .unwrap()
            .unwrap()[0];
        // Ensure that the returned consensus matches the one from testdata2.
        assert_eq!(
            meta.docid,
            DocumentId::digest(testdata2::current_consensus_ns().2.as_bytes())
        );
        // Delete a single router descriptor, so we can determine a missing one
        // using it.
        let removed_descriptor = *testdata2::current_consensus_ns().0.routers[0].doc_digest();
        let removed_descriptor = Sha1::from(removed_descriptor);
        pool.get()
            .unwrap()
            .execute(
                sql!(
                    "
                    DELETE FROM descriptor
                    WHERE sha1 = ?1
                    "
                ),
                params![removed_descriptor],
            )
            .unwrap();
        // Only one should be returned.
        let missing_routers = read_tx(&pool, |tx| meta.missing_routers(tx, iter::empty(), None))
            .unwrap()
            .unwrap();
        assert_eq!(missing_routers, HashSet::from([removed_descriptor]));
        // None should be returned if we exclude the descriptor we just removed.
        let missing_routers = read_tx(&pool, |tx| {
            meta.missing_routers(tx, iter::once(removed_descriptor), None)
        })
        .unwrap()
        .unwrap();
        assert!(missing_routers.is_empty());
        // If we delete all router descriptors we have, we should get all.
        rw_tx(&pool, |tx| {
            tx.execute(sql!("DELETE FROM descriptor"), params![])
        })
        .unwrap()
        .unwrap();
        // Now all should be returned; we verify this by checking that the
        // result is present in all_descriptors, which is a superset.
        let missing_routers = read_tx(&pool, |tx| meta.missing_routers(tx, iter::empty(), None))
            .unwrap()
            .unwrap();
        // This is a superset of missing_routers because it includes router
        // descriptors that are not a part of the current consensus.
        let all_descriptors = testdata2::current_router_descs()
            .iter()
            .map(|x| Sha1::from(x.1.hashes.sha1.unwrap()))
            .collect::<HashSet<_>>();
        assert!(
            missing_routers
                .iter()
                .all(|sha1| all_descriptors.contains(sha1))
        );
        // If we exclude one random descriptor, only that one should be missing.
        let to_exclude = missing_routers.iter().copied().next().unwrap();
        let missing_routers2 = read_tx(&pool, |tx| {
            meta.missing_routers(tx, iter::once(to_exclude), None)
        })
        .unwrap()
        .unwrap();
        assert_eq!(
            missing_routers
                .difference(&missing_routers2)
                .collect::<HashSet<_>>(),
            [&to_exclude].into()
        );
        // If we exclude all descriptors, it should be empty too.
        let missing_routers = read_tx(&pool, |tx| {
            meta.missing_routers(tx, all_descriptors.iter().copied(), None)
        })
        .unwrap()
        .unwrap();
        assert!(missing_routers.is_empty());
    }
    /// Tests whether or not the missing descriptors are actually random.
    #[test]
    fn missing_descriptors_are_random() {
        let pool = testdata2::test_db();
        rw_tx(&pool, |tx| {
            tx.execute(sql!("DELETE FROM descriptor"), ()).unwrap();
            let meta = ConsensusMeta::<Plain>::query(tx).unwrap()[0];
            // Ensure there are more than 1 missing descriptors now.
            let n = meta.missing_routers(tx, iter::empty(), None).unwrap().len();
            assert!(n > 1);
            let mut prev = HashSet::new();
            let mut randomness_works = false;
            for _ in 0..100 {
                let cur = meta.missing_routers(tx, iter::empty(), Some(1)).unwrap();
                if prev != cur {
                    randomness_works = true;
                    break;
                }
                prev = cur;
            }
            assert!(randomness_works);
        })
        .unwrap();
    }
    /// Tests whether the RANDOM() calls are executed before the ordering so
    /// that we have stability.
    #[test]
    fn missing_routers_monotonically_increasing() {
        let pool = testdata2::test_db();
        rw_tx(&pool, |tx| {
            let meta = ConsensusMeta::<Plain>::query(tx).unwrap();
            tx.execute(sql!("DELETE FROM descriptor"), ()).unwrap();
            let mut stmt = tx
                .prepare_cached(ConsensusMeta::<Plain>::MISSING_ROUTERS_QUERY)
                .unwrap();
            let empty = Rc::new(Vec::<Value>::new());
            let res = stmt
                .query_map(params![meta[0].docid, empty, -1], |row| {
                    row.get::<_, i64>(1)
                })
                .unwrap()
                .collect::<Result<Vec<_>, _>>()
                .unwrap();
            // Ensure there is more than 1 missing descriptor because we can
            // only test ordering if n >= 2.
            assert!(res.len() > 1);
            // Verify that the calls are monotonically increasing.
            assert!(res.is_sorted());
        })
        .unwrap();
    }
    /// Tests whether the missing extra-info documents are computed properly.
    // TODO DIRMIRROR: Expand on this once we have proper extra-info support.
    #[test]
    fn missing_extra_infos() {
        let pool = testdata2::test_db();
        let meta = read_tx(&pool, ConsensusMeta::<Plain>::query)
            .unwrap()
            .unwrap()[0];
        // Ensure that the returned consensus matches the one from testdata2.
        assert_eq!(
            meta.docid,
            DocumentId::digest(testdata2::current_consensus_ns().2.as_bytes())
        );
        // We should have no missing extra-infos.
        let missing_extras = read_tx(&pool, |tx| meta.missing_extras(tx, iter::empty(), None))
            .unwrap()
            .unwrap();
        assert!(missing_extras.is_empty());
        // TODO DIRMIRROR: Once we have support for extra-info's, add a test
        // for this here.  Right now, testing this is pretty useless as we
        // cannot add it nicely to the testdata2 module if there isn't even
        // an ExtraInfo struct from tor-netdoc.
    }
    /// Tests whether the missing micro descriptor queue is computed properly.
    ///
    /// For this, we remove existing micro descriptors from the database and
    /// see whether they are determined as missing properly.
    #[test]
    fn missing_micro_descriptors() {
        let pool = testdata2::test_db();
        let meta = read_tx(&pool, ConsensusMeta::<Md>::query).unwrap().unwrap()[0];
        // Ensure that the returned consensus matches the one from testdata2.
        assert_eq!(
            meta.docid,
            DocumentId::digest(testdata2::current_consensus_md().2.as_bytes())
        );
        // Delete a single router descriptor, so we can determine a missing one
        // using it.
        let removed_descriptor = *testdata2::current_consensus_md().0.routers[0].doc_digest();
        let removed_descriptor = Sha256::from(removed_descriptor);
        pool.get()
            .unwrap()
            .execute(
                sql!(
                    "
                    DELETE FROM descriptor
                    WHERE sha2 = ?1
                    "
                ),
                params![removed_descriptor],
            )
            .unwrap();
        // Only one should be returned.
        let missing_micros = read_tx(&pool, |tx| meta.missing_micros(tx, iter::empty(), None))
            .unwrap()
            .unwrap();
        assert_eq!(missing_micros, HashSet::from([removed_descriptor]));
        // If we delete all micro descriptors we have, we should get all.
        rw_tx(&pool, |tx| {
            tx.execute(sql!("DELETE FROM descriptor"), params![])
        })
        .unwrap()
        .unwrap();
        // Now all should be returned; we verify this by checking that the
        // result is present in all_descriptors, which is a superset.
        let missing_micros = read_tx(&pool, |tx| meta.missing_micros(tx, iter::empty(), None))
            .unwrap()
            .unwrap();
        // This is a superset of missing_micros because it includes micro
        // descriptors that are not a part of the current consensus.
        let all_descriptors = testdata2::current_micro_descs()
            .iter()
            .map(|x| Sha256::digest(x.1.as_bytes()))
            .collect::<HashSet<_>>();
        assert!(
            missing_micros
                .iter()
                .all(|sha2| all_descriptors.contains(sha2))
        );
    }
}