Return the number of Signed Certificate Timestamps (SCTs) embedded in the certificate.
422412
def count_scts_in_sct_extension(certificate: cryptography.x509.Certificate) -> Optional[int]:
        
        scts_count = 0
        try:
            # Look for the x509 extension
            sct_ext = certificate.extensions.get_extension_for_oid(
                ExtensionOID.PRECERT_SIGNED_CERTIFICATE_TIMESTAMPS
            )

            if isinstance(sct_ext.value, cryptography.x509.UnrecognizedExtension):
                # The version of OpenSSL on the system is too old and can't parse the SCT extension
                return None

            # Count the number of entries in the extension
            scts_count = len(sct_ext.value)
        except ExtensionNotFound:
            pass

        return scts_count
86509
def getKeySequenceCounter(self):
        """"""
        print '%s call getKeySequenceCounter' % self.port
        keySequence = ''
        keySequence = self.__sendCommand(WPANCTL_CMD + 'getprop -v Network:KeyIndex')[0]
        return keySequence
375734
def _ttl(self):
        """"""
        return self.hlim if isinstance(self, scapy.layers.inet6.IPv6) else self.ttl
186073
private X509Certificate []getClientCertificatesImpl()
    throws CertificateException
  {
    if (! (_s instanceof SSLSocket))
      return null;
    
    SSLSocket sslSocket = (SSLSocket) _s;

    SSLSession sslSession = sslSocket.getSession();
    if (sslSession == null)
      return null;

    try {
      return (X509Certificate []) sslSession.getPeerCertificates();
    } catch (SSLPeerUnverifiedException e) {
      if (log.isLoggable(Level.FINEST))
        log.log(Level.FINEST, e.toString(), e);
      
      return null;
    } catch (Throwable e) {
      log.log(Level.FINER, e.toString(), e);
    }
    
    return null;
  }
375964
def get_timestamps(cols, created_name, updated_name):
  """"""
  has_created = created_name in cols
  has_updated = updated_name in cols
  return (created_name if has_created else None, updated_name if has_updated else None)
75995
def _CollectSignedData(self, extent):
    """"""
    start, length = extent

    self.file.seek(start, os.SEEK_SET)
    buf = self.file.read(length)
    signed_data = []
    # This loop ignores trailing cruft, or too-short signedData chunks.
    while len(buf) >= 8:
      dw_length, w_revision, w_cert_type = struct.unpack('<IHH', buf[:8])
      if dw_length < 8:
        # If the entire blob is smaller than its header, bail out.
        return signed_data
      b_cert = buf[8:dw_length]
      buf = buf[(dw_length + 7) & 0x7ffffff8:]
      signed_data.append((w_revision, w_cert_type, b_cert))
    return signed_data
967
def getStats(self):
        
        stats = protocol.ReadStats()
        stats.aligned_read_count = self.getNumAlignedReads()
        stats.unaligned_read_count = self.getNumUnalignedReads()
        # TODO base_count requires iterating through all reads
        return stats
397740
function getScopeBasedDesignTime(mMetadata, sScopeKey) {
		var mResult = mMetadata;

		if ("default" in mMetadata) {
			mResult = merge(
				{},
				mMetadata.default,
				sScopeKey !== "default" && mMetadata[sScopeKey] || null
			);
		}

		return mResult;
	}
1225949
function() {
    return this.timestamp().substr( -2 ) +
      this.count().toString( this.base ).substr( -4 ) +
      this._fingerprint[0] +
      this._fingerprint[ this._fingerprint.length - 1 ] +
      this.randomBlock().substr( -2 )
  }
120420
def tracked_coordinator_count(self, count_ascps=False):
        '''  '''
        with self._lock:
            _count = 0
            if count_ascps:
                for _coordinator in self._tracked_transfer_coordinators:
                    _count += _coordinator.session_count
            else:
                _count = len(self._tracked_transfer_coordinators)
            return _count
984225
def validate_num_assertions(self):
        
        encrypted_assertion_nodes = OneLogin_Saml2_XML.query(self.document, '//saml:EncryptedAssertion')
        assertion_nodes = OneLogin_Saml2_XML.query(self.document, '//saml:Assertion')

        valid = len(encrypted_assertion_nodes) + len(assertion_nodes) == 1

        if (self.encrypted):
            assertion_nodes = OneLogin_Saml2_XML.query(self.decrypted_document, '//saml:Assertion')
            valid = valid and len(assertion_nodes) == 1

        return valid
