README.md

@@ -121,7 +121,7 @@ The mixin sits in `DeviceTypeDriver` and only declares, so `hasattr(driver, "run
 - `container_engines()` says which engines the host offers (`[{"engine": "docker", "api": "docker-engine"}]`).
 - `open_container_engine(engine)` returns a `ContainerEngineConnection`:
   - `open_api()` is a stream to the engine's API (`docker system dial-stdio`);
-  - `run_cli(args)` and `stream_cli(args)` run the engine's CLI with arguments the caller chooses.
+  - `run_cli(args)` and `stream_cli(args)` run the engine's CLI with arguments the caller chooses. With `privileged=True` the driver runs it the way `run_command` gains root, for an engine that refuses the login user.
 - The driver decides only *how* the engine is reached. Its hook `_container_engine_binary(engine)` returns, for QNAP, the Container Station path, and the caller never sees it.
 
 **What runs on the engine, and what to do with it, is netOrk's** (NetOrk/netork#765): the container model, the Engine API requests, compose, updates. Above the connection everything is specific to the service, so this package abstracts the connection and nothing more.

napalm_device_types/container_engine.py

@@ -65,19 +65,34 @@ class ContainerEngineConnection:
         return self._driver.open_stream(self._command(["system", "dial-stdio"]))
 
     def run_cli(
-        self, args: Sequence[str], *, stdin: Optional[bytes] = None, timeout: float = _CLI_TIMEOUT
+        self,
+        args: Sequence[str],
+        *,
+        stdin: Optional[bytes] = None,
+        timeout: float = _CLI_TIMEOUT,
+        privileged: bool = False,
     ) -> CommandResult:
-        """Run the engine's CLI with *args* and wait for it."""
-        return self._driver.run_command(self._command(args), timeout=timeout, stdin=stdin)
+        """Run the engine's CLI with *args* and wait for it.
 
-    def stream_cli(self, args: Sequence[str], *, merge_stderr: bool = False) -> ByteStream:
+        *privileged* runs it as root, the way the driver gains root for any
+        command: for a call the engine refused to the login user.
+        """
+        return self._driver.run_command(
+            self._command(args), privileged=privileged, timeout=timeout, stdin=stdin
+        )
+
+    def stream_cli(
+        self, args: Sequence[str], *, merge_stderr: bool = False, privileged: bool = False
+    ) -> ByteStream:
         """Start the engine's CLI with *args* and stream its output.
 
         *merge_stderr* folds stderr into the stream, for tools that report
-        progress there (``compose up``).
+        progress there (``compose up``); *privileged* as for :meth:`run_cli`.
         """
         command = self._command(args)
-        return self._driver.open_stream(f"{command} 2>&1" if merge_stderr else command)
+        return self._driver.open_stream(
+            f"{command} 2>&1" if merge_stderr else command, privileged=privileged
+        )
 
 
 class ContainerEngineMixin:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "2.6.0"
+version = "2.7.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_container_engine.py

@@ -131,3 +131,29 @@ def test_an_os_driver_does_not_get_container_engine_access_by_role():
     too, and `hasattr(driver, "open_container_engine")` has to stay truthful."""
     assert not issubclass(OSDriver, ContainerEngineMixin)
     assert not hasattr(OSDriver, "open_container_engine")
+
+
+class TestPrivilegedCli:
+    """A refused engine call can be repeated as root (netork#773): the driver
+    knows the binary and how it gains root, so the connection carries it."""
+
+    def test_run_cli_passes_privileged_to_the_channel(self):
+        driver = FakeDriver()
+
+        driver.open_container_engine("docker").run_cli(["restart", "web"], privileged=True)
+
+        assert driver.commands == [("docker restart web", True, 120, None)]
+
+    def test_run_cli_is_unprivileged_by_default(self):
+        driver = FakeDriver()
+
+        driver.open_container_engine("docker").run_cli(["ps"])
+
+        assert driver.commands[0][1] is False
+
+    def test_stream_cli_passes_privileged_too(self):
+        driver = FakeDriver()
+
+        driver.open_container_engine("docker").stream_cli(["pull", "nginx"], privileged=True)
+
+        assert driver.streams == [("docker pull nginx", True)]