Fix python meterpreter network method exceptions for OSX

.github/workflows/verify.yml

@@ -33,3 +33,44 @@ jobs:
         run: |
           cd c/meterpreter
           script --return --command 'make docker'
+
+  python:
+    strategy:
+      fail-fast: false
+      matrix:
+        os:
+          - macos-11
+          - windows-2019
+          - ubuntu-20.04
+        runtime_version:
+          - 2.7
+          - 3.6
+          - 3.8
+          - 3.11
+
+    timeout-minutes: 40
+    runs-on: ${{ matrix.os }}
+    name: Python ${{ matrix.runtime_version }} ${{ matrix.os }}
+
+    env:
+      # Ignore messages generated by Pip: 'DEPRECATION: Python 2.7 will reach the end of its life on January 1st, 2020'
+      PIP_DISABLE_PIP_VERSION_CHECK: true
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.runtime_version }}
+
+      - name: Install mock
+        if: ${{ matrix.runtime_version <= 3.3 }}
+        run: |
+          pip install mock
+
+      - name: run tests
+        run: |
+          cd python/meterpreter
+          python -m unittest discover -v ./tests

python/meterpreter/.gitignore

@@ -0,0 +1 @@
+*.pyc

python/meterpreter/README.md

@@ -0,0 +1,11 @@
+## Python Meterpreter
+
+Running unit tests:
+
+```
+# Only required if less than python 3.3
+pip install mock
+
+# Run the tests
+python -m unittest discover -v ./tests
+```

python/meterpreter/ext_server_stdapi.py

@@ -2188,8 +2188,8 @@ def stdapi_net_config_get_interfaces(request, response):
         iface_tlv += tlv_pack(TLV_TYPE_MAC_ADDRESS, iface_info.get('hw_addr', '\x00\x00\x00\x00\x00\x00'))
         if 'mtu' in iface_info:
             iface_tlv += tlv_pack(TLV_TYPE_INTERFACE_MTU, iface_info['mtu'])
-        if 'flags' in iface_info:
-            iface_tlv += tlv_pack(TLV_TYPE_INTERFACE_FLAGS, iface_info['flags'])
+        if 'flags_str' in iface_info:
+            iface_tlv += tlv_pack(TLV_TYPE_INTERFACE_FLAGS, iface_info['flags_str'])
         iface_tlv += tlv_pack(TLV_TYPE_INTERFACE_INDEX, iface_info['index'])
         for address in iface_info.get('addrs', []):
             iface_tlv += tlv_pack(TLV_TYPE_IP, address[1])
@@ -2224,7 +2224,8 @@ def stdapi_net_config_get_interfaces_via_netlink():
         for flag in iface_flags_sorted:
             if (iface.flags & flag):
                 flags.append(iface_flags[flag])
-        iface_info['flags'] = ' '.join(flags)
+        iface_info['flags'] = iface.flags
+        iface_info['flags_str'] = ' '.join(flags)
         cursor = ctypes.sizeof(IFINFOMSG)
         while cursor < len(res_data):
             attribute = ctstruct_unpack(RTATTR, res_data[cursor:])
