Merge pull request #1531 from guardicore/1486/powershell-multi-hop

1486/powershell multi hop

envs/monkey_zoo/blackbox/config_templates/powershell.py

@@ -20,7 +20,7 @@ class PowerShell(ConfigTemplate):
             ],
             "basic.credentials.exploit_password_list": ["Passw0rd!"],
             "basic_network.scope.depth": 2,
-            "basic.credentials.exploit_user_list": ["m0nk3y", "m0nk3y-user", ".\\m0nk3y"],
+            "basic.credentials.exploit_user_list": ["m0nk3y", "m0nk3y-user"],
             "internal.classes.finger_classes": ["PingScanner"],
             "internal.network.tcp_scanner.HTTP_PORTS": [],
             "internal.network.tcp_scanner.tcp_target_ports": [],

monkey/infection_monkey/exploit/powershell_utils/credentials.py

@@ -1,7 +1,10 @@
+import logging
 from dataclasses import dataclass
 from enum import Enum
 from itertools import product
-from typing import List, Union
+from typing import List, Tuple, Union
+
+logger = logging.getLogger(__name__)
 
 
 class SecretType(Enum):
@@ -25,16 +28,43 @@ def get_credentials(
     nt_hashes: List[str],
     is_windows: bool,
 ) -> List[Credentials]:
+    username_domain_combinations = _get_username_domain_combinations(usernames, is_windows)
+
     credentials = []
     credentials.extend(_get_empty_credentials(is_windows))
-    credentials.extend(_get_username_only_credentials(usernames, is_windows))
+    credentials.extend(_get_username_only_credentials(username_domain_combinations, is_windows))
-    credentials.extend(_get_username_password_credentials(usernames, passwords))
+    credentials.extend(_get_username_password_credentials(username_domain_combinations, passwords))
-    credentials.extend(_get_username_lm_hash_credentials(usernames, lm_hashes))
+    credentials.extend(_get_username_lm_hash_credentials(username_domain_combinations, lm_hashes))
-    credentials.extend(_get_username_nt_hash_credentials(usernames, nt_hashes))
+    credentials.extend(_get_username_nt_hash_credentials(username_domain_combinations, nt_hashes))
 
     return credentials
 
 
+def _get_username_domain_combinations(usernames: List[str], is_windows) -> List[str]:
+    username_domain_combinations = set(usernames)
+    for u in usernames:
+        username_domain_combinations.add(f".\\{u}")
+
+    if is_windows:
+        try:
+            domain, current_username = _get_current_user_and_domain()
+            username_domain_combinations.add(current_username)
+            username_domain_combinations.add(f"{domain}\\{current_username}")
+            username_domain_combinations.add(f".\\{current_username}")
+            for u in usernames:
+                username_domain_combinations.add(f"{domain}\\{u}")
+        except Exception as ex:
+            logger.error(f"Failed to get the current user's username and domain name: {ex}")
+
+    return list(username_domain_combinations)
+
+
+def _get_current_user_and_domain() -> Tuple[str, str]:
+    import win32api
+
+    return win32api.GetUserNameEx(win32api.NameSamCompatible).split("\\")
+
+
 # On Windows systems, when username == None and password == None, the current user's credentials
 # will be used to attempt to log into the victim only on the first hop, from island
 # to a machine. Propagating after the first hop is not possible at the moment.

monkey/tests/unit_tests/infection_monkey/exploit/conftest.py

@@ -0,0 +1,21 @@
+import sys
+from collections import namedtuple
+from unittest.mock import MagicMock
+
+import pytest
+
+DomainUser = namedtuple("DomainUser", ["domain", "username"])
+
+
+@pytest.fixture(scope="module")
+def local_user():
+    return DomainUser("TEST-DOMAIN", "localuser")
+
+
+@pytest.fixture(scope="module")
+def patch_win32api_get_user_name(local_user):
+    win32api = MagicMock()
+    win32api.GetUserNameEx = MagicMock(return_value=f"{local_user.domain}\\{local_user.username}")
+    win32api.NameSamCompatible = None
+
+    sys.modules["win32api"] = win32api

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

