voa_openpgp/

lookup.rs

//! Helper functionality to find appropriate [`SignatureVerifier`]s from a set of [`OpenpgpCert`].

use std::{
    collections::{BTreeMap, HashMap},
    ops::Add,
    time::SystemTime,
};

use log::{debug, info, trace, warn};
use pgp::{
    packet::{Signature, SignatureType, UserId},
    types::{Fingerprint, KeyDetails, KeyId, Tag},
};
use rpgpie::{certificate::SignatureVerifier, signature::signature_acceptable};

use crate::{OpenpgpCert, OpenpgpSignature};

/// Helper to efficiently look up [`OpenpgpCert`]s by [`Fingerprint`] or by [`KeyId`].
///
/// This struct keeps a lookup entry by the [`Fingerprint`] and [`KeyId`] of *each* component key
/// of each indexed [`OpenpgpCert`] (including ones that don't have the data signing key flag, and
/// invalid ones).
///
/// As a consequence, the resulting lookup configuration doesn't make any semantical guarantees
/// about lookup results, beyond that one component key bears the queried [`Fingerprint`] or
/// [`KeyId`].
///
/// Downstream users must thus perform additional checks if a certificate is actually
/// appropriate to use for a particular purpose.
#[derive(Debug)]
struct CertLookup<'a> {
    by_fingerprint: HashMap<Fingerprint, Vec<&'a OpenpgpCert>>,
    by_key_id: HashMap<KeyId, Vec<&'a OpenpgpCert>>,
}

impl<'a> From<&'a [OpenpgpCert]> for CertLookup<'a> {
    fn from(certs: &'a [OpenpgpCert]) -> Self {
        let mut lookup = Self {
            by_fingerprint: HashMap::new(),
            by_key_id: HashMap::new(),
        };

        // Insert lookup entries for the Fingerprint and KeyId of each component key of each cert
        certs.iter().for_each(|cert| lookup.insert_cert(cert));

        lookup
    }
}

impl<'a> CertLookup<'a> {
    /// Inserts an entry for the [`Fingerprint`] and [`KeyId`] of _each_ component key of an
    /// [`OpenpgpCert`].
    fn insert_cert(&mut self, cert: &'a OpenpgpCert) {
        let primary = &cert.certificate.as_signed_public_key().primary_key;
        self.insert_by_fingerprint(cert, primary.fingerprint());
        self.insert_by_key_id(cert, primary.key_id());

        for subkey in &cert.certificate.as_signed_public_key().public_subkeys {
            self.insert_by_fingerprint(cert, subkey.fingerprint());
            self.insert_by_key_id(cert, subkey.key_id());
        }
    }

    /// Inserts a lookup entry for an [`OpenpgpCert`] and an accompanying [`Fingerprint`].
    fn insert_by_fingerprint(&mut self, cert: &'a OpenpgpCert, fingerprint: Fingerprint) {
        self.by_fingerprint
            .entry(fingerprint)
            .or_default()
            .push(cert);
    }

    /// Inserts a lookup entry for an [`OpenpgpCert`] and an accompanying [`KeyId`].
    fn insert_by_key_id(&mut self, cert: &'a OpenpgpCert, key_id: KeyId) {
        self.by_key_id.entry(key_id).or_default().push(cert);
    }

    /// Returns a list of [`OpenpgpCert`]s that match a provided [`Fingerprint`].
    pub(crate) fn by_fingerprint(&self, fingerprint: &Fingerprint) -> &[&'a OpenpgpCert] {
        self.by_fingerprint
            .get(fingerprint)
            .map_or(&[], |cert| cert)
    }

    /// Returns a list of [`OpenpgpCert`]s that match a provided [`KeyId`].
    pub(crate) fn by_key_id(&self, key_id: &KeyId) -> &[&'a OpenpgpCert] {
        self.by_key_id.get(key_id).map_or(&[], |cert| cert)
    }
}

/// An abstraction over a set of [`OpenpgpCert`], to facilitate signature verification.
///
/// It can efficiently look up component keys that are appropriate for attempting validation of
/// specific signatures.
#[derive(Debug)]
pub struct VerifierLookup<'a> {
    index: CertLookup<'a>,
}

impl<'a> FromIterator<&'a OpenpgpCert> for VerifierLookup<'a> {
    fn from_iter<T: IntoIterator<Item = &'a OpenpgpCert>>(iter: T) -> Self {
        let mut index = CertLookup {
            by_fingerprint: HashMap::new(),
            by_key_id: HashMap::new(),
        };

        for c in iter.into_iter() {
            index.insert_cert(c);
        }

        Self { index }
    }
}

