Merge pull request #3 from yechielb2000/feature/parallel-rce

Author: yechielb2000

README.md

@@ -161,6 +161,33 @@ with RemoteMachine("example.com", "user") as conn:
     all_vars = conn.env.list()
 ```
 
+## Parallel Execution
+
+Run commands on multiple machines concurrently for efficient bulk operations.
+
+```python
+from remote_machine import RemoteMachine
+from remote_machine.parallel import ParallelExecutor
+
+# Create connections to multiple hosts
+clients = [
+    RemoteMachine("host1.example.com", "user", key_path="~/.ssh/id_rsa"),
+    RemoteMachine("host2.example.com", "user", key_path="~/.ssh/id_rsa"),
+    RemoteMachine("host3.example.com", "user", key_path="~/.ssh/id_rsa"),
+]
+
+# Execute command in parallel
+executor = ParallelExecutor(max_workers=5)
+results = executor.run(clients, "uptime")
+
+# Process results
+for result in results:
+    if result.success:
+        print(f"{result.host}: {result.output}")
+    else:
+        print(f"{result.host} failed: {result.error}")
+```
+
 ## Logging & Telemetry
 
 RemoteMachine supports structured logging and optional telemetry for monitoring and observability.

remote_machine/actions/sys.py

@@ -59,11 +59,11 @@ def info(self) -> SystemInfo:
 
     def uname(self) -> UnameInfo:
         """Get system name and information as a dataclass."""
-        sysname = self.protocol.run_command("uname -s", self.state)
-        nodename = self.protocol.run_command("uname -n", self.state)
-        release = self.protocol.run_command("uname -r", self.state)
-        version = self.protocol.run_command("uname -v", self.state)
-        machine = self.protocol.run_command("uname -m", self.state)
+        sysname = self.protocol.run_command("uname -s", self.state).strip()
+        nodename = self.protocol.run_command("uname -n", self.state).strip()
+        release = self.protocol.run_command("uname -r", self.state).strip()
+        version = self.protocol.run_command("uname -v", self.state).strip()
+        machine = self.protocol.run_command("uname -m", self.state).strip()
 
         return UnameInfo(
             sysname=sysname,

remote_machine/core.py

@@ -7,6 +7,7 @@
 
 from remote_machine import actions
 from remote_machine.errors import ProtocolNotAvailable, PermissionDenied
+from remote_machine.errors.error_mapper import ErrorMapper
 from remote_machine.models.capabilities import Capabilities
 from remote_machine.models.remote_state import RemoteState
 from remote_machine.protocols.ssh import SSHProtocol
@@ -146,6 +147,28 @@ def _initialize_state(self) -> None:
         result = ssh.exec("sudo -n true", self.state)
         self.state.has_sudo = result.exit_code == 0
 
+    def run(self, command: str) -> str:
+        """Run a command on the remote machine and return stdout.
+
+        Args:
+            command: Command to execute
+
+        Returns:
+            stdout as string
+
+        Raises:
+            Appropriate exception if command fails
+        """
+        ssh: SSHProtocol = self._protocols["ssh"]
+        result = ssh.exec(command, self.state)
+        ErrorMapper.raise_if_error(result)
+        return result.stdout
+
+    @property
+    def host(self) -> str:
+        """Return the remote host address."""
+        return self._protocols["ssh"].host
+
     def add_ssh_layer(self, ssh: SSHProtocol) -> None:
         """Add an SSHProtocol to the layers (for tunnel chaining)."""
         self._ssh_layers.append(ssh)

remote_machine/parallel/__init__.py

@@ -0,0 +1,6 @@
+"""Parallel execution module."""
+
+from .executor import ParallelExecutor
+from .result import ParallelResult
+
+__all__ = ["ParallelExecutor", "ParallelResult"]

remote_machine/parallel/executor.py

@@ -0,0 +1,75 @@
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Iterable, List
+
+from ..logging import get_logger
+from .result import ParallelResult
+
+logger = get_logger(__name__)
+
+
+class ParallelExecutor:
+    def __init__(self, max_workers: int = 10):
+        if max_workers <= 0:
+            raise ValueError("max_workers must be > 0")
+        self.max_workers = max_workers
+
+    def run(self, clients: Iterable, command: str) -> List[ParallelResult]:
+        """
+        Run the same command across multiple RemoteMachine instances in parallel.
+        """
+
+        results: List[ParallelResult] = []
+
+        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
+            future_map = {
+                executor.submit(self._execute, client, command): client
+                for client in clients
+            }
+
+            for future in as_completed(future_map):
+                result = future.result()
+                results.append(result)
+
+        return results
+
+    def _execute(self, client, command: str) -> ParallelResult:
+        start = time.time()
+        try:
+            output = client.run(command)
+            duration_ms = (time.time() - start) * 1000
+
+            logger.info(
+                "parallel_command_success",
+                extra={
+                    "host": client.host,
+                    "duration_ms": duration_ms,
+                },
+            )
+
+            return ParallelResult(
+                host=client.host,
+                success=True,
+                output=output,
+                error=None,
+                duration_ms=duration_ms,
+            )
+
+        except Exception as e:
+            duration_ms = (time.time() - start) * 1000
+
+            logger.error(
+                "parallel_command_failed",
+                extra={
+                    "host": client.host,
+                    "error": str(e),
+                },
+            )
+
+            return ParallelResult(
+                host=client.host,
+                success=False,
+                output=None,
+                error=e,
+                duration_ms=duration_ms,
+            )

remote_machine/parallel/result.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass
+class ParallelResult:
+    host: str
+    success: bool
+    output: Any | None
+    error: Exception | None
+    duration_ms: float

tests/test_net_actions.py

@@ -23,6 +23,12 @@ def exec(self, command: str, state):
             command=command, stdout="", stderr="", exit_code=0, success=True
         )
 
+    def run_command(self, command: str, state, thread: bool = False) -> str:
+        result = self.exec(command, state)
+        if result.exit_code != 0:
+            raise Exception(f"Command failed: {result.stderr}")
+        return result.stdout
+
 
 def test_interfaces_uses_parser(monkeypatch):
     ip_mod = types.ModuleType("linux_parsers.parsers.network.ip")

tests/test_parallel.py

@@ -0,0 +1,104 @@
+"""Unit tests for parallel execution."""
+
+import pytest
+pytestmark = [pytest.mark.parallel]
+
+import os
+import sys
+import time
+
+# Make the package importable in tests run in this environment
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from remote_machine.parallel import ParallelExecutor, ParallelResult
+from remote_machine.models.command_result import CommandResult
+from remote_machine.models.remote_state import RemoteState
+
+
+class FakeRemoteMachine:
+    def __init__(self, host: str, responses: dict[str, str] = None, should_fail: bool = False):
+        self.host = host
+        self.responses = responses or {}
+        self.should_fail = should_fail
+
+    def run(self, command: str) -> str:
+        if self.should_fail:
+            raise Exception("Simulated failure")
+        return self.responses.get(command, f"executed {command} on {self.host}")
+
+
+def test_parallel_executor_success():
+    """Test successful parallel execution."""
+    clients = [
+        FakeRemoteMachine("host1"),
+        FakeRemoteMachine("host2"),
+        FakeRemoteMachine("host3"),
+    ]
+
+    executor = ParallelExecutor(max_workers=2)
+    results = executor.run(clients, "uptime")
+
+    assert len(results) == 3
+    for result in results:
+        assert result.success
+        assert result.error is None
+        assert result.duration_ms >= 0
+        assert "uptime" in result.output
+        assert result.host in ["host1", "host2", "host3"]
+
+
+def test_parallel_executor_failure():
+    """Test parallel execution with failures."""
+    clients = [
+        FakeRemoteMachine("host1"),
+        FakeRemoteMachine("host2", should_fail=True),
+        FakeRemoteMachine("host3"),
+    ]
+
+    executor = ParallelExecutor(max_workers=2)
+    results = executor.run(clients, "uptime")
+
+    assert len(results) == 3
+
+    success_results = [r for r in results if r.success]
+    failure_results = [r for r in results if not r.success]
+
+    assert len(success_results) == 2
+    assert len(failure_results) == 1
+
+    failure = failure_results[0]
+    assert failure.host == "host2"
+    assert failure.output is None
+    assert isinstance(failure.error, Exception)
+    assert str(failure.error) == "Simulated failure"
+
+
+def test_parallel_executor_max_workers():
+    """Test that max_workers limits concurrency."""
+    # This is hard to test directly, but we can check the parameter validation
+    with pytest.raises(ValueError):
+        ParallelExecutor(max_workers=0)
+
+    with pytest.raises(ValueError):
+        ParallelExecutor(max_workers=-1)
+
+    # Valid
+    executor = ParallelExecutor(max_workers=5)
+    assert executor.max_workers == 5
+
+
+def test_parallel_result_structure():
+    """Test ParallelResult dataclass structure."""
+    result = ParallelResult(
+        host="testhost",
+        success=True,
+        output="output",
+        error=None,
+        duration_ms=123.45
+    )
+
+    assert result.host == "testhost"
+    assert result.success is True
+    assert result.output == "output"
+    assert result.error is None
+    assert result.duration_ms == 123.45

tests/test_ps_actions.py

@@ -22,6 +22,12 @@ def exec(self, command: str, state: RemoteState) -> CommandResult:
                 return CommandResult(command=command, stdout=out, stderr="", exit_code=0)
         return CommandResult(command=command, stdout="", stderr="", exit_code=0)
 
+    def run_command(self, command: str, state: RemoteState, thread: bool = False) -> str:
+        result = self.exec(command, state)
+        if result.exit_code != 0:
+            raise Exception(f"Command failed: {result.stderr}")
+        return result.stdout
+
 
 def test_list_uses_parser(monkeypatch):
     ps_mod = types.ModuleType("linux_parsers.parsers.process.ps")

tests/test_sys_actions.py

@@ -30,6 +30,12 @@ def exec(self, command: str, state: RemoteState) -> CommandResult:
         # default empty
         return CommandResult(command=command, stdout="", stderr="", exit_code=0)
 
+    def run_command(self, command: str, state: RemoteState, thread: bool = False) -> str:
+        result = self.exec(command, state)
+        if result.exit_code != 0:
+            raise Exception(f"Command failed: {result.stderr}")
+        return result.stdout
+
 
 def test_uname_parsing():
     responses = {