outbound v0.6.4 [base]

"""outbound.drivers — реестр: kind -> драйвер, op -> драйвер.

Сквозные режимы применяются здесь, а не в драйверах:
  via:   имя другой точки kind=ssh — ssh идёт через ProxyJump (-J);
         остальным kind via пока запрещён (явная ошибка, не молча).
  proxy: только kind=http (opts.proxy); остальным — явная ошибка.
  dryrun игнорирует via/proxy.
"""
import logging

from . import base as _b
from . import file as _file
from . import http as _http
from . import mgmt as _mgmt
from . import msg as _msg
from . import net as _net
from . import ssh as _ssh

logger = logging.getLogger(__name__)

_DRIVERS = (_net, _http, _ssh, _msg, _file, _mgmt)

KIND2DRV = {}
OP2DRV = {}
for _d in _DRIVERS:
    for _k in getattr(_d, "KINDS", ()):
        KIND2DRV[_k] = _d
    for _op in getattr(_d, "OPS", {}):
        OP2DRV[_op] = _d


def kinds():
    return sorted(KIND2DRV)


def ops():
    return sorted(OP2DRV)


def supports(kind, op):
    """Умеет ли драйвер kind выполнять op (для маршрутизации привязок)."""
    drv = KIND2DRV.get((kind or "").strip())
    return bool(drv and op in getattr(drv, "OPS", {}))


def dispatch(ep, secret, op, params, via_jump=None):
    """Выполнить op на точке. Возвращает dict (всегда, без исключений)."""
    kind = (ep.get("kind") or "").strip()
    drv = KIND2DRV.get(kind)
    if drv is None:
        return _b.err("неизвестный kind точки: %s" % kind)
    if op not in getattr(drv, "OPS", {}):
        return _b.err("точка %s (%s) не умеет op=%s" % (
            ep.get("name"), kind, op))
    if ep.get("proxy") and kind != "http":
        return _b.err("proxy поддерживается только kind=http")
    if via_jump and kind != "ssh":
        return _b.err("via (jump-host) поддерживается только kind=ssh")
    fn = drv.OPS[op]
    try:
        if via_jump and kind == "ssh":
            return fn(ep, secret, params or {}, jump=via_jump)
        return fn(ep, secret, params or {})
    except TypeError:
        # драйвер без jump-kwarg (на будущее) — обычный вызов
        try:
            return fn(ep, secret, params or {})
        except Exception as e:  # noqa: BLE001
            logger.exception("outbound.dispatch %s", op)
            return _b.err("внутренняя ошибка драйвера: %s" % e)
    except Exception as e:  # noqa: BLE001
        logger.exception("outbound.dispatch %s", op)
        return _b.err("внутренняя ошибка драйвера: %s" % e)


def check_health(ep, secret, via_jump=None):
    drv = KIND2DRV.get((ep.get("kind") or "").strip())
    if drv is None or not hasattr(drv, "health"):
        return _b.err("нет health для kind=%s" % ep.get("kind"))
    try:
        if via_jump:
            return drv.health(ep, secret, jump=via_jump)
        return drv.health(ep, secret)
    except TypeError:
        try:
            return drv.health(ep, secret)
        except Exception as e:  # noqa: BLE001
            return _b.err("health: %s" % e)
    except Exception as e:  # noqa: BLE001
        return _b.err("health: %s" % e)

"""outbound.drivers.base — общие помощники драйверов.

Драйвер = модуль с:
  KINDS = ("kind1", ...)   какие kind обслуживает
  OPS = {"op": func}       операции; func(ep, secret, params) -> dict
  health(ep, secret)       проверка точки -> {"ok": bool, "ms": int, ...}
Контракт результата операции: {"ok": True, ...} либо {"ok": False,
"error": "..."}. Секреты в результаты/ошибки не попадают.
"""
import importlib
import logging
import subprocess
import time

logger = logging.getLogger(__name__)


def now_ms():
    return int(time.time() * 1000)


def err(msg, **extra):
    out = {"ok": False, "error": str(msg)[:500]}
    out.update(extra)
    return out


def ok(**extra):
    out = {"ok": True}
    out.update(extra)
    return out


def opt(ep, key, default=None):
    try:
        return (ep.get("opts") or {}).get(key, default)
    except Exception:  # noqa: BLE001
        return default


def addr(ep, key, default=""):
    try:
        return (ep.get("address") or {}).get(key, default)
    except Exception:  # noqa: BLE001
        return default


def timeout(ep, default=15):
    try:
        return int(opt(ep, "timeout", default) or default)
    except (TypeError, ValueError):
        return default


def lazy(modname):
    """Импорт опциональной зависимости: (модуль|None)."""
    try:
        return importlib.import_module(modname)
    except Exception:  # noqa: BLE001
        return None


def missing(pkg, kind):
    return err("драйвер %s: нет пакета %s (requirements.txt модуля "
               "outbound)" % (kind, pkg))


def decode_out(data, codepage=None):
    """Текст из байтов вывода: UTF-8 строго → кодовая страница хоста → replace.

    console-утилиты Windows по SSH (sc, net, ...) отвечают OEM cp866, а
    text=True декодировал их по локали (UTF-8) и русские строки превращались
    в U+FFFD ещё до модулей-потребителей (каша в деталях статуса служб).
    codepage (opts.ssh_codepage) — явная кодовая страница точки: она важнее
    эвристики, потому что вывод в cp1251/Latin-1 эвристика не различает.
    Валидный UTF-8 и ASCII не меняются.
    """
    if isinstance(data, str):
        return data
    data = data or b""
    for enc in (codepage, "utf-8", "cp866"):
        if not enc:
            continue
        try:
            return data.decode(enc)
        except (UnicodeDecodeError, LookupError):
            continue
    return data.decode("utf-8", "replace")


def run(cmd, timeout_s=15, env=None, input_text=None, max_chars=8000,
        codepage=None):
    """subprocess без shell. Возвращает (rc, stdout, stderr).

    Вывод декодируется decode_out (байты -> текст): stdout режется на
    max_chars уже ПОСЛЕ декода (символы, не байты: иначе срез резал бы
    мультибайтные строки и лимиты разъезжались; discover каталогов
    передаёт больше через params max_chars). input_text кодируется в
    байты той же кодовой страницы (stdin sudo -S на Windows читает OEM).
    """
    try:
        stdin = None
        if input_text is not None:
            enc = codepage or "utf-8"
            stdin = str(input_text).encode(enc, "replace")
        r = subprocess.run(cmd, capture_output=True, timeout=timeout_s,
                           env=env, input=stdin)
        out = decode_out(r.stdout, codepage)
        err = decode_out(r.stderr, codepage)
        limit = max(8000, int(max_chars or 8000))
        return r.returncode, out[:limit], err[:2000]
    except subprocess.TimeoutExpired:
        return None, "", "timeout %ss" % timeout_s
    except FileNotFoundError:
        return None, "", "нет бинаря: %s" % (cmd[0] if cmd else "?")
    except Exception as e:  # noqa: BLE001
        return None, "", "%s: %s" % (type(e).__name__, e)

