我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用pyasn1.codec.ber.decoder.decode()。
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
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 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 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 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')
def cbRecvFun(transportDispatcher, transportDomain, transportAddress, wholeMsg, reqPDU=reqPDU, headVars=headVars): 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 and errorStatus != 2: raise Exception(errorStatus) # Format var-binds table varBindTable = pMod.apiPDU.getVarBindTable(reqPDU, rspPDU) # Report SNMP table for tableRow in varBindTable: for name, val in tableRow: print('from: %s, %s = %s' % ( transportAddress, name.prettyPrint(), val.prettyPrint() ) ) # Stop on EOM for oid, val in varBindTable[-1]: if not isinstance(val, pMod.Null): break else: transportDispatcher.jobFinished(1) # Generate request for next row pMod.apiPDU.setVarBinds( reqPDU, [(x, pMod.null) for x, y in varBindTable[-1]] ) pMod.apiPDU.setRequestID(reqPDU, pMod.getNextRequestID()) transportDispatcher.sendMessage( encoder.encode(reqMsg), transportDomain, transportAddress ) global startedAt if time() - startedAt > 3: raise Exception('Request timed out') startedAt = time() return wholeMsg
def cbFun(transportDispatcher, transportDomain, transportAddress, wholeMsg): while wholeMsg: msgVer = int(api.decodeMessageVersion(wholeMsg)) if msgVer in api.protoModules: pMod = api.protoModules[msgVer] else: print('Unsupported SNMP version %s' % msgVer) return reqMsg, wholeMsg = decoder.decode( wholeMsg, asn1Spec=pMod.Message(), ) print('Notification message from %s:%s: ' % ( transportDomain, transportAddress ) ) reqPDU = pMod.apiMessage.getPDU(reqMsg) if reqPDU.isSameTypeWith(pMod.TrapPDU()): if msgVer == api.protoVersion1: print('Enterprise: %s' % (pMod.apiTrapPDU.getEnterprise(reqPDU).prettyPrint())) print('Agent Address: %s' % (pMod.apiTrapPDU.getAgentAddr(reqPDU).prettyPrint())) print('Generic Trap: %s' % (pMod.apiTrapPDU.getGenericTrap(reqPDU).prettyPrint())) print('Specific Trap: %s' % (pMod.apiTrapPDU.getSpecificTrap(reqPDU).prettyPrint())) print('Uptime: %s' % (pMod.apiTrapPDU.getTimeStamp(reqPDU).prettyPrint())) varBinds = pMod.apiTrapPDU.getVarBinds(reqPDU) else: varBinds = pMod.apiPDU.getVarBinds(reqPDU) print('Var-binds:') for oid, val in varBinds: print('%s = %s' % (oid.prettyPrint(), val.prettyPrint())) return wholeMsg
def decodeMessageVersion(wholeMsg): try: seq, wholeMsg = decoder.decode( wholeMsg, asn1Spec=univ.Sequence(), recursiveFlag=False, substrateFun=lambda a, b, c: (a, b[:c]) ) ver, wholeMsg = decoder.decode( wholeMsg, asn1Spec=univ.Integer(), recursiveFlag=False, substrateFun=lambda a, b, c: (a, b[:c]) ) if eoo.endOfOctets.isSameTypeWith(ver): raise ProtocolError('EOO at SNMP version component') return ver except PyAsn1Error: raise ProtocolError('Invalid BER at SNMP version component')
def main(argv): try: infile = argv[0] except: print "usage: ./krb5-downgrade-asreq.py <infile>" sys.exit(0) fin = open(infile, 'r') data = fin.read() data_len = len(data) fin.close() krb_preauth_req, temp = decoder.decode(data[4:]) for i in range(0, len(krb_preauth_req[3][7])): krb_preauth_req[3][7][i] = univ.Integer(1) payload_out = data[:4] payload_out += encoder.encode(krb_preauth_req) # log what we're doing fout = open(infile +".in", "w") fout.write(data) fout.close() fout = open(infile +".out", "w") fout.write(payload_out) fout.close() sys.stdout.write(payload_out) os.remove(infile)
def _get_private_key(private_key_pkcs8_text): """Get an RSA private key object from a pkcs8 representation.""" private_key_pkcs8_text = _to_bytes(private_key_pkcs8_text) 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')
def decodeMessageVersion(wholeMsg): try: seq, wholeMsg = decoder.decode( wholeMsg, asn1Spec=univ.Sequence(), recursiveFlag=0 ) ver, wholeMsg = decoder.decode( wholeMsg, asn1Spec=univ.Integer(), recursiveFlag=0 ) if eoo.endOfOctets.isSameTypeWith(ver): raise ProtocolError('EOO at SNMP version component') return ver except PyAsn1Error: raise ProtocolError('Invalid BER at SNMP version component')
def decodeControlValue(self,encodedControlValue): decodedValue,_ = decoder.decode(encodedControlValue,asn1Spec=DerefResultControlValue()) self.derefRes = {} for deref_res in decodedValue: deref_attr,deref_val,deref_vals = deref_res partial_attrs_dict = dict([ (str(t),map(str,v)) for t,v in deref_vals or [] ]) try: self.derefRes[str(deref_attr)].append((str(deref_val),partial_attrs_dict)) except KeyError: self.derefRes[str(deref_attr)] = [(str(deref_val),partial_attrs_dict)]
def decodeControlValue(self,encodedControlValue): decodedEntry,_ = decoder.decode(encodedControlValue,asn1Spec=SearchResultEntry()) self.dn = str(decodedEntry[0]) self.entry = {} for attr in decodedEntry[1]: self.entry[str(attr[0])] = [ str(attr_value) for attr_value in attr[1] ]
def decodeControlValue(self, encodedControlValue): d = decoder.decode(encodedControlValue, asn1Spec = syncStateValue()) state = d[0].getComponentByName('state') uuid = UUID(bytes=d[0].getComponentByName('entryUUID')) self.cookie = d[0].getComponentByName('cookie') self.state = self.__class__.opnames[int(state)] self.entryUUID = str(uuid) if self.cookie is not None: self.cookie = str(self.cookie)
def decodeControlValue(self, encodedControlValue): d = decoder.decode(encodedControlValue, asn1Spec = syncDoneValue()) self.cookie = d[0].getComponentByName('cookie') self.refreshDeletes = d[0].getComponentByName('refreshDeletes') if self.cookie is not None: self.cookie = str(self.cookie) if self.refreshDeletes is not None: self.refreshDeletes = bool(self.refreshDeletes)
def __init__(self, encodedMessage): d = decoder.decode(encodedMessage, asn1Spec = syncInfoValue()) self.newcookie = None self.refreshDelete = None self.refreshPresent = None self.syncIdSet = None for attr in [ 'newcookie', 'refreshDelete', 'refreshPresent', 'syncIdSet']: comp = d[0].getComponentByName(attr) if comp is not None: if attr == 'newcookie': self.newcookie = str(comp) return val = dict() cookie = comp.getComponentByName('cookie') if cookie is not None: val['cookie'] = str(cookie) if attr.startswith('refresh'): val['refreshDone'] = bool(comp.getComponentByName('refreshDone')) elif attr == 'syncIdSet': uuids = [] ids = comp.getComponentByName('syncUUIDs') for i in range(len(ids)): uuid = UUID(bytes=str(ids.getComponentByPosition(i))) uuids.append(str(uuid)) val['syncUUIDs'] = uuids val['refreshDeletes'] = bool(comp.getComponentByName('refreshDeletes')) setattr(self,attr,val) return
def prune_not_ca(self): from pyasn1.codec.ber import decoder as d for i in list(reversed(range(len(self.certs)))): for e in range(self.certs[i].get_extension_count()): if self.certs[i].get_extension(e).get_short_name() == 'basicConstraints': data = d.decode(self.certs[i].get_extension(e).get_data())[0] ca = False if data: ca = data.getComponentByPosition(0) if not ca: self.certs.pop(i) return True return False
def get_server_cert(self): from pyasn1.codec.ber import decoder as d for i in range(len(self.certs)): for e in range(self.certs[i].get_extension_count()): if self.certs[i].get_extension(e).get_short_name() == 'basicConstraints': data = d.decode(self.certs[i].get_extension(e).get_data())[0] ca = False if data: ca = data.getComponentByPosition(0) if not ca: return self.certs[i]
