Merge pull request #219 from mtury/asn1_hightag

Fix support and add test for ASN.1 (unknown) high-tag-number

scapy/asn1/ber.py

@@ -89,7 +89,8 @@ def BER_num_enc(l, size=1):
             l >>= 7
             size -= 1
         return "".join([chr(k) for k in x])
-def BER_num_dec(s, x=0):
+def BER_num_dec(s, cls_id=0):
+        x = cls_id
         for i, c in enumerate(s):
             c = ord(c)
             x <<= 7
@@ -100,41 +101,71 @@ def BER_num_dec(s, x=0):
             raise BER_Decoding_Error("BER_num_dec: unfinished number description", remaining=s)
         return x, s[i+1:]
 
-# The function below provides low-tag and high-tag identifier partial support.
-# Class and primitive/constructed bit decoding is not supported yet.
-# For now Scapy considers this information to always be part of the identifier
-# e.g. we need BER_id_dec("\x30") to be 0x30 so that Scapy calls a SEQUENCE,
-# even though the real id is 0x10 once bit 6 (constructed) has been removed.
 def BER_id_dec(s):
+    # This returns the tag ALONG WITH THE PADDED CLASS+CONSTRUCTIVE INFO.
+    # Let's recall that bits 8-7 from the first byte of the tag encode
+    # the class information, while bit 6 means primitive or constructive.
+    # For instance, with low-tag-number '\x81', class would be 0b10
+    # ('context-specific') and tag 0x01, but we return 0x81 as a whole.
+    # For '\xff\x02', class would be 0b11 ('private'), constructed, then
+    # padding, then tag 0x02, but we return (0xff>>5)*128^1 + 0x02*128^0.
+    # Why the 5-bit-shifting? Because it provides an unequivocal encoding
+    # on base 128 (note that 0xff would equal 1*128^1 + 127*128^0...),
+    # as we know that bits 5 to 1 are fixed to 1 anyway.
+    # As long as there is no class differentiation, we have to keep this info
+    # encoded in scapy's tag in order to reuse it for packet building.
+    # Note that tags thus may have to be hard-coded with their extended
+    # information, e.g. a SEQUENCE from asn1.py has a direct tag 0x20|16.
         x = ord(s[0])
         if x & 0x1f != 0x1f:
+            # low-tag-number
             return x,s[1:]
-        return BER_num_dec(s[1:], x&0xe0)
+        else:
+            # high-tag-number
+            return BER_num_dec(s[1:], cls_id=x>>5)
+def BER_id_enc(n):
+        if n < 256:
+            # low-tag-number
+            return chr(n)
+        else:
+            # high-tag-number
+            s = BER_num_enc(n)
+            tag = ord(s[0])             # first byte, as an int
+            tag &= 0x07                 # reset every bit from 8 to 4
+            tag <<= 5                   # move back the info bits on top
+            tag |= 0x1f                 # pad with 1s every bit from 5 to 1
+            return chr(tag) + s[1:]
 
 # The functions below provide implicit and explicit tagging support.
 def BER_tagging_dec(s, hidden_tag=None, implicit_tag=None,
                     explicit_tag=None, safe=False):
+    # We output the 'real_tag' if it is different from the (im|ex)plicit_tag.
+    real_tag = None
     if len(s) > 0:
         err_msg = "BER_tagging_dec: observed tag does not match expected tag"
         if implicit_tag is not None:
             ber_id,s = BER_id_dec(s)
             if ber_id != implicit_tag:
                 if not safe:
                     raise BER_Decoding_Error(err_msg, remaining=s)
+                else:
+                    real_tag = ber_id
             s = chr(hidden_tag) + s
         elif explicit_tag is not None:
             ber_id,s = BER_id_dec(s)
             if ber_id != explicit_tag:
                 if not safe:
                     raise BER_Decoding_Error(err_msg, remaining=s)
+                else:
+                    real_tag = ber_id
             l,s = BER_len_dec(s)
-    return s
+    return real_tag, s
 def BER_tagging_enc(s, implicit_tag=None, explicit_tag=None):
     if len(s) > 0:
         if implicit_tag is not None:
