cyberflex-shell/TLV_utils.py

526 lines
17 KiB
Python

import binascii, utils, re, sys
class identifier:
"""An identifier, because I'm too lazy to use quotes all over the place.
Instantiating an object of this type registers a name in the current scope, essentially
making each instantiation of this class equal to
foo = identifier("foo")
even when you only write
identifier("foo")
"""
def __init__(self,name):
self.name = name
sys._getframe(1).f_locals[name] = self
def __str__(self):
return self.name
def __repr__(self):
return "identifier(%r)" % self.name
identifier("context_FCP")
identifier("context_FMD")
identifier("context_FCI")
identifier("recurse")
identifier("binary")
identifier("number")
identifier("ascii")
identifier("utf8")
file_descriptor_byte_descriptions = [
#mask byte no match match
(0x80, 0x80, None, "RFU"),
(0xC0, 0x40, "non shareable", "shareable"),
(0xB8, 0x00, None, "working EF"),
(0xB8, 0x08, None, "internal EF"),
(0xB8, 0x10, None, "Reserved for proprietary uses"),
(0xB8, 0x18, None, "Reserved for proprietary uses"),
(0xB8, 0x20, None, "Reserved for proprietary uses"),
(0xB8, 0x28, None, "Reserved for proprietary uses"),
(0xB8, 0x30, None, "Reserved for proprietary uses"),
(0xB8, 0x38, None, "DF"),
(0x87, 0x00, None, "No file structure information given"),
(0x87, 0x01, None, "Transparent"),
(0x87, 0x02, None, "Linear fixed, no further info"),
(0x87, 0x03, None, "Linear fixed, SIMPLE-TLV"),
(0x87, 0x04, None, "Linear variable, no further info"),
(0x87, 0x05, None, "Linear variable, SIMPLE-TLV"),
(0x87, 0x06, None, "Cyclic, no further info"),
(0x87, 0x07, None, "Cyclic, SIMPLE-TLV"),
]
data_coding_byte_descriptions = [
(0x60, 0x00, None, "one-time write"),
(0x60, 0x20, None, "proprietary"),
(0x60, 0x40, None, "write OR"),
(0x60, 0x60, None, "write AND"),
]
life_cycle_status_byte_descriptions = [
(0xF0, 0x00, "Proprietary", None),
(0xFF, 0x00, None, "No information given"),
(0xFF, 0x01, None, "Creation state"),
(0xFF, 0x03, None, "Initialisation state"),
(0xFD, 0x05, None, "Operational state (activated)"),
(0xFD, 0x04, None, "Operational state (deactivated)"),
(0xFC, 0x0C, None, "Termination state"),
]
def decode_file_descriptor_byte(value, verbose = True):
result = " %s" % utils.hexdump(value, short=True)
if not verbose:
attributes = utils.parse_binary(ord(value[0]), file_descriptor_byte_descriptions, False)
if len(value) > 1:
attributes.append(
"data coding byte, behavior of write functions: %s, data unit size in in nibbles: %i" % (
"".join( utils.parse_binary(ord(value[1]), data_coding_byte_descriptions) ),
2 ** (ord(value[1])&0x07)
)
)
if len(value) > 2:
i = 0
for j in value[2:4]:
i = i * 256 + ord(j)
attributes.append(
"maximum record length: %s" % i
)
if len(value) > 4:
i = 0
for j in value[4:6]:
i = i * 256 + ord(j)
attributes.append(
"number of records: %s" % i
)
return result + " (%s)" % "; ".join(attributes)
else:
result = result + "\nFile descriptor byte:\n"
result = result + "\t" + "\n\t".join(
utils.parse_binary(ord(value[0]), file_descriptor_byte_descriptions, True)
)
if len(value) > 1:
result = result + "\nData coding byte (0x%02X):\n" % ord(value[1])
result = result + "\tBehavior of write functions: %s\n\tData unit size in in nibbles: %i" % (
"".join( utils.parse_binary(ord(value[1]), data_coding_byte_descriptions) ),
2 ** (ord(value[1])&0x07)
)
if len(value) > 2:
i = 0
for j in value[2:4]:
i = i * 256 + ord(j)
result = result + "\nMaximum record length: %s" % i
if len(value) > 4:
i = 0
for j in value[4:6]:
i = i * 256 + ord(j)
result = result + "\nNumber of records: %s" % i
return result
def parse_oid(value):
result = []
def next_arc(data):
bits = ord(data[0]) & 0x7F
while ord(data[0]) & 0x80 != 0:
data = data[1:]
bits = (bits << 7) + (ord(data[0]) & 0x7F)
data = data[1:]
return bits, data
arc, value = next_arc(value)
if arc < 40:
result.append( 0 )
result.append( arc )
elif arc < 80:
result.append( 1 )
result.append( arc-40 )
else:
result.append( 2 )
result.append( arc-80 )
while len(value) > 0:
arc,value = next_arc(value)
result.append( arc )
return tuple(result)
oidCache = {}
def loadOids(filename="oids.txt"):
try:
fp = file(filename, "r")
except (SystemExit,KeyboardInterrupt):
raise
except:
pass
else:
try:
lines = fp.readlines()
finally:
fp.close()
for line in lines:
if line.strip() == "" or line[0] == "#":
continue
parts = line.strip().split(None,2)
if len(parts) < 3:
parts.append(parts[1])
oidCache[parts[0]] = tuple(parts[1:])
def decode_oid(value):
oid = parse_oid(value)
str_rep = ".".join([str(a) for a in oid])
if len(oidCache) == 0:
loadOids()
description = oidCache.get(str_rep, None)
if description is None:
steps = [oid[:e] for e in range(len(oid)-1, 0, -1)]
for step in steps:
new_str_rep = ".".join([str(a) for a in step])
if oidCache.has_key(new_str_rep):
description = ("%s %s" % (oidCache[new_str_rep][0], ".".join([str(a) for a in oid[len(step):]])),)
break
if description is None:
description = ("No description available",)
return " %s (%s)" % (str_rep, description[0])
_gtimere = re.compile(r'(\d{4})(\d\d)(\d\d)(\d\d)(?:(\d\d)(\d\d(?:[.,]\d+)?)?)?(|Z|(?:[+-]\d\d(?:\d\d)?))$')
def decode_generalized_time(value):
matches = _gtimere.match(value)
if not matches:
return " "+value
else:
matches = matches.groups()
result = [" %s-%s-%s %s:" % matches[:4]]
if matches[4] is not None:
result.append("%s:" % matches[4])
if matches[5] is not None:
result.append("%s" % matches[5])
else:
result.append("00")
else:
result.append(":00:00")
if matches[6] == "Z":
result.append(" UTC")
elif matches[6] != "":
result.append(" ")
result.append(matches[6])
if len(matches[6]) < 5:
result.append("00")
return "".join(result)
_utimere = re.compile(r'(\d\d)(\d\d)(\d\d)(\d\d)(?:(\d\d))?(Z|(?:[+-]\d\d(?:\d\d)?))$')
def decode_utc_time(value):
matches = _utimere.match(value)
if not matches:
return " "+value
else:
matches = matches.groups()
result = [" %s-%s-%s %s:" % matches[:4]]
if matches[4] is not None:
result.append("%s:" % matches[4])
if matches[5] is not None:
result.append("%s" % matches[5])
else:
result.append("00")
else:
result.append(":00:00")
if matches[6] == "Z":
result.append(" UTC")
elif matches[6] != "":
result.append(" ")
result.append(matches[6])
if len(matches[6]) < 5:
result.append("00")
return "".join(result)
def decode_bit_string(value):
unused_len = ord(value[0])
value = value[1:]
bits = []
for i in range(len(value)):
v = ord(value[i])
l = 8
if i == len(value)-1:
l = l - unused_len
for j in range(l):
bits.append( (v & 0x80) >> 7 )
v = v << 1
def do_some_bits(slice):
result = []
for index, bit in enumerate(slice):
if index % 4 == 0:
result.append(" ")
if index % 8 == 0:
result.append(" ")
result.append(str(bit))
return result
if len(bits) <= 16:
return " '%s'B" % "".join(do_some_bits(bits)).strip()
else:
step = 32
result = []
head, tail = bits[:step], bits[step:]
offset = 0
while offset == 0 or len(tail) > 0:
result.append("%05x: %s" % (offset, "".join(do_some_bits(head)).strip()))
offset += step
head, tail = tail[:step], tail[step:]
return "\n" + "\n".join(result)
def decode_lcs(value):
value = ord(value[0])
return " 0x%02x\n%s" % (value, "\n".join(
utils.parse_binary(value, life_cycle_status_byte_descriptions, True)
)
)
def decode_sfi(value):
if len(value) == 0: return ""
return " 0x%02x" % (ord(value[0]) >> 3)
tags = {
None: {
0x01: (lambda a: (len(a) > 0 and ord(a[0]) != 0) and " True" or " False", "Boolean"),
0x02: (number, "Integer"),
0x03: (decode_bit_string, "Bit string"),
0x04: (binary, "Octet string"),
0x05: (lambda a: " Null", "Null"),
0x06: (decode_oid, "Object identifier"),
0x0A: (number, "Enumerated"),
0x0C: (utf8, "UTF-8 string"),
0x12: (ascii, "Numeric string"),
0x13: (ascii, "Printable string"),
0x14: (ascii, "Teletex string"), ## FIXME: support escape sequences?
0x15: (ascii, "Videotext string"), ## dito
0x16: (ascii, "IA5String"),
0x17: (decode_utc_time, "UTC time"),
0x18: (decode_generalized_time, "Generalized time"),
0x30: (recurse, "Sequence", None),
0x31: (recurse, "Set", None),
0x62: (recurse, "File Control Parameters", context_FCP),
0x64: (recurse, "File Management Data", context_FMD),
0x6F: (recurse, "File Control Information", context_FCI),
},
context_FCI: {
0x80: (number, "Number of data bytes in the file, excluding structural information"),
0x81: (number, "Number of data bytes in the file, including structural information"),
0x82: (decode_file_descriptor_byte, "File descriptor byte"),
0x83: (binary, "File identifier"),
0x84: (binary, "DF name"),
0x85: (binary, "Proprietary information"),
0x86: (binary, "Security attributes"),
0x87: (binary, "Identifier of an EF containing an extension of the FCI"),
0x88: (decode_sfi, "Short EF identifier"),
0x8A: (decode_lcs, "Life cycle status byte"),
0xA5: (recurse, "Proprietary information", None),
},
}
tags[context_FCP] = tags[context_FCI]
BER_CLASSES = {
0x0: "universal",
0x1: "application",
0x2: "context-specific",
0x3: "private",
}
def tlv_unpack(data):
ber_class = (ord(data[0]) & 0xC0) >> 6
constructed = (ord(data[0]) & 0x20) != 0 ## 0 = primitive, 0x20 = constructed
tag = ord(data[0])
data = data[1:]
if (tag & 0x1F) == 0x1F:
tag = (tag << 8) | ord(data[0])
while ord(data[0]) & 0x80 == 0x80:
data = data[1:]
tag = (tag << 8) | ord(data[0])
data = data[1:]
length = ord(data[0])
if length < 0x80:
data = data[1:]
elif length & 0x80 == 0x80:
length_ = 0
data = data[1:]
for i in range(0,length & 0x7F):
length_ = length_ * 256 + ord(data[0])
data = data[1:]
length = length_
value = data[:length]
rest = data[length:]
return ber_class, constructed, tag, length, value, rest
def decode(data, context = None, level = 0, tags=tags):
result = []
while len(data) > 0:
if ord(data[0]) in (0x00, 0xFF):
data = data[1:]
continue
ber_class, constructed, tag, length, value, data = tlv_unpack(data)
interpretation = tags.get(context, tags.get(None, {})).get(tag, None)
if interpretation is None:
if not constructed: interpretation = [binary, "Unknown field"]
else: interpretation = [recurse, "Unknown structure", ber_class in (0, 1) and context or None]
interpretation[1] = "%s (%s class)" % (interpretation[1], BER_CLASSES[ber_class])
interpretation = tuple(interpretation)
current = ["\t"*level]
current.append("Tag 0x%02X, Len 0x%02X, '%s':" % (tag, length, interpretation[1]))
if interpretation[0] is recurse:
current.append("\n")
current.append( decode(value, interpretation[2], level+1, tags=tags) )
elif interpretation[0] is number:
num = 0
for i in value:
num = num * 256
num = num + ord(i)
current.append( " 0x%02x (%i)" % (num, num))
elif interpretation[0] is ascii:
current.append( " %s" % value)
elif interpretation[0] is utf8:
current.append( " %s" % unicode(value, "utf-8"))
elif interpretation[0] is binary:
if len(value) < 0x10:
current.append( " %s" % utils.hexdump(value, short=True))
else:
current.append( "\n" + "\t"*(level+1) )
current.append( ("\n" + "\t"*(level+1)).join( utils.hexdump(value).splitlines() ) )
elif callable(interpretation[0]):
current.append( ("\n"+"\t"*(level+1)).join(interpretation[0](value).splitlines()) )
result.append( "".join(current) )
return "\n".join(result)
def tlv_find_tag(tlv_data, tag, num_results = None):
"""Find (and return) all instances of tag in the given tlv structure (as returned by unpack).
If num_results is specified then at most that many results will be returned."""
results = []
def find_recursive(tlv_data):
for d in tlv_data:
t,l,v = d[:3]
if t == tag:
results.append(d)
else:
if isinstance(v, list): # FIXME Refactor the whole TLV code into a class
find_recursive(v)
if num_results is not None and len(results) >= num_results:
return
find_recursive(tlv_data)
return results
def unpack(data, with_marks = None, offset = 0, include_filler=False):
result = []
while len(data) > 0:
if ord(data[0]) in (0x00, 0xFF):
if include_filler:
if with_marks is None:
result.append( (ord(data[0]), None, None) )
else:
result.append( (ord(data[0]), None, None, () ) )
data = data[1:]
offset = offset + 1
continue
l = len(data)
ber_class, constructed, tag, length, value, data = tlv_unpack(data)
stop = offset + (l - len(data))
start = stop - length
if with_marks is not None:
marks = []
for type, mark_start, mark_stop in with_marks:
if (mark_start, mark_stop) == (start, stop):
marks.append(type)
marks = (marks, )
else:
marks = ()
if not constructed:
result.append( (tag, length, value) + marks )
else:
result.append( (tag, length, unpack(value, with_marks, offset = start)) + marks )
offset = stop
return result
def pack(tlv_data, recalculate_length = False):
result = []
for data in tlv_data:
tag, length, value = data[:3]
if tag in (0xff, 0x00):
result.append( chr(tag) )
continue
if not isinstance(value, str):
value = pack(value, recalculate_length)
if recalculate_length:
length = len(value)
t = ""
while tag > 0:
t = chr( tag & 0xff ) + t
tag = tag >> 8
if length < 0x7F:
l = chr(length)
else:
l = ""
while length > 0:
l = chr( length & 0xff ) + l
length = length >> 8
assert len(l) < 0x7f
l = chr( 0x80 | len(l) ) + l
result.append(t)
result.append(l)
result.append(value)
return "".join(result)
if __name__ == "__main__":
test = binascii.unhexlify("".join(("6f 2b 83 02 2f 00 81 02 01 00 82 03 05 41 26 85" \
+"02 01 00 86 18 60 00 00 00 ff ff b2 00 00 00 ff" \
+"ff dc 00 00 00 ff ff e4 10 00 00 ff ff").split()))
#decoded = decode(test)
#print decoded
#print decode(file("c100").read())
marks = [ ('[', 5, 8) ]
a = binascii.a2b_hex( "".join( "80 01 aa b0 03 81 01 bb ff ff 00".split() ) )
b = unpack( a, with_marks=marks, include_filler=True)
print b
c = pack(b, recalculate_length = True)
print utils.hexdump(a)
print utils.hexdump(c)
loadOids()