665 lines
23 KiB
Python
Executable file
665 lines
23 KiB
Python
Executable file
# This file is part of Scapy
|
|
# See http://www.secdev.org/projects/scapy for more information
|
|
# Copyright (C) Philippe Biondi <phil@secdev.org>
|
|
# Enhanced by Maxence Tury <maxence.tury@ssi.gouv.fr>
|
|
# This program is published under a GPLv2 license
|
|
|
|
"""
|
|
Classes that implement ASN.1 data structures.
|
|
"""
|
|
|
|
from __future__ import absolute_import
|
|
from scapy.asn1.asn1 import ASN1_Class_UNIVERSAL, ASN1_NULL, ASN1_Error, \
|
|
ASN1_Object, ASN1_INTEGER
|
|
from scapy.asn1.ber import BER_tagging_dec, BER_Decoding_Error, BER_id_dec, \
|
|
BER_tagging_enc
|
|
from scapy.volatile import RandInt, RandChoice, RandNum, RandString, RandOID, \
|
|
GeneralizedTime
|
|
from scapy.compat import orb, raw
|
|
from scapy.base_classes import BasePacket
|
|
from scapy.utils import binrepr
|
|
from scapy import packet
|
|
from functools import reduce
|
|
import scapy.modules.six as six
|
|
from scapy.modules.six.moves import range
|
|
|
|
|
|
class ASN1F_badsequence(Exception):
|
|
pass
|
|
|
|
|
|
class ASN1F_element(object):
|
|
pass
|
|
|
|
|
|
##########################
|
|
# Basic ASN1 Field #
|
|
##########################
|
|
|
|
class ASN1F_field(ASN1F_element):
|
|
holds_packets = 0
|
|
islist = 0
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.ANY
|
|
context = ASN1_Class_UNIVERSAL
|
|
|
|
def __init__(self, name, default, context=None,
|
|
implicit_tag=None, explicit_tag=None,
|
|
flexible_tag=False):
|
|
self.context = context
|
|
self.name = name
|
|
if default is None:
|
|
self.default = None
|
|
elif isinstance(default, ASN1_NULL):
|
|
self.default = default
|
|
else:
|
|
self.default = self.ASN1_tag.asn1_object(default)
|
|
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)
|
|
self.implicit_tag = implicit_tag
|
|
self.explicit_tag = explicit_tag
|
|
# network_tag gets useful for ASN1F_CHOICE
|
|
self.network_tag = implicit_tag or explicit_tag or self.ASN1_tag
|
|
|
|
def i2repr(self, pkt, x):
|
|
return repr(x)
|
|
|
|
def i2h(self, pkt, x):
|
|
return x
|
|
|
|
def any2i(self, pkt, x):
|
|
return x
|
|
|
|
def m2i(self, pkt, s):
|
|
"""
|
|
The good thing about safedec is that it may still decode ASN1
|
|
even if there is a mismatch between the expected tag (self.ASN1_tag)
|
|
and the actual tag; the decoded ASN1 object will simply be put
|
|
into an ASN1_BADTAG object. However, safedec prevents the raising of
|
|
exceptions needed for ASN1F_optional processing.
|
|
Thus we use 'flexible_tag', which should be False with ASN1F_optional.
|
|
|
|
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.
|
|
"""
|
|
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)
|
|
else:
|
|
return codec.dec(s, context=self.context)
|
|
|
|
def i2m(self, pkt, x):
|
|
if x is None:
|
|
return b""
|
|
if isinstance(x, ASN1_Object):
|
|
if (self.ASN1_tag == ASN1_Class_UNIVERSAL.ANY or
|
|
x.tag == ASN1_Class_UNIVERSAL.RAW or
|
|
x.tag == ASN1_Class_UNIVERSAL.ERROR or
|
|
self.ASN1_tag == x.tag):
|
|
s = x.enc(pkt.ASN1_codec)
|
|
else:
|
|
raise ASN1_Error("Encoding Error: got %r instead of an %r for field [%s]" % (x, self.ASN1_tag, self.name)) # noqa: E501
|
|
else:
|
|
s = self.ASN1_tag.get_codec(pkt.ASN1_codec).enc(x)
|
|
return BER_tagging_enc(s, implicit_tag=self.implicit_tag,
|
|
explicit_tag=self.explicit_tag)
|
|
|
|
def extract_packet(self, cls, s):
|
|
if len(s) > 0:
|
|
try:
|
|
c = cls(s)
|
|
except ASN1F_badsequence:
|
|
c = packet.Raw(s)
|
|
cpad = c.getlayer(packet.Raw)
|
|
s = b""
|
|
if cpad is not None:
|
|
s = cpad.load
|
|
del(cpad.underlayer.payload)
|
|
return c, s
|
|
else:
|
|
return None, s
|
|
|
|
def build(self, pkt):
|
|
return self.i2m(pkt, getattr(pkt, self.name))
|
|
|
|
def dissect(self, pkt, s):
|
|
v, s = self.m2i(pkt, s)
|
|
self.set_val(pkt, v)
|
|
return s
|
|
|
|
def do_copy(self, x):
|
|
if hasattr(x, "copy"):
|
|
return x.copy()
|
|
if isinstance(x, list):
|
|
x = x[:]
|
|
for i in range(len(x)):
|
|
if isinstance(x[i], BasePacket):
|
|
x[i] = x[i].copy()
|
|
return x
|
|
|
|
def set_val(self, pkt, val):
|
|
setattr(pkt, self.name, val)
|
|
|
|
def is_empty(self, pkt):
|
|
return getattr(pkt, self.name) is None
|
|
|
|
def get_fields_list(self):
|
|
return [self]
|
|
|
|
def __str__(self):
|
|
return repr(self)
|
|
|
|
def randval(self):
|
|
return RandInt()
|
|
|
|
|
|
############################
|
|
# Simple ASN1 Fields #
|
|
############################
|
|
|
|
class ASN1F_BOOLEAN(ASN1F_field):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.BOOLEAN
|
|
|
|
def randval(self):
|
|
return RandChoice(True, False)
|
|
|
|
|
|
class ASN1F_INTEGER(ASN1F_field):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.INTEGER
|
|
|
|
def randval(self):
|
|
return RandNum(-2**64, 2**64 - 1)
|
|
|
|
|
|
class ASN1F_enum_INTEGER(ASN1F_INTEGER):
|
|
def __init__(self, name, default, enum, context=None,
|
|
implicit_tag=None, explicit_tag=None):
|
|
ASN1F_INTEGER.__init__(self, name, default, context=context,
|
|
implicit_tag=implicit_tag,
|
|
explicit_tag=explicit_tag)
|
|
i2s = self.i2s = {}
|
|
s2i = self.s2i = {}
|
|
if isinstance(enum, list):
|
|
keys = range(len(enum))
|
|
else:
|
|
keys = list(enum)
|
|
if any(isinstance(x, six.string_types) for x in keys):
|
|
i2s, s2i = s2i, i2s
|
|
for k in keys:
|
|
i2s[k] = enum[k]
|
|
s2i[enum[k]] = k
|
|
|
|
def i2m(self, pkt, s):
|
|
if isinstance(s, str):
|
|
s = self.s2i.get(s)
|
|
return super(ASN1F_enum_INTEGER, self).i2m(pkt, s)
|
|
|
|
def i2repr(self, pkt, x):
|
|
if x is not None and isinstance(x, ASN1_INTEGER):
|
|
r = self.i2s.get(x.val)
|
|
if r:
|
|
return "'%s' %s" % (r, repr(x))
|
|
return repr(x)
|
|
|
|
|
|
class ASN1F_BIT_STRING(ASN1F_field):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.BIT_STRING
|
|
|
|
def __init__(self, name, default, default_readable=True, context=None,
|
|
implicit_tag=None, explicit_tag=None):
|
|
if default is not None and default_readable:
|
|
default = b"".join(binrepr(orb(x)).zfill(8).encode("utf8") for x in default) # noqa: E501
|
|
ASN1F_field.__init__(self, name, default, context=context,
|
|
implicit_tag=implicit_tag,
|
|
explicit_tag=explicit_tag)
|
|
|
|
def randval(self):
|
|
return RandString(RandNum(0, 1000))
|
|
|
|
|
|
class ASN1F_STRING(ASN1F_field):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.STRING
|
|
|
|
def randval(self):
|
|
return RandString(RandNum(0, 1000))
|
|
|
|
|
|
class ASN1F_NULL(ASN1F_INTEGER):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.NULL
|
|
|
|
|
|
class ASN1F_OID(ASN1F_field):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.OID
|
|
|
|
def randval(self):
|
|
return RandOID()
|
|
|
|
|
|
class ASN1F_ENUMERATED(ASN1F_enum_INTEGER):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.ENUMERATED
|
|
|
|
|
|
class ASN1F_UTF8_STRING(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.UTF8_STRING
|
|
|
|
|
|
class ASN1F_NUMERIC_STRING(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.NUMERIC_STRING
|
|
|
|
|
|
class ASN1F_PRINTABLE_STRING(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.PRINTABLE_STRING
|
|
|
|
|
|
class ASN1F_T61_STRING(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.T61_STRING
|
|
|
|
|
|
class ASN1F_VIDEOTEX_STRING(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.VIDEOTEX_STRING
|
|
|
|
|
|
class ASN1F_IA5_STRING(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.IA5_STRING
|
|
|
|
|
|
class ASN1F_UTC_TIME(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.UTC_TIME
|
|
|
|
def randval(self):
|
|
return GeneralizedTime()
|
|
|
|
|
|
class ASN1F_GENERALIZED_TIME(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.GENERALIZED_TIME
|
|
|
|
def randval(self):
|
|
return GeneralizedTime()
|
|
|
|
|
|
class ASN1F_ISO646_STRING(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.ISO646_STRING
|
|
|
|
|
|
class ASN1F_UNIVERSAL_STRING(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.UNIVERSAL_STRING
|
|
|
|
|
|
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. # noqa: E501
|
|
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", "flexible_tag"]:
|
|
setattr(self, kwarg, kwargs.get(kwarg))
|
|
ASN1F_field.__init__(self, name, default, context=self.context,
|
|
implicit_tag=self.implicit_tag,
|
|
explicit_tag=self.explicit_tag,
|
|
flexible_tag=self.flexible_tag)
|
|
self.seq = seq
|
|
self.islist = len(seq) > 1
|
|
|
|
def __repr__(self):
|
|
return "<%s%r>" % (self.__class__.__name__, self.seq)
|
|
|
|
def is_empty(self, pkt):
|
|
return all(f.is_empty(pkt) for f in self.seq)
|
|
|
|
def get_fields_list(self):
|
|
return reduce(lambda x, y: x + y.get_fields_list(), self.seq, [])
|
|
|
|
def m2i(self, pkt, s):
|
|
"""
|
|
ASN1F_SEQUENCE behaves transparently, with nested ASN1_objects being
|
|
dissected one by one. Because we use obj.dissect (see loop below)
|
|
instead of obj.m2i (as we trust dissect to do the appropriate set_vals)
|
|
we do not directly retrieve the list of nested objects.
|
|
Thus m2i returns an empty list (along with the proper remainder).
|
|
It is discarded by dissect() and should not be missed elsewhere.
|
|
"""
|
|
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:
|
|
for obj in self.seq:
|
|
obj.set_val(pkt, None)
|
|
else:
|
|
for obj in self.seq:
|
|
try:
|
|
s = obj.dissect(pkt, s)
|
|
except ASN1F_badsequence:
|
|
break
|
|
if len(s) > 0:
|
|
raise BER_Decoding_Error("unexpected remainder", remaining=s)
|
|
return [], remain
|
|
|
|
def dissect(self, pkt, s):
|
|
_, x = self.m2i(pkt, s)
|
|
return x
|
|
|
|
def build(self, pkt):
|
|
s = reduce(lambda x, y: x + y.build(pkt), self.seq, b"")
|
|
return self.i2m(pkt, s)
|
|
|
|
|
|
class ASN1F_SET(ASN1F_SEQUENCE):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.SET
|
|
|
|
|
|
class ASN1F_SEQUENCE_OF(ASN1F_field):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.SEQUENCE
|
|
holds_packets = 1
|
|
islist = 1
|
|
|
|
def __init__(self, name, default, cls, context=None,
|
|
implicit_tag=None, explicit_tag=None):
|
|
self.cls = cls
|
|
ASN1F_field.__init__(self, name, None, context=context,
|
|
implicit_tag=implicit_tag, explicit_tag=explicit_tag) # noqa: E501
|
|
self.default = default
|
|
|
|
def is_empty(self, pkt):
|
|
return ASN1F_field.is_empty(self, pkt)
|
|
|
|
def m2i(self, pkt, s):
|
|
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 = []
|
|
while s:
|
|
c, s = self.extract_packet(self.cls, s)
|
|
lst.append(c)
|
|
if len(s) > 0:
|
|
raise BER_Decoding_Error("unexpected remainder", remaining=s)
|
|
return lst, remain
|
|
|
|
def build(self, pkt):
|
|
val = getattr(pkt, self.name)
|
|
if isinstance(val, ASN1_Object) and val.tag == ASN1_Class_UNIVERSAL.RAW: # noqa: E501
|
|
s = val
|
|
elif val is None:
|
|
s = b""
|
|
else:
|
|
s = b"".join(raw(i) for i in val)
|
|
return self.i2m(pkt, s)
|
|
|
|
def randval(self):
|
|
return packet.fuzz(self.cls())
|
|
|
|
def __repr__(self):
|
|
return "<%s %s>" % (self.__class__.__name__, self.name)
|
|
|
|
|
|
class ASN1F_SET_OF(ASN1F_SEQUENCE_OF):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.SET
|
|
|
|
|
|
class ASN1F_IPADDRESS(ASN1F_STRING):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.IPADDRESS
|
|
|
|
|
|
class ASN1F_TIME_TICKS(ASN1F_INTEGER):
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.TIME_TICKS
|
|
|
|
|
|
#############################
|
|
# Complex ASN1 Fields #
|
|
#############################
|
|
|
|
class ASN1F_optional(ASN1F_element):
|
|
def __init__(self, field):
|
|
field.flexible_tag = False
|
|
self._field = field
|
|
|
|
def __getattr__(self, attr):
|
|
return getattr(self._field, attr)
|
|
|
|
def m2i(self, pkt, s):
|
|
try:
|
|
return self._field.m2i(pkt, s)
|
|
except (ASN1_Error, ASN1F_badsequence, BER_Decoding_Error):
|
|
# ASN1_Error may be raised by ASN1F_CHOICE
|
|
return None, s
|
|
|
|
def dissect(self, pkt, s):
|
|
try:
|
|
return self._field.dissect(pkt, s)
|
|
except (ASN1_Error, ASN1F_badsequence, BER_Decoding_Error):
|
|
self._field.set_val(pkt, None)
|
|
return s
|
|
|
|
def build(self, pkt):
|
|
if self._field.is_empty(pkt):
|
|
return b""
|
|
return self._field.build(pkt)
|
|
|
|
def any2i(self, pkt, x):
|
|
return self._field.any2i(pkt, x)
|
|
|
|
def i2repr(self, pkt, x):
|
|
return self._field.i2repr(pkt, x)
|
|
|
|
|
|
class ASN1F_CHOICE(ASN1F_field):
|
|
"""
|
|
Multiple types are allowed: ASN1_Packet, ASN1F_field and ASN1F_PACKET(),
|
|
See layers/x509.py for examples.
|
|
Other ASN1F_field instances than ASN1F_PACKET instances must not be used.
|
|
"""
|
|
holds_packets = 1
|
|
ASN1_tag = ASN1_Class_UNIVERSAL.ANY
|
|
|
|
def __init__(self, name, default, *args, **kwargs):
|
|
if "implicit_tag" in kwargs:
|
|
err_msg = "ASN1F_CHOICE has been called with an implicit_tag"
|
|
raise ASN1_Error(err_msg)
|
|
self.implicit_tag = None
|
|
for kwarg in ["context", "explicit_tag"]:
|
|
setattr(self, kwarg, kwargs.get(kwarg))
|
|
ASN1F_field.__init__(self, name, None, context=self.context,
|
|
explicit_tag=self.explicit_tag)
|
|
self.default = default
|
|
self.current_choice = None
|
|
self.choices = {}
|
|
self.pktchoices = {}
|
|
for p in args:
|
|
if hasattr(p, "ASN1_root"): # should be ASN1_Packet
|
|
if hasattr(p.ASN1_root, "choices"):
|
|
for k, v in six.iteritems(p.ASN1_root.choices):
|
|
self.choices[k] = v # ASN1F_CHOICE recursion
|
|
else:
|
|
self.choices[p.ASN1_root.network_tag] = p
|
|
elif hasattr(p, "ASN1_tag"):
|
|
if isinstance(p, type): # should be ASN1F_field class
|
|
self.choices[p.ASN1_tag] = p
|
|
else: # should be ASN1F_PACKET instance
|
|
self.choices[p.network_tag] = p
|
|
self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501
|
|
else:
|
|
raise ASN1_Error("ASN1F_CHOICE: no tag found for one field")
|
|
|
|
def m2i(self, pkt, s):
|
|
"""
|
|
First we have to retrieve the appropriate choice.
|
|
Then we extract the field/packet, according to this choice.
|
|
"""
|
|
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)
|
|
tag, _ = BER_id_dec(s)
|
|
if tag not in self.choices:
|
|
if self.flexible_tag:
|
|
choice = ASN1F_field
|
|
else:
|
|
raise ASN1_Error("ASN1F_CHOICE: unexpected field")
|
|
else:
|
|
choice = self.choices[tag]
|
|
if hasattr(choice, "ASN1_root"):
|
|
# we don't want to import ASN1_Packet in this module...
|
|
return self.extract_packet(choice, s)
|
|
elif isinstance(choice, type):
|
|
# XXX find a way not to instantiate the ASN1F_field
|
|
return choice(self.name, b"").m2i(pkt, s)
|
|
else:
|
|
# XXX check properly if this is an ASN1F_PACKET
|
|
return choice.m2i(pkt, s)
|
|
|
|
def i2m(self, pkt, x):
|
|
if x is None:
|
|
s = b""
|
|
else:
|
|
s = raw(x)
|
|
if hash(type(x)) in self.pktchoices:
|
|
imp, exp = self.pktchoices[hash(type(x))]
|
|
s = BER_tagging_enc(s, implicit_tag=imp,
|
|
explicit_tag=exp)
|
|
return BER_tagging_enc(s, explicit_tag=self.explicit_tag)
|
|
|
|
def randval(self):
|
|
randchoices = []
|
|
for p in six.itervalues(self.choices):
|
|
if hasattr(p, "ASN1_root"): # should be ASN1_Packet class
|
|
randchoices.append(packet.fuzz(p()))
|
|
elif hasattr(p, "ASN1_tag"):
|
|
if isinstance(p, type): # should be (basic) ASN1F_field class # noqa: E501
|
|
randchoices.append(p("dummy", None).randval())
|
|
else: # should be ASN1F_PACKET instance
|
|
randchoices.append(p.randval())
|
|
return RandChoice(*randchoices)
|
|
|
|
|
|
class ASN1F_PACKET(ASN1F_field):
|
|
holds_packets = 1
|
|
|
|
def __init__(self, name, default, cls, context=None,
|
|
implicit_tag=None, explicit_tag=None):
|
|
self.cls = cls
|
|
ASN1F_field.__init__(self, name, None, context=context,
|
|
implicit_tag=implicit_tag, explicit_tag=explicit_tag) # noqa: E501
|
|
if cls.ASN1_root.ASN1_tag == ASN1_Class_UNIVERSAL.SEQUENCE:
|
|
if implicit_tag is None and explicit_tag is None:
|
|
self.network_tag = 16 | 0x20
|
|
self.default = default
|
|
|
|
def m2i(self, pkt, s):
|
|
diff_tag, s = BER_tagging_dec(s, hidden_tag=self.cls.ASN1_root.ASN1_tag, # noqa: E501
|
|
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):
|
|
if x is None:
|
|
s = b""
|
|
else:
|
|
s = raw(x)
|
|
return BER_tagging_enc(s, implicit_tag=self.implicit_tag,
|
|
explicit_tag=self.explicit_tag)
|
|
|
|
def randval(self):
|
|
return packet.fuzz(self.cls())
|
|
|
|
|
|
class ASN1F_BIT_STRING_ENCAPS(ASN1F_BIT_STRING):
|
|
"""
|
|
We may emulate simple string encapsulation with explicit_tag=0x04,
|
|
but we need a specific class for bit strings because of unused bits, etc.
|
|
"""
|
|
holds_packets = 1
|
|
|
|
def __init__(self, name, default, cls, context=None,
|
|
implicit_tag=None, explicit_tag=None):
|
|
self.cls = cls
|
|
ASN1F_BIT_STRING.__init__(self, name, None, context=context,
|
|
implicit_tag=implicit_tag,
|
|
explicit_tag=explicit_tag)
|
|
self.default = default
|
|
|
|
def m2i(self, pkt, s):
|
|
bit_string, remain = ASN1F_BIT_STRING.m2i(self, pkt, s)
|
|
if len(bit_string.val) % 8 != 0:
|
|
raise BER_Decoding_Error("wrong bit string", remaining=s)
|
|
p, s = self.extract_packet(self.cls, bit_string.val_readable)
|
|
if len(s) > 0:
|
|
raise BER_Decoding_Error("unexpected remainder", remaining=s)
|
|
return p, remain
|
|
|
|
def i2m(self, pkt, x):
|
|
s = b"" if x is None else raw(x)
|
|
s = b"".join(binrepr(orb(x)).zfill(8).encode("utf8") for x in s)
|
|
return ASN1F_BIT_STRING.i2m(self, pkt, s)
|
|
|
|
|
|
class ASN1F_FLAGS(ASN1F_BIT_STRING):
|
|
def __init__(self, name, default, mapping, context=None,
|
|
implicit_tag=None, explicit_tag=None):
|
|
self.mapping = mapping
|
|
ASN1F_BIT_STRING.__init__(self, name, default,
|
|
default_readable=False,
|
|
context=context,
|
|
implicit_tag=implicit_tag,
|
|
explicit_tag=explicit_tag)
|
|
|
|
def get_flags(self, pkt):
|
|
fbytes = getattr(pkt, self.name).val
|
|
return [self.mapping[i] for i, positional in enumerate(fbytes)
|
|
if positional == '1' and i < len(self.mapping)]
|
|
|
|
def i2repr(self, pkt, x):
|
|
if x is not None:
|
|
pretty_s = ", ".join(self.get_flags(pkt))
|
|
return pretty_s + " " + repr(x)
|
|
return repr(x)
|