Support encrypted SSH private key passphrases

README.md

@@ -43,6 +43,7 @@ This integration does not create devices or entities. It only registers the `ssh
 - `username` (required) — SSH username
 - `password` — SSH password (use instead of key_file)
 - `key_file` — Path to an SSH private key file (use instead of password)
+- `passphrase` — Passphrase for an encrypted private key file (only valid when `key_file` is used)
 - `command` — Command string to execute on the host
 - `input` — Input to send to the `stdin` of the host. If this is a file path, the content of the file will be sent.
 - `check_known_hosts` (default: `true`) — Verify host key against known hosts
@@ -54,6 +55,7 @@ All parameters are optional in the raw schema except `host` and `username` — t
 #### Validation rules enforced by the service
 
 - Either `password` or `key_file` must be provided, but not both
+- `passphrase` may only be provided when `key_file` is used
 - Either `command` or `input` or both must be provided
 - If `key_file` is provided, the file must exist on the Home Assistant filesystem
 - `known_hosts` may not be provided when `check_known_hosts` is `false`

__init__.py

@@ -13,8 +13,17 @@
 from homeassistant.exceptions import ServiceValidationError
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.typing import ConfigType
-from .const import DOMAIN, SERVICE_EXECUTE, CONF_KEY_FILE, CONF_INPUT, CONST_DEFAULT_TIMEOUT, \
-    CONF_CHECK_KNOWN_HOSTS, CONF_KNOWN_HOSTS, CONF_PORT
+from .const import (
+    CONF_CHECK_KNOWN_HOSTS,
+    CONF_INPUT,
+    CONF_KEY_FILE,
+    CONF_KNOWN_HOSTS,
+    CONF_PASSPHRASE,
+    CONF_PORT,
+    CONST_DEFAULT_TIMEOUT,
+    DOMAIN,
+    SERVICE_EXECUTE,
+)
 from .coordinator import SshCommandCoordinator
 
 CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)  # pylint: disable=invalid-name
@@ -23,6 +32,7 @@
 async def _validate_service_data(hass: HomeAssistant, data: dict[str, Any]) -> None:
     has_password: bool = bool(data.get(CONF_PASSWORD))
     has_key_file: bool = bool(data.get(CONF_KEY_FILE))
+    has_passphrase: bool = data.get(CONF_PASSPHRASE) is not None
 
     if not has_password and not has_key_file:
         raise ServiceValidationError(
@@ -38,6 +48,13 @@ async def _validate_service_data(hass: HomeAssistant, data: dict[str, Any]) -> N
             translation_key="password_and_key_file",
         )
 
+    if has_passphrase and not has_key_file:
+        raise ServiceValidationError(
+            "Passphrase can only be used when key_file is provided.",
+            translation_domain=DOMAIN,
+            translation_key="passphrase_requires_key_file",
+        )
+
     has_command: bool = bool(data.get(CONF_COMMAND))
     has_input: bool = bool(data.get(CONF_INPUT))
 
@@ -73,6 +90,7 @@ async def _validate_service_data(hass: HomeAssistant, data: dict[str, Any]) -> N
             vol.Required(CONF_USERNAME): str,
             vol.Optional(CONF_PASSWORD): str,
             vol.Optional(CONF_KEY_FILE): str,
+            vol.Optional(CONF_PASSPHRASE): str,
             vol.Optional(CONF_COMMAND): str,
             vol.Optional(CONF_INPUT): str,
             vol.Optional(CONF_CHECK_KNOWN_HOSTS, default=True): bool,

const.py

@@ -5,6 +5,7 @@
 CONF_PORT = "port"
 CONF_CLIENT_KEYS = "client_keys"
 CONF_KEY_FILE = "key_file"
+CONF_PASSPHRASE = "passphrase"
 CONF_INPUT = "input"
 CONF_CHECK = "check"
 CONF_CHECK_KNOWN_HOSTS = "check_known_hosts"

coordinator.py

@@ -34,7 +34,15 @@
         )
         sys.modules["fido2.client.windows"] = None  # type: ignore[assignment]
 
-from asyncssh import HostKeyNotVerifiable, KeyImportError, PermissionDenied, connect, read_known_hosts, DEFAULT_PORT
+from asyncssh import (
+    HostKeyNotVerifiable,
+    KeyEncryptionError,
+    KeyImportError,
+    PermissionDenied,
+    connect,
+    read_known_hosts,
+    DEFAULT_PORT,
+)
 
 from .const import CONF_CONNECTION_TIMEOUT
 from homeassistant.const import CONF_USERNAME, CONF_PASSWORD, CONF_HOST, CONF_COMMAND, CONF_TIMEOUT
@@ -44,6 +52,7 @@
 from .const import (
     DOMAIN,
     CONF_KEY_FILE,
+    CONF_PASSPHRASE,
     CONF_INPUT,
     CONF_CHECK_KNOWN_HOSTS,
     CONF_KNOWN_HOSTS,
@@ -75,6 +84,7 @@ async def async_execute(self, data: dict[str, Any]) -> dict[str, Any]:
         username = data.get(CONF_USERNAME)
         password = data.get(CONF_PASSWORD)
         key_file = data.get(CONF_KEY_FILE)
+        passphrase = data.get(CONF_PASSPHRASE)
         command = data.get(CONF_COMMAND)
         input_data = data.get(CONF_INPUT)
         check_known_hosts = data.get(CONF_CHECK_KNOWN_HOSTS, True)
@@ -94,6 +104,8 @@ async def async_execute(self, data: dict[str, Any]) -> dict[str, Any]:
             CONF_KNOWN_HOSTS: await self._resolve_known_hosts(check_known_hosts, known_hosts),
             CONF_CONNECTION_TIMEOUT: timeout,
         }
+        if passphrase is not None:
+            conn_kwargs[CONF_PASSPHRASE] = passphrase
 
         run_kwargs: dict[str, Any] = {
             CONF_COMMAND: command,
@@ -114,6 +126,13 @@ async def async_execute(self, data: dict[str, Any]) -> dict[str, Any]:
                 translation_domain=DOMAIN,
                 translation_key="host_key_not_verifiable",
             ) from exc
+        except KeyEncryptionError as exc:
+            _LOGGER.warning("Invalid passphrase for %s@%s: %s", username, host, exc)
+            raise ServiceValidationError(
+                "The key file passphrase is invalid.",
+                translation_domain=DOMAIN,
+                translation_key="invalid_key_passphrase",
+            ) from exc
         except KeyImportError as exc:
             _LOGGER.warning("Invalid key file for %s@%s: %s", username, host, exc)
             raise ServiceValidationError(

services.yaml

@@ -28,12 +28,20 @@ execute:
       required: false
       selector:
         text:
+          type: password
     key_file:
       name: Key File
       description: Path to the SSH private key file for key-based authentication (optional if using username/password).
       required: false
       selector:
         text:
+    passphrase:
+      name: Passphrase
+      description: Passphrase for the private key file. Only valid when `key_file` is set.
+      required: false
+      selector:
+        text:
+          type: password
     command:
       name: Command
       description: The command to execute on the host.

strings.json

@@ -30,6 +30,10 @@
                 "name": "Key File",
                 "description": "Path to the SSH private key file for key-based authentication."
             },
+            "passphrase": {
+                "name": "Passphrase",
+                "description": "Passphrase for the SSH private key file, if the key is encrypted."
+            },
             "command": {
                 "name": "Command",
                 "description": "The command to execute on the machine."
@@ -66,6 +70,12 @@
     "key_file_not_found": {
       "message": "Could not find key file."
     },
+    "passphrase_requires_key_file": {
+      "message": "Passphrase can only be used when key_file is provided."
+    },
+    "invalid_key_passphrase": {
+      "message": "The key file passphrase is invalid."
+    },
     "invalid_key_file": {
       "message": "The key file is not a valid private key."
     },

tests/unit_tests/test_async_execute.py

@@ -12,7 +12,7 @@
 absolute_plugin_path = str(Path(__file__).parent.parent.parent.parent.absolute())
 sys.path.insert(0, absolute_plugin_path)
 
-from asyncssh import HostKeyNotVerifiable, KeyImportError, PermissionDenied
+from asyncssh import HostKeyNotVerifiable, KeyEncryptionError, KeyImportError, PermissionDenied
 
 from homeassistant.exceptions import ServiceValidationError
 
@@ -109,6 +109,24 @@ async def test_invalid_key_file(self):
 
         self.assertEqual(ctx.exception.translation_key, "invalid_key_file")
 
+    async def test_invalid_passphrase(self):
+        data = {
+            "host": "192.0.2.1",
+            "username": "user",
+            "key_file": "/home/user/.ssh/id_rsa",
+            "passphrase": "wrong",
+            "command": "echo hello",
+            "check_known_hosts": False,
+        }
+        service_call = self._make_service_call(data)
+
+        with patch("pathlib.Path.exists", return_value=True):
+            with patch("ssh_command.coordinator.connect", return_value=_MockConnectRaises(KeyEncryptionError("Incorrect passphrase"))):
+                with self.assertRaises(ServiceValidationError) as ctx:
+                    await self.handler(service_call)
+
+        self.assertEqual(ctx.exception.translation_key, "invalid_key_passphrase")
+
     async def test_permission_denied(self):
         service_call = self._make_service_call(SERVICE_DATA_BASE)
 
@@ -145,6 +163,44 @@ async def test_other_oserror_is_reraised(self):
             with self.assertRaises(OSError):
                 await self.handler(service_call)
 
+    async def test_passphrase_is_forwarded_to_connect(self):
+        mock_conn = self._make_mock_conn(stdout="ok", stderr="", exit_status=0)
+        data = {
+            "host": "192.0.2.1",
+            "username": "user",
+            "key_file": "/home/user/.ssh/id_rsa",
+            "passphrase": "topsecret",
+            "command": "echo hello",
+            "check_known_hosts": False,
+        }
+        service_call = self._make_service_call(data)
+
+        with patch("pathlib.Path.exists", return_value=True):
+            with patch("ssh_command.coordinator.connect", return_value=_MockConnect(mock_conn)) as mock_connect:
+                await self.handler(service_call)
+
+        call_kwargs = mock_connect.call_args[1]
+        self.assertEqual(call_kwargs["passphrase"], "topsecret")
+
+    async def test_empty_passphrase_is_forwarded_to_connect(self):
+        mock_conn = self._make_mock_conn(stdout="ok", stderr="", exit_status=0)
+        data = {
+            "host": "192.0.2.1",
+            "username": "user",
+            "key_file": "/home/user/.ssh/id_rsa",
+            "passphrase": "",
+            "command": "echo hello",
+            "check_known_hosts": False,
+        }
+        service_call = self._make_service_call(data)
+
+        with patch("pathlib.Path.exists", return_value=True):
+            with patch("ssh_command.coordinator.connect", return_value=_MockConnect(mock_conn)) as mock_connect:
+                await self.handler(service_call)
+
+        call_kwargs = mock_connect.call_args[1]
+        self.assertEqual(call_kwargs["passphrase"], "")
+
     async def test_input_from_file(self):
         with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as tf:
             tf.write("file content\n")

tests/unit_tests/test_validate_service_data.py

@@ -34,6 +34,15 @@ async def test_password_and_key_file_raises(self):
             await _validate_service_data(self.mock_hass, {"password": "secret", "key_file": "/home/user/.ssh/id_rsa", "command": "echo hi"})
         self.assertEqual(ctx.exception.translation_key, "password_and_key_file")
 
+    async def test_passphrase_without_key_file_raises(self):
+        with self.assertRaises(ServiceValidationError) as ctx:
+            await _validate_service_data(self.mock_hass, {"password": "secret", "passphrase": "topsecret", "command": "echo hi"})
+        self.assertEqual(ctx.exception.translation_key, "passphrase_requires_key_file")
+
+    async def test_empty_passphrase_with_key_file_is_allowed(self):
+        with patch("pathlib.Path.exists", return_value=True):
+            await _validate_service_data(self.mock_hass, {"key_file": "/home/user/.ssh/id_rsa", "passphrase": "", "input": "some text"})
+
     async def test_no_command_no_input_raises(self):
         with self.assertRaises(ServiceValidationError) as ctx:
             await _validate_service_data(self.mock_hass, {"password": "secret"})

translations/de.json

@@ -30,6 +30,10 @@
                 "name": "Schlüsseldatei",
                 "description": "Der Pfad zur privaten Schlüsseldatei für die SSH-Anmeldung"
             },
+            "passphrase": {
+                "name": "Passphrase",
+                "description": "Passphrase für die SSH-Schlüsseldatei, falls der Schlüssel verschlüsselt ist."
+            },
             "command": {
                 "name": "Befehl",
                 "description": "Der Befehl, der auf dem Host ausgeführt werden soll."
@@ -66,6 +70,12 @@
     "key_file_not_found": {
       "message": "Konnte Schlüsseldatei nicht finden."
     },
+    "passphrase_requires_key_file": {
+      "message": "Passphrase kann nur verwendet werden, wenn key_file angegeben ist."
+    },
+    "invalid_key_passphrase": {
+      "message": "Die Passphrase für die Schlüsseldatei ist ungültig."
+    },
     "invalid_key_file": {
       "message": "Die Schlüsseldatei ist kein gültiger privater Schlüssel."
     },

translations/en.json

@@ -30,6 +30,10 @@
                 "name": "Key File",
                 "description": "Path to the SSH private key file for key-based authentication."
             },
+            "passphrase": {
+                "name": "Passphrase",
+                "description": "Passphrase for the SSH private key file, if the key is encrypted."
+            },
             "command": {
                 "name": "Command",
                 "description": "The command to execute on the machine."
@@ -66,6 +70,12 @@
     "key_file_not_found": {
       "message": "Could not find key file."
     },
+    "passphrase_requires_key_file": {
+      "message": "Passphrase can only be used when key_file is provided."
+    },
+    "invalid_key_passphrase": {
+      "message": "The key file passphrase is invalid."
+    },
     "invalid_key_file": {
       "message": "The key file is not a valid private key."
     },