"""outbound.drivers.file — sftp/smb/ftp/nfs.

sftp: бинарь OpenSSH (batchmode), без pip. smb: pysmb (опционально).
ftp: ftplib (stdlib). nfs: точка монтирования на хосте бота (nfs-common),
драйвер делает файловые операции по mount_path + проверяет, что это mount.
"""
import io as _io
import os as _os
import tempfile as _tf

from . import base as b

KINDS = ("sftp", "smb", "ftp", "nfs")


def _local_safe(local):
    """Проверка локального пути для get/put: без выхода за пределы.

    Абсолютные пути разрешены (бэкапы работают с абсолютными путями
    под BACKUP_DIR), но запрещены: пусто, NUL-байт, ~ (дом. каталог),
    ..-сегменты в любом стиле разделителей. Возвращает очищенный путь
    или "" (плохой путь).
    """
    p = (local or "").strip()
    if not p or "\x00" in p:
        return ""
    if p.startswith("~"):
        return ""
    segs = p.replace("\\", "/").split("/")
    if any(s == ".." for s in segs):
        return ""
    return p


def _remote_nl_ok(*vals):
    """Нет CR/LF в remote-путях (инъекция команд в ftp/sftp-batch)."""
    for v in vals:
        if v and ("\r" in v or "\n" in v):
            return False
    return True


def _ssh_target(ep):
    user = (b.addr(ep, "user") or "").strip()
    host = (b.addr(ep, "host") or "").strip()
    try:
        port = int(b.addr(ep, "port") or 22)
    except (TypeError, ValueError):
        port = 22
    if not host:
        return None
    return ("%s@%s" % (user, host) if user else host), port


# --- sftp (бинарь) ---

def _sftp_batch(ep, secret, batch):
    tgt = _ssh_target(ep)
    if not tgt:
        return None, b.err("sftp: нужен address.host")
    target, port = tgt
    key = (b.addr(ep, "key") or "").strip()
    cmd = ["sftp", "-P", str(port), "-o",
           "StrictHostKeyChecking=%s" % b.opt(ep, "strict_host_key",
                                              "accept-new"),
           "-o", "ConnectTimeout=%s" % min(b.timeout(ep), 30),
           "-b", "-"]
    if key:
        cmd += ["-i", key]
    cmd.append(target)
    env = None
    if secret and not key:
        cmd = ["sshpass", "-e"] + cmd
        import os as _o
        env = dict(_o.environ)
        env["SSHPASS"] = secret
    rc, out, err = b.run(cmd, timeout_s=b.timeout(ep) + 30, env=env,
                         input_text=batch)
    if rc is None:
        return None, b.err("sftp: %s" % err)
    if rc != 0:
        return None, b.err("sftp rc=%s: %s" % (rc, (err or out)[:300]))
    return out, None


def sftp_list(ep, secret, params):
    path = ((params or {}).get("path") or ".").strip()
    out, err = _sftp_batch(ep, secret, "ls -l %s\n" % _q(path))
    if err:
        return err
    return b.ok(path=path, listing=out.strip()[:4000])


def sftp_get(ep, secret, params):
    remote = ((params or {}).get("remote") or "").strip()
    local = _local_safe((params or {}).get("local"))
    if not remote or not local:
        return b.err("sftp.get: нужны params.remote и params.local "
                     "(local — без .. и ~)")
    if not _remote_nl_ok(remote, local):
        return b.err("sftp.get: переводы строк в путях запрещены")
    out, err = _sftp_batch(ep, secret, "get %s %s\n" % (_q(remote),
                                                        _q(local)))
    if err:
        return err
    return b.ok(remote=remote, local=local)


def sftp_put(ep, secret, params):
    local = _local_safe((params or {}).get("local"))
    remote = ((params or {}).get("remote") or "").strip()
    if not local or not remote or not _os.path.isfile(local):
        return b.err("sftp.put: нужен существующий params.local и "
                     "params.remote")
    if not _remote_nl_ok(remote, local):
        return b.err("sftp.put: переводы строк в путях запрещены")
    out, err = _sftp_batch(ep, secret, "put %s %s\n" % (_q(local),
                                                        _q(remote)))
    if err:
        return err
    return b.ok(local=local, remote=remote)


def _q(s):
    # перевод строки разорвал бы sftp-batch на две команды — режем сразу
    s = str(s or "").replace("\r", "").replace("\n", "")
    return "'%s'" % s.replace("'", "'\\''")


# --- smb (pysmb) ---

def _smb_conn(ep, secret):
    smb = b.lazy("smb.SMBConnection")
    if not smb:
        return None, b.missing("pysmb", "smb")
    host = (b.addr(ep, "host") or "").strip()
    user = (b.addr(ep, "user") or "").strip()
    share = (b.addr(ep, "share") or "").strip()
    if not host or not user or not share:
        return None, b.err("smb: нужны address.host/user/share")
    try:
        conn = smb.SMBConnection(user, secret or "", "bot", host,
                                 use_ntlm_v2=True)
        if not conn.connect(host, int(b.addr(ep, "port") or 445),
                            timeout=b.timeout(ep)):
            return None, b.err("smb: %s не отвечает" % host)
        return (conn, share), None
    except Exception as e:  # noqa: BLE001
        return None, b.err("smb.connect: %s" % e)


def smb_list(ep, secret, params):
    path = ((params or {}).get("path") or "/").strip() or "/"
    conn, err = _smb_conn(ep, secret)
    if err:
        return err
    conn_obj, share = conn
    try:
        items = conn_obj.listPath(share, path, timeout=b.timeout(ep))
        names = [i.filename for i in items
                 if i.filename not in (".", "..")]
        return b.ok(path=path, names=names[:200])
    except Exception as e:  # noqa: BLE001
        return b.err("smb.list: %s" % e)
    finally:
        try:
            conn_obj.close()
        except Exception:  # noqa: BLE001
            pass


def smb_get(ep, secret, params):
    remote = ((params or {}).get("remote") or "").strip()
    local = _local_safe((params or {}).get("local"))
    if not remote or not local:
        return b.err("smb.get: нужны params.remote и params.local "
                     "(local — без .. и ~)")
    conn, err = _smb_conn(ep, secret)
    if err:
        return err
    conn_obj, share = conn
    try:
        with open(local, "wb") as f:
            conn_obj.retrieveFile(share, remote, f, timeout=b.timeout(ep))
        return b.ok(remote=remote, local=local)
    except Exception as e:  # noqa: BLE001
        return b.err("smb.get: %s" % e)
    finally:
        try:
            conn_obj.close()
        except Exception:  # noqa: BLE001
            pass