-            s = chr(implicit_tag) + s[1:]
+            s = BER_id_enc(implicit_tag) + s[1:]
         elif explicit_tag is not None:
-            s = chr(explicit_tag) + BER_len_enc(len(s)) + s
+            s = BER_id_enc(explicit_tag) + BER_len_enc(len(s)) + s
     return s
 
 #####[ BER classes ]#####

scapy/asn1fields.py

@@ -15,8 +15,6 @@
 from base_classes import BasePacket
 from utils import binrepr
 
-FLEXIBLE_TAGS = False
-
 class ASN1F_badsequence(Exception):
     pass
 
@@ -35,7 +33,8 @@ class ASN1F_field(ASN1F_element):
     context = ASN1_Class_UNIVERSAL
 
     def __init__(self, name, default, context=None,
-                 implicit_tag=None, explicit_tag=None):
+                 implicit_tag=None, explicit_tag=None,
+                 flexible_tag=False):
         self.context = context
         self.name = name
         if default is None:
@@ -44,7 +43,7 @@ def __init__(self, name, default, context=None,
             self.default = default
         else:
             self.default = self.ASN1_tag.asn1_object(default)
-        self.flexible_tag = False or FLEXIBLE_TAGS
+        self.flexible_tag = flexible_tag
         if (implicit_tag is not None) and (explicit_tag is not None):
             err_msg = "field cannot be both implicitly and explicitly tagged"
             raise ASN1_Error(err_msg)
@@ -70,13 +69,18 @@ def m2i(self, pkt, s):
 
         Regarding other fields, we might need to know whether encoding went
         as expected or not. Noticeably, input methods from cert.py expect
-        certain exceptions to be raised. Hence default flexible_tag is False,
-        but we provide a FLEXIBLE_TAGS switch for debugging purposes.
+        certain exceptions to be raised. Hence default flexible_tag is False.
         """
-        s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag,
-                            implicit_tag=self.implicit_tag,
-                            explicit_tag=self.explicit_tag,
-                            safe=self.flexible_tag)
+        diff_tag, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag,
+                                      implicit_tag=self.implicit_tag,
+                                      explicit_tag=self.explicit_tag,
+                                      safe=self.flexible_tag)
+        if diff_tag is not None:
+            # this implies that flexible_tag was True
+            if self.implicit_tag is not None:
+                self.implicit_tag = diff_tag
+            elif self.explicit_tag is not None:
+                self.explicit_tag = diff_tag
         codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)
         if self.flexible_tag:
             return codec.safedec(s, context=self.context)
@@ -251,19 +255,32 @@ class ASN1F_BMP_STRING(ASN1F_STRING):
     ASN1_tag = ASN1_Class_UNIVERSAL.BMP_STRING
 
 class ASN1F_SEQUENCE(ASN1F_field):
+# Here is how you could decode a SEQUENCE
+# with an unknown, private high-tag prefix :
+# class PrivSeq(ASN1_Packet):
+#     ASN1_codec = ASN1_Codecs.BER
+#     ASN1_root = ASN1F_SEQUENCE(
+#                       <asn1 field #0>,
+#                       ...
+#                       <asn1 field #N>,
+#                       explicit_tag=0,
+#                       flexible_tag=True)
+# Because we use flexible_tag, the value of the explicit_tag does not matter.
     ASN1_tag = ASN1_Class_UNIVERSAL.SEQUENCE
     holds_packets = 1
     def __init__(self, *seq, **kwargs):
         name = "dummy_seq_name"
         default = [field.default for field in seq]
-        for kwarg in ["context", "implicit_tag", "explicit_tag"]:
+        for kwarg in ["context", "implicit_tag",
+                      "explicit_tag", "flexible_tag"]:
             if kwarg in kwargs:
                 setattr(self, kwarg, kwargs[kwarg])
             else:
                 setattr(self, kwarg, None)
         ASN1F_field.__init__(self, name, default, context=self.context,
                              implicit_tag=self.implicit_tag,
-                             explicit_tag=self.explicit_tag)
+                             explicit_tag=self.explicit_tag,
+                             flexible_tag=self.flexible_tag)
         self.seq = seq
         self.islist = len(seq) > 1
     def __repr__(self):
@@ -284,10 +301,15 @@ def m2i(self, pkt, s):
         Thus m2i returns an empty list (along with the proper remainder).
         It is discarded by dissect() and should not be missed elsewhere.
         """
-        s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag,
-                            implicit_tag=self.implicit_tag,
-                            explicit_tag=self.explicit_tag,
-                            safe=self.flexible_tag)
+        diff_tag, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag,
+                                      implicit_tag=self.implicit_tag,
+                                      explicit_tag=self.explicit_tag,
+                                      safe=self.flexible_tag)
+        if diff_tag is not None:
+            if self.implicit_tag is not None:
+                self.implicit_tag = diff_tag
+            elif self.explicit_tag is not None:
+                self.explicit_tag = diff_tag
         codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)
         i,s,remain = codec.check_type_check_len(s)
         if len(s) == 0:
