_synta/
decoder.rs

1//! Python wrapper for ASN.1 decoder
2
3use pyo3::prelude::*;
4use pyo3::types::PyBytes;
5
6use synta::traits::Decode;
7use synta::{Decoder, Encoding, TagClass};
8
9use super::error::SyntaErr;
10
11fn parse_tag_class(s: &str) -> PyResult<TagClass> {
12    match s {
13        "Universal" => Ok(TagClass::Universal),
14        "Context" => Ok(TagClass::ContextSpecific),
15        "Application" => Ok(TagClass::Application),
16        "Private" => Ok(TagClass::Private),
17        other => Err(pyo3::exceptions::PyValueError::new_err(format!(
18            "Unknown tag class '{other}'; expected 'Universal', 'Context', 'Application', or 'Private'",
19        ))),
20    }
21}
22
23fn tag_class_str(class: TagClass) -> &'static str {
24    match class {
25        TagClass::Universal => "Universal",
26        TagClass::Application => "Application",
27        TagClass::ContextSpecific => "Context",
28        TagClass::Private => "Private",
29    }
30}
31
32use super::types::{
33    PyBitString, PyBmpString, PyBoolean, PyGeneralString, PyGeneralizedTime, PyIA5String,
34    PyInteger, PyNull, PyNumericString, PyObjectIdentifier, PyOctetString, PyPrintableString,
35    PyReal, PyTeletexString, PyUniversalString, PyUtcTime, PyUtf8String, PyVisibleString,
36};
37use super::PyEncoding;
38
39/// ASN.1 Decoder
40///
41/// Decodes ASN.1 DER/BER encoded data.
42///
43/// Example:
44///     >>> decoder = Decoder(b'\\x02\\x01\\x2A', Encoding.DER)
45///     >>> integer = decoder.decode_integer()
46///     >>> print(integer.to_int())
47///     42
48#[pyclass(name = "Decoder")]
49pub struct PyDecoder {
50    // Store owned data to ensure lifetime validity
51    data: Vec<u8>,
52    // Position in the data (to track progress)
53    position: usize,
54    // Encoding type
55    encoding: Encoding,
56}
57
58impl PyDecoder {
59    /// Advance the position by `by` bytes, asserting the invariant in debug builds.
60    fn advance(&mut self, by: usize) {
61        self.position += by;
62        debug_assert!(
63            self.position <= self.data.len(),
64            "decoder position {} overshot data length {}",
65            self.position,
66            self.data.len()
67        );
68    }
69
70    /// Decode a single owned value of type `T`, then advance the parent position.
71    ///
72    /// This encapsulates the repeated pattern:
73    ///   1. create a sub-`Decoder` over the remaining slice
74    ///   2. call `T::decode` to decode a value
75    ///   3. advance `self.position` by however many bytes were consumed
76    ///
77    /// `T` must implement `Decode<'a>` for *any* lifetime `'a` (i.e. it must own
78    /// its decoded representation, not borrow from the decoder's input buffer).
79    fn decode_one<T>(&mut self) -> PyResult<T>
80    where
81        T: for<'a> Decode<'a>,
82    {
83        let mut decoder = Decoder::new(&self.data[self.position..], self.encoding);
84        let result = T::decode(&mut decoder).map_err(SyntaErr)?;
85        self.advance(decoder.position());
86        Ok(result)
87    }
88}
89
90#[pymethods]
91impl PyDecoder {
92    /// Create a new decoder
93    ///
94    /// Args:
95    ///     data: The ASN.1 encoded data to decode
96    ///     encoding: The encoding type (DER, BER, or CER)
97    #[new]
98    fn new(data: Vec<u8>, encoding: PyEncoding) -> Self {
99        Self {
100            data,
101            position: 0,
102            encoding: encoding.into(),
103        }
104    }
105
106    /// Decode an INTEGER
107    fn decode_integer(&mut self) -> PyResult<PyInteger> {
108        self.decode_one::<synta::Integer>()
109            .map(|inner| PyInteger { inner })
110    }
111
112    /// Decode an OCTET STRING
113    fn decode_octet_string(&mut self) -> PyResult<PyOctetString> {
114        self.decode_one::<synta::OctetString>()
115            .map(|inner| PyOctetString { inner })
116    }
117
118    /// Decode an OBJECT IDENTIFIER
119    fn decode_oid(&mut self) -> PyResult<PyObjectIdentifier> {
120        self.decode_one::<synta::ObjectIdentifier>()
121            .map(PyObjectIdentifier::from_oid)
122    }
123
124    /// Decode a BIT STRING
125    fn decode_bit_string(&mut self) -> PyResult<PyBitString> {
126        self.decode_one::<synta::BitString>()
127            .map(|inner| PyBitString { inner })
128    }
129
130    /// Decode a BOOLEAN
131    fn decode_boolean(&mut self) -> PyResult<PyBoolean> {
132        self.decode_one::<synta::Boolean>()
133            .map(|inner| PyBoolean { inner })
134    }
135
136    /// Decode a UTCTime
137    fn decode_utc_time(&mut self) -> PyResult<PyUtcTime> {
138        self.decode_one::<synta::UtcTime>()
139            .map(|inner| PyUtcTime { inner })
140    }
141
142    /// Decode a GeneralizedTime
143    fn decode_generalized_time(&mut self) -> PyResult<PyGeneralizedTime> {
144        self.decode_one::<synta::GeneralizedTime>()
145            .map(|inner| PyGeneralizedTime { inner })
146    }
147
148    /// Decode a NULL
149    fn decode_null(&mut self) -> PyResult<PyNull> {
150        self.decode_one::<synta::Null>()?;
151        Ok(PyNull)
152    }
153
154    /// Decode a REAL
155    fn decode_real(&mut self) -> PyResult<PyReal> {
156        self.decode_one::<synta::Real>()
157            .map(|inner| PyReal { inner })
158    }
159
160    /// Decode a UTF8String
161    fn decode_utf8_string(&mut self) -> PyResult<PyUtf8String> {
162        self.decode_one::<synta::Utf8String>()
163            .map(|inner| PyUtf8String { inner })
164    }
165
166    /// Decode a PrintableString
167    fn decode_printable_string(&mut self) -> PyResult<PyPrintableString> {
168        self.decode_one::<synta::PrintableString>()
169            .map(|inner| PyPrintableString { inner })
170    }
171
172    /// Decode an IA5String
173    fn decode_ia5_string(&mut self) -> PyResult<PyIA5String> {
174        self.decode_one::<synta::IA5String>()
175            .map(|inner| PyIA5String { inner })
176    }
177
178    /// Decode a NumericString (tag 18)
179    fn decode_numeric_string(&mut self) -> PyResult<PyNumericString> {
180        self.decode_one::<synta::NumericString>()
181            .map(|inner| PyNumericString { inner })
182    }
183
184    /// Decode a TeletexString / T61String (tag 20)
185    fn decode_teletex_string(&mut self) -> PyResult<PyTeletexString> {
186        self.decode_one::<synta::TeletexString>()
187            .map(|inner| PyTeletexString { inner })
188    }
189
190    /// Decode a VisibleString (tag 26)
191    fn decode_visible_string(&mut self) -> PyResult<PyVisibleString> {
192        self.decode_one::<synta::VisibleString>()
193            .map(|inner| PyVisibleString { inner })
194    }
195
196    /// Decode a GeneralString (tag 27)
197    fn decode_general_string(&mut self) -> PyResult<PyGeneralString> {
198        self.decode_one::<synta::GeneralString>()
199            .map(|inner| PyGeneralString { inner })
200    }
201
202    /// Decode a UniversalString (tag 28)
203    fn decode_universal_string(&mut self) -> PyResult<PyUniversalString> {
204        self.decode_one::<synta::UniversalString>()
205            .map(|inner| PyUniversalString { inner })
206    }
207
208    /// Decode a BMPString (tag 30)
209    fn decode_bmp_string(&mut self) -> PyResult<PyBmpString> {
210        self.decode_one::<synta::BmpString>()
211            .map(|inner| PyBmpString { inner })
212    }
213
214    /// Decode the next ASN.1 element as a Python ``str``.
215    ///
216    /// Accepts any of the standard ASN.1 string types and converts to a
217    /// native Python ``str`` using the appropriate encoding for each tag:
218    ///
219    /// * UTF8String (12), PrintableString (19), IA5String (22),
220    ///   NumericString (18), VisibleString (26), GeneralString (27) —
221    ///   decoded as UTF-8 (lossily; all are ASCII or UTF-8 subsets in practice)
222    /// * TeletexString / T61String (20) — each byte decoded as a Latin-1
223    ///   code point (U+0000–U+00FF)
224    /// * BMPString (30) — decoded from UCS-2 big-endian (BMP only)
225    /// * UniversalString (28) — decoded from UCS-4 big-endian
226    ///
227    /// Raises ``ValueError`` for any other tag, and ``EOFError`` when no data
228    /// remains.
229    ///
230    /// This is the single-call replacement for the duck-typing pattern that
231    /// ``decode_any()`` requires:
232    ///
233    /// ```python
234    /// # Before — three-way probe after decode_any():
235    /// val = decoder.decode_any()
236    /// if hasattr(val, 'as_str'):
237    ///     s = val.as_str()
238    /// elif hasattr(val, 'to_bytes'):   # TeletexString / BmpString
239    ///     s = val.to_bytes().decode('latin-1')
240    /// else:
241    ///     raise ValueError(f"expected a string type, got {type(val)}")
242    ///
243    /// # After — one call, correct encoding for all nine string types:
244    /// s = decoder.decode_any_str()
245    /// ```
246    fn decode_any_str<'py>(
247        &mut self,
248        py: Python<'py>,
249    ) -> PyResult<Bound<'py, pyo3::types::PyString>> {
250        use pyo3::exceptions::{PyEOFError, PyValueError};
251        use pyo3::types::PyString;
252        use synta::tag::*;
253
254        if self.position >= self.data.len() {
255            return Err(PyEOFError::new_err("No more data"));
256        }
257
258        let (tag_num, val_bytes, advance) = {
259            let mut decoder = Decoder::new(&self.data[self.position..], self.encoding);
260            let tag = decoder.read_tag().map_err(SyntaErr)?;
261            let length = decoder.read_length().map_err(SyntaErr)?;
262            let len = length.definite().map_err(SyntaErr)?;
263            let val = decoder.read_bytes(len).map_err(SyntaErr)?.to_vec();
264            (tag.number(), val, decoder.position())
265        };
266
267        let s: String = match tag_num {
268            TAG_UTF8_STRING | TAG_PRINTABLE_STRING | TAG_IA5_STRING | TAG_VISIBLE_STRING
269            | TAG_NUMERIC_STRING | TAG_GENERAL_STRING => {
270                String::from_utf8_lossy(&val_bytes).into_owned()
271            }
272
273            // TeletexString / T61String: treat each byte as a Latin-1 codepoint.
274            TAG_TELETEX_STRING => val_bytes.iter().map(|&b| b as char).collect(),
275
276            // BMPString: UCS-2 big-endian (Basic Multilingual Plane only).
277            TAG_BMP_STRING => val_bytes
278                .chunks_exact(2)
279                .filter_map(|pair| char::from_u32(u16::from_be_bytes([pair[0], pair[1]]) as u32))
280                .collect(),
281
282            // UniversalString: UCS-4 big-endian.
283            TAG_UNIVERSAL_STRING => val_bytes
284                .chunks_exact(4)
285                .filter_map(|quad| {
286                    char::from_u32(u32::from_be_bytes([quad[0], quad[1], quad[2], quad[3]]))
287                })
288                .collect(),
289
290            other => {
291                return Err(PyValueError::new_err(format!(
292                    "tag {other} is not an ASN.1 string type; \
293                     expected UTF8String (12), PrintableString (19), IA5String (22), \
294                     NumericString (18), TeletexString (20), VisibleString (26), \
295                     GeneralString (27), BMPString (30), or UniversalString (28)"
296                )));
297            }
298        };
299
300        self.advance(advance);
301        Ok(PyString::new(py, &s))
302    }
303
304    /// Enter a SEQUENCE and return a child Decoder positioned over its contents.
305    ///
306    /// Reads and validates the SEQUENCE tag and length from the current position,
307    /// advances this decoder past the whole SEQUENCE, and returns a new Decoder
308    /// whose data is exactly the content bytes of the SEQUENCE.  Call typed
309    /// ``decode_*`` methods on the child decoder to read elements one by one;
310    /// use ``is_empty()`` to detect the end of the SEQUENCE.
311    fn decode_sequence(&mut self) -> PyResult<PyDecoder> {
312        use pyo3::exceptions::PyValueError;
313        use synta::tag::TAG_SEQUENCE;
314        use synta::TagClass;
315
316        let (data, advance) = {
317            let mut decoder = Decoder::new(&self.data[self.position..], self.encoding);
318            let tag = decoder.read_tag().map_err(SyntaErr)?;
319            if tag.class() != TagClass::Universal
320                || tag.number() != TAG_SEQUENCE
321                || !tag.is_constructed()
322            {
323                return Err(PyValueError::new_err(format!(
324                    "Expected SEQUENCE (0x30), got tag {:?}",
325                    tag
326                )));
327            }
328            let length = decoder.read_length().map_err(SyntaErr)?;
329            let len = length.definite().map_err(SyntaErr)?;
330            let data = decoder.read_bytes(len).map_err(SyntaErr)?.to_vec();
331            (data, decoder.position())
332            // decoder drops here, releasing the borrow of self.data
333        };
334        self.advance(advance);
335        Ok(PyDecoder {
336            data,
337            position: 0,
338            encoding: self.encoding,
339        })
340    }
341
342    /// Consume an explicit context tag ``[tag_num]`` and return a child Decoder
343    /// positioned over the tagged content.
344    ///
345    /// Reads the outer explicit tag+length from the current position, checks that the
346    /// tag number matches ``tag_num`` and that the class is Context-specific, then
347    /// returns a new Decoder whose data is the content of the tagged element.
348    fn decode_explicit_tag(&mut self, tag_num: u32) -> PyResult<PyDecoder> {
349        use pyo3::exceptions::PyValueError;
350        use synta::TagClass;
351
352        let (data, advance) = {
353            let mut decoder = Decoder::new(&self.data[self.position..], self.encoding);
354            let tag = decoder.read_tag().map_err(SyntaErr)?;
355            if tag.class() != TagClass::ContextSpecific || tag.number() != tag_num {
356                return Err(PyValueError::new_err(format!(
357                    "Expected explicit context tag [{}], got {:?}",
358                    tag_num, tag
359                )));
360            }
361            let length = decoder.read_length().map_err(SyntaErr)?;
362            let len = length.definite().map_err(SyntaErr)?;
363            let data = decoder.read_bytes(len).map_err(SyntaErr)?.to_vec();
364            (data, decoder.position())
365            // decoder drops here, releasing the borrow of self.data
366        };
367        self.advance(advance);
368        Ok(PyDecoder {
369            data,
370            position: 0,
371            encoding: self.encoding,
372        })
373    }
374
375    /// Enter a SET and return a child Decoder positioned over its contents.
376    ///
377    /// Symmetric counterpart to ``decode_sequence()`` for SET (tag 0x31).
378    fn decode_set(&mut self) -> PyResult<PyDecoder> {
379        use pyo3::exceptions::PyValueError;
380        use synta::tag::TAG_SET;
381
382        let (data, advance) = {
383            let mut decoder = Decoder::new(&self.data[self.position..], self.encoding);
384            let tag = decoder.read_tag().map_err(SyntaErr)?;
385            if tag.class() != TagClass::Universal
386                || tag.number() != TAG_SET
387                || !tag.is_constructed()
388            {
389                return Err(PyValueError::new_err(format!(
390                    "Expected SET (0x31), got tag {:?}",
391                    tag
392                )));
393            }
394            let length = decoder.read_length().map_err(SyntaErr)?;
395            let len = length.definite().map_err(SyntaErr)?;
396            let data = decoder.read_bytes(len).map_err(SyntaErr)?.to_vec();
397            (data, decoder.position())
398        };
399        self.advance(advance);
400        Ok(PyDecoder {
401            data,
402            position: 0,
403            encoding: self.encoding,
404        })
405    }
406
407    /// Consume an implicit tag ``[tag_num]`` and return a child Decoder over the
408    /// value content bytes.
409    ///
410    /// For implicit tags the original type tag is replaced by a context-specific
411    /// (or other class) tag.  This method validates the tag, strips the T+L, and
412    /// returns a Decoder over the raw value bytes.  The caller must supply the
413    /// expected ``tag_class`` (``"Context"``, ``"Application"``, ``"Private"``,
414    /// or ``"Universal"``).
415    ///
416    /// To decode the content as a typed value after stripping, pass the returned
417    /// bytes to a new ``Decoder``:
418    ///
419    /// ```python
420    /// raw = decoder.decode_implicit_tag(0, "Context")
421    /// inner = synta.Decoder(raw, synta.Encoding.DER)
422    /// value = inner.decode_integer()
423    /// ```
424    fn decode_implicit_tag(&mut self, tag_num: u32, tag_class: &str) -> PyResult<PyDecoder> {
425        use pyo3::exceptions::PyValueError;
426
427        let class = parse_tag_class(tag_class)?;
428
429        let (data, advance) = {
430            let mut decoder = Decoder::new(&self.data[self.position..], self.encoding);
431            let tag = decoder.read_tag().map_err(SyntaErr)?;
432            if tag.class() != class || tag.number() != tag_num {
433                return Err(PyValueError::new_err(format!(
434                    "Expected implicit tag [{tag_num} {tag_class}], got {:?}",
435                    tag
436                )));
437            }
438            let length = decoder.read_length().map_err(SyntaErr)?;
439            let len = length.definite().map_err(SyntaErr)?;
440            let data = decoder.read_bytes(len).map_err(SyntaErr)?.to_vec();
441            (data, decoder.position())
442        };
443        self.advance(advance);
444        Ok(PyDecoder {
445            data,
446            position: 0,
447            encoding: self.encoding,
448        })
449    }
450
451    /// Inspect the next tag without consuming any bytes.
452    ///
453    /// Returns ``(tag_number, tag_class, is_constructed)`` where ``tag_class``
454    /// is one of ``"Universal"``, ``"Context"``, ``"Application"``, or
455    /// ``"Private"``.  Raises ``EOFError`` if no data remains.
456    ///
457    /// Use this to implement CHOICE dispatch or to test for optional fields:
458    ///
459    /// ```python
460    /// tag_num, tag_class, _ = decoder.peek_tag()
461    /// if tag_class == "Context" and tag_num == 0:
462    ///     version = decoder.decode_explicit_tag(0)
463    /// ```
464    fn peek_tag(&self) -> PyResult<(u32, String, bool)> {
465        use pyo3::exceptions::PyEOFError;
466        if self.position >= self.data.len() {
467            return Err(PyEOFError::new_err("No more data to peek"));
468        }
469        let mut decoder = Decoder::new(&self.data[self.position..], self.encoding);
470        let tag = decoder.read_tag().map_err(SyntaErr)?;
471        Ok((
472            tag.number(),
473            tag_class_str(tag.class()).to_string(),
474            tag.is_constructed(),
475        ))
476    }
477
478    /// Read the complete next TLV (tag + length + value) as raw bytes and
479    /// advance past it.
480    ///
481    /// Useful for capturing an element whose type is unknown or whose decoding
482    /// should be deferred:
483    ///
484    /// ```python
485    /// raw_tlv = decoder.decode_raw_tlv()
486    /// inner = synta.Decoder(raw_tlv, synta.Encoding.DER)
487    /// ```
488    fn decode_raw_tlv<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
489        use pyo3::exceptions::PyEOFError;
490        if self.position >= self.data.len() {
491            return Err(PyEOFError::new_err("No more data"));
492        }
493        let total = {
494            let mut decoder = Decoder::new(&self.data[self.position..], self.encoding);
495            decoder.read_tag().map_err(SyntaErr)?;
496            let length = decoder.read_length().map_err(SyntaErr)?;
497            let content_len = length.definite().map_err(SyntaErr)?;
498            decoder.read_bytes(content_len).map_err(SyntaErr)?;
499            decoder.position()
500        };
501        let end = self.position + total;
502        let bytes = PyBytes::new(py, &self.data[self.position..end]);
503        self.advance(total);
504        Ok(bytes)
505    }
506
507    /// Return all remaining bytes from the current position as a Python bytes object.
508    ///
509    /// After ``decode_implicit_tag()`` strips the context tag and length, the
510    /// child ``Decoder``'s data contains only the raw value bytes (no TLV
511    /// header).  For **constructed** implicit types (e.g.
512    /// ``[0] IMPLICIT SEQUENCE``) the child can be decoded element-by-element
513    /// using the ``decode_*`` methods.  For **primitive** implicit types (e.g.
514    /// ``[2] IMPLICIT IA5String`` for dNSName, ``[7] IMPLICIT OCTET STRING``
515    /// for iPAddress) the child data is just the bare value bytes —
516    /// ``remaining_bytes()`` retrieves them directly without a new
517    /// encode/decode round-trip.
518    ///
519    /// Example — decode a dNSName GeneralName::
520    ///
521    ///     tag_num, tag_class, _ = decoder.peek_tag()
522    ///     if tag_class == "Context" and tag_num == 2:
523    ///         child = decoder.decode_implicit_tag(2, "Context")
524    ///         dns_name = child.remaining_bytes().decode("ascii")
525    fn remaining_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
526        PyBytes::new(py, &self.data[self.position..])
527    }
528
529    /// Decode the next ASN.1 element as a dynamically-typed Python object.
530    ///
531    /// Sequences and Sets are returned as Python lists.
532    /// Tagged elements are returned as ``TaggedElement`` objects; use
533    /// ``isinstance(x, synta.TaggedElement)`` to detect them.
534    fn decode_any(&mut self, py: Python) -> PyResult<Py<PyAny>> {
535        let (py_obj, advance) = {
536            let mut decoder = Decoder::new(&self.data[self.position..], self.encoding);
537            let element = synta::Element::decode(&mut decoder).map_err(SyntaErr)?;
538            let advance = decoder.position();
539            let py_obj = super::types::element_to_pyobject(element, py)?;
540            (py_obj, advance)
541        };
542        self.advance(advance);
543        Ok(py_obj)
544    }
545
546    /// Check if there is more data to decode
547    fn is_empty(&self) -> bool {
548        self.position == self.data.len()
549    }
550
551    /// Get the current position in the data
552    fn position(&self) -> usize {
553        self.position
554    }
555
556    /// Get the remaining bytes
557    fn remaining(&self) -> usize {
558        self.data.len().saturating_sub(self.position)
559    }
560
561    fn __repr__(&self) -> String {
562        format!(
563            "Decoder(position={}, remaining={}, encoding={:?})",
564            self.position,
565            self.remaining(),
566            self.encoding
567        )
568    }
569}
_synta/decoder.rs

_synta/
decoder.rs
