Python rsa 模块，verify() 实例源码

def decryptAndVerify(self, ciphertextPickle, senderPublic):
        try:
            if (ciphertextPickle):
                decoded = pickle.loads(ciphertextPickle)
                if (decoded):
                    signature = decoded[0]
                    content = decoded[1]
                    if (rsa.verify(content, signature, senderPublic)):
                        debug("crypt", "Message signature was verified. Decrypting..")
                        return rsa.decrypt(content, self.DRMPrivateKey)
            else:
                debug("crypt", "Message was empty. \"No Updates\"")
        except Exception as e:
            debug("crypt", "Error decrypting message. -> %s" % str(e))

    # Load the keys

def perform_operation(self, indata, pub_key, cli_args):
        '''Decrypts files.'''

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def test_sign(self):
        """Test sign works."""
        # rsa_pk = M2Crypto.RSA.gen_key(2048, 65537)
        rsa_keys = rsa.newkeys(2048, 65537)
        rsa_pk = rsa_keys[1]
        rsa_pub = rsa_keys[0]
        salt = 'salt'
        data = {"flags": 8,
                "name": "MyAwesomeVM",
                "ram": 512,
                "secret": "mg041na39123",
                "userData": "[amiconfig]\nplugins=cernvm\n[cernvm]\nusers=user:users;password",
                "vcpus": 1,
                "version": "1.5"}
        strBuffer = vmcp.calculate_buffer(data, salt)

        with patch('rsa.PrivateKey.load_pkcs1', return_value=rsa_pk):
            with patch('pybossa.vmcp.open', mock_open(read_data=''), create=True) as m:
                out = vmcp.sign(data, salt, 'testkey')
                err_msg = "There should be a key named signature"
                assert out.get('signature'), err_msg

                err_msg = "The signature should not be empty"
                assert out['signature'] is not None, err_msg
                assert out['signature'] != '', err_msg

                err_msg = "The signature should be the same"
                signature = base64.b64decode(out['signature'])
                assert rsa.verify(strBuffer, signature, rsa_pub) == 1, err_msg

                # The output must be convertible into json object
                import json
                assert_not_raises(Exception, json.dumps, out)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        '''Decrypts files.'''

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def validate(self, bucket, key, public_key, digest_data, inflated_digest):
        """Validates a digest file.

        Throws a DigestError when the digest is invalid.

        :param bucket: Bucket of the digest file
        :param key: Key of the digest file
        :param public_key: Public key bytes.
        :param digest_data: Dict of digest data returned when JSON
            decoding a manifest.
        :param inflated_digest: Inflated digest file contents as bytes.
        """
        try:
            decoded_key = base64.b64decode(public_key)
            public_key = rsa.PublicKey.load_pkcs1(decoded_key, format='DER')
            to_sign = self._create_string_to_sign(digest_data, inflated_digest)
            signature_bytes = binascii.unhexlify(digest_data['_signature'])
            rsa.verify(to_sign, signature_bytes, public_key)
        except PyAsn1Error:
            raise DigestError(
                ('Digest file\ts3://%s/%s\tINVALID: Unable to load PKCS #1 key'
                 ' with fingerprint %s')
                % (bucket, key, digest_data['digestPublicKeyFingerprint']))
        except rsa.pkcs1.VerificationError:
            # Note from the Python-RSA docs: Never display the stack trace of
            # a rsa.pkcs1.VerificationError exception. It shows where in the
            # code the exception occurred, and thus leaks information about
            # the key.
            raise DigestSignatureError(bucket, key)

def setup_services(self, parsed_globals):
        self._source_region = parsed_globals.region
        # Use the the same region as the region of the CLI to get locations.
        self.s3_client_provider = S3ClientProvider(
            self._session, self._source_region)
        client_args = {'region_name': parsed_globals.region,
                       'verify': parsed_globals.verify_ssl}
        if parsed_globals.endpoint_url is not None:
            client_args['endpoint_url'] = parsed_globals.endpoint_url
        self.cloudtrail_client = self._session.create_client(
            'cloudtrail', **client_args)

def verify(self, message, signature):

        # assume we're dealing with a base64 encoded signature
        signature = b64decode(signature.encode())

        message = self.ensure_bytes(message)

        # ensure we have a public key we can use use to verify
        if not self._public_key:
            raise Exception("Key object does not have public key defined, and thus cannot be used to verify.")

        return self._verify(message, signature)

def _verify(self, message, signature):

        try:
            PYRSA.verify(message, signature, self._public_key)
            return True
        except PYRSA.pkcs1.VerificationError:
            return False

