[Hyper-v platform] Enabling port forwarding when switch type is Internal

lisa/sut_orchestrator/hyperv/platform_.py

@@ -10,6 +10,7 @@
 from lisa.node import RemoteNode
 from lisa.platform_ import Platform
 from lisa.tools import Cp, HyperV, Mkdir, Mount, PowerShell
+from lisa.tools.hyperv import HypervSwitchType
 from lisa.util import LisaException, constants
 from lisa.util.logger import Logger, get_logger
 from lisa.util.parallel import run_in_parallel
@@ -309,10 +310,19 @@ def _deploy_environment(self, environment: Environment, log: Logger) -> None:
             hv.start_vm(name=vm_name, extra_args=extra_args)
 
             ip_addr = hv.get_ip_address(vm_name)
+            port = 22
+            # If the switch type is internal, we need to add a NAT mapping to access the
+            # VM from the outside of HyperV host.
+            if default_switch.type == HypervSwitchType.INTERNAL:
+                port = hv.add_nat_mapping(
+                    nat_name=default_switch.name,
+                    internal_ip=ip_addr,
+                )
+                ip_addr = node_context.host.public_address
             username = self.runbook.admin_username
             password = self.runbook.admin_password
             node.set_connection_info(
-                address=ip_addr, username=username, password=password
+                address=ip_addr, username=username, password=password, public_port=port
             )
             # In some cases, we observe that resize vhd resizes the entire disk
             # but fails to expand the partition size.

lisa/tools/hyperv.py

@@ -5,7 +5,7 @@
 import time
 from dataclasses import dataclass
 from enum import Enum
-from typing import Dict, Optional
+from typing import Dict, Optional, Set
 
 from assertpy import assert_that
 from dataclasses_json import dataclass_json
@@ -35,6 +35,12 @@ class HyperV(Tool):
     # 192.168.5.12
     IP_REGEX = r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}"
     _default_switch: Optional[VMSwitch] = None
+    _external_forwarding_port_start = 50000
+    _assigned_nat_ports: Set[int] = set()
+    internal_nat_router = "192.168.5.1"
+    internal_nat_subnet = "192.168.5.0/24"
+    internal_nat_dhcp_ip_start = "192.168.5.50"
+    internal_nat_dhcp_ip_end = "192.168.5.100"
 
     @property
     def command(self) -> str:
@@ -293,6 +299,40 @@ def create_nat(self, name: str, ip_range: str) -> None:
             force_run=True,
         )
 
+    def add_nat_mapping(self, nat_name: str, internal_ip: str) -> int:
+        external_port = self._get_next_available_nat_port()
+
+        # delete existing NAT mapping
+        self.delete_nat_mapping(external_port)
+
+        # create a new NAT
+        self.node.tools[PowerShell].run_cmdlet(
+            f"Add-NetNatStaticMapping -NatName {nat_name} -Protocol TCP "
+            f"-ExternalIPAddress 0.0.0.0 -InternalIPAddress {internal_ip} "
+            f"-InternalPort 22 -ExternalPort {external_port}",
+            force_run=True,
+        )
+        return external_port
+
+    def delete_nat_mapping(self, external_port: int) -> None:
+        # get the NAT mapping id for the port
+        mapping_id = self.node.tools[PowerShell].run_cmdlet(
+            f"Get-NetNatStaticMapping | "
+            f"Where-Object {{$_.ExternalPort -eq {external_port}}}"
+            f" | Select-Object -ExpandProperty StaticMappingID",
+            force_run=True,
+        )
+        if mapping_id:
+            # delete the NAT mapping if it exists
+            self.node.tools[PowerShell].run_cmdlet(
+                f"Remove-NetNatStaticMapping -StaticMappingID {mapping_id} "
+                "-Confirm:$false",
+                force_run=True,
+            )
+            self._release_nat_port(external_port)
+        else:
+            self._log.debug(f"Mapping for port {external_port} does not exist")
+
     def delete_nat_networking(self, switch_name: str, nat_name: str) -> None:
         # Delete switch
         self.delete_switch(switch_name)
@@ -313,13 +353,13 @@ def setup_nat_networking(self, switch_name: str, nat_name: str) -> None:
 
         # set switch interface as gateway for NAT
         self.node.tools[PowerShell].run_cmdlet(
-            "New-NetIPAddress -IPAddress 192.168.5.1 "
+            f"New-NetIPAddress -IPAddress {self.internal_nat_router} "
             f"-InterfaceIndex {interface_index} -PrefixLength 24",
             force_run=True,
         )
 
         # create a new NAT
-        self.create_nat(nat_name, "192.168.5.0/24")
+        self.create_nat(nat_name, self.internal_nat_subnet)
 
     def get_ip_address(self, name: str) -> str:
         # verify vm is running
@@ -432,13 +472,13 @@ def enable_internal_dhcp(self, dhcp_scope_name: str = "DHCPInternalNAT") -> None
 
         # Configure the DHCP server to use the internal NAT network
         powershell.run_cmdlet(
-            f'Add-DhcpServerV4Scope -Name "{dhcp_scope_name}" -StartRange 192.168.0.50 -EndRange 192.168.0.100 -SubnetMask 255.255.255.0',  # noqa: E501
+            f'Add-DhcpServerV4Scope -Name "{dhcp_scope_name}" -StartRange {self.internal_nat_dhcp_ip_start} -EndRange {self.internal_nat_dhcp_ip_end} -SubnetMask 255.255.255.0',  # noqa: E501
             force_run=True,
         )
 
         # Set the DHCP server options
         powershell.run_cmdlet(
-            "Set-DhcpServerV4OptionValue -Router 192.168.0.1 -DnsServer 168.63.129.16",
+            f"Set-DhcpServerV4OptionValue -Router {self.internal_nat_router} -DnsServer 168.63.129.16",
             force_run=True,
         )
 
@@ -482,3 +522,19 @@ def _run_hyperv_cmdlet(
 
     def _check_exists(self) -> bool:
         return self.node.tools[WindowsFeatureManagement].is_installed("Hyper-V")
+
+    def _get_next_available_nat_port(self) -> int:
+        # Start checking from the external forwarding port start and
+        # find the first available one
+        port = self._external_forwarding_port_start
+        while port in self._assigned_nat_ports:
+            port += 1
+        self._assigned_nat_ports.add(port)  # Assign this port
+        return port
+
+    def _release_nat_port(self, port: int) -> None:
+        # Release a port if used
+        if port in self._assigned_nat_ports:
+            self._assigned_nat_ports.remove(port)
+        else:
+            print(f"Port {port} was not assigned.")