Skip to main content

_synta/types/
primitives.rs

1//! Python wrappers for primitive ASN.1 types: INTEGER, OCTET STRING, BIT STRING,
2//! BOOLEAN, UTCTime, GeneralizedTime, REAL, NULL.
3
4use std::collections::hash_map::DefaultHasher;
5use std::hash::{Hash, Hasher};
6
7use pyo3::exceptions::{PyOverflowError, PyValueError};
8use pyo3::prelude::*;
9use pyo3::types::PyBytes;
10
11use synta::{
12    BitString, Boolean, FromDer, GeneralizedTime, Integer, Null, OctetString, Real, ToDer, UtcTime,
13};
14
15use crate::error::SyntaErr;
16
17/// Python wrapper for ASN.1 INTEGER
18#[pyclass(name = "Integer")]
19#[derive(Debug, Clone)]
20pub struct PyInteger {
21    pub(crate) inner: Integer,
22}
23
24#[pymethods]
25impl PyInteger {
26    /// Create a new Integer from a Python int
27    #[new]
28    fn new(value: i64) -> Self {
29        Self {
30            inner: Integer::from_i64(value),
31        }
32    }
33
34    /// Convert to Python int (i64)
35    fn to_int(&self) -> PyResult<i64> {
36        self.inner
37            .as_i64()
38            .map_err(|_| PyOverflowError::new_err("Integer too large for i64"))
39    }
40
41    /// Get the raw bytes (big-endian two's complement)
42    fn to_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
43        PyBytes::new(py, self.inner.as_bytes())
44    }
45
46    /// Create Integer from raw bytes
47    #[staticmethod]
48    fn from_bytes(bytes: &[u8]) -> Self {
49        Self {
50            inner: Integer::from_bytes(bytes),
51        }
52    }
53
54    /// Create Integer from an unsigned 64-bit value
55    #[staticmethod]
56    fn from_u64(value: u64) -> Self {
57        Self {
58            inner: Integer::from_u64(value),
59        }
60    }
61
62    /// Return the DER encoding of this ``Integer``.
63    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
64        Ok(PyBytes::new(py, &self.inner.to_der().map_err(SyntaErr)?))
65    }
66
67    /// Parse a DER-encoded ``Integer``.
68    ///
69    /// :raises ValueError: if the bytes cannot be decoded.
70    #[staticmethod]
71    fn from_der(data: &[u8]) -> PyResult<Self> {
72        let inner = Integer::from_der(data).map_err(SyntaErr)?;
73        Ok(Self { inner })
74    }
75
76    /// Convert to Python int (i128) for larger integers
77    fn to_i128(&self) -> PyResult<i128> {
78        self.inner
79            .as_i128()
80            .map_err(|_| PyOverflowError::new_err("Integer too large for i128"))
81    }
82
83    fn __eq__(&self, other: &Self) -> bool {
84        // DER encodes integers with minimal bytes (no redundant leading zeros),
85        // so comparing the raw bytes is equivalent to comparing by value.
86        self.inner.as_bytes() == other.inner.as_bytes()
87    }
88
89    fn __hash__(&self) -> u64 {
90        let mut h = DefaultHasher::new();
91        self.inner.as_bytes().hash(&mut h);
92        h.finish()
93    }
94
95    fn __repr__(&self) -> PyResult<String> {
96        match self.inner.as_i64() {
97            Ok(val) => Ok(format!("Integer({})", val)),
98            Err(_) => Ok(format!("Integer(<{} bytes>)", self.inner.as_bytes().len())),
99        }
100    }
101
102    fn __str__(&self) -> PyResult<String> {
103        match self.inner.as_i64() {
104            Ok(val) => Ok(val.to_string()),
105            Err(_) => Ok(format!("<integer {} bytes>", self.inner.as_bytes().len())),
106        }
107    }
108}
109
110/// Python wrapper for ASN.1 OCTET STRING
111#[pyclass(name = "OctetString")]
112#[derive(Debug, Clone)]
113pub struct PyOctetString {
114    pub(crate) inner: OctetString,
115}
116
117#[pymethods]
118impl PyOctetString {
119    /// Create a new OctetString from bytes
120    #[new]
121    fn new(data: Vec<u8>) -> Self {
122        Self {
123            inner: OctetString::new(data),
124        }
125    }
126
127    /// Get the bytes as a Python bytes object
128    fn to_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
129        PyBytes::new(py, self.inner.as_bytes())
130    }
131
132    /// Get the length
133    fn __len__(&self) -> usize {
134        self.inner.as_bytes().len()
135    }
136
137    fn __eq__(&self, other: &Self) -> bool {
138        self.inner == other.inner
139    }
140
141    fn __repr__(&self) -> String {
142        format!("OctetString(<{} bytes>)", self.inner.as_bytes().len())
143    }
144
145    /// Return the DER encoding of this ``OctetString``.
146    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
147        Ok(PyBytes::new(py, &self.inner.to_der().map_err(SyntaErr)?))
148    }
149
150    /// Parse a DER-encoded ``OctetString``.
151    ///
152    /// :raises ValueError: if the bytes cannot be decoded.
153    #[staticmethod]
154    fn from_der(data: &[u8]) -> PyResult<Self> {
155        let inner = OctetString::from_der(data).map_err(SyntaErr)?;
156        Ok(Self { inner })
157    }
158}
159
160/// Python wrapper for ASN.1 BIT STRING
161#[pyclass(name = "BitString")]
162#[derive(Debug, Clone)]
163pub struct PyBitString {
164    pub(crate) inner: BitString,
165}
166
167#[pymethods]
168impl PyBitString {
169    /// Create a new BitString from bytes and unused bits count
170    #[new]
171    fn new(data: Vec<u8>, unused_bits: u8) -> PyResult<Self> {
172        if unused_bits > 7 {
173            return Err(PyValueError::new_err("unused_bits must be 0-7"));
174        }
175        let inner = BitString::new(data, unused_bits)
176            .map_err(|e| PyValueError::new_err(format!("Invalid BitString: {:?}", e)))?;
177        Ok(Self { inner })
178    }
179
180    /// Get the bytes
181    fn to_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
182        PyBytes::new(py, self.inner.as_bytes())
183    }
184
185    /// Get the number of unused bits in the last byte
186    fn unused_bits(&self) -> u8 {
187        self.inner.unused_bits()
188    }
189
190    /// Get the number of bits
191    fn bit_len(&self) -> usize {
192        self.inner.bit_len()
193    }
194
195    fn __len__(&self) -> usize {
196        self.inner.bit_len()
197    }
198
199    fn __eq__(&self, other: &Self) -> bool {
200        self.inner == other.inner
201    }
202
203    fn __repr__(&self) -> String {
204        format!(
205            "BitString(<{} bits, {} unused>)",
206            self.inner.bit_len(),
207            self.inner.unused_bits()
208        )
209    }
210
211    /// Return the DER encoding of this ``BitString``.
212    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
213        Ok(PyBytes::new(py, &self.inner.to_der().map_err(SyntaErr)?))
214    }
215
216    /// Parse a DER-encoded ``BitString``.
217    ///
218    /// :raises ValueError: if the bytes cannot be decoded.
219    #[staticmethod]
220    fn from_der(data: &[u8]) -> PyResult<Self> {
221        let inner = BitString::from_der(data).map_err(SyntaErr)?;
222        Ok(Self { inner })
223    }
224}
225
226/// Python wrapper for ASN.1 BOOLEAN
227#[pyclass(name = "Boolean")]
228#[derive(Debug, Clone, Copy)]
229pub struct PyBoolean {
230    pub(crate) inner: Boolean,
231}
232
233#[pymethods]
234impl PyBoolean {
235    /// Create a new Boolean
236    #[new]
237    fn new(value: bool) -> Self {
238        Self {
239            inner: Boolean::new(value),
240        }
241    }
242
243    /// Get the boolean value
244    fn value(&self) -> bool {
245        self.inner.value()
246    }
247
248    fn __bool__(&self) -> bool {
249        self.inner.value()
250    }
251
252    fn __eq__(&self, other: &Self) -> bool {
253        self.inner.value() == other.inner.value()
254    }
255
256    fn __hash__(&self) -> isize {
257        // Matches Python's hash(True) == 1, hash(False) == 0.
258        self.inner.value() as isize
259    }
260
261    fn __repr__(&self) -> String {
262        format!("Boolean({})", self.inner.value())
263    }
264
265    /// Return the DER encoding of this ``Boolean``.
266    #[allow(clippy::wrong_self_convention)]
267    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
268        Ok(PyBytes::new(py, &self.inner.to_der().map_err(SyntaErr)?))
269    }
270
271    /// Parse a DER-encoded ``Boolean``.
272    ///
273    /// :raises ValueError: if the bytes cannot be decoded.
274    #[staticmethod]
275    fn from_der(data: &[u8]) -> PyResult<Self> {
276        let inner = Boolean::from_der(data).map_err(SyntaErr)?;
277        Ok(Self { inner })
278    }
279}
280
281/// Python wrapper for ASN.1 UTCTime
282#[pyclass(name = "UtcTime")]
283#[derive(Debug, Clone)]
284pub struct PyUtcTime {
285    pub(crate) inner: UtcTime,
286}
287
288#[pymethods]
289impl PyUtcTime {
290    /// Create a new UTCTime (year must be in 1950-2049 range)
291    #[new]
292    fn new(year: u16, month: u8, day: u8, hour: u8, minute: u8, second: u8) -> PyResult<Self> {
293        let inner = UtcTime::new(year, month, day, hour, minute, second)
294            .map_err(|e| PyValueError::new_err(format!("Invalid UTCTime: {:?}", e)))?;
295        Ok(Self { inner })
296    }
297
298    #[getter]
299    fn year(&self) -> u16 {
300        self.inner.year
301    }
302
303    #[getter]
304    fn month(&self) -> u8 {
305        self.inner.month
306    }
307
308    #[getter]
309    fn day(&self) -> u8 {
310        self.inner.day
311    }
312
313    #[getter]
314    fn hour(&self) -> u8 {
315        self.inner.hour
316    }
317
318    #[getter]
319    fn minute(&self) -> u8 {
320        self.inner.minute
321    }
322
323    #[getter]
324    fn second(&self) -> u8 {
325        self.inner.second
326    }
327
328    fn __str__(&self) -> String {
329        self.inner.to_string()
330    }
331
332    fn __repr__(&self) -> String {
333        format!("UtcTime('{}')", self.inner)
334    }
335
336    /// Return the DER encoding of this ``UtcTime``.
337    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
338        Ok(PyBytes::new(py, &self.inner.to_der().map_err(SyntaErr)?))
339    }
340
341    /// Parse a DER-encoded ``UtcTime``.
342    ///
343    /// :raises ValueError: if the bytes cannot be decoded.
344    #[staticmethod]
345    fn from_der(data: &[u8]) -> PyResult<Self> {
346        let inner = UtcTime::from_der(data).map_err(SyntaErr)?;
347        Ok(Self { inner })
348    }
349}
350
351impl std::fmt::Display for PyUtcTime {
352    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
353        write!(f, "{}", self.inner)
354    }
355}
356
357/// Python wrapper for ASN.1 GeneralizedTime
358#[pyclass(name = "GeneralizedTime")]
359#[derive(Debug, Clone)]
360pub struct PyGeneralizedTime {
361    pub(crate) inner: GeneralizedTime,
362}
363
364#[pymethods]
365impl PyGeneralizedTime {
366    /// Create a new GeneralizedTime
367    ///
368    /// Args:
369    ///     milliseconds: Optional fractional seconds in milliseconds (0-999)
370    #[new]
371    fn new(
372        year: u16,
373        month: u8,
374        day: u8,
375        hour: u8,
376        minute: u8,
377        second: u8,
378        milliseconds: Option<u16>,
379    ) -> PyResult<Self> {
380        let inner = GeneralizedTime::new(year, month, day, hour, minute, second, milliseconds)
381            .map_err(|e| PyValueError::new_err(format!("Invalid GeneralizedTime: {:?}", e)))?;
382        Ok(Self { inner })
383    }
384
385    #[getter]
386    fn year(&self) -> u16 {
387        self.inner.year
388    }
389
390    #[getter]
391    fn month(&self) -> u8 {
392        self.inner.month
393    }
394
395    #[getter]
396    fn day(&self) -> u8 {
397        self.inner.day
398    }
399
400    #[getter]
401    fn hour(&self) -> u8 {
402        self.inner.hour
403    }
404
405    #[getter]
406    fn minute(&self) -> u8 {
407        self.inner.minute
408    }
409
410    #[getter]
411    fn second(&self) -> u8 {
412        self.inner.second
413    }
414
415    #[getter]
416    fn milliseconds(&self) -> Option<u16> {
417        self.inner.milliseconds
418    }
419
420    fn __str__(&self) -> String {
421        self.inner.to_string()
422    }
423
424    fn __repr__(&self) -> String {
425        format!("GeneralizedTime('{}')", self.inner)
426    }
427
428    /// Return the DER encoding of this ``GeneralizedTime``.
429    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
430        Ok(PyBytes::new(py, &self.inner.to_der().map_err(SyntaErr)?))
431    }
432
433    /// Parse a DER-encoded ``GeneralizedTime``.
434    ///
435    /// :raises ValueError: if the bytes cannot be decoded.
436    #[staticmethod]
437    fn from_der(data: &[u8]) -> PyResult<Self> {
438        let inner = GeneralizedTime::from_der(data).map_err(SyntaErr)?;
439        Ok(Self { inner })
440    }
441}
442
443impl std::fmt::Display for PyGeneralizedTime {
444    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
445        write!(f, "{}", self.inner)
446    }
447}
448
449/// Python wrapper for ASN.1 REAL
450#[pyclass(name = "Real")]
451#[derive(Debug, Clone, Copy)]
452pub struct PyReal {
453    pub(crate) inner: Real,
454}
455
456#[pymethods]
457impl PyReal {
458    /// Create a new Real from a Python float
459    #[new]
460    fn new(value: f64) -> Self {
461        Self {
462            inner: Real::new(value),
463        }
464    }
465
466    /// Get the f64 value
467    fn value(&self) -> f64 {
468        self.inner.value()
469    }
470
471    /// Return True if this is positive infinity
472    fn is_infinite(&self) -> bool {
473        self.inner.value().is_infinite()
474    }
475
476    /// Return True if this is NaN (not-a-number)
477    fn is_nan(&self) -> bool {
478        self.inner.value().is_nan()
479    }
480
481    /// Return True if the value is finite
482    fn is_finite(&self) -> bool {
483        self.inner.value().is_finite()
484    }
485
486    fn __float__(&self) -> f64 {
487        self.inner.value()
488    }
489
490    fn __repr__(&self) -> String {
491        format!("Real({})", self.inner.value())
492    }
493
494    fn __str__(&self) -> String {
495        self.inner.value().to_string()
496    }
497
498    fn __eq__(&self, other: &Self) -> bool {
499        // NaN != NaN by IEEE 754 — mirror Python float behaviour
500        self.inner.value() == other.inner.value()
501    }
502
503    fn __hash__(&self, py: Python<'_>) -> PyResult<isize> {
504        // Delegate to Python's float.__hash__ so that hash(Real(1.0)) == hash(1.0) == hash(1),
505        // matching the Python data model guarantee for numeric types.
506        self.inner.value().into_pyobject(py)?.hash()
507    }
508
509    /// Return the DER encoding of this ``Real``.
510    #[allow(clippy::wrong_self_convention)]
511    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
512        Ok(PyBytes::new(py, &self.inner.to_der().map_err(SyntaErr)?))
513    }
514
515    /// Parse a DER-encoded ``Real``.
516    ///
517    /// :raises ValueError: if the bytes cannot be decoded.
518    #[staticmethod]
519    fn from_der(data: &[u8]) -> PyResult<Self> {
520        let inner = Real::from_der(data).map_err(SyntaErr)?;
521        Ok(Self { inner })
522    }
523}
524
525/// Python wrapper for ASN.1 NULL
526#[pyclass(name = "Null")]
527#[derive(Debug, Clone, Copy)]
528pub struct PyNull;
529
530#[pymethods]
531impl PyNull {
532    #[new]
533    fn new() -> Self {
534        Self
535    }
536
537    fn __repr__(&self) -> String {
538        "Null()".to_string()
539    }
540
541    fn __eq__(&self, _other: &Self) -> bool {
542        true
543    }
544
545    fn __hash__(&self) -> isize {
546        // All Null values are equal; use a fixed hash consistent with __eq__.
547        0
548    }
549
550    /// Return the DER encoding of this ``Null``.
551    #[allow(clippy::wrong_self_convention)]
552    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
553        Ok(PyBytes::new(py, &Null.to_der().map_err(SyntaErr)?))
554    }
555
556    /// Parse a DER-encoded ``Null``.
557    ///
558    /// :raises ValueError: if the bytes cannot be decoded.
559    #[staticmethod]
560    fn from_der(data: &[u8]) -> PyResult<Self> {
561        Null::from_der(data).map_err(SyntaErr)?;
562        Ok(Self)
563    }
564}