blob: cdcb09567316db79de6fa99abb2115f2ce7c967e (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
# /config/custom_components/patch_asyncssh/__init__.py
from __future__ import annotations
import logging
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass, config):
"""Patch asyncssh at HA startup, then return True to finish setup."""
try:
import asyncssh
# Re-expose ReadHostKeysPolicy at top-level for libs that expect it
try:
from asyncssh.hostkeys import ReadHostKeysPolicy
asyncssh.ReadHostKeysPolicy = ReadHostKeysPolicy # type: ignore[attr-defined]
except Exception:
pass
# Patch only once
if not getattr(asyncssh, "_ha_patched_connect", False):
_real_connect = asyncssh.connect
def _patched_connect(*args, **kwargs):
# Expand allowed *server* host key algorithms to match Dropbear
algs = list(
kwargs.get("server_host_key_algs")
or kwargs.get("host_key_algs")
or []
)
wanted = ["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-256"]
for a in wanted:
if a not in algs:
algs.append(a)
kwargs["server_host_key_algs"] = algs or wanted
return _real_connect(*args, **kwargs)
asyncssh.connect = _patched_connect # type: ignore[assignment]
asyncssh._ha_patched_connect = True # type: ignore[attr-defined]
_LOGGER.info(
"patch_asyncssh: asyncssh.connect patched (added algs %s)",
["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-256"],
)
else:
_LOGGER.debug("patch_asyncssh: asyncssh.connect already patched")
except Exception:
_LOGGER.exception("patch_asyncssh: failed to patch asyncssh")
return True
|
