def extract_details_from_filepath (nrt_file_path):
"""
Extract some details of call events
"""
# Build a NRT structure from the DER-serialized ASN.1 string
nrt = None
with open (nrt_file_path, mode='rb') as nrt_file:
nrt_file_content = nrt_file.read()
# Undo DER serialization, reconstruct NRT structure
try:
nrt, rest_of_input = ber_decoder.decode (nrt_file_content,
asn1Spec = Nrtrde.Nrtrde())
except SubstrateUnderrunError:
print ("Error in decoding '" + nrt_file_path + "'. Skipping it")
nrt = None
# Translate the NRT structure into a Python one
py_nrt = None
if nrt is not None:
py_nrt = extract_details_from_nrt (nrt)
#
return py_nrt
python类decode()的实例源码
def extract_details_from_stdin ():
"""
Extract details (ie, NRT structures) from stdin
"""
# Extract the NRT structures from the stdin
nrt_file = sys.stdin.buffer.readlines()
for nrt_file_content in nrt_file:
# Undo DER serialization, reconstruct NRT structure
nrt = None
try:
nrt, rest_of_input = ber_decoder.decode (nrt_file_content,
asn1Spec = Nrtrde.Nrtrde())
except (SubstrateUnderrunError, PyAsn1Error) as err:
print ("Error in decoding standard input: " + str(err))
nrt = None
continue
# Translate the NRT structure into a Python one
py_nrt = None
if nrt is not None:
py_nrt = extract_details_from_nrt (nrt)
#
yield py_nrt
def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
wholeMsg, reqPDU=reqPDU):
while wholeMsg:
rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
rspPDU = pMod.apiMessage.getPDU(rspMsg)
# Match response to request
if pMod.apiPDU.getRequestID(reqPDU) == pMod.apiPDU.getRequestID(rspPDU):
# Check for SNMP errors reported
errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
if errorStatus:
print(errorStatus.prettyPrint())
else:
print('INFORM message delivered, response var-binds follow')
for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
transportDispatcher.jobFinished(1)
return wholeMsg
def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
wholeMsg, reqPDU=reqPDU):
while wholeMsg:
rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
rspPDU = pMod.apiMessage.getPDU(rspMsg)
# Match response to request
if pMod.apiPDU.getRequestID(reqPDU) == pMod.apiPDU.getRequestID(rspPDU):
# Check for SNMP errors reported
errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
if errorStatus:
print(errorStatus.prettyPrint())
else:
for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
transportDispatcher.jobFinished(1)
return wholeMsg
def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
wholeMsg, reqPDU=reqPDU):
while wholeMsg:
rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
rspPDU = pMod.apiMessage.getPDU(rspMsg)
# Match response to request
if pMod.apiPDU.getRequestID(reqPDU) == pMod.apiPDU.getRequestID(rspPDU):
# Check for SNMP errors reported
errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
if errorStatus:
print(errorStatus.prettyPrint())
else:
for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
transportDispatcher.jobFinished(1)
return wholeMsg
def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
wholeMsg, reqPDU=reqPDU):
while wholeMsg:
rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
rspPDU = pMod.apiMessage.getPDU(rspMsg)
# Match response to request
if pMod.apiPDU.getRequestID(reqPDU) == pMod.apiPDU.getRequestID(rspPDU):
# Check for SNMP errors reported
errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
if errorStatus:
print(errorStatus.prettyPrint())
else:
for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
transportDispatcher.jobFinished(1)
return wholeMsg
def recv(dispatcher, domain, address, msg):
while msg:
msg_recv, msg = decoder.decode(msg, asn1Spec=pmod.Message())
pdu_recv = pmod.apiMessage.getPDU(msg_recv)
# match response to request as we're broadcasting
if pmod.apiPDU.getRequestID(pdu_send) == pmod.apiPDU.getRequestID(pdu_recv):
ipaddr = address[0]; device = '?'; uptime = '?'; status = '?'; prstat = 0
# retrieve device properties
for oid, val in pmod.apiPDU.getVarBinds(pdu_recv):
oid, val = oid.prettyPrint(), val.prettyPrint()
# skip non-printer devices
if oid == '1.3.6.1.2.1.25.3.2.1.2.1' and val != '1.3.6.1.2.1.25.3.1.5': return
# harvest device information
if oid == '1.3.6.1.2.1.25.3.2.1.3.1': device = val
if oid == '1.3.6.1.2.1.1.3.0': uptime = conv().elapsed(val, 100, True)
if oid == '1.3.6.1.2.1.43.16.5.1.2.1.1': status = val
if oid == '1.3.6.1.2.1.25.3.2.1.5.1' and val: prstat = val[:1]
dispatcher.jobFinished(1)
results[ipaddr] = [device, uptime, status, prstat]
def decodeControlValue(self,encodedControlValue):
ppolicyValue,_ = decoder.decode(encodedControlValue,asn1Spec=PasswordPolicyResponseValue())
warning = ppolicyValue.getComponentByName('warning')
if warning is None:
self.timeBeforeExpiration,self.graceAuthNsRemaining = None,None
else:
timeBeforeExpiration = warning.getComponentByName('timeBeforeExpiration')
if timeBeforeExpiration!=None:
self.timeBeforeExpiration = int(timeBeforeExpiration)
else:
self.timeBeforeExpiration = None
graceAuthNsRemaining = warning.getComponentByName('graceAuthNsRemaining')
if graceAuthNsRemaining!=None:
self.graceAuthNsRemaining = int(graceAuthNsRemaining)
else:
self.graceAuthNsRemaining = None
error = ppolicyValue.getComponentByName('error')
if error is None:
self.error = None
else:
self.error = int(error)
def decodeControlValue(self,encodedControlValue):
ecncValue,_ = decoder.decode(encodedControlValue,asn1Spec=EntryChangeNotificationValue())
self.changeType = int(ecncValue.getComponentByName('changeType'))
if len(ecncValue)==3:
self.previousDN = str(ecncValue.getComponentByName('previousDN'))
self.changeNumber = int(ecncValue.getComponentByName('changeNumber'))
elif len(ecncValue)==2:
if self.changeType==8:
self.previousDN = str(ecncValue.getComponentByName('previousDN'))
self.changeNumber = None
else:
self.previousDN = None
self.changeNumber = int(ecncValue.getComponentByName('changeNumber'))
else:
self.previousDN,self.changeNumber = None,None
return (self.changeType,self.previousDN,self.changeNumber)
def libsnmp_encode_decode():
try:
import libsnmp.rfc1905 as libsnmp_rfc1905
def decode():
libsnmp_rfc1905.Message().decode(ENCODED_MESSAGE)
encode_time = float('inf')
decode_time = timeit.timeit(decode, number=ITERATIONS)
except ImportError:
encode_time = float('inf')
decode_time = float('inf')
print('Unable to import libsnmp.')
except SyntaxError:
encode_time = float('inf')
decode_time = float('inf')
print('Syntax error in libsnmp.')
return encode_time, decode_time
def pyasn1_encode_decode():
try:
from pysnmp.proto import api
from pyasn1.codec.ber import decoder
snmp_v1 = api.protoModules[api.protoVersion1].Message()
def decode():
decoder.decode(ENCODED_MESSAGE, asn1Spec=snmp_v1)
encode_time = float('inf')
decode_time = timeit.timeit(decode, number=ITERATIONS)
except ImportError:
encode_time = float('inf')
decode_time = float('inf')
print('Unable to import pyasn1.')
return encode_time, decode_time
def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
wholeMsg, reqPDU=reqPDU):
while wholeMsg:
rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
rspPDU = pMod.apiMessage.getPDU(rspMsg)
# Match response to request
if pMod.apiPDU.getRequestID(reqPDU)==pMod.apiPDU.getRequestID(rspPDU):
# Check for SNMP errors reported
errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
if errorStatus:
print(errorStatus.prettyPrint())
else:
for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
transportDispatcher.jobFinished(1)
return wholeMsg
def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
wholeMsg, reqPDU=reqPDU):
while wholeMsg:
rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
rspPDU = pMod.apiMessage.getPDU(rspMsg)
# Match response to request
if pMod.apiPDU.getRequestID(reqPDU)==pMod.apiPDU.getRequestID(rspPDU):
# Check for SNMP errors reported
errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
if errorStatus:
print(errorStatus.prettyPrint())
else:
for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
transportDispatcher.jobFinished(1)
return wholeMsg
def TraverseRdn(rdn):
"""Traverses RDN structure and returns string encoding of the DN.
Args:
rdn: ASN.1 SET (or SEQUENCE) containing RDNs (relative distinguished
names), as identified by type / value pairs. A typical input would
be of type X.509 RelativeDistinguishedName.
Returns:
A dict representing the Distinguished Name.
"""
val = dict()
for n in rdn:
# Note that this does not work for e.g. DC which is present
# multiple times.
# For a real DN parser, make sure to follow the spec in regards
# to multiple occurence of a field in subsequent RDNs, maintaining
# original ordering etc.
# TODO(user): What about elements other than [0]??
name = DistinguishedName.OidToName(n[0]['type'])
value = decoder.decode(n[0]['value'])
if name in val:
val[name] = str(value[0]) + ', ' + val.get(name, '')
else:
val[name] = str(value[0])
return val
def _ParseTimestamp(self, time_asn1):
# Parses countersignature timestamp according to RFC3280, section 4.1.2.5+
timestamp_choice, rest = decoder.decode(time_asn1,
asn1Spec=pkcs7.SigningTime())
if rest: raise Asn1Error('Extra unparsed content.')
return timestamp_choice.ToPythonEpochTime()
def _ParseOpusInfo(self, opus_info_asn1):
spc_opus_info, rest = decoder.decode(opus_info_asn1,
asn1Spec=spc.SpcSpOpusInfo())
if rest: raise Asn1Error('Extra unparsed content.')
if spc_opus_info['programName']:
# According to spec, this should always be a Unicode string. However,
# the ASN.1 syntax allows both ASCII and Unicode. So, let's be careful.
opus_prog_name = spc_opus_info['programName']
uni_name = opus_prog_name['unicode']
ascii_name = opus_prog_name['ascii']
if ascii_name and uni_name:
# WTF? This is supposed to be a CHOICE
raise Asn1Error('Both elements of a choice are present.')
elif uni_name:
program_name = str(uni_name).decode('utf-16-be')
elif ascii_name:
program_name = str(ascii_name)
else:
raise Asn1Error('No element of opusInfo choice is present.')
else:
# According to spec, there should always be a program name,
# and be it zero-length. But let's be gentle, since ASN.1 marks
# this field als optional.
program_name = None
# Again, according to Authenticode spec, the moreInfo field should always
# be there and point to an ASCII string with a URL.
if spc_opus_info['moreInfo']:
more_info = spc_opus_info['moreInfo']
if more_info['url']:
more_info_link = str(more_info['url'])
else:
raise Asn1Error('Expected a URL in moreInfo.')
else:
more_info_link = None
return program_name, more_info_link
def _ParseCountersig(self, unauth_attrs):
attr = unauth_attrs[0]
if oids.OID_TO_CLASS.get(attr['type']) is not pkcs7.CountersignInfo:
raise Asn1Error('Unexpected countersign OID.')
values = attr['values']
if len(values) != 1:
raise Asn1Error('Expected one CS value, got %d.' % len(values))
counter_sig_info, rest = decoder.decode(values[0],
asn1Spec=pkcs7.CountersignInfo())
if rest: raise Asn1Error('Extra unparsed content.')
return counter_sig_info
def _ParseAuthAttrs(self, auth_attrs, required):
results = dict.fromkeys(required)
for attr in auth_attrs:
if (attr['type'] in oids.OID_TO_CLASS and
oids.OID_TO_CLASS.get(attr['type']) in required):
# There are more than those I require, but I don't know what they are,
# and what to do with them. The spec does not talk about them.
# One example:
# 1.3.6.1.4.1.311.2.1.11 contains as value 1.3.6.1.4.1.311.2.1.21
# SPC_STATEMENT_TYPE_OBJID SPC_INDIVIDUAL_SP_KEY_PURPOSE_OBJID
results[oids.OID_TO_CLASS.get(attr['type'])] = attr['values']
if None in results.itervalues():
raise Asn1Error('Missing mandatory field(s) in auth_attrs.')
# making sure that the auth_attrs were processed in correct order
# they need to be sorted in ascending order in the SET, when DER encoded
# This also makes sure that the tag on Attributes is correct.
a = [der_encoder.encode(i) for i in auth_attrs]
a.sort()
attrs_for_hash = pkcs7.Attributes()
for i in range(len(auth_attrs)):
d, _ = decoder.decode(a[i], asn1Spec=pkcs7.Attribute())
attrs_for_hash.setComponentByPosition(i, d)
encoded_attrs = der_encoder.encode(attrs_for_hash)
return results, encoded_attrs
def ValidateHashes(self, computed_content_hash):
"""Compares computed against expected hashes.
This method makes sure the chain of hashes is correct. The chain
consists of Authenticode hash of the actual binary payload, as checked
against the hash in SpcInfo to the hash of SpcInfo as stored in the
AuthAttrs, and the hash of EncryptedDigest as stored in the counter-
signature AuthAttrs, if present.
Args:
computed_content_hash: Authenticode hash of binary, as provided by
fingerprinter.
Raises:
Asn1Error: if hash validation fails.
"""
if computed_content_hash != self.spc_info['messageDigest']['digest']:
raise Asn1Error('1: Validation of content hash failed.')
spc_blob = self.signed_data['contentInfo']['content']
# According to RFC2315, 9.3, identifier (tag) and length need to be
# stripped for hashing. We do this by having the parser just strip
# out the SEQUENCE part of the spcIndirectData.
# Alternatively this could be done by re-encoding and concatenating
# the individual elements in spc_value, I _think_.
_, hashable_spc_blob = decoder.decode(spc_blob, recursiveFlag=0)
spc_blob_hash = self.digest_algorithm(str(hashable_spc_blob)).digest()
if spc_blob_hash != self.expected_spc_info_hash:
raise Asn1Error('2: Validation of SpcInfo hash failed.')
# Can't check authAttr hash against encrypted hash, done implicitly in
# M2's pubkey.verify. This can be added by explicit decryption of
# encryptedDigest, if really needed. (See sample code for RSA in
# 'verbose_authenticode_sig.py')
if self.has_countersignature:
# Validates the hash value found in the authenticated attributes of the
# counter signature against the hash of the outer signature.
auth_attr_hash = self.digest_algorithm(self.encrypted_digest).digest()
if auth_attr_hash != self.expected_auth_attrs_hash:
raise Asn1Error('3: Validation of countersignature hash failed.')
def OidFromAttid(prefixTable, attr):
# separate the ATTRTYP into two parts
upperWord = attr / 65536
lowerWord = attr % 65536
# search in the prefix table to find the upperWord, if found,
# construct the binary OID by appending lowerWord to the end of
# found prefix.
binaryOID = None
for j, item in enumerate(prefixTable):
if item['ndx'] == upperWord:
binaryOID = item['prefix']['elements'][:item['prefix']['length']]
if lowerWord < 128:
binaryOID.append(chr(lowerWord))
else:
if lowerWord >= 32768:
lowerWord -= 32768
binaryOID.append(chr(((lowerWord/128) % 128)+128))
binaryOID.append(chr(lowerWord%128))
break
if binaryOID is None:
return None
return str(decoder.decode('\x06' + chr(len(binaryOID)) + ''.join(binaryOID), asn1Spec = univ.ObjectIdentifier())[0])
def decodeControlValue(self):
decodedControlValue, _ = decoder.decode(self['controlValue'], asn1Spec=SimplePagedResultsControlValue())
self._size, self._cookie = decodedControlValue[0], decodedControlValue[1]
return decodedControlValue
def recv(self):
REQUEST_SIZE = 8192
data = ''
done = False
while not done:
recvData = self._socket.recv(REQUEST_SIZE)
if len(recvData) < REQUEST_SIZE:
done = True
data += recvData
response = []
while len(data) > 0:
try:
message, remaining = decoder.decode(data, asn1Spec=LDAPMessage())
except SubstrateUnderrunError:
# We need more data
remaining = data + self._socket.recv(REQUEST_SIZE)
else:
if message['messageID'] == 0: # unsolicited notification
name = message['protocolOp']['extendedResp']['responseName'] or message['responseName']
notification = KNOWN_NOTIFICATIONS.get(name, "Unsolicited Notification '%s'" % name)
if name == NOTIFICATION_DISCONNECT: # Server has disconnected
self.close()
raise LDAPSessionError(
error=int(message['protocolOp']['extendedResp']['resultCode']),
errorString='%s -> %s: %s' % (notification,
message['protocolOp']['extendedResp']['resultCode'].prettyPrint(),
message['protocolOp']['extendedResp']['diagnosticMessage'])
)
response.append(message)
data = remaining
self._messageId += 1
return response
def TraverseRdn(rdn):
"""Traverses RDN structure and returns string encoding of the DN.
Args:
rdn: ASN.1 SET (or SEQUENCE) containing RDNs (relative distinguished
names), as identified by type / value pairs. A typical input would
be of type X.509 RelativeDistinguishedName.
Returns:
A dict representing the Distinguished Name.
"""
val = dict()
for n in rdn:
# Note that this does not work for e.g. DC which is present
# multiple times.
# For a real DN parser, make sure to follow the spec in regards
# to multiple occurence of a field in subsequent RDNs, maintaining
# original ordering etc.
# TODO(user): What about elements other than [0]??
name = DistinguishedName.OidToName(n[0]['type'])
value = decoder.decode(n[0]['value'])
if name in val:
val[name] = str(value[0]) + ', ' + val.get(name, '')
else:
val[name] = str(value[0])
return val
def _ParseTimestamp(self, time_asn1):
# Parses countersignature timestamp according to RFC3280, section 4.1.2.5+
timestamp_choice, rest = decoder.decode(time_asn1,
asn1Spec=pkcs7.SigningTime())
if rest: raise Asn1Error('Extra unparsed content.')
return timestamp_choice.ToPythonEpochTime()
def _ParseOpusInfo(self, opus_info_asn1):
spc_opus_info, rest = decoder.decode(opus_info_asn1,
asn1Spec=spc.SpcSpOpusInfo())
if rest: raise Asn1Error('Extra unparsed content.')
if spc_opus_info['programName']:
# According to spec, this should always be a Unicode string. However,
# the ASN.1 syntax allows both ASCII and Unicode. So, let's be careful.
opus_prog_name = spc_opus_info['programName']
uni_name = opus_prog_name['unicode']
ascii_name = opus_prog_name['ascii']
if ascii_name and uni_name:
# WTF? This is supposed to be a CHOICE
raise Asn1Error('Both elements of a choice are present.')
elif uni_name:
program_name = str(uni_name).decode('utf-16-be')
elif ascii_name:
program_name = str(ascii_name)
else:
raise Asn1Error('No element of opusInfo choice is present.')
else:
# According to spec, there should always be a program name,
# and be it zero-length. But let's be gentle, since ASN.1 marks
# this field als optional.
program_name = None
# Again, according to Authenticode spec, the moreInfo field should always
# be there and point to an ASCII string with a URL.
if spc_opus_info['moreInfo']:
more_info = spc_opus_info['moreInfo']
if more_info['url']:
more_info_link = str(more_info['url'])
else:
raise Asn1Error('Expected a URL in moreInfo.')
else:
more_info_link = None
return program_name, more_info_link
def _ParseCountersig(self, unauth_attrs):
attr = unauth_attrs[0]
if oids.OID_TO_CLASS.get(attr['type']) is not pkcs7.CountersignInfo:
raise Asn1Error('Unexpected countersign OID.')
values = attr['values']
if len(values) != 1:
raise Asn1Error('Expected one CS value, got %d.' % len(values))
counter_sig_info, rest = decoder.decode(values[0],
asn1Spec=pkcs7.CountersignInfo())
if rest: raise Asn1Error('Extra unparsed content.')
return counter_sig_info
def _ParseAuthAttrs(self, auth_attrs, required):
results = dict.fromkeys(required)
for attr in auth_attrs:
if (attr['type'] in oids.OID_TO_CLASS and
oids.OID_TO_CLASS.get(attr['type']) in required):
# There are more than those I require, but I don't know what they are,
# and what to do with them. The spec does not talk about them.
# One example:
# 1.3.6.1.4.1.311.2.1.11 contains as value 1.3.6.1.4.1.311.2.1.21
# SPC_STATEMENT_TYPE_OBJID SPC_INDIVIDUAL_SP_KEY_PURPOSE_OBJID
results[oids.OID_TO_CLASS.get(attr['type'])] = attr['values']
if None in results.itervalues():
raise Asn1Error('Missing mandatory field(s) in auth_attrs.')
# making sure that the auth_attrs were processed in correct order
# they need to be sorted in ascending order in the SET, when DER encoded
# This also makes sure that the tag on Attributes is correct.
a = [der_encoder.encode(i) for i in auth_attrs]
a.sort()
attrs_for_hash = pkcs7.Attributes()
for i in range(len(auth_attrs)):
d, _ = decoder.decode(a[i], asn1Spec=pkcs7.Attribute())
attrs_for_hash.setComponentByPosition(i, d)
encoded_attrs = der_encoder.encode(attrs_for_hash)
return results, encoded_attrs
def ValidateHashes(self, computed_content_hash):
"""Compares computed against expected hashes.
This method makes sure the chain of hashes is correct. The chain
consists of Authenticode hash of the actual binary payload, as checked
against the hash in SpcInfo to the hash of SpcInfo as stored in the
AuthAttrs, and the hash of EncryptedDigest as stored in the counter-
signature AuthAttrs, if present.
Args:
computed_content_hash: Authenticode hash of binary, as provided by
fingerprinter.
Raises:
Asn1Error: if hash validation fails.
"""
if computed_content_hash != self.spc_info['messageDigest']['digest']:
raise Asn1Error('1: Validation of content hash failed.')
spc_blob = self.signed_data['contentInfo']['content']
# According to RFC2315, 9.3, identifier (tag) and length need to be
# stripped for hashing. We do this by having the parser just strip
# out the SEQUENCE part of the spcIndirectData.
# Alternatively this could be done by re-encoding and concatenating
# the individual elements in spc_value, I _think_.
_, hashable_spc_blob = decoder.decode(spc_blob, recursiveFlag=0)
spc_blob_hash = self.digest_algorithm(str(hashable_spc_blob)).digest()
if spc_blob_hash != self.expected_spc_info_hash:
raise Asn1Error('2: Validation of SpcInfo hash failed.')
# Can't check authAttr hash against encrypted hash, done implicitly in
# M2's pubkey.verify. This can be added by explicit decryption of
# encryptedDigest, if really needed. (See sample code for RSA in
# 'verbose_authenticode_sig.py')
if self.has_countersignature:
# Validates the hash value found in the authenticated attributes of the
# counter signature against the hash of the outer signature.
auth_attr_hash = self.digest_algorithm(self.encrypted_digest).digest()
if auth_attr_hash != self.expected_auth_attrs_hash:
raise Asn1Error('3: Validation of countersignature hash failed.')
def toString (byte_string, field_name):
"""
Reformat the string
"""
result_string = ''
try:
result_string = byte_string.decode ('utf-8')
except AttributeError:
print ("The '" + field_name + "' field is not a UTF-8 string: '" + str (byte_string)+ "'")
return result_string
def _get_private_key(private_key_pkcs8_text):
"""Get an RSA private key object from a pkcs8 representation."""
der = rsa.pem.load_pem(private_key_pkcs8_text, 'PRIVATE KEY')
asn1_private_key, _ = decoder.decode(der, asn1Spec=PrivateKeyInfo())
return rsa.PrivateKey.load_pkcs1(
asn1_private_key.getComponentByName('privateKey').asOctets(),
format='DER')
