Agent: Add public credentials property to CredentialsTelem

This commit is contained in:
Mike Salvatore 2022-03-28 09:45:45 -04:00 committed by Ilija Lazoroski
parent 5060ddb5d1
commit eb6342e2f8
2 changed files with 32 additions and 14 deletions

View File

@ -17,6 +17,10 @@ class CredentialsTelem(BaseTelem):
""" """
self._credentials = credentials self._credentials = credentials
@property
def credentials(self) -> Iterable[Credentials]:
return iter(self._credentials)
def send(self, log_data=True): def send(self, log_data=True):
super().send(log_data=False) super().send(log_data=False)

View File

@ -1,37 +1,51 @@
import json import json
import pytest
from infection_monkey.credential_collectors import Password, SSHKeypair, Username from infection_monkey.credential_collectors import Password, SSHKeypair, Username
from infection_monkey.i_puppet import Credentials from infection_monkey.i_puppet import Credentials
from infection_monkey.telemetry.credentials_telem import CredentialsTelem from infection_monkey.telemetry.credentials_telem import CredentialsTelem
USERNAME = "m0nkey"
PASSWORD = "mmm"
PUBLIC_KEY = "pub_key"
PRIVATE_KEY = "priv_key"
def test_credential_telem_send(spy_send_telemetry):
username = "m0nkey" @pytest.fixture
password = "mmm" def credentials_for_test():
public_key = "pub_key"
private_key = "priv_key" return Credentials(
[Username(USERNAME)], [Password(PASSWORD), SSHKeypair(PRIVATE_KEY, PUBLIC_KEY)]
)
def test_credential_telem_send(spy_send_telemetry, credentials_for_test):
expected_data = [ expected_data = [
{ {
"identities": [{"username": username, "credential_type": "USERNAME"}], "identities": [{"username": USERNAME, "credential_type": "USERNAME"}],
"secrets": [ "secrets": [
{"password": password, "credential_type": "PASSWORD"}, {"password": PASSWORD, "credential_type": "PASSWORD"},
{ {
"private_key": "pub_key", "private_key": PRIVATE_KEY,
"public_key": "priv_key", "public_key": PUBLIC_KEY,
"credential_type": "SSH_KEYPAIR", "credential_type": "SSH_KEYPAIR",
}, },
], ],
} }
] ]
credentials = Credentials( telem = CredentialsTelem([credentials_for_test])
[Username(username)], [Password(password), SSHKeypair(public_key, private_key)]
)
telem = CredentialsTelem([credentials])
telem.send() telem.send()
expected_data = json.dumps(expected_data, cls=telem.json_encoder) expected_data = json.dumps(expected_data, cls=telem.json_encoder)
assert spy_send_telemetry.data == expected_data assert spy_send_telemetry.data == expected_data
assert spy_send_telemetry.telem_category == "credentials" assert spy_send_telemetry.telem_category == "credentials"
def test_credentials_property(credentials_for_test):
telem = CredentialsTelem([credentials_for_test])
assert len(list(telem.credentials)) == 1
assert list(telem.credentials)[0] == credentials_for_test