Raise UnknownBERTag instead of printing and hanging (fixes #170, #240, #243)

docs/source/NEWS.rst

@@ -43,6 +43,12 @@ Fixes
 - Fixed ``LDAPEntry.toWire`` including the ``objectClass`` attribute
   twice when the attribute key was stored as ``bytes`` rather than
   ``str`` (#220).
+- ``pureber.berDecodeObject`` now raises ``UnknownBERTag`` for
+  unregistered top-level tags instead of writing to stdout and returning
+  ``None``. Twisted then tears down the transport, so a plaintext LDAP
+  server accessed over TLS (or any other framing mismatch) fails fast
+  rather than hanging the client. Also removes a stray ``print`` from
+  production code (#170, #240, #243).
 
 
 21.2.0 (2021-02-28)

ldaptor/protocols/pureber.py

@@ -379,8 +379,7 @@ def berDecodeObject(context, m):
             r = berclass.fromBER(tag=i, content=m2, berdecoder=inh)
             return (r, 1 + lenlen + length)
         else:
-            print(str(UnknownBERTag(i, context)))  # TODO
-            return (None, 1 + lenlen + length)
+            raise UnknownBERTag(i, context)
     return (None, 0)
 
 

ldaptor/protocols/pureldap.py

@@ -673,7 +673,9 @@ def asText(self):
         if final is None:
             final = ""
 
-        return "(" + self.type.decode() + "=" + "*".join([initial] + any + [final]) + ")"
+        return (
+            "(" + self.type.decode() + "=" + "*".join([initial] + any + [final]) + ")"
+        )
 
 
 class LDAPFilter_greaterOrEqual(LDAPAttributeValueAssertion):
@@ -854,13 +856,15 @@ class LDAPFilter_extensibleMatch(LDAPMatchingRuleAssertion):
     tag = CLASS_CONTEXT | 0x09
 
     def asText(self):
-        return '(' + \
-               (self.type.value.decode() if self.type else '') + \
-               (':dn' if self.dnAttributes and self.dnAttributes.value else '') + \
-               ((':' + self.matchingRule.value.decode()) if self.matchingRule else '') + \
-               ':=' + \
-               self.escaper(self.matchValue.value.decode()) + \
-               ')'
+        return (
+            "("
+            + (self.type.value.decode() if self.type else "")
+            + (":dn" if self.dnAttributes and self.dnAttributes.value else "")
+            + ((":" + self.matchingRule.value.decode()) if self.matchingRule else "")
+            + ":="
+            + self.escaper(self.matchValue.value.decode())
+            + ")"
+        )
 
 
 class LDAPBERDecoderContext_Filter(BERDecoderContext):

ldaptor/test/test_ldapfilter.py

@@ -152,7 +152,7 @@ def test_escape_backslash(self):
         text = r"(filename=C:\5cMyFile)"
         filt = pureldap.LDAPFilter_equalityMatch(
             attributeDesc=pureldap.LDAPAttributeDescription(value=b"filename"),
-            assertionValue=pureldap.LDAPAssertionValue(value=br"C:\MyFile"),
+            assertionValue=pureldap.LDAPAssertionValue(value=rb"C:\MyFile"),
         )
         self.assertEqual(ldapfilter.parseFilter(text), filt)
         self.assertEqual(filt.asText(), text)

ldaptor/test/test_pureber.py

@@ -562,3 +562,21 @@ def testDecdeInvalidInput(self):
         self.assertEqual(
             (None, 0), pureber.berDecodeObject(pureber.BERDecoderContext(), "")
         )
+
+
+class BERUnknownTag(unittest.TestCase):
+    def test_unknownTagRaises(self):
+        """
+        Decoding a BER stream whose top-level tag is not registered raises
+        UnknownBERTag instead of silently returning ``None``. This lets the
+        transport tear down the connection (e.g. when a plaintext LDAP
+        server is spoken to over TLS) rather than hang.
+        """
+        # Tag 0x16 (BERIA5String) is not in the default LDAP decoder context.
+        m = s(0x16, 0x03, 0x66, 0x6F, 0x6F)
+        self.assertRaises(
+            pureber.UnknownBERTag,
+            pureber.berDecodeObject,
+            pureber.BERDecoderContext(),
+            m,
+        )