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

_synta/
decoder.rs
