certificate_authorization.py 文件源码

def verify_peer_token(self, peer_token, peer, peer_type):
        token_time = peer_token[:16]
        token_sign = peer_token[16:]
        current_date = int(datetime.datetime.now().timestamp() * 1000)
        token_date = int(token_time, 16)
        if current_date > token_date:
            return False

        date = bytes.fromhex(token_time)

        peer_info = b''.join([peer.peer_id.encode('utf-8'),
                              peer.target.encode('utf-8'),
                              peer.group_id.encode('utf-8')]) + bytes([peer_type])

        peer_cert = x509.load_der_x509_certificate(bytes.fromhex(peer.cert), default_backend())
        peer_pub = peer_cert.public_key().public_bytes(encoding=serialization.Encoding.DER,
                                                       format=PublicFormat.SubjectPublicKeyInfo)

        token_bytes = peer_info + date + peer_pub
        logging.debug("TBS Token(V) : %s", token_bytes.hex())
        signature = bytes.fromhex(token_sign)

        return self.verify_data_with_cert(cert=self.__ca_cert, data=token_bytes, signature=signature)