impl<'a> VerifierLookup<'a> {
    /// The signature types that we consider meaningful for certifications on identities.
    ///
    /// Note that the set excludes `SignatureType::CertPersona` which is specified to not carry any
    /// meaningful information.
    const CERTIFICATION_SIGNATURE_TYPES: &'static [SignatureType] = &[
        SignatureType::CertPositive,
        SignatureType::CertGeneric,
        SignatureType::CertCasual,
        SignatureType::CertRevocation,
    ];

    /// Creates a new [`VerifierLookup`] from a list of [`OpenpgpCert`]s.
    pub fn new(certs: &'a [OpenpgpCert]) -> Self {
        Self {
            index: certs.into(),
        }
    }

    /// Returns a list of [`SignatureVerifier`] and [`OpenpgpCert`] tuples, that are reasonable
    /// candidates for attempting to verify an [`OpenpgpSignature`].
    ///
    /// Looks up all component keys that match the [IssuerFingerprint] and/or [Issuer] subpackets in
    /// `signature`, and returns them as [`SignatureVerifier`] objects.
    /// For informational purposes, each [`SignatureVerifier`] is accompanied by a reference to the
    /// [`OpenpgpCert`] that contains it.
    ///
    /// Callers of this function will usually want to validate `signature` with the returned
    /// [`SignatureVerifier`]s.
    ///
    /// [IssuerFingerprint]: https://www.rfc-editor.org/rfc/rfc9580.html#issuer-fingerprint-subpacket
    /// [Issuer]: https://www.rfc-editor.org/rfc/rfc9580.html#name-issuer-key-id
    pub(crate) fn get_matching_verifiers(
        &self,
        signature: &OpenpgpSignature,
    ) -> Vec<(SignatureVerifier, &'a OpenpgpCert)> {
        let Some(created) = signature.creation_time() else {
            warn!("Skipping signature without creation time");
            return Vec::new();
        };

        let mut verifiers = Vec::new();

        // Iterate over all candidate certificates that contain component keys that match
        // an issuer identity that `signature` lists.
        // (This also performs rpgpie policy checks on the signature.)
        for cert in self.candidate_certs(&signature.detached.signature) {
            // OpenPGP semantics: enumerate all component keys that are valid for data signing
            // purposes at the creation time of this signature.
            for verifier in cert
                .certificate
                .valid_signing_capable_component_keys_at(&created.into())
            {
                // This set of valid component keys may contain some that do not match the issuer
                // hints in the signature.
                // Here, we filter the valid verifiers by the signature's issuer hints again.
                if Self::matches_issuer(signature, &verifier) {
                    verifiers.push((verifier, cert))
                }
            }
        }

        debug!(
            "get_matching_verifiers for {:?}: {}",
            &signature.source,
            verifiers
                .iter()
                .map(|(verifier, cert)| format!(
                    "{}/{}",
                    verifier.as_componentkey().fingerprint(),
                    cert.certificate.fingerprint()
                ))
                .collect::<Vec<_>>()
                .join(", ")
        );

        verifiers
    }

    /// Returns a list of [`OpenpgpCert`]s that match either the
    /// [IssuerFingerprint] or [IssuerKeyId] subpackets in a [`Signature`].
    ///
    /// All component keys of each [`OpenpgpCert`] are considered when matching against the
    /// [IssuerFingerprint] and [IssuerKeyId] subpackets of the `signature`. The returned
    /// [`OpenpgpCert`]s are deduplicated by their primary [`Fingerprint`].
    ///
    /// ## Note
    ///
    /// The lookup does not enforce any semantics constraints. It does not guarantee
    /// validity of certificates or component keys for any particular purpose. Semantics checks
    /// must be performed separately, after this lookup.
    ///
    /// [IssuerFingerprint]: https://www.rfc-editor.org/rfc/rfc9580.html#issuer-fingerprint-subpacket
    /// [IssuerKeyId]: https://www.rfc-editor.org/rfc/rfc9580.html#name-issuer-key-id
    fn candidate_certs(&self, signature: &Signature) -> Vec<&'a OpenpgpCert> {
        // We use a map with the key as a string-representation of the fingerprint.
        // The fingerprint is used both to ensure unique results and a stable ordering.
        let mut candidates = BTreeMap::new();

        // Try to find the signer by Issuer Fingerprint subpacket
        for fp in signature.issuer_fingerprint() {
            for &cert in self.index.by_fingerprint(fp) {
                candidates.insert(cert.certificate.fingerprint().to_string(), cert);
            }
        }

        // Try to find the signer by Issuer KeyId subpacket
        for key_id in signature.issuer() {
            for &cert in self.index.by_key_id(key_id) {
                candidates.insert(cert.certificate.fingerprint().to_string(), cert);
            }
        }

        trace!(
            "candidate_certs for {signature:#?}: {:?}",
            candidates
                .keys()
                .map(ToString::to_string)
                .collect::<Vec<_>>()
                .join(", ")
        );

        candidates.into_values().collect()
    }

    /// Checks if a [`SignatureVerifier`] matches any [IssuerFingerprint] or [Issuer] subpacket in
    /// an [`OpenpgpSignature`].
    ///
    /// [IssuerFingerprint]: https://www.rfc-editor.org/rfc/rfc9580.html#issuer-fingerprint-subpacket
    /// [Issuer]: https://www.rfc-editor.org/rfc/rfc9580.html#name-issuer-key-id
    fn matches_issuer(signature: &OpenpgpSignature, verifier: &SignatureVerifier) -> bool {
        let issuer_fingerprint = signature.detached.signature.issuer_fingerprint();
        let issuer_key_id = signature.detached.signature.issuer();

        issuer_fingerprint.contains(&&verifier.as_componentkey().fingerprint())
            || issuer_key_id.contains(&&verifier.as_componentkey().key_id())
    }

    /// Returns pairs of [`OpenpgpCert`] and [`Signature`]s for third-party UserId certifications.
    ///
    /// Filters `signatures` (a slice of third-party certifications over `target` and
    /// `target_user`) by policy, as well as temporal and cryptographic validity (at
    /// `reference_time`).
    /// The validated signatures are grouped by signer certificate.
    ///
    /// ## Notes
    ///
    /// - A certifying signature must pass [`rpgpie`]'s policy checks (i.e. cryptographic mechanisms
    ///   that are considered weak at signature creation time are rejected).
    /// - If a certifying signature has a "signature expiration time" that is after the reference
    ///   time, that certifying signature is ignored (except for certification revocation
    ///   signatures, which may not expire).
    /// - The certifying signature *may* be younger than the data signature that is authenticated.
    /// - The certifying certificate *may* be younger than the data signature that is authenticated.
    pub fn valid_userid_certifications(
        &self,
        signatures: &[&'a Signature],
        target: &OpenpgpCert,
        target_user: &UserId,
        reference_time: SystemTime,
    ) -> Vec<(&'a OpenpgpCert, Vec<&'a Signature>)> {
        let mut map = HashMap::new();

        for &sig in signatures {
            if let Some(typ) = sig.typ()
                && !Self::CERTIFICATION_SIGNATURE_TYPES.contains(&typ)
            {
                continue;
            }

            // Policy check - is the signature using acceptable algorithms?
            if !signature_acceptable(sig) {
                continue;
            }

            let Some(sig_created) = sig.created() else {
                continue;
            };

            if sig.typ() != Some(SignatureType::CertRevocation) {
                // Filter out certifications that expire before reference_time.
                // (However, we don't accept signature expiration in revocations)
                if let Some(exp) = sig.signature_expiration_time()
                    && !exp.is_zero()
                {
                    let expires = sig_created.add(*exp);

                    if SystemTime::from(expires) <= reference_time {
                        trace!("skipping expired sig {:?}", sig);
                        continue;
                    }
                }
            }

            // Find the signer that has made this signature, if any.
            // This includes a check for cryptographic validity of the signature.
            let Some(signer) = self.lookup_third_party_certifier(sig, target, target_user) else {
                // We found no signer, go to the next signature
                continue;
            };

            trace!("  found signer: {:?}", signer.certificate.fingerprint());

            // Ignore certification if signer is not valid at its creation time.
            //
            // (However, we do allow certifiers that were not valid "yet" at the *data signature*'s
            // creation time.)
            if !matches!(signer.certificate.primary_valid_at(sig_created), Ok(true)) {
                continue;
            }

            // Check that the signer's primary has key flag `0x01`
            // that allows issuing third-party certifications
            if let Some(self_sig) = signer
                .certificate
                .active_certificate_self_signature_at(sig_created)
            {
                if !self_sig.key_flags().certify() {
                    info!(
                        "  skipping signer {:?} because it's missing the 'certifications' key flag",
                        signer.certificate.fingerprint()
                    );
                    continue;
                }
            } else {
                info!(
                    "  skipping signer {:?} because we found no active self-signature",
                    signer.certificate.fingerprint()
                );
                continue;
            }

            // Store this certifying signature in map, using the signer certificate's fingerprint
            // as the map key
            let (_, sigs) = map
                .entry(signer.certificate.fingerprint())
                .or_insert((signer, Vec::new()));
            sigs.push(sig);
        }

        map.into_values().collect()
    }

    /// Returns a matching [`OpenpgpCert`] for a third-party certification over a [`UserId`].
    ///
    /// The considered certificate must be cryptographically valid and must have issued `sig` as a
    /// (third-party) certification over `target` and `target_user`.
    ///
    /// If a signer is found, a reference to it is returned, otherwise [`None`].
    fn lookup_third_party_certifier(
        &self,
        sig: &Signature,
        target: &OpenpgpCert,
        target_user: &UserId,
    ) -> Option<&'a OpenpgpCert> {
        self.candidate_certs(sig).into_iter().find(|&candidate| {
            sig.verify_third_party_certification(
                &target.certificate.as_signed_public_key().primary_key,
                &candidate.certificate.as_signed_public_key().primary_key,
                Tag::UserId,
                target_user,
            )
            .is_ok()
        })
    }
}