p17860254
/
monkey
forked from p15670423/monkey
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Agent, Island: Use pydantic credentials and methods 

Since the interface of credential serialization changed, code was modified to use the new interface
This commit is contained in:
vakarisz 2022-09-01 14:10:54 +03:00 committed by vakaris_zilius
parent f018b85f56
commit d73cbee591
13 changed files with 92 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
envs/monkey_zoo/blackbox/island_client/monkey_island_client.py
View File

@ -30,7 +30,7 @@ class MonkeyIslandClient(object):
def get_propagation_credentials(self) -> Sequence[Credentials]: def get_propagation_credentials(self) -> Sequence[Credentials]:
response = self.requests.get("api/propagation-credentials") response = self.requests.get("api/propagation-credentials")
return [Credentials.from_mapping(credentials) for credentials in response.json()] return [Credentials(**credentials) for credentials in response.json()]
@avoid_race_condition @avoid_race_condition
def import_config(self, test_configuration: TestConfiguration): def import_config(self, test_configuration: TestConfiguration):
@ -61,7 +61,7 @@ class MonkeyIslandClient(object):
@avoid_race_condition @avoid_race_condition
def _import_credentials(self, propagation_credentials: Credentials): def _import_credentials(self, propagation_credentials: Credentials):
serialized_propagation_credentials = [ serialized_propagation_credentials = [
Credentials.to_mapping(credentials) for credentials in propagation_credentials Credentials.dict(credentials) for credentials in propagation_credentials
] ]
response = self.requests.put_json( response = self.requests.put_json(
"/api/propagation-credentials/configured-credentials", "/api/propagation-credentials/configured-credentials",

12
monkey/infection_monkey/credential_collectors/mimikatz_collector/mimikatz_credential_collector.py
View File

@ -55,22 +55,22 @@ class MimikatzCredentialCollector(ICredentialCollector):
identity = None identity = None
if wc.username: if wc.username:
identity = Username(wc.username) identity = Username(username=wc.username)
if wc.password: if wc.password:
password = Password(wc.password) password = Password(password=wc.password)
credentials.append(Credentials(identity, password)) credentials.append(Credentials(identity=identity, secret=password))
if wc.lm_hash: if wc.lm_hash:
lm_hash = LMHash(lm_hash=wc.lm_hash) lm_hash = LMHash(lm_hash=wc.lm_hash)
credentials.append(Credentials(identity, lm_hash)) credentials.append(Credentials(identity=identity, secret=lm_hash))
if wc.ntlm_hash: if wc.ntlm_hash:
ntlm_hash = NTHash(nt_hash=wc.ntlm_hash) ntlm_hash = NTHash(nt_hash=wc.ntlm_hash)
credentials.append(Credentials(identity, ntlm_hash)) credentials.append(Credentials(identity=identity, secret=ntlm_hash))
if len(credentials) == 0 and identity is not None: if len(credentials) == 0 and identity is not None:
credentials.append(Credentials(identity, None)) credentials.append(Credentials(identity=identity, secret=None))
return credentials return credentials

7
monkey/infection_monkey/credential_collectors/ssh_collector/ssh_handler.py
View File

@ -149,7 +149,7 @@ def to_credentials(ssh_info: Iterable[Dict]) -> Sequence[Credentials]:
secret = None secret = None
if info.get("name", ""): if info.get("name", ""):
identity = Username(info["name"]) identity = Username(username=info["name"])
ssh_keypair = {} ssh_keypair = {}
for key in ["public_key", "private_key"]: for key in ["public_key", "private_key"]:
@ -158,11 +158,12 @@ def to_credentials(ssh_info: Iterable[Dict]) -> Sequence[Credentials]:
if len(ssh_keypair): if len(ssh_keypair):
secret = SSHKeypair( secret = SSHKeypair(
ssh_keypair.get("private_key", ""), ssh_keypair.get("public_key", "") private_key=ssh_keypair.get("private_key", ""),
public_key=ssh_keypair.get("public_key", ""),
) )
if any([identity, secret]): if any([identity, secret]):
ssh_credentials.append(Credentials(identity, secret)) ssh_credentials.append(Credentials(identity=identity, secret=secret))
return ssh_credentials return ssh_credentials

3
monkey/infection_monkey/telemetry/credentials_telem.py
View File

@ -1,4 +1,3 @@
import json
from typing import Iterable from typing import Iterable
from common.common_consts.telem_categories import TelemCategoryEnum from common.common_consts.telem_categories import TelemCategoryEnum
@ -24,4 +23,4 @@ class CredentialsTelem(BaseTelem):
super().send(log_data=False) super().send(log_data=False)
def get_data(self): def get_data(self):
return [json.loads(Credentials.to_json(c)) for c in self._credentials] return [c.dict(simplify=True) for c in self._credentials]

4
monkey/monkey_island/cc/repository/mongo_credentials_repository.py
View File

@ -59,7 +59,7 @@ class MongoCredentialsRepository(ICredentialsRepository):
for encrypted_credentials in list_collection_result: for encrypted_credentials in list_collection_result:
del encrypted_credentials[MONGO_OBJECT_ID_KEY] del encrypted_credentials[MONGO_OBJECT_ID_KEY]
plaintext_credentials = self._decrypt_credentials_mapping(encrypted_credentials) plaintext_credentials = self._decrypt_credentials_mapping(encrypted_credentials)
collection_result.append(Credentials.from_mapping(plaintext_credentials)) collection_result.append(Credentials(**plaintext_credentials))
return collection_result return collection_result
except Exception as err: except Exception as err:
@ -68,7 +68,7 @@ class MongoCredentialsRepository(ICredentialsRepository):
def _save_credentials_to_collection(self, credentials: Sequence[Credentials], collection): def _save_credentials_to_collection(self, credentials: Sequence[Credentials], collection):
try: try:
for c in credentials: for c in credentials:
encrypted_credentials = self._encrypt_credentials_mapping(Credentials.to_mapping(c)) encrypted_credentials = self._encrypt_credentials_mapping(c.dict())
collection.insert_one(encrypted_credentials) collection.insert_one(encrypted_credentials)
except Exception as err: except Exception as err:
raise StorageError(err) raise StorageError(err)

2
monkey/monkey_island/cc/resources/propagation_credentials.py
View File

@ -29,7 +29,7 @@ class PropagationCredentials(AbstractResource):
return propagation_credentials, HTTPStatus.OK return propagation_credentials, HTTPStatus.OK
def put(self, collection=None): def put(self, collection=None):
credentials = [Credentials.from_mapping(c) for c in request.json] credentials = [Credentials.parse_raw(c) for c in request.json]
if collection == _configured_collection: if collection == _configured_collection:
self._credentials_repository.remove_configured_credentials() self._credentials_repository.remove_configured_credentials()
self._credentials_repository.save_configured_credentials(credentials) self._credentials_repository.save_configured_credentials(credentials)

28
monkey/tests/unit_tests/infection_monkey/credential_collectors/test_mimikatz_collector.py
View File

@ -41,9 +41,9 @@ def test_pypykatz_result_parsing(monkeypatch):
win_creds = [WindowsCredentials(username="user", password="secret", ntlm_hash="", lm_hash="")] win_creds = [WindowsCredentials(username="user", password="secret", ntlm_hash="", lm_hash="")]
patch_pypykatz(win_creds, monkeypatch) patch_pypykatz(win_creds, monkeypatch)
username = Username("user") username = Username(username="user")
password = Password("secret") password = Password(password="secret")
expected_credentials = Credentials(username, password) expected_credentials = Credentials(identity=username, secret=password)
collected_credentials = collect_credentials() collected_credentials = collect_credentials()
assert len(collected_credentials) == 1 assert len(collected_credentials) == 1
@ -70,10 +70,13 @@ def test_pypykatz_result_parsing_defaults(monkeypatch):
patch_pypykatz(win_creds, monkeypatch) patch_pypykatz(win_creds, monkeypatch)
# Expected credentials # Expected credentials
username = Username("user2") username = Username(username="user2")
password = Password("secret2") password = Password(password="secret2")
lm_hash = LMHash("0182BD0BD4444BF8FC83B5D9042EED2E") lm_hash = LMHash(lm_hash="0182BD0BD4444BF8FC83B5D9042EED2E")
expected_credentials = [Credentials(username, password), Credentials(username, lm_hash)] expected_credentials = [
Credentials(identity=username, secret=password),
Credentials(identity=username, secret=lm_hash),
]
collected_credentials = collect_credentials() collected_credentials = collect_credentials()
assert len(collected_credentials) == 2 assert len(collected_credentials) == 2
@ -91,9 +94,12 @@ def test_pypykatz_result_parsing_no_identities(monkeypatch):
] ]
patch_pypykatz(win_creds, monkeypatch) patch_pypykatz(win_creds, monkeypatch)
lm_hash = LMHash("0182BD0BD4444BF8FC83B5D9042EED2E") lm_hash = LMHash(lm_hash="0182BD0BD4444BF8FC83B5D9042EED2E")
nt_hash = NTHash("E9F85516721DDC218359AD5280DB4450") nt_hash = NTHash(nt_hash="E9F85516721DDC218359AD5280DB4450")
expected_credentials = [Credentials(None, lm_hash), Credentials(None, nt_hash)] expected_credentials = [
Credentials(identity=None, secret=lm_hash),
Credentials(identity=None, secret=nt_hash),
]
collected_credentials = collect_credentials() collected_credentials = collect_credentials()
assert len(collected_credentials) == 2 assert len(collected_credentials) == 2
@ -112,7 +118,7 @@ def test_pypykatz_result_parsing_no_secrets(monkeypatch):
] ]
patch_pypykatz(win_creds, monkeypatch) patch_pypykatz(win_creds, monkeypatch)
expected_credentials = [Credentials(Username(username), None)] expected_credentials = [Credentials(identity=Username(username=username), secret=None)]
collected_credentials = collect_credentials() collected_credentials = collect_credentials()
assert len(collected_credentials) == 1 assert len(collected_credentials) == 1

