Merge pull request #2080 from guardicore/1965-stateful-authentication-service
1965 stateful authentication service
This commit is contained in:
commit
93deaebff3
|
@ -9,3 +9,4 @@ from .pba_results import PbaResults
|
|||
from monkey_island.cc.models.report.report import Report
|
||||
from .stolen_credentials import StolenCredentials
|
||||
from .simulation import Simulation, SimulationSchema, IslandMode
|
||||
from .user_credentials import UserCredentials
|
||||
|
|
|
@ -3,7 +3,7 @@ from __future__ import annotations
|
|||
from typing import Dict
|
||||
|
||||
|
||||
class UserCreds:
|
||||
class UserCredentials:
|
||||
def __init__(self, username, password_hash):
|
||||
self.username = username
|
||||
self.password_hash = password_hash
|
|
@ -6,6 +6,7 @@ from .i_agent_binary_repository import IAgentBinaryRepository
|
|||
from .i_agent_configuration_repository import IAgentConfigurationRepository
|
||||
from .i_simulation_repository import ISimulationRepository
|
||||
from .i_credentials_repository import ICredentialsRepository
|
||||
from .i_user_repository import IUserRepository
|
||||
|
||||
|
||||
from .local_storage_file_repository import LocalStorageFileRepository
|
||||
|
@ -16,4 +17,5 @@ from .file_repository_logging_decorator import FileRepositoryLoggingDecorator
|
|||
from .agent_binary_repository import AgentBinaryRepository
|
||||
from .file_agent_configuration_repository import FileAgentConfigurationRepository
|
||||
from .file_simulation_repository import FileSimulationRepository
|
||||
from .json_file_user_repository import JSONFileUserRepository
|
||||
from .mongo_credentials_repository import MongoCredentialsRepository
|
||||
|
|
|
@ -1,9 +1,9 @@
|
|||
import abc
|
||||
|
||||
from .user_creds import UserCreds
|
||||
from monkey_island.cc.models import UserCredentials
|
||||
|
||||
|
||||
class IUserDatastore(metaclass=abc.ABCMeta):
|
||||
class IUserRepository(metaclass=abc.ABCMeta):
|
||||
"""
|
||||
Allows user credentials to be stored and retrieved.
|
||||
"""
|
||||
|
@ -17,7 +17,7 @@ class IUserDatastore(metaclass=abc.ABCMeta):
|
|||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def add_user(self, credentials: UserCreds):
|
||||
def add_user(self, credentials: UserCredentials):
|
||||
"""
|
||||
Adds a new user to the datastore.
|
||||
:param UserCreds credentials: New user credentials to persistant storage.
|
||||
|
@ -26,7 +26,7 @@ class IUserDatastore(metaclass=abc.ABCMeta):
|
|||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def get_user_credentials(self, username: str) -> UserCreds:
|
||||
def get_user_credentials(self, username: str) -> UserCredentials:
|
||||
"""
|
||||
Gets the user matching `username` from storage.
|
||||
:param str username: The username for which credentials will be retrieved
|
|
@ -6,15 +6,14 @@ from common.utils.exceptions import (
|
|||
InvalidRegistrationCredentialsError,
|
||||
UnknownUserError,
|
||||
)
|
||||
from monkey_island.cc.models import UserCredentials
|
||||
from monkey_island.cc.repository import IUserRepository
|
||||
from monkey_island.cc.server_utils.file_utils import open_new_securely_permissioned_file
|
||||
|
||||
from .i_user_datastore import IUserDatastore
|
||||
from .user_creds import UserCreds
|
||||
|
||||
CREDENTIALS_FILE = "credentials.json"
|
||||
|
||||
|
||||
class JsonFileUserDatastore(IUserDatastore):
|
||||
class JSONFileUserRepository(IUserRepository):
|
||||
def __init__(self, data_dir: Path):
|
||||
self._credentials = None
|
||||
self._credentials_file = data_dir / CREDENTIALS_FILE
|
||||
|
@ -22,16 +21,16 @@ class JsonFileUserDatastore(IUserDatastore):
|
|||
if self._credentials_file.exists():
|
||||
self._credentials = self._load_from_file()
|
||||
|
||||
def _load_from_file(self) -> UserCreds:
|
||||
def _load_from_file(self) -> UserCredentials:
|
||||
with open(self._credentials_file, "r") as f:
|
||||
credentials_dict = json.load(f)
|
||||
|
||||
return UserCreds(credentials_dict["user"], credentials_dict["password_hash"])
|
||||
return UserCredentials(credentials_dict["user"], credentials_dict["password_hash"])
|
||||
|
||||
def has_registered_users(self) -> bool:
|
||||
return self._credentials is not None
|
||||
|
||||
def add_user(self, credentials: UserCreds):
|
||||
def add_user(self, credentials: UserCredentials):
|
||||
if credentials is None:
|
||||
raise InvalidRegistrationCredentialsError("Credentials can not be 'None'")
|
||||
elif not credentials.username:
|
||||
|
@ -51,7 +50,7 @@ class JsonFileUserDatastore(IUserDatastore):
|
|||
with open_new_securely_permissioned_file(self._credentials_file, "w") as f:
|
||||
json.dump(self._credentials.to_dict(), f, indent=2)
|
||||
|
||||
def get_user_credentials(self, username: str) -> UserCreds:
|
||||
def get_user_credentials(self, username: str) -> UserCredentials:
|
||||
if self._credentials is None or self._credentials.username != username:
|
||||
raise UnknownUserError(f"User {username} does not exist.")
|
||||
|
|
@ -29,6 +29,9 @@ class Authenticate(AbstractResource):
|
|||
|
||||
urls = ["/api/auth"]
|
||||
|
||||
def __init__(self, authentication_service: AuthenticationService):
|
||||
self._authentication_service = authentication_service
|
||||
|
||||
def post(self):
|
||||
"""
|
||||
Example request: \
|
||||
|
@ -41,7 +44,7 @@ class Authenticate(AbstractResource):
|
|||
username, password = get_username_password_from_request(request)
|
||||
|
||||
try:
|
||||
AuthenticationService.authenticate(username, password)
|
||||
self._authentication_service.authenticate(username, password)
|
||||
access_token = create_access_token(username)
|
||||
except IncorrectCredentialsError:
|
||||
return make_response({"error": "Invalid credentials"}, 401)
|
||||
|
|
|
@ -14,14 +14,17 @@ class Registration(AbstractResource):
|
|||
|
||||
urls = ["/api/registration"]
|
||||
|
||||
def __init__(self, authentication_service: AuthenticationService):
|
||||
self._authentication_service = authentication_service
|
||||
|
||||
def get(self):
|
||||
return {"needs_registration": AuthenticationService.needs_registration()}
|
||||
return {"needs_registration": self._authentication_service.needs_registration()}
|
||||
|
||||
def post(self):
|
||||
username, password = get_username_password_from_request(request)
|
||||
|
||||
try:
|
||||
AuthenticationService.register_new_user(username, password)
|
||||
self._authentication_service.register_new_user(username, password)
|
||||
return make_response({"error": ""}, 200)
|
||||
# API Spec: HTTP status code for AlreadyRegisteredError should be 409 (CONFLICT)
|
||||
except (InvalidRegistrationCredentialsError, AlreadyRegisteredError) as e:
|
||||
|
|
|
@ -1,5 +1,4 @@
|
|||
from .authentication.authentication_service import AuthenticationService
|
||||
from .authentication.json_file_user_datastore import JsonFileUserDatastore
|
||||
from .authentication_service import AuthenticationService
|
||||
|
||||
from .aws import AWSService
|
||||
from .island_mode_service import IslandModeService
|
||||
|
|
|
@ -7,63 +7,50 @@ from common.utils.exceptions import (
|
|||
InvalidRegistrationCredentialsError,
|
||||
UnknownUserError,
|
||||
)
|
||||
from monkey_island.cc.models import UserCredentials
|
||||
from monkey_island.cc.repository import IUserRepository
|
||||
from monkey_island.cc.server_utils.encryption import (
|
||||
reset_datastore_encryptor,
|
||||
unlock_datastore_encryptor,
|
||||
)
|
||||
from monkey_island.cc.setup.mongo.database_initializer import reset_database
|
||||
|
||||
from .i_user_datastore import IUserDatastore
|
||||
from .user_creds import UserCreds
|
||||
|
||||
|
||||
class AuthenticationService:
|
||||
DATA_DIR = None
|
||||
user_datastore = None
|
||||
def __init__(self, data_dir: Path, user_datastore: IUserRepository):
|
||||
self._data_dir = data_dir
|
||||
self._user_datastore = user_datastore
|
||||
|
||||
# TODO: A number of these services should be instance objects instead of
|
||||
# static/singleton hybrids. At the moment, this requires invasive refactoring that's
|
||||
# not a priority.
|
||||
@classmethod
|
||||
def initialize(cls, data_dir: Path, user_datastore: IUserDatastore):
|
||||
cls.DATA_DIR = data_dir
|
||||
cls.user_datastore = user_datastore
|
||||
def needs_registration(self) -> bool:
|
||||
return not self._user_datastore.has_registered_users()
|
||||
|
||||
@classmethod
|
||||
def needs_registration(cls) -> bool:
|
||||
return not cls.user_datastore.has_registered_users()
|
||||
|
||||
@classmethod
|
||||
def register_new_user(cls, username: str, password: str):
|
||||
def register_new_user(self, username: str, password: str):
|
||||
if not username or not password:
|
||||
raise InvalidRegistrationCredentialsError("Username or password can not be empty.")
|
||||
|
||||
credentials = UserCreds(username, _hash_password(password))
|
||||
cls.user_datastore.add_user(credentials)
|
||||
cls._reset_datastore_encryptor(username, password)
|
||||
credentials = UserCredentials(username, _hash_password(password))
|
||||
self._user_datastore.add_user(credentials)
|
||||
self._reset_datastore_encryptor(username, password)
|
||||
reset_database()
|
||||
|
||||
@classmethod
|
||||
def authenticate(cls, username: str, password: str):
|
||||
def authenticate(self, username: str, password: str):
|
||||
try:
|
||||
registered_user = cls.user_datastore.get_user_credentials(username)
|
||||
registered_user = self._user_datastore.get_user_credentials(username)
|
||||
except UnknownUserError:
|
||||
raise IncorrectCredentialsError()
|
||||
|
||||
if not _credentials_match_registered_user(username, password, registered_user):
|
||||
raise IncorrectCredentialsError()
|
||||
|
||||
cls._unlock_datastore_encryptor(username, password)
|
||||
self._unlock_datastore_encryptor(username, password)
|
||||
|
||||
@classmethod
|
||||
def _unlock_datastore_encryptor(cls, username: str, password: str):
|
||||
def _unlock_datastore_encryptor(self, username: str, password: str):
|
||||
secret = _get_secret_from_credentials(username, password)
|
||||
unlock_datastore_encryptor(cls.DATA_DIR, secret)
|
||||
unlock_datastore_encryptor(self._data_dir, secret)
|
||||
|
||||
@classmethod
|
||||
def _reset_datastore_encryptor(cls, username: str, password: str):
|
||||
def _reset_datastore_encryptor(self, username: str, password: str):
|
||||
secret = _get_secret_from_credentials(username, password)
|
||||
reset_datastore_encryptor(cls.DATA_DIR, secret)
|
||||
reset_datastore_encryptor(self._data_dir, secret)
|
||||
|
||||
|
||||
def _hash_password(plaintext_password: str) -> str:
|
||||
|
@ -74,7 +61,7 @@ def _hash_password(plaintext_password: str) -> str:
|
|||
|
||||
|
||||
def _credentials_match_registered_user(
|
||||
username: str, password: str, registered_user: UserCreds
|
||||
username: str, password: str, registered_user: UserCredentials
|
||||
) -> bool:
|
||||
return (registered_user.username == username) and _password_matches_hash(
|
||||
password, registered_user.password_hash
|
|
@ -24,6 +24,8 @@ from monkey_island.cc.repository import (
|
|||
ICredentialsRepository,
|
||||
IFileRepository,
|
||||
ISimulationRepository,
|
||||
IUserRepository,
|
||||
JSONFileUserRepository,
|
||||
LocalStorageFileRepository,
|
||||
MongoCredentialsRepository,
|
||||
RetrievalError,
|
||||
|
@ -40,7 +42,7 @@ from monkey_island.cc.services.telemetry.processing.processing import (
|
|||
)
|
||||
from monkey_island.cc.setup.mongo.mongo_setup import MONGO_URL
|
||||
|
||||
from . import AuthenticationService, JsonFileUserDatastore
|
||||
from . import AuthenticationService
|
||||
from .reporting.report import ReportService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
@ -64,7 +66,6 @@ def initialize_services(data_dir: Path) -> DIContainer:
|
|||
|
||||
# This is temporary until we get DI all worked out.
|
||||
PostBreachFilesService.initialize(container.resolve(IFileRepository))
|
||||
AuthenticationService.initialize(data_dir, JsonFileUserDatastore(data_dir))
|
||||
ReportService.initialize(container.resolve(AWSService))
|
||||
|
||||
return container
|
||||
|
@ -95,6 +96,7 @@ def _register_repositories(container: DIContainer, data_dir: Path):
|
|||
container.register_instance(
|
||||
ICredentialsRepository, container.resolve(MongoCredentialsRepository)
|
||||
)
|
||||
container.register_instance(IUserRepository, container.resolve(JSONFileUserRepository))
|
||||
|
||||
|
||||
def _decorate_file_repository(file_repository: IFileRepository) -> IFileRepository:
|
||||
|
@ -140,6 +142,7 @@ def _register_services(container: DIContainer):
|
|||
container.register_instance(AWSService, container.resolve(AWSService))
|
||||
container.register_instance(LocalMonkeyRunService, container.resolve(LocalMonkeyRunService))
|
||||
container.register_instance(IslandModeService, container.resolve(IslandModeService))
|
||||
container.register_instance(AuthenticationService, container.resolve(AuthenticationService))
|
||||
|
||||
|
||||
def _patch_credentials_parser(container: DIContainer):
|
||||
|
|
|
@ -1,36 +1,36 @@
|
|||
from monkey_island.cc.services.authentication.user_creds import UserCreds
|
||||
from monkey_island.cc.models import UserCredentials
|
||||
|
||||
TEST_USER = "Test"
|
||||
TEST_HASH = "abc1231234"
|
||||
|
||||
|
||||
def test_bool_true():
|
||||
assert UserCreds(TEST_USER, TEST_HASH)
|
||||
assert UserCredentials(TEST_USER, TEST_HASH)
|
||||
|
||||
|
||||
def test_bool_false_empty_password_hash():
|
||||
assert not UserCreds(TEST_USER, "")
|
||||
assert not UserCredentials(TEST_USER, "")
|
||||
|
||||
|
||||
def test_bool_false_empty_user():
|
||||
assert not UserCreds("", TEST_HASH)
|
||||
assert not UserCredentials("", TEST_HASH)
|
||||
|
||||
|
||||
def test_bool_false_empty_user_and_password_hash():
|
||||
assert not UserCreds("", "")
|
||||
assert not UserCredentials("", "")
|
||||
|
||||
|
||||
def test_to_dict_empty_creds():
|
||||
user_creds = UserCreds("", "")
|
||||
user_creds = UserCredentials("", "")
|
||||
assert user_creds.to_dict() == {}
|
||||
|
||||
|
||||
def test_to_dict_full_creds():
|
||||
user_creds = UserCreds(TEST_USER, TEST_HASH)
|
||||
user_creds = UserCredentials(TEST_USER, TEST_HASH)
|
||||
assert user_creds.to_dict() == {"user": TEST_USER, "password_hash": TEST_HASH}
|
||||
|
||||
|
||||
def test_member_values(monkeypatch):
|
||||
creds = UserCreds(TEST_USER, TEST_HASH)
|
||||
creds = UserCredentials(TEST_USER, TEST_HASH)
|
||||
assert creds.username == TEST_USER
|
||||
assert creds.password_hash == TEST_HASH
|
|
@ -9,12 +9,12 @@ from common.utils.exceptions import (
|
|||
InvalidRegistrationCredentialsError,
|
||||
UnknownUserError,
|
||||
)
|
||||
from monkey_island.cc.server_utils.file_utils import is_windows_os
|
||||
from monkey_island.cc.services.authentication.json_file_user_datastore import (
|
||||
from monkey_island.cc.models import UserCredentials
|
||||
from monkey_island.cc.repository.json_file_user_repository import (
|
||||
CREDENTIALS_FILE,
|
||||
JsonFileUserDatastore,
|
||||
JSONFileUserRepository,
|
||||
)
|
||||
from monkey_island.cc.services.authentication.user_creds import UserCreds
|
||||
from monkey_island.cc.server_utils.file_utils import is_windows_os
|
||||
|
||||
USERNAME = "test"
|
||||
PASSWORD_HASH = "DEADBEEF"
|
||||
|
@ -22,12 +22,12 @@ PASSWORD_HASH = "DEADBEEF"
|
|||
|
||||
@pytest.fixture
|
||||
def empty_datastore(tmp_path):
|
||||
return JsonFileUserDatastore(tmp_path)
|
||||
return JSONFileUserRepository(tmp_path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def populated_datastore(data_for_tests_dir):
|
||||
return JsonFileUserDatastore(data_for_tests_dir)
|
||||
return JSONFileUserRepository(data_for_tests_dir)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
@ -46,14 +46,14 @@ def test_has_registered_users_after_registration(populated_datastore):
|
|||
def test_add_user(empty_datastore, credentials_file_path):
|
||||
datastore = empty_datastore
|
||||
|
||||
datastore.add_user(UserCreds(USERNAME, PASSWORD_HASH))
|
||||
datastore.add_user(UserCredentials(USERNAME, PASSWORD_HASH))
|
||||
assert datastore.has_registered_users()
|
||||
assert credentials_file_path.exists()
|
||||
|
||||
|
||||
@pytest.mark.skipif(is_windows_os(), reason="Tests Posix (not Windows) permissions.")
|
||||
def test_add_user__term_posix(empty_datastore, credentials_file_path):
|
||||
empty_datastore.add_user(UserCreds(USERNAME, PASSWORD_HASH))
|
||||
empty_datastore.add_user(UserCredentials(USERNAME, PASSWORD_HASH))
|
||||
st = os.stat(credentials_file_path)
|
||||
|
||||
expected_mode = stat.S_IRUSR | stat.S_IWUSR
|
||||
|
@ -66,7 +66,7 @@ def test_add_user__term_posix(empty_datastore, credentials_file_path):
|
|||
def test_add_user__term_windows(empty_datastore, credentials_file_path):
|
||||
datastore = empty_datastore
|
||||
|
||||
datastore.add_user(UserCreds(USERNAME, PASSWORD_HASH))
|
||||
datastore.add_user(UserCredentials(USERNAME, PASSWORD_HASH))
|
||||
assert_windows_permissions(str(credentials_file_path))
|
||||
|
||||
|
||||
|
@ -77,24 +77,24 @@ def test_add_user__None_creds(empty_datastore):
|
|||
|
||||
def test_add_user__empty_username(empty_datastore):
|
||||
with pytest.raises(InvalidRegistrationCredentialsError):
|
||||
empty_datastore.add_user(UserCreds("", PASSWORD_HASH))
|
||||
empty_datastore.add_user(UserCredentials("", PASSWORD_HASH))
|
||||
|
||||
|
||||
def test_add_user__empty_password_hash(empty_datastore):
|
||||
with pytest.raises(InvalidRegistrationCredentialsError):
|
||||
empty_datastore.add_user(UserCreds(USERNAME, ""))
|
||||
empty_datastore.add_user(UserCredentials(USERNAME, ""))
|
||||
|
||||
|
||||
def test_add_user__already_registered(populated_datastore):
|
||||
with pytest.raises(AlreadyRegisteredError):
|
||||
populated_datastore.add_user(UserCreds("new_user", "new_hash"))
|
||||
populated_datastore.add_user(UserCredentials("new_user", "new_hash"))
|
||||
|
||||
|
||||
def test_get_user_credentials_from_file(tmp_path):
|
||||
empty_datastore = JsonFileUserDatastore(tmp_path)
|
||||
empty_datastore.add_user(UserCreds(USERNAME, PASSWORD_HASH))
|
||||
empty_datastore = JSONFileUserRepository(tmp_path)
|
||||
empty_datastore.add_user(UserCredentials(USERNAME, PASSWORD_HASH))
|
||||
|
||||
populated_datastore = JsonFileUserDatastore(tmp_path)
|
||||
populated_datastore = JSONFileUserRepository(tmp_path)
|
||||
stored_user = populated_datastore.get_user_credentials(USERNAME)
|
||||
|
||||
assert stored_user.username == USERNAME
|
|
@ -0,0 +1,24 @@
|
|||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from tests.common import StubDIContainer
|
||||
|
||||
from monkey_island.cc.services import AuthenticationService
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_authentication_service():
|
||||
mock_service = MagicMock(spec=AuthenticationService)
|
||||
mock_service.authenticate = MagicMock()
|
||||
|
||||
return mock_service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def flask_client(build_flask_client, mock_authentication_service):
|
||||
container = StubDIContainer()
|
||||
|
||||
container.register_instance(AuthenticationService, mock_authentication_service)
|
||||
|
||||
with build_flask_client(container) as flask_client:
|
||||
yield flask_client
|
|
@ -11,16 +11,6 @@ PASSWORD = "test_password"
|
|||
TEST_REQUEST = f'{{"username": "{USERNAME}", "password": "{PASSWORD}"}}'
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_authentication_service(monkeypatch):
|
||||
mock_service = MagicMock()
|
||||
mock_service.authenticate = MagicMock()
|
||||
|
||||
monkeypatch.setattr("monkey_island.cc.resources.auth.auth.AuthenticationService", mock_service)
|
||||
|
||||
return mock_service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def make_auth_request(flask_client):
|
||||
url = Authenticate.urls[0]
|
||||
|
|
|
@ -12,19 +12,6 @@ USERNAME = "test_user"
|
|||
PASSWORD = "test_password"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_authentication_service(monkeypatch):
|
||||
mock_service = MagicMock()
|
||||
mock_service.register_new_user = MagicMock()
|
||||
mock_service.needs_registration = MagicMock()
|
||||
|
||||
monkeypatch.setattr(
|
||||
"monkey_island.cc.resources.auth.registration.AuthenticationService", mock_service
|
||||
)
|
||||
|
||||
return mock_service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def make_registration_request(flask_client):
|
||||
def inner(request_body):
|
||||
|
|
|
@ -8,17 +8,16 @@ from common.utils.exceptions import (
|
|||
InvalidRegistrationCredentialsError,
|
||||
UnknownUserError,
|
||||
)
|
||||
from monkey_island.cc.services import AuthenticationService
|
||||
from monkey_island.cc.services.authentication import authentication_service
|
||||
from monkey_island.cc.services.authentication.i_user_datastore import IUserDatastore
|
||||
from monkey_island.cc.services.authentication.user_creds import UserCreds
|
||||
from monkey_island.cc.models import UserCredentials
|
||||
from monkey_island.cc.repository import IUserRepository
|
||||
from monkey_island.cc.services import AuthenticationService, authentication_service
|
||||
|
||||
USERNAME = "user1"
|
||||
PASSWORD = "test"
|
||||
PASSWORD_HASH = "$2b$12$YsGjjuJFddYJ6z5S5/nMCuKkCzKHB1AWY9SXkQ02i25d8TgdhIRS2"
|
||||
|
||||
|
||||
class MockUserDatastore(IUserDatastore):
|
||||
class MockUserDatastore(IUserRepository):
|
||||
def __init__(self, has_registered_users, add_user, get_user_credentials):
|
||||
self._has_registered_users = has_registered_users
|
||||
self._add_user = add_user
|
||||
|
@ -27,10 +26,10 @@ class MockUserDatastore(IUserDatastore):
|
|||
def has_registered_users(self):
|
||||
return self._has_registered_users()
|
||||
|
||||
def add_user(self, credentials: UserCreds):
|
||||
def add_user(self, credentials: UserCredentials):
|
||||
return self._add_user(credentials)
|
||||
|
||||
def get_user_credentials(self, username: str) -> UserCreds:
|
||||
def get_user_credentials(self, username: str) -> UserCredentials:
|
||||
return self._get_user_credentials(username)
|
||||
|
||||
|
||||
|
@ -69,8 +68,7 @@ def test_needs_registration__true(tmp_path):
|
|||
has_registered_users = False
|
||||
mock_user_datastore = MockUserDatastore(lambda: has_registered_users, None, None)
|
||||
|
||||
a_s = AuthenticationService()
|
||||
a_s.initialize(tmp_path, mock_user_datastore)
|
||||
a_s = AuthenticationService(tmp_path, mock_user_datastore)
|
||||
|
||||
assert a_s.needs_registration()
|
||||
|
||||
|
@ -79,8 +77,7 @@ def test_needs_registration__false(tmp_path):
|
|||
has_registered_users = True
|
||||
mock_user_datastore = MockUserDatastore(lambda: has_registered_users, None, None)
|
||||
|
||||
a_s = AuthenticationService()
|
||||
a_s.initialize(tmp_path, mock_user_datastore)
|
||||
a_s = AuthenticationService(tmp_path, mock_user_datastore)
|
||||
|
||||
assert not a_s.needs_registration()
|
||||
|
||||
|
@ -92,8 +89,7 @@ def test_register_new_user__fails(
|
|||
):
|
||||
mock_user_datastore = MockUserDatastore(lambda: True, MagicMock(side_effect=error), None)
|
||||
|
||||
a_s = AuthenticationService()
|
||||
a_s.initialize(tmp_path, mock_user_datastore)
|
||||
a_s = AuthenticationService(tmp_path, mock_user_datastore)
|
||||
|
||||
with pytest.raises(error):
|
||||
a_s.register_new_user(USERNAME, PASSWORD)
|
||||
|
@ -107,8 +103,7 @@ def test_register_new_user__empty_password_fails(
|
|||
):
|
||||
mock_user_datastore = MockUserDatastore(lambda: False, None, None)
|
||||
|
||||
a_s = AuthenticationService()
|
||||
a_s.initialize(tmp_path, mock_user_datastore)
|
||||
a_s = AuthenticationService(tmp_path, mock_user_datastore)
|
||||
|
||||
with pytest.raises(InvalidRegistrationCredentialsError):
|
||||
a_s.register_new_user(USERNAME, "")
|
||||
|
@ -122,8 +117,7 @@ def test_register_new_user(tmp_path, mock_reset_datastore_encryptor, mock_reset_
|
|||
mock_add_user = MagicMock()
|
||||
mock_user_datastore = MockUserDatastore(lambda: False, mock_add_user, None)
|
||||
|
||||
a_s = AuthenticationService()
|
||||
a_s.initialize(tmp_path, mock_user_datastore)
|
||||
a_s = AuthenticationService(tmp_path, mock_user_datastore)
|
||||
|
||||
a_s.register_new_user(USERNAME, PASSWORD)
|
||||
|
||||
|
@ -141,11 +135,10 @@ def test_authenticate__success(tmp_path, mock_unlock_datastore_encryptor):
|
|||
mock_user_datastore = MockUserDatastore(
|
||||
lambda: True,
|
||||
None,
|
||||
lambda _: UserCreds(USERNAME, PASSWORD_HASH),
|
||||
lambda _: UserCredentials(USERNAME, PASSWORD_HASH),
|
||||
)
|
||||
|
||||
a_s = AuthenticationService()
|
||||
a_s.initialize(tmp_path, mock_user_datastore)
|
||||
a_s = AuthenticationService(tmp_path, mock_user_datastore)
|
||||
|
||||
# If authentication fails, this function will raise an exception and the test will fail.
|
||||
a_s.authenticate(USERNAME, PASSWORD)
|
||||
|
@ -162,11 +155,10 @@ def test_authenticate__failed_wrong_credentials(
|
|||
mock_user_datastore = MockUserDatastore(
|
||||
lambda: True,
|
||||
None,
|
||||
lambda _: UserCreds(USERNAME, PASSWORD_HASH),
|
||||
lambda _: UserCredentials(USERNAME, PASSWORD_HASH),
|
||||
)
|
||||
|
||||
a_s = AuthenticationService()
|
||||
a_s.initialize(tmp_path, mock_user_datastore)
|
||||
a_s = AuthenticationService(tmp_path, mock_user_datastore)
|
||||
|
||||
with pytest.raises(IncorrectCredentialsError):
|
||||
a_s.authenticate(username, password)
|
||||
|
@ -179,8 +171,7 @@ def test_authenticate__failed_no_registered_user(tmp_path, mock_unlock_datastore
|
|||
lambda: True, None, MagicMock(side_effect=UnknownUserError)
|
||||
)
|
||||
|
||||
a_s = AuthenticationService()
|
||||
a_s.initialize(tmp_path, mock_user_datastore)
|
||||
a_s = AuthenticationService(tmp_path, mock_user_datastore)
|
||||
|
||||
with pytest.raises(IncorrectCredentialsError):
|
||||
a_s.authenticate(USERNAME, PASSWORD)
|
||||
|
|
Loading…
Reference in New Issue