#![allow(rustdoc::invalid_rust_codeblocks)]
use pyo3::prelude::*;
pub mod certificate;
pub mod crypto;
pub mod crypto_keys;
pub mod decoder;
pub mod encoder;
pub mod error;
pub mod ext_builders;
pub mod otp;
#[cfg(feature = "pkcs11-mgmt")]
pub mod pkcs11;
pub mod types;
pub mod x509_verification;
pub use certificate::*;
pub use decoder::*;
pub use encoder::*;
pub use error::*;
pub use types::*;
pub(crate) use synta_python_common::install_submodule;
#[pyclass(name = "Encoding")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PyEncoding {
DER,
BER,
CER,
}
impl From<PyEncoding> for synta::Encoding {
fn from(enc: PyEncoding) -> Self {
match enc {
PyEncoding::DER => synta::Encoding::Der,
PyEncoding::BER => synta::Encoding::Ber,
PyEncoding::CER => synta::Encoding::Cer,
}
}
}
impl From<synta::Encoding> for PyEncoding {
fn from(enc: synta::Encoding) -> Self {
match enc {
synta::Encoding::Der => PyEncoding::DER,
synta::Encoding::Ber => PyEncoding::BER,
synta::Encoding::Cer => PyEncoding::CER,
}
}
}
#[pymodule]
fn _synta(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<PyEncoding>()?;
m.add("SyntaError", py.get_type::<SyntaError>())?;
m.add_class::<PyDecoder>()?;
m.add_class::<PyEncoder>()?;
m.add_class::<PyInteger>()?;
m.add_class::<PyOctetString>()?;
m.add_class::<PyBitString>()?;
m.add_class::<PyBoolean>()?;
m.add_class::<PyReal>()?;
m.add_class::<PyUtcTime>()?;
m.add_class::<PyGeneralizedTime>()?;
m.add_class::<PyNull>()?;
m.add_class::<PyUtf8String>()?;
m.add_class::<PyPrintableString>()?;
m.add_class::<PyIA5String>()?;
m.add_class::<PyNumericString>()?;
m.add_class::<PyTeletexString>()?;
m.add_class::<PyVisibleString>()?;
m.add_class::<PyGeneralString>()?;
m.add_class::<PyUniversalString>()?;
m.add_class::<PyBmpString>()?;
m.add_class::<PyTaggedElement>()?;
m.add_class::<PyRawElement>()?;
certificate::register_module(m)?;
m.add_function(wrap_pyfunction!(pem_to_der, m)?)?;
m.add_function(wrap_pyfunction!(der_to_pem, m)?)?;
m.add_function(wrap_pyfunction!(parse_general_names, m)?)?;
m.add_function(wrap_pyfunction!(parse_name_attrs, m)?)?;
m.add_function(wrap_pyfunction!(encode_extended_key_usage, m)?)?;
m.add_function(wrap_pyfunction!(encode_subject_alt_names, m)?)?;
m.add_function(wrap_pyfunction!(name_der_equal, m)?)?;
m.add_function(wrap_pyfunction!(digest, m)?)?;
m.add_function(wrap_pyfunction!(format_dn, m)?)?;
m.add_function(wrap_pyfunction!(format_dn_slash, m)?)?;
m.add_function(wrap_pyfunction!(find_extension_value, m)?)?;
m.add_function(wrap_pyfunction!(encode_general_names, m)?)?;
m.add_function(wrap_pyfunction!(signing_algorithm_der, m)?)?;
m.add_function(wrap_pyfunction!(key_usage_bit, m)?)?;
m.add_function(wrap_pyfunction!(decode_public_key_info, m)?)?;
m.add_class::<crypto_keys::PyPublicKey>()?;
m.add_class::<crypto_keys::PyPrivateKey>()?;
crypto::register_crypto_module(m)?;
ext_builders::register_ext_module(m)?;
x509_verification::register_x509_module(m)?;
#[cfg(feature = "pkcs11-mgmt")]
pkcs11::register_pkcs11_module(m)?;
m.add("__version__", env!("CARGO_PKG_VERSION"))?;
Ok(())
}
#[pyfunction]
fn parse_general_names<'py>(
py: Python<'py>,
san_der: &[u8],
) -> PyResult<Bound<'py, pyo3::types::PyList>> {
use pyo3::types::{PyBytes, PyList, PyTuple};
let list = PyList::empty(py);
for (tag_num, content) in synta_certificate::parse_general_names(san_der) {
let tuple = PyTuple::new(
py,
[
tag_num.into_pyobject(py)?.into_any(),
PyBytes::new(py, &content).into_any(),
],
)?;
list.append(tuple)?;
}
Ok(list)
}
#[pyfunction]
fn parse_name_attrs<'py>(
py: Python<'py>,
name_der: &[u8],
) -> PyResult<Bound<'py, pyo3::types::PyList>> {
use pyo3::types::{PyList, PyTuple};
let attrs = synta_certificate::name::parse_name_attrs(name_der);
let list = PyList::empty(py);
for (oid, value) in attrs {
let tuple = PyTuple::new(
py,
[
oid.into_pyobject(py)?.into_any(),
value.into_pyobject(py)?.into_any(),
],
)?;
list.append(tuple)?;
}
Ok(list)
}
#[pyfunction]
fn der_to_pem<'py>(py: Python<'py>, der: &[u8], label: &str) -> Bound<'py, pyo3::types::PyBytes> {
pyo3::types::PyBytes::new(py, &synta_certificate::der_to_pem(label, der))
}
#[pyfunction]
fn pem_to_der<'py>(
py: Python<'py>,
data: &[u8],
) -> PyResult<pyo3::Bound<'py, pyo3::types::PyList>> {
let blocks = synta_certificate::pem_blocks(data);
if blocks.is_empty() {
return Err(pyo3::exceptions::PyValueError::new_err(
"no PEM block found in input",
));
}
let list = pyo3::types::PyList::empty(py);
for (_, block) in &blocks {
list.append(pyo3::types::PyBytes::new(py, block))?;
}
Ok(list)
}
#[pyfunction]
fn digest<'py>(
py: Python<'py>,
algorithm: &str,
data: &[u8],
) -> PyResult<pyo3::Bound<'py, pyo3::types::PyBytes>> {
use synta_certificate::{default_data_hasher, DataHasher};
let d = default_data_hasher()
.hash_data(algorithm, data)
.map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
Ok(pyo3::types::PyBytes::new(py, &d))
}
#[pyfunction]
fn format_dn(name_der: &[u8]) -> String {
synta_certificate::name::format_dn(name_der)
}
#[pyfunction]
fn format_dn_slash(name_der: &[u8]) -> String {
synta_certificate::name::format_dn_slash(name_der)
}
#[pyfunction]
fn find_extension_value<'py>(
py: Python<'py>,
ext_seq_der: &[u8],
oid: &Bound<'_, PyAny>,
) -> PyResult<Py<PyAny>> {
use std::str::FromStr;
use synta::ObjectIdentifier;
let oid_val: ObjectIdentifier =
if let Ok(oid_ref) = oid.extract::<pyo3::PyRef<crate::types::PyObjectIdentifier>>() {
oid_ref.inner.clone()
} else if let Ok(s) = oid.extract::<String>() {
ObjectIdentifier::from_str(&s)
.map_err(|_| pyo3::exceptions::PyValueError::new_err(format!("invalid OID: {s}")))?
} else {
return Err(pyo3::exceptions::PyTypeError::new_err(
"oid must be a str or ObjectIdentifier",
));
};
match synta_certificate::find_extension_value(ext_seq_der, oid_val.components()) {
Some(bytes) => Ok(pyo3::types::PyBytes::new(py, bytes).into_any().unbind()),
None => Ok(py.None()),
}
}
#[pyfunction]
fn encode_general_names<'py>(
py: Python<'py>,
entries: &Bound<'_, pyo3::types::PyList>,
) -> PyResult<Py<PyAny>> {
let mut rust_entries: Vec<(u32, Vec<u8>)> = Vec::with_capacity(entries.len());
for item in entries.iter() {
let tuple = item.cast::<pyo3::types::PyTuple>().map_err(|_| {
pyo3::exceptions::PyValueError::new_err("each entry must be a (int, bytes) tuple")
})?;
if tuple.len() != 2 {
return Err(pyo3::exceptions::PyValueError::new_err(
"each entry must be a 2-tuple (tag_number, bytes)",
));
}
let tag_num: u32 = tuple
.get_item(0)?
.extract()
.map_err(|_| pyo3::exceptions::PyValueError::new_err("tag_number must be an int"))?;
let value: Vec<u8> = tuple
.get_item(1)?
.extract()
.map_err(|_| pyo3::exceptions::PyValueError::new_err("value must be bytes"))?;
rust_entries.push((tag_num, value));
}
let refs: Vec<(u32, &[u8])> = rust_entries
.iter()
.map(|(t, v)| (*t, v.as_slice()))
.collect();
match synta_certificate::encode_general_names(&refs) {
Some(encoded) => Ok(pyo3::types::PyBytes::new(py, &encoded).into_any().unbind()),
None => Ok(py.None()),
}
}
#[pyfunction]
fn signing_algorithm_der<'py>(
py: Python<'py>,
key_oid: &Bound<'_, PyAny>,
hash_algo: &str,
) -> PyResult<Py<PyAny>> {
use std::str::FromStr;
use synta::ObjectIdentifier;
let oid_val: ObjectIdentifier =
if let Ok(oid_ref) = key_oid.extract::<pyo3::PyRef<crate::types::PyObjectIdentifier>>() {
oid_ref.inner.clone()
} else if let Ok(s) = key_oid.extract::<String>() {
ObjectIdentifier::from_str(&s)
.map_err(|_| pyo3::exceptions::PyValueError::new_err(format!("invalid OID: {s}")))?
} else {
return Err(pyo3::exceptions::PyTypeError::new_err(
"key_oid must be a str or ObjectIdentifier",
));
};
match synta_certificate::signing_algorithm_der(&oid_val, hash_algo) {
Some(der) => Ok(pyo3::types::PyBytes::new(py, &der).into_any().unbind()),
None => Ok(py.None()),
}
}
#[pyfunction]
fn key_usage_bit(ku_value_bytes: &[u8], bit_n: usize) -> PyResult<bool> {
let mut dec = synta::Decoder::new(ku_value_bytes, synta::Encoding::Der);
let ku: synta_certificate::KeyUsage = dec.decode().map_err(|e| {
pyo3::exceptions::PyValueError::new_err(format!("invalid KeyUsage DER: {e}"))
})?;
Ok(synta_certificate::key_usage_bit(&ku, bit_n))
}
#[pyfunction]
fn decode_public_key_info<'py>(
py: Python<'py>,
spki_der: &[u8],
) -> PyResult<Bound<'py, pyo3::types::PyDict>> {
use pyo3::types::{PyBytes, PyDict};
use synta::{Decoder, Encoding};
use synta_certificate::SubjectPublicKeyInfo;
let mut dec = Decoder::new(spki_der, Encoding::Der);
let spki: SubjectPublicKeyInfo<'_> = dec
.decode()
.map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("invalid SPKI DER: {e}")))?;
let alg_oid = spki
.algorithm
.algorithm
.components()
.iter()
.map(|n| n.to_string())
.collect::<Vec<_>>()
.join(".");
let key_bytes = spki.subject_public_key.as_bytes();
let key_bit_len = spki.subject_public_key.bit_len();
let info = synta_certificate::decode_public_key_info(
&spki.algorithm.algorithm,
spki.algorithm.parameters.as_ref(),
key_bytes,
key_bit_len,
);
let dict = PyDict::new(py);
dict.set_item("algorithm_oid", &alg_oid)?;
match info {
synta_certificate::PublicKeyInfo::Rsa {
modulus,
exponent,
bit_count,
} => {
dict.set_item("key_bytes", PyBytes::new(py, &modulus))?;
dict.set_item("modulus", PyBytes::new(py, &modulus))?;
dict.set_item("exponent", exponent)?;
dict.set_item("bit_count", bit_count)?;
}
synta_certificate::PublicKeyInfo::Ec {
key_bytes,
bit_count,
curve_short_name,
curve_nist_name,
curve_oid_str,
} => {
dict.set_item("key_bytes", PyBytes::new(py, &key_bytes))?;
dict.set_item("bit_count", bit_count)?;
dict.set_item("curve_oid", &curve_oid_str)?;
match curve_short_name {
Some(name) => dict.set_item("curve_short_name", name)?,
None => dict.set_item("curve_short_name", py.None())?,
}
match curve_nist_name {
Some(name) => dict.set_item("curve_nist_name", name)?,
None => dict.set_item("curve_nist_name", py.None())?,
}
}
synta_certificate::PublicKeyInfo::Unknown {
key_bytes,
bit_count,
..
} => {
dict.set_item("key_bytes", PyBytes::new(py, &key_bytes))?;
dict.set_item("bit_count", bit_count)?;
}
}
Ok(dict)
}