use std::{borrow::Cow, io::Write};
use asn1_rs::FromDer;
use async_generic::async_generic;
use c2pa_raw_crypto::{
ec_utils::parse_ec_der_sig, validator_for_signing_alg, RawSignatureValidationError, SigningAlg,
};
use coset::CoseSign1;
use x509_parser::{
der_parser::oid, oid_registry::Oid, prelude::X509Certificate, x509::AlgorithmIdentifier,
};
use crate::{
crypto::{
asn1::rfc3161::TstInfo,
base64::encode,
cose::{
cert_chain_from_sign1, check_end_entity_certificate_profile, parse_cose_sign1,
signing_alg_from_sign1, CertificateInfo, CertificateTrustError, CertificateTrustPolicy,
CoseError, TrustAnchorType,
},
},
log_item,
status_tracker::StatusTracker,
validation_results::validation_codes::{
ALGORITHM_UNSUPPORTED, CLAIM_SIGNATURE_OUTSIDE_VALIDITY, SIGNING_CREDENTIAL_INVALID,
SIGNING_CREDENTIAL_TRUSTED, SIGNING_CREDENTIAL_UNTRUSTED,
},
};
#[derive(Debug)]
#[non_exhaustive]
pub enum Verifier<'a> {
VerifyTrustPolicy(Cow<'a, CertificateTrustPolicy>),
VerifyCertificateProfileOnly(Cow<'a, CertificateTrustPolicy>),
IgnoreProfileAndTrustPolicy,
}
const EC_PUBLICKEY_OID: Oid<'static> = oid!(1.2.840 .10045 .2 .1);
const RSA_OID: Oid<'static> = oid!(1.2.840 .113549 .1 .1 .1);
const RSASSA_PSS_OID: Oid<'static> = oid!(1.2.840 .113549 .1 .1 .10);
const ED25519_OID: Oid<'static> = oid!(1.3.101 .112);
const PRIME256V1_OID: Oid<'static> = oid!(1.2.840 .10045 .3 .1 .7);
const SECP384R1_OID: Oid<'static> = oid!(1.3.132 .0 .34);
const SECP521R1_OID: Oid<'static> = oid!(1.3.132 .0 .35);
fn spki_matches_signing_alg(alg: SigningAlg, spki: &AlgorithmIdentifier) -> bool {
let key_alg = &spki.algorithm;
let ec_named_curve_is = |curve: &Oid| {
*key_alg == EC_PUBLICKEY_OID
&& spki
.parameters
.as_ref()
.and_then(|p| p.as_oid().ok())
.is_some_and(|c| c == *curve)
};
match alg {
SigningAlg::Es256 => ec_named_curve_is(&PRIME256V1_OID),
SigningAlg::Es384 => ec_named_curve_is(&SECP384R1_OID),
SigningAlg::Es512 => ec_named_curve_is(&SECP521R1_OID),
SigningAlg::Ps256 | SigningAlg::Ps384 | SigningAlg::Ps512 => {
*key_alg == RSA_OID || *key_alg == RSASSA_PSS_OID
}
SigningAlg::Ed25519 => *key_alg == ED25519_OID,
_ => false,
}
}
impl Verifier<'_> {
#[async_generic]
pub fn verify_signature(
&self,
cose_sign1: &[u8],
data: &[u8],
additional_data: &[u8],
tst_info: Option<&TstInfo>,
validation_log: &mut StatusTracker,
) -> Result<CertificateInfo, CoseError> {
let mut sign1 = parse_cose_sign1(cose_sign1, data, validation_log)?;
let Ok(alg) = signing_alg_from_sign1(&sign1) else {
log_item!(
"Cose_Sign1",
"unsupported or missing Cose algorithm",
"verify_cose"
)
.validation_status(ALGORITHM_UNSUPPORTED)
.failure_no_throw(validation_log, CoseError::UnsupportedSigningAlgorithm);
return Err(CoseError::UnsupportedSigningAlgorithm);
};
if let (SigningAlg::Es256 | SigningAlg::Es384 | SigningAlg::Es512, true) =
(alg, parse_ec_der_sig(&sign1.signature).is_some())
{
log_item!(
"Cose_Sign1",
"unsupported signature format (EC signature should be in P1363 r|s format)",
"verify_cose"
)
.validation_status(SIGNING_CREDENTIAL_INVALID)
.failure_no_throw(validation_log, CoseError::InvalidEcdsaSignature);
return Err(CoseError::InvalidEcdsaSignature);
}
let trust_log_start = validation_log.logged_items().len();
let result = if _sync {
self.verify_trust(&sign1, tst_info, validation_log)
} else {
self.verify_trust_async(&sign1, tst_info, validation_log)
.await
};
let override_ekus = match result {
Ok((trust_type, Some(trust_uri))) => match self {
Self::VerifyTrustPolicy(ref ctp) => {
if let Some(anchor) = ctp.get_anchor_set(trust_type, &trust_uri) {
anchor.trust_config.clone()
} else {
None
}
}
_ => None,
},
_ => None,
};
let profile_log_start = validation_log.logged_items().len();
let profile_failed = if _sync {
self.verify_profile(&sign1, tst_info, override_ekus, validation_log)
} else {
self.verify_profile_async(&sign1, tst_info, override_ekus, validation_log)
.await
}
.is_err();
if profile_failed
|| validation_log.logged_items()[profile_log_start..]
.iter()
.any(|item| item.validation_status.as_deref() == Some(SIGNING_CREDENTIAL_INVALID))
{
validation_log.remove_status_since(trust_log_start, SIGNING_CREDENTIAL_TRUSTED);
}
sign1.payload = Some(data.to_vec());
let tbs = sign1.tbs_data(additional_data);
let certs = cert_chain_from_sign1(&sign1)?;
let end_entity_cert_der = &certs[0];
let (_rem, sign_cert) = X509Certificate::from_der(end_entity_cert_der)
.map_err(|_| CoseError::CborParsingError("invalid X509 certificate".to_string()))?;
let pk = sign_cert.public_key();
let pk_der = pk.raw;
if !spki_matches_signing_alg(alg, &pk.algorithm) {
log_item!(
"Cose_Sign1",
"certificate key algorithm does not match COSE signing algorithm",
"verify_cose"
)
.validation_status(SIGNING_CREDENTIAL_INVALID)
.failure_no_throw(
validation_log,
CoseError::RawSignatureValidationError(
RawSignatureValidationError::InvalidPublicKey,
),
);
return Err(RawSignatureValidationError::InvalidPublicKey.into());
}
let Some(validator) = validator_for_signing_alg(alg) else {
return Err(CoseError::UnsupportedSigningAlgorithm);
};
validator.validate(&sign1.signature, &tbs, pk_der)?;
let subject = sign_cert
.subject()
.iter_organization()
.map(|attr| attr.as_str())
.last()
.and_then(|attr| attr.ok())
.map(|a| a.to_string());
let common_name = sign_cert
.subject()
.iter_common_name()
.map(|attr| attr.as_str())
.last()
.and_then(|attr| attr.ok())
.map(|a| a.to_string());
Ok(CertificateInfo {
alg: Some(alg),
date: tst_info.map(|t| t.gen_time.clone().into()),
cert_serial_number: Some(sign_cert.serial.clone()),
issuer_org: subject,
common_name,
validated: true,
cert_chain: dump_cert_chain(&certs)?,
revocation_status: Some(true),
..Default::default()
})
}
#[async_generic]
pub(crate) fn verify_profile(
&self,
sign1: &CoseSign1,
tst_info: Option<&TstInfo>,
additional_ekus: Option<String>,
validation_log: &mut StatusTracker,
) -> Result<(), CoseError> {
let ctp = match self {
Self::VerifyTrustPolicy(ref ctp) => ctp,
Self::VerifyCertificateProfileOnly(ref ctp) => ctp,
Self::IgnoreProfileAndTrustPolicy => {
return Ok(());
}
};
let certs = cert_chain_from_sign1(sign1)?;
let end_entity_cert_der = &certs[0];
if let Some(ekus) = additional_ekus {
let mut adjusted_ctp = ctp.clone();
adjusted_ctp.to_mut().add_valid_ekus(ekus.as_bytes());
Ok(check_end_entity_certificate_profile(
end_entity_cert_der,
adjusted_ctp.as_ref(),
validation_log,
tst_info,
)?)
} else {
Ok(check_end_entity_certificate_profile(
end_entity_cert_der,
ctp.as_ref(),
validation_log,
tst_info,
)?)
}
}
#[async_generic]
pub(crate) fn verify_trust(
&self,
sign1: &CoseSign1,
tst_info_res: Option<&TstInfo>,
validation_log: &mut StatusTracker,
) -> Result<(TrustAnchorType, Option<String>), CoseError> {
let ctp = match self {
Self::VerifyTrustPolicy(ref ctp) => ctp,
Self::VerifyCertificateProfileOnly(ref _ctp) => {
return Ok((TrustAnchorType::NoCheck, None));
}
Self::IgnoreProfileAndTrustPolicy => {
return Ok((TrustAnchorType::NoCheck, None));
}
};
let certs = cert_chain_from_sign1(sign1)?;
let end_entity_cert_der = &certs[0];
let chain_der = &certs[1..];
let signing_time_epoch = tst_info_res.map(|tst_info| {
let dt: chrono::DateTime<chrono::Utc> = tst_info.gen_time.clone().into();
dt.timestamp()
});
let verify_result = if _sync {
ctp.check_certificate_trust(chain_der, end_entity_cert_der, signing_time_epoch)
} else {
ctp.check_certificate_trust_async(chain_der, end_entity_cert_der, signing_time_epoch)
.await
};
match verify_result {
Ok((tat, trust_uri)) => {
log_item!(
"",
format!(
"signing certificate trusted, found in [{}] trust anchors",
&trust_uri
),
"verify_cose"
)
.validation_status(SIGNING_CREDENTIAL_TRUSTED)
.set_trust_list_uri(&trust_uri)
.success(validation_log);
Ok((tat, Some(trust_uri)))
}
Err(e) => {
if e == CertificateTrustError::CaCertificateOutsideValidity {
log_item!(
"",
"certificate chain outside validity period",
"verify_cose"
)
.validation_status(CLAIM_SIGNATURE_OUTSIDE_VALIDITY)
.failure_no_throw(
validation_log,
CertificateTrustError::CaCertificateOutsideValidity,
);
}
Err(
log_item!("", "signing certificate untrusted", "verify_cose")
.validation_status(SIGNING_CREDENTIAL_UNTRUSTED)
.failure_as_err(validation_log, e.into()),
)
}
}
}
}
impl Default for Verifier<'_> {
fn default() -> Self {
Self::VerifyTrustPolicy(Cow::Owned(CertificateTrustPolicy::default()))
}
}
fn dump_cert_chain(certs: &[Vec<u8>]) -> Result<Vec<u8>, CoseError> {
let mut writer = Vec::new();
let line_len = 64;
let cert_begin = "-----BEGIN CERTIFICATE-----";
let cert_end = "-----END CERTIFICATE-----";
for der_bytes in certs {
let cert_base_str = encode(der_bytes);
let cert_lines = cert_base_str
.chars()
.collect::<Vec<char>>()
.chunks(line_len)
.map(|chunk| chunk.iter().collect::<String>())
.collect::<Vec<_>>();
writer
.write_fmt(format_args!("{cert_begin}\n"))
.map_err(|_e| CoseError::InternalError("could not write PEM".to_string()))?;
for l in cert_lines {
writer
.write_fmt(format_args!("{l}\n"))
.map_err(|_e| CoseError::InternalError("could not write PEM".to_string()))?;
}
writer
.write_fmt(format_args!("{cert_end}\n"))
.map_err(|_e| CoseError::InternalError("could not write PEM".to_string()))?;
}
Ok(writer)
}
#[cfg(test)]
mod tests {
#![allow(clippy::unwrap_used)]
use c2pa_raw_crypto::SigningAlg;
#[cfg(all(target_arch = "wasm32", not(target_os = "wasi")))]
use wasm_bindgen_test::wasm_bindgen_test;
use x509_parser::prelude::{FromDer, Pem, X509Certificate};
use super::spki_matches_signing_alg;
const ES256: &[u8] = include_bytes!("../../../tests/fixtures/crypto/raw_signature/es256.pub");
const ES384: &[u8] = include_bytes!("../../../tests/fixtures/crypto/raw_signature/es384.pub");
const ES512: &[u8] = include_bytes!("../../../tests/fixtures/crypto/raw_signature/es512.pub");
const PS256: &[u8] = include_bytes!("../../../tests/fixtures/crypto/raw_signature/ps256.pub");
const PS384: &[u8] = include_bytes!("../../../tests/fixtures/crypto/raw_signature/ps384.pub");
const ED25519: &[u8] = include_bytes!("../../../tests/fixtures/certs/ed25519.pub");
const ED448: &[u8] = include_bytes!("../../../tests/fixtures/crypto/raw_signature/ed448.pub");
fn key_matches(cert_pem: &[u8], alg: SigningAlg) -> bool {
let pem = Pem::iter_from_buffer(cert_pem).next().unwrap().unwrap();
let (_, cert) = X509Certificate::from_der(&pem.contents).unwrap();
spki_matches_signing_alg(alg, &cert.public_key().algorithm)
}
#[test]
fn matching_keys_accepted() {
assert!(key_matches(ED25519, SigningAlg::Ed25519));
assert!(key_matches(ES256, SigningAlg::Es256));
assert!(key_matches(ES384, SigningAlg::Es384));
assert!(key_matches(ES512, SigningAlg::Es512));
assert!(key_matches(PS256, SigningAlg::Ps256));
assert!(key_matches(PS384, SigningAlg::Ps384));
}
#[test]
fn wrong_key_type_rejected() {
assert!(!key_matches(ED448, SigningAlg::Ed25519));
assert!(!key_matches(ES256, SigningAlg::Ed25519));
assert!(!key_matches(ED25519, SigningAlg::Es256));
assert!(!key_matches(PS256, SigningAlg::Es256));
assert!(!key_matches(ES256, SigningAlg::Ps256));
assert!(!key_matches(PS256, SigningAlg::Ed25519));
}
#[test]
fn wrong_ec_curve_rejected() {
assert!(!key_matches(ES256, SigningAlg::Es384));
assert!(!key_matches(ES256, SigningAlg::Es512));
assert!(!key_matches(ES384, SigningAlg::Es256));
assert!(!key_matches(ES512, SigningAlg::Es256));
}
#[c2pa_macros::c2pa_test_async]
async fn ca_outside_validity_reports_both_failure_codes() {
use std::borrow::Cow;
use coset::cbor::value::Value;
use super::*;
use crate::crypto::{
asn1::{rfc3161::MessageImprint, AlgorithmIdentifier, GeneralizedTime},
cert_chain_pem_to_der,
};
let tst_info = TstInfo {
version: 1.into(),
policy: "1.2.3".parse().unwrap(),
message_imprint: MessageImprint {
hash_algorithm: AlgorithmIdentifier {
algorithm: "2.16.840.1.101.3.4.2.1".parse().unwrap(),
},
hashed_message: bcder::OctetString::new(bytes::Bytes::from_static(&[0; 32])),
},
serial_number: 1.into(),
gen_time: GeneralizedTime::from_der_bytes(b"\x18\x0f20260901000000Z").unwrap(),
accuracy: None,
ordering: None,
nonce: None,
tsa: None,
extensions: None,
};
for (chain_pem, root_pem, outside_validity) in [
(
include_bytes!(
"../../../tests/fixtures/crypto/cose/chain_trust_expired_intermediate.pub"
)
.as_slice(),
include_bytes!("../../../tests/fixtures/crypto/cose/chain_trust_root.pub")
.as_slice(),
true,
),
(
include_bytes!(
"../../../tests/fixtures/crypto/cose/chain_trust_future_intermediate.pub"
)
.as_slice(),
include_bytes!("../../../tests/fixtures/crypto/cose/chain_trust_future_root.pub")
.as_slice(),
true,
),
(
include_bytes!("../../../tests/fixtures/crypto/cose/chain_trust_valid.pub")
.as_slice(),
include_bytes!("../../../tests/fixtures/crypto/cose/chain_trust_root.pub")
.as_slice(),
false,
),
] {
let certs = cert_chain_pem_to_der(chain_pem).unwrap();
let mut sign1 = CoseSign1::default();
sign1.protected.header.rest.push((
coset::Label::Int(coset::iana::HeaderParameter::X5Chain as i64),
Value::Array(certs.into_iter().map(Value::Bytes).collect()),
));
let mut policy = CertificateTrustPolicy::new();
policy
.add_trust_anchors(
root_pem,
"http://some_trust_list",
TrustAnchorType::Manifest,
None,
)
.unwrap();
let verifier = Verifier::VerifyTrustPolicy(Cow::Borrowed(&policy));
for is_async in [false, true] {
let mut log = StatusTracker::default();
let result = if is_async {
verifier
.verify_trust_async(&sign1, Some(&tst_info), &mut log)
.await
} else {
verifier.verify_trust(&sign1, Some(&tst_info), &mut log)
};
if outside_validity {
assert!(matches!(
result,
Err(CoseError::CertificateTrustError(
CertificateTrustError::CaCertificateOutsideValidity
))
));
assert!(log.has_status(CLAIM_SIGNATURE_OUTSIDE_VALIDITY));
assert!(log.has_status(SIGNING_CREDENTIAL_UNTRUSTED));
assert!(!log.has_status(SIGNING_CREDENTIAL_TRUSTED));
assert_eq!(log.logged_items().len(), 2);
} else {
assert_eq!(
result.unwrap(),
(
TrustAnchorType::Manifest,
Some("http://some_trust_list".into())
)
);
assert!(log.has_status(SIGNING_CREDENTIAL_TRUSTED));
assert!(!log.has_any_error());
assert!(!log.has_status(CLAIM_SIGNATURE_OUTSIDE_VALIDITY));
assert!(!log.has_status(SIGNING_CREDENTIAL_UNTRUSTED));
}
}
}
}
#[c2pa_macros::c2pa_test_async]
async fn profile_failure_is_not_trusted() {
use std::borrow::Cow;
use c2pa_raw_crypto::signer_from_private_key;
use super::Verifier;
use crate::{
crypto::{
cert_chain_pem_to_der,
cose::{CertificateTrustPolicy, TrustAnchorType},
},
settings::Settings,
signer::RawSignerWrapper,
status_tracker::StatusTracker,
validation_results::validation_codes::{
SIGNING_CREDENTIAL_INVALID, SIGNING_CREDENTIAL_TRUSTED,
},
};
let cert_chain = include_bytes!("../../../tests/fixtures/crypto/cose/disallowed_eku.pub");
let private_key = include_bytes!("../../../tests/fixtures/crypto/cose/disallowed_eku.priv");
let root = include_bytes!("../../../tests/fixtures/crypto/cose/key_usage_root.pub");
let signer = RawSignerWrapper::new(
signer_from_private_key(private_key, SigningAlg::Es256).unwrap(),
cert_chain_pem_to_der(cert_chain).unwrap(),
None,
);
let mut claim = crate::claim::Claim::new("profile_test", Some("contentauth"), 1);
claim.build().unwrap();
let claim_bytes = claim.data().unwrap();
let mut settings = Settings::default();
settings.verify.verify_trust = false;
let cose_bytes =
crate::cose_sign::sign_claim(&claim_bytes, &signer, 10000, &settings).unwrap();
let sign1 =
super::parse_cose_sign1(&cose_bytes, &claim_bytes, &mut StatusTracker::default())
.unwrap();
for (allow_eku, require_missing_eku) in [(false, false), (true, true), (true, false)] {
let mut ctp = CertificateTrustPolicy::new();
ctp.add_trust_anchors(
root,
"http://some_trust_list",
TrustAnchorType::Manifest,
None,
)
.unwrap();
if allow_eku {
ctp.add_valid_ekus(b"1.3.6.1.5.5.7.3.36");
}
if require_missing_eku {
ctp.add_mandatory_ekus(b"1.3.6.1.5.5.7.3.8");
}
let verifier = Verifier::VerifyTrustPolicy(Cow::Owned(ctp));
let invalid = !allow_eku || require_missing_eku;
for use_async in [false, true] {
let mut profile_log = StatusTracker::default();
let profile = if use_async {
verifier
.verify_profile_async(&sign1, None, None, &mut profile_log)
.await
} else {
verifier.verify_profile(&sign1, None, None, &mut profile_log)
};
assert_eq!(profile.is_err(), !allow_eku);
assert_eq!(profile_log.has_status(SIGNING_CREDENTIAL_INVALID), invalid);
let mut log = StatusTracker::default();
crate::log_item!(
"previous claim",
"trusted",
"profile_failure_is_not_trusted"
)
.validation_status(SIGNING_CREDENTIAL_TRUSTED)
.success(&mut log);
crate::log_item!(
"previous claim",
"invalid",
"profile_failure_is_not_trusted"
)
.validation_status(SIGNING_CREDENTIAL_INVALID)
.failure_no_throw(&mut log, super::CoseError::UnsupportedSigningAlgorithm);
let verification_log_start = log.logged_items().len();
let result = if use_async {
verifier
.verify_signature_async(&cose_bytes, &claim_bytes, b"", None, &mut log)
.await
} else {
verifier.verify_signature(&cose_bytes, &claim_bytes, b"", None, &mut log)
};
assert!(result.unwrap().validated);
assert_eq!(
log.logged_items()[verification_log_start..]
.iter()
.any(|item| item.validation_status.as_deref()
== Some(SIGNING_CREDENTIAL_INVALID)),
invalid
);
assert_eq!(
log.logged_items()
.iter()
.filter(|item| item.validation_status.as_deref()
== Some(SIGNING_CREDENTIAL_TRUSTED))
.count(),
if invalid { 1 } else { 2 }
);
assert_eq!(log.logged_items()[0].label, "previous claim");
}
}
}
#[cfg(any(target_os = "linux", target_os = "macos"))]
#[test]
fn profile_failure_without_key_usage_is_not_trusted() {
use std::borrow::Cow;
use c2pa_raw_crypto::signer_from_private_key;
use super::Verifier;
use crate::{
crypto::cose::{
cose_signer::RawSignerCoseSigner, sign, CertificateTrustPolicy, TimeStampStorage,
TrustAnchorType,
},
status_tracker::StatusTracker,
utils::ephemeral_cert::generate_ephemeral_chain_with_ee_skip,
validation_results::validation_codes::{
SIGNING_CREDENTIAL_INVALID, SIGNING_CREDENTIAL_TRUSTED,
},
};
let chain =
generate_ephemeral_chain_with_ee_skip("profile-test.example.com", &["key_usage"])
.unwrap();
let raw_signer =
signer_from_private_key(chain.ee_private_key_pem.as_bytes(), SigningAlg::Ed25519)
.unwrap();
let cert_chain = vec![chain.ee_der.clone(), chain.ca_der.clone()];
let signer = RawSignerCoseSigner::new(raw_signer.as_ref(), &cert_chain);
let claim_bytes = b"claim";
let cose_bytes =
sign(&signer, claim_bytes, None, TimeStampStorage::V2_sigTst2_CTT).unwrap();
let mut ctp = CertificateTrustPolicy::default();
ctp.add_trust_anchors(
format!(
"-----BEGIN CERTIFICATE-----\n{}\n-----END CERTIFICATE-----\n",
crate::crypto::base64::encode(&chain.ca_der)
)
.as_bytes(),
"http://some_trust_list",
TrustAnchorType::Manifest,
None,
)
.unwrap();
let mut log = StatusTracker::default();
let _ = Verifier::VerifyTrustPolicy(Cow::Owned(ctp)).verify_signature(
&cose_bytes,
claim_bytes,
b"",
None,
&mut log,
);
assert!(log.has_status(SIGNING_CREDENTIAL_INVALID));
assert!(!log.has_status(SIGNING_CREDENTIAL_TRUSTED));
}
}