Agent: Separate AuthOptions from Credentials

monkey/infection_monkey/exploit/powershell.py

@@ -14,7 +14,9 @@ from infection_monkey.exploit.consts import WIN_ARCH_32, WIN_ARCH_64
 from infection_monkey.exploit.HostExploiter import HostExploiter
 from infection_monkey.exploit.powershell_utils import utils
 from infection_monkey.exploit.powershell_utils.auth_options import AuthOptions
+from infection_monkey.exploit.powershell_utils.auth_options_generators import get_auth_options
 from infection_monkey.exploit.powershell_utils.credential_generators import get_credentials
+from infection_monkey.exploit.powershell_utils.credentials import Credentials
 from infection_monkey.exploit.powershell_utils.utils import (
     IClient,
     get_client_based_on_auth_options,
@@ -58,13 +60,11 @@ class PowerShellExploiter(HostExploiter):
             return False
 
         credentials = get_credentials(
-            self._config.exploit_user_list,
+            self._config.exploit_user_list, self._config.exploit_password_list, is_windows_os()
-            self._config.exploit_password_list,
-            is_windows_os(),
-            is_https=is_https,
         )
+        auth_options = get_auth_options(credentials, is_https)
 
-        self.client = self._authenticate_via_brute_force(credentials)
+        self.client = self._authenticate_via_brute_force(credentials, auth_options)
         if not self.client:
             return False
 
@@ -91,44 +91,50 @@ class PowerShellExploiter(HostExploiter):
             raise PowerShellRemotingDisabledError("Powershell remoting seems to be disabled.")
 
     def _try_http(self):
-        auth_options_http = AuthOptions(
+        credentials = Credentials(
             username=self._config.exploit_user_list[0],
             password=self._config.exploit_password_list[0],
-            is_https=False,
         )
-        self._authenticate(auth_options_http)
+        auth_options = AuthOptions(
+            ssl=False,
+        )
+        self._authenticate(credentials, auth_options)
 
     def _try_https(self):
-        auth_options_http = AuthOptions(
+        credentials = Credentials(
             username=self._config.exploit_user_list[0],
             password=self._config.exploit_password_list[0],
-            is_https=True,
         )
-        self._authenticate(auth_options_http)
+        auth_options = AuthOptions(
+            ssl=True,
+        )
+        self._authenticate(credentials, auth_options)
 
-    def _authenticate_via_brute_force(self, credentials: [AuthOptions]) -> Optional[IClient]:
+    def _authenticate_via_brute_force(
-        for credential in credentials:
+        self, credentials: [Credentials], auth_options: [AuthOptions]
+    ) -> Optional[IClient]:
+        for (creds, opts) in zip(credentials, auth_options):
             try:
-                client = self._authenticate(credential)
+                client = self._authenticate(creds, opts)
 
                 LOG.info(
                     f"Successfully logged into {self.host.ip_addr} using Powershell. User: "
-                    f"{credential.username}"
+                    f"{creds.username}"
                 )
-                self.report_login_attempt(True, credential.username, credential.password)
+                self.report_login_attempt(True, creds.username, creds.password)
 
                 return client
             except Exception as ex:  # noqa: F841
                 LOG.debug(
                     f"Error logging into {self.host.ip_addr} using Powershell. User: "
-                    f"{credential.username}, Error: {ex}"
+                    f"{creds.username}, Error: {ex}"
                 )
-                self.report_login_attempt(False, credential.username, credential.password)
+                self.report_login_attempt(False, creds.username, creds.password)
 
         return None
 
-    def _authenticate(self, auth_options: AuthOptions) -> IClient:
+    def _authenticate(self, credentials: Credentials, auth_options: AuthOptions) -> IClient:
-        client = get_client_based_on_auth_options(self.host.ip_addr, auth_options)
+        client = get_client_based_on_auth_options(self.host.ip_addr, credentials, auth_options)
 
         # attempt to execute dir command to know if authentication was successful
         client.execute_cmd("dir")

monkey/infection_monkey/exploit/powershell_utils/auth_options.py

@@ -1,9 +1,6 @@
 from dataclasses import dataclass
-from typing import Union
 
 
 @dataclass
 class AuthOptions:
-    username: Union[str, None]
+    ssl: bool
-    password: Union[str, None]
-    is_https: bool

monkey/infection_monkey/exploit/powershell_utils/auth_options_generators.py

@@ -0,0 +1,19 @@
+from typing import List
+
+from infection_monkey.exploit.powershell_utils.auth_options import AuthOptions
+from infection_monkey.exploit.powershell_utils.credentials import Credentials
+
+
+def get_auth_options(credentials: List[Credentials], ssl: bool) -> List[AuthOptions]:
+    auth_options = []
+
+    for cred in credentials:
+        opts = AuthOptions(ssl)
+
+        # Passwordless login only works with SSL false
+        if cred.password == "":
+            opts.ssl = False
+
+        auth_options.append(opts)
+
+    return auth_options

monkey/infection_monkey/exploit/powershell_utils/credential_generators.py

@@ -1,46 +1,41 @@
 from itertools import product
 from typing import List
 
-from infection_monkey.exploit.powershell_utils.auth_options import AuthOptions
+from infection_monkey.exploit.powershell_utils.credentials import Credentials
 
 
 def get_credentials(
-    usernames: List[str], passwords: List[str], is_windows: bool, is_https: bool
+    usernames: List[str], passwords: List[str], is_windows: bool
-) -> List[AuthOptions]:
+) -> List[Credentials]:
     credentials = []
     credentials.extend(_get_empty_credentials(is_windows))
     credentials.extend(_get_username_only_credentials(usernames, is_windows))
-    credentials.extend(_get_username_password_credentials(usernames, passwords, is_https=is_https))
+    credentials.extend(_get_username_password_credentials(usernames, passwords))
 
     return credentials
 
 
-def _get_empty_credentials(is_windows: bool) -> List[AuthOptions]:
+def _get_empty_credentials(is_windows: bool) -> List[Credentials]:
     if is_windows:
-        return [AuthOptions(username=None, password=None, is_https=False)]
+        return [Credentials(username=None, password=None)]
 
     return []
 
 
-def _get_username_only_credentials(usernames: List[str], is_windows: bool) -> List[AuthOptions]:
+def _get_username_only_credentials(usernames: List[str], is_windows: bool) -> List[Credentials]:
-    credentials = [
+    credentials = [Credentials(username=username, password="") for username in usernames]
-        AuthOptions(username=username, password="", is_https=False) for username in usernames
-    ]
 
     if is_windows:
         credentials.extend(
-            [AuthOptions(username=username, password=None, is_https=True) for username in usernames]
+            [Credentials(username=username, password=None) for username in usernames]
         )
 
     return credentials
 
 
 def _get_username_password_credentials(
-    usernames: List[str], passwords: List[str], is_https: bool
+    usernames: List[str], passwords: List[str]
-) -> List[AuthOptions]:
+) -> List[Credentials]:
     username_password_pairs = product(usernames, passwords)
 
-    return [
+    return [Credentials(credentials[0], credentials[1]) for credentials in username_password_pairs]
-        AuthOptions(credentials[0], credentials[1], is_https=is_https)
-        for credentials in username_password_pairs
-    ]

monkey/infection_monkey/exploit/powershell_utils/credentials.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+from typing import Union
+
+
+@dataclass
+class Credentials:
+    username: Union[str, None]
+    password: Union[str, None]

monkey/infection_monkey/exploit/powershell_utils/utils.py

@@ -2,6 +2,7 @@ from pypsrp.client import Client
 from typing_extensions import Protocol
 
 from infection_monkey.exploit.powershell_utils.auth_options import AuthOptions
+from infection_monkey.exploit.powershell_utils.credentials import Credentials
 from infection_monkey.model import DROPPER_ARG, RUN_MONKEY, VictimHost
 from infection_monkey.utils.commands import build_monkey_commandline
 
@@ -34,22 +35,19 @@ class IClient(Protocol):
         pass
 
 
-def get_client_based_on_auth_options(ip_addr: str, auth_options: AuthOptions) -> IClient:
+def get_client_based_on_auth_options(
-
+    ip_addr: str, credentials: Credentials, auth_options: AuthOptions
+) -> IClient:
     # Passwordless login only works with SSL false, AUTH_BASIC and ENCRYPTION_NEVER
-    if auth_options.password == "":
+    auth = AUTH_NEGOTIATE if credentials.password != "" else AUTH_BASIC
-        ssl = False
+    encryption = ENCRYPTION_AUTO if credentials.password != "" else ENCRYPTION_NEVER
-    else:
-        ssl = auth_options.is_https
-    auth = AUTH_NEGOTIATE if auth_options.password != "" else AUTH_BASIC
-    encryption = ENCRYPTION_AUTO if auth_options.password != "" else ENCRYPTION_NEVER
 
     return Client(
         ip_addr,
-        username=auth_options.username,
+        username=credentials.username,
-        password=auth_options.password,
+        password=credentials.password,
         cert_validation=False,
-        ssl=ssl,
+        ssl=auth_options.ssl,
         auth=auth,
         encryption=encryption,
         connection_timeout=CONNECTION_TIMEOUT,

monkey/tests/unit_tests/infection_monkey/exploit/powershell_utils/test_auth_options_generators.py

@@ -0,0 +1,45 @@
+# from infection_monkey.exploit.powershell_utils.auth_options import AuthOptions
+from infection_monkey.exploit.powershell_utils.auth_options_generators import get_auth_options
+from infection_monkey.exploit.powershell_utils.credentials import Credentials
+
+CREDENTIALS = [
+    Credentials("user1", "password1"),
+    Credentials("user2", ""),
+    Credentials("user3", None),
+]
+
+
+def test_get_auth_options__ssl_true_with_password():
+    auth_options = get_auth_options(CREDENTIALS, ssl=True)
+
+    assert auth_options[0].ssl
+
+
+def test_get_auth_options__ssl_true_empty_password():
+    auth_options = get_auth_options(CREDENTIALS, ssl=True)
+
+    assert not auth_options[1].ssl
+
+
+def test_get_auth_options__ssl_true_none_password():
+    auth_options = get_auth_options(CREDENTIALS, ssl=True)
+
+    assert auth_options[2].ssl
+
+
+def test_get_auth_options__ssl_false_with_password():
+    auth_options = get_auth_options(CREDENTIALS, ssl=False)
+
+    assert not auth_options[0].ssl
+
+
+def test_get_auth_options__ssl_false_empty_password():
+    auth_options = get_auth_options(CREDENTIALS, ssl=False)
+
+    assert not auth_options[1].ssl
+
+
+def test_get_auth_options__ssl_false_none_password():
+    auth_options = get_auth_options(CREDENTIALS, ssl=False)
+
+    assert not auth_options[2].ssl

monkey/tests/unit_tests/infection_monkey/exploit/powershell_utils/test_credential_generators.py

@@ -1,45 +1,45 @@
-from infection_monkey.exploit.powershell_utils.auth_options import AuthOptions
 from infection_monkey.exploit.powershell_utils.credential_generators import get_credentials
+from infection_monkey.exploit.powershell_utils.credentials import Credentials
 
 TEST_USERNAMES = ["user1", "user2"]
 TEST_PASSWORDS = ["p1", "p2"]
 
 
 def test_get_credentials__empty_windows_true():
-    credentials = get_credentials([], [], True, True)
+    credentials = get_credentials([], [], True)
 
     assert len(credentials) == 1
-    assert credentials[0] == AuthOptions(username=None, password=None, is_https=False)
+    assert credentials[0] == Credentials(username=None, password=None)
 
 
 def test_get_credentials__empty_windows_false():
-    credentials = get_credentials([], [], False, True)
+    credentials = get_credentials([], [], False)
 
     assert len(credentials) == 0
 
 
 def test_get_credentials__username_only_windows_true():
-    credentials = get_credentials(TEST_USERNAMES, [], True, True)
+    credentials = get_credentials(TEST_USERNAMES, [], True)
 
     assert len(credentials) == 5
-    assert AuthOptions(username=TEST_USERNAMES[0], password="", is_https=False) in credentials
+    assert Credentials(username=TEST_USERNAMES[0], password="") in credentials
-    assert AuthOptions(username=TEST_USERNAMES[1], password="", is_https=False) in credentials
+    assert Credentials(username=TEST_USERNAMES[1], password="") in credentials
-    assert AuthOptions(username=TEST_USERNAMES[0], password=None, is_https=True) in credentials
+    assert Credentials(username=TEST_USERNAMES[0], password=None) in credentials
-    assert AuthOptions(username=TEST_USERNAMES[1], password=None, is_https=True) in credentials
+    assert Credentials(username=TEST_USERNAMES[1], password=None) in credentials
 
 
 def test_get_credentials__username_only_windows_false():
-    credentials = get_credentials(TEST_USERNAMES, [], False, True)
+    credentials = get_credentials(TEST_USERNAMES, [], False)
 
     assert len(credentials) == 2
-    assert AuthOptions(username=TEST_USERNAMES[0], password="", is_https=False) in credentials
+    assert Credentials(username=TEST_USERNAMES[0], password="") in credentials
-    assert AuthOptions(username=TEST_USERNAMES[1], password="", is_https=False) in credentials
+    assert Credentials(username=TEST_USERNAMES[1], password="") in credentials
 
 
 def test_get_credentials__username_password_windows_true():
-    credentials = get_credentials(TEST_USERNAMES, TEST_PASSWORDS, True, True)
+    credentials = get_credentials(TEST_USERNAMES, TEST_PASSWORDS, True)
 
     assert len(credentials) == 9
     for user in TEST_USERNAMES:
         for password in TEST_PASSWORDS:
-            assert AuthOptions(username=user, password=password, is_https=True) in credentials
+            assert Credentials(username=user, password=password) in credentials