_synta/

lib.rs

//! Python bindings for Synta ASN.1 library
//!
//! This module provides PyO3-based Python bindings for the Synta ASN.1 library,
//! enabling high-performance ASN.1 parsing and encoding from Python.

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 krb5;
pub mod otp;
pub mod pkinit;
pub mod types;
pub mod x509_verification;

// Re-export for convenience
pub use certificate::*;
pub use decoder::*;
pub use encoder::*;
pub use error::*;
pub use types::*;

/// Register a submodule into both the parent module and ``sys.modules``.
///
/// Sets the module's ``__name__`` to ``qualified_name`` (e.g. ``"synta.oids"``)
/// and optionally its ``__doc__``, adds it as an attribute on ``parent`` under
/// the leaf name (the part after the last ``'.'``), and installs it in
/// ``sys.modules[qualified_name]``.
///
/// This ensures ``import synta.oids``, ``from synta import oids``, and
/// ``repr(synta.oids)`` all behave correctly.
pub(crate) fn install_submodule<'py>(
    parent: &Bound<'py, PyModule>,
    m: &Bound<'py, PyModule>,
    qualified_name: &str,
    doc: Option<&str>,
) -> PyResult<()> {
    let py = parent.py();
    m.setattr("__name__", qualified_name)?;
    if let Some(docstr) = doc {
        m.setattr("__doc__", docstr)?;
    }
    let (package, leaf) = qualified_name
        .rsplit_once('.')
        .unwrap_or(("", qualified_name));
    m.setattr("__package__", package)?;
    let sys_modules = py.import("sys")?.getattr("modules")?;
    sys_modules.set_item(qualified_name, m)?;
    parent.add(leaf, m)?;
    Ok(())
}

/// ASN.1 encoding rules
///
/// Specifies which encoding rules to use when encoding or decoding ASN.1 data.
#[pyclass(name = "Encoding")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PyEncoding {
    /// Distinguished Encoding Rules (deterministic subset of BER)
    DER,
    /// Basic Encoding Rules (most flexible)
    BER,
    /// Canonical Encoding Rules (similar to DER but for streaming)
    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,
        }
    }
}

/// Synta: High-performance ASN.1 parser and encoder
///
/// This module provides ASN.1 parsing, decoding, and encoding capabilities
/// with support for DER (Distinguished Encoding Rules) and BER (Basic Encoding Rules).
///
/// Example:
///     >>> import synta
///     >>> # Decode an integer
///     >>> decoder = synta.Decoder(b'\\x02\\x01\\x2A', synta.Encoding.DER)
///     >>> integer = decoder.decode_integer()
///     >>> print(integer.to_int())
///     42
#[pymodule]
fn _synta(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
    // Add encoding enum
    m.add_class::<PyEncoding>()?;

    // Add error types
    m.add("SyntaError", py.get_type::<SyntaError>())?;

    // Add decoder and encoder
    m.add_class::<PyDecoder>()?;
    m.add_class::<PyEncoder>()?;

    // Add primitive types
    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>()?;
    // New string types
    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>()?;

    // Add certificate types (Certificate, CertificationRequest, CertificateList, OCSPResponse)
    // and the pem_to_der helper function.
    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)?)?;

    // PublicKey and PrivateKey classes
    m.add_class::<crypto_keys::PyPublicKey>()?;
    m.add_class::<crypto_keys::PyPrivateKey>()?;

    // Symmetric crypto submodule (synta.crypto)
    crypto::register_crypto_module(m)?;

    // X.509 extension value builders submodule (synta.ext)
    ext_builders::register_ext_module(m)?;

    // Kerberos V5 submodule (synta.krb5)
    krb5::register_krb5_module(m)?;

    // X.509 verification submodule (synta.x509)
    x509_verification::register_x509_module(m)?;

    // Add version
    m.add("__version__", env!("CARGO_PKG_VERSION"))?;

    Ok(())
}

/// Parse a DER-encoded GeneralNames SEQUENCE into ``(tag_number, content_bytes)`` pairs.
///
/// ``san_der`` must be the **complete DER bytes** of the ``SEQUENCE OF GeneralName``
/// value — exactly what you get from the SAN extension's ``extn_value`` octet-string
/// content, or from ``Certificate.get_extension_value_der("2.5.29.17")``.
///
/// Returns a ``list`` of ``(tag_number: int, content: bytes)`` tuples, one per
/// ``GeneralName`` alternative.  Tag numbers follow RFC 5280:
///
/// * 0 — otherName (constructed; ``content`` is the full ``OtherNameValue`` TLV)
/// * 1 — rfc822Name (email); ``content`` is raw IA5String bytes
/// * 2 — dNSName; ``content`` is raw IA5String bytes
/// * 3 — x400Address
/// * 4 — directoryName; ``content`` is the Name SEQUENCE TLV — pass to ``parse_name_attrs()``
/// * 5 — ediPartyName
/// * 6 — uniformResourceIdentifier; ``content`` is raw IA5String bytes
/// * 7 — iPAddress; ``content`` is 4 bytes (IPv4) or 16 bytes (IPv6)
/// * 8 — registeredID; ``content`` is raw OID value bytes
///
/// Tag constants are available in the :mod:`synta.general_name` submodule
/// (e.g. ``synta.general_name.DNS_NAME == 2``), making dispatch readable
/// without hardcoded magic numbers:
///
/// ```python,ignore
/// import ipaddress
/// import synta.general_name as gn
///
/// san_der = cert.get_extension_value_der("2.5.29.17")
/// for tag_num, content in synta.parse_general_names(san_der):
///     if tag_num == gn.DNS_NAME:
///         print("DNS:", content.decode("ascii"))
///     elif tag_num == gn.IP_ADDRESS:
///         print("IP:", ipaddress.ip_address(content))
///     elif tag_num == gn.RFC822_NAME:
///         print("email:", content.decode("ascii"))
///     elif tag_num == gn.DIRECTORY_NAME:
///         attrs = synta.parse_name_attrs(content)
///         print("DirName:", attrs)
///     elif tag_num == gn.URI:
///         print("URI:", content.decode("ascii"))
/// ```
///
/// Returns an empty list if ``san_der`` cannot be parsed as a DER SEQUENCE.
#[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)
}

