_synta/certificate/

tsp.rs

//! Python bindings for RFC 3161 Time-Stamp Protocol (TSP) builders and parsed types.
//!
//! Exposes [`PyTimeStampReqBuilder`] as a Python class for constructing
//! DER-encoded `TimeStampReq` structures, and [`PyTimeStampResp`] for
//! decoding DER-encoded `TimeStampResp` structures.

use std::sync::OnceLock;

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

use synta::{Decoder, Encoding};

use crate::error::SyntaErr;

// ── PyTimeStampReqBuilder ─────────────────────────────────────────────────────

/// Python-facing wrapper for [`synta_certificate::TimeStampReqBuilder`].
///
/// Builds a DER-encoded ``TimeStampReq`` (RFC 3161 §2.4.1).
///
/// The only required field is ``message_imprint`` (set via
/// :meth:`message_imprint` or :meth:`message_imprint_with_alg_der`).
/// All other fields are optional.
///
/// Example::
///
///     import synta
///     import synta.oids as oids
///
///     # SHA-256 hash of the content to timestamp
///     import hashlib
///     digest = hashlib.sha256(b"data").digest()
///
///     req_der = (
///         synta.TimeStampReqBuilder()
///         .message_imprint(list(oids.SHA256.components()), digest)
///         .nonce(b"\\x01\\x02\\x03\\x04\\x05\\x06\\x07\\x08")
///         .cert_req(True)
///         .build()
///     )
#[pyclass(name = "TimeStampReqBuilder")]
pub struct PyTimeStampReqBuilder {
    inner: synta_certificate::TimeStampReqBuilder,
    built: bool,
}

#[pymethods]
impl PyTimeStampReqBuilder {
    /// Create a new, empty ``TimeStampReqBuilder``.
    #[new]
    fn new() -> Self {
        Self {
            inner: synta_certificate::TimeStampReqBuilder::new(),
            built: false,
        }
    }

