use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyBytes;
use synta_certificate::{BackendPrivateKey, BackendPublicKey, PrivateKey};
#[pyclass(frozen, name = "PublicKey")]
pub struct PyPublicKey {
pub(crate) inner: BackendPublicKey,
}
#[pymethods]
impl PyPublicKey {
#[staticmethod]
fn from_pem(data: &[u8]) -> PyResult<Self> {
let inner =
BackendPublicKey::from_pem(data).map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
fn from_rsa_components(n: &[u8], e: &[u8]) -> PyResult<Self> {
let inner = BackendPublicKey::from_rsa_components(n, e)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
fn from_ec_components(x: &[u8], y: &[u8], curve: &str) -> PyResult<Self> {
let inner = BackendPublicKey::from_ec_components(x, y, curve)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
fn from_der(data: &[u8]) -> PyResult<Self> {
let inner =
BackendPublicKey::from_der(data).map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
fn to_pem<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
let pem = self
.inner
.to_pem()
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &pem))
}
fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
let der = self
.inner
.to_der()
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &der))
}
#[getter]
fn key_type(&self) -> &'static str {
self.inner.key_type()
}
#[getter]
fn key_size(&self) -> Option<i64> {
self.inner.key_bit_size()
}
#[getter]
fn modulus<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
match self
.inner
.rsa_modulus()
.map_err(|e| PyValueError::new_err(format!("{e}")))?
{
Some(n) => Ok(Some(PyBytes::new(py, &n))),
None => Ok(None),
}
}
#[getter]
fn public_exponent<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
match self
.inner
.rsa_public_exponent()
.map_err(|e| PyValueError::new_err(format!("{e}")))?
{
Some(e) => Ok(Some(PyBytes::new(py, &e))),
None => Ok(None),
}
}
#[getter]
fn curve_name(&self) -> PyResult<Option<&'static str>> {
self.inner
.ec_curve_name()
.map_err(|e| PyValueError::new_err(format!("{e}")))
}
#[getter]
fn x<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
match self
.inner
.ec_affine_coordinates()
.map_err(|e| PyValueError::new_err(format!("{e}")))?
{
Some((xv, _)) => Ok(Some(PyBytes::new(py, &xv))),
None => Ok(None),
}
}
#[getter]
fn y<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
match self
.inner
.ec_affine_coordinates()
.map_err(|e| PyValueError::new_err(format!("{e}")))?
{
Some((_, yv)) => Ok(Some(PyBytes::new(py, &yv))),
None => Ok(None),
}
}
#[pyo3(signature = (plaintext, hash_algorithm = "sha256"))]
fn rsa_oaep_encrypt<'py>(
&self,
py: Python<'py>,
plaintext: &[u8],
hash_algorithm: &str,
) -> PyResult<Bound<'py, PyBytes>> {
let ct = self
.inner
.rsa_oaep_encrypt(plaintext, hash_algorithm)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &ct))
}
fn rsa_pkcs1v15_encrypt<'py>(
&self,
py: Python<'py>,
plaintext: &[u8],
) -> PyResult<Bound<'py, PyBytes>> {
let ct = self
.inner
.rsa_pkcs1v15_encrypt(plaintext)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &ct))
}
#[pyo3(signature = (signature, data, algorithm = None, context = None))]
fn verify_signature(
&self,
signature: &[u8],
data: &[u8],
algorithm: Option<&str>,
context: Option<&[u8]>,
) -> PyResult<()> {
let kt = self.inner.key_type();
if matches!(kt, "ml-dsa-44" | "ml-dsa-65" | "ml-dsa-87") {
self.inner
.verify_ml_dsa_with_context(data, signature, context.unwrap_or(b""))
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
return Ok(());
}
self.inner
.verify_message(data, signature, algorithm)
.map_err(|e| PyValueError::new_err(format!("{e}")))
}
fn verify_certificate_signature(
&self,
tbs_der: &[u8],
sig_alg_der: &[u8],
signature: &[u8],
) -> PyResult<()> {
self.inner
.verify_signature(tbs_der, sig_alg_der, signature)
.map_err(|e| PyValueError::new_err(format!("{e}")))
}
fn kem_encapsulate<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, pyo3::types::PyTuple>> {
use pyo3::types::PyTuple;
let (ct, ss) = self
.inner
.ml_kem_encapsulate()
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
let items = [PyBytes::new(py, &ct), PyBytes::new(py, &ss)];
PyTuple::new(py, items)
}
fn __repr__(&self) -> String {
let kt = self.inner.key_type();
let bits = match kt {
"ed25519" | "ed448" | "ml-dsa-44" | "ml-dsa-65" | "ml-dsa-87" | "ml-kem-512"
| "ml-kem-768" | "ml-kem-1024" => String::new(),
_ => self
.inner
.key_bit_size()
.map(|b| format!(", key_size={b}"))
.unwrap_or_default(),
};
format!("PublicKey(key_type={kt:?}{bits})")
}
}
#[pyclass(frozen, name = "PrivateKey")]
pub struct PyPrivateKey {
pub(crate) inner: BackendPrivateKey,
}
#[pymethods]
impl PyPrivateKey {
#[staticmethod]
#[pyo3(signature = (data, password = None))]
fn from_pem(data: &[u8], password: Option<&[u8]>) -> PyResult<Self> {
let inner = BackendPrivateKey::from_pem(data, password)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
fn from_der(data: &[u8]) -> PyResult<Self> {
let inner =
BackendPrivateKey::from_der(data).map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[pyo3(signature = (password = None))]
fn to_pem<'py>(
&self,
py: Python<'py>,
password: Option<&[u8]>,
) -> PyResult<Bound<'py, PyBytes>> {
let pem = self
.inner
.to_pem(password)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &pem))
}
fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
let der = self
.inner
.to_der()
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &der))
}
fn to_pkcs8_encrypted<'py>(
&self,
py: Python<'py>,
password: &[u8],
) -> PyResult<Bound<'py, PyBytes>> {
let der = self
.inner
.to_pkcs8_encrypted(password)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &der))
}
#[staticmethod]
fn from_pkcs8_encrypted(data: &[u8], password: &[u8]) -> PyResult<Self> {
let inner = BackendPrivateKey::from_pkcs8_encrypted(data, password)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
#[cfg(any(feature = "openssl", feature = "nss"))]
fn from_pkcs11_uri(uri: &str) -> PyResult<Self> {
let inner = BackendPrivateKey::from_pkcs11_uri(uri)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[getter]
fn key_type(&self) -> &'static str {
self.inner.key_type()
}
#[getter]
fn key_size(&self) -> Option<i64> {
self.inner.key_bit_size()
}
#[getter]
fn public_key(&self) -> PyResult<PyPublicKey> {
let bpk = self
.inner
.public_key()
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyPublicKey { inner: bpk })
}
#[pyo3(signature = (ciphertext, hash_algorithm = "sha256"))]
fn rsa_oaep_decrypt<'py>(
&self,
py: Python<'py>,
ciphertext: &[u8],
hash_algorithm: &str,
) -> PyResult<Bound<'py, PyBytes>> {
let pt = self
.inner
.rsa_oaep_decrypt(ciphertext, hash_algorithm)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &pt))
}
fn rsa_pkcs1v15_decrypt<'py>(
&self,
py: Python<'py>,
ciphertext: &[u8],
) -> PyResult<Bound<'py, PyBytes>> {
let pt = self
.inner
.rsa_pkcs1v15_decrypt(ciphertext)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &pt))
}
#[staticmethod]
#[pyo3(signature = (key_size, public_exponent = 65537))]
fn generate_rsa(key_size: u32, public_exponent: u32) -> PyResult<Self> {
let inner = BackendPrivateKey::generate_rsa(key_size, public_exponent)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
#[pyo3(signature = (curve = "P-256"))]
fn generate_ec(curve: &str) -> PyResult<Self> {
let inner = BackendPrivateKey::generate_ec(curve)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
fn generate_ed25519() -> PyResult<Self> {
let inner = BackendPrivateKey::generate_ed25519()
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
fn generate_ed448() -> PyResult<Self> {
let inner = BackendPrivateKey::generate_ed448()
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
fn generate_ml_dsa(parameter_set: &str) -> PyResult<Self> {
let inner = BackendPrivateKey::generate_ml_dsa(parameter_set)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
fn generate_composite_ml_dsa(sub_arc: u32) -> PyResult<Self> {
let inner = synta_certificate::BackendPrivateKey::generate_composite_ml_dsa(sub_arc)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
#[staticmethod]
fn generate_ml_kem(parameter_set: &str) -> PyResult<Self> {
let inner = BackendPrivateKey::generate_ml_kem(parameter_set)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(Self { inner })
}
fn kem_decapsulate<'py>(
&self,
py: Python<'py>,
ciphertext: &[u8],
) -> PyResult<Bound<'py, PyBytes>> {
let ss = self
.inner
.ml_kem_decapsulate(ciphertext)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &ss))
}
#[pyo3(signature = (data, algorithm = None, context = None))]
fn sign<'py>(
&self,
py: Python<'py>,
data: &[u8],
algorithm: Option<&str>,
context: Option<&[u8]>,
) -> PyResult<Bound<'py, PyBytes>> {
let kt = self.inner.key_type();
if matches!(kt, "ml-dsa-44" | "ml-dsa-65" | "ml-dsa-87") {
let sig = self
.inner
.sign_ml_dsa_with_context(data, context.unwrap_or(b""))
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
return Ok(PyBytes::new(py, &sig));
}
let alg = algorithm.unwrap_or("sha256");
let signer = self.inner.as_signer(alg);
let sig = signer
.sign_tbs_erased(data)
.map_err(|e| PyValueError::new_err(format!("{e}")))?;
Ok(PyBytes::new(py, &sig))
}
fn __repr__(&self) -> String {
let kt = self.inner.key_type();
let bits = match kt {
"ed25519" | "ed448" | "ml-dsa-44" | "ml-dsa-65" | "ml-dsa-87" | "ml-kem-512"
| "ml-kem-768" | "ml-kem-1024" => String::new(),
_ => self
.inner
.key_bit_size()
.map(|b| format!(", key_size={b}"))
.unwrap_or_default(),
};
format!("PrivateKey(key_type={kt:?}{bits})")
}
}
impl synta_certificate::PrivateKey for PyPrivateKey {
fn public_key_spki_der(&self) -> Result<Vec<u8>, synta_certificate::PrivateKeyError> {
self.inner.public_key_spki_der()
}
fn as_signer(&self, algorithm: &str) -> Box<dyn synta_certificate::ErasedCertificateSigner> {
self.inner.as_signer(algorithm)
}
}