def init(self):
import DataStore, readconf, logging, sys
self.conf.update({ "debug": None, "logging": None })
self.conf.update(DataStore.CONFIG_DEFAULTS)
args, argv = readconf.parse_argv(self.argv, self.conf, strict=False)
if argv and argv[0] in ('-h', '--help'):
print self.usage()
return None, []
logging.basicConfig(
stream=sys.stdout, level=logging.DEBUG, format="%(message)s")
if args.logging is not None:
import logging.config as logging_config
logging_config.dictConfig(args.logging)
store = DataStore.new(args)
return store, argv
# Abstract hex-binary conversions for eventual porting to Python 3.
python类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 encrypt(key, filename):
chunksize = 64 * 1024
outFile = os.path.join(os.path.dirname(filename), ".hell"+os.path.basename(filename))
filesize = str(os.path.getsize(filename)).zfill(16)
IV = ''
for i in range(16):
IV += chr(random.randint(0, 0xFF))
encryptor = AES.new(key, AES.MODE_CBC, IV)
with open(filename, "rb") as infile:
with open(outFile, "wb") as outfile:
outfile.write(filesize)
outfile.write(IV)
while True:
chunk = infile.read(chunksize)
if len(chunk) == 0:
break
elif len(chunk) % 16 !=0:
chunk += ' ' * (16 - (len(chunk) % 16))
outfile.write(encryptor.encrypt(chunk))
def runTest(self):
"""SHA256: 512/520 MiB test"""
from Crypto.Hash import SHA256
zeros = bchr(0x00) * (1024*1024)
h = SHA256.new(zeros)
for i in xrange(511):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('9acca8e8c22201155389f65abbf6bc9723edc7384ead80503839f49dcc56d767', h.hexdigest()) # 512 MiB
for i in xrange(8):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('abf51ad954b246009dfe5a50ecd582fd5b8f1b8b27f30393853c3ef721e7fa6e', h.hexdigest()) # 520 MiB
def decrypt_secret(secret, key):
"""Python implementation of SystemFunction005.
Decrypts a block of data with DES using given key.
Note that key can be longer than 7 bytes."""
decrypted_data = ''
j = 0 # key index
for i in range(0,len(secret),8):
enc_block = secret[i:i+8]
block_key = key[j:j+7]
des_key = str_to_key(block_key)
des = DES.new(des_key, DES.MODE_ECB)
decrypted_data += des.decrypt(enc_block)
j += 7
if len(key[j:j+7]) < 7:
j = len(key[j:j+7])
(dec_data_len,) = unpack("<L", decrypted_data[:4])
return decrypted_data[8:8+dec_data_len]
def decrypt_aes(secret, key):
sha = SHA256.new()
sha.update(key)
for _i in range(1, 1000+1):
sha.update(secret[28:60])
aeskey = sha.digest()
data = ""
for i in range(60, len(secret), 16):
aes = AES.new(aeskey, AES.MODE_CBC, "\x00"*16)
buf = secret[i : i + 16]
if len(buf) < 16:
buf += (16-len(buf)) * "\00"
data += aes.decrypt(buf)
return data
def runTest(self):
"""SHA256: 512/520 MiB test"""
from Crypto.Hash import SHA256
zeros = bchr(0x00) * (1024*1024)
h = SHA256.new(zeros)
for i in xrange(511):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('9acca8e8c22201155389f65abbf6bc9723edc7384ead80503839f49dcc56d767', h.hexdigest()) # 512 MiB
for i in xrange(8):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('abf51ad954b246009dfe5a50ecd582fd5b8f1b8b27f30393853c3ef721e7fa6e', h.hexdigest()) # 520 MiB
def generate(bits, progress_func=None):
"""
Generate a new private RSA key. This factory function can be used to
generate a new host key or authentication key.
@param bits: number of bits the generated key should be.
@type bits: int
@param progress_func: an optional function to call at key points in
key generation (used by C{pyCrypto.PublicKey}).
@type progress_func: function
@return: new private key
@rtype: L{RSAKey}
"""
signing_key = ECDSA.generate()
key = ECDSAKey(vals=(signing_key, signing_key.get_verifying_key()))
return key
def lmots_sig_to_pub(sig, S, lmots_type, message):
signature = LmotsSignature.deserialize(sig)
if (signature.type != lmots_type):
raise ValueError(err_unknown_typecode)
n, p, w, ls = lmots_params[lmots_type]
hashQ = H(S + signature.C + message + D_MESG)
V = hashQ + checksum(hashQ, w, ls)
hash = SHA256.new()
hash.update(S)
for i, y in enumerate(signature.y):
tmp = y
for j in xrange(coef(V, i, w), 2**w - 1):
tmp = H(S + tmp + u16str(i) + u8str(j) + D_ITER)
hash.update(tmp)
hash.update(D_PBLC)
return hash.digest()
# ***************************************************************
# |
# LMS N-time signatures functions |
# |
# ***************************************************************
def analyseBlock(self, block):
signerPublicKey = self.network.getNodePublicKey(block.node)
try:
pkcs1_15.new(signerPublicKey).verify(SHA256.new(block.hash.encode('utf-8')), block.signature)
except (ValueError, TypeError):
return False
if block.hash[:block.threshold] != '0'*block.threshold:
return False
if block.flags == 0x11:
if not reduce(lambda x, y: x and y, map(self.verifyTransaction, block.transactions)):
return False
return True
#listening service corresponding to getForkedBlocks
def runTest(self):
"""SHA256: 512/520 MiB test"""
from Crypto.Hash import SHA256
zeros = bchr(0x00) * (1024*1024)
h = SHA256.new(zeros)
for i in xrange(511):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('9acca8e8c22201155389f65abbf6bc9723edc7384ead80503839f49dcc56d767', h.hexdigest()) # 512 MiB
for i in xrange(8):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('abf51ad954b246009dfe5a50ecd582fd5b8f1b8b27f30393853c3ef721e7fa6e', h.hexdigest()) # 520 MiB
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 runTest(self):
"""SHA256: 512/520 MiB test"""
from Crypto.Hash import SHA256
zeros = bchr(0x00) * (1024*1024)
h = SHA256.new(zeros)
for i in xrange(511):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('9acca8e8c22201155389f65abbf6bc9723edc7384ead80503839f49dcc56d767', h.hexdigest()) # 512 MiB
for i in xrange(8):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('abf51ad954b246009dfe5a50ecd582fd5b8f1b8b27f30393853c3ef721e7fa6e', h.hexdigest()) # 520 MiB
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 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 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 runTest(self):
"""SHA256: 512/520 MiB test"""
from Crypto.Hash import SHA256
zeros = bchr(0x00) * (1024*1024)
h = SHA256.new(zeros)
for i in xrange(511):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('9acca8e8c22201155389f65abbf6bc9723edc7384ead80503839f49dcc56d767', h.hexdigest()) # 512 MiB
for i in xrange(8):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('abf51ad954b246009dfe5a50ecd582fd5b8f1b8b27f30393853c3ef721e7fa6e', h.hexdigest()) # 520 MiB
def decrypt_secret(secret, key):
"""Python implementation of SystemFunction005.
Decrypts a block of data with DES using given key.
Note that key can be longer than 7 bytes."""
decrypted_data = ''
j = 0 # key index
for i in range(0,len(secret),8):
enc_block = secret[i:i+8]
block_key = key[j:j+7]
des_key = str_to_key(block_key)
des = DES.new(des_key, DES.MODE_ECB)
decrypted_data += des.decrypt(enc_block)
j += 7
if len(key[j:j+7]) < 7:
j = len(key[j:j+7])
(dec_data_len,) = unpack("<L", decrypted_data[:4])
return decrypted_data[8:8+dec_data_len]
def decrypt_aes(secret, key):
sha = SHA256.new()
sha.update(key)
for _i in range(1, 1000+1):
sha.update(secret[28:60])
aeskey = sha.digest()
data = ""
for i in range(60, len(secret), 16):
aes = AES.new(aeskey, AES.MODE_CBC, "\x00"*16)
buf = secret[i : i + 16]
if len(buf) < 16:
buf += (16-len(buf)) * "\00"
data += aes.decrypt(buf)
return data
def decrypt_secret(secret, key):
"""Python implementation of SystemFunction005.
Decrypts a block of data with DES using given key.
Note that key can be longer than 7 bytes."""
decrypted_data = ''
j = 0 # key index
for i in range(0,len(secret),8):
enc_block = secret[i:i+8]
block_key = key[j:j+7]
des_key = str_to_key(block_key)
des = DES.new(des_key, DES.MODE_ECB)
decrypted_data += des.decrypt(enc_block)
j += 7
if len(key[j:j+7]) < 7:
j = len(key[j:j+7])
(dec_data_len,) = unpack("<L", decrypted_data[:4])
return decrypted_data[8:8+dec_data_len]
def rsa_decrypt(priv_key, rsa_ciphertext):
"""Decrypt the RSA ciphertext.
This function decrypt the ciphertext using RSA-OAEP algorithm and private
key of the recipient.
:parameter:
priv_key : RSA key object
The RSA private key used to decrypt.
rsa_ciphertext : string
Ciphertext to decrypt.
:return: A string, the plaintext after decryption.
"""
try:
# create PKCS1 OAEP cipher to perform decryption
cipherer = PKCS1_OAEP.new(priv_key)
# decode then decrypt the ciphertext
rsa_plaintext = cipherer.decrypt(b64decode(rsa_ciphertext))
except (ValueError, TypeError):
print("[error] RSA decryption failed")
return False
return rsa_plaintext
def rsa_sign(priv_key, payload):
"""Sign the payload.
This function sign the payload using RSA-PSS algorithm and private key
of the source.
:parameter:
priv_key : RSA key object
The rsa private key use to sign the payload.
payload : string
The payload to sign.
:return: A string, the RSA-PSS signature of the payload.
"""
# prepare the SHA256 hash
h = SHA256.new()
# create the SHA256 hash of the payload
h.update(payload)
# prepare the PKCS1-PSS signature with the private key
signer = PKCS1_PSS.new(priv_key)
# sign the hash
thesignature = signer.sign(h)
return thesignature
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 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 authenticate_owner_password(self, password):
# Algorithm 3.7
password = (password + self.PASSWORD_PADDING)[:32]
hash = md5.md5(password)
if self.r >= 3:
for _ in range(50):
hash = md5.md5(hash.digest())
n = 5
if self.r >= 3:
n = self.length // 8
key = hash.digest()[:n]
if self.r == 2:
user_password = ARC4.new(key).decrypt(self.o)
else:
user_password = self.o
for i in range(19, -1, -1):
k = b''.join(chr(ord(c) ^ i) for c in key)
user_password = ARC4.new(k).decrypt(user_password)
return self.authenticate_user_password(user_password)
def authenticate(self, password):
password = password.encode('utf-8')[:127]
hash = SHA256.new(password)
hash.update(self.o_validation_salt)
hash.update(self.u)
if hash.digest() == self.o_hash:
hash = SHA256.new(password)
hash.update(self.o_key_salt)
hash.update(self.u)
return AES.new(hash.digest(), mode=AES.MODE_CBC, IV=b'\x00' * 16).decrypt(self.oe)
hash = SHA256.new(password)
hash.update(self.u_validation_salt)
if hash.digest() == self.u_hash:
hash = SHA256.new(password)
hash.update(self.u_key_salt)
return AES.new(hash.digest(), mode=AES.MODE_CBC, IV=b'\x00' * 16).decrypt(self.ue)
return None
def runTest(self):
"""SHA256: 512/520 MiB test"""
from Crypto.Hash import SHA256
zeros = bchr(0x00) * (1024*1024)
h = SHA256.new(zeros)
for i in xrange(511):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('9acca8e8c22201155389f65abbf6bc9723edc7384ead80503839f49dcc56d767', h.hexdigest()) # 512 MiB
for i in xrange(8):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('abf51ad954b246009dfe5a50ecd582fd5b8f1b8b27f30393853c3ef721e7fa6e', h.hexdigest()) # 520 MiB
def runTest(self):
"""SHA256: 512/520 MiB test"""
from Crypto.Hash import SHA256
zeros = bchr(0x00) * (1024*1024)
h = SHA256.new(zeros)
for i in range(511):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('9acca8e8c22201155389f65abbf6bc9723edc7384ead80503839f49dcc56d767', h.hexdigest()) # 512 MiB
for i in range(8):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('abf51ad954b246009dfe5a50ecd582fd5b8f1b8b27f30393853c3ef721e7fa6e', h.hexdigest()) # 520 MiB
def decode(encryptedValues, key):
if key==None:
values = encryptedValues
else:
hashKey = SHA256.new()
hashKey.update(key.encode('utf_8'))
key = hashKey.digest()
decoder = AES.new(key, AES.MODE_ECB)
values = []
for obj in encryptedValues:
number = ''
for s0 in decoder.decrypt(obj).decode("utf-8"):
if s0.isdigit() or s0=='.':
number += s0
else:
break
values.append(float(number))
return values
def runTest(self):
"""SHA256: 512/520 MiB test"""
from Crypto.Hash import SHA256
zeros = bchr(0x00) * (1024*1024)
h = SHA256.new(zeros)
for i in xrange(511):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('9acca8e8c22201155389f65abbf6bc9723edc7384ead80503839f49dcc56d767', h.hexdigest()) # 512 MiB
for i in xrange(8):
h.update(zeros)
# This test vector is from PyCrypto's old testdata.py file.
self.assertEqual('abf51ad954b246009dfe5a50ecd582fd5b8f1b8b27f30393853c3ef721e7fa6e', h.hexdigest()) # 520 MiB
