CLI: init cmd_with_root

pkgs/clan-cli/clan_cli/flash/automount.py

@@ -7,14 +7,13 @@ from pathlib import Path
 from clan_cli.cmd import Log, RunOpts, run
 from clan_cli.errors import ClanError
 from clan_cli.machines.machines import Machine
-from clan_cli.nix import nix_shell
 
 log = logging.getLogger(__name__)
 
 
 @contextmanager
 def pause_automounting(
-    devices: list[Path], machine: Machine, sudo: str = "pkexec"
+    devices: list[Path], machine: Machine, request_graphical: bool = False
 ) -> Generator[None, None, None]:
     """
     Pause automounting on the device for the duration of this context
@@ -33,21 +32,32 @@ def pause_automounting(
         raise ClanError(msg)
 
     str_devs = [str(dev) for dev in devices]
-    cmd = nix_shell(
-        ["nixpkgs#pkexec"] if sudo == "pkexec" else [],
-        [sudo, str(inhibit_path), "enable", *str_devs],
-    )
+    cmd = [str(inhibit_path), "enable", *str_devs]
 
     result = run(
         cmd,
         RunOpts(
-            log=Log.BOTH, check=False, needs_user_terminal=True, prefix=machine.name
+            log=Log.BOTH,
+            check=False,
+            needs_user_terminal=True,
+            prefix=machine.name,
+            requires_root_perm=True,
+            graphical_perm=request_graphical,
         ),
     )
     if result.returncode != 0:
         machine.error("Failed to inhibit automounting")
     yield None
-    cmd = [sudo, str(inhibit_path), "disable", *str_devs]
-    result = run(cmd, RunOpts(log=Log.BOTH, check=False, prefix=machine.name))
+    cmd = [str(inhibit_path), "disable", *str_devs]
+    result = run(
+        cmd,
+        RunOpts(
+            log=Log.BOTH,
+            check=False,
+            prefix=machine.name,
+            requires_root_perm=True,
+            graphical_perm=request_graphical,
+        ),
+    )
     if result.returncode != 0:
         machine.error("Failed to re-enable automounting")

pkgs/clan-cli/clan_cli/flash/flash.py

@@ -2,14 +2,13 @@ import importlib
 import json
 import logging
 import os
-import shutil
 from dataclasses import dataclass, field
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import Any
 
 from clan_cli.api import API
-from clan_cli.cmd import Log, RunOpts, run
+from clan_cli.cmd import Log, RunOpts, cmd_with_root, run
 from clan_cli.errors import ClanError
 from clan_cli.facts.generate import generate_facts
 from clan_cli.facts.secret_modules import SecretStoreBase
@@ -47,11 +46,10 @@ def flash_machine(
     write_efi_boot_entries: bool,
     debug: bool,
     extra_args: list[str] | None = None,
-    use_user_permission: bool = False,
+    graphical: bool = False,
 ) -> None:
     devices = [Path(disk.device) for disk in disks]
-    sudo = "pkexec" if use_user_permission else "sudo"
-    with pause_automounting(devices, machine, sudo):
+    with pause_automounting(devices, machine, request_graphical=graphical):
         if extra_args is None:
             extra_args = []
         system_config_nix: dict[str, Any] = {}
@@ -110,12 +108,13 @@ def flash_machine(
             disko_install = []
 
             if os.geteuid() != 0:
-                # Use pkexec to elevate permissions if not running as root
-                perm_prefix = "pkexec" if use_user_permission else "exec sudo"
-                if shutil.which("sudo") is None:
-                    msg = "sudo is required to run disko-install as a non-root user"
-                    raise ClanError(msg)
-                wrapper = f'set -x; disko_install=$(command -v disko-install); {perm_prefix} "$disko_install" "$@"'
+                wrapper = " ".join(
+                    [
+                        "disko_install=$(command -v disko-install);",
+                        "exec",
+                        *cmd_with_root(['"$disko_install" "$@"'], graphical=graphical),
+                    ]
+                )
                 disko_install.extend(["bash", "-c", wrapper])
 
             disko_install.append("disko-install")