15
monkey/tests/unit_tests/infection_monkey/credential_collectors/test_ssh_credentials_collector.py
View File

@ -31,7 +31,6 @@ def test_ssh_credentials_empty_results(monkeypatch, ssh_creds, patch_telemetry_m
def test_ssh_info_result_parsing(monkeypatch, patch_telemetry_messenger): def test_ssh_info_result_parsing(monkeypatch, patch_telemetry_messenger):
ssh_creds = [ ssh_creds = [
{ {
"name": "ubuntu", "name": "ubuntu",
@ -56,13 +55,15 @@ def test_ssh_info_result_parsing(monkeypatch, patch_telemetry_messenger):
patch_ssh_handler(ssh_creds, monkeypatch) patch_ssh_handler(ssh_creds, monkeypatch)
# Expected credentials # Expected credentials
username = Username("ubuntu") username = Username(username="ubuntu")
username2 = Username("mcus") username2 = Username(username="mcus")
username3 = Username("guest") username3 = Username(username="guest")
ssh_keypair1 = SSHKeypair("ExtremelyGoodPrivateKey", "SomePublicKeyUbuntu") ssh_keypair1 = SSHKeypair(
ssh_keypair2 = SSHKeypair("", "AnotherPublicKey") private_key="ExtremelyGoodPrivateKey", public_key="SomePublicKeyUbuntu"
ssh_keypair3 = SSHKeypair("PrivKey", "PubKey") )
ssh_keypair2 = SSHKeypair(private_key="", public_key="AnotherPublicKey")
ssh_keypair3 = SSHKeypair(private_key="PrivKey", public_key="PubKey")
expected = [ expected = [
Credentials(identity=username, secret=ssh_keypair1), Credentials(identity=username, secret=ssh_keypair1),

6
monkey/tests/unit_tests/infection_monkey/credential_store/test_add_credentials_from_event.py
View File

@ -8,7 +8,11 @@ from infection_monkey.credential_repository import (
add_credentials_from_event_to_propagation_credentials_repository, add_credentials_from_event_to_propagation_credentials_repository,
) )
credentials = [Credentials(identity=Username("test_username"), secret=Password("some_password"))] credentials = [
Credentials(
identity=Username(username="test_username"), secret=Password(password="some_password")
)
]
credentials_stolen_event = CredentialsStolenEvent( credentials_stolen_event = CredentialsStolenEvent(
source=UUID("f811ad00-5a68-4437-bd51-7b5cc1768ad5"), source=UUID("f811ad00-5a68-4437-bd51-7b5cc1768ad5"),

29
monkey/tests/unit_tests/infection_monkey/credential_store/test_aggregating_propagation_credentials_repository.py
View File

@ -2,13 +2,13 @@ from unittest.mock import MagicMock
import pytest import pytest
from tests.data_for_tests.propagation_credentials import ( from tests.data_for_tests.propagation_credentials import (
CREDENTIALS,
LM_HASH, LM_HASH,
NT_HASH, NT_HASH,
PASSWORD_1, PASSWORD_1,
PASSWORD_2, PASSWORD_2,
PASSWORD_3, PASSWORD_3,
PRIVATE_KEY, PRIVATE_KEY,
PROPAGATION_CREDENTIALS,
PUBLIC_KEY, PUBLIC_KEY,
SPECIAL_USERNAME, SPECIAL_USERNAME,
USERNAME, USERNAME,
@ -17,7 +17,6 @@ from tests.data_for_tests.propagation_credentials import (
from common.credentials import Credentials, LMHash, NTHash, Password, SSHKeypair, Username from common.credentials import Credentials, LMHash, NTHash, Password, SSHKeypair, Username
from infection_monkey.credential_repository import AggregatingPropagationCredentialsRepository from infection_monkey.credential_repository import AggregatingPropagationCredentialsRepository
CONTROL_CHANNEL_CREDENTIALS = PROPAGATION_CREDENTIALS
TRANSFORMED_CONTROL_CHANNEL_CREDENTIALS = { TRANSFORMED_CONTROL_CHANNEL_CREDENTIALS = {
"exploit_user_list": {USERNAME, SPECIAL_USERNAME}, "exploit_user_list": {USERNAME, SPECIAL_USERNAME},
"exploit_password_list": {PASSWORD_1, PASSWORD_2, PASSWORD_3}, "exploit_password_list": {PASSWORD_1, PASSWORD_2, PASSWORD_3},
@ -41,26 +40,32 @@ STOLEN_PRIVATE_KEY_1 = "some_private_key_1"
STOLEN_PRIVATE_KEY_2 = "some_private_key_2" STOLEN_PRIVATE_KEY_2 = "some_private_key_2"
STOLEN_CREDENTIALS = [ STOLEN_CREDENTIALS = [
Credentials( Credentials(
identity=Username(STOLEN_USERNAME_1), identity=Username(username=STOLEN_USERNAME_1),
secret=Password(PASSWORD_1), secret=Password(password=PASSWORD_1),
), ),
Credentials(identity=Username(STOLEN_USERNAME_1), secret=Password(STOLEN_PASSWORD_1)),
Credentials( Credentials(
identity=Username(STOLEN_USERNAME_2), identity=Username(username=STOLEN_USERNAME_1), secret=Password(password=STOLEN_PASSWORD_1)
),
Credentials(
identity=Username(username=STOLEN_USERNAME_2),
secret=SSHKeypair(public_key=STOLEN_PUBLIC_KEY_1, private_key=STOLEN_PRIVATE_KEY_1), secret=SSHKeypair(public_key=STOLEN_PUBLIC_KEY_1, private_key=STOLEN_PRIVATE_KEY_1),
), ),
Credentials( Credentials(
identity=None, identity=None,
secret=Password(STOLEN_PASSWORD_2), secret=Password(password=STOLEN_PASSWORD_2),
), ),
Credentials(identity=Username(STOLEN_USERNAME_2), secret=LMHash(STOLEN_LM_HASH)), Credentials(
Credentials(identity=Username(STOLEN_USERNAME_2), secret=NTHash(STOLEN_NT_HASH)), identity=Username(username=STOLEN_USERNAME_2), secret=LMHash(lm_hash=STOLEN_LM_HASH)
Credentials(identity=Username(STOLEN_USERNAME_3), secret=None), ),
Credentials(
identity=Username(username=STOLEN_USERNAME_2), secret=NTHash(nt_hash=STOLEN_NT_HASH)
),
Credentials(identity=Username(username=STOLEN_USERNAME_3), secret=None),
] ]
STOLEN_SSH_KEYS_CREDENTIALS = [ STOLEN_SSH_KEYS_CREDENTIALS = [
Credentials( Credentials(
identity=Username(USERNAME), identity=Username(username=USERNAME),
secret=SSHKeypair(public_key=STOLEN_PUBLIC_KEY_2, private_key=STOLEN_PRIVATE_KEY_2), secret=SSHKeypair(public_key=STOLEN_PUBLIC_KEY_2, private_key=STOLEN_PRIVATE_KEY_2),
) )
] ]
@ -69,7 +74,7 @@ STOLEN_SSH_KEYS_CREDENTIALS = [
@pytest.fixture @pytest.fixture
def aggregating_credentials_repository() -> AggregatingPropagationCredentialsRepository: def aggregating_credentials_repository() -> AggregatingPropagationCredentialsRepository:
control_channel = MagicMock() control_channel = MagicMock()
control_channel.get_credentials_for_propagation.return_value = CONTROL_CHANNEL_CREDENTIALS control_channel.get_credentials_for_propagation.return_value = CREDENTIALS
return AggregatingPropagationCredentialsRepository(control_channel) return AggregatingPropagationCredentialsRepository(control_channel)

5
monkey/tests/unit_tests/infection_monkey/telemetry/test_credentials_telem.py
View File

@ -13,13 +13,12 @@ PRIVATE_KEY = "priv_key"
@pytest.fixture @pytest.fixture
def credentials_for_test(): def credentials_for_test():
return Credentials(identity=Username(username=USERNAME), secret=Password(password=PASSWORD))
return Credentials(Username(USERNAME), Password(PASSWORD))
def test_credential_telem_send(spy_send_telemetry, credentials_for_test): def test_credential_telem_send(spy_send_telemetry, credentials_for_test):
expected_data = [Credentials.to_mapping(credentials_for_test)] expected_data = [credentials_for_test.dict(simplify=True)]
telem = CredentialsTelem([credentials_for_test]) telem = CredentialsTelem([credentials_for_test])
telem.send() telem.send()

18
monkey/tests/unit_tests/monkey_island/cc/repository/test_mongo_credentials_repository.py
View File

@ -6,14 +6,14 @@ import pytest
from pymongo import MongoClient from pymongo import MongoClient
from pymongo.collection import Collection from pymongo.collection import Collection
from pymongo.database import Database from pymongo.database import Database
from tests.data_for_tests.propagation_credentials import PROPAGATION_CREDENTIALS from tests.data_for_tests.propagation_credentials import CREDENTIALS
from common.credentials import Credentials from common.credentials import Credentials
from monkey_island.cc.repository import MongoCredentialsRepository from monkey_island.cc.repository import MongoCredentialsRepository
from monkey_island.cc.server_utils.encryption import ILockableEncryptor from monkey_island.cc.server_utils.encryption import ILockableEncryptor
CONFIGURED_CREDENTIALS = PROPAGATION_CREDENTIALS[0:3] CONFIGURED_CREDENTIALS = CREDENTIALS[0:3]
STOLEN_CREDENTIALS = PROPAGATION_CREDENTIALS[3:] STOLEN_CREDENTIALS = CREDENTIALS[3:]
def reverse(data: bytes) -> bytes: def reverse(data: bytes) -> bytes:
@ -59,9 +59,9 @@ def test_mongo_repository_get_all(mongo_repository):
def test_mongo_repository_configured(mongo_repository): def test_mongo_repository_configured(mongo_repository):
mongo_repository.save_configured_credentials(PROPAGATION_CREDENTIALS) mongo_repository.save_configured_credentials(CREDENTIALS)
actual_configured_credentials = mongo_repository.get_configured_credentials() actual_configured_credentials = mongo_repository.get_configured_credentials()
assert actual_configured_credentials == PROPAGATION_CREDENTIALS assert actual_configured_credentials == CREDENTIALS
mongo_repository.remove_configured_credentials() mongo_repository.remove_configured_credentials()
actual_configured_credentials = mongo_repository.get_configured_credentials() actual_configured_credentials = mongo_repository.get_configured_credentials()
@ -82,7 +82,7 @@ def test_mongo_repository_all(mongo_repository):
mongo_repository.save_configured_credentials(CONFIGURED_CREDENTIALS) mongo_repository.save_configured_credentials(CONFIGURED_CREDENTIALS)
mongo_repository.save_stolen_credentials(STOLEN_CREDENTIALS) mongo_repository.save_stolen_credentials(STOLEN_CREDENTIALS)
actual_credentials = mongo_repository.get_all_credentials() actual_credentials = mongo_repository.get_all_credentials()
assert actual_credentials == PROPAGATION_CREDENTIALS assert actual_credentials == CREDENTIALS
mongo_repository.remove_all_credentials() mongo_repository.remove_all_credentials()
@ -91,7 +91,7 @@ def test_mongo_repository_all(mongo_repository):
assert mongo_repository.get_configured_credentials() == [] assert mongo_repository.get_configured_credentials() == []
@pytest.mark.parametrize("credentials", PROPAGATION_CREDENTIALS) @pytest.mark.parametrize("credentials", CREDENTIALS)
def test_configured_secrets_encrypted( def test_configured_secrets_encrypted(
mongo_repository: MongoCredentialsRepository, mongo_repository: MongoCredentialsRepository,
mongo_client: MongoClient, mongo_client: MongoClient,
@ -101,14 +101,14 @@ def test_configured_secrets_encrypted(
check_if_stored_credentials_encrypted(mongo_client, credentials) check_if_stored_credentials_encrypted(mongo_client, credentials)
@pytest.mark.parametrize("credentials", PROPAGATION_CREDENTIALS) @pytest.mark.parametrize("credentials", CREDENTIALS)
def test_stolen_secrets_encrypted(mongo_repository, mongo_client, credentials: Credentials): def test_stolen_secrets_encrypted(mongo_repository, mongo_client, credentials: Credentials):
mongo_repository.save_stolen_credentials([credentials]) mongo_repository.save_stolen_credentials([credentials])
check_if_stored_credentials_encrypted(mongo_client, credentials) check_if_stored_credentials_encrypted(mongo_client, credentials)
def check_if_stored_credentials_encrypted(mongo_client: MongoClient, original_credentials): def check_if_stored_credentials_encrypted(mongo_client: MongoClient, original_credentials):
original_credentials_mapping = Credentials.to_mapping(original_credentials) original_credentials_mapping = original_credentials.dict()
raw_credentials = get_all_credentials_in_mongo(mongo_client) raw_credentials = get_all_credentials_in_mongo(mongo_client)
for rc in raw_credentials: for rc in raw_credentials:

41
monkey/tests/unit_tests/monkey_island/cc/resources/test_propagation_credentials.py
View File

@ -5,15 +5,10 @@ from urllib.parse import urljoin
import pytest import pytest
from tests.common import StubDIContainer from tests.common import StubDIContainer
from tests.data_for_tests.propagation_credentials import ( from tests.data_for_tests.propagation_credentials import LM_HASH, NT_HASH, PASSWORD_1, PASSWORD_2
LM_HASH_CREDENTIALS,
NT_HASH_CREDENTIALS,
PASSWORD_CREDENTIALS_1,
PASSWORD_CREDENTIALS_2,
)
from tests.monkey_island import InMemoryCredentialsRepository from tests.monkey_island import InMemoryCredentialsRepository
from common.credentials import Credentials from common.credentials import Credentials, LMHash, NTHash, Password
from monkey_island.cc.repository import ICredentialsRepository from monkey_island.cc.repository import ICredentialsRepository
from monkey_island.cc.resources import PropagationCredentials from monkey_island.cc.resources import PropagationCredentials
from monkey_island.cc.resources.propagation_credentials import ( from monkey_island.cc.resources.propagation_credentials import (
@ -24,6 +19,10 @@ from monkey_island.cc.resources.propagation_credentials import (
ALL_CREDENTIALS_URL = PropagationCredentials.urls[0] ALL_CREDENTIALS_URL = PropagationCredentials.urls[0]
CONFIGURED_CREDENTIALS_URL = urljoin(ALL_CREDENTIALS_URL + "/", _configured_collection) CONFIGURED_CREDENTIALS_URL = urljoin(ALL_CREDENTIALS_URL + "/", _configured_collection)
STOLEN_CREDENTIALS_URL = urljoin(ALL_CREDENTIALS_URL + "/", _stolen_collection) STOLEN_CREDENTIALS_URL = urljoin(ALL_CREDENTIALS_URL + "/", _stolen_collection)
CREDENTIALS_1 = Credentials(identity=None, secret=Password(password=PASSWORD_1))
CREDENTIALS_2 = Credentials(identity=None, secret=LMHash(lm_hash=LM_HASH))
CREDENTIALS_3 = Credentials(identity=None, secret=NTHash(nt_hash=NT_HASH))
CREDENTIALS_4 = Credentials(identity=None, secret=Password(password=PASSWORD_2))
@pytest.fixture @pytest.fixture
@ -42,20 +41,18 @@ def flask_client(build_flask_client, credentials_repository):
def test_propagation_credentials_endpoint_get(flask_client, credentials_repository): def test_propagation_credentials_endpoint_get(flask_client, credentials_repository):
credentials_repository.save_configured_credentials( credentials_repository.save_configured_credentials([CREDENTIALS_1, CREDENTIALS_2])
[PASSWORD_CREDENTIALS_1, NT_HASH_CREDENTIALS] credentials_repository.save_stolen_credentials([CREDENTIALS_3, CREDENTIALS_4])
)
credentials_repository.save_stolen_credentials([LM_HASH_CREDENTIALS, PASSWORD_CREDENTIALS_2])
resp = flask_client.get(ALL_CREDENTIALS_URL) resp = flask_client.get(ALL_CREDENTIALS_URL)
actual_propagation_credentials = [Credentials.from_mapping(creds) for creds in resp.json] actual_propagation_credentials = [Credentials(**creds) for creds in resp.json]
assert resp.status_code == HTTPStatus.OK assert resp.status_code == HTTPStatus.OK
assert len(actual_propagation_credentials) == 4 assert len(actual_propagation_credentials) == 4
assert PASSWORD_CREDENTIALS_1 in actual_propagation_credentials assert CREDENTIALS_1 in actual_propagation_credentials
assert LM_HASH_CREDENTIALS in actual_propagation_credentials assert CREDENTIALS_2 in actual_propagation_credentials
assert NT_HASH_CREDENTIALS in actual_propagation_credentials assert CREDENTIALS_3 in actual_propagation_credentials
assert PASSWORD_CREDENTIALS_2 in actual_propagation_credentials assert CREDENTIALS_4 in actual_propagation_credentials
def pre_populate_repository( def pre_populate_repository(
@ -69,24 +66,22 @@ def pre_populate_repository(
@pytest.mark.parametrize("url", [CONFIGURED_CREDENTIALS_URL, STOLEN_CREDENTIALS_URL]) @pytest.mark.parametrize("url", [CONFIGURED_CREDENTIALS_URL, STOLEN_CREDENTIALS_URL])
def test_propagation_credentials_endpoint__get_stolen(flask_client, credentials_repository, url): def test_propagation_credentials_endpoint__get_stolen(flask_client, credentials_repository, url):
pre_populate_repository( pre_populate_repository(url, credentials_repository, [CREDENTIALS_1, CREDENTIALS_2])
url, credentials_repository, [PASSWORD_CREDENTIALS_1, LM_HASH_CREDENTIALS]
)
resp = flask_client.get(url) resp = flask_client.get(url)
actual_propagation_credentials = [Credentials.from_mapping(creds) for creds in resp.json] actual_propagation_credentials = [Credentials(**creds) for creds in resp.json]
assert resp.status_code == HTTPStatus.OK assert resp.status_code == HTTPStatus.OK
assert len(actual_propagation_credentials) == 2 assert len(actual_propagation_credentials) == 2
assert actual_propagation_credentials[0] == PASSWORD_CREDENTIALS_1 assert actual_propagation_credentials[0].secret.password == PASSWORD_1
assert actual_propagation_credentials[1] == LM_HASH_CREDENTIALS assert actual_propagation_credentials[1].secret.lm_hash == LM_HASH
def test_configured_propagation_credentials_endpoint_put(flask_client, credentials_repository): def test_configured_propagation_credentials_endpoint_put(flask_client, credentials_repository):
pre_populate_repository( pre_populate_repository(
CONFIGURED_CREDENTIALS_URL, CONFIGURED_CREDENTIALS_URL,
credentials_repository, credentials_repository,
[PASSWORD_CREDENTIALS_1, LM_HASH_CREDENTIALS], [CREDENTIALS_1, CREDENTIALS_2],
) )
resp = flask_client.put(CONFIGURED_CREDENTIALS_URL, json=[]) resp = flask_client.put(CONFIGURED_CREDENTIALS_URL, json=[])
assert resp.status_code == HTTPStatus.NO_CONTENT assert resp.status_code == HTTPStatus.NO_CONTENT