use super::helpers::rp_id_hash;
use crate::{AttestationTrustPolicy, AttestationType, Passki, PasskiError};
use aws_lc_rs::digest::{SHA256, digest};
use aws_lc_rs::rand::SystemRandom;
use aws_lc_rs::signature::{ECDSA_P256_SHA256_ASN1_SIGNING, EcdsaKeyPair, KeyPair};
use ciborium::Value;
use std::str::FromStr;
use std::time::{Duration, SystemTime};
use x509_cert::Version;
use x509_cert::der::asn1::{Any, BitString, OctetString};
use x509_cert::der::oid::{AssociatedOid, ObjectIdentifier};
use x509_cert::der::{Encode, Sequence};
use x509_cert::ext::Extension;
use x509_cert::ext::pkix::BasicConstraints;
use x509_cert::name::Name;
use x509_cert::serial_number::SerialNumber;
use x509_cert::spki::{AlgorithmIdentifierOwned, SubjectPublicKeyInfoOwned};
use x509_cert::time::{Time, Validity};
const OID_EC_PUBLIC_KEY: ObjectIdentifier = ObjectIdentifier::new_unwrap("1.2.840.10045.2.1");
const OID_PRIME256V1: ObjectIdentifier = ObjectIdentifier::new_unwrap("1.2.840.10045.3.1.7");
const OID_ECDSA_SHA256: ObjectIdentifier = ObjectIdentifier::new_unwrap("1.2.840.10045.4.3.2");
#[derive(Sequence)]
struct TbsCertificate {
#[asn1(context_specific = "0", tag_mode = "EXPLICIT")]
version: Version,
serial_number: SerialNumber,
signature: AlgorithmIdentifierOwned,
issuer: Name,
validity: Validity,
subject: Name,
subject_public_key_info: SubjectPublicKeyInfoOwned,
#[asn1(context_specific = "3", tag_mode = "EXPLICIT", optional = "true")]
extensions: Option<Vec<Extension>>,
}
#[derive(Sequence)]
struct Certificate {
tbs_certificate: TbsCertificate,
signature_algorithm: AlgorithmIdentifierOwned,
signature: BitString,
}
#[derive(Clone, Copy)]
enum Role {
Ca(Option<u8>),
Leaf,
}
struct TestCert {
key_pair: EcdsaKeyPair,
name: Name,
der: Vec<u8>,
}
fn generate_key() -> EcdsaKeyPair {
let rng = SystemRandom::new();
let pkcs8 = EcdsaKeyPair::generate_pkcs8(&ECDSA_P256_SHA256_ASN1_SIGNING, &rng).unwrap();
EcdsaKeyPair::from_pkcs8(&ECDSA_P256_SHA256_ASN1_SIGNING, pkcs8.as_ref()).unwrap()
}
fn spki(key_pair: &EcdsaKeyPair) -> SubjectPublicKeyInfoOwned {
SubjectPublicKeyInfoOwned {
algorithm: AlgorithmIdentifierOwned {
oid: OID_EC_PUBLIC_KEY,
parameters: Some(Any::from(OID_PRIME256V1)),
},
subject_public_key: BitString::from_bytes(key_pair.public_key().as_ref()).unwrap(),
}
}
fn basic_constraints(role: Role) -> Extension {
let constraints = match role {
Role::Ca(path_len_constraint) => BasicConstraints {
ca: true,
path_len_constraint,
},
Role::Leaf => BasicConstraints {
ca: false,
path_len_constraint: None,
},
};
Extension {
extn_id: BasicConstraints::OID,
critical: true,
extn_value: OctetString::new(constraints.to_der().unwrap()).unwrap(),
}
}
fn validity(not_before: SystemTime, not_after: SystemTime) -> Validity {
Validity::new(
Time::try_from(not_before).unwrap(),
Time::try_from(not_after).unwrap(),
)
}
fn current() -> Validity {
let now = SystemTime::now();
validity(
now - Duration::from_secs(3600),
now + Duration::from_secs(3600),
)
}
fn expired() -> Validity {
let now = SystemTime::now();
validity(
now - Duration::from_secs(7200),
now - Duration::from_secs(3600),
)
}
fn not_yet_valid() -> Validity {
let now = SystemTime::now();
validity(
now + Duration::from_secs(3600),
now + Duration::from_secs(7200),
)
}
fn issue_certificate(
issuer_key: &EcdsaKeyPair,
issuer_name: &Name,
subject_key: &EcdsaKeyPair,
subject_name: &Name,
role: Role,
validity: Validity,
) -> Vec<u8> {
let signature_algorithm = AlgorithmIdentifierOwned {
oid: OID_ECDSA_SHA256,
parameters: None,
};
let tbs_certificate = TbsCertificate {
version: Version::V3,
serial_number: SerialNumber::new(&[0x01, 0x02, 0x03, 0x04]).unwrap(),
signature: signature_algorithm.clone(),
issuer: issuer_name.clone(),
validity,
subject: subject_name.clone(),
subject_public_key_info: spki(subject_key),
extensions: Some(vec![basic_constraints(role)]),
};
let rng = SystemRandom::new();
let signature = issuer_key
.sign(&rng, &tbs_certificate.to_der().unwrap())
.unwrap();
Certificate {
tbs_certificate,
signature_algorithm,
signature: BitString::from_bytes(signature.as_ref()).unwrap(),
}
.to_der()
.unwrap()
}
impl TestCert {
fn root(common_name: &str) -> Self {
Self::root_with(common_name, Role::Ca(None), current())
}
fn root_with(common_name: &str, role: Role, validity: Validity) -> Self {
let key_pair = generate_key();
let name = Name::from_str(&format!("CN={}", common_name)).unwrap();
let der = issue_certificate(&key_pair, &name, &key_pair, &name, role, validity);
Self {
key_pair,
name,
der,
}
}
fn issue(&self, common_name: &str, role: Role, validity: Validity) -> Self {
let key_pair = generate_key();
let name = Name::from_str(&format!("CN={}", common_name)).unwrap();
let der = issue_certificate(&self.key_pair, &self.name, &key_pair, &name, role, validity);
Self {
key_pair,
name,
der,
}
}
fn ca(&self, common_name: &str) -> Self {
self.issue(common_name, Role::Ca(None), current())
}
fn leaf(&self, common_name: &str) -> Self {
self.issue(common_name, Role::Leaf, current())
}
}
fn packed_attestation_over(chain: &[&TestCert], client_data_hash: &[u8]) -> Vec<u8> {
let cose_key = vec![
(Value::Integer(1.into()), Value::Integer(2.into())), (Value::Integer(3.into()), Value::Integer((-7).into())), (Value::Integer((-1).into()), Value::Integer(1.into())), (Value::Integer((-2).into()), Value::Bytes(vec![2u8; 32])), (Value::Integer((-3).into()), Value::Bytes(vec![3u8; 32])), ];
let mut cose_key_bytes = Vec::new();
ciborium::into_writer(&Value::Map(cose_key), &mut cose_key_bytes).unwrap();
let mut auth_data = Vec::new();
auth_data.extend_from_slice(&rp_id_hash("localhost"));
auth_data.push(0x45); auth_data.extend_from_slice(&[0, 0, 0, 0]); auth_data.extend_from_slice(&[7u8; 16]); auth_data.extend_from_slice(&[0, 16]); auth_data.extend_from_slice(&[1u8; 16]); auth_data.extend_from_slice(&cose_key_bytes);
let mut signed = auth_data.clone();
signed.extend_from_slice(client_data_hash);
let rng = SystemRandom::new();
let signature = chain[0].key_pair.sign(&rng, &signed).unwrap();
let att_stmt = vec![
(Value::Text("alg".to_string()), Value::Integer((-7).into())),
(
Value::Text("sig".to_string()),
Value::Bytes(signature.as_ref().to_vec()),
),
(
Value::Text("x5c".to_string()),
Value::Array(
chain
.iter()
.map(|cert| Value::Bytes(cert.der.clone()))
.collect(),
),
),
];
let att_obj = vec![
(
Value::Text("fmt".to_string()),
Value::Text("packed".to_string()),
),
(Value::Text("authData".to_string()), Value::Bytes(auth_data)),
(Value::Text("attStmt".to_string()), Value::Map(att_stmt)),
];
let mut bytes = Vec::new();
ciborium::into_writer(&Value::Map(att_obj), &mut bytes).unwrap();
bytes
}
fn packed_attestation(chain: &[&TestCert]) -> (Vec<u8>, Vec<u8>) {
let client_data_hash = digest(&SHA256, b"trust path client data").as_ref().to_vec();
let bytes = packed_attestation_over(chain, &client_data_hash);
(bytes, client_data_hash)
}
fn passki() -> Passki {
Passki::new("localhost", &["http://localhost:3000"], "Test")
}
fn passki_trusting(roots: &[&TestCert], policy: AttestationTrustPolicy) -> Passki {
let ders: Vec<&[u8]> = roots.iter().map(|root| root.der.as_slice()).collect();
passki().with_attestation_trust(&ders, policy).unwrap()
}
#[test]
fn test_default_policy_leaves_a_chain_unverified() {
let foreign = TestCert::root("Passki Foreign Root");
let leaf = foreign.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf]);
let parsed = passki()
.verify_attestation(&bytes, &client_data_hash)
.unwrap();
assert_eq!(parsed.attestation_type, AttestationType::Unverified);
}
#[test]
fn test_chain_to_installed_root_is_basic() {
let root = TestCert::root("Passki Test Root");
let leaf = root.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf]);
let parsed = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap();
assert_eq!(parsed.attestation_type, AttestationType::Basic);
assert_eq!(parsed.aaguid, [7u8; 16]);
}
#[test]
fn test_chain_through_an_intermediate_is_accepted() {
let root = TestCert::root("Passki Test Root");
let intermediate = root.ca("Passki Test Intermediate");
let leaf = intermediate.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf, &intermediate]);
let parsed = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap();
assert_eq!(parsed.attestation_type, AttestationType::Basic);
}
#[test]
fn test_root_included_in_x5c_is_accepted() {
let root = TestCert::root("Passki Test Root");
let leaf = root.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf, &root]);
let parsed = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap();
assert_eq!(parsed.attestation_type, AttestationType::Basic);
}
#[test]
fn test_self_signed_root_in_x5c_without_an_anchor_is_rejected() {
let foreign = TestCert::root("Passki Foreign Root");
let leaf = foreign.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf, &foreign]);
let root = TestCert::root("Passki Test Root");
let err = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(matches!(err, PasskiError::UntrustedAttestation));
}
#[test]
fn test_certificate_from_a_foreign_root_is_rejected() {
let foreign = TestCert::root("Passki Foreign Root");
let leaf = foreign.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf]);
let root = TestCert::root("Passki Test Root");
let err = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(matches!(err, PasskiError::UntrustedAttestation));
}
#[test]
fn test_anchor_with_the_same_name_but_a_different_key_is_rejected() {
let root = TestCert::root("Passki Test Root");
let leaf = root.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf]);
let impostor = TestCert::root("Passki Test Root");
let err = passki_trusting(&[&impostor], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(matches!(err, PasskiError::UntrustedAttestation));
}
#[test]
fn test_broken_signature_link_inside_the_chain_is_rejected() {
let root = TestCert::root("Passki Test Root");
let signing = root.ca("Passki Test Intermediate");
let decoy = root.ca("Passki Test Intermediate");
let leaf = signing.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf, &decoy]);
let err = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(
matches!(&err, PasskiError::InvalidCertificateChain(m) if m.contains("broken signature link")),
"unexpected error: {}",
err
);
}
#[test]
fn test_wrong_issuer_name_is_rejected() {
let root = TestCert::root("Passki Test Root");
let intermediate = root.ca("Passki Test Intermediate");
let other = root.ca("Passki Other Intermediate");
let leaf = intermediate.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf, &other]);
let err = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(
matches!(&err, PasskiError::InvalidCertificateChain(m) if m.contains("issuer name")),
"unexpected error: {}",
err
);
}
#[test]
fn test_expired_certificate_is_rejected() {
let root = TestCert::root("Passki Test Root");
let leaf = root.issue("Passki Attestation", Role::Leaf, expired());
let (bytes, client_data_hash) = packed_attestation(&[&leaf]);
let err = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(
matches!(&err, PasskiError::InvalidCertificateChain(m) if m.contains("expired")),
"unexpected error: {}",
err
);
}
#[test]
fn test_not_yet_valid_certificate_is_rejected() {
let root = TestCert::root("Passki Test Root");
let leaf = root.issue("Passki Attestation", Role::Leaf, not_yet_valid());
let (bytes, client_data_hash) = packed_attestation(&[&leaf]);
let err = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(
matches!(&err, PasskiError::InvalidCertificateChain(m) if m.contains("not valid yet")),
"unexpected error: {}",
err
);
}
#[test]
fn test_expired_anchor_still_anchors() {
let root = TestCert::root_with("Passki Test Root", Role::Ca(None), expired());
let leaf = root.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf]);
let parsed = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap();
assert_eq!(parsed.attestation_type, AttestationType::Basic);
}
#[test]
fn test_non_ca_issuer_is_rejected() {
let root = TestCert::root("Passki Test Root");
let impostor = root.leaf("Passki Not A CA");
let leaf = impostor.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf, &impostor]);
let err = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(
matches!(&err, PasskiError::InvalidCertificateChain(m) if m.contains("not a CA")),
"unexpected error: {}",
err
);
}
#[test]
fn test_path_length_constraint_is_enforced() {
let root = TestCert::root_with("Passki Test Root", Role::Ca(Some(0)), current());
let intermediate = root.ca("Passki Test Intermediate");
let leaf = intermediate.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf, &intermediate]);
let err = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(
matches!(&err, PasskiError::InvalidCertificateChain(m) if m.contains("path length")),
"unexpected error: {}",
err
);
}
#[test]
fn test_chain_without_any_anchor_installed_is_rejected() {
let root = TestCert::root("Passki Test Root");
let leaf = root.leaf("Passki Attestation");
let (bytes, client_data_hash) = packed_attestation(&[&leaf]);
let passki = passki()
.with_attestation_trust(&[] as &[&[u8]], AttestationTrustPolicy::VerifyWhenPresent)
.unwrap();
let err = passki
.verify_attestation(&bytes, &client_data_hash)
.unwrap_err();
assert!(matches!(err, PasskiError::UntrustedAttestation));
}
#[test]
fn test_unparsable_anchor_is_rejected() {
let result = passki().with_attestation_trust(
&[&[0xFF, 0xFE, 0xFD][..]],
AttestationTrustPolicy::VerifyWhenPresent,
);
assert!(matches!(
result.err(),
Some(PasskiError::InvalidCertificate(_))
));
}
#[test]
fn test_none_format_reports_no_attestation() {
let attestation_obj = super::helpers::create_test_attestation_object(-7, 0x45);
let parsed = passki()
.verify_attestation(&attestation_obj, &[0u8; 32])
.unwrap();
assert_eq!(parsed.attestation_type, AttestationType::None);
}
#[test]
fn test_required_policy_rejects_a_statement_without_certificates() {
let root = TestCert::root("Passki Test Root");
let attestation_obj = super::helpers::create_test_attestation_object(-7, 0x45);
let err = passki_trusting(&[&root], AttestationTrustPolicy::Required)
.verify_attestation(&attestation_obj, &[0u8; 32])
.unwrap_err();
assert!(matches!(err, PasskiError::MissingAttestationChain));
}
#[test]
fn test_verify_when_present_accepts_a_statement_without_certificates() {
let root = TestCert::root("Passki Test Root");
let attestation_obj = super::helpers::create_test_attestation_object(-7, 0x45);
let parsed = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent)
.verify_attestation(&attestation_obj, &[0u8; 32])
.unwrap();
assert_eq!(parsed.attestation_type, AttestationType::None);
}
#[test]
fn test_registration_stores_the_attestation_type() {
use crate::{RegistrationCredential, RegistrationOptions};
let root = TestCert::root("Passki Test Root");
let leaf = root.leaf("Passki Attestation");
let passki = passki_trusting(&[&root], AttestationTrustPolicy::VerifyWhenPresent);
let (_, state) = passki
.start_passkey_registration(
b"test_user_id_1234567890",
"alice@example.com",
"Alice",
RegistrationOptions::default(),
)
.unwrap();
let client_data_json =
super::helpers::create_test_client_data_json(&state.challenge, "http://localhost:3000");
let client_data_hash = digest(&SHA256, &client_data_json).as_ref().to_vec();
let attestation_obj = packed_attestation_over(&[&leaf], &client_data_hash);
let credential = RegistrationCredential {
credential_id: Passki::base64_encode(&[1u8; 16]),
public_key: Passki::base64_encode(&attestation_obj),
client_data_json: Passki::base64_encode(&client_data_json),
client_extension_results: None,
authenticator_attachment: None,
transports: Vec::new(),
};
let stored = passki
.finish_passkey_registration(&credential, &state)
.unwrap();
assert_eq!(stored.attestation_type, AttestationType::Basic);
assert_eq!(stored.aaguid, [7u8; 16]);
}
#[test]
fn test_attestation_type_round_trips_through_json() {
let stored = crate::StoredPasskey {
credential_id: vec![1u8; 16],
public_key: vec![2u8; 32],
counter: 0,
algorithm: -7,
aaguid: [7u8; 16],
attestation_type: AttestationType::AttCa,
transports: Vec::new(),
rk: None,
large_blob_supported: None,
be: false,
bs: false,
};
let json = serde_json::to_string(&stored).unwrap();
assert!(json.contains(r#""attestation_type":"att-ca""#));
let decoded: crate::StoredPasskey = serde_json::from_str(&json).unwrap();
assert_eq!(decoded.attestation_type, AttestationType::AttCa);
}
#[test]
fn test_stored_passkey_without_attestation_type_deserializes() {
let json = r#"{
"credential_id": [1],
"public_key": [2],
"counter": 0,
"algorithm": -7
}"#;
let decoded: crate::StoredPasskey = serde_json::from_str(json).unwrap();
assert_eq!(decoded.attestation_type, AttestationType::None);
}