Skip to main content

_synta/
encoder.rs

1//! Python wrapper for ASN.1 encoder
2
3use pyo3::prelude::*;
4
5use synta::Encoder;
6
7use super::error::SyntaErr;
8
9use super::types::{
10    PyBitString, PyBmpString, PyBoolean, PyGeneralString, PyGeneralizedTime, PyIA5String,
11    PyInteger, PyNull, PyNumericString, PyObjectIdentifier, PyOctetString, PyPrintableString,
12    PyReal, PyRelativeOid, PyTeletexString, PyUniversalString, PyUtcTime, PyUtf8String,
13    PyVisibleString,
14};
15use super::PyEncoding;
16
17/// ASN.1 Encoder
18///
19/// Encodes ASN.1 data in DER/BER format.
20///
21/// Example:
22///     >>> encoder = Encoder(Encoding.DER)
23///     >>> encoder.encode_integer(42)
24///     >>> output = encoder.finish()
25#[pyclass(name = "Encoder")]
26pub struct PyEncoder {
27    encoder: Encoder,
28}
29
30#[pymethods]
31impl PyEncoder {
32    /// Create a new encoder
33    ///
34    /// Args:
35    ///     encoding: The encoding type (DER, BER, or CER)
36    #[new]
37    fn new(encoding: PyEncoding) -> Self {
38        Self {
39            encoder: Encoder::new(encoding.into()),
40        }
41    }
42
43    /// Encode an INTEGER
44    ///
45    /// Accepts any Python ``int`` regardless of magnitude.  Values that fit in
46    /// ``i64`` or ``i128`` are encoded via the native typed constructors;
47    /// larger values (e.g. 20-byte X.509 serial numbers up to 160 bits) are
48    /// converted through Python's ``int.to_bytes()`` using signed two's
49    /// complement, then passed directly to the encoder.
50    ///
51    /// Args:
52    ///     value: The integer value to encode (any Python int)
53    fn encode_integer(
54        &mut self,
55        py: Python,
56        value: &Bound<'_, pyo3::types::PyAny>,
57    ) -> PyResult<()> {
58        // Fast path: fits in i64 (covers most ASN.1 integer fields)
59        if let Ok(v) = value.extract::<i64>() {
60            self.encoder
61                .encode(&synta::Integer::from_i64(v))
62                .map_err(SyntaErr)?;
63            return Ok(());
64        }
65        // Fits in i128 (covers 8–16 byte random serial numbers)
66        if let Ok(v) = value.extract::<i128>() {
67            self.encoder
68                .encode(&synta::Integer::from_i128(v))
69                .map_err(SyntaErr)?;
70            return Ok(());
71        }
72        // Arbitrarily large: convert via Python int methods.
73        // ceil((bit_length + 1) / 8) == (bit_length + 8) / 8 bytes are
74        // sufficient for signed two's complement encoding of any integer.
75        let bit_len: usize = value
76            .call_method0(pyo3::intern!(py, "bit_length"))?
77            .extract()?;
78        let byte_len = ((bit_len + 8) / 8).max(1);
79        let kwargs = pyo3::types::PyDict::new(py);
80        kwargs.set_item(pyo3::intern!(py, "signed"), true)?;
81        let raw: Vec<u8> = value
82            .call_method(
83                pyo3::intern!(py, "to_bytes"),
84                (byte_len, pyo3::intern!(py, "big")),
85                Some(&kwargs),
86            )?
87            .extract()?;
88        self.encoder
89            .encode(&synta::Integer::from_bytes(&raw))
90            .map_err(SyntaErr)?;
91        Ok(())
92    }
93
94    /// Encode an INTEGER from a PyInteger object
95    ///
96    /// Args:
97    ///     value: The PyInteger object to encode
98    fn encode_integer_object(&mut self, value: &PyInteger) -> PyResult<()> {
99        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
100        Ok(())
101    }
102
103    /// Encode an OCTET STRING
104    ///
105    /// Args:
106    ///     data: The bytes to encode
107    fn encode_octet_string(&mut self, data: &[u8]) -> PyResult<()> {
108        let octet_string = synta::OctetStringRef::new(data);
109        self.encoder.encode(&octet_string).map_err(SyntaErr)?;
110        Ok(())
111    }
112
113    /// Encode an OCTET STRING from a PyOctetString object
114    ///
115    /// Args:
116    ///     value: The PyOctetString object to encode
117    fn encode_octet_string_object(&mut self, value: &PyOctetString) -> PyResult<()> {
118        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
119        Ok(())
120    }
121
122    /// Encode an OBJECT IDENTIFIER
123    ///
124    /// Args:
125    ///     value: The PyObjectIdentifier to encode
126    fn encode_oid(&mut self, value: &PyObjectIdentifier) -> PyResult<()> {
127        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
128        Ok(())
129    }
130
131    /// Encode an OBJECT IDENTIFIER from a PyObjectIdentifier object (alias for encode_oid)
132    fn encode_oid_object(&mut self, value: &PyObjectIdentifier) -> PyResult<()> {
133        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
134        Ok(())
135    }
136
137    /// Encode a RELATIVE-OID (tag 0x0D)
138    ///
139    /// Args:
140    ///     value: The :class:`RelativeOid` to encode
141    fn encode_relative_oid(&mut self, value: &PyRelativeOid) -> PyResult<()> {
142        use synta::traits::Encode;
143        value.inner.encode(&mut self.encoder).map_err(SyntaErr)?;
144        Ok(())
145    }
146
147    /// Encode a BIT STRING
148    ///
149    /// Args:
150    ///     value: The PyBitString to encode
151    fn encode_bit_string(&mut self, value: &PyBitString) -> PyResult<()> {
152        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
153        Ok(())
154    }
155
156    /// Encode a BIT STRING from a PyBitString object (alias for encode_bit_string)
157    fn encode_bit_string_object(&mut self, value: &PyBitString) -> PyResult<()> {
158        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
159        Ok(())
160    }
161
162    /// Encode a BOOLEAN
163    ///
164    /// Args:
165    ///     value: The boolean value to encode
166    fn encode_boolean(&mut self, value: bool) -> PyResult<()> {
167        let boolean = synta::Boolean::new(value);
168        self.encoder.encode(&boolean).map_err(SyntaErr)?;
169        Ok(())
170    }
171
172    /// Encode a BOOLEAN from a PyBoolean object
173    ///
174    /// Args:
175    ///     value: The PyBoolean object to encode
176    fn encode_boolean_object(&mut self, value: &PyBoolean) -> PyResult<()> {
177        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
178        Ok(())
179    }
180
181    /// Encode a UTCTime
182    ///
183    /// Args:
184    ///     value: The PyUtcTime to encode
185    fn encode_utc_time(&mut self, value: &PyUtcTime) -> PyResult<()> {
186        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
187        Ok(())
188    }
189
190    /// Encode a UTCTime from a PyUtcTime object (alias for encode_utc_time)
191    fn encode_utc_time_object(&mut self, value: &PyUtcTime) -> PyResult<()> {
192        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
193        Ok(())
194    }
195
196    /// Encode a GeneralizedTime
197    ///
198    /// Args:
199    ///     value: The PyGeneralizedTime to encode
200    fn encode_generalized_time(&mut self, value: &PyGeneralizedTime) -> PyResult<()> {
201        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
202        Ok(())
203    }
204
205    /// Encode a GeneralizedTime from a PyGeneralizedTime object (alias for encode_generalized_time)
206    fn encode_generalized_time_object(&mut self, value: &PyGeneralizedTime) -> PyResult<()> {
207        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
208        Ok(())
209    }
210
211    /// Encode a REAL
212    ///
213    /// Args:
214    ///     value: The float value to encode
215    fn encode_real(&mut self, value: f64) -> PyResult<()> {
216        let real = synta::Real::new(value);
217        self.encoder.encode(&real).map_err(SyntaErr)?;
218        Ok(())
219    }
220
221    /// Encode a REAL from a PyReal object
222    ///
223    /// Args:
224    ///     value: The PyReal object to encode
225    fn encode_real_object(&mut self, value: &PyReal) -> PyResult<()> {
226        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
227        Ok(())
228    }
229
230    /// Encode a NULL
231    fn encode_null(&mut self) -> PyResult<()> {
232        self.encoder.encode(&synta::Null).map_err(SyntaErr)?;
233        Ok(())
234    }
235
236    /// Encode a NULL from a PyNull object
237    fn encode_null_object(&mut self, _value: &PyNull) -> PyResult<()> {
238        self.encoder.encode(&synta::Null).map_err(SyntaErr)?;
239        Ok(())
240    }
241
242    /// Encode a UTF8String
243    ///
244    /// Args:
245    ///     value: The string to encode
246    fn encode_utf8_string(&mut self, value: &str) -> PyResult<()> {
247        let s = synta::Utf8String::new(value.to_string());
248        self.encoder.encode(&s).map_err(SyntaErr)?;
249        Ok(())
250    }
251
252    /// Encode a UTF8String from a PyUtf8String object
253    fn encode_utf8_string_object(&mut self, value: &PyUtf8String) -> PyResult<()> {
254        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
255        Ok(())
256    }
257
258    /// Encode a PrintableString
259    ///
260    /// Args:
261    ///     value: The string to encode (must contain only PrintableString-valid characters)
262    fn encode_printable_string(&mut self, value: &str) -> PyResult<()> {
263        let s = synta::PrintableString::new(value.to_string()).map_err(|e| {
264            pyo3::exceptions::PyValueError::new_err(format!("Invalid PrintableString: {:?}", e))
265        })?;
266        self.encoder.encode(&s).map_err(SyntaErr)?;
267        Ok(())
268    }
269
270    /// Encode a PrintableString from a PyPrintableString object
271    fn encode_printable_string_object(&mut self, value: &PyPrintableString) -> PyResult<()> {
272        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
273        Ok(())
274    }
275
276    /// Encode an IA5String
277    ///
278    /// Args:
279    ///     value: The string to encode (must be ASCII only)
280    fn encode_ia5_string(&mut self, value: &str) -> PyResult<()> {
281        let s = synta::IA5String::new(value.to_string()).map_err(|e| {
282            pyo3::exceptions::PyValueError::new_err(format!("Invalid IA5String: {:?}", e))
283        })?;
284        self.encoder.encode(&s).map_err(SyntaErr)?;
285        Ok(())
286    }
287
288    /// Encode an IA5String from a PyIA5String object
289    fn encode_ia5_string_object(&mut self, value: &PyIA5String) -> PyResult<()> {
290        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
291        Ok(())
292    }
293
294    /// Encode a NumericString (tag 18)
295    fn encode_numeric_string(&mut self, value: &str) -> PyResult<()> {
296        let s = synta::NumericString::new(value.to_string()).map_err(|e| {
297            pyo3::exceptions::PyValueError::new_err(format!("Invalid NumericString: {:?}", e))
298        })?;
299        self.encoder.encode(&s).map_err(SyntaErr)?;
300        Ok(())
301    }
302
303    /// Encode a NumericString from a PyNumericString object
304    fn encode_numeric_string_object(&mut self, value: &PyNumericString) -> PyResult<()> {
305        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
306        Ok(())
307    }
308
309    /// Encode a TeletexString / T61String (tag 20) from raw bytes
310    fn encode_teletex_string(&mut self, data: &[u8]) -> PyResult<()> {
311        let s = synta::TeletexString::new(data.to_vec());
312        self.encoder.encode(&s).map_err(SyntaErr)?;
313        Ok(())
314    }
315
316    /// Encode a TeletexString from a PyTeletexString object
317    fn encode_teletex_string_object(&mut self, value: &PyTeletexString) -> PyResult<()> {
318        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
319        Ok(())
320    }
321
322    /// Encode a VisibleString (tag 26)
323    fn encode_visible_string(&mut self, value: &str) -> PyResult<()> {
324        let s = synta::VisibleString::new(value.to_string()).map_err(|e| {
325            pyo3::exceptions::PyValueError::new_err(format!("Invalid VisibleString: {:?}", e))
326        })?;
327        self.encoder.encode(&s).map_err(SyntaErr)?;
328        Ok(())
329    }
330
331    /// Encode a VisibleString from a PyVisibleString object
332    fn encode_visible_string_object(&mut self, value: &PyVisibleString) -> PyResult<()> {
333        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
334        Ok(())
335    }
336
337    /// Encode a GeneralString (tag 27) from raw bytes
338    fn encode_general_string(&mut self, data: &[u8]) -> PyResult<()> {
339        let s = synta::GeneralString::new(data.to_vec());
340        self.encoder.encode(&s).map_err(SyntaErr)?;
341        Ok(())
342    }
343
344    /// Encode a GeneralString from a PyGeneralString object
345    fn encode_general_string_object(&mut self, value: &PyGeneralString) -> PyResult<()> {
346        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
347        Ok(())
348    }
349
350    /// Encode a UniversalString (tag 28)
351    fn encode_universal_string(&mut self, value: &str) -> PyResult<()> {
352        let s = synta::UniversalString::new(value.to_string());
353        self.encoder.encode(&s).map_err(SyntaErr)?;
354        Ok(())
355    }
356
357    /// Encode a UniversalString from a PyUniversalString object
358    fn encode_universal_string_object(&mut self, value: &PyUniversalString) -> PyResult<()> {
359        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
360        Ok(())
361    }
362
363    /// Encode a BMPString (tag 30)
364    ///
365    /// Raises ValueError if the string contains code points outside the BMP (> U+FFFF).
366    fn encode_bmp_string(&mut self, value: &str) -> PyResult<()> {
367        let s = synta::BmpString::new(value.to_string()).map_err(|e| {
368            pyo3::exceptions::PyValueError::new_err(format!("Invalid BMPString: {:?}", e))
369        })?;
370        self.encoder.encode(&s).map_err(SyntaErr)?;
371        Ok(())
372    }
373
374    /// Encode a BMPString from a PyBmpString object
375    fn encode_bmp_string_object(&mut self, value: &PyBmpString) -> PyResult<()> {
376        self.encoder.encode(&value.inner).map_err(SyntaErr)?;
377        Ok(())
378    }
379
380    /// Wrap pre-encoded bytes in a SEQUENCE TLV (tag 0x30) and append to the output.
381    ///
382    /// Use this to build SEQUENCE structures: encode the inner elements into a
383    /// separate Encoder, call ``finish()`` to get the inner bytes, then pass those
384    /// bytes to ``encode_sequence()`` on the outer Encoder.
385    ///
386    /// Example::
387    ///
388    ///     inner = synta.Encoder(synta.Encoding.DER)
389    ///     inner.encode_integer(42)
390    ///     inner.encode_boolean(True)
391    ///     outer = synta.Encoder(synta.Encoding.DER)
392    ///     outer.encode_sequence(inner.finish())
393    ///     result = outer.finish()
394    fn encode_sequence(&mut self, inner_bytes: &[u8]) -> PyResult<()> {
395        use synta::tag::TAG_SEQUENCE;
396        self.encoder
397            .write_tag(synta::Tag::universal_constructed(TAG_SEQUENCE))
398            .map_err(SyntaErr)?;
399        self.encoder
400            .write_length(inner_bytes.len())
401            .map_err(SyntaErr)?;
402        self.encoder.write_bytes(inner_bytes);
403        Ok(())
404    }
405
406    /// Wrap pre-encoded bytes in a SET TLV (tag 0x31) and append to the output.
407    fn encode_set(&mut self, inner_bytes: &[u8]) -> PyResult<()> {
408        use synta::tag::TAG_SET;
409        self.encoder
410            .write_tag(synta::Tag::universal_constructed(TAG_SET))
411            .map_err(SyntaErr)?;
412        self.encoder
413            .write_length(inner_bytes.len())
414            .map_err(SyntaErr)?;
415        self.encoder.write_bytes(inner_bytes);
416        Ok(())
417    }
418
419    /// Wrap pre-encoded bytes with an explicit context tag ``[tag_num]`` and append
420    /// to the output.
421    ///
422    /// ``tag_class`` must be one of ``"Context"`` (default for ``[N]`` notation),
423    /// ``"Application"``, or ``"Private"``.
424    ///
425    /// Example — encode ``[1] EXPLICIT SEQUENCE { ... }``::
426    ///
427    ///     enc.encode_explicit_tag(1, "Context", sequence_bytes)
428    fn encode_explicit_tag(
429        &mut self,
430        tag_num: u32,
431        tag_class: &str,
432        inner_bytes: &[u8],
433    ) -> PyResult<()> {
434        use pyo3::exceptions::PyValueError;
435        use synta::tag::TagClass;
436
437        let class = match tag_class {
438            "Context" => TagClass::ContextSpecific,
439            "Application" => TagClass::Application,
440            "Private" => TagClass::Private,
441            other => {
442                return Err(PyValueError::new_err(format!(
443                    "Unknown tag class '{}'; expected 'Context', 'Application', or 'Private'",
444                    other
445                )))
446            }
447        };
448        // Explicit tags are always constructed (they wrap another TLV)
449        let tag = synta::Tag::new(class, true, tag_num);
450        self.encoder.write_tag(tag).map_err(SyntaErr)?;
451        self.encoder
452            .write_length(inner_bytes.len())
453            .map_err(SyntaErr)?;
454        self.encoder.write_bytes(inner_bytes);
455        Ok(())
456    }
457
458    /// Wrap pre-encoded value bytes with an implicit tag and append to the output.
459    ///
460    /// For implicit tagging, the original type tag is replaced with the given
461    /// tag; the value bytes (the content, *not* the full original TLV) are
462    /// written as-is.  Set ``is_constructed=True`` when the underlying type is
463    /// a SEQUENCE, SET, or other constructed type.
464    ///
465    /// ``tag_class`` must be one of ``"Context"`` (default for ``[N]`` notation),
466    /// ``"Application"``, or ``"Private"``.
467    ///
468    /// Example — encode ``[1] IMPLICIT INTEGER`` with value bytes ``b'\\x2a'``:
469    ///
470    ///     enc.encode_implicit_tag(1, "Context", False, b'\\x2a')
471    fn encode_implicit_tag(
472        &mut self,
473        tag_num: u32,
474        tag_class: &str,
475        is_constructed: bool,
476        value_bytes: &[u8],
477    ) -> PyResult<()> {
478        use pyo3::exceptions::PyValueError;
479        use synta::tag::TagClass;
480
481        let class = match tag_class {
482            "Context" => TagClass::ContextSpecific,
483            "Application" => TagClass::Application,
484            "Private" => TagClass::Private,
485            other => {
486                return Err(PyValueError::new_err(format!(
487                    "Unknown tag class '{other}'; expected 'Context', 'Application', or 'Private'",
488                )))
489            }
490        };
491        let tag = synta::Tag::new(class, is_constructed, tag_num);
492        self.encoder.write_tag(tag).map_err(SyntaErr)?;
493        self.encoder
494            .write_length(value_bytes.len())
495            .map_err(SyntaErr)?;
496        self.encoder.write_bytes(value_bytes);
497        Ok(())
498    }
499
500    /// Finish encoding and return the encoded bytes
501    ///
502    /// Returns:
503    ///     The encoded ASN.1 data as bytes
504    fn finish<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, pyo3::types::PyBytes>> {
505        // Capture encoding before the move so the replacement uses the same rules.
506        let encoding = self.encoder.encoding();
507        let encoder = std::mem::replace(&mut self.encoder, Encoder::new(encoding));
508        let bytes = encoder.finish().map_err(SyntaErr)?;
509        Ok(pyo3::types::PyBytes::new(py, &bytes))
510    }
511
512    fn __repr__(&self) -> String {
513        format!("Encoder(encoding={:?})", self.encoder.encoding())
514    }
515}