107 lines
5.7 KiB
Python
107 lines
5.7 KiB
Python
"""RPC exposure and inventory outputs never imply successful exploitation."""
|
|
|
|
import unittest
|
|
|
|
from rpcscope.catalog import (
|
|
CHECKS, CREDENTIALLESS_CHECKS, EXPOSURE_CHECKS, INFO_CHECKS,
|
|
RECOMMENDED_CHECKS, resolve_checks,
|
|
)
|
|
from rpcscope.checks import classify_nxc
|
|
|
|
|
|
HOST = "192.0.2.40"
|
|
|
|
|
|
def nxc_line(module, message, host=HOST, port=445):
|
|
return f"{module:16} {host} {port} HOST1 [+] {message}\n"
|
|
|
|
|
|
class RpcExposureTests(unittest.TestCase):
|
|
def test_presets_are_explicit_and_keep_existing_credentialless_selection(self):
|
|
self.assertEqual(tuple(check.key for check in resolve_checks(["credentialless"])), CREDENTIALLESS_CHECKS)
|
|
self.assertEqual(tuple(check.key for check in resolve_checks(["safe"])), CREDENTIALLESS_CHECKS)
|
|
self.assertEqual(tuple(check.key for check in resolve_checks(["exposure"])), EXPOSURE_CHECKS)
|
|
self.assertEqual(tuple(check.key for check in resolve_checks(["info"])), INFO_CHECKS)
|
|
self.assertEqual(tuple(check.key for check in resolve_checks(["recommended"])), RECOMMENDED_CHECKS)
|
|
self.assertEqual(tuple(check.key for check in resolve_checks(["info,oxid"])), INFO_CHECKS)
|
|
|
|
def test_bind_checks_supply_only_a_method_and_no_cve_claim(self):
|
|
for key in EXPOSURE_CHECKS:
|
|
with self.subTest(key=key):
|
|
definition = CHECKS[key]
|
|
self.assertEqual(definition.nxc_module, "coerce_plus")
|
|
self.assertEqual(len(definition.nxc_options), 1)
|
|
self.assertTrue(definition.nxc_options[0].startswith("METHOD="))
|
|
self.assertEqual(definition.cves, ())
|
|
self.assertFalse(definition.requires_credentials)
|
|
self.assertEqual(definition.positive_status, "exposed")
|
|
self.assertNotIn("LISTENER", repr(definition.nxc_options))
|
|
self.assertNotIn("ALWAYS", repr(definition.nxc_options))
|
|
|
|
def test_recommended_includes_longlogon_once_as_precondition_only(self):
|
|
checks = resolve_checks(["recommended,longlogon"])
|
|
self.assertEqual(len(checks), 12)
|
|
self.assertEqual(sum(check.key == "longlogon" for check in checks), 1)
|
|
self.assertIn("longlogon", [check.key for check in resolve_checks(["recommended"])])
|
|
self.assertEqual(CHECKS["longlogon"].kind, "precondition")
|
|
self.assertEqual(CHECKS["longlogon"].positive_status, "precondition-met")
|
|
self.assertEqual(resolve_checks([]), [])
|
|
|
|
def test_each_bind_marker_is_only_exposure_and_matches_selected_method(self):
|
|
for key in EXPOSURE_CHECKS:
|
|
with self.subTest(key=key):
|
|
method = CHECKS[key].nxc_options[0].partition("=")[2]
|
|
text = nxc_line("COERCE_PLUS", f"VULNERABLE, {method}")
|
|
result = classify_nxc(key, HOST, text)
|
|
self.assertEqual(result.status, "exposed")
|
|
self.assertIn("bind accepted", result.evidence)
|
|
for other in set(EXPOSURE_CHECKS) - {key}:
|
|
self.assertEqual(classify_nxc(other, HOST, text).status, "inconclusive")
|
|
|
|
def test_coercion_success_words_alone_are_not_bind_confirmation(self):
|
|
for message in ("VULNERABLE", "Exploit Success, lsarpc\\EfsRpcOpenFileRaw", "Target is not vulnerable to PetitPotam"):
|
|
with self.subTest(message=message):
|
|
self.assertEqual(classify_nxc("petitpotam", HOST, nxc_line("COERCE_PLUS", message)).status, "inconclusive")
|
|
|
|
def test_bind_requires_exact_target_module_port_and_successful_tool_exit(self):
|
|
for text in (
|
|
nxc_line("COERCE_PLUS", "VULNERABLE, PetitPotam", host="192.0.2.41"),
|
|
nxc_line("SMB", "VULNERABLE, PetitPotam"),
|
|
nxc_line("COERCE_PLUS", "VULNERABLE, PetitPotam", port=80),
|
|
"COERCE_PLUS VULNERABLE, PetitPotam\n",
|
|
):
|
|
with self.subTest(text=text):
|
|
self.assertEqual(classify_nxc("petitpotam", HOST, text).status, "inconclusive")
|
|
self.assertEqual(classify_nxc("petitpotam", HOST, nxc_line("COERCE_PLUS", "VULNERABLE, PetitPotam"), 124).status, "inconclusive")
|
|
|
|
def test_spooler_marker_is_inventory_only(self):
|
|
for port in (135, 445):
|
|
result = classify_nxc("spooler", HOST, nxc_line("SPOOLER", "Spooler service enabled", port=port))
|
|
self.assertEqual(result.status, "observed")
|
|
self.assertIn("patch status was not established", result.evidence)
|
|
for message in ("No endpoints found", "Spooler service disabled", "Protocol failed: access denied"):
|
|
self.assertEqual(classify_nxc("spooler", HOST, nxc_line("SPOOLER", message)).status, "inconclusive")
|
|
|
|
def test_oxid_only_keeps_valid_attributed_ip_addresses(self):
|
|
text = (
|
|
nxc_line("IOXIDRESOLVER", "Address: 10.2.3.4\x00", port=135)
|
|
+ nxc_line("IOXIDRESOLVER", "Address: 2001:db8::12\x00", port=445)
|
|
+ nxc_line("IOXIDRESOLVER", "Address: 10.2.3.4", port=135)
|
|
+ nxc_line("IOXIDRESOLVER", "Address: 999.2.3.4", port=135)
|
|
+ nxc_line("IOXIDRESOLVER", "Address: another-machine", port=135)
|
|
+ nxc_line("IOXIDRESOLVER", "Address: 10.2.3.9", host="192.0.2.41")
|
|
+ nxc_line("SMB", "Address: 10.2.3.10")
|
|
)
|
|
result = classify_nxc("oxid", HOST, text)
|
|
self.assertEqual(result.status, "observed")
|
|
self.assertEqual(result.evidence, "Advertised interface IPs: 10.2.3.4, 2001:db8::12")
|
|
|
|
def test_oxid_malformed_or_unattributed_output_remains_inconclusive(self):
|
|
for text in ("", "Address: 10.2.3.4\n", nxc_line("IOXIDRESOLVER", "Address: not-an-ip")):
|
|
with self.subTest(text=text):
|
|
self.assertEqual(classify_nxc("oxid", HOST, text).status, "inconclusive")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|