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

_synta/
encoder.rs