@@ -2271,19 +2272,20 @@ def stdapi_net_config_get_interfaces_via_osx_ifconfig():
     proc_h = subprocess.Popen('/sbin/ifconfig', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
     if proc_h.wait():
         raise Exception('ifconfig exited with non-zero status')
-    output = proc_h.stdout.read()
+    output = str(proc_h.stdout.read())
 
     interfaces = []
     iface = {}
     for line in output.split('\n'):
-        match = re.match(r'^([a-z0-9]+): flags=(\d+)<[A-Z,]*> mtu (\d+)\s*$', line)
+        match = re.match(r'^([a-z0-9]+): flags=(\d+)<([A-Z,]*)> mtu (\d+)\s*$', line)
         if match is not None:
             if iface:
                 interfaces.append(iface)
             iface = {}
             iface['name'] = match.group(1)
             iface['flags'] = int(match.group(2))
-            iface['mtu'] = int(match.group(3))
+            iface['flags_str'] = match.group(3)
+            iface['mtu'] = int(match.group(4))
             iface['index'] = len(interfaces)
             continue
         match = re.match(r'^\s+ether (([a-f0-9]{2}:){5}[a-f0-9]{2})\s*$', line)
@@ -2487,7 +2489,7 @@ def stdapi_net_config_get_routes_via_osx_netstat():
     proc_h = subprocess.Popen(['/usr/sbin/netstat', '-rn'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
     if proc_h.wait():
         raise Exception('netstat exited with non-zero status')
-    output = proc_h.stdout.read()
+    output = str(proc_h.stdout.read())
 
     routes = []
     state = None
@@ -2524,7 +2526,7 @@ def stdapi_net_config_get_routes_via_osx_netstat():
             continue
         if destination == 'default':
             destination = all_nets
-        if re.match('link#\d+', gateway) or re.match('([0-9a-f]{1,2}:){5}[0-9a-f]{1,2}', gateway):
+        if re.match('link#\\d+', gateway) or re.match('([0-9a-f]{1,2}:){5}[0-9a-f]{1,2}', gateway):
             gateway = all_nets[:-2]
         if '/' in destination:
             destination, netmask_bits = destination.rsplit('/', 1)

python/meterpreter/tests/test_ext_server_stdapi.py

@@ -0,0 +1,168 @@
+# coding=utf-8
+
+import unittest
+import code
+import sys
+import socket
+
+if sys.version_info >= (3, 3):
+    from unittest import mock
+else:
+    import mock
+
+ERROR_SUCCESS = 0
+
+
+def create_meterpreter_context():
+    with open("meterpreter.py", "rb") as file:
+        # Read and patch out the Meterpreter socket connection logic side effect onwards
+        source = file.read()
+        source_without_socket_connection = source[
+            0 : source.index(b"# PATCH-SETUP-ENCRYPTION #")
+        ]
+
+        context = {}
+        exec(source_without_socket_connection, context, context)
+    return context
+
+
+def create_ext_server_stdapi_context(meterpreter, meterpreter_context):
+    with open("ext_server_stdapi.py", "rb") as file:
+        extension_content = file.read()
+
+        context = {}
+        context.update(meterpreter_context["EXPORTED_SYMBOLS"])
+        context["meterpreter"] = meterpreter
+        exec(extension_content, context, context)
+    return context
+
+
+class MockMeterpreter:
+    def __init__(self):
+        self.extension_functions = {}
+
+    def register_extension(self, extension_name):
+        pass
+
+    def register_function(self, func):
+        self.extension_functions[func.__name__] = func
+        return func
+
+
+class TestExtServerStdApi(unittest.TestCase):
+    def setUp(self):
+        self.ext_server_stdapi = create_ext_server_stdapi_context(
+            MockMeterpreter(), create_meterpreter_context()
+        )
+
+    @mock.patch("subprocess.Popen")
+    def test_stdapi_net_config_get_routes_via_osx_netstat(self, mock_popen):
+        popen_read_value = b"""
+Routing tables
+
+Internet:
+Destination        Gateway            Flags        Netif Expire
+default            10.79.0.1          UGScg          en0
+
+Internet6:
+Destination                             Gateway                         Flags         Netif Expire
+default                                 fe80::%utun0                    UGcIg         utun0
+""".lstrip()
+
+        process_mock = mock.Mock()
+        attrs = {
+            "stdout.read.return_value": popen_read_value,
+            "wait.return_value": ERROR_SUCCESS,
+        }
+        process_mock.configure_mock(**attrs)
+        mock_popen.return_value = process_mock
+
+        result = self.ext_server_stdapi[
+            "stdapi_net_config_get_routes_via_osx_netstat"
+        ]()
+
+        expected = [
+            {
+                "gateway": b"\nO\x00\x01",
+                "iface": "en0",
+                "metric": 0,
+                "netmask": b"\x00\x00\x00\x00",
+                "subnet": b"\x00\x00\x00\x00",
+            },
+            {
+                "gateway": b"\xfe\x80\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
+                "iface": "utun0",
+                "metric": 0,
+                "netmask": b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
+                "subnet": b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
+            },
+        ]
+
+        self.assertEqual(result, expected)
+
+    @mock.patch("subprocess.Popen")
+    def test_stdapi_net_config_get_interfaces_via_osx_ifconfig(self, mock_popen):
+        popen_read_value = b"""
+en0: flags=8863<UP,BROADCAST,SMART,RUNNING,SIMPLEX,MULTICAST> mtu 1500
+        options=400<CHANNEL_IO>
+        ether 11:22:33:44:55:66
+        inet 192.168.1.166 netmask 0xffffff00 broadcast 192.168.1.255
+        media: autoselect
+        status: active
+""".lstrip()
+
+        process_mock = mock.Mock()
+        attrs = {
+            "stdout.read.return_value": popen_read_value,
+            "wait.return_value": ERROR_SUCCESS,
+        }
+        process_mock.configure_mock(**attrs)
+        mock_popen.return_value = process_mock
+
+        result = self.ext_server_stdapi[
+            "stdapi_net_config_get_interfaces_via_osx_ifconfig"
+        ]()
+
+        expected = [
+            {
+                "addrs": [
+                    (
+                        socket.AF_INET,
+                        b"\xc0\xa8\x01\xa6",
+                        b"\xff\xff\xff\x00",
+                    )
+                ],
+                "flags": 8863,
+                "flags_str": "UP,BROADCAST,SMART,RUNNING,SIMPLEX,MULTICAST",
+                "hw_addr": '\x11"3DUf',
+                "index": 0,
+                "mtu": 1500,
+                "name": "en0",
+            }
+        ]
+
+        self.assertEqual(result, expected)
+
+    def test_stdapi_net_config_get_interfaces(self):
+        request = bytes()
+        response = bytes()
+        self.assertErrorSuccess("stdapi_net_config_get_interfaces", request, response)
+
+    def test_stdapi_net_config_get_routes(self):
+        request = bytes()
+        response = bytes()
+        self.assertErrorSuccess("stdapi_net_config_get_routes", request, response)
+
+    def assertErrorSuccess(self, method_name, request, response):
+        request = bytes()
+        response = bytes()
+        result = self.ext_server_stdapi[method_name](request, response)
+
+        self.assertEqual(result[0], ERROR_SUCCESS)
+        self.assertIsInstance(result[1], bytes)
+
+        return bytes
+
+
+if __name__ == "__main__":
+    unittest.main()