synta-python 0.2.5

//! Python bindings for PKCS #8 / RFC 5958 private key structures.
//!
//! Exposes ``OneAsymmetricKey`` (aka ``PrivateKeyInfo``) into the
//! ``synta.pkcs8`` submodule for parsing DER-encoded private key files.

use std::sync::OnceLock;

use pyo3::prelude::*;
use pyo3::types::PyBytes;

use synta::{Decoder, Encoding};

use crate::error::SyntaErr;
use crate::types::PyObjectIdentifier;

// ── PyOneAsymmetricKey ────────────────────────────────────────────────────────

/// A PKCS #8 / RFC 5958 ``OneAsymmetricKey`` (also known as
/// ``PrivateKeyInfo``) structure.
///
/// Parses a DER-encoded private key envelope that contains:
///
/// - ``version`` — 0 for v1, 1 for v2 (RFC 5958)
/// - ``private_key_algorithm`` — the algorithm OID (e.g. RSA, EC, Ed25519)
/// - ``private_key`` — the raw key material as an OCTET STRING
/// - ``attributes`` (optional) — ``[0] IMPLICIT`` bag of attributes
/// - ``public_key`` (optional) — ``[1] IMPLICIT`` bit string (v2 only)
///
/// Example:
///
/// ```python,ignore
/// import synta.pkcs8 as pkcs8
///
/// with open("key.p8", "rb") as f:
///     key = pkcs8.OneAsymmetricKey.from_der(f.read())
/// print(key.version, key.private_key_algorithm)
/// print(key.private_key.hex())
/// ```
#[pyclass(frozen, name = "OneAsymmetricKey")]
pub struct PyOneAsymmetricKey {
    _data: Py<PyBytes>,
    raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::pkcs8_types::OneAsymmetricKey<'static>>>,
    // per-field caches
    alg_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
}

impl PyOneAsymmetricKey {
    fn key(&self) -> PyResult<&synta_certificate::pkcs8_types::OneAsymmetricKey<'static>> {
        if let Some(v) = self.inner.get() {
            return Ok(v.as_ref());
        }
        let mut dec = Decoder::new(self.raw, Encoding::Der);
        let decoded = dec
            .decode::<synta_certificate::pkcs8_types::OneAsymmetricKey<'_>>()
            .map_err(SyntaErr)?;
        // SAFETY: raw is pinned by _data for the lifetime of self.
        let decoded: synta_certificate::pkcs8_types::OneAsymmetricKey<'static> =
            unsafe { std::mem::transmute(decoded) };
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }
}