@@ -1,97 +1,116 @@
+import sys
+from unittest.mock import MagicMock
+
+import pytest
+
 from infection_monkey.exploit.powershell_utils.credentials import (
     Credentials,
     SecretType,
     get_credentials,
 )
 
+# Use the path_win32api_get_user_name fixture for all tests in this module
+pytestmark = pytest.mark.usefixtures("patch_win32api_get_user_name")
+
 TEST_USERNAMES = ["user1", "user2"]
 TEST_PASSWORDS = ["p1", "p2"]
 TEST_LM_HASHES = ["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"]
 TEST_NT_HASHES = ["cccccccccccccccccccccccccccccccc", "dddddddddddddddddddddddddddddddd"]
 
 
-def test_get_credentials__empty_windows_true():
+@pytest.fixture(scope="module")
-    credentials = get_credentials([], [], [], [], True)
+def windows_false_usernames():
+    usernames = TEST_USERNAMES.copy()
+    usernames.extend([f".\\{u}" for u in TEST_USERNAMES])
 
-    assert len(credentials) == 1
+    return usernames
-    assert credentials[0] == Credentials(username=None, secret=None, secret_type=SecretType.CACHED)
+
+
+@pytest.fixture(scope="module")
+def windows_true_usernames(local_user):
+    usernames = TEST_USERNAMES.copy()
+    usernames.append(local_user.username)
+    usernames = (
+        usernames
+        + [f".\\{u}" for u in usernames]
+        + [f"{local_user.domain}\\{u}" for u in usernames]
+    )
+
+    return usernames
+
+
+def test_get_credentials__empty_windows_true():
+    results = get_credentials([], [], [], [], True)
+
+    assert Credentials(username=None, secret=None, secret_type=SecretType.CACHED) in results
 
 
 def test_get_credentials__empty_windows_false():
-    credentials = get_credentials([], [], [], [], False)
+    results = get_credentials([], [], [], [], False)
 
-    assert len(credentials) == 0
+    assert len(results) == 0
 
 
-def test_get_credentials__username_only_windows_true():
+def assert_secrets_in_results(usernames, secrets, secret_type, results):
-    credentials = get_credentials(TEST_USERNAMES, [], [], [], True)
+    for u in usernames:
-
+        for s in secrets:
-    assert len(credentials) == 5
+            assert Credentials(username=u, secret=s, secret_type=secret_type) in results
-    assert (
-        Credentials(username=TEST_USERNAMES[0], secret="", secret_type=SecretType.PASSWORD)
-        in credentials
-    )
-    assert (
-        Credentials(username=TEST_USERNAMES[1], secret="", secret_type=SecretType.PASSWORD)
-        in credentials
-    )
-    assert (
-        Credentials(username=TEST_USERNAMES[0], secret=None, secret_type=SecretType.CACHED)
-        in credentials
-    )
-    assert (
-        Credentials(username=TEST_USERNAMES[1], secret=None, secret_type=SecretType.CACHED)
-        in credentials
-    )
 
 
-def test_get_credentials__username_only_windows_false():
+def test_get_credentials__username_only_windows_true(windows_true_usernames):
-    credentials = get_credentials(TEST_USERNAMES, [], [], [], False)
+    results = get_credentials(TEST_USERNAMES, [], [], [], True)
 
-    assert len(credentials) == 2
+    assert len(results) == 19
-    assert (
+    assert_secrets_in_results(windows_true_usernames, [""], SecretType.PASSWORD, results)
-        Credentials(username=TEST_USERNAMES[0], secret="", secret_type=SecretType.PASSWORD)
+    assert_secrets_in_results(windows_true_usernames, [None], SecretType.CACHED, results)
-        in credentials
-    )
-    assert (
-        Credentials(username=TEST_USERNAMES[1], secret="", secret_type=SecretType.PASSWORD)
-        in credentials
-    )
 
 
-def test_get_credentials__username_password_windows_true():
+def test_get_credentials__username_only_windows_false(windows_false_usernames):
-    credentials = get_credentials(TEST_USERNAMES, TEST_PASSWORDS, [], [], True)
+    results = get_credentials(TEST_USERNAMES, [], [], [], False)
 
