Commit 7d46c727 authored by Matti Picus's avatar Matti Picus
Browse files

remove unused rpython _hashlib module, we now use cffi

parent 6bb004a8914f
from __future__ import with_statement
from rpython.rlib import rgc, ropenssl
from rpython.rlib.objectmodel import we_are_translated
from rpython.rlib.rstring import StringBuilder
from rpython.rtyper.lltypesystem import lltype, rffi
from rpython.tool.sourcetools import func_renamer
from pypy.interpreter.baseobjspace import W_Root
from pypy.interpreter.error import OperationError, oefmt
from pypy.interpreter.gateway import unwrap_spec, interp2app, WrappedDefault
from pypy.interpreter.typedef import TypeDef, GetSetProperty
from pypy.module.thread.os_lock import Lock
algorithms = ('md5', 'sha1', 'sha224', 'sha256', 'sha384', 'sha512')
def hash_name_mapper_callback(obj_name, userdata):
if not obj_name:
return
# Ignore aliased names, they pollute the list and OpenSSL appears
# to have a its own definition of alias as the resulting list
# still contains duplicate and alternate names for several
# algorithms.
if rffi.cast(lltype.Signed, obj_name[0].c_alias):
return
name = rffi.charp2str(obj_name[0].c_name)
global_name_fetcher.meth_names.append(name)
class NameFetcher:
def setup(self):
self.meth_names = []
def _cleanup_(self):
self.__dict__.clear()
global_name_fetcher = NameFetcher()
def fetch_names(space):
global_name_fetcher.setup()
ropenssl.init_digests()
ropenssl.OBJ_NAME_do_all(ropenssl.OBJ_NAME_TYPE_MD_METH,
hash_name_mapper_callback, None)
meth_names = global_name_fetcher.meth_names
global_name_fetcher.meth_names = None
return space.call_function(space.w_frozenset, space.newlist(
[space.newtext(name) for name in meth_names]))
class W_Hash(W_Root):
NULL_CTX = lltype.nullptr(ropenssl.EVP_MD_CTX.TO)
ctx = NULL_CTX
def __init__(self, space, name, copy_from=NULL_CTX):
self.name = name
digest_type = self.digest_type_by_name(space)
self.digest_size = ropenssl.EVP_MD_size(digest_type)
# Allocate a lock for each HASH object.
# An optimization would be to not release the GIL on small requests,
# and use a custom lock only when needed.
self.lock = Lock(space)
ctx = ropenssl.EVP_MD_CTX_new()
if ctx is None:
raise MemoryError
rgc.add_memory_pressure(ropenssl.HASH_MALLOC_SIZE + self.digest_size,
self)
try:
if copy_from:
if not ropenssl.EVP_MD_CTX_copy(ctx, copy_from):
raise ValueError
else:
ropenssl.EVP_DigestInit(ctx, digest_type)
self.ctx = ctx
except:
ropenssl.EVP_MD_CTX_free(ctx)
raise
self.register_finalizer(space)
def _finalize_(self):
ctx = self.ctx
if ctx:
self.ctx = lltype.nullptr(ropenssl.EVP_MD_CTX.TO)
ropenssl.EVP_MD_CTX_free(ctx)
def digest_type_by_name(self, space):
digest_type = ropenssl.EVP_get_digestbyname(self.name)
if not digest_type:
raise oefmt(space.w_ValueError, "unknown hash function")
return digest_type
def descr_repr(self, space):
addrstring = self.getaddrstring(space)
return space.newtext("<%s HASH object at 0x%s>" % (
self.name, addrstring))
@unwrap_spec(string='bufferstr')
def update(self, space, string):
with rffi.scoped_nonmovingbuffer(string) as buf:
with self.lock:
# XXX try to not release the GIL for small requests
ropenssl.EVP_DigestUpdate(self.ctx, buf, len(string))
def copy(self, space):
"Return a copy of the hash object."
with self.lock:
return W_Hash(space, self.name, copy_from=self.ctx)
def digest(self, space):
"Return the digest value as a string of binary data."
digest = self._digest(space)
return space.newbytes(digest)
def hexdigest(self, space):
"Return the digest value as a string of hexadecimal digits."
digest = self._digest(space)
hexdigits = '0123456789abcdef'
result = StringBuilder(self.digest_size * 2)
for c in digest:
result.append(hexdigits[(ord(c) >> 4) & 0xf])
result.append(hexdigits[ ord(c) & 0xf])
return space.newtext(result.build())
def get_digest_size(self, space):
return space.newint(self.digest_size)
def get_block_size(self, space):
digest_type = self.digest_type_by_name(space)
block_size = ropenssl.EVP_MD_block_size(digest_type)
return space.newint(block_size)
def get_name(self, space):
return space.newtext(self.name)
def _digest(self, space):
ctx = ropenssl.EVP_MD_CTX_new()
if ctx is None:
raise MemoryError
try:
with self.lock:
if not ropenssl.EVP_MD_CTX_copy(ctx, self.ctx):
raise ValueError
digest_size = self.digest_size
with rffi.scoped_alloc_buffer(digest_size) as buf:
ropenssl.EVP_DigestFinal(ctx, buf.raw, None)
return buf.str(digest_size)
finally:
ropenssl.EVP_MD_CTX_free(ctx)
W_Hash.typedef = TypeDef(
'HASH',
__repr__=interp2app(W_Hash.descr_repr),
update=interp2app(W_Hash.update),
copy=interp2app(W_Hash.copy),
digest=interp2app(W_Hash.digest),
hexdigest=interp2app(W_Hash.hexdigest),
#
digest_size=GetSetProperty(W_Hash.get_digest_size),
digestsize=GetSetProperty(W_Hash.get_digest_size),
block_size=GetSetProperty(W_Hash.get_block_size),
name=GetSetProperty(W_Hash.get_name),
)
W_Hash.typedef.acceptable_as_base_class = False
@unwrap_spec(name='text', string='bufferstr')
def new(space, name, string=''):
w_hash = W_Hash(space, name)
w_hash.update(space, string)
return w_hash
# shortcut functions
def make_new_hash(name, funcname):
@func_renamer(funcname)
@unwrap_spec(string='bufferstr')
def new_hash(space, string=''):
return new(space, name, string)
return new_hash
for _name in algorithms:
_newname = 'new_%s' % (_name,)
globals()[_newname] = make_new_hash(_name, _newname)
HAS_FAST_PKCS5_PBKDF2_HMAC = ropenssl.PKCS5_PBKDF2_HMAC is not None
if HAS_FAST_PKCS5_PBKDF2_HMAC:
@unwrap_spec(name='text', password='bytes', salt='bytes', rounds=int,
w_dklen=WrappedDefault(None))
def pbkdf2_hmac(space, name, password, salt, rounds, w_dklen):
digest = ropenssl.EVP_get_digestbyname(name)
if not digest:
raise oefmt(space.w_ValueError, "unknown hash function")
if space.is_w(w_dklen, space.w_None):
dklen = ropenssl.EVP_MD_size(digest)
else:
dklen = space.int_w(w_dklen)
if dklen < 1:
raise oefmt(space.w_ValueError,
"key length must be greater than 0.")
with rffi.scoped_alloc_buffer(dklen) as buf:
r = ropenssl.PKCS5_PBKDF2_HMAC(
password, len(password), salt, len(salt), rounds, digest,
dklen, buf.raw)
if not r:
raise ValueError
return space.newbytes(buf.str(dklen))
from pypy.interpreter.mixedmodule import MixedModule
from pypy.module._hashlib.interp_hashlib import (
algorithms, fetch_names, HAS_FAST_PKCS5_PBKDF2_HMAC)
class Module(MixedModule):
interpleveldefs = {
'new' : 'interp_hashlib.new',
}
appleveldefs = {
}
for name in algorithms:
interpleveldefs['openssl_' + name] = 'interp_hashlib.new_' + name
if HAS_FAST_PKCS5_PBKDF2_HMAC:
interpleveldefs['pbkdf2_hmac'] = 'interp_hashlib.pbkdf2_hmac'
def startup(self, space):
w_meth_names = fetch_names(space)
space.setattr(self, space.newtext('openssl_md_meth_names'), w_meth_names)
class AppTestHashlib:
spaceconfig = {
"usemodules": ['_hashlib', 'array', 'struct', 'binascii'],
}
def test_method_names(self):
import _hashlib
assert isinstance(_hashlib.openssl_md_meth_names, frozenset)
assert "md5" in _hashlib.openssl_md_meth_names
def test_simple(self):
import _hashlib
assert _hashlib.new('md5').__class__.__name__ == 'HASH'
assert len(_hashlib.new('md5').hexdigest()) == 32
def test_attributes(self):
import hashlib
for name, (expected_size, expected_block_size) in {
'md5': (16, 64),
'sha1': (20, 64),
'sha224': (28, 64),
'sha256': (32, 64),
'sha384': (48, 128),
'sha512': (64, 128),
}.items():
h = hashlib.new(name)
assert h.name == name
assert h.digest_size == expected_size
assert h.digestsize == expected_size
assert h.block_size == expected_block_size
#
h.update('abc')
h2 = h.copy()
h.update('def')
digest = h.digest()
hexdigest = h.hexdigest()
h2.update('d')
h2.update('ef')
assert digest == h2.digest()
assert hexdigest == h2.hexdigest()
assert len(digest) == h.digest_size
assert len(hexdigest) == h.digest_size * 2
c_digest = digest
c_hexdigest = hexdigest
# also test the pure Python implementation
py_new = getattr(hashlib, '__get_builtin_constructor')
h = py_new(name)('')
assert h.digest_size == expected_size
assert h.digestsize == expected_size
assert h.block_size == expected_block_size
#
h.update('abc')
h2 = h.copy()
h.update('def')
digest = h.digest()
hexdigest = h.hexdigest()
h2.update('d')
h2.update('ef')
assert digest == h2.digest()
assert hexdigest == h2.hexdigest()
# compare both implementations
assert c_digest == digest
assert c_hexdigest == hexdigest
def test_shortcut(self):
import hashlib
assert repr(hashlib.md5()).startswith("<md5 HASH object")
def test_unicode(self):
import _hashlib
assert _hashlib.new('sha1', u'xxx').__class__.__name__ == 'HASH'
def test_uppercase(self):
import _hashlib
h = _hashlib.new('MD5')
assert h.digest_size == 16
assert len(h.hexdigest()) == 32
def test_buffer(self):
import _hashlib, array
b = array.array('b', 'x' * 10)
h = _hashlib.new('md5', b)
h.update(b)
assert h.digest() == _hashlib.openssl_md5('x' * 20).digest()
_hashlib.openssl_sha1(b).digest()
def test_extra_algorithms(self):
expected_results = {
"md5": "bb649c83dd1ea5c9d9dec9a18df0ffe9",
"md4": "c275b8454684ea416b93d7a418b43176",
"mdc2": None, # XXX find the correct expected value
"sha": "e2b0a8609b47c58e5d984c9ccfe69f9b654b032b",
"ripemd160": "cc4a5ce1b3df48aec5d22d1f16b894a0b894eccc",
"whirlpool": ("1a22b79fe5afda02c63a25927193ed01dc718b74"
"026e597608ce431f9c3d2c9e74a7350b7fbb7c5d"
"4effe5d7a31879b8b7a10fd2f544c4ca268ecc6793923583"),
}
import _hashlib
test_string = "Nobody inspects the spammish repetition"
for hash_name, expected in sorted(expected_results.items()):
try:
m = _hashlib.new(hash_name)
except ValueError as e:
print 'skipped %s: %s' % (hash_name, e)
continue
m.update(test_string)
got = m.hexdigest()
assert got and type(got) is str and len(got) % 2 == 0
got.decode('hex')
if expected is not None:
assert got == expected
def test_pbkdf2(self):
try:
from _hashlib import pbkdf2_hmac
except ImportError:
skip("Requires OpenSSL >= 1.1")
out = pbkdf2_hmac('sha1', 'password', 'salt', 1)
assert out == '0c60c80f961f0e71f3a9b524af6012062fe037a6'.decode('hex')
out = pbkdf2_hmac('sha1', 'password', 'salt', 2, None)
assert out == 'ea6c014dc72d6f8ccd1ed92ace1d41f0d8de8957'.decode('hex')
from pypy.objspace.fake.checkmodule import checkmodule
def test_checkmodule():
checkmodule('_hashlib')
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment