_synta/

crypto_keys.rs

//! Python bindings for generic public and private key operations.
//!
//! Exposes [`PyPublicKey`] and [`PyPrivateKey`] as pyo3 classes supporting
//! RSA, EC, EdDSA, and DSA keys via the `synta-certificate` backend traits.
//! No direct `openssl::*` imports are used here.

use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyBytes;
use synta_certificate::{BackendPrivateKey, BackendPublicKey, KeyDecryptor, KeyEncryptor};

// ── PublicKey ─────────────────────────────────────────────────────────────────

/// An asymmetric public key.
///
/// Supports RSA, EC (P-256, P-384, P-521), Ed25519, Ed448, and DSA keys.
/// Load from PEM or SubjectPublicKeyInfo DER; serialize back to PEM or DER.
/// RSA keys can encrypt data with OAEP or PKCS\#1 v1.5 padding.
///
/// ```python,ignore
/// import synta
///
/// # Load an RSA public key from a PEM file:
/// with open("rsa_pub.pem", "rb") as f:
///     pub = synta.PublicKey.from_pem(f.read())
/// print(pub.key_type)   # "rsa"
/// print(pub.key_size)   # e.g. 2048
///
/// # Encrypt with OAEP (SHA-256):
/// ct = pub.rsa_oaep_encrypt(b"secret", "sha256")
///
/// # Load an EC public key from SPKI DER:
/// ec_pub = synta.PublicKey.from_der(spki_der)
/// print(ec_pub.curve_name)  # "P-256"
/// ```
#[pyclass(frozen, name = "PublicKey")]
pub struct PyPublicKey {
    pub(crate) inner: BackendPublicKey,
}

