fix: use Multipass-compatible instance names (#712)

Signed-off-by: Callahan Kovacs <callahan.kovacs@canonical.com>

Author: mr-cal

craft_providers/executor.py

@@ -18,14 +18,17 @@
 """Executor module."""
 
 import contextlib
+import hashlib
 import io
 import logging
 import pathlib
+import re
 import subprocess
 from abc import ABC, abstractmethod
 from typing import Dict, Generator, List, Optional
 
 import craft_providers.util.temp_paths
+from craft_providers.errors import ProviderError
 
 logger = logging.getLogger(__name__)
 
@@ -187,3 +190,61 @@ def is_running(self) -> bool:
 
         :returns: True if instance is running.
         """
+
+
+def get_instance_name(name: str, error_class: type[ProviderError]) -> str:
+    """Get an instance-friendly name from a name.
+
+    LXD and Multipass instance names have the same naming convention
+    as Linux hostnames.
+
+    Naming convention:
+    - between 1 and 63 characters long
+    - made up exclusively of letters, numbers, and hyphens from the ASCII table
+    - not begin with a digit or a hyphen
+    - not end with a hyphen
+
+    To create an instance name, invalid characters are removed, the name is
+    truncated to 40 characters, then a hash is appended:
+    <truncated-name>-<hash-of-name>
+    └     1 - 40   ┘1└     20     ┘
+
+    :param name: the name to convert
+    :param error_class: the exception class to raise if name is invalid
+
+    :raises error_class: if name contains no alphanumeric characters
+    :returns: the instance name
+    """
+    # remove anything that is not an alphanumeric character or hyphen
+    name_with_valid_chars = re.sub(r"[^\w-]", "", name, flags=re.ASCII)
+    if not name_with_valid_chars:
+        raise error_class(
+            brief=f"failed to create an instance with name {name!r}.",
+            details="name must contain at least one alphanumeric character",
+        )
+
+    # trim digits and hyphens from the beginning and hyphens from the end
+    trimmed_name = re.compile(r"^[0-9-]*(?P<valid_name>.*?)[-]*$").search(
+        name_with_valid_chars
+    )
+    if not trimmed_name or not trimmed_name.group("valid_name"):
+        raise error_class(
+            brief=f"failed to create an instance with name {name!r}.",
+            details="name must contain at least one alphanumeric character",
+        )
+    valid_name = trimmed_name.group("valid_name")
+
+    # if the original name satisfies the naming convention, then use the original name
+    if name == valid_name and len(name) <= 63:
+        instance_name = name
+
+    # else, continue converting the name
+    else:
+        # truncate to 40 characters
+        truncated_name = valid_name[:40]
+        # hash the entire name, not the truncated name
+        hashed_name = hashlib.sha1(name.encode()).hexdigest()[:20]
+        instance_name = f"{truncated_name}-{hashed_name}"
+
+    logger.debug("Converted name %r to instance name %r", name, instance_name)
+    return instance_name

craft_providers/lxd/lxd_instance.py

@@ -17,20 +17,18 @@
 
 """LXD Instance Executor."""
 
-import hashlib
 import io
 import logging
 import os
 import pathlib
-import re
 import shutil
 import subprocess
 import tempfile
 from typing import Any, Dict, List, Optional
 
 from craft_providers.const import TIMEOUT_SIMPLE
 from craft_providers.errors import details_from_called_process_error
-from craft_providers.executor import Executor
+from craft_providers.executor import Executor, get_instance_name
 from craft_providers.lxd.errors import LXDError
 from craft_providers.lxd.lxc import LXC
 from craft_providers.util import env_cmd
@@ -39,7 +37,14 @@
 
 
 class LXDInstance(Executor):
-    """LXD Instance Lifecycle."""
+    """Wrapper for a LXD Instance.
+
+    :ivar name: The provided name for the instance.
+    :ivar instance_name: The normalized name actually used for the instance.
+    :ivar project: The name of the LXD project.
+    :ivar remote: The name of the LXD remote.
+    :ivar lxc: The LXC wrapper to use.
+    """
 
     def __init__(
         self,
@@ -55,11 +60,13 @@ def __init__(
         To comply with LXD naming conventions, the supplied name is converted to a
         LXD-compatible name before creating the instance.
 
-        :param name: Unique name of the lxd instance
-        :param default_command_environment: command environment
-        :param project: name of lxd project
-        :param remote: name of lxd remote
-        :param lxc: LXC instance object
+        :param name: The name of the LXDInstance.
+        :param default_command_environment: The command environment.
+        :param project: The name of the LXD project.
+        :param remote: The name of the LXD remote.
+        :param lxc: The LXC wrapper to use.
+
+        :raises LXDError: If the name is invalid.
         """
         super().__init__()
 
@@ -69,7 +76,7 @@ def __init__(
             self.default_command_environment = {}
 
         self.name = name
-        self._set_instance_name()
+        self.instance_name = get_instance_name(name, LXDError)
         self.project = project
         self.remote = remote
 
@@ -78,57 +85,6 @@ def __init__(
         else:
             self.lxc = lxc
 
-    def _set_instance_name(self) -> None:
-        """Convert a name to a LXD-compatible name.
-
-        LXD naming convention:
-        - between 1 and 63 characters long
-        - made up exclusively of letters, numbers, and hyphens from the ASCII table
-        - not begin with a digit or a hyphen
-        - not end with a hyphen
-
-        To create a LXD-compatible name, invalid characters are removed, the name is
-        truncated to 40 characters, then a hash is appended:
-        <truncated-name>-<hash-of-name>
-        └     1 - 40   ┘1└     20     ┘
-
-        :param name: name of instance
-        :raises LXDError: if name contains no alphanumeric characters
-        """
-        # remove anything that is not an alphanumeric characters or hyphen
-        name_with_valid_chars = re.sub(r"[^\w-]", "", self.name)
-        if not name_with_valid_chars:
-            raise LXDError(
-                brief=f"failed to create LXD instance with name {self.name!r}.",
-                details="name must contain at least one alphanumeric character",
-            )
-
-        # trim digits and hyphens from the beginning and hyphens from the end
-        trimmed_name = re.compile(r"^[0-9-]*(?P<valid_name>.*?)[-]*$").search(
-            name_with_valid_chars
-        )
-        if not trimmed_name or not trimmed_name.group("valid_name"):
-            raise LXDError(
-                brief=f"failed to create LXD instance with name {self.name!r}.",
-                details="name must contain at least one alphanumeric character",
-            )
-        valid_name = trimmed_name.group("valid_name")
-
-        # if the original name meets LXD's naming convention, then use the original name
-        if self.name == valid_name and len(self.name) <= 63:
-            instance_name = self.name
-
-        # else, continue converting the name
-        else:
-            # truncate to 40 characters
-            truncated_name = valid_name[:40]
-            # hash the entire name, not the truncated name
-            hashed_name = hashlib.sha1(self.name.encode()).hexdigest()[:20]
-            instance_name = f"{truncated_name}-{hashed_name}"
-
-        self.instance_name = instance_name
-        logger.debug("Set LXD instance name to %r", instance_name)
-
     def _finalize_lxc_command(
         self,
         command: List[str],

craft_providers/multipass/multipass_instance.py

@@ -26,7 +26,7 @@
 from craft_providers.const import TIMEOUT_COMPLEX, TIMEOUT_SIMPLE
 from craft_providers.util import env_cmd
 
-from .. import Executor
+from ..executor import Executor, get_instance_name
 from .errors import MultipassError
 from .multipass import Multipass
 
@@ -59,9 +59,10 @@ def _rootify_multipass_command(
 
 
 class MultipassInstance(Executor):
-    """Multipass Instance Lifecycle.
+    """Wrapper for a Multipass instance.
 
-    :param name: Name of multipass instance.
+    :ivar name: The provided name for the instance.
+    :ivar instance_name: The normalized name actually used for the instance.
     """
 
     def __init__(
@@ -70,9 +71,17 @@ def __init__(
         name: str,
         multipass: Optional[Multipass] = None,
     ) -> None:
+        """Set up the wrapper class.
+
+        :param name: The name of the MultipassInstance.
+        :param multipass: The Multipass wrapper to use.
+
+        :raises MultipassError: If the name is invalid.
+        """
         super().__init__()
 
         self.name = name
+        self.instance_name = get_instance_name(name, MultipassError)
 
         if multipass is not None:
             self._multipass = multipass
@@ -146,7 +155,7 @@ def push_file_io(
             tmp_file_path = self._create_temp_file()
 
             self._multipass.transfer_source_io(
-                source=content, destination=f"{self.name}:{tmp_file_path}"
+                source=content, destination=f"{self.instance_name}:{tmp_file_path}"
             )
 
             # now that the file has been transferred, its ownership can be set
@@ -174,15 +183,15 @@ def push_file_io(
             raise MultipassError(
                 brief=(
                     f"Failed to create file {destination.as_posix()!r}"
-                    f" in Multipass instance {self.name!r}."
+                    f" in Multipass instance {self.instance_name!r}."
                 ),
                 details=errors.details_from_called_process_error(error),
             ) from error
 
     def delete(self) -> None:
         """Delete instance and purge."""
         return self._multipass.delete(
-            instance_name=self.name,
+            instance_name=self.instance_name,
             purge=True,
         )
 
@@ -215,7 +224,7 @@ def execute_popen(
         :returns: Popen instance.
         """
         return self._multipass.exec(
-            instance_name=self.name,
+            instance_name=self.instance_name,
             command=_rootify_multipass_command(command, cwd=cwd, env=env),
             runner=subprocess.Popen,
             timeout=timeout,
@@ -255,7 +264,7 @@ def execute_run(
         :raises subprocess.CalledProcessError: if command fails and check is True.
         """
         return self._multipass.exec(
-            instance_name=self.name,
+            instance_name=self.instance_name,
             command=_rootify_multipass_command(command, cwd=cwd, env=env),
             runner=subprocess.run,
             timeout=timeout,
@@ -272,7 +281,7 @@ def exists(self) -> bool:
         """
         vm_list = self._multipass.list()
 
-        return self.name in vm_list
+        return self.instance_name in vm_list
 
     def _get_info(self) -> Dict[str, Any]:
         """Get configuration and state for instance.
@@ -282,15 +291,15 @@ def _get_info(self) -> Dict[str, Any]:
 
         :raises MultipassError: If unable to parse VM info.
         """
-        info_data = self._multipass.info(instance_name=self.name).get("info")
+        info_data = self._multipass.info(instance_name=self.instance_name).get("info")
 
-        if info_data is None or self.name not in info_data:
+        if info_data is None or self.instance_name not in info_data:
             raise MultipassError(
                 brief="Malformed multipass info",
                 details=f"Returned data: {info_data!r}",
             )
 
-        return info_data[self.name]
+        return info_data[self.instance_name]
 
     def is_mounted(
         self, *, host_source: pathlib.Path, target: pathlib.PurePath
@@ -349,7 +358,7 @@ def launch(
         :raises MultipassError: On unexpected failure.
         """
         self._multipass.launch(
-            instance_name=self.name,
+            instance_name=self.instance_name,
             image=image,
             cpus=str(cpus),
             disk=f"{disk_gb!s}G",
@@ -376,7 +385,7 @@ def mount(
 
         self._multipass.mount(
             source=host_source,
-            target=f"{self.name}:{target.as_posix()}",
+            target=f"{self.instance_name}:{target.as_posix()}",
         )
 
     def pull_file(self, *, source: pathlib.PurePath, destination: pathlib.Path) -> None:
@@ -402,7 +411,8 @@ def pull_file(self, *, source: pathlib.PurePath, destination: pathlib.Path) -> N
             raise FileNotFoundError(f"Directory not found: {str(destination.parent)!r}")
 
         self._multipass.transfer(
-            source=f"{self.name}:{source.as_posix()}", destination=str(destination)
+            source=f"{self.instance_name}:{source.as_posix()}",
+            destination=str(destination),
         )
 
     def push_file(self, *, source: pathlib.Path, destination: pathlib.PurePath) -> None:
@@ -441,7 +451,7 @@ def push_file(self, *, source: pathlib.Path, destination: pathlib.PurePath) -> N
 
             # push the file to a location where the `ubuntu` user has access
             self._multipass.transfer(
-                source=str(source), destination=f"{self.name}:{tmp_file_path}"
+                source=str(source), destination=f"{self.instance_name}:{tmp_file_path}"
             )
 
             # if the destination was a directory, then we need to specify the filename
@@ -461,7 +471,7 @@ def push_file(self, *, source: pathlib.Path, destination: pathlib.PurePath) -> N
             raise MultipassError(
                 brief=(
                     f"Failed to push file {destination.as_posix()!r}"
-                    f" into Multipass instance {self.name!r}."
+                    f" into Multipass instance {self.instance_name!r}."
                 ),
                 details=errors.details_from_called_process_error(error),
             ) from error
@@ -471,7 +481,7 @@ def start(self) -> None:
 
         :raises MultipassError: On unexpected failure.
         """
-        self._multipass.start(instance_name=self.name)
+        self._multipass.start(instance_name=self.instance_name)
 
     def stop(self, *, delay_mins: int = 0) -> None:
         """Stop instance.
@@ -480,7 +490,7 @@ def stop(self, *, delay_mins: int = 0) -> None:
 
         :raises MultipassError: On unexpected failure.
         """
-        self._multipass.stop(instance_name=self.name, delay_mins=delay_mins)
+        self._multipass.stop(instance_name=self.instance_name, delay_mins=delay_mins)
 
     def unmount(self, target: pathlib.Path) -> None:
         """Unmount mount target shared with host.
@@ -489,7 +499,7 @@ def unmount(self, target: pathlib.Path) -> None:
 
         :raises MultipassError: On failure to unmount target.
         """
-        mount = f"{self.name}:{target.as_posix()}"
+        mount = f"{self.instance_name}:{target.as_posix()}"
 
         self._multipass.umount(mount=mount)
 
@@ -498,4 +508,4 @@ def unmount_all(self) -> None:
 
         :raises MultipassError: On failure to unmount target.
         """
-        self._multipass.umount(mount=self.name)
+        self._multipass.umount(mount=self.instance_name)