-    assert len(credentials) == 9
+    assert len(results) == 4
-    for user in TEST_USERNAMES:
+    assert_secrets_in_results(windows_false_usernames, [""], SecretType.PASSWORD, results)
-        for password in TEST_PASSWORDS:
-            assert (
-                Credentials(username=user, secret=password, secret_type=SecretType.PASSWORD)
-                in credentials
-            )
 
 
-def test_get_credentials__username_lm_hash_windows_false():
+def test_get_credentials__username_password_windows_true(windows_true_usernames):
-    credentials = get_credentials(TEST_USERNAMES, TEST_PASSWORDS, TEST_LM_HASHES, [], False)
+    results = get_credentials(TEST_USERNAMES, TEST_PASSWORDS, [], [], True)
 
-    assert len(credentials) == 10
+    assert_secrets_in_results(windows_true_usernames, TEST_PASSWORDS, SecretType.PASSWORD, results)
-    for user in TEST_USERNAMES:
-        for lm_hash in TEST_LM_HASHES:
-            assert (
-                Credentials(username=user, secret=lm_hash, secret_type=SecretType.LM_HASH)
-                in credentials
-            )
 
 
-def test_get_credentials__username_nt_hash_windows_false():
+def test_get_credentials__username_lm_hash_windows_false(windows_false_usernames):
-    credentials = get_credentials(
+    results = get_credentials(TEST_USERNAMES, TEST_PASSWORDS, TEST_LM_HASHES, [], False)
-        TEST_USERNAMES, TEST_PASSWORDS, TEST_LM_HASHES, TEST_NT_HASHES, False
-    )
 
-    assert len(credentials) == 14
+    assert len(results) == 20
-    for user in TEST_USERNAMES:
+    assert_secrets_in_results(windows_false_usernames, TEST_LM_HASHES, SecretType.LM_HASH, results)
-        for nt_hash in TEST_NT_HASHES:
+
-            assert (
+
-                Credentials(username=user, secret=nt_hash, secret_type=SecretType.NT_HASH)
+def test_get_credentials__username_lm_hash_windows_true(windows_true_usernames):
-                in credentials
+    results = get_credentials(TEST_USERNAMES, TEST_PASSWORDS, TEST_LM_HASHES, [], True)
-            )
+
+    assert_secrets_in_results(windows_true_usernames, TEST_LM_HASHES, SecretType.LM_HASH, results)
+
+
+def test_get_credentials__username_nt_hash_windows_false(windows_false_usernames):
+    results = get_credentials(TEST_USERNAMES, TEST_PASSWORDS, TEST_LM_HASHES, TEST_NT_HASHES, False)
+
+    assert len(results) == 28
+    assert_secrets_in_results(windows_false_usernames, TEST_NT_HASHES, SecretType.NT_HASH, results)
+
+
+def test_get_credentials__username_nt_hash_windows_true(windows_true_usernames):
+    results = get_credentials(TEST_USERNAMES, TEST_PASSWORDS, TEST_LM_HASHES, TEST_NT_HASHES, True)
+
+    assert_secrets_in_results(windows_true_usernames, TEST_NT_HASHES, SecretType.NT_HASH, results)
+
+
+def test_get_credentials__get_username_failure(windows_false_usernames):
+    win32api = MagicMock()
+    win32api.GetUserNameEx = MagicMock(side_effect=Exception("win32api test failure"))
+    win32api.NameSamCompatible = None
+    sys.modules["win32api"] = win32api
+
+    results = get_credentials(TEST_USERNAMES, [], [], [], True)
+
+    assert len(results) == 9
+    assert_secrets_in_results(windows_false_usernames, [""], SecretType.PASSWORD, results)

monkey/tests/unit_tests/infection_monkey/exploit/test_powershell.py

@@ -9,6 +9,9 @@ from infection_monkey.exploit.powershell_utils.auth_options import AuthOptions
 from infection_monkey.exploit.powershell_utils.credentials import Credentials
 from infection_monkey.model.host import VictimHost
 
+# Use the path_win32api_get_user_name fixture for all tests in this module
+pytestmark = pytest.mark.usefixtures("patch_win32api_get_user_name")
+
 USER_LIST = ["user1", "user2"]
 PASSWORD_LIST = ["pass1", "pass2"]
 LM_HASH_LIST = ["bogo_lm_1"]