def _verify(self, message, signature):

        try:
            self._public_key.verify(hashlib.sha256(message).digest(), signature, algo="sha256")
            return True
        except M2RSA.RSAError:
            return False

def _verify(self, message, signature):
        try:
            self._public_key.verify(signature, message, crypto_padding.PKCS1v15(), crypto_hashes.SHA256())
            return True
        except crypto_exceptions.InvalidSignature:
            return False

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def verify(self):
        """ Verifies the signature of transaction
        """
        try:
            rsa.verify(self.message, self.signature, self.sender)
        except VerificationError:
            print("Verification failed", file=sys.stderr)
            return False

        return True

def verify(self):
        acc = ''
        for t in self.transactions:
            acc += str(t.hash)

        return int(sha256((str(self.nonce)+acc).encode('utf-8')).hexdigest()[0:6],16) < GAMER_BAWA

def perform_operation(self, indata, pub_key, cli_args):
        '''Decrypts files.'''

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def appendMessage(self, message):
        if (message):
            self.messages.append(message)
        return

# Keep hostlist, verify public keys, and invalidate hosts
# after a certain amount of time

def verifyTrustedSignature(self, data, signature):
        result = False
        try:
            rsa.verify(data, signature, self.DRMTrustedKey)
            result = True
        except Exception as e:
            debug("crypt", "Verifying the signature of the remote host's message. -> %s" % str(e))
        return result

    # Encrypt to recipient's public key, and sign with our private key

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        '''Decrypts files.'''

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        '''Decrypts files.'''

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def perform_operation(self, indata, pub_key, cli_args):
        """Verifies files."""

        signature_file = cli_args[1]

        with open(signature_file, 'rb') as sigfile:
            signature = sigfile.read()

        try:
            rsa.verify(indata, signature, pub_key)
        except rsa.VerificationError:
            raise SystemExit('Verification failed.')

        print('Verification OK', file=sys.stderr)

def main():
    if len(sys.argv) == 2 and sys.argv[1] == 'rsa_old':
        import_rsa(0)
    else:
        import_rsa(1)

    (sig, message) = forge()
    (priv, pub) = genkeys()
    verify(message, sig, priv)

def verify(message, sig, priv):
    try:
        rsa.verify(message, sig, priv)
        print("L'identite de l'expediteur a bien ete confirmee")
    except:
        print("Impossible de confirmer l'expediteur du message")

def verifySessionKey():
    if not "application/json" in request.headers["Content-Type"]:
        return None, None, make_response("Expected content-type JSON", 400)

    data = request.json
    for key in ("appid", "key", "_sig"):
        if not key in data:
            return make_response("Missing argument: {key}".format(key=key), 400)

    appid = str(data["appid"])
    if not "appversion" in data:
        appversion = "any"
    else:
        appversion = str(data["appversion"])
    key = str(data["key"])

    # calculate message that was signed
    message = "{appid}:{appversion}:{key}".format(**locals())

    # decode signature
    import base64
    signature = data["_sig"]
    signature = base64.decodestring("\n".join([signature[x:x+64] for x in range(0, len(signature), 64)]))

    # fetch and validate app information
    lookup_key = appid + ":" + appversion
    apps = _get_registered_apps()
    if not lookup_key in apps or not apps[lookup_key]["enabled"] or not "pubkey" in apps[lookup_key]:
        octoprint.server.appSessionManager.remove(key)
        return make_response("Invalid app: {lookup_key}".format(lookup_key=lookup_key), 401)

    pubkey_string = apps[lookup_key]["pubkey"]
    pubkey_string = "\n".join([pubkey_string[x:x+64] for x in range(0, len(pubkey_string), 64)])
    try:
        pubkey = rsa.PublicKey.load_pkcs1("-----BEGIN RSA PUBLIC KEY-----\n" + pubkey_string + "\n-----END RSA PUBLIC KEY-----\n")
    except:
        octoprint.server.appSessionManager.remove(key)
        return make_response("Invalid pubkey stored in server", 500)

    # verify signature
    try:
        rsa.verify(message, signature, pubkey)
    except rsa.VerificationError:
        octoprint.server.appSessionManager.remove(key)
        return make_response("Invalid signature", 401)

    # generate new session key and return it
    result = octoprint.server.appSessionManager.verify(key)
    if not result:
        return make_response("Invalid key or already verified", 401)

    verified_key, valid_until = result
    return jsonify(key=verified_key, validUntil=valid_until)