def main ():
n = int (sys.argv[2], 16)
keysize = n.bit_length() / 16
with open (sys.argv[1], "rb") as f:
chunk = f.read (16384)
while chunk:
for offset in xrange (0, len (chunk) - keysize):
p = long (''.join (["%02x" % ord (chunk[x]) for x in xrange (offset + keysize - 1, offset - 1, -1)]).strip(), 16)
if gmpy.is_prime (p) and p != n and n % p == 0:
e = 65537
q = n / p
phi = (p - 1) * (q - 1)
d = gmpy.invert (e, phi)
dp = d % (p - 1)
dq = d % (q - 1)
qinv = gmpy.invert (q, p)
seq = Sequence()
for x in [0, n, e, d, p, q, dp, dq, qinv]:
seq.setComponentByPosition (len (seq), Integer (x))
print "\n\n-----BEGIN RSA PRIVATE KEY-----\n%s-----END RSA PRIVATE KEY-----\n\n" % base64.encodestring(encoder.encode (seq))
sys.exit (0)
chunk = f.read (16384)
print "private key not found :("
python类encode()的实例源码
def build_as_req(target_realm, user_name, key, current_time, nonce, pac_request=None):
req_body = build_req_body(target_realm, 'krbtgt', '', nonce, cname=user_name)
pa_ts = build_pa_enc_timestamp(current_time, key)
as_req = AsReq()
as_req['pvno'] = 5
as_req['msg-type'] = 10
as_req['padata'] = None
as_req['padata'][0] = None
as_req['padata'][0]['padata-type'] = 2
as_req['padata'][0]['padata-value'] = encode(pa_ts)
if pac_request is not None:
pa_pac_request = KerbPaPacRequest()
pa_pac_request['include-pac'] = pac_request
as_req['padata'][1] = None
as_req['padata'][1]['padata-type'] = 128
as_req['padata'][1]['padata-value'] = encode(pa_pac_request)
as_req['req-body'] = _v(4, req_body)
return as_req
def _ValidatePubkeyGeneric(self, signing_cert, digest_alg, payload,
enc_digest):
m2_cert = M2_X509.load_cert_der_string(der_encoder.encode(signing_cert))
pubkey = m2_cert.get_pubkey()
pubkey.reset_context(digest_alg().name)
pubkey.verify_init()
pubkey.verify_update(payload)
v = pubkey.verify_final(enc_digest)
if v != 1:
self.openssl_error = M2_Err.get_error()
# Let's try a special case. I have no idea how I would determine when
# to use this instead of the above code, so I'll always try. The
# observed problem was that for one countersignature (RSA on MD5),
# the encrypted digest did not contain an ASN.1 structure, but the
# raw hash value instead.
try:
rsa = pubkey.get_rsa()
except ValueError:
# It's not an RSA key, just fall through...
pass
else:
clear = rsa.public_decrypt(enc_digest, M2_RSA.pkcs1_padding)
if digest_alg(payload).digest() == clear:
return 1
return v
def toTGT(self):
tgt_rep = AS_REP()
tgt_rep['pvno'] = 5
tgt_rep['msg-type'] = int(constants.ApplicationTagNumbers.AP_REP.value)
tgt_rep['crealm'] = self['server'].realm['data']
# Fake EncryptedData
tgt_rep['enc-part'] = None
tgt_rep['enc-part']['etype'] = 1
tgt_rep['enc-part']['cipher'] = ''
seq_set(tgt_rep, 'cname', self['client'].toPrincipal().components_to_asn1)
ticket = types.Ticket()
ticket.from_asn1(self.ticket['data'])
seq_set(tgt_rep,'ticket', ticket.to_asn1)
cipher = crypto._enctype_table[self['key']['keytype']]()
tgt = dict()
tgt['KDC_REP'] = encoder.encode(tgt_rep)
tgt['cipher'] = cipher
tgt['sessionKey'] = crypto.Key(cipher.enctype, str(self['key']['keyvalue']))
return tgt
def toTGS(self):
tgs_rep = TGS_REP()
tgs_rep['pvno'] = 5
tgs_rep['msg-type'] = int(constants.ApplicationTagNumbers.TGS_REP.value)
tgs_rep['crealm'] = self['server'].realm['data']
# Fake EncryptedData
tgs_rep['enc-part'] = None
tgs_rep['enc-part']['etype'] = 1
tgs_rep['enc-part']['cipher'] = ''
seq_set(tgs_rep, 'cname', self['client'].toPrincipal().components_to_asn1)
ticket = types.Ticket()
ticket.from_asn1(self.ticket['data'])
seq_set(tgs_rep,'ticket', ticket.to_asn1)
cipher = crypto._enctype_table[self['key']['keytype']]()
tgs = dict()
tgs['KDC_REP'] = encoder.encode(tgs_rep)
tgs['cipher'] = cipher
tgs['sessionKey'] = crypto.Key(cipher.enctype, str(self['key']['keyvalue']))
return tgs
def check_dir(self, service, path, password = None):
path = string.replace(path,'/', '\\')
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
smb = NewSMBPacket()
smb['Tid'] = tid
smb['Mid'] = 0
cmd = SMBCommand(SMB.SMB_COM_CHECK_DIRECTORY)
cmd['Parameters'] = ''
cmd['Data'] = SMBCheckDirectory_Data(flags = self.__flags2)
cmd['Data']['DirectoryName'] = path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else path
smb.addCommand(cmd)
self.sendSMB(smb)
while 1:
s = self.recvSMB()
if s.isValidAnswer(SMB.SMB_COM_CHECK_DIRECTORY):
return
finally:
self.disconnect_tree(tid)
def rmdir(self, service, path, password = None):
path = string.replace(path,'/', '\\')
# Check that the directory exists
self.check_dir(service, path, password)
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
path = path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else path
smb = NewSMBPacket()
smb['Tid'] = tid
createDir = SMBCommand(SMB.SMB_COM_DELETE_DIRECTORY)
createDir['Data'] = SMBDeleteDirectory_Data(flags=self.__flags2)
createDir['Data']['DirectoryName'] = path
smb.addCommand(createDir)
self.sendSMB(smb)
while 1:
s = self.recvSMB()
if s.isValidAnswer(SMB.SMB_COM_DELETE_DIRECTORY):
return
finally:
self.disconnect_tree(tid)
def mkdir(self, service, path, password = None):
path = string.replace(path,'/', '\\')
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
path = path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else path
smb = NewSMBPacket()
smb['Tid'] = tid
smb['Mid'] = 0
createDir = SMBCommand(SMB.SMB_COM_CREATE_DIRECTORY)
createDir['Data'] = SMBCreateDirectory_Data(flags=self.__flags2)
createDir['Data']['DirectoryName'] = path
smb.addCommand(createDir)
self.sendSMB(smb)
smb = self.recvSMB()
if smb.isValidAnswer(SMB.SMB_COM_CREATE_DIRECTORY):
return 1
return 0
finally:
self.disconnect_tree(tid)
def rename(self, service, old_path, new_path, password = None):
old_path = string.replace(old_path,'/', '\\')
new_path = string.replace(new_path,'/', '\\')
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
smb = NewSMBPacket()
smb['Tid'] = tid
smb['Mid'] = 0
renameCmd = SMBCommand(SMB.SMB_COM_RENAME)
renameCmd['Parameters'] = SMBRename_Parameters()
renameCmd['Parameters']['SearchAttributes'] = ATTR_SYSTEM | ATTR_HIDDEN | ATTR_DIRECTORY
renameCmd['Data'] = SMBRename_Data(flags = self.__flags2)
renameCmd['Data']['OldFileName'] = old_path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else old_path
renameCmd['Data']['NewFileName'] = new_path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else new_path
smb.addCommand(renameCmd)
self.sendSMB(smb)
smb = self.recvSMB()
if smb.isValidAnswer(SMB.SMB_COM_RENAME):
return 1
return 0
finally:
self.disconnect_tree(tid)
def _prepare_auth_attributes_to_digest(auth_attributes_instance):
"""
Prepares autheticated attributes field to digesting process.
Replaces implicit tag with SET tag.
"""
implicit_tag = chr(0xa0) # implicit tag of the set of authAtt
set_tag = chr(0x31) # tag of the ASN type "set"
# encode authentcatdAttributes instance into DER
attrs = encoder.encode(auth_attributes_instance)
# remove implicit tag
if (attrs[0] == implicit_tag):
attrs = attrs.lstrip(implicit_tag)
attrs = str(set_tag) + attrs
return attrs
def main ():
n = int (sys.argv[2], 16)
keysize = n.bit_length() / 16
with open (sys.argv[1], "rb") as f:
chunk = f.read (16384)
while chunk:
for offset in xrange (0, len (chunk) - keysize):
p = long (''.join (["%02x" % ord (chunk[x]) for x in xrange (offset + keysize - 1, offset - 1, -1)]).strip(), 16)
if gmpy.is_prime (p) and p != n and n % p == 0:
e = 65537
q = n / p
phi = (p - 1) * (q - 1)
d = gmpy.invert (e, phi)
dp = d % (p - 1)
dq = d % (q - 1)
qinv = gmpy.invert (q, p)
seq = Sequence()
for x in [0, n, e, d, p, q, dp, dq, qinv]:
seq.setComponentByPosition (len (seq), Integer (x))
print "\n\n-----BEGIN RSA PRIVATE KEY-----\n%s-----END RSA PRIVATE KEY-----\n\n" % base64.encodestring(encoder.encode (seq))
sys.exit (0)
chunk = f.read (16384)
print "private key not found :("
def generateNegotiateSecurityBlob(ntlm_data):
mech_token = univ.OctetString(ntlm_data).subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 2))
mech_types = MechTypeList().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))
mech_types.setComponentByPosition(0, univ.ObjectIdentifier('1.3.6.1.4.1.311.2.2.10'))
n = NegTokenInit().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))
n.setComponentByName('mechTypes', mech_types)
n.setComponentByName('mechToken', mech_token)
nt = NegotiationToken()
nt.setComponentByName('negTokenInit', n)
ct = ContextToken()
ct.setComponentByName('thisMech', univ.ObjectIdentifier('1.3.6.1.5.5.2'))
ct.setComponentByName('innerContextToken', nt)
return encoder.encode(ct)
def ssh_gss_oids(self, mode="client"):
"""
This method returns a single OID, because we only support the
Kerberos V5 mechanism.
:param str mode: Client for client mode and server for server mode
:return: A byte sequence containing the number of supported
OIDs, the length of the OID and the actual OID encoded with
DER
:rtype: Bytes
:note: In server mode we just return the OID length and the DER encoded
OID.
"""
OIDs = self._make_uint32(1)
krb5_OID = encoder.encode(ObjectIdentifier(self._krb5_mech))
OID_len = self._make_uint32(len(krb5_OID))
if mode == "server":
return OID_len + krb5_OID
return OIDs + OID_len + krb5_OID
def generateNegotiateSecurityBlob(ntlm_data):
mech_token = univ.OctetString(ntlm_data).subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 2))
mech_types = MechTypeList().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))
mech_types.setComponentByPosition(0, univ.ObjectIdentifier('1.3.6.1.4.1.311.2.2.10'))
n = NegTokenInit().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))
n.setComponentByName('mechTypes', mech_types)
n.setComponentByName('mechToken', mech_token)
nt = NegotiationToken()
nt.setComponentByName('negTokenInit', n)
ct = ContextToken()
ct.setComponentByName('thisMech', univ.ObjectIdentifier('1.3.6.1.5.5.2'))
ct.setComponentByName('innerContextToken', nt)
return encoder.encode(ct)
def toTGT(self):
tgt_rep = AS_REP()
tgt_rep['pvno'] = 5
tgt_rep['msg-type'] = int(constants.ApplicationTagNumbers.AS_REP.value)
tgt_rep['crealm'] = self['server'].realm['data']
# Fake EncryptedData
tgt_rep['enc-part'] = None
tgt_rep['enc-part']['etype'] = 1
tgt_rep['enc-part']['cipher'] = ''
seq_set(tgt_rep, 'cname', self['client'].toPrincipal().components_to_asn1)
ticket = types.Ticket()
ticket.from_asn1(self.ticket['data'])
seq_set(tgt_rep,'ticket', ticket.to_asn1)
cipher = crypto._enctype_table[self['key']['keytype']]()
tgt = dict()
tgt['KDC_REP'] = encoder.encode(tgt_rep)
tgt['cipher'] = cipher
tgt['sessionKey'] = crypto.Key(cipher.enctype, str(self['key']['keyvalue']))
return tgt
def rmdir(self, service, path, password = None):
path = string.replace(path,'/', '\\')
# Check that the directory exists
self.check_dir(service, path, password)
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
path = path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else path
smb = NewSMBPacket()
smb['Tid'] = tid
createDir = SMBCommand(SMB.SMB_COM_DELETE_DIRECTORY)
createDir['Data'] = SMBDeleteDirectory_Data(flags=self.__flags2)
createDir['Data']['DirectoryName'] = path
smb.addCommand(createDir)
self.sendSMB(smb)
while 1:
s = self.recvSMB()
if s.isValidAnswer(SMB.SMB_COM_DELETE_DIRECTORY):
return
finally:
self.disconnect_tree(tid)
def mkdir(self, service, path, password = None):
path = string.replace(path,'/', '\\')
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
path = path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else path
smb = NewSMBPacket()
smb['Tid'] = tid
smb['Mid'] = 0
createDir = SMBCommand(SMB.SMB_COM_CREATE_DIRECTORY)
createDir['Data'] = SMBCreateDirectory_Data(flags=self.__flags2)
createDir['Data']['DirectoryName'] = path
smb.addCommand(createDir)
self.sendSMB(smb)
smb = self.recvSMB()
if smb.isValidAnswer(SMB.SMB_COM_CREATE_DIRECTORY):
return 1
return 0
finally:
self.disconnect_tree(tid)
def rename(self, service, old_path, new_path, password = None):
old_path = string.replace(old_path,'/', '\\')
new_path = string.replace(new_path,'/', '\\')
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
smb = NewSMBPacket()
smb['Tid'] = tid
smb['Mid'] = 0
renameCmd = SMBCommand(SMB.SMB_COM_RENAME)
renameCmd['Parameters'] = SMBRename_Parameters()
renameCmd['Parameters']['SearchAttributes'] = ATTR_SYSTEM | ATTR_HIDDEN | ATTR_DIRECTORY
renameCmd['Data'] = SMBRename_Data(flags = self.__flags2)
renameCmd['Data']['OldFileName'] = old_path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else old_path
renameCmd['Data']['NewFileName'] = new_path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else new_path
smb.addCommand(renameCmd)
self.sendSMB(smb)
smb = self.recvSMB()
if smb.isValidAnswer(SMB.SMB_COM_RENAME):
return 1
return 0
finally:
self.disconnect_tree(tid)
def generateNegotiateSecurityBlob(ntlm_data):
mech_token = univ.OctetString(ntlm_data).subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 2))
mech_types = MechTypeList().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))
mech_types.setComponentByPosition(0, univ.ObjectIdentifier('1.3.6.1.4.1.311.2.2.10'))
n = NegTokenInit().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))
n.setComponentByName('mechTypes', mech_types)
n.setComponentByName('mechToken', mech_token)
nt = NegotiationToken()
nt.setComponentByName('negTokenInit', n)
ct = ContextToken()
ct.setComponentByName('thisMech', univ.ObjectIdentifier('1.3.6.1.5.5.2'))
ct.setComponentByName('innerContextToken', nt)
return encoder.encode(ct)
securityblob.py 文件源码
项目：plugin.video.streamondemand-pureita
作者: orione7
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def generateNegotiateSecurityBlob(ntlm_data):
mech_token = univ.OctetString(ntlm_data).subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 2))
mech_types = MechTypeList().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))
mech_types.setComponentByPosition(0, univ.ObjectIdentifier('1.3.6.1.4.1.311.2.2.10'))
n = NegTokenInit().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))
n.setComponentByName('mechTypes', mech_types)
n.setComponentByName('mechToken', mech_token)
nt = NegotiationToken()
nt.setComponentByName('negTokenInit', n)
ct = ContextToken()
ct.setComponentByName('thisMech', univ.ObjectIdentifier('1.3.6.1.5.5.2'))
ct.setComponentByName('innerContextToken', nt)
return encoder.encode(ct)
def ssh_gss_oids(self, mode="client"):
"""
This method returns a single OID, because we only support the
Kerberos V5 mechanism.
:param str mode: Client for client mode and server for server mode
:return: A byte sequence containing the number of supported
OIDs, the length of the OID and the actual OID encoded with
DER
:rtype: Bytes
:note: In server mode we just return the OID length and the DER encoded
OID.
"""
OIDs = self._make_uint32(1)
krb5_OID = encoder.encode(ObjectIdentifier(self._krb5_mech))
OID_len = self._make_uint32(len(krb5_OID))
if mode == "server":
return OID_len + krb5_OID
return OIDs + OID_len + krb5_OID
def ssh_gss_oids(self, mode="client"):
"""
This method returns a single OID, because we only support the
Kerberos V5 mechanism.
:param str mode: Client for client mode and server for server mode
:return: A byte sequence containing the number of supported
OIDs, the length of the OID and the actual OID encoded with
DER
:rtype: Bytes
:note: In server mode we just return the OID length and the DER encoded
OID.
"""
OIDs = self._make_uint32(1)
krb5_OID = encoder.encode(ObjectIdentifier(self._krb5_mech))
OID_len = self._make_uint32(len(krb5_OID))
if mode == "server":
return OID_len + krb5_OID
return OIDs + OID_len + krb5_OID
def toTGT(self):
tgt_rep = AS_REP()
tgt_rep['pvno'] = 5
tgt_rep['msg-type'] = int(constants.ApplicationTagNumbers.AP_REP.value)
tgt_rep['crealm'] = self['server'].realm['data']
# Fake EncryptedData
tgt_rep['enc-part'] = None
tgt_rep['enc-part']['etype'] = 1
tgt_rep['enc-part']['cipher'] = ''
seq_set(tgt_rep, 'cname', self['client'].toPrincipal().components_to_asn1)
ticket = types.Ticket()
ticket.from_asn1(self.ticket['data'])
seq_set(tgt_rep,'ticket', ticket.to_asn1)
cipher = crypto._enctype_table[self['key']['keytype']]()
tgt = dict()
tgt['KDC_REP'] = encoder.encode(tgt_rep)
tgt['cipher'] = cipher
tgt['sessionKey'] = crypto.Key(cipher.enctype, str(self['key']['keyvalue']))
return tgt
def check_dir(self, service, path, password = None):
path = string.replace(path,'/', '\\')
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
smb = NewSMBPacket()
smb['Tid'] = tid
smb['Mid'] = 0
cmd = SMBCommand(SMB.SMB_COM_CHECK_DIRECTORY)
cmd['Parameters'] = ''
cmd['Data'] = SMBCheckDirectory_Data(flags = self.__flags2)
cmd['Data']['DirectoryName'] = path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else path
smb.addCommand(cmd)
self.sendSMB(smb)
while 1:
s = self.recvSMB()
if s.isValidAnswer(SMB.SMB_COM_CHECK_DIRECTORY):
return
finally:
self.disconnect_tree(tid)
def mkdir(self, service, path, password = None):
path = string.replace(path,'/', '\\')
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
path = path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else path
smb = NewSMBPacket()
smb['Tid'] = tid
smb['Mid'] = 0
createDir = SMBCommand(SMB.SMB_COM_CREATE_DIRECTORY)
createDir['Data'] = SMBCreateDirectory_Data(flags=self.__flags2)
createDir['Data']['DirectoryName'] = path
smb.addCommand(createDir)
self.sendSMB(smb)
smb = self.recvSMB()
if smb.isValidAnswer(SMB.SMB_COM_CREATE_DIRECTORY):
return 1
return 0
finally:
self.disconnect_tree(tid)
def rename(self, service, old_path, new_path, password = None):
old_path = string.replace(old_path,'/', '\\')
new_path = string.replace(new_path,'/', '\\')
tid = self.tree_connect_andx('\\\\' + self.__remote_name + '\\' + service, password)
try:
smb = NewSMBPacket()
smb['Tid'] = tid
smb['Mid'] = 0
renameCmd = SMBCommand(SMB.SMB_COM_RENAME)
renameCmd['Parameters'] = SMBRename_Parameters()
renameCmd['Parameters']['SearchAttributes'] = ATTR_SYSTEM | ATTR_HIDDEN | ATTR_DIRECTORY
renameCmd['Data'] = SMBRename_Data(flags = self.__flags2)
renameCmd['Data']['OldFileName'] = old_path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else old_path
renameCmd['Data']['NewFileName'] = new_path.encode('utf-16le') if self.__flags2 & SMB.FLAGS2_UNICODE else new_path
smb.addCommand(renameCmd)
self.sendSMB(smb)
smb = self.recvSMB()
if smb.isValidAnswer(SMB.SMB_COM_RENAME):
return 1
return 0
finally:
self.disconnect_tree(tid)
def test_aes_encryption_consistency(self):
# test encryption-decryption for each test message
for msg in self.TEST_MSGS:
# precreate simple ASN.1 structure to encode and encrypt
test_der = encode(OctetString(msg))
# generate random IV and a key
iv = urandom(16)
key = urandom(16)
# encrypt and decrypt message
ct = self.crypto_obj._Crypto__encrypt_with_aes(test_der, key, iv)
pt = self.crypto_obj._Crypto__decrypt_with_aes(ct, key, iv)
# check whether they are equal
self.assertEqual(test_der, pt)
def get_certificate(self, filename):
"""
Return a certificate object by giving the name in the apk file
"""
pkcs7message = self.get_file(filename)
message, _ = decode(pkcs7message)
cert = encode(message[1][3])
# Remove the first identifier·
# byte 0 == identifier, skip
# byte 1 == length. If byte1 & 0x80 > 1, we have long format
# The length of to read bytes is then coded
# in byte1 & 0x7F
cert = cert[2 + (cert[1] & 0x7F) if cert[1] & 0x80 > 1 else 2:]
certificate = x509.load_der_x509_certificate(cert, default_backend())
return certificate
def get_public_resources(self, package_name, locale='\x00\x00'):
self._analyse()
buff = '<?xml version="1.0" encoding="utf-8"?>\n'
buff += '<resources>\n'
try:
for i in self.values[package_name][locale]["public"]:
buff += '<public type="%s" name="%s" id="0x%08x" />\n' % (
i[0], i[1], i[2])
except KeyError:
pass
buff += '</resources>\n'
return buff.encode('utf-8')
def ssh_gss_oids(self, mode="client"):
"""
This method returns a single OID, because we only support the
Kerberos V5 mechanism.
:param str mode: Client for client mode and server for server mode
:return: A byte sequence containing the number of supported
OIDs, the length of the OID and the actual OID encoded with
DER
:note: In server mode we just return the OID length and the DER encoded
OID.
"""
OIDs = self._make_uint32(1)
krb5_OID = encoder.encode(ObjectIdentifier(self._krb5_mech))
OID_len = self._make_uint32(len(krb5_OID))
if mode == "server":
return OID_len + krb5_OID
return OIDs + OID_len + krb5_OID