def smb_put(ep, secret, params):
    remote = ((params or {}).get("remote") or "").strip()
    local = _local_safe((params or {}).get("local"))
    if not remote or not local:
        return b.err("smb.put: нужны params.remote и params.local "
                     "(local — без .. и ~)")
    if ".." in remote.split("/"):
        return b.err("smb.put: remote вне шары запрещён")
    conn, err = _smb_conn(ep, secret)
    if err:
        return err
    conn_obj, share = conn
    try:
        with open(local, "rb") as f:
            conn_obj.storeFile(share, remote, f, timeout=b.timeout(ep))
        return b.ok(remote=remote, local=local)
    except Exception as e:  # noqa: BLE001
        return b.err("smb.put: %s" % e)
    finally:
        try:
            conn_obj.close()
        except Exception:  # noqa: BLE001
            pass


def smb_mkdir(ep, secret, params):
    remote = ((params or {}).get("remote") or "").strip().strip("/")
    if not remote:
        return b.err("smb.mkdir: нужен params.remote")
    if ".." in remote.split("/"):
        return b.err("smb.mkdir: вне шары запрещён")
    conn, err = _smb_conn(ep, secret)
    if err:
        return err
    conn_obj, share = conn
    try:
        try:
            conn_obj.createDirectory(share, remote, timeout=b.timeout(ep))
        except Exception:
            pass  # уже есть — не ошибка
        return b.ok(remote=remote)
    except Exception as e:  # noqa: BLE001
        return b.err("smb.mkdir: %s" % e)
    finally:
        try:
            conn_obj.close()
        except Exception:  # noqa: BLE001
            pass


def smb_delete(ep, secret, params):
    remote = ((params or {}).get("remote") or "").strip()
    if not remote:
        return b.err("smb.delete: нужен params.remote")
    if ".." in remote.split("/"):
        return b.err("smb.delete: remote вне шары запрещён")
    conn, err = _smb_conn(ep, secret)
    if err:
        return err
    conn_obj, share = conn
    try:
        conn_obj.deleteFiles(share, remote, timeout=b.timeout(ep))
        return b.ok(remote=remote)
    except Exception as e:  # noqa: BLE001
        return b.err("smb.delete: %s" % e)
    finally:
        try:
            conn_obj.close()
        except Exception:  # noqa: BLE001
            pass


# --- ftp (stdlib) ---

def _ftp_conn(ep, secret):
    from ftplib import FTP
    host = (b.addr(ep, "host") or "").strip()
    user = (b.addr(ep, "user") or "anonymous").strip()
    try:
        port = int(b.addr(ep, "port") or 21)
    except (TypeError, ValueError):
        port = 21
    if not host:
        return None, b.err("ftp: нужен address.host")
    try:
        ftp = FTP()
        ftp.connect(host, port, timeout=b.timeout(ep))
        ftp.login(user, secret or "")
        return ftp, None
    except Exception as e:  # noqa: BLE001
        return None, b.err("ftp.connect: %s" % e)


def ftp_list(ep, secret, params):
    path = ((params or {}).get("path") or "").strip()
    ftp, err = _ftp_conn(ep, secret)
    if err:
        return err
    try:
        names = ftp.nlst(path or None)
        return b.ok(path=path or "/", names=names[:200])
    except Exception as e:  # noqa: BLE001
        return b.err("ftp.list: %s" % e)
    finally:
        try:
            ftp.quit()
        except Exception:  # noqa: BLE001
            pass


def ftp_get(ep, secret, params):
    remote = ((params or {}).get("remote") or "").strip()
    local = _local_safe((params or {}).get("local"))
    if not remote or not local:
        return b.err("ftp.get: нужны params.remote и params.local "
                     "(local — без .. и ~)")
    if not _remote_nl_ok(remote):
        # ftplib склеивает "RETR " + remote: CR/LF = инъекция команд
        return b.err("ftp.get: переводы строк в remote запрещены")
    ftp, err = _ftp_conn(ep, secret)
    if err:
        return err
    try:
        with open(local, "wb") as f:
            ftp.retrbinary("RETR " + remote, f.write)
        return b.ok(remote=remote, local=local)
    except Exception as e:  # noqa: BLE001
        return b.err("ftp.get: %s" % e)
    finally:
        try:
            ftp.quit()
        except Exception:  # noqa: BLE001
            pass


# --- nfs (точка монтирования на хосте бота) ---

def _nfs_root(ep):
    root = (b.addr(ep, "mount_path") or "").strip()
    if not root or not _os.path.isdir(root):
        return None, b.err("nfs: address.mount_path не смонтирован: %s"
                           % root)
    if not _os.path.ismount(root):
        return None, b.err("nfs: %s не mount (stale?)" % root)
    return root, None


def _nfs_safe(root, rel):
    rel = str(rel or "").strip()
    if rel == root or rel.startswith(root.rstrip("/") + "/"):
        full = _os.path.realpath(rel)
    else:
        full = _os.path.realpath(_os.path.join(root, rel.lstrip("/")))
    if full != root and not full.startswith(root + _os.sep):
        return None
    return full


def nfs_list(ep, secret, params):
    root, err = _nfs_root(ep)
    if err:
        return err
    full = _nfs_safe(root, (params or {}).get("path") or "")
    if not full or not _os.path.isdir(full):
        return b.err("nfs.list: плохой путь")
    try:
        return b.ok(path=full, names=sorted(_os.listdir(full))[:200])
    except Exception as e:  # noqa: BLE001
        return b.err("nfs.list: %s" % e)


def nfs_read(ep, secret, params):
    root, err = _nfs_root(ep)
    if err:
        return err
    full = _nfs_safe(root, (params or {}).get("path") or "")
    if not full or not _os.path.isfile(full):
        return b.err("nfs.read: плохой путь")
    try:
        if _os.path.getsize(full) > 512 * 1024:
            return b.err("nfs.read: файл >512K (забирай кусками позже)")
        with open(full, "r", encoding="utf-8", errors="replace") as f:
            return b.ok(path=full, text=f.read()[:8000])
    except Exception as e:  # noqa: BLE001
        return b.err("nfs.read: %s" % e)


# --- fs.*: единый файловый слой (обходы, fileman) ---

def _int(v):
    try:
        return int(str(v).strip())
    except (TypeError, ValueError):
        return 0


def _text_file(path, max_b):
    with open(path, "rb") as f:
        data = f.read(max_b + 1)
    return data[:max_b].decode("utf-8", errors="replace"), len(data) > max_b


