Handle KeyImportError for invalid SSH private key files

coordinator.py

@@ -15,7 +15,7 @@
 from pathlib import Path
 from typing import Any
 
-from asyncssh import HostKeyNotVerifiable, PermissionDenied, connect, read_known_hosts
+from asyncssh import HostKeyNotVerifiable, KeyImportError, PermissionDenied, connect, read_known_hosts
 
 from homeassistant.const import CONF_USERNAME, CONF_PASSWORD, CONF_HOST, CONF_COMMAND, CONF_TIMEOUT
 from homeassistant.core import HomeAssistant
@@ -92,6 +92,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 KeyImportError as exc:
+            _LOGGER.warning("Invalid key file for %s@%s: %s", username, host, exc)
+            raise ServiceValidationError(
+                "The key file is not a valid private key.",
+                translation_domain=DOMAIN,
+                translation_key="invalid_key_file",
+            ) from exc
         except PermissionDenied as exc:
             _LOGGER.warning("SSH login failed for %s@%s: %s", username, host, exc)
             raise ServiceValidationError(

strings.json

@@ -59,6 +59,9 @@
     "key_file_not_found": {
       "message": "Could not find key file."
     },
+    "invalid_key_file": {
+      "message": "The key file is not a valid private key."
+    },
     "known_hosts_with_check_disabled": {
       "message": "Known hosts provided while check known hosts is disabled."
     },

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, PermissionDenied
+from asyncssh import HostKeyNotVerifiable, KeyImportError, PermissionDenied
 
 from homeassistant.exceptions import ServiceValidationError
 
@@ -100,6 +100,15 @@ async def test_host_key_not_verifiable(self):
 
         self.assertEqual(ctx.exception.translation_key, "host_key_not_verifiable")
 
+    async def test_invalid_key_file(self):
+        service_call = self._make_service_call(SERVICE_DATA_BASE)
+
+        with patch("ssh_command.coordinator.connect", return_value=_MockConnectRaises(KeyImportError("Invalid private key"))):
+            with self.assertRaises(ServiceValidationError) as ctx:
+                await self.handler(service_call)
+
+        self.assertEqual(ctx.exception.translation_key, "invalid_key_file")
+
     async def test_permission_denied(self):
         service_call = self._make_service_call(SERVICE_DATA_BASE)
 

translations/de.json

@@ -55,6 +55,9 @@
     "key_file_not_found": {
       "message": "Konnte Schlüsseldatei nicht finden."
     },
+    "invalid_key_file": {
+      "message": "Die Schlüsseldatei ist kein gültiger privater Schlüssel."
+    },
     "known_hosts_with_check_disabled": {
       "message": "Bekannte Hosts wurden angegeben, obwohl die Überprüfung deaktiviert ist."
     },

translations/en.json

@@ -55,6 +55,9 @@
     "key_file_not_found": {
       "message": "Could not find key file."
     },
+    "invalid_key_file": {
+      "message": "The key file is not a valid private key."
+    },
     "known_hosts_with_check_disabled": {
       "message": "Known hosts provided while check known hosts is disabled."
     },