Support python-asn1 3.x

.github/workflows/python-tests.yml

@@ -18,6 +18,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        asn1-version: ["asn1<3", "asn1>=3"]
 
     steps:
     - uses: actions/checkout@v4
@@ -33,5 +34,8 @@ jobs:
     - name: Install the project
       run: uv sync --dev
 
+    - name: Install python-asn1 (${{ matrix.asn1-version }})
+      run: uv pip install '${{ matrix.asn1-version }}'
+
     - name: Run tests
-      run: uv run pytest
+      run: uv run --no-sync pytest

pyimg4/_asn1compat.py

@@ -0,0 +1,108 @@
+"""Compatibility layer over python-asn1 2.x and 3.x.
+
+python-asn1 3.0 changed behavior PyIMG4 depends on
+(https://github.com/andrivet/python-asn1/issues/297):
+
+- ``Decoder.read()`` on a constructed element recursively decodes it and
+  returns a ``list`` instead of the raw content octets.
+- ``Encoder.write()`` with ``Types.Constructed`` iterates the value and
+  encodes each item instead of emitting pre-encoded octets verbatim.
+- ``Decoder.eof()`` is relative to the whole input instead of the entered
+  container, and ``Decoder.leave()`` no longer skips unread elements.
+
+Image4 requires byte-exact round-trips of nested DER (signed manifest
+bodies, X.509 certificates), so the raw-octet behavior of 2.x is recreated
+here. The 3.x public API cannot read or emit raw pre-encoded content, so
+that branch uses private python-asn1 internals; their presence is checked
+at import time so an incompatible future release fails loudly instead of
+producing corrupt output.
+"""
+
+import asn1
+
+
+def _detect_legacy() -> bool:
+    # asn1 2.x emits bytes passed to a constructed write() verbatim as the
+    # element content; 3.x re-encodes each byte as an ASN.1 Integer.
+    encoder = asn1.Encoder()
+    encoder.start()
+    try:
+        encoder.write(
+            b'\x02\x01\x00',
+            asn1.Numbers.Sequence,
+            asn1.Types.Constructed,
+            asn1.Classes.Universal,
+        )
+        return encoder.output() == b'\x30\x03\x02\x01\x00'
+    except Exception:
+        return False
+
+
+_LEGACY = _detect_legacy()
+
+if not _LEGACY:
+    _REQUIRED_INTERNALS = (
+        (asn1.Decoder, '_decode_length'),
+        (asn1.Decoder, '_read_bytes'),
+        (asn1.Encoder, '_emit_tag'),
+        (asn1.Encoder, '_emit_length'),
+        (asn1.Encoder, '_emit'),
+    )
+    if any(
+        not callable(getattr(owner, name, None)) for owner, name in _REQUIRED_INTERNALS
+    ):
+        raise ImportError(
+            'The installed python-asn1 version is not supported by pyimg4: it '
+            'neither behaves like the 2.x API nor exposes the 3.x internals '
+            'pyimg4 relies on. Please report this at '
+            'https://github.com/m1stadev/PyIMG4/issues.'
+        )
+
+
+def _read_raw(decoder: asn1.Decoder) -> bytes:
+    # Replicates asn1 3.x Decoder.read() up to (and without) value decoding.
+    tag = decoder.peek()
+    if tag is None:
+        raise asn1.Error('ASN1 decoding error: premature end of input.')
+
+    decoder._tag = None
+    length = decoder._decode_length(tag.typ)
+    if length < 0:
+        raise asn1.Error('ASN1 decoding error: indefinite lengths are not supported.')
+
+    return decoder._read_bytes(length)
+
+
+def _read_constructed(decoder: asn1.Decoder) -> bytes:
+    """Read the next (constructed) element, returning its raw content octets."""
+    if _LEGACY:
+        return decoder.read()[1]
+
+    return _read_raw(decoder)
+
+
+def _write_constructed(
+    encoder: asn1.Encoder, content: bytes, nr: int, cls: int
+) -> None:
+    """Write pre-encoded content octets wrapped in a constructed tag."""
+    if _LEGACY:
+        encoder.write(content, nr, asn1.Types.Constructed, cls)
+        return
+
+    encoder._emit_tag(nr, asn1.Types.Constructed, cls)
+    encoder._emit_length(len(content))
+    encoder._emit(content)
+
+
+def _at_end(decoder: asn1.Decoder) -> bool:
+    """Return whether the current container (or input) has no elements left."""
+    return decoder.peek() is None
+
+
+def _leave(decoder: asn1.Decoder) -> None:
+    """Skip any unread elements, then leave the current container."""
+    if not _LEGACY:
+        while decoder.peek() is not None:
+            _read_raw(decoder)
+
+    decoder.leave()

pyimg4/parser.py

@@ -7,6 +7,7 @@
 import lzss
 from Crypto.Cipher import AES
 
+from ._asn1compat import _at_end, _leave, _read_constructed, _write_constructed
 from .errors import CompressionError, UnexpectedDataError, UnexpectedTagError
 from .types import Compression, KeybagType, Payload
 
@@ -187,8 +188,8 @@ def _parse(self) -> None:
 
         self._decoder.enter()
 
-        while not self._decoder.eof():
-            self._properties.append(self._property(self._decoder.read()[1]))
+        while not _at_end(self._decoder):
+            self._properties.append(self._property(_read_constructed(self._decoder)))
 
     @property
     def fourcc(self) -> str:
@@ -256,10 +257,10 @@ def output(self) -> bytes:
                             self._decoder.peek().nr, asn1.Classes.Private
                         ):
                             self._decoder.enter()
-                            self._encoder.write(
-                                self._decoder.read()[1],
+                            _write_constructed(
+                                self._encoder,
+                                _read_constructed(self._decoder),
                                 asn1.Numbers.Sequence,
-                                asn1.Types.Constructed,
                                 asn1.Classes.Universal,
                             )
 
@@ -355,17 +356,17 @@ def _parse(self) -> None:
             raise UnexpectedTagError(self._decoder.peek(), asn1.Numbers.Set)
 
         self._decoder.enter()
-        while not self._decoder.eof():
-            data = ManifestImageProperties(self._decoder.read()[1])
+        while not _at_end(self._decoder):
+            data = ManifestImageProperties(_read_constructed(self._decoder))
             if data.fourcc == 'MANP':
                 self._properties = list(data.properties)
             else:
                 self._images.append(data)
 
         for _ in range(4):
-            self._decoder.leave()
+            _leave(self._decoder)
 
-        if self._decoder.eof():
+        if _at_end(self._decoder):
             # IM4M has no signature/certificates
             # TODO: Make this cleaner
             self._signature = None
@@ -381,9 +382,9 @@ def _parse(self) -> None:
             raise UnexpectedTagError(self._decoder.peek(), asn1.Numbers.Sequence)
 
         # TODO: Parse certificates
-        self._certificates = self._decoder.read()[1]
+        self._certificates = _read_constructed(self._decoder)
 
-        if not self._decoder.eof():
+        if not _at_end(self._decoder):
             raise ValueError(
                 f'Unexpected data found at end of Image4 manifest: {self._decoder.peek().nr.name.upper()}'
             )
@@ -553,10 +554,10 @@ def output(self) -> bytes:
                     self._decoder.peek().nr, asn1.Classes.Private
                 ):
                     self._decoder.enter()
-                    self._encoder.write(
-                        self._decoder.read()[1],
+                    _write_constructed(
+                        self._encoder,
+                        _read_constructed(self._decoder),
                         asn1.Numbers.Sequence,
-                        asn1.Types.Constructed,
                         asn1.Classes.Universal,
                     )
 
@@ -567,10 +568,10 @@ def output(self) -> bytes:
                 asn1.Classes.Universal,
             )
 
-            self._encoder.write(
+            _write_constructed(
+                self._encoder,
                 self.certificates,
                 asn1.Numbers.Sequence,
-                asn1.Types.Constructed,
                 asn1.Classes.Universal,
             )
         return self._encoder.output()
@@ -636,10 +637,10 @@ def output(self) -> bytes:
                         self._decoder.peek().nr, asn1.Classes.Private
                     ):
                         self._decoder.enter()