def _sftp_entries(out):
    """`ls -l` sftp → [{name, kind, size}] (формат OpenSSH)."""
    items = []
    for ln in (out or "").splitlines():
        s = ln.strip()
        if not s or s.startswith("sftp>"):
            continue
        toks = s.split(None, 8)
        if len(toks) < 9 or len(toks[0]) != 10 or toks[0][0] not in "-dl":
            continue
        items.append({"name": toks[8],
                      "kind": "dir" if toks[0][0] == "d" else (
                          "link" if toks[0][0] == "l" else "file"),
                      "size": _int(toks[4]), "mtime": 0})
    return items


def _sftp_fs_list(ep, secret, params):
    path = ((params or {}).get("path") or "/").strip() or "/"
    out, err = _sftp_batch(ep, secret, "ls -l %s\n" % _q(path))
    if err:
        return err
    items = _sftp_entries(out)
    return b.ok(path=path, items=items[:2000], total=len(items),
                truncated=len(items) > 2000)


def _sftp_fs_read(ep, secret, params):
    remote = ((params or {}).get("path") or "").strip()
    if not remote:
        return b.err("fs.read: нужен params.path")
    max_b = min(max(1024, _int((params or {}).get("max_bytes"))
                    or 200000), 2000000)
    fd, tmp = _tf.mkstemp(prefix="obfs-")
    _os.close(fd)
    try:
        _out, err = _sftp_batch(ep, secret, "get %s %s\n" % (_q(remote),
                                                            _q(tmp)))
        if err:
            return err
        text, trunc = _text_file(tmp, max_b)
        return b.ok(path=remote, text=text, truncated=trunc, size=len(text))
    finally:
        try:
            _os.unlink(tmp)
        except OSError:
            pass


def _smb_fs_list(ep, secret, params):
    path = ((params or {}).get("path") or "/").strip() or "/"
    conn, err = _smb_conn(ep, secret)
    if err:
        return err
    conn_obj, share = conn
    try:
        items = [{"name": i.filename,
                  "kind": "dir" if i.isDirectory else "file",
                  "size": int(i.file_size or 0),
                  "mtime": int(i.last_write_time or 0)}
                 for i in conn_obj.listPath(share, path,
                                            timeout=b.timeout(ep))
                 if i.filename not in (".", "..")]
        items.sort(key=lambda x: x["name"])
        return b.ok(path=path, items=items[:2000], total=len(items),
                    truncated=len(items) > 2000)
    except Exception as e:  # noqa: BLE001
        return b.err("fs.list: %s" % e)
    finally:
        try:
            conn_obj.close()
        except Exception:  # noqa: BLE001
            pass


def _smb_fs_read(ep, secret, params):
    remote = ((params or {}).get("path") or "").strip()
    if not remote:
        return b.err("fs.read: нужен params.path")
    max_b = min(max(1024, _int((params or {}).get("max_bytes"))
                    or 200000), 2000000)
    conn, err = _smb_conn(ep, secret)
    if err:
        return err
    conn_obj, share = conn
    try:
        attrs = conn_obj.getAttributes(share, remote)
        if int(attrs.file_size or 0) > 4 * 1024 * 1024:
            return b.err("fs.read: файл >4M")
        buf = _io.BytesIO()
        conn_obj.retrieveFile(share, remote, buf, timeout=b.timeout(ep))
        data = buf.getvalue()
        return b.ok(path=remote,
                    text=data[:max_b].decode("utf-8", errors="replace"),
                    truncated=len(data) > max_b, size=len(data))
    except Exception as e:  # noqa: BLE001
        return b.err("fs.read: %s" % e)
    finally:
        try:
            conn_obj.close()
        except Exception:  # noqa: BLE001
            pass


def _ftp_fs_list(ep, secret, params):
    path = ((params or {}).get("path") or "").strip() or "/"
    ftp, err = _ftp_conn(ep, secret)
    if err:
        return err
    try:
        items = []
        for name, facts in ftp.mlsd(path):
            if name in (".", ".."):
                continue
            typ = (facts.get("type") or "").lower()
            items.append({"name": name,
                          "kind": ("dir" if typ in ("dir", "cdir", "pdir")
                                   else "link" if typ == "os.link"
                                   else "file"),
                          "size": _int(facts.get("size")), "mtime": 0})
        items.sort(key=lambda x: x["name"])
        return b.ok(path=path, items=items[:2000], total=len(items),
                    truncated=len(items) > 2000)
    except Exception as e:  # noqa: BLE001
        return b.err("fs.list: ftp без MLSD: %s" % e)
    finally:
        try:
            ftp.quit()
        except Exception:  # noqa: BLE001
            pass


def _ftp_fs_read(ep, secret, params):
    remote = ((params or {}).get("path") or "").strip()
    if not remote:
        return b.err("fs.read: нужен params.path")
    max_b = min(max(1024, _int((params or {}).get("max_bytes"))
                    or 200000), 2000000)
    ftp, err = _ftp_conn(ep, secret)
    if err:
        return err
    try:
        buf = _io.BytesIO()
        ftp.retrbinary("RETR " + remote, buf.write)
        data = buf.getvalue()
        return b.ok(path=remote,
                    text=data[:max_b].decode("utf-8", errors="replace"),
                    truncated=len(data) > max_b, size=len(data))
    except Exception as e:  # noqa: BLE001
        return b.err("fs.read: %s" % e)
    finally:
        try:
            ftp.quit()
        except Exception:  # noqa: BLE001
            pass


def _nfs_fs_list(ep, secret, params):
    root, err = _nfs_root(ep)
    if err:
        return err
    full = _nfs_safe(root, (params or {}).get("path") or "")
    if not full or not _os.path.isdir(full):
        return b.err("fs.list: плохой путь")
    try:
        items = []
        with _os.scandir(full) as it:
            for e in it:
                try:
                    st = e.stat(follow_symlinks=False)
                    size, mt = st.st_size, int(st.st_mtime)
                except OSError:
                    size, mt = 0, 0
                kind = ("dir" if e.is_dir(follow_symlinks=False)
                        else "file" if e.is_file(follow_symlinks=False)
                        else "link")
                items.append({"name": e.name, "kind": kind, "size": size,
                              "mtime": mt})
        items.sort(key=lambda x: x["name"])
        return b.ok(path=full, items=items[:2000], total=len(items),
                    truncated=len(items) > 2000)
    except Exception as e:  # noqa: BLE001
        return b.err("fs.list: %s" % e)


def _nfs_fs_read(ep, secret, params):
    root, err = _nfs_root(ep)
    if err:
        return err
    full = _nfs_safe(root, (params or {}).get("path") or "")
    if not full or not _os.path.isfile(full):
        return b.err("fs.read: плохой путь")
    max_b = min(max(1024, _int((params or {}).get("max_bytes"))
                    or 200000), 2000000)
    try:
        text, trunc = _text_file(full, max_b)
        return b.ok(path=full, text=text, truncated=trunc, size=len(text))
    except OSError as e:
        return b.err("fs.read: %s" % e)