31085
def _get_v4_signed_headers(self):
        """"""
        if self.aws_session is None:
            boto_session = session.Session()
            creds = boto_session.get_credentials()
        else:
            creds = self.aws_session.get_credentials()
        if creds is None:
            raise CerberusClientException("Unable to locate AWS credentials")
        readonly_credentials = creds.get_frozen_credentials()

        # hardcode get-caller-identity request
        data = OrderedDict((('Action','GetCallerIdentity'), ('Version', '2011-06-15')))
        url = 'https://sts.{}.amazonaws.com/'.format(self.region)
        request_object = awsrequest.AWSRequest(method='POST', url=url, data=data)

        signer = auth.SigV4Auth(readonly_credentials, 'sts', self.region)
        signer.add_auth(request_object)
        return request_object.headers
237135
def get_count(self, prefix=''):
        """"""
        return sum([self.counters[key]
                    for key in self.messages if key.startswith(prefix)])
635586
def getUserSignupDate(self):
        
        userinfo = self.getUserInfo()
        timestamp = int(float(userinfo["signupTimeSec"]))
        return time.strftime("%m/%d/%Y %H:%M", time.gmtime(timestamp))
9126
def get_num_names(include_expired=False, proxy=None, hostport=None):
    
    assert proxy or hostport, 'Need proxy or hostport'
    if proxy is None:
        proxy = connect_hostport(hostport)

    schema = {
        'type': 'object',
        'properties': {
            'count': {
                'type': 'integer',
                'minimum': 0,
            },
        },
        'required': [
            'count',
        ],
    }

    count_schema = json_response_schema(schema)

    resp = {}
    try:
        if include_expired:
            resp = proxy.get_num_names_cumulative()
        else:
            resp = proxy.get_num_names()

        resp = json_validate(count_schema, resp)
        if json_is_error(resp):
            return resp

    except ValidationError as e:
        if BLOCKSTACK_DEBUG:
            log.exception(e)

        resp = {'error': 'Server response did not match expected schema.  You are likely communicating with an out-of-date Blockstack node.', 'http_status': 502}
        return resp

    except socket.timeout:
        log.error("Connection timed out")
        resp = {'error': 'Connection to remote host timed out.', 'http_status': 503}
        return resp

    except socket.error as se:
        log.error("Connection error {}".format(se.errno))
        resp = {'error': 'Connection to remote host failed.', 'http_status': 502}
        return resp

    except Exception as ee:
        if BLOCKSTACK_DEBUG:
            log.exception(ee)

        log.error("Caught exception while connecting to Blockstack node: {}".format(ee))
        resp = {'error': 'Failed to contact Blockstack node.  Try again with `--debug`.', 'http_status': 500}
        return resp

    return resp['count']
919440
def check_signature(self, msgbuf, srcSystem, srcComponent):
            ''''''
            if isinstance(msgbuf, array.array):
                msgbuf = msgbuf.tostring()
            timestamp_buf = msgbuf[-12:-6]
            link_id = msgbuf[-13]
            (tlow, thigh) = struct.unpack('<IH', timestamp_buf)
            timestamp = tlow + (thigh<<32)

            # see if the timestamp is acceptable
            stream_key = (link_id,srcSystem,srcComponent)
            if stream_key in self.signing.stream_timestamps:
                if timestamp <= self.signing.stream_timestamps[stream_key]:
                    # reject old timestamp
                    # print('old timestamp')
                    return False
            else:
                # a new stream has appeared. Accept the timestamp if it is at most
                # one minute behind our current timestamp
                if timestamp + 6000*1000 < self.signing.timestamp:
                    # print('bad new stream ', timestamp/(100.0*1000*60*60*24*365), self.signing.timestamp/(100.0*1000*60*60*24*365))
                    return False
                self.signing.stream_timestamps[stream_key] = timestamp
                # print('new stream')

            h = hashlib.new('sha256')
            h.update(self.signing.secret_key)
            h.update(msgbuf[:-6])
            sig1 = str(h.digest())[:6]
            sig2 = str(msgbuf)[-6:]
            if sig1 != sig2:
                # print('sig mismatch')
                return False

            # the timestamp we next send with is the max of the received timestamp and
            # our current timestamp
            self.signing.timestamp = max(self.signing.timestamp, timestamp)
            return True
562081
def getTimestampUTC():
    
    utc     = datetime.datetime.utcnow()
    ts_sec  = calendar.timegm( utc.timetuple() )
    ts_usec = utc.microsecond
    return ts_sec, ts_usec