diff options
Diffstat (limited to 'tests/python_dependencies/impacket/spnego.py')
-rw-r--r-- | tests/python_dependencies/impacket/spnego.py | 154 |
1 files changed, 77 insertions, 77 deletions
diff --git a/tests/python_dependencies/impacket/spnego.py b/tests/python_dependencies/impacket/spnego.py index 43e65fb5c..98069f579 100644 --- a/tests/python_dependencies/impacket/spnego.py +++ b/tests/python_dependencies/impacket/spnego.py @@ -1,3 +1,4 @@ +from __future__ import print_function # Copyright (c) 2003-2016 CORE Security Technologies # # This software is provided under under a slightly modified version @@ -13,7 +14,7 @@ from struct import pack, unpack, calcsize ############### GSS Stuff ################ -GSS_API_SPNEGO_UUID = '\x2b\x06\x01\x05\x05\x02' +GSS_API_SPNEGO_UUID = '\x2b\x06\x01\x05\x05\x02' ASN1_SEQUENCE = 0x30 ASN1_AID = 0x60 ASN1_OID = 0x06 @@ -33,59 +34,59 @@ MechTypes = { TypesMech = dict((v,k) for k, v in MechTypes.iteritems()) def asn1encode(data = ''): - #res = asn1.SEQUENCE(str).encode() - #import binascii - #print '\nalex asn1encode str: %s\n' % binascii.hexlify(str) - if 0 <= len(data) <= 0x7F: - res = pack('B', len(data)) + data - elif 0x80 <= len(data) <= 0xFF: - res = pack('BB', 0x81, len(data)) + data - elif 0x100 <= len(data) <= 0xFFFF: - res = pack('!BH', 0x82, len(data)) + data - elif 0x10000 <= len(data) <= 0xffffff: - res = pack('!BBH', 0x83, len(data) >> 16, len(data) & 0xFFFF) + data - elif 0x1000000 <= len(data) <= 0xffffffff: - res = pack('!BL', 0x84, len(data)) + data - else: - raise Exception('Error in asn1encode') - return str(res) + #res = asn1.SEQUENCE(str).encode() + #import binascii + #print '\nalex asn1encode str: %s\n' % binascii.hexlify(str) + if 0 <= len(data) <= 0x7F: + res = pack('B', len(data)) + data + elif 0x80 <= len(data) <= 0xFF: + res = pack('BB', 0x81, len(data)) + data + elif 0x100 <= len(data) <= 0xFFFF: + res = pack('!BH', 0x82, len(data)) + data + elif 0x10000 <= len(data) <= 0xffffff: + res = pack('!BBH', 0x83, len(data) >> 16, len(data) & 0xFFFF) + data + elif 0x1000000 <= len(data) <= 0xffffffff: + res = pack('!BL', 0x84, len(data)) + data + else: + raise Exception('Error in asn1encode') + return str(res) def asn1decode(data = ''): - len1 = unpack('B', data[:1])[0] - data = data[1:] - if len1 == 0x81: - pad = calcsize('B') - len2 = unpack('B',data[:pad])[0] - data = data[pad:] - ans = data[:len2] - elif len1 == 0x82: - pad = calcsize('H') - len2 = unpack('!H', data[:pad])[0] - data = data[pad:] - ans = data[:len2] - elif len1 == 0x83: - pad = calcsize('B') + calcsize('!H') - len2, len3 = unpack('!BH', data[:pad]) - data = data[pad:] - ans = data[:len2 << 16 + len3] - elif len1 == 0x84: - pad = calcsize('!L') - len2 = unpack('!L', data[:pad])[0] - data = data[pad:] - ans = data[:len2] - # 1 byte length, string <= 0x7F - else: - pad = 0 - ans = data[:len1] - return ans, len(ans)+pad+1 + len1 = unpack('B', data[:1])[0] + data = data[1:] + if len1 == 0x81: + pad = calcsize('B') + len2 = unpack('B',data[:pad])[0] + data = data[pad:] + ans = data[:len2] + elif len1 == 0x82: + pad = calcsize('H') + len2 = unpack('!H', data[:pad])[0] + data = data[pad:] + ans = data[:len2] + elif len1 == 0x83: + pad = calcsize('B') + calcsize('!H') + len2, len3 = unpack('!BH', data[:pad]) + data = data[pad:] + ans = data[:len2 << 16 + len3] + elif len1 == 0x84: + pad = calcsize('!L') + len2 = unpack('!L', data[:pad])[0] + data = data[pad:] + ans = data[:len2] + # 1 byte length, string <= 0x7F + else: + pad = 0 + ans = data[:len1] + return ans, len(ans)+pad+1 class GSSAPI: -# Generic GSSAPI Header Format +# Generic GSSAPI Header Format def __init__(self, data = None): self.fields = {} self['UUID'] = GSS_API_SPNEGO_UUID if data: - self.fromString(data) + self.fromString(data) pass def __setitem__(self,key,value): @@ -115,27 +116,27 @@ class GSSAPI: if next_byte != ASN1_AID: raise Exception('Unknown AID=%x' % next_byte) data = data[1:] - decode_data, total_bytes = asn1decode(data) + decode_data, total_bytes = asn1decode(data) # Now we should have a OID tag - next_byte = unpack('B',decode_data[:1])[0] + next_byte = unpack('B',decode_data[:1])[0] if next_byte != ASN1_OID: raise Exception('OID tag not found %x' % next_byte) decode_data = decode_data[1:] # Now the OID contents, should be SPNEGO UUID - uuid, total_bytes = asn1decode(decode_data) + uuid, total_bytes = asn1decode(decode_data) self['OID'] = uuid # the rest should be the data self['Payload'] = decode_data[total_bytes:] #pass - + def dump(self): for i in self.fields.keys(): - print "%s: {%r}" % (i,self[i]) + print("%s: {%r}" % (i,self[i])) def getData(self): ans = pack('B',ASN1_AID) ans += asn1encode( - pack('B',ASN1_OID) + + pack('B',ASN1_OID) + asn1encode(self['UUID']) + self['Payload'] ) return ans @@ -163,7 +164,7 @@ class SPNEGO_NegTokenResp: def __init__(self, data = None): self.fields = {} if data: - self.fromString(data) + self.fromString(data) pass def __setitem__(self,key,value): @@ -198,7 +199,7 @@ class SPNEGO_NegTokenResp: if next_byte != ASN1_MECH_TYPE: # MechType not found, could be an AUTH answer if next_byte != ASN1_RESPONSE_TOKEN: - raise Exception('MechType/ResponseToken tag not found %x' % next_byte) + raise Exception('MechType/ResponseToken tag not found %x' % next_byte) else: decode_data2 = decode_data[1:] decode_data2, total_bytes = asn1decode(decode_data2) @@ -245,30 +246,30 @@ class SPNEGO_NegTokenResp: def dump(self): for i in self.fields.keys(): - print "%s: {%r}" % (i,self[i]) - + print("%s: {%r}" % (i,self[i])) + def getData(self): ans = pack('B',SPNEGO_NegTokenResp.SPNEGO_NEG_TOKEN_RESP) - if self.fields.has_key('NegResult') and self.fields.has_key('SupportedMech'): + if 'NegResult' in self.fields and 'SupportedMech' in self.fields: # Server resp ans += asn1encode( pack('B', ASN1_SEQUENCE) + asn1encode( pack('B',SPNEGO_NegTokenResp.SPNEGO_NEG_TOKEN_TARG) + asn1encode( - pack('B',ASN1_ENUMERATED) + + pack('B',ASN1_ENUMERATED) + asn1encode( self['NegResult'] )) + pack('B',ASN1_SUPPORTED_MECH) + - asn1encode( + asn1encode( pack('B',ASN1_OID) + asn1encode(self['SupportedMech'])) + pack('B',ASN1_RESPONSE_TOKEN ) + asn1encode( pack('B', ASN1_OCTET_STRING) + asn1encode(self['ResponseToken'])))) - elif self.fields.has_key('NegResult'): + elif 'NegResult' in self.fields: # Server resp ans += asn1encode( - pack('B', ASN1_SEQUENCE) + + pack('B', ASN1_SEQUENCE) + asn1encode( pack('B', SPNEGO_NegTokenResp.SPNEGO_NEG_TOKEN_TARG) + asn1encode( @@ -285,24 +286,24 @@ class SPNEGO_NegTokenResp: return ans class SPNEGO_NegTokenInit(GSSAPI): - # http://tools.ietf.org/html/rfc4178#page-8 + # http://tools.ietf.org/html/rfc4178#page-8 # NegTokeInit :: = SEQUENCE { - # mechTypes [0] MechTypeList, + # mechTypes [0] MechTypeList, # reqFlags [1] ContextFlags OPTIONAL, - # mechToken [2] OCTET STRING OPTIONAL, + # mechToken [2] OCTET STRING OPTIONAL, # mechListMIC [3] OCTET STRING OPTIONAL, # } SPNEGO_NEG_TOKEN_INIT = 0xa0 def fromString(self, data = 0): GSSAPI.fromString(self, data) payload = self['Payload'] - next_byte = unpack('B', payload[:1])[0] + next_byte = unpack('B', payload[:1])[0] if next_byte != SPNEGO_NegTokenInit.SPNEGO_NEG_TOKEN_INIT: raise Exception('NegTokenInit not found %x' % next_byte) payload = payload[1:] decode_data, total_bytes = asn1decode(payload) # Now we should have a SEQUENCE Tag - next_byte = unpack('B', decode_data[:1])[0] + next_byte = unpack('B', decode_data[:1])[0] if next_byte != ASN1_SEQUENCE: raise Exception('SEQUENCE tag not found %x' % next_byte) decode_data = decode_data[1:] @@ -321,14 +322,14 @@ class SPNEGO_NegTokenInit(GSSAPI): # And finally we should have the MechTypes self['MechTypes'] = [] while decode_data: - next_byte = unpack('B', decode_data[:1])[0] - if next_byte != ASN1_OID: - # Not a valid OID, there must be something else we won't unpack - break - decode_data = decode_data[1:] - item, total_bytes = asn1decode(decode_data) - self['MechTypes'].append(item) - decode_data = decode_data[total_bytes:] + next_byte = unpack('B', decode_data[:1])[0] + if next_byte != ASN1_OID: + # Not a valid OID, there must be something else we won't unpack + break + decode_data = decode_data[1:] + item, total_bytes = asn1decode(decode_data) + self['MechTypes'].append(item) + decode_data = decode_data[total_bytes:] # Do we have MechTokens as well? decode_data = remaining_data[total_bytes3:] @@ -352,7 +353,7 @@ class SPNEGO_NegTokenInit(GSSAPI): mechToken = '' # Do we have tokens to send? - if self.fields.has_key('MechToken'): + if 'MechToken' in self.fields: mechToken = pack('B', ASN1_MECH_TOKEN) + asn1encode( pack('B', ASN1_OCTET_STRING) + asn1encode( self['MechToken'])) @@ -363,10 +364,9 @@ class SPNEGO_NegTokenInit(GSSAPI): asn1encode( pack('B', ASN1_MECH_TYPE) + asn1encode( - pack('B', ASN1_SEQUENCE) + + pack('B', ASN1_SEQUENCE) + asn1encode(mechTypes)) + mechToken )) self['Payload'] = ans return GSSAPI.getData(self) - |