wot-network-rpgpie 0.0.2

//! Helper module for third party certifications
//!
//! TODO: Can some of this logic be upstreamed to rpgpie?

use std::{collections::HashMap, ops::Add};

use chrono::{DateTime, Utc};
use pgp::{
    packet::{PublicKey, Signature, SignatureType, UserId},
    types::{Fingerprint, KeyId, Tag},
};
use rpgpie::certificate::Checked;
use wot_network::{Certificate, Network};

use crate::util::{pick_latest, regexes_from, to_certificate, to_identity, user_id_valid};

/// `ThirdPartyLookup` is used to find the issuer of third party signatures (or to determine
/// that the signer is unknown in the scope of this network).
///
/// Internally, `ThirdPartyLookup` acts as a lookup table for certificates by [`Fingerprint`] or
/// [`KeyId`].
pub(crate) struct ThirdPartyLookup<'a> {
    /// A map from (primary) `Fingerprint` to `Checked` certificates
    by_fp: HashMap<Fingerprint, &'a Checked>,

    /// A list of `Checked` per `KeyId` for efficient lookup,
    /// (while robustly handling potential duplicate `KeyId`s).
    by_keyid: HashMap<KeyId, Vec<&'a Checked>>,
}

impl<'a> ThirdPartyLookup<'a> {
    /// Initialize a lookup struct for all `checked`.
    ///
    /// This allows efficient lookup of certificates by both `Fingerprint` and `KeyId`.
    pub(crate) fn new(checked: &'a Vec<Checked>) -> Self {
        let mut by_fp = HashMap::new();
        let mut by_keyid = HashMap::new();

        for c in checked {
            // store for lookup by fingerprint
            by_fp.insert(c.fingerprint().clone(), c);

            // store for lookup by keyid
            let values = by_keyid.entry(c.key_id()).or_insert(vec![]);
            values.push(c);
        }

        ThirdPartyLookup { by_fp, by_keyid }
    }

    /// Find a cryptographically valid signer for `sig` as a third party certification over
    /// `signee` and `user_id`.
    ///
    /// If a signer is found, a reference to it is returned, otherwise None.
    pub(crate) fn find_third_party_userid_signer(
        &self,
        sig: &Signature,
        signee: &PublicKey,
        user_id: &UserId,
    ) -> Option<&Checked> {
        // Helper closure: verify signature as user id certification
        let verify = |signee: &PublicKey, signer: &PublicKey| {
            sig.verify_third_party_certification(signee, signer, Tag::UserId, user_id)
                .is_ok()
        };

        self.find_third_party_signer(sig, signee, &verify)
    }

    /// Find a cryptographically valid signer for `sig` as a third party direct key signature over
    /// `signee`.
    ///
    /// If a signer is found, a reference to it is returned, otherwise None.
    ///
    /// NOTE: This function currently doesn't find signers, waiting for an rPGP release >0.17!
    pub(crate) fn find_third_party_dks_signer(
        &self,
        sig: &Signature,
        signee: &PublicKey,
    ) -> Option<&Checked> {
        // Helper closure: verify signature as third party direct certification
        let verify = |_signee: &PublicKey, _signer: &PublicKey| {
            false

            // FIXME: wait for rPGP release >0.17
            // sig.verify_key_third_party(signee, signer).is_ok()
        };

        self.find_third_party_signer(sig, signee, &verify)
    }

    /// Third-party signer lookup (used for both user id certifications and direct key signatures)
    fn find_third_party_signer(
        &self,
        sig: &Signature,
        signee: &PublicKey,
        verify: &dyn Fn(&PublicKey, &PublicKey) -> bool,
    ) -> Option<&Checked> {
        // Possible optimization:
        // Remember (e.g. by HashSet of fingerprint), which possible signers we already tried and
        // didn't get a positive check result for, and don't try those again.
        //
        // This would save time if there are many issuer/issuer fingerprint subpackets that don't
        // verify as ok.

        // Try to find the signer by Issuer Fingerprint subpacket
        for fp in sig.issuer_fingerprint() {
            if let Some(c) = self.by_fp.get(fp) {
                if verify(signee, &c.as_signed_public_key().primary_key) {
                    return Some(c);
                }
            }
        }

        // Try to find the signer by Issuer KeyId subpacket
        for keyid in sig.issuer() {
            for &c in self.by_keyid.get(keyid).unwrap_or(&vec![]) {
                if verify(signee, &c.as_signed_public_key().primary_key) {
                    return Some(c);
                }
            }
        }

        None // We found no cryptographically valid signer key
    }
}

/// Process all third-party signatures on `target` over the user id `user`
pub(crate) fn handle_userid_third_party(
    net: &mut Network,
    tplookup: &ThirdPartyLookup,
    target: &Checked,
    user: &UserId,
    sigs: &Vec<&Signature>,
    reference_time: &DateTime<Utc>,
) {
    // Search for a user id with the same content in the target certificate
    if let Some(target_user) = target.user_ids().iter().find(|su| &su.id == user) {
        // If the target certificate considers the user id invalid, we skip it
        // (and thus reject all third-party certifications over that user id)
        if !user_id_valid(target_user, reference_time) {
            return;
        }
    }

    let Some(id) = to_identity(user) else {
        // Reject/skip non-utf8 User IDs
        return;
    };

    if !sigs.is_empty() {
        log::debug!("{} is certified by:", id);
    }

    // Signatures filtered by validity (including cryptographic) and grouped by signer
    let valid_third_party =
        valid_userid_third_party_sigs(tplookup, sigs, target, user, reference_time);

    // Iterate over all signers in valid_third_party
    for (signer, sigs) in valid_third_party {
        // We consider this signer's certification revoked if a revocation exists at reference time
        if sigs
            .iter()
            .any(|s| s.typ() == Some(SignatureType::CertRevocation))
        {
            continue;
        }

        // Make a binding and a delegation edge based on the latest certification by this signer
        let latest = pick_latest(&sigs);

        net.add_binding(signer.clone(), to_certificate(target), id.clone());
        add_delegation(net, latest, signer, target);
    }
}