#[pymethods]
impl PyPublicKey {
    /// Load a public key from PEM-encoded SubjectPublicKeyInfo data.
    ///
    /// Supports RSA, EC (P-256, P-384, P-521), Ed25519, Ed448, and DSA keys.
    ///
    /// ```python,ignore
    /// with open("pubkey.pem", "rb") as f:
    ///     pub = synta.PublicKey.from_pem(f.read())
    /// ```
    #[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 })
    }

    /// Load a public key from a DER-encoded SubjectPublicKeyInfo structure.
    ///
    /// ```python,ignore
    /// with open("pubkey.der", "rb") as f:
    ///     pub = synta.PublicKey.from_der(f.read())
    /// ```
    #[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 })
    }

    /// Serialize this public key to PEM-encoded SubjectPublicKeyInfo.
    ///
    /// ```python,ignore
    /// pem = pub.to_pem()
    /// open("pubkey.pem", "wb").write(pem)
    /// ```
    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))
    }

    /// Serialize this public key to DER-encoded SubjectPublicKeyInfo.
    ///
    /// ```python,ignore
    /// der = pub.to_der()
    /// open("pubkey.der", "wb").write(der)
    /// ```
    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))
    }

    /// The key algorithm as a lowercase string.
    ///
    /// Returns one of ``"rsa"``, ``"ec"``, ``"ed25519"``, ``"ed448"``,
    /// ``"dsa"``, or ``"unknown"``.
    #[getter]
    fn key_type(&self) -> &'static str {
        self.inner.key_type()
    }

    /// The key size in bits, or ``None`` for EdDSA keys.
    ///
    /// For RSA this is the modulus bit-length; for EC this is the field
    /// bit-length.  Returns ``None`` for Ed25519 and Ed448.
    #[getter]
    fn key_size(&self) -> Option<i64> {
        self.inner.key_bit_size()
    }

    /// The RSA modulus ``n`` as big-endian bytes, or ``None`` for non-RSA keys.
    #[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),
        }
    }

    /// The RSA public exponent ``e`` as big-endian bytes, or ``None`` for
    /// non-RSA keys.
    ///
    /// The most common value is ``b'\x01\x00\x01'`` (65537).
    #[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),
        }
    }

    /// The NIST curve name for EC keys, or ``None`` for non-EC keys.
    ///
    /// Returns ``"P-256"``, ``"P-384"``, ``"P-521"``, or ``"unknown"`` for
    /// EC keys on other curves.
    #[getter]
    fn curve_name(&self) -> PyResult<Option<&'static str>> {
        self.inner
            .ec_curve_name()
            .map_err(|e| PyValueError::new_err(format!("{e}")))
    }

    /// The affine X coordinate of the EC public key as big-endian bytes, or
    /// ``None`` for non-EC keys.
    #[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),
        }
    }

    /// The affine Y coordinate of the EC public key as big-endian bytes, or
    /// ``None`` for non-EC keys.
    #[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),
        }
    }

    /// Encrypt ``plaintext`` with RSA-OAEP using the specified hash algorithm.
    ///
    /// ``hash_algorithm`` must be one of ``"sha1"``, ``"sha224"``,
    /// ``"sha256"``, ``"sha384"``, or ``"sha512"``.
    ///
    /// Raises :exc:`ValueError` if this key is not an RSA key.
    ///
    /// ```python,ignore
    /// ct = pub.rsa_oaep_encrypt(b"secret data", "sha256")
    /// ```
    #[pyo3(signature = (plaintext, hash_algorithm = "sha256"))]
    fn rsa_oaep_encrypt<'py>(
        &self,
        py: Python<'py>,
        plaintext: &[u8],
        hash_algorithm: &str,
    ) -> PyResult<Bound<'py, PyBytes>> {
        use synta_certificate::OpensslRsaOaepEncryptor;
        let ct = OpensslRsaOaepEncryptor::new(self.inner.spki_der(), hash_algorithm)
            .map_err(|e| PyValueError::new_err(format!("{e}")))?
            .encrypt_key(plaintext)
            .map_err(|e| PyValueError::new_err(format!("{e}")))?;
        Ok(PyBytes::new(py, &ct))
    }

    /// Encrypt ``plaintext`` with RSA PKCS\#1 v1.5 padding.
    ///
    /// Raises :exc:`ValueError` if this key is not an RSA key.
    ///
    /// ```python,ignore
    /// ct = pub.rsa_pkcs1v15_encrypt(b"secret data")
    /// ```
    fn rsa_pkcs1v15_encrypt<'py>(
        &self,
        py: Python<'py>,
        plaintext: &[u8],
    ) -> PyResult<Bound<'py, PyBytes>> {
        use synta_certificate::OpensslRsaPkcs1Encryptor;
        let ct = OpensslRsaPkcs1Encryptor::new(self.inner.spki_der())
            .map_err(|e| PyValueError::new_err(format!("{e}")))?
            .encrypt_key(plaintext)
            .map_err(|e| PyValueError::new_err(format!("{e}")))?;
        Ok(PyBytes::new(py, &ct))
    }

    /// Verify a signature over ``data``.
    ///
    /// ``algorithm`` is the hash algorithm used during signing.  It must be one
    /// of ``"sha1"``, ``"sha224"``, ``"sha256"``, ``"sha384"``, or
    /// ``"sha512"`` for RSA (PKCS\#1 v1.5) and ECDSA keys.  For Ed25519 and
    /// Ed448 keys pass ``None`` (or omit the argument) — no pre-hash is used.
    ///
    /// Raises :exc:`ValueError` if the signature is invalid or the algorithm
    /// combination is unsupported.
    ///
    /// ```python,ignore
    /// pub.verify_signature(sig, data, "sha256")   # RSA or ECDSA
    /// ed_pub.verify_signature(sig, data)          # Ed25519 / Ed448
    /// ```
    #[pyo3(signature = (signature, data, algorithm = None))]
    fn verify_signature(
        &self,
        signature: &[u8],
        data: &[u8],
        algorithm: Option<&str>,
    ) -> PyResult<()> {
        self.inner
            .verify_message(data, signature, algorithm)
            .map_err(|e| PyValueError::new_err(format!("{e}")))
    }

    fn __repr__(&self) -> String {
        let kt = self.inner.key_type();
        let bits = match kt {
            "ed25519" | "ed448" => String::new(),
            _ => self
                .inner
                .key_bit_size()
                .map(|b| format!(", key_size={b}"))
                .unwrap_or_default(),
        };
        format!("PublicKey(key_type={kt:?}{bits})")
    }
}

// ── PrivateKey ────────────────────────────────────────────────────────────────

