use std::time::{Duration, SystemTime, UNIX_EPOCH};
use x509_cert::Certificate;
use x509_cert::der::Decode;
use x509_cert::der::oid::{AssociatedOid, ObjectIdentifier};
use x509_cert::ext::pkix::{BasicConstraints, KeyUsage};
use crate::attestation::{cert_extension, parse_cert, tbs_der, verify_with_key};
use crate::types::*;
const OID_ECDSA_SHA256: ObjectIdentifier = ObjectIdentifier::new_unwrap("1.2.840.10045.4.3.2");
const OID_ECDSA_SHA384: ObjectIdentifier = ObjectIdentifier::new_unwrap("1.2.840.10045.4.3.3");
const OID_RSA_SHA256: ObjectIdentifier = ObjectIdentifier::new_unwrap("1.2.840.113549.1.1.11");
const OID_RSA_SHA384: ObjectIdentifier = ObjectIdentifier::new_unwrap("1.2.840.113549.1.1.12");
const OID_ED25519: ObjectIdentifier = ObjectIdentifier::new_unwrap("1.3.101.112");
fn cose_alg_for_signature_oid(oid: &ObjectIdentifier) -> Result<i32> {
match *oid {
OID_ECDSA_SHA256 => Ok(ALG_ES256),
OID_ECDSA_SHA384 => Ok(ALG_ES384),
OID_RSA_SHA256 => Ok(ALG_RS256),
OID_RSA_SHA384 => Ok(ALG_RS384),
OID_ED25519 => Ok(ALG_EDDSA),
_ => Err(PasskiError::InvalidCertificateChain(format!(
"unsupported signature algorithm {}",
oid
))),
}
}
fn spki_bytes(cert: &Certificate) -> Result<&[u8]> {
cert.tbs_certificate()
.subject_public_key_info()
.subject_public_key
.as_bytes()
.ok_or_else(|| {
PasskiError::InvalidCertificateChain(
"issuer public key is not byte-aligned".to_string(),
)
})
}
fn check_issued_by(der: &[u8], cert: &Certificate, issuer: &Certificate) -> Result<()> {
if cert.tbs_certificate().issuer() != issuer.tbs_certificate().subject() {
return Err(PasskiError::InvalidCertificateChain(
"issuer name does not match the subject of the next certificate".to_string(),
));
}
let alg = cose_alg_for_signature_oid(&cert.signature_algorithm().oid)?;
let signature = cert.signature().as_bytes().ok_or_else(|| {
PasskiError::InvalidCertificateChain("signature is not byte-aligned".to_string())
})?;
verify_with_key(alg, spki_bytes(issuer)?, tbs_der(der)?, signature)
.map_err(|_| PasskiError::InvalidCertificateChain("broken signature link".to_string()))
}
fn check_validity(cert: &Certificate, now: Duration) -> Result<()> {
let validity = cert.tbs_certificate().validity();
if now < validity.not_before.to_unix_duration() {
return Err(PasskiError::InvalidCertificateChain(
"certificate is not valid yet".to_string(),
));
}
if now > validity.not_after.to_unix_duration() {
return Err(PasskiError::InvalidCertificateChain(
"certificate has expired".to_string(),
));
}
Ok(())
}
fn check_can_issue(cert: &Certificate, intermediates_below: usize) -> Result<()> {
let value = cert_extension(cert, &BasicConstraints::OID).ok_or_else(|| {
PasskiError::InvalidCertificateChain(
"issuing certificate has no Basic Constraints".to_string(),
)
})?;
let basic_constraints = BasicConstraints::from_der(value).map_err(|e| {
PasskiError::InvalidCertificateChain(format!("Invalid Basic Constraints: {}", e))
})?;
if !basic_constraints.ca {
return Err(PasskiError::InvalidCertificateChain(
"issuing certificate is not a CA".to_string(),
));
}
if let Some(max) = basic_constraints.path_len_constraint
&& intermediates_below > max as usize
{
return Err(PasskiError::InvalidCertificateChain(
"path length constraint exceeded".to_string(),
));
}
if let Some(value) = cert_extension(cert, &KeyUsage::OID) {
let key_usage = KeyUsage::from_der(value).map_err(|e| {
PasskiError::InvalidCertificateChain(format!("Invalid key usage: {}", e))
})?;
if !key_usage.key_cert_sign() {
return Err(PasskiError::InvalidCertificateChain(
"issuing certificate may not sign certificates".to_string(),
));
}
}
Ok(())
}
fn now() -> Duration {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
}
pub(crate) fn validate_trust_path(chain: &[Vec<u8>], anchors: &[Certificate]) -> Result<()> {
if anchors.is_empty() {
return Err(PasskiError::UntrustedAttestation);
}
let mut certificates = chain
.iter()
.map(|der| parse_cert(der))
.collect::<Result<Vec<_>>>()?;
let root_included = certificates.len() > 1
&& certificates
.last()
.is_some_and(|last| anchors.contains(last));
if root_included {
certificates.pop();
}
let deadline = now();
for certificate in &certificates {
check_validity(certificate, deadline)?;
}
for (i, link) in certificates.windows(2).enumerate() {
check_issued_by(&chain[i], &link[0], &link[1])?;
check_can_issue(&link[1], i)?;
}
let last = certificates.last().ok_or_else(|| {
PasskiError::InvalidCertificateChain("chain contains no certificates".to_string())
})?;
let intermediates_below = certificates.len() - 1;
let anchor = anchors
.iter()
.find(|anchor| check_issued_by(&chain[intermediates_below], last, anchor).is_ok())
.ok_or(PasskiError::UntrustedAttestation)?;
check_can_issue(anchor, intermediates_below)
}