aboutsummaryrefslogtreecommitdiff
path: root/tests/python_dependencies/impacket/spnego.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/python_dependencies/impacket/spnego.py')
-rw-r--r--tests/python_dependencies/impacket/spnego.py372
1 files changed, 372 insertions, 0 deletions
diff --git a/tests/python_dependencies/impacket/spnego.py b/tests/python_dependencies/impacket/spnego.py
new file mode 100644
index 000000000..f177d18f7
--- /dev/null
+++ b/tests/python_dependencies/impacket/spnego.py
@@ -0,0 +1,372 @@
+# Copyright (c) 2003-2016 CORE Security Technologies
+#
+# This software is provided under under a slightly modified version
+# of the Apache Software License. See the accompanying LICENSE file
+# for more information.
+#
+# Author: Alberto Solino (beto@coresecurity.com)
+#
+# Description:
+# SPNEGO functions used by SMB, SMB2/3 and DCERPC
+#
+
+from struct import pack, unpack, calcsize
+
+############### GSS Stuff ################
+GSS_API_SPNEGO_UUID = '\x2b\x06\x01\x05\x05\x02'
+ASN1_SEQUENCE = 0x30
+ASN1_AID = 0x60
+ASN1_OID = 0x06
+ASN1_OCTET_STRING = 0x04
+ASN1_MECH_TYPE = 0xa0
+ASN1_MECH_TOKEN = 0xa2
+ASN1_SUPPORTED_MECH = 0xa1
+ASN1_RESPONSE_TOKEN = 0xa2
+ASN1_ENUMERATED = 0x0a
+MechTypes = {
+'+\x06\x01\x04\x01\x827\x02\x02\x1e': 'SNMPv2-SMI::enterprises.311.2.2.30',
+'+\x06\x01\x04\x01\x827\x02\x02\n': 'NTLMSSP - Microsoft NTLM Security Support Provider',
+'*\x86H\x82\xf7\x12\x01\x02\x02': 'MS KRB5 - Microsoft Kerberos 5',
+'*\x86H\x86\xf7\x12\x01\x02\x02': 'KRB5 - Kerberos 5',
+'*\x86H\x86\xf7\x12\x01\x02\x02\x03': 'KRB5 - Kerberos 5 - User to User'
+}
+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)
+
+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
+
+class GSSAPI:
+# Generic GSSAPI Header Format
+ def __init__(self, data = None):
+ self.fields = {}
+ self['UUID'] = GSS_API_SPNEGO_UUID
+ if data:
+ self.fromString(data)
+ pass
+
+ def __setitem__(self,key,value):
+ self.fields[key] = value
+
+ def __getitem__(self, key):
+ return self.fields[key]
+
+ def __delitem__(self, key):
+ del self.fields[key]
+
+ def __len__(self):
+ return len(self.getData())
+
+ def __str__(self):
+ return len(self.getData())
+
+ def fromString(self, data = None):
+ # Manual parse of the GSSAPI Header Format
+ # It should be something like
+ # AID = 0x60 TAG, BER Length
+ # OID = 0x06 TAG
+ # GSSAPI OID
+ # UUID data (BER Encoded)
+ # Payload
+ next_byte = unpack('B',data[:1])[0]
+ if next_byte != ASN1_AID:
+ raise Exception('Unknown AID=%x' % next_byte)
+ data = data[1:]
+ decode_data, total_bytes = asn1decode(data)
+ # Now we should have a OID tag
+ 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)
+ 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])
+
+ def getData(self):
+ ans = pack('B',ASN1_AID)
+ ans += asn1encode(
+ pack('B',ASN1_OID) +
+ asn1encode(self['UUID']) +
+ self['Payload'] )
+ return ans
+
+class SPNEGO_NegTokenResp:
+ # http://tools.ietf.org/html/rfc4178#page-9
+ # NegTokenResp ::= SEQUENCE {
+ # negState [0] ENUMERATED {
+ # accept-completed (0),
+ # accept-incomplete (1),
+ # reject (2),
+ # request-mic (3)
+ # } OPTIONAL,
+ # -- REQUIRED in the first reply from the target
+ # supportedMech [1] MechType OPTIONAL,
+ # -- present only in the first reply from the target
+ # responseToken [2] OCTET STRING OPTIONAL,
+ # mechListMIC [3] OCTET STRING OPTIONAL,
+ # ...
+ # }
+ # This structure is not prepended by a GSS generic header!
+ SPNEGO_NEG_TOKEN_RESP = 0xa1
+ SPNEGO_NEG_TOKEN_TARG = 0xa0
+
+ def __init__(self, data = None):
+ self.fields = {}
+ if data:
+ self.fromString(data)
+ pass
+
+ def __setitem__(self,key,value):
+ self.fields[key] = value
+
+ def __getitem__(self, key):
+ return self.fields[key]
+
+ def __delitem__(self, key):
+ del self.fields[key]
+
+ def __len__(self):
+ return len(self.getData())
+
+ def __str__(self):
+ return len(self.getData())
+
+ def fromString(self, data = 0):
+ payload = data
+ next_byte = unpack('B', payload[:1])[0]
+ if next_byte != SPNEGO_NegTokenResp.SPNEGO_NEG_TOKEN_RESP:
+ raise Exception('NegTokenResp not found %x' % next_byte)
+ payload = payload[1:]
+ decode_data, total_bytes = asn1decode(payload)
+ 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:]
+ decode_data, total_bytes = asn1decode(decode_data)
+ next_byte = unpack('B',decode_data[:1])[0]
+
+ 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)
+ else:
+ decode_data2 = decode_data[1:]
+ decode_data2, total_bytes = asn1decode(decode_data2)
+ next_byte = unpack('B', decode_data2[:1])[0]
+ if next_byte != ASN1_ENUMERATED:
+ raise Exception('Enumerated tag not found %x' % next_byte)
+ item, total_bytes2 = asn1decode(decode_data)
+ self['NegResult'] = item
+ decode_data = decode_data[1:]
+ decode_data = decode_data[total_bytes:]
+
+ # Do we have more data?
+ if len(decode_data) == 0:
+ return
+
+ next_byte = unpack('B', decode_data[:1])[0]
+ if next_byte != ASN1_SUPPORTED_MECH:
+ if next_byte != ASN1_RESPONSE_TOKEN:
+ raise Exception('Supported Mech/ResponseToken tag not found %x' % next_byte)
+ else:
+ decode_data2 = decode_data[1:]
+ decode_data2, total_bytes = asn1decode(decode_data2)
+ next_byte = unpack('B', decode_data2[:1])[0]
+ if next_byte != ASN1_OID:
+ raise Exception('OID tag not found %x' % next_byte)
+ decode_data2 = decode_data2[1:]
+ item, total_bytes2 = asn1decode(decode_data2)
+ self['SuportedMech'] = item
+
+ decode_data = decode_data[1:]
+ decode_data = decode_data[total_bytes:]
+ next_byte = unpack('B', decode_data[:1])[0]
+ if next_byte != ASN1_RESPONSE_TOKEN:
+ raise Exception('Response token tag not found %x' % next_byte)
+
+ decode_data = decode_data[1:]
+ decode_data, total_bytes = asn1decode(decode_data)
+ next_byte = unpack('B', decode_data[:1])[0]
+ if next_byte != ASN1_OCTET_STRING:
+ raise Exception('Octet string token tag not found %x' % next_byte)
+ decode_data = decode_data[1:]
+ decode_data, total_bytes = asn1decode(decode_data)
+ self['ResponseToken'] = decode_data
+
+ def dump(self):
+ for i in self.fields.keys():
+ 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'):
+ # Server resp
+ ans += asn1encode(
+ pack('B', ASN1_SEQUENCE) +
+ asn1encode(
+ pack('B',SPNEGO_NegTokenResp.SPNEGO_NEG_TOKEN_TARG) +
+ asn1encode(
+ pack('B',ASN1_ENUMERATED) +
+ asn1encode( self['NegResult'] )) +
+ pack('B',ASN1_SUPPORTED_MECH) +
+ 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'):
+ # Server resp
+ ans += asn1encode(
+ pack('B', ASN1_SEQUENCE) +
+ asn1encode(
+ pack('B', SPNEGO_NegTokenResp.SPNEGO_NEG_TOKEN_TARG) +
+ asn1encode(
+ pack('B',ASN1_ENUMERATED) +
+ asn1encode( self['NegResult'] ))))
+ else:
+ # Client resp
+ ans += asn1encode(
+ pack('B', ASN1_SEQUENCE) +
+ asn1encode(
+ pack('B', ASN1_RESPONSE_TOKEN) +
+ asn1encode(
+ pack('B', ASN1_OCTET_STRING) + asn1encode(self['ResponseToken']))))
+ return ans
+
+class SPNEGO_NegTokenInit(GSSAPI):
+ # http://tools.ietf.org/html/rfc4178#page-8
+ # NegTokeInit :: = SEQUENCE {
+ # mechTypes [0] MechTypeList,
+ # reqFlags [1] ContextFlags 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]
+ 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]
+ if next_byte != ASN1_SEQUENCE:
+ raise Exception('SEQUENCE tag not found %x' % next_byte)
+ decode_data = decode_data[1:]
+ decode_data, total_bytes2 = asn1decode(decode_data)
+ next_byte = unpack('B',decode_data[:1])[0]
+ if next_byte != ASN1_MECH_TYPE:
+ raise Exception('MechType tag not found %x' % next_byte)
+ decode_data = decode_data[1:]
+ remaining_data = decode_data
+ decode_data, total_bytes3 = asn1decode(decode_data)
+ 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:]
+ decode_data, total_bytes4 = asn1decode(decode_data)
+ # 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:]
+
+ # Do we have MechTokens as well?
+ decode_data = remaining_data[total_bytes3:]
+ if len(decode_data) > 0:
+ next_byte = unpack('B', decode_data[:1])[0]
+ if next_byte == ASN1_MECH_TOKEN:
+ # We have tokens in here!
+ decode_data = decode_data[1:]
+ decode_data, total_bytes = asn1decode(decode_data)
+ next_byte = unpack('B', decode_data[:1])[0]
+ if next_byte == ASN1_OCTET_STRING:
+ decode_data = decode_data[1:]
+ decode_data, total_bytes = asn1decode(decode_data)
+ self['MechToken'] = decode_data
+
+ def getData(self):
+ mechTypes = ''
+ for i in self['MechTypes']:
+ mechTypes += pack('B', ASN1_OID)
+ mechTypes += asn1encode(i)
+
+ mechToken = ''
+ # Do we have tokens to send?
+ if self.fields.has_key('MechToken'):
+ mechToken = pack('B', ASN1_MECH_TOKEN) + asn1encode(
+ pack('B', ASN1_OCTET_STRING) + asn1encode(
+ self['MechToken']))
+
+ ans = pack('B',SPNEGO_NegTokenInit.SPNEGO_NEG_TOKEN_INIT)
+ ans += asn1encode(
+ pack('B', ASN1_SEQUENCE) +
+ asn1encode(
+ pack('B', ASN1_MECH_TYPE) +
+ asn1encode(
+ pack('B', ASN1_SEQUENCE) +
+ asn1encode(mechTypes)) + mechToken ))
+
+
+ self['Payload'] = ans
+ return GSSAPI.getData(self)
+
generated by cgit v1.2.3 (git 2.25.1) at 2024-07-01 01:35:15 +0000