Add dda.utils.platform.get_machine_id function

Author: ofek

CHANGELOG.md

@@ -12,6 +12,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 - The paths used to search for local commands are no longer added to the Python search path and instead a sibling directory `pythonpath` is used
 
+***Added:***
+
+- Add `dda.utils.platform.get_machine_id` function
+
 ***Fixed:***
 
 - Properly persist Python search path modifications for local commands when using subprocesses

docs/reference/api/platform.md

@@ -0,0 +1,9 @@
+# Platform reference
+
+-----
+
+::: dda.utils.platform
+    options:
+      show_if_no_docstring: false
+      show_root_heading: false
+      show_root_toc_entry: false

mkdocs.yml

@@ -67,9 +67,10 @@ nav:
     - Command: reference/api/command.md
     - Application: reference/api/app.md
     - Process: reference/api/process.md
-    - Tools: reference/api/tools.md
     - Filesystem: reference/api/fs.md
+    - Platform: reference/api/platform.md
     - Retries: reference/api/retry.md
+    - Tools: reference/api/tools.md
     - CI: reference/api/ci.md
     - Config: reference/api/config.md
     - Constants: reference/api/constants.md

src/dda/utils/platform.py

@@ -0,0 +1,158 @@
+# SPDX-FileCopyrightText: 2024-present Datadog, Inc. <dev@datadoghq.com>
+#
+# SPDX-License-Identifier: MIT
+from __future__ import annotations
+
+import contextlib
+import os
+import sys
+from functools import cache
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from uuid import UUID
+
+
+if sys.platform == "win32":
+    __PLATFORM_ID = "windows"
+    __PLATFORM_NAME = "Windows"
+    __DEFAULT_SHELL = os.environ.get("SHELL", os.environ.get("COMSPEC", "cmd"))
+
+    def __join_command_args(args: list[str]) -> str:
+        import subprocess
+
+        return subprocess.list2cmdline(args)
+
+    def __get_machine_id() -> UUID | None:
+        import winreg
+        from uuid import UUID
+
+        with winreg.OpenKey(
+            winreg.HKEY_LOCAL_MACHINE,
+            r"SOFTWARE\Microsoft\Cryptography",
+            0,
+            winreg.KEY_READ | winreg.KEY_WOW64_64KEY,
+        ) as key:
+            value, _ = winreg.QueryValueEx(key, "MachineGuid")
+            return UUID(value)
+
+
+elif sys.platform == "darwin":
+    __PLATFORM_ID = "macos"
+    __PLATFORM_NAME = "macOS"
+    __DEFAULT_SHELL = os.environ.get("SHELL", "zsh")
+
+    def __join_command_args(args: list[str]) -> str:
+        import shlex
+
+        return shlex.join(args)
+
+    def __get_machine_id() -> UUID | None:
+        import re
+        import subprocess
+        from uuid import UUID
+
+        process = subprocess.run(
+            ["ioreg", "-c", "IOPlatformExpertDevice", "-d2"],  # noqa: S607
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            check=True,
+        )
+        match = re.search(rb'"IOPlatformUUID"\s*=\s*"([^"]+)"', process.stdout)
+        return UUID(match.group(1).decode("utf-8")) if match else None
+
+else:
+    __PLATFORM_ID = "linux"
+    __PLATFORM_NAME = "Linux"
+    __DEFAULT_SHELL = os.environ.get("SHELL", "bash")
+
+    def __join_command_args(args: list[str]) -> str:
+        import shlex
+
+        return shlex.join(args)
+
+    def __get_machine_id() -> UUID | None:
+        from uuid import UUID
+
+        for path in (
+            # https://utcc.utoronto.ca/~cks/space/blog/linux/DMIDataInSysfs
+            # Section 7.2.1 of https://www.dmtf.org/sites/default/files/standards/documents/DSP0134_3.0.0.pdf
+            # https://cloud.google.com/compute/docs/instances/get-uuid#linux
+            "/sys/class/dmi/id/product_uuid",
+            # https://www.freedesktop.org/software/systemd/man/latest/machine-id.html
+            "/etc/machine-id",
+            # https://wiki.debian.org/MachineId
+            "/var/lib/dbus/machine-id",
+        ):
+            with contextlib.suppress(Exception), open(path, encoding="utf-8") as f:
+                return UUID(f.read().strip())
+
+        return None
+
+
+PLATFORM_ID = __PLATFORM_ID
+"""
+A short identifier for the current platform. Known values:
+
+- `linux`
+- `windows`
+- `macos`
+"""
+PLATFORM_NAME = __PLATFORM_NAME
+"""
+The human readable name of the current platform. Known values:
+
+- Linux
+- Windows
+- macOS
+"""
+DEFAULT_SHELL = __DEFAULT_SHELL
+"""
+The default shell for the current platform. Values are taken from environment variables, with
+platform-specific fallbacks.
+
+Platform | Environment variables | Fallback
+--- | --- | ---
+`linux` | `SHELL` | `bash`
+`windows` | `SHELL`, `COMSPEC` | `cmd`
+`macos` | `SHELL` | `zsh`
+"""
+
+
+def join_command_args(args: list[str]) -> str:
+    """
+    Create a valid shell command from a list of arguments.
+
+    Parameters:
+        args: A list of command line arguments.
+
+    Returns:
+        A single string of command line arguments.
+    """
+    return __join_command_args(args)
+
+
+@cache
+def get_machine_id() -> UUID:
+    """
+    Get a unique identifier for the current machine that is consistent across reboots and different
+    processes. The following platform-specific methods are given priority:
+
+    Platform | Method
+    --- | ---
+    `linux` | The [`/sys/class/dmi/id/product_uuid`](https://utcc.utoronto.ca/~cks/space/blog/linux/DMIDataInSysfs), [`/etc/machine-id`](https://www.freedesktop.org/software/systemd/man/latest/machine-id.html) or [`/var/lib/dbus/machine-id`](https://wiki.debian.org/MachineId) files
+    `windows` | The `HKEY_LOCAL_MACHINE\\SOFTWARE\\Microsoft\\Cryptography\\MachineGuid` registry key
+    `macos` | The [`IOPlatformUUID`](https://developer.apple.com/documentation/iokit/kioplatformuuidkey/) property of the [`IOPlatformExpertDevice`](https://developer.apple.com/library/archive/technotes/tn1103/_index.html) node in the [I/O Registry](https://developer.apple.com/library/archive/documentation/DeviceDrivers/Conceptual/IOKitFundamentals/TheRegistry/TheRegistry.html#//apple_ref/doc/uid/TP0000014-TP9)
+
+    As a fallback, the ID will be generated using the MAC address.
+    """
+    machine_id: UUID | None = None
+    with contextlib.suppress(Exception):
+        machine_id = __get_machine_id()
+
+    if machine_id is not None:
+        return machine_id
+
+    import uuid
+
+    return uuid.uuid5(uuid.NAMESPACE_DNS, str(uuid.getnode()))

src/dda/utils/platform/__init__.py

@@ -47,3 +47,25 @@ def test_name(self):
     def test_default_shell(self):
         expected_shell = os.environ.get("SHELL", "bash")
         assert platform.DEFAULT_SHELL == expected_shell  # noqa: SIM300
+
+
+class TestMachineId:
+    def test_platform_specific(self):
+        machine_id = str(platform.get_machine_id())
+        assert str(platform.get_machine_id()) == machine_id
+
+        parts = machine_id.split("-")
+        lengths = list(map(len, parts))
+        assert lengths == [8, 4, 4, 4, 12]
+
+    def test_fallback(self, mocker):
+        mocker.patch("dda.utils.platform.__get_machine_id", return_value=None)
+
+        platform.get_machine_id.cache_clear()
+        machine_id = str(platform.get_machine_id())
+        platform.get_machine_id.cache_clear()
+        assert str(platform.get_machine_id()) == machine_id
+
+        parts = machine_id.split("-")
+        lengths = list(map(len, parts))
+        assert lengths == [8, 4, 4, 4, 12]