@@ -325,10 +347,15 @@ def __init__(self, name, default, cls, context=None,
     def is_empty(self, pkt):
         return ASN1F_field.is_empty(self, pkt)
     def m2i(self, pkt, s):
-        s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag,
-                            implicit_tag=self.implicit_tag,
-                            explicit_tag=self.explicit_tag,
-                            safe=self.flexible_tag)
+        diff_tag, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag,
+                                      implicit_tag=self.implicit_tag,
+                                      explicit_tag=self.explicit_tag,
+                                      safe=self.flexible_tag)
+        if diff_tag is not None:
+            if self.implicit_tag is not None:
+                self.implicit_tag = diff_tag
+            elif self.explicit_tag is not None:
+                self.explicit_tag = diff_tag
         codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)
         i,s,remain = codec.check_type_check_len(s)
         lst = []
@@ -368,7 +395,7 @@ class ASN1F_TIME_TICKS(ASN1F_INTEGER):
 #############################
 
 class ASN1F_optional(ASN1F_element):
-    def __init__(self, field, by_default=False):
+    def __init__(self, field):
         field.flexible_tag = False
         self._field = field
     def __getattr__(self, attr):
@@ -440,9 +467,8 @@ def m2i(self, pkt, s):
         """
         if len(s) == 0:
             raise ASN1_Error("ASN1F_CHOICE: got empty string")
-        s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag,
-                            explicit_tag=self.explicit_tag,
-                            safe=self.flexible_tag)
+        _,s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag,
+                              explicit_tag=self.explicit_tag)
         tag,_ = BER_id_dec(s)
         if tag not in self.choices:
             if self.flexible_tag:
@@ -484,10 +510,15 @@ def __init__(self, name, default, cls, context=None,
                 self.network_tag = 16|0x20
         self.default = default
     def m2i(self, pkt, s):
-        s = BER_tagging_dec(s, hidden_tag=self.cls.ASN1_root.ASN1_tag,
-                            implicit_tag=self.implicit_tag,
-                            explicit_tag=self.explicit_tag,
-                            safe=self.flexible_tag)
+        diff_tag, s = BER_tagging_dec(s, hidden_tag=self.cls.ASN1_root.ASN1_tag,
+                                      implicit_tag=self.implicit_tag,
+                                      explicit_tag=self.explicit_tag,
+                                      safe=self.flexible_tag)
+        if diff_tag is not None:
+            if self.implicit_tag is not None:
+                self.implicit_tag = diff_tag
+            elif self.explicit_tag is not None:
+                self.explicit_tag = diff_tag
         p,s = self.extract_packet(self.cls, s)
         return p,s
     def i2m(self, pkt, x):

scapy/layers/x509.py

@@ -19,6 +19,16 @@ class ASN1P_INTEGER(ASN1_Packet):
     ASN1_codec = ASN1_Codecs.BER
     ASN1_root = ASN1F_INTEGER("number", 0)
 
+class ASN1P_PRIVSEQ(ASN1_Packet):
+    # This class gets used in x509.uts
+    # It showcases the private high-tag decoding capacities of scapy.
+    ASN1_codec = ASN1_Codecs.BER
+    ASN1_root = ASN1F_SEQUENCE(
+            ASN1F_IA5_STRING("str", ""),
+            ASN1F_STRING("int", 0),
+            explicit_tag=0,
+            flexible_tag=True)
+
 
 #######################
 ##### RSA packets #####
@@ -36,7 +46,7 @@ class RSAPublicKey(ASN1_Packet):
                     ASN1F_INTEGER("publicExponent", 3))
 
 class RSAOtherPrimeInfo(ASN1_Packet):
-    ASN1_codec = ASN1_Codecs.DER
+    ASN1_codec = ASN1_Codecs.BER
     ASN1_root = ASN1F_SEQUENCE(
                     ASN1F_INTEGER("prime", 0),
                     ASN1F_INTEGER("exponent", 0),
@@ -610,9 +620,9 @@ def __init__(self, **kargs):
                    explicit_tag=0x04)]
         ASN1F_SEQUENCE.__init__(self, *seq, **kargs)
     def dissect(self, pkt, s):
-        s = BER_tagging_dec(s, implicit_tag=self.implicit_tag,
-                            explicit_tag=self.explicit_tag,
-                            safe=self.flexible_tag)
+        _,s = BER_tagging_dec(s, implicit_tag=self.implicit_tag,
+                              explicit_tag=self.explicit_tag,
+                              safe=self.flexible_tag)
         codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)
         i,s,remain = codec.check_type_check_len(s)
         extnID = self.seq[0]

test/x509.uts

@@ -1,7 +1,15 @@
 % Tests for X.509 objects
 # 
-# Launch me with:
-# sudo bash test/run_tests -t test/x509.uts -F
+# Try me with:
+# bash test/run_tests -t test/x509.uts -F
+
+########### ASN.1 border case #######################################
+
++ General BER decoding tests
+= Decoding an ASN.1 SEQUENCE with an unknown, high-tag identifier
+s = '\xff\x84\x92\xb9\x86H\x1e0\x1c\x16\x04BNCH\x04\x14\xb7\xca\x01wO\x9b\xbaz\xbb\xb5\x92\x87>T\xb2\xc3g\xc1]\xfb'
+p = ASN1P_PRIVSEQ(s)
+
 
 ########### Key class ###############################################
 
@@ -136,7 +144,7 @@ str(x) == c
 tbs = x.tbsCertList
 tbs.version == None
 
-= CRL class : Signature algorithm (as advertised by TBSCRLificate)
+= CRL class : Signature algorithm (as advertised by TBSCertList)
 assert(type(tbs.signature) is X509_AlgorithmIdentifier)
 tbs.signature.algorithm == ASN1_OID("sha1-with-rsa-signature")
 
@@ -178,6 +186,7 @@ x.signatureAlgorithm.algorithm == ASN1_OID("sha1-with-rsa-signature")
 x.signatureValue == ASN1_BIT_STRING('"\xc9\xf6\xbb\x1d\xa1\xa5=$\xc7\xff\xb0"\x11\xb3p\x06[\xc5U\xdd3v\xa0\x98"\x08cDi\xcfOG%w\x99\x12\x84\xd2\x19\xae \x94\xca,T\x9ak\x81\xd2\x038\xa6Z\x95\x8d*\xe2a\xce\xdb\x19\xcdu\'Y&|V\xe1\xe4\x80q\x1aI\xb2\xaa\xcdI[\xda\x0f\xa8\xff\xce<\n\xfc\xc9\xad\xc6\xde\xc8@d\x0c&\t#\x90\xb7\x9c\xb9P\x03\x8fK\x18\x9f\xb0\xe0e\x0f`\x1c\x1ag\xe5\x85\xc4%\xf5\x0b\xc93\x82R\xe6', readable=True)
 
 = CRL class : Default X509_CRL from scratch
-str(X509_CRL(str(X509_CRL()))) == str(X509_CRL())
+s = str(X509_CRL())
+str(X509_CRL(s)) == s