我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用Crypto.Signature.PKCS1_v1_5.new()。
def verify(self, message, signature): """Verifies a message against a signature. Args: message: string, The message to verify. signature: string, The signature on the message. Returns: True if message was signed by the private key associated with the public key that this object was constructed with. """ try: return PKCS1_v1_5.new(self._pubkey).verify( SHA256.new(message), signature) except: return False
def verifySignature(self): key = RSA.importKey(open(self.raven_public_key).read()) # Compile the parts to hash together parts = self.rav_str.split("!") parts.pop() # Remove the last two items related to signing parts.pop() to_hash = "!".join(parts) # Now hash it and verify our_hash = SHA.new(to_hash) #print our_hash verifier = PKCS1_v1_5.new(key) # Obtain the correct form of the signature signature = urllib.unquote(self.raven_signature) signature = signature.replace("-","+") signature = signature.replace(".","/") signature = signature.replace("_","=") signature = base64.b64decode(signature) if verifier.verify(our_hash, signature): return True else: return False
def testSign1(self): for i in range(len(self._testData)): row = self._testData[i] # Build the key if isStr(row[0]): key = RSA.importKey(row[0]) else: comps = [ long(rws(row[0][x]),16) for x in ('n','e','d') ] key = RSA.construct(comps) h = row[3].new() # Data to sign can either be in hex form or not try: h.update(t2b(row[1])) except: h.update(b(row[1])) # The real test signer = PKCS.new(key) self.failUnless(signer.can_sign()) s = signer.sign(h) self.assertEqual(s, t2b(row[2]))
def testVerify1(self): for i in range(len(self._testData)): row = self._testData[i] # Build the key if isStr(row[0]): key = RSA.importKey(row[0]).publickey() else: comps = [ long(rws(row[0][x]),16) for x in ('n','e') ] key = RSA.construct(comps) h = row[3].new() # Data to sign can either be in hex form or not try: h.update(t2b(row[1])) except: h.update(b(row[1])) # The real test verifier = PKCS.new(key) self.failIf(verifier.can_sign()) result = verifier.verify(h, t2b(row[2])) self.failUnless(result)
def _build_auth_token_data( self, auth_token_ticket, authenticator, private_key, **kwargs ): auth_token = dict( authenticator=authenticator, ticket=auth_token_ticket, **kwargs ) auth_token = json.dumps(auth_token, sort_keys=True) if six.PY3: auth_token = auth_token.encode('utf-8') digest = SHA256.new() digest.update(auth_token) auth_token = base64.b64encode(auth_token) rsa_key = RSA.importKey(private_key) signer = PKCS1_v1_5.new(rsa_key) auth_token_signature = signer.sign(digest) auth_token_signature = base64.b64encode(auth_token_signature) return auth_token, auth_token_signature
def submit(self, nonce): command = { 'command': 'submission', 'args': { 'nonce': nonce, 'wallet_id': self.wallet_id}} message = json.dumps(command) await self.socket.send(message) message = await self.socket.recv() response = json.loads(message) print("Submission result: {0}".format(response)) if 'error' in response: print("Error during submission : {0}".format(response["error"])) if 'challenge_name' in response: # we got a new challenge challenge = Challenge() challenge.fill_from_challenge(response) return challenge return None
def sign(message, priv_key, hashAlg="SHA-256"): global hash hash = hashAlg signer = PKCS1_v1_5.new(priv_key) if (hash == "SHA-512"): digest = SHA512.new() elif (hash == "SHA-384"): digest = SHA384.new() elif (hash == "SHA-256"): digest = SHA256.new() elif (hash == "SHA-1"): digest = SHA.new() else: digest = MD5.new() digest.update(message) return signer.sign(digest)
def testSign1(self): for i in range(len(self._testData)): row = self._testData[i] # Build the key if isStr(row[0]): key = RSA.importKey(row[0]) else: comps = [ int(rws(row[0][x]),16) for x in ('n','e','d') ] key = RSA.construct(comps) h = row[3].new() # Data to sign can either be in hex form or not try: h.update(t2b(row[1])) except: h.update(b(row[1])) # The real test signer = PKCS.new(key) self.failUnless(signer.can_sign()) s = signer.sign(h) self.assertEqual(s, t2b(row[2]))
def testVerify1(self): for i in range(len(self._testData)): row = self._testData[i] # Build the key if isStr(row[0]): key = RSA.importKey(row[0]).publickey() else: comps = [ int(rws(row[0][x]),16) for x in ('n','e') ] key = RSA.construct(comps) h = row[3].new() # Data to sign can either be in hex form or not try: h.update(t2b(row[1])) except: h.update(b(row[1])) # The real test verifier = PKCS.new(key) self.failIf(verifier.can_sign()) result = verifier.verify(h, t2b(row[2])) self.failUnless(result)
def verify_signature(self, data_dict): def _verify_signature(message, signature, sign_type): # Only 'RSA' Type Supported if sign_type.upper() == 'RSA': h = SHA.new(message.encode('utf-8')) verifier = PKCS1_v1_5.new(RSA.importKey(base64.decodestring(self.alipay_verification_public_key))) return verifier.verify(h, base64.decodestring(signature.encode('utf-8'))) else: raise Exception(Template('Sign type $sign_type not supported').substitute(sign_type=sign_type)) _sign = data_dict.get('sign', '') _sign_type = data_dict.get('sign_type', '') # Filter out the sign, sign_type field, and join other key value in format k1=v1&k2=v2 (keys in order) _string_to_verify = "&".join(['{}={}'.format(key, str(value)) for (key, value) in sorted(data_dict.items()) if key not in {'sign', 'sign_type'}]) return _verify_signature(message=_string_to_verify, signature=_sign, sign_type=_sign_type)
def encrypt(self, data, pubkey): # all or nothing data transformPacket = AONTencrypt(data) # message key messageKey = random_aes_32_key() if not isinstance(pubkey, RSA.RsaKey): key = RSA.importKey(pubkey) else: key = pubkey skey = PKCS1_OAEP.new(key) frontMatter = skey.encrypt(messageKey) # , 16) # print('RSA calls AES Encrypt with private key: %s' % key.publickey().exportKey()) # encrypted message backMatter = default_aes(messageKey).encrypt(transformPacket) # put together message = frontMatter + backMatter return message
def testSign1(self): for i in range(len(self._testData)): row = self._testData[i] # Build the key if isStr(row[0]): key = RSA.importKey(row[0]) else: comps = [ int(rws(row[0][x]),16) for x in ('n','e','d') ] key = RSA.construct(comps) h = row[3].new() # Data to sign can either be in hex form or not try: h.update(t2b(row[1])) except: h.update(b(row[1])) # The real test signer = PKCS.new(key) self.assertTrue(signer.can_sign()) s = signer.sign(h) self.assertEqual(s, t2b(row[2]))
def testVerify1(self): for i in range(len(self._testData)): row = self._testData[i] # Build the key if isStr(row[0]): key = RSA.importKey(row[0]).publickey() else: comps = [ int(rws(row[0][x]),16) for x in ('n','e') ] key = RSA.construct(comps) h = row[3].new() # Data to sign can either be in hex form or not try: h.update(t2b(row[1])) except: h.update(b(row[1])) # The real test verifier = PKCS.new(key) self.assertFalse(verifier.can_sign()) result = verifier.verify(h, t2b(row[2])) self.assertTrue(result)
def submit(self, nonce): command = {'command': 'submission', 'args': {'nonce': nonce, 'wallet_id': self.wallet_id}} message = json.dumps(command) await self.socket.send(message) message = await self.socket.recv() response = json.loads(message) print("Submission result: {0}".format(response)) if 'error' in response: print("Error during submission : {0}".format(response["error"])) if 'challenge_name' in response: # we got a new challenge challenge = Challenge() challenge.fill_from_challenge(response) return challenge return None
def create_transaction(self, recipient, amount): hasher = SHA256.new() hasher.update("{0},{1},{2:.5f}".format(self.wallet_id, recipient, amount)) signature = self.sign_message(hasher) command = {'command': 'create_transaction', 'args': {'source': self.wallet_id, 'recipient': recipient, 'amount': "{0:.5f}".format(amount), signature: signature}} message = json.dumps(command) await self.socket.send(message) message = await self.socket.recv() response = json.loads(message) if 'error' in response: print("Error during create_transaction : {0}".format(response['error'])) if 'id' in response: return response['id'] return None
def verify_signature(data, signature, x509_certificate): """Verifies a signature using the given x.509 public key certificate.""" # PyCrypto 2.6 doesn't support x.509 certificates directly, so we'll need # to extract the public key from it manually. # This code is based on https://github.com/google/oauth2client/blob/master # /oauth2client/_pycrypto_crypt.py pem_lines = x509_certificate.replace(b' ', b'').split() cert_der = base64.urlsafe_b64decode(b''.join(pem_lines[1:-1])) cert_seq = DerSequence() cert_seq.decode(cert_der) tbs_seq = DerSequence() tbs_seq.decode(cert_seq[0]) public_key = RSA.importKey(tbs_seq[6]) signer = PKCS1_v1_5.new(public_key) digest = SHA256.new(data) return signer.verify(digest, signature)
def sign_dict(dict_, private_key): """Serializes a dict to JSON and sign it using RSA.""" try: with open(private_key) as fp: key = RSA.importKey(fp.read()) except (FileNotFoundError, ValueError, IndexError, TypeError): raise ValueError('Invalid private key file.') signer = PKCS1_v1_5.new(key) # encodes message message = SHA256.new(json.dumps(dict_, sort_keys=True).encode()) # sign signature = signer.sign(message) return base64.b64encode(signature).decode()
def verify_archive(self, dest): self.assertTrue(os.path.exists(dest)) self.assertTrue(zipfile.is_zipfile(dest)) with zipfile.ZipFile(dest) as archive: files = archive.namelist() member = archive.extract(self.obj_sha256) sig_fn = archive.extract('signature') metadata_fn = archive.extract('metadata') self.addCleanup(os.remove, member) self.addCleanup(os.remove, sig_fn) self.addCleanup(os.remove, metadata_fn) self.assertEqual(len(files), 3) self.assertIn(self.obj_sha256, files) self.assertFalse(os.path.islink(member)) with open(metadata_fn) as fp: message = SHA256.new(fp.read().encode()) with open(sig_fn) as fp: signature = base64.b64decode(fp.read()) verifier = PKCS1_v1_5.new(self.private_key) self.assertTrue(verifier.verify(message, signature))
def test_can_sign_dict(self): _, fn = tempfile.mkstemp() self.addCleanup(os.remove, fn) key = RSA.generate(1024) with open(fn, 'wb') as fp: fp.write(key.exportKey()) dict_ = {} message = SHA256.new(json.dumps(dict_).encode()) result = utils.sign_dict(dict_, fn) signature = base64.b64decode(result) verifier = PKCS1_v1_5.new(key) is_valid = verifier.verify(message, signature) self.assertTrue(is_valid)