/// An asymmetric private key.
///
/// Supports RSA, EC (P-256, P-384, P-521), Ed25519, Ed448, and DSA keys.
/// Load from PEM (optionally password-protected) or unencrypted PKCS\#8 DER;
/// serialize back to PEM (optionally encrypted with AES-256-CBC) or
/// unencrypted PKCS\#8 DER.  RSA keys can decrypt ciphertext with OAEP or
/// PKCS\#1 v1.5 padding.
///
/// ```python,ignore
/// import synta
///
/// # Load an encrypted RSA private key from PEM:
/// with open("rsa_key.pem", "rb") as f:
///     priv = synta.PrivateKey.from_pem(f.read(), password=b"secret")
///
/// # Extract the public key:
/// pub = priv.public_key
///
/// # Decrypt RSA-OAEP ciphertext:
/// plaintext = priv.rsa_oaep_decrypt(ciphertext, "sha256")
/// ```
#[pyclass(frozen, name = "PrivateKey")]
pub struct PyPrivateKey {
    pub(crate) inner: BackendPrivateKey,
}

#[pymethods]
impl PyPrivateKey {
    /// Load a private key from PEM-encoded data.
    ///
    /// Supports RSA, EC, Ed25519, Ed448, and DSA keys in both PKCS\#8
    /// (``-----BEGIN PRIVATE KEY-----``) and traditional
    /// (``-----BEGIN RSA PRIVATE KEY-----`` etc.) PEM formats.
    ///
    /// If the PEM block is password-protected, pass the passphrase as
    /// ``password``.
    ///
    /// ```python,ignore
    /// # Unencrypted key:
    /// priv = synta.PrivateKey.from_pem(open("key.pem", "rb").read())
    ///
    /// # Encrypted key:
    /// priv = synta.PrivateKey.from_pem(open("key.pem", "rb").read(), password=b"pass")
    /// ```
    #[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 })
    }

    /// Load an unencrypted private key from PKCS\#8 DER bytes.
    ///
    /// ```python,ignore
    /// with open("key.der", "rb") as f:
    ///     priv = synta.PrivateKey.from_der(f.read())
    /// ```
    #[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 })
    }

    /// Serialize this private key to PEM-encoded PKCS\#8.
    ///
    /// If ``password`` is provided the output is encrypted with AES-256-CBC.
    ///
    /// ```python,ignore
    /// # Unencrypted:
    /// pem = priv.to_pem()
    ///
    /// # Encrypted:
    /// pem = priv.to_pem(password=b"my-passphrase")
    /// ```
    #[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))
    }

    /// Serialize this private key to unencrypted PKCS\#8 DER.
    ///
    /// ```python,ignore
    /// der = priv.to_der()
    /// open("key.der", "wb").write(der)
    /// ```
    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))
    }

    /// Serialize this private key to encrypted PKCS\#8 DER
    /// (``EncryptedPrivateKeyInfo``, RFC 5958 §3).
    ///
    /// ```python,ignore
    /// der = priv.to_pkcs8_encrypted(b"my-passphrase")
    /// open("key.p8e", "wb").write(der)
    ///
    /// # Round-trip:
    /// priv2 = synta.PrivateKey.from_pkcs8_encrypted(der, b"my-passphrase")
    /// assert priv2.to_der() == priv.to_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))
    }

    /// Load a private key from an encrypted PKCS\#8 DER blob
    /// (``EncryptedPrivateKeyInfo``).
    ///
    /// ```python,ignore
    /// der = open("key.p8e", "rb").read()
    /// priv = synta.PrivateKey.from_pkcs8_encrypted(der, b"my-passphrase")
    /// ```
    #[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 })
    }

    /// The key algorithm as a lowercase string.
    ///
    /// Returns one of ``"rsa"``, ``"ec"``, ``"ed25519"``, ``"ed448"``,
    /// ``"dsa"``, or ``"unknown"``.
    #[getter]
    fn key_type(&self) -> &'static str {
        self.inner.key_type()
    }

    /// The key size in bits, or ``None`` for EdDSA keys.
    #[getter]
    fn key_size(&self) -> Option<i64> {
        self.inner.key_bit_size()
    }

    /// The public key corresponding to this private key.
    ///
    /// ```python,ignore
    /// pub = priv.public_key
    /// ct = pub.rsa_oaep_encrypt(b"data", "sha256")
    /// ```
    #[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 })
    }

    /// Decrypt ``ciphertext`` with RSA-OAEP using the specified hash algorithm.
    ///
    /// ``hash_algorithm`` must be one of ``"sha1"``, ``"sha224"``,
    /// ``"sha256"``, ``"sha384"``, or ``"sha512"``.
    ///
    /// Raises :exc:`ValueError` if this key is not an RSA key.
    ///
    /// ```python,ignore
    /// plaintext = priv.rsa_oaep_decrypt(ciphertext, "sha256")
    /// ```
    #[pyo3(signature = (ciphertext, hash_algorithm = "sha256"))]
    fn rsa_oaep_decrypt<'py>(
        &self,
        py: Python<'py>,
        ciphertext: &[u8],
        hash_algorithm: &str,
    ) -> PyResult<Bound<'py, PyBytes>> {
        use synta_certificate::OpensslRsaOaepDecryptor;
        let pkcs8 = self
            .inner
            .to_der()
            .map_err(|e| PyValueError::new_err(format!("{e}")))?;
        let pt = OpensslRsaOaepDecryptor::new(&pkcs8, hash_algorithm)
            .map_err(|e| PyValueError::new_err(format!("{e}")))?
            .decrypt_key(ciphertext)
            .map_err(|e| PyValueError::new_err(format!("{e}")))?;
        Ok(PyBytes::new(py, &pt))
    }

    /// Decrypt ``ciphertext`` with RSA PKCS\#1 v1.5 padding.
    ///
    /// Raises :exc:`ValueError` if this key is not an RSA key.
    ///
    /// ```python,ignore
    /// plaintext = priv.rsa_pkcs1v15_decrypt(ciphertext)
    /// ```
    fn rsa_pkcs1v15_decrypt<'py>(
        &self,
        py: Python<'py>,
        ciphertext: &[u8],
    ) -> PyResult<Bound<'py, PyBytes>> {
        use synta_certificate::OpensslRsaPkcs1Decryptor;
        let pkcs8 = self
            .inner
            .to_der()
            .map_err(|e| PyValueError::new_err(format!("{e}")))?;
        let pt = OpensslRsaPkcs1Decryptor::new(&pkcs8)
            .map_err(|e| PyValueError::new_err(format!("{e}")))?
            .decrypt_key(ciphertext)
            .map_err(|e| PyValueError::new_err(format!("{e}")))?;
        Ok(PyBytes::new(py, &pt))
    }

    /// Generate a new RSA private key.
    ///
    /// ``key_size`` is the modulus bit-length (e.g. 2048, 3072, 4096).
    /// ``public_exponent`` defaults to 65537.
    ///
    /// ```python,ignore
    /// priv = synta.PrivateKey.generate_rsa(2048)
    /// ```
    #[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 })
    }

    /// Generate a new EC private key on the specified named curve.
    ///
    /// ``curve`` must be one of ``"P-256"``, ``"P-384"``, or ``"P-521"``.
    /// Raises :exc:`ValueError` for unknown curve names.
    ///
    /// ```python,ignore
    /// priv = synta.PrivateKey.generate_ec("P-256")
    /// ```
    #[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 })
    }

    fn __repr__(&self) -> String {
        let kt = self.inner.key_type();
        let bits = match kt {
            "ed25519" | "ed448" => String::new(),
            _ => self
                .inner
                .key_bit_size()
                .map(|b| format!(", key_size={b}"))
                .unwrap_or_default(),
        };
        format!("PrivateKey(key_type={kt:?}{bits})")
    }
}

// ── PrivateKey trait impl ─────────────────────────────────────────────────────

/// Implement the backend-agnostic [`synta_certificate::PrivateKey`] trait for
/// [`PyPrivateKey`] by delegating to [`synta_certificate::BackendPrivateKey`].
///
/// This allows Python binding code (e.g. `cert_builder.rs`) to call
/// `key.as_signer(algorithm)` without importing backend-specific types.
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)
    }
}