    /// Set the ``messageImprint`` from a hash algorithm OID (as a list of
    /// integer arcs) and raw hash bytes.
    ///
    /// A ``NULL`` parameters field is added automatically (standard for all
    /// SHA-2 hash algorithms).
    ///
    /// :param hash_alg_oid: OID arc components as a list of ints, e.g.
    ///     ``list(synta.oids.SHA256.components())``.
    /// :param hashed_message: raw hash bytes (no OCTET STRING wrapper).
    /// :raises ValueError: if the OID is invalid or encoding fails
    ///     (deferred to :meth:`build`).
    fn message_imprint<'py>(
        slf: Bound<'py, Self>,
        hash_alg_oid: Vec<u32>,
        hashed_message: &[u8],
    ) -> Bound<'py, Self> {
        {
            let mut guard = slf.borrow_mut();
            let old = std::mem::replace(
                &mut guard.inner,
                synta_certificate::TimeStampReqBuilder::new(),
            );
            guard.inner = old.message_imprint(&hash_alg_oid, hashed_message);
        }
        slf
    }

    /// Set the ``messageImprint`` from a pre-encoded ``AlgorithmIdentifier``
    /// DER TLV and raw hash bytes.
    ///
    /// :param alg_der: complete ``AlgorithmIdentifier`` SEQUENCE TLV bytes.
    /// :param hashed_message: raw hash bytes (no OCTET STRING wrapper).
    /// :raises ValueError: if decoding fails (deferred to :meth:`build`).
    fn message_imprint_with_alg_der<'py>(
        slf: Bound<'py, Self>,
        alg_der: &[u8],
        hashed_message: &[u8],
    ) -> Bound<'py, Self> {
        {
            let mut guard = slf.borrow_mut();
            let old = std::mem::replace(
                &mut guard.inner,
                synta_certificate::TimeStampReqBuilder::new(),
            );
            guard.inner = old.message_imprint_with_alg_der(alg_der, hashed_message);
        }
        slf
    }

    /// Set the optional ``reqPolicy`` OID by arc components.
    ///
    /// :param oid_components: OID arc components as a list of ints.
    /// :raises ValueError: if the OID is invalid (deferred to :meth:`build`).
    fn req_policy<'py>(slf: Bound<'py, Self>, oid_components: Vec<u32>) -> Bound<'py, Self> {
        {
            let mut guard = slf.borrow_mut();
            let old = std::mem::replace(
                &mut guard.inner,
                synta_certificate::TimeStampReqBuilder::new(),
            );
            guard.inner = old.req_policy(&oid_components);
        }
        slf
    }

    /// Set the optional ``nonce`` value.
    ///
    /// :param nonce_bytes: big-endian two's-complement representation of the
    ///     nonce integer.
    fn nonce<'py>(slf: Bound<'py, Self>, nonce_bytes: &[u8]) -> Bound<'py, Self> {
        {
            let mut guard = slf.borrow_mut();
            let old = std::mem::replace(
                &mut guard.inner,
                synta_certificate::TimeStampReqBuilder::new(),
            );
            guard.inner = old.nonce(nonce_bytes);
        }
        slf
    }

    /// Set the ``certReq`` flag.
    ///
    /// When ``True``, requests that the TSA include its signing certificate
    /// in the time-stamp response.
    ///
    /// :param val: ``True`` to request certificate inclusion.
    fn cert_req<'py>(slf: Bound<'py, Self>, val: bool) -> Bound<'py, Self> {
        {
            let mut guard = slf.borrow_mut();
            let old = std::mem::replace(
                &mut guard.inner,
                synta_certificate::TimeStampReqBuilder::new(),
            );
            guard.inner = old.cert_req(val);
        }
        slf
    }

    /// Build the DER-encoded ``TimeStampReq`` SEQUENCE.
    ///
    /// :returns: DER bytes of the complete ``TimeStampReq``.
    /// :raises ValueError: if ``message_imprint`` was not set, if any OID
    ///     was invalid, if DER encoding fails, or if called more than once.
    fn build<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if self.built {
            return Err(pyo3::exceptions::PyValueError::new_err(
                "build() has already been called; create a new builder",
            ));
        }
        self.built = true;
        let inner = std::mem::replace(
            &mut self.inner,
            synta_certificate::TimeStampReqBuilder::new(),
        );
        let der = inner
            .build()
            .map_err(pyo3::exceptions::PyValueError::new_err)?;
        Ok(PyBytes::new(py, &der))
    }

    fn __repr__(&self) -> String {
        "TimeStampReqBuilder()".to_string()
    }
}

// ── PyTimeStampResp ───────────────────────────────────────────────────────────

/// RFC 3161 Time-Stamp Response (TSP, RFC 3161 §2.4.2).
///
/// A ``TimeStampResp`` carries a ``PKIStatusInfo`` (indicating whether the
/// request was granted or rejected) and an optional ``timeStampToken``
/// (a CMS ``ContentInfo`` wrapping a DER-encoded ``SignedData``).
///
/// ```python,ignore
/// import synta
///
/// with open("response.tsr", "rb") as f:
///     resp = synta.TimeStampResp.from_der(f.read())
///
/// print(resp.status)          # 0 = granted
/// token = resp.time_stamp_token  # bytes or None
/// ```
#[pyclass(frozen, name = "TimeStampResp")]
pub struct PyTimeStampResp {
    /// Owning reference to the Python ``bytes`` object whose buffer backs `raw`.
    ///
    /// Kept alive for the entire lifetime of this struct so that `raw` is
    /// never dangling.
    _data: Py<PyBytes>,
    /// Slice into the payload of `_data`.  Typed `'static` because CPython
    /// bytes objects have non-relocating buffers and `_data` keeps the object
    /// alive; see the safety comments in [`Self::from_der`].
    raw: &'static [u8],
    /// Lazily decoded `TimeStampResp`; initialised on the first property access.
    inner: OnceLock<Box<synta_certificate::tsp_types::TimeStampResp<'static>>>,
    /// Cached value of `status.status` as an `i64` (avoids re-decoding).
    status_cache: OnceLock<i64>,
    /// Cached re-encoded `timeStampToken` DER bytes, or `None` if absent.
    time_stamp_token_cache: OnceLock<Option<Py<PyBytes>>>,
}