/// Walk a DER-encoded X.500 Name SEQUENCE and return ``(dotted_oid, value_str)`` pairs.
///
/// ``name_der`` must be the **complete TLV** bytes of the Name SEQUENCE (tag + length
/// + value), as returned by ``Certificate.issuer_raw_der`` or
/// ``Certificate.subject_raw_der``, or from a ``directoryName`` entry in
/// ``parse_general_names()``.
///
/// Returns a ``list`` of ``(oid: str, value: str)`` tuples in DER traversal order
/// (outermost RDN first, innermost ATV first within each RDN).  The OID is always
/// in dotted-decimal notation (e.g. ``"2.5.4.3"``).  The value string is decoded
/// using the appropriate per-tag encoding: UTF-8 for most types, Latin-1 for
/// TeletexString, UCS-2 big-endian for BMPString, and UCS-4 big-endian for
/// UniversalString.
///
/// This replaces manual ``Decoder`` iteration over the Name structure and is the
/// structured-data counterpart to the ``Certificate.issuer`` string property:
///
/// ```python
/// # Inspect subject attributes directly:
/// attrs = synta.parse_name_attrs(cert.subject_raw_der)
/// # → [("2.5.4.6", "US"), ("2.5.4.10", "Example Corp"), ("2.5.4.3", "Root CA")]
///
/// # Build a cryptography.x509.Name for comparison or re-use:
/// from cryptography.x509 import Name, NameAttribute, ObjectIdentifier
/// name = Name([
///     NameAttribute(ObjectIdentifier(oid), val)
///     for oid, val in synta.parse_name_attrs(cert.subject_raw_der)
/// ])
/// ```
///
/// Returns an empty list if ``name_der`` cannot be parsed.
#[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)
}

/// Encode DER bytes as a PEM block.
///
/// Returns :class:`bytes` containing a ``-----BEGIN {label}-----`` /
/// ``-----END {label}-----`` block with standard 64-character base64 lines.
/// This is the low-level inverse of :func:`pem_to_der`.
///
/// For serialising parsed objects use the class-level
/// ``Certificate.to_pem()``, ``CertificationRequest.to_pem()``, etc., which
/// fill in the correct label automatically.
///
/// ```python
/// with open("cert.der", "rb") as f:
///     der = f.read()
/// pem = synta.der_to_pem(der, "CERTIFICATE")
/// ```
#[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))
}

/// Decode PEM blocks to DER bytes.
///
/// Strips ``-----BEGIN ...-----`` / ``-----END ...-----`` boundary lines and
/// decodes the base64 body of every PEM block found in the input.  Implemented
/// in pure Rust — no external dependencies required.
///
/// Always returns :class:`list` [:class:`bytes`] — one entry per PEM block.
/// Raises :exc:`ValueError` if no PEM block is found.
///
/// ```python,ignore
/// # Single block — index into the list:
/// der = synta.pem_to_der(open("cert.pem", "rb").read())[0]
/// cert = synta.Certificate.from_der(der)
///
/// # Bundle / chain:
/// ders = synta.pem_to_der(open("bundle.pem", "rb").read())
/// certs = [synta.Certificate.from_der(d) for d in ders]
/// ```
#[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)
}

/// Compute a hash digest of arbitrary bytes.
///
/// Returns a :class:`bytes` object containing the raw (binary) digest.
/// ``algorithm`` must be one of ``"sha1"``, ``"sha224"``, ``"sha256"``,
/// ``"sha384"``, ``"sha512"``, or ``"md5"``.  Raises :exc:`ValueError` for
/// unknown algorithm names or crypto backend errors.
///
/// ```python,ignore
/// import synta
///
/// # Hash a certificate DER blob:
/// digest_bytes = synta.digest("sha256", cert_der)
/// print(digest_bytes.hex())
///
/// # Hash an arbitrary byte string:
/// digest_bytes = synta.digest("sha1", b"hello world")
/// ```
#[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))
}