Skip to content

Commit

Permalink
add multithreading test
Browse files Browse the repository at this point in the history
  • Loading branch information
ngoldbaum committed Nov 26, 2024
1 parent 091a872 commit ba2bcd4
Show file tree
Hide file tree
Showing 2 changed files with 46 additions and 4 deletions.
2 changes: 1 addition & 1 deletion README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -285,7 +285,7 @@ Compatibility
-------------

This library should be compatible with py-bcrypt and it will run on Python
3.6+, and PyPy 3.
3.8+ (including free-threaded builds), and PyPy 3.

Security
--------
Expand Down
48 changes: 45 additions & 3 deletions tests/test_bcrypt.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
import pytest
import random
import uuid

from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass

import bcrypt
import pytest

_test_vectors = [
(
Expand Down Expand Up @@ -171,7 +176,7 @@
]


def test_gensalt_basic(monkeypatch):
def test_gensalt_basic():
salt = bcrypt.gensalt()
assert salt.startswith(b"$2b$12$")

Expand Down Expand Up @@ -219,7 +224,7 @@ def test_gensalt_bad_prefix():
bcrypt.gensalt(prefix=b"bad")


def test_gensalt_2a_prefix(monkeypatch):
def test_gensalt_2a_prefix():
salt = bcrypt.gensalt(prefix=b"2a")
assert salt.startswith(b"$2a$12$")

Expand Down Expand Up @@ -495,3 +500,40 @@ def test_2a_wraparound_bug():
)
== b"$2a$04$R1lJ2gkNaoPGdafE.H.16.1MKHPvmKwryeulRe225LKProWYwt9Oi"
)


@pytest.mark.thread_unsafe()
def test_multithreading():
def get_id():
return uuid.uuid4().bytes

class User:
def __init__(self, id_, pw):
self.id_ = id_
self.salt = bcrypt.gensalt(4)
self.hash_ = bcrypt.hashpw(pw, self.salt)
self.key = bcrypt.kdf(pw, self.salt, 32, 50)
assert self.check(pw)

def check(self, pw):
return bcrypt.checkpw(pw, self.hash_)

# use UUIDs as both ID and passwords
NUM_USERS = 50
ids = [get_id() for _ in range(NUM_USERS)]
pws = {id_: get_id() for id_, _ in zip(ids, range(NUM_USERS))}

user_creator = ThreadPoolExecutor(max_workers=4)

def create_user(id_, pw):
return id_, User(id_, pw)

creator_futures = [
user_creator.submit(create_user, id_, pw) for id_, pw in pws.items()]

users = [future.result() for future in creator_futures]

for id_, user in users:
assert bcrypt.hashpw(pws[id_], user.salt) == user.hash_
assert user.check(pws[id_])
assert bcrypt.kdf(pws[id_], user.salt, 32, 50) == user.key

0 comments on commit ba2bcd4

Please sign in to comment.