Add `SET|SEQUENCE OF ANY` encoding support (#165)

For example:

  AttributeTypeAndValues ::= SEQUENCE {
    type OBJECT IDENTIFIER,
    values SET OF ANY DEFINED BY type
  }

This patch adds support of the above ASN.1 syntax to BER/DER/CER
codecs.

It appears that to implement this feature properly, `SetOf`/`SequenceOf`
pyasn1 types need to have `.componentType` wrapped into something
similar to `NamedType` that `Set`/`Sequence` have. That additional
layer would then carry the open type meta information. Without it,
`Sequence`/`Set` codec needs to signal `SetOf`/`SequenceOf` codec
of the open type being processed, which is a slight hack.

A other inconvenience is that when `SetOf`/`SequenceOf` deal with
an open type component, they should not verify types on component
assignment. Without open type property in `SetOf`/`SequenceOf`,
the code checks for `Any` component type which is another hack.

The above shortcomings should be addressed in the follow up patch.

2 files changed, 285 insertions, 8 deletions

diff --git a/tests/codec/ber/test_decoder.py b/tests/codec/ber/test_decoder.py
index bff246e..95f6c91 100644
--- a/tests/codec/ber/test_decoder.py
+++ b/tests/codec/ber/test_decoder.py
@@ -972,6 +972,159 @@ class SequenceDecoderWithExplicitlyTaggedOpenTypesTestCase(BaseTestCase):
         assert s[1] == univ.OctetString(hexValue='02010C')
 
 
+class SequenceDecoderWithUnaggedSetOfOpenTypesTestCase(BaseTestCase):
+    def setUp(self):
+        openType = opentype.OpenType(
+            'id',
+            {1: univ.Integer(),
+             2: univ.OctetString()}
+        )
+        self.s = univ.Sequence(
+            componentType=namedtype.NamedTypes(
+                namedtype.NamedType('id', univ.Integer()),
+                namedtype.NamedType('blob', univ.SetOf(componentType=univ.Any()),
+                                    openType=openType)
+            )
+        )
+
+    def testDecodeOpenTypesChoiceOne(self):
+        s, r = decoder.decode(
+            ints2octs((48, 8, 2, 1, 1, 49, 3, 2, 1, 12)), asn1Spec=self.s,
+            decodeOpenTypes=True
+        )
+        assert not r
+        assert s[0] == 1
+        assert s[1][0] == 12
+
+    def testDecodeOpenTypesChoiceTwo(self):
+        s, r = decoder.decode(
+            ints2octs((48, 18, 2, 1, 2, 49, 13, 4, 11, 113, 117, 105, 99,
+                       107, 32, 98, 114, 111, 119, 110)), asn1Spec=self.s,
+            decodeOpenTypes=True
+        )
+        assert not r
+        assert s[0] == 2
+        assert s[1][0] == univ.OctetString('quick brown')
+
+    def testDecodeOpenTypesUnknownType(self):
+        try:
+            s, r = decoder.decode(
+                ints2octs((48, 6, 2, 1, 2, 6, 1, 39)), asn1Spec=self.s,
+                decodeOpenTypes=True
+            )
+
+        except PyAsn1Error:
+            pass
+
+        else:
+            assert False, 'unknown open type tolerated'
+
+    def testDecodeOpenTypesUnknownId(self):
+        s, r = decoder.decode(
+            ints2octs((48, 8, 2, 1, 3, 49, 3, 2, 1, 12)), asn1Spec=self.s,
+            decodeOpenTypes=True
+        )
+        assert not r
+        assert s[0] == 3
+        assert s[1][0] == univ.OctetString(hexValue='02010c')
+
+    def testDontDecodeOpenTypesChoiceOne(self):
+        s, r = decoder.decode(
+            ints2octs((48, 8, 2, 1, 1, 49, 3, 2, 1, 12)), asn1Spec=self.s
+        )
+        assert not r
+        assert s[0] == 1
+        assert s[1][0] == ints2octs((2, 1, 12))
+
+    def testDontDecodeOpenTypesChoiceTwo(self):
+        s, r = decoder.decode(
+            ints2octs((48, 18, 2, 1, 2, 49, 13, 4, 11, 113, 117, 105, 99,
+                       107, 32, 98, 114, 111, 119, 110)), asn1Spec=self.s
+        )
+        assert not r
+        assert s[0] == 2
+        assert s[1][0] == ints2octs((4, 11, 113, 117, 105, 99, 107, 32, 98, 114,
+                                     111, 119, 110))
+
+
+class SequenceDecoderWithImplicitlyTaggedSetOfOpenTypesTestCase(BaseTestCase):
+    def setUp(self):
+        openType = opentype.OpenType(
+            'id',
+            {1: univ.Integer(),
+             2: univ.OctetString()}
+        )
+        self.s = univ.Sequence(
+            componentType=namedtype.NamedTypes(
+                namedtype.NamedType('id', univ.Integer()),
+                namedtype.NamedType(
+                    'blob', univ.SetOf(
+                        componentType=univ.Any().subtype(
+                            implicitTag=tag.Tag(
+                                tag.tagClassContext, tag.tagFormatSimple, 3))),
+                    openType=openType
+                )
+            )
+        )
+
+    def testDecodeOpenTypesChoiceOne(self):
+        s, r = decoder.decode(
+            ints2octs((48, 10, 2, 1, 1, 49, 5, 131, 3, 2, 1, 12)),
+            asn1Spec=self.s, decodeOpenTypes=True
+        )
+        assert not r
+        assert s[0] == 1
+        assert s[1][0] == 12
+
+    def testDecodeOpenTypesUnknownId(self):
+        s, r = decoder.decode(
+            ints2octs((48, 10, 2, 1, 3, 49, 5, 131, 3, 2, 1, 12)),
+            asn1Spec=self.s, decodeOpenTypes=True
+        )
+        assert not r
+        assert s[0] == 3
+        assert s[1][0] == univ.OctetString(hexValue='02010C')
+
+
+class SequenceDecoderWithExplicitlyTaggedSetOfOpenTypesTestCase(BaseTestCase):
+    def setUp(self):
+        openType = opentype.OpenType(
+            'id',
+            {1: univ.Integer(),
+             2: univ.OctetString()}
+        )
+        self.s = univ.Sequence(
+            componentType=namedtype.NamedTypes(
+                namedtype.NamedType('id', univ.Integer()),
+                namedtype.NamedType(
+                    'blob', univ.SetOf(
+                        componentType=univ.Any().subtype(
+                            explicitTag=tag.Tag(
+                                tag.tagClassContext, tag.tagFormatSimple, 3))),
+                    openType=openType
+                )
+            )
+        )
+
+    def testDecodeOpenTypesChoiceOne(self):
+        s, r = decoder.decode(
+            ints2octs((48, 10, 2, 1, 1, 49, 5, 131, 3, 2, 1, 12)),
+            asn1Spec=self.s, decodeOpenTypes=True
+        )
+        assert not r
+        assert s[0] == 1
+        assert s[1][0] == 12
+
+    def testDecodeOpenTypesUnknownId(self):
+        s, r = decoder.decode(
+            ints2octs( (48, 10, 2, 1, 3, 49, 5, 131, 3, 2, 1, 12)),
+            asn1Spec=self.s, decodeOpenTypes=True
+        )
+        assert not r
+        assert s[0] == 3
+        assert s[1][0] == univ.OctetString(hexValue='02010C')
+
+
 class SetDecoderTestCase(BaseTestCase):
     def setUp(self):
         BaseTestCase.setUp(self)
@@ -1439,8 +1592,8 @@ class ErrorOnDecodingTestCase(BaseTestCase):
 
         except PyAsn1Error:
             exc = sys.exc_info()[1]
-            assert (isinstance(exc, PyAsn1Error),
-                    'Unexpected exception raised %r' % (exc,))
+            assert isinstance(exc, PyAsn1Error), (
+                'Unexpected exception raised %r' % (exc,))
 
         else:
             assert False, 'Unexpected decoder result %r' % (asn1Object,)
@@ -1453,12 +1606,12 @@ class ErrorOnDecodingTestCase(BaseTestCase):
         asn1Object, rest = decode(ints2octs(
             (31, 8, 2, 1, 1, 131, 3, 2, 1, 12)))
 
-        assert (isinstance(asn1Object, univ.Any),
-                'Unexpected raw dump type %r' % (asn1Object,))
-        assert (asn1Object.asNumbers() == (31, 8, 2, 1, 1),
-                'Unexpected raw dump value %r' % (asn1Object,))
-        assert (rest == ints2octs((131, 3, 2, 1, 12)),
-                'Unexpected rest of substrate after raw dump %r' % rest)
+        assert isinstance(asn1Object, univ.Any), (
+            'Unexpected raw dump type %r' % (asn1Object,))
+        assert asn1Object.asNumbers() == (31, 8, 2, 1, 1), (
+            'Unexpected raw dump value %r' % (asn1Object,))
+        assert rest == ints2octs((131, 3, 2, 1, 12)), (
+            'Unexpected rest of substrate after raw dump %r' % rest)
 
 
 suite = unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
diff --git a/tests/codec/ber/test_encoder.py b/tests/codec/ber/test_encoder.py
index da14830..26236af 100644
--- a/tests/codec/ber/test_encoder.py
+++ b/tests/codec/ber/test_encoder.py
@@ -854,6 +854,130 @@ class SequenceEncoderWithExplicitlyTaggedOpenTypesTestCase(BaseTestCase):
         )
 
 