impl PyTimeStampResp {
    /// Decode and cache the `TimeStampResp` from `self.raw`.
    ///
    /// The first call decodes the DER bytes and stores the result in
    /// `self.inner`; subsequent calls return the cached value immediately.
    /// Returns `Err(ValueError)` if DER decoding fails.
    fn tsr(&self) -> PyResult<&synta_certificate::tsp_types::TimeStampResp<'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::tsp_types::TimeStampResp<'static>>()
            .map_err(SyntaErr)?;
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }
}

#[pymethods]
impl PyTimeStampResp {
    /// Parse a DER-encoded ``TimeStampResp`` SEQUENCE.
    ///
    /// Validates the envelope tag and length eagerly; field decoding is deferred
    /// to the first property access.
    ///
    /// :param data: DER bytes of the ``TimeStampResp``.
    /// :raises ValueError: if the bytes cannot be decoded.
    #[staticmethod]
    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        // SAFETY: `py_bytes` holds a strong reference (Py<PyBytes>) that keeps
        // the Python bytes object alive for the lifetime of this struct.
        // CPython bytes objects have a fixed-address, non-relocating payload
        // buffer (CPython has no moving GC).  The slice lifetime is extended to
        // 'static; the actual safety invariants are:
        //   (1) All reads of `raw` go through `&self`; no borrow of the struct
        //       can outlive the struct, so `raw` is never read after drop begins.
        //   (2) `raw: &'static [u8]` has no destructor (fat pointer, no heap
        //       allocation), so field drop order does not cause use-after-free.
        //   (3) `inner` contains only borrow-typed fields; dropping
        //       Box<TimeStampResp<'static>> does not read through the contained
        //       &'static slices (borrows have no destructors).
        // CPython-only: does not hold for PyPy or GraalPy.
        let raw: &'static [u8] = unsafe {
            let s = py_bytes.bind(py).as_bytes();
            std::slice::from_raw_parts(s.as_ptr(), s.len())
        };
        {
            let mut d = Decoder::new(raw, Encoding::Der);
            d.read_tag()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
            d.read_length()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        }
        Ok(Self {
            _data: py_bytes,
            raw,
            inner: OnceLock::new(),
            status_cache: OnceLock::new(),
            time_stamp_token_cache: OnceLock::new(),
        })
    }

    /// Return the original DER bytes passed to :meth:`from_der`.
    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        Ok(PyBytes::new(py, self.raw))
    }

    /// ``PKIStatus`` integer value.
    ///
    /// Named values per RFC 3161 §2.4.2:
    ///
    /// - ``0`` — ``granted``
    /// - ``1`` — ``grantedWithMods``
    /// - ``2`` — ``rejection``
    /// - ``3`` — ``waiting``
    /// - ``4`` — ``revocationWarning``
    /// - ``5`` — ``revocationNotification``
    #[getter]
    fn status(&self) -> PyResult<i64> {
        if let Some(v) = self.status_cache.get() {
            return Ok(*v);
        }
        let v = self.tsr()?.status.status.as_i64().map_err(SyntaErr)?;
        let _ = self.status_cache.set(v);
        Ok(v)
    }

    /// Raw DER bytes of the ``timeStampToken`` ``ContentInfo``, or ``None`` if
    /// the response carries no token (e.g. rejection or waiting status).
    #[getter]
    fn time_stamp_token<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.time_stamp_token_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed: Option<Py<PyBytes>> = match self.tsr()?.time_stamp_token.as_ref() {
            None => None,
            Some(ci) => {
                let der = ci.to_der().map_err(SyntaErr)?;
                Some(PyBytes::new(py, &der).unbind())
            }
        };
        let cached = computed.as_ref().map(|b| b.clone_ref(py));
        let _ = self.time_stamp_token_cache.set(cached);
        Ok(computed.map(|b| b.into_bound(py)))
    }

    fn __repr__(&self) -> PyResult<String> {
        let status = self.tsr()?.status.status.as_i64().map_err(SyntaErr)?;
        Ok(format!("TimeStampResp(status={status})"))
    }
}

// ── register ──────────────────────────────────────────────────────────────────

/// Register TSP builder classes into the given module.
pub(super) fn register_tsp_classes(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_class::<PyTimeStampReqBuilder>()?;
    m.add_class::<PyTimeStampResp>()?;
    Ok(())
}