def _nfs_fs_grep(ep, secret, params):
    import re as _re
    root, err = _nfs_root(ep)
    if err:
        return err
    full = _nfs_safe(root, (params or {}).get("path") or "")
    pat = ((params or {}).get("pattern") or "").strip()
    if not full or not pat:
        return b.err("fs.grep: нужны params.path и params.pattern")
    maxn = min(max(1, _int((params or {}).get("max")) or 100), 1000)
    try:
        rx = _re.compile(pat)
    except _re.error as e:
        return b.err("fs.grep: плохой шаблон: %s" % e)
    files = []
    if _os.path.isfile(full):
        files = [full]
    elif bool((params or {}).get("recursive")):
        for dirpath, dirnames, filenames in _os.walk(full):
            dirnames[:] = sorted(d for d in dirnames
                                 if not d.startswith(".")
                                 and d not in (".git", "node_modules",
                                               "__pycache__"))
            files += [_os.path.join(dirpath, f) for f in sorted(filenames)
                      if not f.startswith(".")]
    else:
        return b.err("fs.grep: каталог без recursive")
    out = []
    for fp in files:
        if len(out) >= maxn:
            break
        try:
            with open(fp, "r", encoding="utf-8", errors="replace") as fh:
                for n, line in enumerate(fh, 1):
                    if len(line) > 4000:
                        continue
                    if rx.search(line):
                        out.append({"path": _os.path.relpath(fp, root),
                                    "line": n, "text": line.rstrip("\n")[:300]})
                        if len(out) >= maxn:
                            break
        except OSError:
            continue
    return b.ok(path=full, matches=out, total=len(out),
                truncated=len(out) >= maxn)


def _fs_list(ep, secret, params):
    kind = (ep.get("kind") or "").strip()
    fn = {"sftp": _sftp_fs_list, "smb": _smb_fs_list, "ftp": _ftp_fs_list,
          "nfs": _nfs_fs_list}.get(kind)
    if fn is None:
        return b.err("fs.list: %s не поддерживает" % kind)
    return fn(ep, secret, params)


def _fs_read(ep, secret, params):
    kind = (ep.get("kind") or "").strip()
    fn = {"sftp": _sftp_fs_read, "smb": _smb_fs_read, "ftp": _ftp_fs_read,
          "nfs": _nfs_fs_read}.get(kind)
    if fn is None:
        return b.err("fs.read: %s не поддерживает" % kind)
    return fn(ep, secret, params)


def _fs_grep(ep, secret, params):
    kind = (ep.get("kind") or "").strip()
    if kind == "nfs":
        return _nfs_fs_grep(ep, secret, params)
    return b.err("fs.grep: только ssh/nfs (у %s нет поиска на источнике)"
                 % kind)


OPS = {
    "sftp.list": sftp_list,
    "sftp.get": sftp_get,
    "sftp.put": sftp_put,
    "smb.list": smb_list,
    "smb.get": smb_get,
    "smb.put": smb_put,
    "smb.mkdir": smb_mkdir,
    "smb.delete": smb_delete,
    "ftp.list": ftp_list,
    "ftp.get": ftp_get,
    "nfs.list": nfs_list,
    "nfs.read": nfs_read,
    "fs.list": _fs_list,
    "fs.read": _fs_read,
    "fs.grep": _fs_grep,
}


def health(ep, secret):
    kind = (ep.get("kind") or "").strip()
    if kind == "sftp":
        out, err = _sftp_batch(ep, secret, "pwd\n")
        return b.ok() if err is None else err
    if kind == "smb":
        r = smb_list(ep, secret, {"path": (b.addr(ep, "health_path")
                                           or "/")})
        return b.ok() if r.get("ok") else r
    if kind == "ftp":
        r = ftp_list(ep, secret, {})
        return b.ok() if r.get("ok") else r
    if kind == "nfs":
        root, err = _nfs_root(ep)
        return b.ok(mount=root) if err is None else err
    return b.err("file: нет health для kind=%s" % kind)

"""outbound.drivers.http — http/https через urllib3 (зависимость ядра де-факто).

Операции: http.request (JSON/текст), http.stream (буфер с лимитом — для
будущего SSE потребители пока читают собранное тело + чанки-мету).
Прокси: opts.proxy (http://...) либо SOCKS при наличии PySocks.
"""
import json as _json

from . import base as b

KINDS = ("http",)


def _loopback_host(host):
    """Строго loopback: localhost/::1/127.x.x.x (без префикс-матчей вида 127.evil.com)."""
    h = (host or "").strip("[]").lower()
    if h in ("localhost", "::1"):
        return True
    parts = h.split(".")
    return (len(parts) == 4 and parts[0] == "127"
            and all(p.isdigit() and 0 <= int(p) <= 255 for p in parts[1:]))


def _resolve_verify(ep, params, url):
    """verify для запроса: дефолт из биндинга; params verify:false —
    только для loopback (самоподписанный серт панели), иначе игнор."""
    verify = (params or {}).get("verify")
    if verify is None:
        return b.opt(ep, "verify", True)
    if verify is False:
        try:
            host = (url or "").split("://", 1)[1].split("/", 1)[0]
            host = host.split("@")[-1].split(":")[0]
        except Exception:  # noqa: BLE001
            host = ""
        if not _loopback_host(host):
            return True
        return False
    return True


def _manager(ep, verify=True):
    u3 = b.lazy("urllib3")
    if not u3:
        return None, b.missing("urllib3", "http")
    proxy = (b.opt(ep, "proxy") or "").strip()
    if proxy:
        pm = u3.ProxyManager(proxy, cert_reqs="CERT_REQUIRED" if verify
                             else "CERT_NONE")
    else:
        pm = u3.PoolManager(cert_reqs="CERT_REQUIRED" if verify
                            else "CERT_NONE")
    return pm, None


def _headers(ep, secret, params):
    h = dict((b.opt(ep, "headers") or {}))
    h.update((params or {}).get("headers") or {})
    auth = (b.opt(ep, "auth") or "").strip().lower()
    if auth == "bearer" and secret:
        h["Authorization"] = "Bearer " + secret
    elif auth == "basic" and secret:
        import base64 as _b64
        h["Authorization"] = "Basic " + _b64.b64encode(
            secret.encode()).decode()
    elif auth == "token" and secret:
        h["Authorization"] = "token " + secret
    return h