+class SequenceEncoderWithUntaggedSetOfOpenTypesTestCase(BaseTestCase):
+    def setUp(self):
+        BaseTestCase.setUp(self)
+
+        openType = opentype.OpenType(
+            'id',
+            {1: univ.Integer(),
+             2: univ.OctetString()}
+        )
+        self.s = univ.Sequence(
+            componentType=namedtype.NamedTypes(
+                namedtype.NamedType('id', univ.Integer()),
+                namedtype.NamedType('blob', univ.SetOf(
+                    componentType=univ.Any()), openType=openType)
+            )
+        )
+
+    def testEncodeOpenTypeChoiceOne(self):
+        self.s.clear()
+
+        self.s[0] = 1
+        self.s[1].append(univ.Integer(12))
+
+        assert encoder.encode(self.s, asn1Spec=self.s) == ints2octs(
+            (48, 8, 2, 1, 1, 49, 3, 2, 1, 12)
+        )
+
+    def testEncodeOpenTypeChoiceTwo(self):
+        self.s.clear()
+
+        self.s[0] = 2
+        self.s[1].append(univ.OctetString('quick brown'))
+
+        assert encoder.encode(self.s, asn1Spec=self.s) == ints2octs(
+            (48, 18, 2, 1, 2, 49, 13, 4, 11, 113, 117, 105, 99,
+             107, 32, 98, 114, 111, 119, 110)
+        )
+
+    def testEncodeOpenTypeUnknownId(self):
+        self.s.clear()
+
+        self.s[0] = 2
+        self.s[1].append(univ.ObjectIdentifier('1.3.6'))
+
+        try:
+            encoder.encode(self.s, asn1Spec=self.s)
+
+        except PyAsn1Error:
+            assert False, 'incompatible open type tolerated'
+
+    def testEncodeOpenTypeIncompatibleType(self):
+        self.s.clear()
+
+        self.s[0] = 2
+        self.s[1].append(univ.ObjectIdentifier('1.3.6'))
+
+        try:
+            encoder.encode(self.s, asn1Spec=self.s)
+
+        except PyAsn1Error:
+            assert False, 'incompatible open type tolerated'
+
+
+class SequenceEncoderWithImplicitlyTaggedSetOfOpenTypesTestCase(BaseTestCase):
+    def setUp(self):
+        BaseTestCase.setUp(self)
+
+        openType = opentype.OpenType(
+            'id',
+            {1: univ.Integer(),
+             2: univ.OctetString()}
+        )
+        self.s = univ.Sequence(
+            componentType=namedtype.NamedTypes(
+                namedtype.NamedType('id', univ.Integer()),
+                namedtype.NamedType('blob', univ.SetOf(
+                    componentType=univ.Any().subtype(
+                        implicitTag=tag.Tag(
+                            tag.tagClassContext, tag.tagFormatSimple, 3))),
+                    openType=openType)
+            )
+        )
+
+    def testEncodeOpenTypeChoiceOne(self):
+        self.s.clear()
+
+        self.s[0] = 1
+        self.s[1].append(univ.Integer(12))
+
+        assert encoder.encode(self.s, asn1Spec=self.s) == ints2octs(
+            (48, 10, 2, 1, 1, 49, 5, 131, 3, 2, 1, 12)
+        )
+
+
+class SequenceEncoderWithExplicitlyTaggedSetOfOpenTypesTestCase(BaseTestCase):
+    def setUp(self):
+        BaseTestCase.setUp(self)
+
+        openType = opentype.OpenType(
+            'id',
+            {1: univ.Integer(),
+             2: univ.OctetString()}
+        )
+        self.s = univ.Sequence(
+            componentType=namedtype.NamedTypes(
+                namedtype.NamedType('id', univ.Integer()),
+                namedtype.NamedType('blob', univ.SetOf(
+                    componentType=univ.Any().subtype(
+                        explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 3))),
+                    openType=openType)
+            )
+        )
+
+    def testEncodeOpenTypeChoiceOne(self):
+        self.s.clear()
+
+        self.s[0] = 1
+        self.s[1].append(univ.Integer(12))
+
+        assert encoder.encode(self.s, asn1Spec=self.s) == ints2octs(
+            (48, 10, 2, 1, 1, 49, 5, 163, 3, 2, 1, 12)
+        )
+
+
 class SequenceEncoderWithComponentsSchemaTestCase(BaseTestCase):
     def setUp(self):
         BaseTestCase.setUp(self)