run: add --ip-address option

docs/development.md

@@ -143,6 +143,35 @@ The following example uses the qemu driver, and connects using vmnet-run:
 [  16.630] INFO VM is ready at test-vmnet-helper.local
 ```
 
+VMs use DHCP by default. To assign a static IP address, restrict the DHCP range,
+then select an address outside of that range:
+
+```console
+% ./run test \
+    --start-address 192.168.200.1 \
+    --end-address 192.168.200.127 \
+    --subnet-mask 255.255.255.0 \
+    --ip-address 192.168.200.128
+```
+
+> [!NOTE]
+> Setting `--ip-address` to a value inside the DHCP range may work, but may
+> cause conflicts.
+
+> [!NOTE]
+> `--ip-address` must be difrerent from `--start-address`, but in the same
+> subnet.
+
+When changing a VM's IP address or switching to DHCP, the instance ID and host
+key will be reset. Remove the old host key before you ssh again:
+
+```console
+% ssh-keygen -R test-vmnet-helper.local
+# Host test-vmnet-helper.local found: line 161
+/Users/user/.ssh/known_hosts updated.
+Original contents retained as /Users/user/.ssh/known_hosts.old
+```
+
 ### Performance tuning
 
 By default, VMs use interrupt-driven packet processing. During high

run

@@ -14,6 +14,7 @@ import signal
 import sys
 
 import testing
+import testing.validate
 
 VMNET_OFFLOAD_AUTO = {
     "vfkit": "off",
@@ -45,6 +46,7 @@ def main():
     )
     p.add_argument(
         "--start-address",
+        type=testing.validate.private_ipv4_address,
         help=(
             "The starting IPv4 address (string) to use for the interface. This "
             "address is used as the gateway address. The subsequent address up "
@@ -55,13 +57,16 @@ def main():
     )
     p.add_argument(
         "--end-address",
+        type=testing.validate.private_ipv4_address,
         help=(
             "The DHCP IPv4 range end address (string) to use for the interface. "
             "The address must be in the private IP range (RFC 1918)."
         ),
     )
     p.add_argument(
-        "--subnet-mask", help="The IPv4 subnet mask (string) to use on the interface."
+        "--subnet-mask",
+        type=testing.validate.subnet_mask,
+        help="The IPv4 subnet mask (string) to use on the interface.",
     )
     p.add_argument(
         "--shared-interface",
@@ -109,7 +114,7 @@ def main():
     p.add_argument(
         "--cpus",
         metavar="N",
-        type=cpus,
+        type=testing.validate.cpus,
         default=2,
         help="Number of vpus (2)",
     )
@@ -139,10 +144,19 @@ def main():
     p.add_argument(
         "--dns-servers",
         metavar="ADDR",
-        type=ip_list,
+        type=testing.validate.ip_list,
         default=["8.8.8.8", "1.1.1.1"],
         help="Comma-separated DNS servers for the VM (8.8.8.8,1.1.1.1)",
     )
+    p.add_argument(
+        "--ip-address",
+        type=testing.validate.private_ipv4_address,
+        help=(
+            "The static IPv4 address to assign to the vm. Requires the --start-address, "
+            "--end-address, --subnet-mask options, and should be outside of the requested "
+            "DHCP range. When unset, the vm uses DHCP."
+        ),
+    )
 
     # Performance tuning
 
@@ -159,21 +173,8 @@ def main():
     args = p.parse_args()
     setup_logging(args.verbose)
 
-    if args.network_name:
-        if args.operation_mode:
-            p.error("--network cannot be used with --operation-mode")
-        if args.start_address:
-            p.error("--network cannot be used with --start-address")
-        if args.end_address:
-            p.error("--network cannot be used with --end-address")
-        if args.subnet_mask:
-            p.error("--network cannot be used with --subnet-mask")
-
-    if args.operation_mode == "bridged":
-        if not args.shared_interface:
-            p.error("--shared-interface required for --operation-mode=bridged")
-        if args.enable_isolation:
-            p.error("--enable-isolation not compatible with --operation-mode=bridged")
+    testing.validate.network_options(p, args)
+    testing.validate.operation_mode(p, args)
 
     signal.signal(signal.SIGTERM, terminate)
     signal.signal(signal.SIGINT, terminate)
@@ -269,18 +270,6 @@ def run_with_runner(args):
         vm.stop()
 
 
-def ip_list(s):
-    # TODO: validate that values are IP addresses.
-    return s.split(",")
-
-
-def cpus(s):
-    n = int(s)
-    if n < 1:
-        raise ValueError(f"Invalid number of cpus: '{s}'")
-    return n
-
-
 def terminate(signo, frame):
     sys.exit(1)
 

testing/cidata.py

@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: The vmnet-helper authors
 # SPDX-License-Identifier: Apache-2.0
 
+import ipaddress
 import glob
 import logging
 import os
@@ -156,24 +157,38 @@ def create_network_config(vm):
     """
     Create cloud-init network-config dict.
     """
-    return {
+    data = {
         "version": 2,
         "ethernets": {
             "eth0": {
                 "match": {
                     "macaddress": vm.mac_address,
                 },
-                "dhcp4": True,
-                "dhcp-identifier": "mac",
-                "dhcp4-overrides": {
-                    "use-dns": False,
-                },
+                "dhcp4": vm.args.ip_address is None,
                 "nameservers": {
                     "addresses": vm.dns_servers,
                 },
             },
         },
     }
+    dhcp_data = {
+        "dhcp-identifier": "mac",
+        "dhcp4-overrides": {
+            "use-dns": False,
+        },
+    }
+    if vm.args.ip_address:
+        data["ethernets"]["eth0"]["addresses"] = [
+            ipaddress.IPv4Interface(
+                (vm.args.ip_address, vm.args.subnet_mask)
+            ).with_prefixlen
+        ]
+        data["ethernets"]["eth0"]["routes"] = [
+            {"to": "default", "via": str(vm.args.start_address)}
+        ]
+    else:
+        data["ethernets"]["eth0"].update(dhcp_data)
+    return data
 
 
 def file_matches(data, iso_path, file_path):

testing/validate.py

@@ -0,0 +1,128 @@
+# SPDX-FileCopyrightText: The vmnet-helper authors
+# SPDX-License-Identifier: Apache-2.0
+
+import ipaddress
+
+
+def ip_list(s):
+    return [str(ipaddress.IPv4Address(x)) for x in s.split(",")]
+
+
+def cpus(s):
+    n = int(s)
+    if n < 1:
+        raise ValueError(f"Invalid number of cpus: '{s}'")
+    return n
+
+
+def subnet_mask(s):
+    """
+    Raises NetmaskValueError if 's' is not a valid netmask.
+
+    Returns the unmodified string expected by ipaddress.IPv4Network.
+    """
+    ipaddress.IPv4Network(f"0.0.0.0/{s}")
+    return s
+
+
+def _one_dhcp_option_set(args):
+    """
+    Returns True if one or more DHCP option is set.
+    """
+    return args.start_address or args.end_address or args.subnet_mask
+
+
+def _all_dhcp_options_set(args):
+    """
+    Returns True if all DHCP options are set.
+    """
+    return args.start_address and args.end_address and args.subnet_mask
+
+
+def _dhcp_options(p, args):
+    if _one_dhcp_option_set(args) and not _all_dhcp_options_set(args):
+        p.error(
+            "--start-address, --end-address, --subnet-mask must all be set or all omitted"
+        )
+
+
+def _bridged_mode(p, args):
+    if not args.shared_interface:
+        p.error("--shared-interface required for --operation-mode=bridged")
+    if args.enable_isolation:
+        p.error("--enable-isolation not compatible with --operation-mode=bridged")
+
+
+def _shared_mode(p, args):
+    _dhcp_options(p, args)
+
+
+def _host_mode(p, args):
+    _dhcp_options(p, args)
+
+
+def operation_mode(p, args):
+    if args.operation_mode == "shared" or not args.operation_mode:
+        _shared_mode(p, args)
+    elif args.operation_mode == "host":
+        _host_mode(p, args)
+    elif args.operation_mode == "bridged":
+        _bridged_mode(p, args)
+
+
+def network_options(p, args):
+    """
+    Validate the network options passed to run.
+    """
+    if args.network_name:
+        if args.operation_mode:
+            p.error("--network cannot be used with --operation-mode")
+        if args.start_address:
+            p.error("--network cannot be used with --start-address")
+        if args.end_address:
+            p.error("--network cannot be used with --end-address")
+        if args.subnet_mask:
+            p.error("--network cannot be used with --subnet-mask")
+        if args.ip_address:
+            p.error("--network cannot be used with --ip-address")
+
+    if args.ip_address and not _all_dhcp_options_set(args):
+        p.error("--ip-address requires --start-address, --end-address, --subnet-mask")
+
+    if _all_dhcp_options_set(args):
+        # vmnet does not enforce the order of --start-address and --end-address.
+        network = ipaddress.IPv4Interface(
+            (args.start_address, args.subnet_mask)
+        ).network
+        if args.end_address not in network:
+            p.error("--start-address and --end-address must be in the same subnet")
+        # --ip-address inside the DHCP range may cause conflicts, but works.
+        if args.ip_address:
+            if args.ip_address not in network:
+                p.error(
+                    "--ip-address, --start-address and --end-address "
+                    "must be in the same subnet",
+                )
+            # Only reserve --start-address, since it gets assigned to the host.
+            if args.ip_address == args.start_address:
+                p.error("--ip-address must be different from --start-address")
+
+
+_RFC1918_NETWORKS = [
+    ipaddress.IPv4Network("10.0.0.0/8"),
+    ipaddress.IPv4Network("172.16.0.0/12"),
+    ipaddress.IPv4Network("192.168.0.0/16"),
+]
+
+
+def private_ipv4_address(ip):
+    """
+    Validates that "ip" is in the RFC 1918 private range.
+
+    Returns an ipaddress.IPv4Address object, or raises ValueError if validation fails.
+    """
+    address = ipaddress.IPv4Address(ip)
+    for network in _RFC1918_NETWORKS:
+        if address in network:
+            return address
+    raise ValueError(f"{ip} is not a valid RFC 1918 IP address")

testing/vm.py

@@ -43,6 +43,7 @@ def __init__(self, args, mac_address, fd=None, socket=None, runner=None):
         self.memory = args.memory
         self.distro = args.distro
         self.dns_servers = args.dns_servers
+        self.ip_address = args.ip_address
         self.busy_poll = args.busy_poll
         self.serial = store.vm_path(self.vm_name, "serial.log")
         self.enable_offloading = args.enable_offloading