def http_request(ep, secret, params):
    params = params or {}
    base = (b.addr(ep, "base_url") or "").rstrip("/")
    path = (params.get("path") or "/").strip() or "/"
    if not path.startswith("/"):
        path = "/" + path
    url = (params.get("url") or (base + path)).strip()
    if not url.startswith(("http://", "https://")):
        return b.err("http.request: bad url")
    pm, err = _manager(ep, _resolve_verify(ep, params, url))
    if err:
        return err
    try:
        timeout = int(params.get("timeout") or 0)
    except (TypeError, ValueError):
        timeout = 0
    if timeout:
        timeout = max(2, min(timeout, 300))
    else:
        timeout = b.timeout(ep)
    method = (params.get("method") or "GET").upper()
    if method not in ("GET", "POST", "PUT", "PATCH", "DELETE", "HEAD"):
        return b.err("http.request: метод %s запрещён" % method)
    body = params.get("json")
    raw = params.get("body")
    multi = params.get("multipart")
    data = None
    ctype = None
    if multi is not None:
        # {"поле": "текст" | {"filename","data_b64","content_type"}}
        # (base64 — JSON-безопасно, работает и поверх HTTP API).
        import base64 as _b64
        from urllib3.filepost import encode_multipart_formdata as _enc
        fields = {}
        for k, v in (multi if isinstance(multi, dict) else {}).items():
            if isinstance(v, dict) and "data_b64" in v:
                try:
                    blob = _b64.b64decode(v.get("data_b64") or "")
                except Exception:
                    return b.err("http.request: битый base64 в поле %s" % k)
                fields[k] = (v.get("filename") or k,
                             blob, v.get("content_type")
                             or "application/octet-stream")
            else:
                fields[k] = str(v)
        try:
            data, ctype = _enc(fields)
        except Exception as e:  # noqa: BLE001
            return b.err("http.request: multipart: %s" % e)
    elif body is not None:
        data = _json.dumps(body, ensure_ascii=False).encode()
        ctype = "application/json"
    elif raw is not None:
        data = raw.encode() if isinstance(raw, str) else bytes(raw)
    headers = _headers(ep, secret, params)
    if ctype:
        headers.setdefault("Content-Type", ctype)
    t0 = b.now_ms()
    try:
        r = pm.request(method, url, body=data, headers=headers,
                       timeout=timeout, retries=0)
        # Кап тела 2МБ: транскрипты opencode (истории длинных исполнений)
        # упирались в 200КБ — обрезанный JSON не разбирался и молча
        # терялся (см. agent_tasks session_messages).
        text = (r.data or b"")[:2000000].decode("utf-8", "replace")
        out = b.ok(status=r.status, ms=b.now_ms() - t0)
        if "application/json" in (r.headers.get("Content-Type") or ""):
            try:
                out["json"] = _json.loads(text)
            except Exception:  # noqa: BLE001
                out["text"] = text[:8000]
        else:
            out["text"] = text[:8000]
        return out
    except Exception as e:  # noqa: BLE001
        return b.err("http %s %s: %s" % (method, url.split("?")[0], e),
                     ms=b.now_ms() - t0)


