use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyBytes;
use synta_certificate::{
default_key_id_hasher, encode_authority_key_identifier, encode_basic_constraints,
encode_key_usage, encode_subject_key_identifier, AuthorityInformationAccessBuilder,
ExtendedKeyUsageBuilder, KeyIdMethod, SubjectAlternativeNameBuilder,
};
pub const KEYID_RFC5280: u8 = 0;
pub const KEYID_RFC7093M1: u8 = 1;
pub const KEYID_RFC7093M2: u8 = 2;
pub const KEYID_RFC7093M3: u8 = 3;
pub const KEYID_RFC7093M4: u8 = 4;
pub const KU_DIGITAL_SIGNATURE: u16 = 1 << 0;
pub const KU_NON_REPUDIATION: u16 = 1 << 1;
pub const KU_KEY_ENCIPHERMENT: u16 = 1 << 2;
pub const KU_DATA_ENCIPHERMENT: u16 = 1 << 3;
pub const KU_KEY_AGREEMENT: u16 = 1 << 4;
pub const KU_KEY_CERT_SIGN: u16 = 1 << 5;
pub const KU_CRL_SIGN: u16 = 1 << 6;
pub const KU_ENCIPHER_ONLY: u16 = 1 << 7;
pub const KU_DECIPHER_ONLY: u16 = 1 << 8;
fn method_from_int(m: u8) -> PyResult<KeyIdMethod> {
match m {
KEYID_RFC5280 => Ok(KeyIdMethod::Rfc5280Sha1),
KEYID_RFC7093M1 => Ok(KeyIdMethod::Rfc7093Method1Sha256),
KEYID_RFC7093M2 => Ok(KeyIdMethod::Rfc7093Method2Sha384),
KEYID_RFC7093M3 => Ok(KeyIdMethod::Rfc7093Method3Sha512),
KEYID_RFC7093M4 => Ok(KeyIdMethod::Rfc7093Method4 {
algorithm_oid: synta_certificate::ID_SHA256.to_vec(),
}),
other => Err(PyValueError::new_err(format!(
"unknown key identifier method {other}: \
use KEYID_RFC5280, KEYID_RFC7093M1, KEYID_RFC7093M2, \
KEYID_RFC7093M3, or KEYID_RFC7093M4"
))),
}
}
#[pyfunction]
#[pyo3(signature = (ca = false, path_length = None))]
fn basic_constraints<'py>(
py: Python<'py>,
ca: bool,
path_length: Option<u64>,
) -> PyResult<Bound<'py, PyBytes>> {
let der = encode_basic_constraints(ca, path_length)
.ok_or_else(|| PyValueError::new_err("BasicConstraints DER encoding failed"))?;
Ok(PyBytes::new(py, &der))
}
#[pyfunction]
fn key_usage<'py>(py: Python<'py>, bits: u16) -> PyResult<Bound<'py, PyBytes>> {
let der = encode_key_usage(bits)
.ok_or_else(|| PyValueError::new_err("KeyUsage DER encoding failed"))?;
Ok(PyBytes::new(py, &der))
}
#[pyfunction]
#[pyo3(signature = (spki_der, method = KEYID_RFC5280))]
fn subject_key_identifier<'py>(
py: Python<'py>,
spki_der: &[u8],
method: u8,
) -> PyResult<Bound<'py, PyBytes>> {
let m = method_from_int(method)?;
encode_subject_key_identifier(spki_der, m, &default_key_id_hasher())
.map(|v| PyBytes::new(py, &v))
.ok_or_else(|| {
PyValueError::new_err(
"failed to encode SubjectKeyIdentifier: \
invalid SubjectPublicKeyInfo DER or hash error",
)
})
}
#[pyfunction]
#[pyo3(signature = (issuer_spki_der, method = KEYID_RFC5280))]
fn authority_key_identifier<'py>(
py: Python<'py>,
issuer_spki_der: &[u8],
method: u8,
) -> PyResult<Bound<'py, PyBytes>> {
let m = method_from_int(method)?;
encode_authority_key_identifier(issuer_spki_der, m, &default_key_id_hasher())
.map(|v| PyBytes::new(py, &v))
.ok_or_else(|| {
PyValueError::new_err(
"failed to encode AuthorityKeyIdentifier: \
invalid SubjectPublicKeyInfo DER or hash error",
)
})
}
#[pyclass(name = "SubjectAlternativeNameBuilder")]
pub struct PySubjectAlternativeNameBuilder {
inner: SubjectAlternativeNameBuilder,
built: bool,
}
#[pymethods]
impl PySubjectAlternativeNameBuilder {
#[new]
fn new() -> Self {
Self {
inner: SubjectAlternativeNameBuilder::new(),
built: false,
}
}
fn dns_name<'py>(slf: Bound<'py, Self>, name: &str) -> Bound<'py, Self> {
let old = std::mem::replace(
&mut slf.borrow_mut().inner,
SubjectAlternativeNameBuilder::new(),
);
slf.borrow_mut().inner = old.dns_name(name);
slf
}
fn rfc822_name<'py>(slf: Bound<'py, Self>, email: &str) -> Bound<'py, Self> {
let old = std::mem::replace(
&mut slf.borrow_mut().inner,
SubjectAlternativeNameBuilder::new(),
);
slf.borrow_mut().inner = old.rfc822_name(email);
slf
}
fn uri<'py>(slf: Bound<'py, Self>, uri: &str) -> Bound<'py, Self> {
let old = std::mem::replace(
&mut slf.borrow_mut().inner,
SubjectAlternativeNameBuilder::new(),
);
slf.borrow_mut().inner = old.uri(uri);
slf
}
fn ip_address<'py>(slf: Bound<'py, Self>, addr: &[u8]) -> Bound<'py, Self> {
let old = std::mem::replace(
&mut slf.borrow_mut().inner,
SubjectAlternativeNameBuilder::new(),
);
slf.borrow_mut().inner = old.ip_address(addr);
slf
}
fn directory_name<'py>(slf: Bound<'py, Self>, name_der: &[u8]) -> Bound<'py, Self> {
let old = std::mem::replace(
&mut slf.borrow_mut().inner,
SubjectAlternativeNameBuilder::new(),
);
slf.borrow_mut().inner = old.directory_name(name_der);
slf
}
fn build<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
if self.built {
return Err(PyValueError::new_err(
"SubjectAlternativeNameBuilder.build() has already been called; \
create a new builder instance",
));
}
self.built = true;
let builder = std::mem::replace(&mut self.inner, SubjectAlternativeNameBuilder::new());
let der = builder.build().map_err(PyValueError::new_err)?;
Ok(PyBytes::new(py, &der))
}
}
#[pyclass(name = "AuthorityInformationAccessBuilder")]
pub struct PyAuthorityInformationAccessBuilder {
inner: AuthorityInformationAccessBuilder,
built: bool,
}
#[pymethods]
impl PyAuthorityInformationAccessBuilder {
#[new]
fn new() -> Self {
Self {
inner: AuthorityInformationAccessBuilder::new(),
built: false,
}
}
fn ocsp<'py>(slf: Bound<'py, Self>, uri: &str) -> Bound<'py, Self> {
let old = std::mem::replace(
&mut slf.borrow_mut().inner,
AuthorityInformationAccessBuilder::new(),
);
slf.borrow_mut().inner = old.ocsp(uri);
slf
}
fn ca_issuers<'py>(slf: Bound<'py, Self>, uri: &str) -> Bound<'py, Self> {
let old = std::mem::replace(
&mut slf.borrow_mut().inner,
AuthorityInformationAccessBuilder::new(),
);
slf.borrow_mut().inner = old.ca_issuers(uri);
slf
}
fn build<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
if self.built {
return Err(PyValueError::new_err(
"AuthorityInformationAccessBuilder.build() has already been called; \
create a new builder instance",
));
}
self.built = true;
let builder = std::mem::replace(&mut self.inner, AuthorityInformationAccessBuilder::new());
let der = builder.build().map_err(PyValueError::new_err)?;
Ok(PyBytes::new(py, &der))
}
}
#[pyclass(name = "ExtendedKeyUsageBuilder")]
pub struct PyExtendedKeyUsageBuilder {
inner: ExtendedKeyUsageBuilder,
built: bool,
}
#[pymethods]
impl PyExtendedKeyUsageBuilder {
#[new]
fn new() -> Self {
Self {
inner: ExtendedKeyUsageBuilder::new(),
built: false,
}
}
fn server_auth<'py>(slf: Bound<'py, Self>) -> Bound<'py, Self> {
let old = std::mem::replace(&mut slf.borrow_mut().inner, ExtendedKeyUsageBuilder::new());
slf.borrow_mut().inner = old.server_auth();
slf
}
fn client_auth<'py>(slf: Bound<'py, Self>) -> Bound<'py, Self> {
let old = std::mem::replace(&mut slf.borrow_mut().inner, ExtendedKeyUsageBuilder::new());
slf.borrow_mut().inner = old.client_auth();
slf
}
fn code_signing<'py>(slf: Bound<'py, Self>) -> Bound<'py, Self> {
let old = std::mem::replace(&mut slf.borrow_mut().inner, ExtendedKeyUsageBuilder::new());
slf.borrow_mut().inner = old.code_signing();
slf
}
fn email_protection<'py>(slf: Bound<'py, Self>) -> Bound<'py, Self> {
let old = std::mem::replace(&mut slf.borrow_mut().inner, ExtendedKeyUsageBuilder::new());
slf.borrow_mut().inner = old.email_protection();
slf
}
fn time_stamping<'py>(slf: Bound<'py, Self>) -> Bound<'py, Self> {
let old = std::mem::replace(&mut slf.borrow_mut().inner, ExtendedKeyUsageBuilder::new());
slf.borrow_mut().inner = old.time_stamping();
slf
}
fn ocsp_signing<'py>(slf: Bound<'py, Self>) -> Bound<'py, Self> {
let old = std::mem::replace(&mut slf.borrow_mut().inner, ExtendedKeyUsageBuilder::new());
slf.borrow_mut().inner = old.ocsp_signing();
slf
}
fn add_oid<'py>(slf: Bound<'py, Self>, comps: Vec<u32>) -> Bound<'py, Self> {
let old = std::mem::replace(&mut slf.borrow_mut().inner, ExtendedKeyUsageBuilder::new());
slf.borrow_mut().inner = old.add_oid(&comps);
slf
}
fn build<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
if self.built {
return Err(PyValueError::new_err(
"ExtendedKeyUsageBuilder.build() has already been called; \
create a new builder instance",
));
}
self.built = true;
let builder = std::mem::replace(&mut self.inner, ExtendedKeyUsageBuilder::new());
let der = builder.build().map_err(PyValueError::new_err)?;
Ok(PyBytes::new(py, &der))
}
}
pub fn register_ext_module(parent: &Bound<'_, PyModule>) -> PyResult<()> {
let py = parent.py();
let m = PyModule::new(py, "ext")?;
m.add("KEYID_RFC5280", KEYID_RFC5280)?;
m.add("KEYID_RFC7093M1", KEYID_RFC7093M1)?;
m.add("KEYID_RFC7093M2", KEYID_RFC7093M2)?;
m.add("KEYID_RFC7093M3", KEYID_RFC7093M3)?;
m.add("KEYID_RFC7093M4", KEYID_RFC7093M4)?;
m.add("KU_DIGITAL_SIGNATURE", KU_DIGITAL_SIGNATURE)?;
m.add("KU_NON_REPUDIATION", KU_NON_REPUDIATION)?;
m.add("KU_KEY_ENCIPHERMENT", KU_KEY_ENCIPHERMENT)?;
m.add("KU_DATA_ENCIPHERMENT", KU_DATA_ENCIPHERMENT)?;
m.add("KU_KEY_AGREEMENT", KU_KEY_AGREEMENT)?;
m.add("KU_KEY_CERT_SIGN", KU_KEY_CERT_SIGN)?;
m.add("KU_CRL_SIGN", KU_CRL_SIGN)?;
m.add("KU_ENCIPHER_ONLY", KU_ENCIPHER_ONLY)?;
m.add("KU_DECIPHER_ONLY", KU_DECIPHER_ONLY)?;
m.add_function(wrap_pyfunction!(basic_constraints, &m)?)?;
m.add_function(wrap_pyfunction!(key_usage, &m)?)?;
m.add_function(wrap_pyfunction!(subject_key_identifier, &m)?)?;
m.add_function(wrap_pyfunction!(authority_key_identifier, &m)?)?;
m.add_class::<PySubjectAlternativeNameBuilder>()?;
m.add_class::<PyAuthorityInformationAccessBuilder>()?;
m.add_class::<PyExtendedKeyUsageBuilder>()?;
m.add("SAN", m.getattr("SubjectAlternativeNameBuilder")?)?;
m.add("AIA", m.getattr("AuthorityInformationAccessBuilder")?)?;
m.add("EKU", m.getattr("ExtendedKeyUsageBuilder")?)?;
crate::install_submodule(
parent,
&m,
"synta.ext",
Some(
"X.509 v3 extension value DER builders: basic_constraints, key_usage, \
subject_key_identifier, authority_key_identifier; \
SubjectAlternativeNameBuilder (alias: SAN), \
AuthorityInformationAccessBuilder (alias: AIA), \
ExtendedKeyUsageBuilder (alias: EKU); \
KEYID_* method constants (RFC 5280 §4.2.1.2 / RFC 7093); \
KU_* key-usage bitmask constants (RFC 5280 §4.2.1.3).",
),
)
}