driver/qemudriver: add support for netdev option in add_port_forward

In case there are more `user` network interfaces defined, allow to
specific the netdev. If omitted, an empty string is passed and the
behaviour is like before.

This is useful for testing OpenWrt inside a QEMU instance, since OpenWrt
expects by default a LAN interface on eth0 and a WAN interface (with
uplink) on eth1. Previously the port forward would always be added to
eth1, which doesn't support SSH due to firewall policies. By adding the
netdev to the forward function, the SSHDriver works as expected.

Signed-off-by: Paul Spooren <mail@aparcar.org>
[bst: rebased, prevent double space in QMP command, adjust commit
message to fit labgrid's style]
Signed-off-by: Bastian Krause <bst@pengutronix.de>

Author: aparcar

CHANGES.rst

@@ -8,7 +8,9 @@ New Features in 25.1
   instead of a line feed (``\n``). barebox v2025.03.0 onwards handles
   CTRL+D specially to halt autoboot countdown without running interactive
   hooks like bringing up network interfaces automatically.
-
+- The `QEMUDriver` now supports a ``netdev`` argument which can be added to
+  ``add_port_forward()`` and ``remove_port_forward()``  in case there is more
+  than one network interface defined.
 - Guermok HDMI to USB 3.0 capture dongle supported
 
 Breaking changes in 25.1

labgrid/driver/qemudriver.py

@@ -315,21 +315,25 @@ def monitor_command(self, command, arguments={}):
                 "Can't use monitor command on non-running target")
         return self.qmp.execute(command, arguments)
 
-    def _add_port_forward(self, proto, local_address, local_port, remote_address, remote_port):
+    def _add_port_forward(self, proto, local_address, local_port, remote_address, remote_port, netdev):
+        command = ("hostfwd_add", netdev, f"{proto}:{local_address}:{local_port}-{remote_address}:{remote_port}")
+        command = filter(None, command)
         self.monitor_command(
             "human-monitor-command",
-            {"command-line": f"hostfwd_add {proto}:{local_address}:{local_port}-{remote_address}:{remote_port}"},
+            {"command-line": " ".join(command)},
         )
 
-    def add_port_forward(self, proto, local_address, local_port, remote_address, remote_port):
-        self._add_port_forward(proto, local_address, local_port, remote_address, remote_port)
-        self._forwarded_ports[(proto, local_address, local_port)] = (proto, local_address, local_port, remote_address, remote_port)
+    def add_port_forward(self, proto, local_address, local_port, remote_address, remote_port, netdev=""):
+        self._add_port_forward(proto, local_address, local_port, remote_address, remote_port, netdev)
+        self._forwarded_ports[(proto, local_address, local_port, netdev)] = (proto, local_address, local_port, remote_address, remote_port, netdev)
 
-    def remove_port_forward(self, proto, local_address, local_port):
-        del self._forwarded_ports[(proto, local_address, local_port)]
+    def remove_port_forward(self, proto, local_address, local_port, netdev=""):
+        del self._forwarded_ports[(proto, local_address, local_port, netdev)]
+        command = ("hostfwd_remove", netdev, f"{proto}:{local_address}:{local_port}")
+        command = filter(None, command)
         self.monitor_command(
             "human-monitor-command",
-            {"command-line": f"hostfwd_remove {proto}:{local_address}:{local_port}"},
+            {"command-line": " ".join(command)},
         )
 
     def _read(self, size=1, timeout=10, max_size=None):

tests/test_qemudriver.py

@@ -90,3 +90,25 @@ def test_qemu_read_write(qemu_target, qemu_driver, qemu_mock, qemu_version_mock)
     qemu_driver.write(b'abc')
 
     qemu_target.deactivate(qemu_driver)
+
+def test_qemu_port_forwarding(qemu_target, qemu_driver, qemu_mock, qemu_version_mock):
+    qemu_target.activate(qemu_driver)
+
+    qemu_driver.on()
+    qemu_driver.add_port_forward('tcp', '127.0.0.1', 8080, '127.0.0.1', 80)
+    assert ('tcp', '127.0.0.1', 8080, '') in qemu_driver._forwarded_ports.keys()
+    qemu_driver.remove_port_forward('tcp', '127.0.0.1', 8080)
+    assert qemu_driver._forwarded_ports == {}
+
+    qemu_target.deactivate(qemu_driver)
+
+def test_qemu_port_forwarding_with_netdev(qemu_target, qemu_driver, qemu_mock, qemu_version_mock):
+    qemu_target.activate(qemu_driver)
+
+    qemu_driver.on()
+    qemu_driver.add_port_forward('tcp', '127.0.0.1', 8080, '127.0.0.1', 80, netdev='netdev0')
+    assert ('tcp', '127.0.0.1', 8080, 'netdev0') in qemu_driver._forwarded_ports.keys()
+    qemu_driver.remove_port_forward('tcp', '127.0.0.1', 8080, netdev='netdev0')
+    assert qemu_driver._forwarded_ports == {}
+
+    qemu_target.deactivate(qemu_driver)