def http_stream(ep, secret, params):
    """Чтение тела чанками с лимитом (задел под SSE-миграцию opencode).

    Возвращает собранный текст + число чанков; настоящий построчный SSE
    пока остаётся в модуле opencode.
    """
    params = params or {}
    try:
        limit = int(params.get("limit_kb", 512))
    except (TypeError, ValueError):
        limit = 512
    limit = max(16, min(limit, 4096)) * 1024
    base = (b.addr(ep, "base_url") or "").rstrip("/")
    path = (params.get("path") or "/").strip() or "/"
    url = (params.get("url") or (base + path)).strip()
    pm, err = _manager(ep, _resolve_verify(ep, params, url))
    if err:
        return err
    t0 = b.now_ms()
    try:
        r = pm.request("GET", url, headers=_headers(ep, secret, params),
                       timeout=b.timeout(ep), retries=0, preload_content=False)
        chunks = 0
        buf = bytearray()
        for chunk in r.stream(65536):
            chunks += 1
            buf += chunk
            if len(buf) >= limit:
                break
        r.release_conn()
        text = bytes(buf).decode("utf-8", "replace")
        return b.ok(status=r.status, chunks=chunks,
                    truncated=len(buf) >= limit,
                    text=text[:8000], ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        return b.err("http.stream %s: %s" % (url.split("?")[0], e),
                     ms=b.now_ms() - t0)


OPS = {
    "http.request": http_request,
    "http.stream": http_stream,
    "http.sse": None,  # подставляется ниже (генератор)
}


def sse_lines(ep, secret, params):
    """SSE/event-stream построчно: генератор data:-пейлоадов (текст).

    Фрейминг — здесь, парсинг JSON — у потребителя. params: path|url,
    method, json/body, headers, timeout (дедлайн всего чтения, до 900с),
    max_lines, max_bytes. Закрытие генератора (close/break) рвёт соединение
    (finally + release_conn). Ошибка HTTP — исключение на первом next().
    """
    params = params or {}
    base = (b.addr(ep, "base_url") or "").rstrip("/")
    path = (params.get("path") or "/").strip() or "/"
    if not path.startswith("/"):
        path = "/" + path
    _sse_url = (params.get("url") or (base + path)).strip()
    pm, err = _manager(ep, _resolve_verify(ep, params, _sse_url))
    if err:
        def _e():
            raise RuntimeError(err.get("error"))
            yield
        return _e()
    try:
        timeout = int(params.get("timeout") or 0)
    except (TypeError, ValueError):
        timeout = 0
    timeout = max(5, min(timeout or 120, 900))
    try:
        max_lines = int(params.get("max_lines") or 50000)
    except (TypeError, ValueError):
        max_lines = 50000
    max_lines = max(1, min(max_lines, 200000))
    try:
        max_bytes = int(params.get("max_bytes") or 16777216)
    except (TypeError, ValueError):
        max_bytes = 16777216
    max_bytes = max(65536, min(max_bytes, 67108864))
    url = _sse_url
    if not url.startswith(("http://", "https://")):
        def _e2():
            raise RuntimeError("http.sse: bad url")
            yield
        return _e2()
    method = (params.get("method") or "GET").upper()
    if method not in ("GET", "POST"):
        def _e3():
            raise RuntimeError("http.sse: метод %s запрещён" % method)
            yield
        return _e3()
    body = params.get("json")
    raw = params.get("body")
    data = None
    if body is not None:
        import json as _j
        data = _j.dumps(body, ensure_ascii=False).encode()
    elif raw is not None:
        data = raw.encode() if isinstance(raw, str) else bytes(raw)
    headers = _headers(ep, secret, params)
    if body is not None:
        headers.setdefault("Content-Type", "application/json")
    import time as _t

    # Eager-подписка: соединение открывается ЗДЕСЬ (до первого next),
    # иначе быстрые дельты прилетают раньше подписки и теряются.
    import urllib3.exceptions as _ue
    from urllib3 import Timeout as _UTimeout
    try:
        # connect — короткий, read — под длинный TTFT/прогрев модели.
        # Раньше единый timeout рвал чтение между чанками (групповой запрос
        # умирал с "Read timed out", плейсхолдер висел).
        _to = _UTimeout(connect=min(20, timeout), read=timeout)
        r = pm.request(method, url, body=data, headers=headers,
                       timeout=_to, retries=0, preload_content=False)
    except Exception as e:  # noqa: BLE001
        raise RuntimeError("http.sse %s: %s" % (url.split("?")[0], e))
    if r.status >= 400:
        try:
            body_txt = (r.read() or b"")[:2000].decode("utf-8", "replace")
        except Exception:  # noqa: BLE001
            body_txt = ""
        try:
            r.release_conn()
        except Exception:  # noqa: BLE001
            pass
        raise RuntimeError("HTTP %d: %s" % (r.status, body_txt[:200]))

    def gen():
        buf = b""
        total = 0
        lines = 0
        deadline = _t.monotonic() + timeout
        try:
            for chunk in r.stream(65536):
                if not chunk:
                    continue
                total += len(chunk)
                if total > max_bytes:
                    raise RuntimeError("http.sse: лимит %d байт" % max_bytes)
                if _t.monotonic() > deadline:
                    raise RuntimeError("http.sse: дедлайн %dс" % timeout)
                buf += chunk
                while b"\n" in buf:
                    line, buf = buf.split(b"\n", 1)
                    line = line.strip()
                    if not line.startswith(b"data:"):
                        continue
                    payload = line[5:].strip().decode("utf-8", "replace")
                    if not payload:
                        continue
                    lines += 1
                    if lines > max_lines:
                        raise RuntimeError("http.sse: лимит %d строк" % max_lines)
                    yield payload
        except _ue.ReadTimeoutError:
            raise RuntimeError("http.sse: таймаут чтения")
        finally:
            try:
                r.release_conn()
            except Exception:  # noqa: BLE001
                pass

    return gen()


OPS["http.sse"] = sse_lines


def health(ep, secret):
    hcfg = ep.get("health") or {}
    hpath = (hcfg.get("health_path") or b.opt(ep, "health_path")
             or "/").strip() or "/"
    r = http_request(ep, secret, {"path": hpath, "method": "GET"})
    if not r.get("ok"):
        return r
    expect = hcfg.get("expect_status", b.opt(ep, "expect_status", [200]))
    try:
        expect = [int(x) for x in expect]
    except (TypeError, ValueError):
        expect = [200]
    if r.get("status") in expect:
        return b.ok(ms=r.get("ms", 0), status=r.get("status"))
    return b.err("http health: status=%s, ждали %s" % (r.get("status"),
                                                       expect))

"""outbound.drivers.mgmt — winrm/docker/proxmox/telnet. Плюс dry-run.

winrm: pywinrm (опционально). docker: Engine API по unix-socket (stdlib
http.client) или TCP. proxmox: REST (ticket либо API-token) поверх urllib3.
telnet: сырые сокеты с expect (stdlib, без telnetlib). dry-run: заглушка.
"""
import json as _json
import socket as _socket
import time as _time

from . import base as b

KINDS = ("winrm", "docker", "proxmox", "telnet", "dryrun")


# --- winrm ---

def winrm_exec(ep, secret, params):
    wr = b.lazy("winrm")
    if not wr:
        return b.missing("pywinrm", "winrm")
    host = (b.addr(ep, "host") or "").strip()
    user = (b.addr(ep, "user") or "").strip()
    if not host or not user:
        return b.err("winrm.exec: нужны address.host/user")
    cmd = ((params or {}).get("cmd") or "").strip()
    if not cmd:
        return b.err("winrm.exec: нужен params.cmd")
    scheme = (b.addr(ep, "scheme") or "http").strip()
    try:
        port = int(b.addr(ep, "port") or (5986 if scheme == "https" else 5985))
    except (TypeError, ValueError):
        port = 5985
    t0 = b.now_ms()
    try:
        sess = wr.Session("%s://%s:%s/wsman" % (scheme, host, port),
                          auth=(user, secret or ""),
                          transport="ntlm",
                          server_cert_validation="ignore"
                          if b.opt(ep, "verify", True) is False else "validate")
        r = sess.run_cmd(cmd)
        out = b.ok(rc=r.status_code,
                   output=(r.std_out or b"").decode("utf-8", "replace"
                                                    )[:8000],
                   ms=b.now_ms() - t0)
        if r.status_code != 0:
            out["ok"] = False
            out["error"] = "rc=%s: %s" % (
                r.status_code,
                (r.std_err or b"").decode("utf-8", "replace")[:300])
        return out
    except Exception as e:  # noqa: BLE001
        return b.err("winrm %s: %s" % (host, e), ms=b.now_ms() - t0)


# --- docker (Engine API) ---

def _docker_call(ep, method, path, body=None):
    import http.client as _hc
    sock = (b.addr(ep, "socket") or "").strip()
    host = (b.addr(ep, "host") or "").strip()
    t0 = b.now_ms()
    try:
        if sock:
            conn = _hc.HTTPConnection("localhost", timeout=b.timeout(ep))
            conn.sock = _socket.socket(_socket.AF_UNIX,
                                       _socket.SOCK_STREAM)
            conn.sock.settimeout(b.timeout(ep))
            conn.sock.connect(sock)
        elif host:
            try:
                port = int(b.addr(ep, "port") or 2375)
            except (TypeError, ValueError):
                port = 2375
            conn = _hc.HTTPConnection(host, port, timeout=b.timeout(ep))
        else:
            return b.err("docker: нужны address.socket или address.host")
        data = None
        headers = {}
        if body is not None:
            data = _json.dumps(body)
            headers["Content-Type"] = "application/json"
        conn.request(method, "/v1.43" + path, body=data, headers=headers)
        r = conn.getresponse()
        text = r.read()[:100000].decode("utf-8", "replace")
        try:
            payload = _json.loads(text) if text else None
        except Exception:  # noqa: BLE001
            payload = None
        return b.ok(status=r.status, json=payload, ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        return b.err("docker %s %s: %s" % (method, path, e))


def docker_ps(ep, secret, params):
    all_c = (params or {}).get("all", True)
    r = _docker_call(ep, "GET", "/containers/json?all=%s" % (
        "1" if all_c else "0"))
    if not r.get("ok"):
        return r
    items = [{"id": (c.get("Id") or "")[:12], "name": (c.get("Names")
                                                      or ["?"])[0].lstrip("/"),
              "state": c.get("State"), "status": c.get("Status")}
             for c in (r.get("json") or [])]
    return b.ok(containers=items)


def docker_action(ep, secret, params):
    name = ((params or {}).get("name") or "").strip()
    act = ((params or {}).get("act") or "").strip()
    if act not in ("start", "stop", "restart"):
        return b.err("docker.action: act = start|stop|restart")
    if not name:
        return b.err("docker.action: нужен params.name")
    return _docker_call(ep, "POST", "/containers/%s/%s" % (name, act))


# --- proxmox (REST: ticket или API-token) ---

def _pve(ep, secret, method, path, body=None):
    from . import http as _h
    user = (b.addr(ep, "user") or "").strip()
    is_token = "!" in user  # PVE API-токен: user@realm!tokenid
    fake = {"opts": {"auth": "token" if is_token else "bearer",
                     "timeout": b.timeout(ep),
                     "verify": b.opt(ep, "verify", True),
                     "headers": {}},
            "address": {"base_url": "https://%s:%s/api2/json" % (
                b.addr(ep, "host"), b.addr(ep, "port") or 8006)}}
    sec = secret
    if not is_token:
        # ticket-режим: сначала логин, токен в куку
        login = _h.http_request(
            {"opts": {"timeout": b.timeout(ep),
                      "verify": b.opt(ep, "verify", True)},
             "address": fake["address"]},
            None, {"path": "/access/ticket", "method": "POST",
                   "body": "username=%s&password=%s" % (user, secret or "")})
        if not login.get("ok"):
            return b.err("proxmox login: %s" % login.get("error"))
        data = (login.get("json") or {}).get("data") or {}
        ticket = data.get("ticket", "")
        fake["opts"]["headers"] = {
            "Cookie": "PVEAuthCookie=" + ticket,
            "CSRFPreventionToken": data.get("CSRFPreventionToken", "")}
        sec = None
    else:
        fake["opts"]["headers"] = {"Authorization": "PVEAPIToken=%s=%s" % (
            user, secret or "")}
        sec = None
    params = {"path": path, "method": method}
    if body is not None:
        params["json"] = body
    return _h.http_request(fake, sec, params)


def proxmox_get(ep, secret, params):
    path = ((params or {}).get("path") or "").strip()
    if not path.startswith("/"):
        return b.err("proxmox.get: path с ведущим /")
    return _pve(ep, secret, "GET", path)


def proxmox_action(ep, secret, params):
    """Действие с ВМ/контейнером: {node, type=qemu|lxc, vmid, act}."""
    params = params or {}
    node = (params.get("node") or "").strip()
    vmid = str(params.get("vmid") or "").strip()
    act = (params.get("act") or "").strip()
    vtype = (params.get("type") or "qemu").strip()
    if act not in ("start", "stop", "shutdown", "reboot"):
        return b.err("proxmox.action: act=start|stop|shutdown|reboot")
    if not node or not vmid:
        return b.err("proxmox.action: нужны node и vmid")
    return _pve(ep, secret, "POST", "/nodes/%s/%s/%s/status/%s" % (
        node, vtype, vmid, act))


# --- telnet (сырые сокеты + expect) ---

def telnet_exec(ep, secret, params):
    host = (b.addr(ep, "host") or "").strip()
    try:
        port = int(b.addr(ep, "port") or 23)
    except (TypeError, ValueError):
        port = 23
    script = (params or {}).get("script") or []
    if not host or not script:
        return b.err("telnet.exec: нужны address.host и params.script "
                     "[{send|expect, ...}]")
    user = (b.addr(ep, "user") or "").strip()
    t0 = b.now_ms()
    try:
        s = _socket.create_connection((host, port), timeout=b.timeout(ep))
        s.settimeout(b.timeout(ep))
        buf = b""
        transcript = []

        def _read_until(marker, timeout_s):
            nonlocal buf
            end = _time.time() + timeout_s
            want = marker.encode()
            while _time.time() < end:
                if want in buf:
                    idx = buf.index(want) + len(want)
                    chunk, buf = buf[:idx], buf[idx:]
                    return chunk
                try:
                    data = s.recv(4096)
                except Exception:  # noqa: BLE001
                    break
                if not data:
                    break
                buf += _telnet_strip(data)
            chunk, buf = buf, b""
            return chunk

        _read_until("ogin:", 8)
        if user:
            s.sendall((user + "\n").encode())
            _read_until("assword:", 8)
            if secret:
                s.sendall((secret + "\n").encode())
        _read_until("#", 8)
        _read_until(">", 3)
        for step in script:
            if "expect" in step:
                chunk = _read_until(step["expect"], b.timeout(ep))
                transcript.append(chunk.decode("utf-8",
                                               "replace")[-2000:])
            elif "send" in step:
                s.sendall((step["send"] + "\n").encode())
                _time.sleep(0.3)
        try:
            s.sendall(b"exit\n")
        except Exception:  # noqa: BLE001
            pass
        s.close()
        return b.ok(transcript="\n".join(transcript)[-8000:],
                    ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        return b.err("telnet %s: %s" % (host, e))


def _telnet_strip(data):
    """Вырезать IAC-последовательности."""
    out = bytearray()
    i = 0
    while i < len(data):
        if data[i] == 0xFF and i + 2 < len(data):
            i += 3
        else:
            out.append(data[i])
            i += 1
    return bytes(out)


# --- dry-run ---

def dryrun_call(ep, secret, params):
    return b.ok(dry_run=True, endpoint=ep.get("name"),
                echo=(params or {}))


OPS = {
    "winrm.exec": winrm_exec,
    "docker.ps": docker_ps,
    "docker.action": docker_action,
    "proxmox.get": proxmox_get,
    "proxmox.action": proxmox_action,
    "telnet.exec": telnet_exec,
    "dryrun.call": dryrun_call,
}


def health(ep, secret):
    kind = (ep.get("kind") or "").strip()
    if kind == "docker":
        r = _docker_call(ep, "GET", "/version")
        return b.ok(version=((r.get("json") or {}).get("Version", "?"))
                    ) if r.get("ok") else r
    if kind == "proxmox":
        r = _pve(ep, secret, "GET", "/version")
        return b.ok() if r.get("ok") else r
    if kind == "winrm":
        r = winrm_exec(ep, secret, {"cmd": "echo outbound-ok"})
        return b.ok() if r.get("ok") else r
    if kind == "telnet":
        from . import net as _n
        host = (b.addr(ep, "host") or "").strip()
        return _n.tcp_open({"address": {"host": host,
                                        "port": b.addr(ep, "port") or 23},
                            "opts": ep.get("opts")}, None, {})
    if kind == "dryrun":
        return b.ok(dry_run=True)
    return b.err("mgmt: нет health для kind=%s" % kind)