use std::convert::TryFrom;
use std::str::FromStr;
use std::sync::{Arc, Mutex, MutexGuard};
use anyhow::{anyhow, Context, Result};
use openpgp_card::{algorithm::AlgoSimple, KeyType};
use openpgp_card_pcsc::PcscBackend;
use openpgp_card_sequoia::state::{Admin, Open, Transaction};
use openpgp_card_sequoia::util::public_key_material_to_key;
use openpgp_card_sequoia::{sq_util, Card, PublicKey};
use sequoia_openpgp::packet::key::{KeyRole, PrimaryRole, SubordinateRole};
use sequoia_openpgp::packet::prelude::SignatureBuilder;
use sequoia_openpgp::packet::{Signature, UserID};
use sequoia_openpgp::policy::StandardPolicy;
use sequoia_openpgp::types::{
Features, HashAlgorithm, KeyFlags, SignatureType, SymmetricAlgorithm,
};
use sequoia_openpgp::{Cert, Packet};
use crate::backend;
use crate::backend::{Backend, CertificationBackend};
use crate::pgp;
use crate::storage::UninitDb;
pub(crate) fn card_matches(transaction: &mut Card<Transaction>, ca_cert: &Cert) -> Result<String> {
let fps = transaction.fingerprints()?;
let auth = fps
.authentication()
.context("No AUT key on card".to_string())?;
let auth_fp = auth.to_string();
let cardholder_name = transaction.cardholder_name()?;
if cardholder_name.as_deref() != Some("OpenPGP CA") {
return Err(anyhow::anyhow!(
"Expected cardholder name 'OpenPGP CA' on OpenPGP card, found '{}'.",
cardholder_name.unwrap_or_default()
));
}
if ca_cert.userids().next().is_none() {
return Err(anyhow::anyhow!(
"Expect CA certificate to contain at least one User ID, but found none."
));
}
let pubkey =
pgp::cert_to_armored(ca_cert).context("Failed to transform CA cert to armored pubkey")?;
if ca_cert.fingerprint().to_hex() != auth_fp {
return Err(anyhow::anyhow!(format!(
"Auth key slot on card {} doesn't match primary (cert) fingerprint {}.",
auth_fp,
ca_cert.fingerprint().to_hex()
)));
}
Ok(pubkey)
}
pub(crate) fn check_if_card_matches(card_ident: &str, ca_cert: &Cert) -> Result<String> {
let backend = PcscBackend::open_by_ident(card_ident, None)?;
let mut card: Card<Open> = backend.into();
let mut transaction = card.transaction()?;
card_matches(&mut transaction, ca_cert).context(format!("On card {card_ident}"))
}
pub(crate) struct CardBackend {
pin: String,
ident: String,
card: Arc<Mutex<Option<Card<Open>>>>,
}
impl CertificationBackend for CardBackend {
fn certify(
&self,
op: &mut dyn FnMut(&mut dyn sequoia_openpgp::crypto::Signer) -> Result<()>,
) -> Result<()> {
let mut card = self.card()?;
let card = card
.as_mut()
.expect("CardCa::card() should always return a Some(_)");
let mut open = card.transaction()?;
open.verify_user(self.pin.as_bytes())?;
let mut user = open
.user_card()
.ok_or_else(|| anyhow!("Unexpected: can't get card in user mode"))?;
let mut signer =
user.authenticator(&|| println!("Touch confirmation needed for certification"))?;
op(&mut signer as &mut dyn sequoia_openpgp::crypto::Signer)?;
Ok(())
}
fn sign(
&self,
op: &mut dyn FnMut(&mut dyn sequoia_openpgp::crypto::Signer) -> Result<()>,
) -> Result<()> {
let mut card = self.card()?;
let card = card
.as_mut()
.expect("CardCa::card() should always return a Some(_)");
let mut open = card.transaction()?;
open.verify_user_for_signing(self.pin.as_bytes())?;
let mut sign = open
.signing_card()
.ok_or_else(|| anyhow!("Unexpected: can't get card in signing mode"))?;
let mut signer = sign.signer(&|| println!("Touch confirmation needed for signing"))?;
op(&mut signer as &mut dyn sequoia_openpgp::crypto::Signer)?;
Ok(())
}
}
impl CardBackend {
pub(crate) fn new(ident: &str, pin: &str) -> Result<Self> {
Ok(Self {
pin: pin.to_string(),
ident: ident.to_string(),
card: Arc::new(Mutex::new(None)),
})
}
fn card(&self) -> Result<MutexGuard<'_, Option<Card<Open>>>> {
let mut card = self.card.try_lock().map_err(|e| {
anyhow::anyhow!(format!("Couldn't get lock for card in CardCa::card() {e}"))
})?;
if card.is_none() {
let backend = PcscBackend::open_by_ident(&self.ident, None)?;
let c: Card<Open> = backend.into();
*card = Some(c);
}
Ok(card)
}
pub(crate) fn ca_init(
db: &UninitDb,
domainname: &str,
card_ident: &str,
pin: &str,
pubkey: &str,
fingerprint: &str,
) -> Result<()> {
let backend = Backend::Card(backend::Card {
ident: card_ident.to_string(),
user_pin: pin.to_string(),
});
db.ca_insert(
domainname,
pubkey,
fingerprint,
backend.to_config().as_deref(),
)
}
pub(crate) fn ca_replace_in_place(
db: &UninitDb,
card_ident: &str,
pin: &str,
pubkey: &str,
) -> Result<()> {
let backend = Backend::Card(backend::Card {
ident: card_ident.to_string(),
user_pin: pin.to_string(),
});
let ca_new = Cert::from_str(pubkey)?;
let (_, mut cacert) = db.ca_cert()?;
if ca_new.fingerprint().to_string() != cacert.fingerprint {
return Err(anyhow::anyhow!(
"Can't replace CA cert, new fingerprint {} differs from existing fingerprint {}.",
ca_new.fingerprint(),
cacert.fingerprint
));
}
cacert.priv_cert = pubkey.to_string();
cacert.backend = backend.to_config();
db.cacert_update(&cacert)
}
}
const PW3_DEFAULT: &str = "12345678";
pub(crate) fn check_card_empty(open: &Card<Transaction>) -> Result<bool> {
let fps = open.fingerprints()?;
if fps.signature().is_some() || fps.decryption().is_some() || fps.authentication().is_some() {
Ok(false)
} else {
Ok(true)
}
}
pub(crate) fn generate_on_card(
ident: &str,
domain: &str,
user_id: String,
algo: Option<AlgoSimple>,
) -> Result<(Cert, String)> {
let backend = PcscBackend::open_by_ident(ident, None)?;
let mut card: Card<Open> = backend.into();
let mut transaction = card.transaction()?;
if !check_card_empty(&transaction)? {
return Err(anyhow!(
"The OpenPGP card contains key material, please reset it before use with OpenPGP CA."
));
}
let algo = match algo {
Some(a) => Some(a),
None => Some(AlgoSimple::RSA4k),
};
println!(
"Generating {}key material on the card, this might take a while.",
if let Some(algo) = algo {
format!("{algo:?} ")
} else {
"".to_string()
}
);
println!();
if transaction.verify_admin(PW3_DEFAULT.as_bytes()).is_err() {
return Err(anyhow!(
"Failed to get Admin access to the card with the default PIN."
));
};
if let Ok(mut admin) = transaction
.admin_card()
.ok_or_else(|| anyhow!("Couldn't get admin access"))
{
let (pkm, ts) = admin.generate_key_simple(KeyType::Signing, algo)?;
admin.set_name("OpenPGP CA")?;
let key_sig = public_key_material_to_key(&pkm, KeyType::Signing, &ts, None, None)?;
let (pkm, ts) = admin.generate_key_simple(KeyType::Authentication, algo)?;
let key_aut = public_key_material_to_key(&pkm, KeyType::Authentication, &ts, None, None)?;
let new_pin = set_user_and_admin_pin(&mut admin, PW3_DEFAULT)?;
let cert = {
let mut pp = vec![];
fn set_signer_metadata(sb: SignatureBuilder) -> Result<SignatureBuilder> {
sb.set_features(Features::sequoia())?
.set_preferred_hash_algorithms(vec![
HashAlgorithm::SHA512,
HashAlgorithm::SHA256,
])?
.set_preferred_symmetric_algorithms(vec![
SymmetricAlgorithm::AES256,
SymmetricAlgorithm::AES128,
])
}
fn certify_on_card(
user_pin: &str,
card: &mut Card<Transaction>,
auth_pubkey: PublicKey,
op: &mut dyn Fn(&mut dyn sequoia_openpgp::crypto::Signer) -> Result<Signature>,
) -> Result<Signature> {
let pw1 = user_pin.as_bytes();
card.verify_user(pw1)?;
if let Some(mut user) = card.user_card() {
let mut card_signer = user.authenticator_from_public(auth_pubkey, &|| {
println!("Need touch confirmation for certification.")
});
let s = op(&mut card_signer)?;
Ok(s)
} else {
Err(anyhow!("Failed to open card for certification"))
}
}
fn sign_on_card(
user_pin: &str,
card: &mut Card<Transaction>,
sig_pubkey: PublicKey,
op: &mut dyn Fn(&mut dyn sequoia_openpgp::crypto::Signer) -> Result<Signature>,
) -> Result<Signature> {
let pw1 = user_pin.as_bytes();
card.verify_user_for_signing(pw1)?;
if let Some(mut sign) = card.signing_card() {
let mut card_signer = sign.signer_from_public(sig_pubkey, &|| {
println!("Need touch confirmation for signing.")
});
let s = op(&mut card_signer)?;
Ok(s)
} else {
Err(anyhow!("Failed to open card for signing"))
}
}
let pri = PrimaryRole::convert_key(key_aut.clone());
pp.push(Packet::from(pri));
let s = certify_on_card(&new_pin, &mut transaction, key_aut.clone(), &mut |signer| {
let sb = SignatureBuilder::new(SignatureType::DirectKey).set_key_flags(
KeyFlags::empty().set_certification(),
)?;
let sb = set_signer_metadata(sb)?;
let sb = pgp::add_ca_domain_notation(sb, domain)?;
sb.sign_direct_key(signer, key_aut.role_as_primary())
})?;
pp.push(s.into());
let uid: UserID = user_id.into();
pp.push(uid.clone().into());
let cert = Cert::try_from(pp.clone())?;
let s = certify_on_card(&new_pin, &mut transaction, key_aut.clone(), &mut |signer| {
let sb = SignatureBuilder::new(SignatureType::PositiveCertification)
.set_key_flags(
KeyFlags::empty().set_certification(),
)?;
let sb = set_signer_metadata(sb)?;
let sb = pgp::add_ca_domain_notation(sb, domain)?;
uid.bind(signer, &cert, sb)
})?;
pp.push(s.into());
let cert = Cert::try_from(pp.clone())?;
let sub_sig = SubordinateRole::convert_key(key_sig.clone());
let bs = sign_on_card(&new_pin, &mut transaction, key_sig, &mut |signer| {
let sb = SignatureBuilder::new(SignatureType::PrimaryKeyBinding)
.set_hash_algo(HashAlgorithm::SHA512)
.set_reference_time(None);
sb.sign_primary_key_binding(&mut *signer, &cert.primary_key(), &sub_sig)
})?;
let cert = Cert::try_from(pp.clone())?;
pp.push(Packet::from(sub_sig.clone()));
let s = certify_on_card(&new_pin, &mut transaction, key_aut.clone(), &mut |signer| {
let sb = SignatureBuilder::new(SignatureType::SubkeyBinding)
.set_key_flags(KeyFlags::empty().set_signing())?
.set_embedded_signature(bs.clone())?;
sub_sig.bind(signer, &cert, sb)
})?;
pp.push(s.into());
Cert::try_from(pp)
}?;
Ok((cert, new_pin))
} else {
Err(anyhow!("Failed to open card in admin mode."))
}
}
pub(crate) fn import_to_card(ident: &str, key: &Cert) -> Result<String> {
let backend = PcscBackend::open_by_ident(ident, None)?;
let mut card: Card<Open> = backend.into();
let mut transaction = card.transaction()?;
if !check_card_empty(&transaction)? {
return Err(anyhow!(
"The OpenPGP card contains key material, please reset it before use with OpenPGP CA."
));
}
transaction.verify_admin(PW3_DEFAULT.as_bytes())?;
if let Ok(mut admin) = transaction
.admin_card()
.ok_or_else(|| anyhow!("Couldn't get admin access"))
{
let policy = StandardPolicy::new();
let mut certifier: Vec<_> = key
.keys()
.with_policy(&policy, None)
.secret()
.for_certification()
.collect();
match certifier.len() {
1 => {
let cert = certifier
.pop()
.expect("Certifier count matched len()==1, this should never happen");
let dec = sq_util::subkey_by_type(key, &policy, KeyType::Decryption)?;
let sig = sq_util::subkey_by_type(key, &policy, KeyType::Signing)?;
admin.upload_key(cert, KeyType::Authentication, None)?;
if let Some(dec) = dec {
admin.upload_key(dec, KeyType::Decryption, None)?;
}
if let Some(sig) = sig {
admin.upload_key(sig, KeyType::Signing, None)?;
}
admin.set_name("OpenPGP CA")?;
let new_pin = set_user_and_admin_pin(&mut admin, PW3_DEFAULT)?;
Ok(new_pin)
}
0 => Err(anyhow::anyhow!("No certification capable key found in key")),
_ => Err(anyhow::anyhow!(
"More than one certification capable key found in key"
)),
}
} else {
Err(anyhow!("Failed to open card in admin mode."))
}
}
fn set_user_and_admin_pin(card: &mut Card<Admin>, admin_pin: &str) -> Result<String> {
let new_pin = random_user_pin();
card.as_open()
.change_admin_pin(admin_pin.as_bytes(), new_pin.as_bytes())?;
card.as_open().verify_admin(new_pin.as_bytes())?;
card.reset_user_pin(new_pin.as_bytes())?;
Ok(new_pin)
}
fn random_user_pin() -> String {
use rand::Rng;
let mut rng = rand::thread_rng();
let i: u64 = rng.gen_range(0..=99_999_999);
format!("{i:08}")
}
pub(crate) fn verify_user_pin(ident: &str, pin: &str) -> Result<()> {
let backend = PcscBackend::open_by_ident(ident, None)?;
let mut card: Card<Open> = backend.into();
let mut transaction = card.transaction()?;
transaction.verify_user(pin.as_bytes())?;
Ok(())
}