voa_openpgp/trust/

anchor.rs

//! Implementation of the basic "trust anchor" verification method.

use std::{collections::HashSet, time::SystemTime};

use log::{debug, error, trace, warn};
use pgp::packet::SignatureType;
use voa_config::openpgp::{DomainName, NumCertifications, TrustAnchorMode};

use crate::{
    OpenpgpCert,
    lookup::VerifierLookup,
    trust::{
        model::ArtifactVerifierValidity,
        util::{find_signed_user, fingerprint_matches, identity_bound_at, user_id_matches_any},
    },
};

/// Signature verification object that uses a basic trust anchor-based trust model.
#[derive(Debug)]
pub(crate) struct TrustAnchorModel<'a> {
    /// Lookup table for the set of trust anchors
    pub(crate) trust_anchor_lookup: VerifierLookup<'a>,

    /// Artifact verifiers are only considered if they contain a validly self-bound identity in
    /// one of these domain names, at the reference time.
    /// In addition, such an identity must be certified by the required number of trust
    /// anchors.
    pub(crate) identity_domain_names: &'a HashSet<DomainName>,

    /// Number of valid certifications an artifact verifiers needs to be used for signature
    /// verification
    pub(crate) required_certifications: NumCertifications,

    /// Lookup table for the set of artifact verifiers
    pub(crate) artifact_verifier_lookup: VerifierLookup<'a>,
}

impl<'a> TrustAnchorModel<'a> {
    /// Creates a new [`TrustAnchorModel`].
    ///
    /// Applies filters from `config` on `trust_anchors`.
    /// Filters for `artifact_verifiers` are only applied during signature verification.
    pub(crate) fn new(
        config: &'a TrustAnchorMode,
        artifact_verifiers: &'a [OpenpgpCert],
        trust_anchors: &'a [OpenpgpCert],
    ) -> Self {
        // NOTE: We don't filter certificates for validity here.
        // (Both for certifications and data signature verification, we care about certificate
        // validity at signature creation time.)

        let anchor_fingerprints = config.trust_anchor_fingerprint_matches();

        // Filter trust anchors by fingerprint
        let trust_anchor_lookup = VerifierLookup::from_iter(
            trust_anchors
                .iter()
                .filter(|cert| fingerprint_matches(cert, anchor_fingerprints)),
        );

        TrustAnchorModel {
            trust_anchor_lookup,
            identity_domain_names: config.artifact_verifier_identity_domain_matches(),
            required_certifications: config.required_certifications(),
            artifact_verifier_lookup: VerifierLookup::new(artifact_verifiers),
        }
    }
}

impl ArtifactVerifierValidity for TrustAnchorModel<'_> {
    /// Check if `cert` is valid under the basic "trust anchor" model.
    ///
    /// This checks for an acceptable identity, which must:
    /// - Be validly self-bound.
    /// - Match `identity_domain_names`, if that set has any entries.
    /// - Have the required number of certifications from valid trust anchors.
    ///
    /// Note that this function considers third-party certification revocations permanent.
    /// If a certifier revokes his third-party certification on a specific user id, this function
    /// will not accept any future certifications from that third party over the same user id.
    fn check(&self, cert: &OpenpgpCert, reference_time: SystemTime) -> bool {
        // Iterate over all user ids, and the third-party signatures over them
        for (uid, sigs) in cert.certificate.user_id_third_party_certifications() {
            // Check that the UserId matches a domain in `identity_domain_names`

            // Transform the user id into a UTF-8 string
            let Ok(userid) = String::from_utf8(uid.id().to_vec()) else {
                warn!(
                    "Invalid non-UTF8 User ID {} in certificate {}",
                    String::from_utf8_lossy(uid.id()),
                    cert.certificate.fingerprint()
                );
                continue;
            };

            // Check if userid is acceptable, based on `identity_domain_names`
            if !user_id_matches_any(&userid, self.identity_domain_names) {
                trace!(
                    "Skipping User ID {userid}, because it doesn't match one of {}",
                    self.identity_domain_names
                        .iter()
                        .map(ToString::to_string)
                        .collect::<Vec<_>>()
                        .join(", ")
                );
                continue;
            }

            debug!(
                "The User ID {userid} has {} third-party certifications in total.",
                sigs.len()
            );

            // Get self-signatures for this user id
            if let Some(su) = find_signed_user(cert, &uid) {
                // Check that the identity is validly self-bound at `reference_time`
                if !identity_bound_at(cert, &su.signatures, reference_time) {
                    trace!(
                        "Skipping User ID {userid}, because it is not validly self-bound at {reference_time:?}",
                    );
                    continue;
                }
            } else {
                error!("Couldn't find SignedUser for {userid}. This should never happen.");
                continue;
            }

            // Third-party certifications by valid certifiers (from our trust anchor set).
            // These certifications are cryptographically valid, and passed rpgpie's policy and
            // temporal checks.
            let valid_certifications = self.trust_anchor_lookup.valid_userid_certifications(
                &sigs,
                cert,
                &uid,
                reference_time,
            );

            debug!(
                "The UserID {userid} has {} valid third-party certifications.",
                valid_certifications.len()
            );

            // Produce a list of confirming certifiers.
            // If a signer has revoked their certification, or not issued a certification that is
            // valid at the reference time, then that signer is not added to this list.
            let active_certifiers = {
                let mut active = Vec::new();

                for (signer, sigs) in valid_certifications {
                    // We consider this signer's certification revoked if any revocation exists
                    // (at reference time)
                    if sigs
                        .iter()
                        .any(|s| s.typ() == Some(SignatureType::CertRevocation))
                    {
                        trace!(
                            "Signer {} has revoked their certification of {userid}",
                            signer.certificate.fingerprint()
                        );
                        continue;
                    }

                    // If there is any certifying signature from this signer we consider that signer
                    // an active certifier of this user id.
                    if !sigs.is_empty() {
                        active.push(signer);
                    }
                }

                active
            };

            if !active_certifiers.is_empty() {
                trace!(
                    "Found third-party certifications on UserID {userid} from {}",
                    active_certifiers
                        .iter()
                        .map(|cert| cert.certificate.fingerprint())
                        .map(ToString::to_string)
                        .collect::<Vec<_>>()
                        .join(", ")
                );
            }

            // We consider each `active` certifier as one unit of trust in `target`.
            // If we have less third-party certifications than
            // `required_certifications`, we continue to the next UserID.
            if active_certifiers.len() < self.required_certifications.get().get() {
                debug!(
                    "Too few certifications {} on User ID {userid} of {} at {:?} ",
                    active_certifiers.len(),
                    cert.certificate.fingerprint(),
                    reference_time.duration_since(std::time::UNIX_EPOCH),
                );
                continue;
            }

            // We have found the required number of third-party certifications (or more)
            return true;
        }

        false
    }

    fn lookup(&self) -> &VerifierLookup<'_> {
        &self.artifact_verifier_lookup
    }
}