pub(crate) fn handle_dks_third_party(
    net: &mut Network,
    tplookup: &ThirdPartyLookup,
    target: &Checked,
    dks_third: &Vec<&Signature>,
    reference_time: &DateTime<Utc>,
) {
    if !dks_third.is_empty() {
        log::trace!("found {} dks", dks_third.len());
    }

    let valid_dks = valid_dks_third_party_sigs(tplookup, dks_third, target, reference_time);

    for (signer, sigs) in valid_dks {
        // Third party direct key signatures have no revocation mechanism.
        // We always handle the latest valid direct signature, per signer.
        // (If a signer wants to remove e.g. a trust signature, they will make a new
        // third-party direct signature, without that subpacket.)
        let latest = pick_latest(&sigs);
        add_delegation(net, latest, signer, target);
    }
}

/// Filter the third party signatures in `third` by policy, as well as temporal and cryptographic
/// validity, and group them by signer certificate.
pub fn valid_userid_third_party_sigs<'a>(
    tplookup: &'a ThirdPartyLookup<'_>,
    signatures: &[&'a Signature],
    target: &Checked,
    target_user: &UserId,
    reference_time: &DateTime<Utc>,
) -> HashMap<Certificate, Vec<&'a Signature>> {
    // We only handle (some) certification signature types
    const FILTER: &[Option<SignatureType>] = &[
        Some(SignatureType::CertPositive),
        Some(SignatureType::CertGeneric),
        Some(SignatureType::CertCasual),
        Some(SignatureType::CertRevocation),
    ];

    let lookup = |sig| {
        tplookup.find_third_party_userid_signer(
            sig,
            &target.as_signed_public_key().primary_key,
            target_user,
        )
    };

    collect_third_party(signatures, reference_time, Box::new(lookup), FILTER)
}

fn valid_dks_third_party_sigs<'a>(
    tplookup: &'a ThirdPartyLookup,
    signatures: &[&'a Signature],
    target: &Checked,
    reference_time: &DateTime<Utc>,
) -> HashMap<Certificate, Vec<&'a Signature>> {
    // We only handle third-party direct key signatures (and their revocations) here
    const FILTER: &[Option<SignatureType>] = &[
        Some(SignatureType::Key),
        Some(SignatureType::CertRevocation),
    ];

    let lookup =
        |sig| tplookup.find_third_party_dks_signer(sig, &target.as_signed_public_key().primary_key);

    collect_third_party(signatures, reference_time, Box::new(lookup), FILTER)
}

/// Collect all valid third party certifications, for further processing, grouped by signer.
fn collect_third_party<'a>(
    signatures: &[&'a Signature],
    reference_time: &DateTime<Utc>,
    lookup: Box<impl Fn(&'a Signature) -> Option<&'a Checked>>,
    sigtype_filter: &[Option<SignatureType>],
) -> HashMap<Certificate, Vec<&'a Signature>> {
    let mut map = HashMap::new();

    for &sig in signatures {
        if !sigtype_filter.contains(&sig.typ()) {
            continue;
        }

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

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

        // Filter out if signature creation time is after reference_time
        if sig_created > reference_time {
            log::trace!("skipping signature from the future {:?}", sig);
            continue;
        }

        // Filter out if signature has an expiration time that is before reference_time
        if let Some(exp) = sig.signature_expiration_time() {
            if exp.num_seconds() != 0 {
                let expires = sig_created.add(*exp);

                if expires <= *reference_time {
                    log::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) = lookup(sig) else {
            // We found no signer, go to the next signature
            continue;
        };

        log::debug!("  found signer: {:?}", signer.fingerprint());

        // TODO: check that the signer's primary has key flag `0x01`

        // Ignore signature if signer is invalid at its creation time
        if !signer.primary_valid_at(sig_created).unwrap_or(false) {
            continue;
        }

        // Ignore signature if signer is invalid at reference time.
        // (NOTE: this disagrees with s-w cert_expired, cert_revoked_soft)
        if !signer.primary_valid_at(reference_time).unwrap_or(false) {
            continue;
        }

        // Store this signature in map
        let signer_cert = to_certificate(signer);
        let sigs: &mut Vec<&Signature> = map.entry(signer_cert).or_default();
        sigs.push(sig);
    }

    map
}

/// Add delegation edge to `net,` based on the "trust signature" and "regex" subpackets of `sig`
fn add_delegation(net: &mut Network, sig: &Signature, signer: Certificate, target: &Checked) {
    // Is this certification a "trust signature" (with depth > 0)?
    if let Some((depth, amount)) = sig.trust_signature() {
        if depth > 0 {
            // We model the "trust signature" aspect of a certifying signature with
            // a separate delegation edge in the `Network`.

            // TODO: map the amount based on a user-provided configuration?
            //  -> "how many partials are needed for one full?"

            let regexes = regexes_from(sig);

            net.add_delegation(
                signer,
                to_certificate(target),
                amount,
                depth.into(),
                regexes,
            );
        }
    }
}