-                        self._encoder.write(
-                            self._decoder.read()[1],
+                        _write_constructed(
+                            self._encoder,
+                            _read_constructed(self._decoder),
                             asn1.Numbers.Sequence,
-                            asn1.Types.Constructed,
                             asn1.Classes.Universal,
                         )
 
@@ -683,28 +684,28 @@ def _parse(self) -> None:
         if self._decoder.peek().nr != asn1.Numbers.Sequence:
             raise UnexpectedTagError(self._decoder.peek(), asn1.Numbers.Sequence)
 
-        self._encoder.write(
-            self._decoder.read()[1],
+        _write_constructed(
+            self._encoder,
+            _read_constructed(self._decoder),
             asn1.Numbers.Sequence,
-            asn1.Types.Constructed,
             asn1.Classes.Universal,
         )
         self.im4p = IM4P(self._encoder.output())  # IM4P
 
         if self._decoder.peek().cls != asn1.Classes.Context:
             raise UnexpectedTagError(self._decoder.peek(), asn1.Classes.Context)
 
-        self.im4m = IM4M(self._decoder.read()[1])  # IM4M
+        self.im4m = IM4M(_read_constructed(self._decoder))  # IM4M
 
-        if self._decoder.eof():
+        if _at_end(self._decoder):
             self.im4r = None
 
         elif self._decoder.peek().cls != asn1.Classes.Context:
             raise UnexpectedTagError(self._decoder.peek(), asn1.Classes.Context)
 
         else:
-            self.im4r = IM4R(self._decoder.read()[1])  # IM4R
-        if not self._decoder.eof():
+            self.im4r = IM4R(_read_constructed(self._decoder))  # IM4R
+        if not _at_end(self._decoder):
             raise ValueError(
                 f'Unexpected data found at end of Image4: {self._decoder.peek().nr.name.upper()}'
             )
@@ -757,28 +758,28 @@ def output(self) -> bytes:
                 raise ValueError('No IM4P is set.')
 
             self._decoder.start(self.im4p.output())
-            self._encoder.write(
-                self._decoder.read()[1],
+            _write_constructed(
+                self._encoder,
+                _read_constructed(self._decoder),
                 asn1.Numbers.Sequence,
-                asn1.Types.Constructed,
                 asn1.Classes.Universal,
             )
 
             if self.im4m is None:
                 raise ValueError('No IM4M is set.')
 
-            self._encoder.write(
+            _write_constructed(
+                self._encoder,
                 self.im4m.output(),
                 0,
-                asn1.Types.Constructed,
                 asn1.Classes.Context,
             )
 
             if self.im4r is not None:
-                self._encoder.write(
+                _write_constructed(
+                    self._encoder,
                     self.im4r.output(),
                     1,
-                    asn1.Types.Constructed,
                     asn1.Classes.Context,
                 )
 
@@ -851,7 +852,7 @@ def _parse(self) -> None:
         self.payload = self._decoder.read()[1]
 
         if (
-            not self._decoder.eof()
+            not _at_end(self._decoder)
             and self._decoder.peek().nr == asn1.Numbers.OctetString
         ):
             kbag_decoder = asn1.Decoder()
@@ -862,13 +863,16 @@ def _parse(self) -> None:
 
             kbag_decoder.enter()
 
-            while not kbag_decoder.eof():
+            while not _at_end(kbag_decoder):
                 if kbag_decoder.peek().nr != asn1.Numbers.Sequence:
                     raise UnexpectedTagError(kbag_decoder.peek(), asn1.Numbers.Sequence)
 
-                self.payload.add_keybag(Keybag(kbag_decoder.read()[1]))
+                self.payload.add_keybag(Keybag(_read_constructed(kbag_decoder)))
 
-        if not self._decoder.eof() and self._decoder.peek().nr == asn1.Numbers.Sequence:
+        if (
+            not _at_end(self._decoder)
+            and self._decoder.peek().nr == asn1.Numbers.Sequence
+        ):
             self._decoder.enter()
 
             if (
@@ -877,9 +881,12 @@ def _parse(self) -> None:
             ):
                 self.payload.size = self._decoder.read()[1]
 
-            self._decoder.leave()
+            _leave(self._decoder)
 
-        if not self._decoder.eof() and self._decoder.peek().cls == asn1.Classes.Context:
+        if (
+            not _at_end(self._decoder)
+            and self._decoder.peek().cls == asn1.Classes.Context
+        ):
             self._decoder.enter()
 
             if self._decoder.peek().nr != asn1.Numbers.Sequence:
@@ -892,10 +899,12 @@ def _parse(self) -> None:
                 raise UnexpectedTagError(self._decoder.peek(), asn1.Numbers.Set)
 
             self._decoder.enter()
-            while not self._decoder.eof():
-                self._properties.append(PayloadProperty(self._decoder.read()[1]))
+            while not _at_end(self._decoder):
+                self._properties.append(
+                    PayloadProperty(_read_constructed(self._decoder))
+                )
 
-        if not self._decoder.eof():
+        if not _at_end(self._decoder):
             raise ValueError(
                 f'Unexpected data found at end of Image4 payload: {self._decoder.peek().nr.name.upper()}'
             )
@@ -1054,10 +1063,10 @@ def output(self) -> bytes:
                                     self._decoder.peek().nr, asn1.Classes.Private
                                 ):
                                     self._decoder.enter()
-                                    self._encoder.write(
-                                        self._decoder.read()[1],
+                                    _write_constructed(
+                                        self._encoder,
+                                        _read_constructed(self._decoder),
                                         asn1.Numbers.Sequence,
-                                        asn1.Types.Constructed,
                                         asn1.Classes.Universal,
                                     )
 
@@ -1109,7 +1118,7 @@ def _parse(self) -> None:
 
         self.key = self._decoder.read()[1]
 
-        if not self._decoder.eof():
+        if not _at_end(self._decoder):
             raise ValueError(
                 f'Unexpected data found at end of keybag: {self._decoder.peek().nr.name.upper()}'
             )

pyproject.toml

@@ -17,7 +17,6 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "asn1>=2.7.0",
-    "asn1<3.0.0",
     "click>=8.1.7",
     "pycryptodome>=3.18.0",
     "pylzss==0.3.4",