#[pymethods]
impl PyOneAsymmetricKey {
    /// Parse a DER-encoded ``OneAsymmetricKey`` SEQUENCE.
    ///
    /// :param data: DER bytes (the content of a PKCS #8 ``.p8`` or ``.key``
    ///     file).
    /// :raises ValueError: if the bytes cannot be decoded.
    #[staticmethod]
    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        {
            let raw = py_bytes.as_bytes(py);
            Decoder::new(raw, Encoding::Der)
                .decode::<synta_certificate::pkcs8_types::OneAsymmetricKey<'_>>()
                .map_err(SyntaErr)?;
        }
        let raw: &'static [u8] = unsafe { std::mem::transmute(py_bytes.as_bytes(py)) };
        Ok(Self {
            _data: py_bytes,
            raw,
            inner: OnceLock::new(),
            alg_oid_cache: OnceLock::new(),
        })
    }

    /// Re-encode the ``OneAsymmetricKey`` to DER bytes.
    ///
    /// For a round-tripped object this returns the original bytes verbatim.
    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        Ok(PyBytes::new(py, &self.key()?.to_der().map_err(SyntaErr)?))
    }

    /// Integer version: 0 for v1 (``PrivateKeyInfo``), 1 for v2 (RFC 5958).
    #[getter]
    fn version(&self) -> PyResult<i64> {
        Ok(self.key()?.version.as_i64().map_err(SyntaErr)?)
    }

    /// The private-key algorithm :class:`~synta.ObjectIdentifier`.
    ///
    /// For RSA keys this is ``1.2.840.113549.1.1.1``; for EC keys it is
    /// ``1.2.840.10045.2.1``; for Ed25519 it is ``1.3.101.112``.
    #[getter]
    fn private_key_algorithm<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(c) = self.alg_oid_cache.get() {
            return Ok(c.clone_ref(py).into_bound(py));
        }
        let k = self.key()?;
        let oid_obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(k.private_key_algorithm.algorithm.clone()),
        )?;
        let _ = self.alg_oid_cache.set(oid_obj.clone_ref(py));
        Ok(oid_obj.into_bound(py))
    }

    /// Raw key material bytes (the OCTET STRING value of ``privateKey``).
    ///
    /// The internal encoding of these bytes depends on the algorithm:
    /// RFC 8410 (Ed25519/Ed448) wraps an OCTET STRING, while SEC 1 (EC)
    /// uses an ``ECPrivateKey`` SEQUENCE.
    #[getter]
    fn private_key<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        Ok(PyBytes::new(py, self.key()?.private_key.as_bytes()))
    }

    /// Raw DER bytes of the ``[0] IMPLICIT`` attributes bag, or ``None``.
    ///
    /// The bytes include the implicit tag; feed them to a ``synta.Decoder``
    /// for further inspection.
    #[getter]
    fn attributes_der<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        Ok(self
            .key()?
            .attributes
            .as_ref()
            .map(|a| PyBytes::new(py, a.as_bytes())))
    }

    /// Raw DER bytes of the ``[1] IMPLICIT`` public key bit string, or ``None``.
    ///
    /// Present only for v2 (RFC 5958) keys that include the public key.
    #[getter]
    fn public_key_der<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        Ok(self
            .key()?
            .public_key
            .as_ref()
            .map(|pk| PyBytes::new(py, pk.as_bytes())))
    }

    /// Algorithm parameters DER bytes, or ``None`` if absent.
    ///
    /// For RSA keys the parameters field is absent (``None``).  For EC keys
    /// it contains a ``namedCurve`` OID.
    #[getter]
    fn alg_parameters_der<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        let k = self.key()?;
        Ok(k.private_key_algorithm.parameters.as_ref().map(|p| {
            let mut enc = synta::Encoder::new(Encoding::Der);
            if enc.encode(p).is_err() {
                return PyBytes::new(py, &[]);
            }
            PyBytes::new(py, &enc.finish().unwrap_or_default())
        }))
    }

    fn __repr__(&self, py: Python<'_>) -> PyResult<String> {
        let alg = self.private_key_algorithm(py)?;
        Ok(format!(
            "OneAsymmetricKey(version={}, algorithm={})",
            self.version()?,
            alg.borrow().inner,
        ))
    }
}

// ── register_pkcs8_submodule ──────────────────────────────────────────────────

/// Build and register the ``synta.pkcs8`` submodule.
pub(super) fn register_pkcs8_submodule(parent: &Bound<'_, PyModule>) -> PyResult<()> {
    let py = parent.py();
    let m = PyModule::new(py, "pkcs8")?;

    m.add_class::<PyOneAsymmetricKey>()?;

    // ``PrivateKeyInfo`` is the RFC 5958 alias for ``OneAsymmetricKey``.
    // Expose it as a module-level alias so both names work.
    m.add("PrivateKeyInfo", m.getattr("OneAsymmetricKey")?)?;

    crate::install_submodule(
        parent,
        &m,
        "synta.pkcs8",
        Some(concat!(
            "synta.pkcs8 — PKCS #8 / RFC 5958 private key structures.\n\n",
            "Provides OneAsymmetricKey (PrivateKeyInfo) for parsing DER-encoded\n",
            "private key envelopes produced by OpenSSL and other PKI tools.",
        )),
    )
}