outbound v0.5.2 [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 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 run(cmd, timeout_s=15, env=None, input_text=None):
    """subprocess без shell. Возвращает (rc, stdout, stderr)."""
    try:
        r = subprocess.run(
            cmd, capture_output=True, text=True, errors="replace",
            timeout=timeout_s, env=env, input=input_text)
        return r.returncode, (r.stdout or "")[:8000], (r.stderr or "")[: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 os as _os

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 = (rel or "").strip().lstrip("/")
    full = _os.path.realpath(_os.path.join(root, rel))
    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)


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,
}


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)
        text = (r.data or b"")[:200000].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
    try:
        r = pm.request(method, url, body=data, headers=headers,
                       timeout=timeout, 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)

"""outbound.drivers.msg — mqtt (paho, опционально) и ws (свой мини-клиент stdlib).

mqtt: publish + subscribe-разовый (слушаем N секунд, собираем сообщения).
ws: текстовые фреймы (RFC 6455, без extensions), wss через ssl.
"""
import socket as _socket
import ssl as _ssl

from . import base as b

KINDS = ("mqtt", "ws")


# --- mqtt ---

def _mqtt_client(ep, secret):
    paho = b.lazy("paho.mqtt.client")
    if not paho:
        return None, b.missing("paho-mqtt", "mqtt")
    user = (b.addr(ep, "user") or "").strip()
    try:
        cli = paho.Client()
    except Exception as e:  # noqa: BLE001
        return None, b.err("mqtt: %s" % e)
    if user:
        cli.username_pw_set(user, secret or None)
    tls = b.opt(ep, "tls", False)
    if tls:
        try:
            cli.tls_set()
        except Exception as e:  # noqa: BLE001
            return None, b.err("mqtt tls: %s" % e)
    return cli, None


def _mqtt_addr(ep):
    host = (b.addr(ep, "host") or "").strip()
    try:
        port = int(b.addr(ep, "port") or 1883)
    except (TypeError, ValueError):
        port = 1883
    return host, port


def mqtt_publish(ep, secret, params):
    params = params or {}
    topic = (params.get("topic") or "").strip()
    if not topic:
        return b.err("mqtt.publish: нужен params.topic")
    payload = params.get("payload", "")
    if not isinstance(payload, str):
        import json as _j
        payload = _j.dumps(payload, ensure_ascii=False)
    try:
        qos = int(params.get("qos", 0))
    except (TypeError, ValueError):
        qos = 0
    host, port = _mqtt_addr(ep)
    cli, err = _mqtt_client(ep, secret)
    if err:
        return err
    t0 = b.now_ms()
    try:
        cli.connect(host, port, keepalive=30)
        info = cli.publish(topic, payload, qos=max(0, min(qos, 2)))
        info.wait_for_publish(timeout=b.timeout(ep))
        cli.disconnect()
        return b.ok(topic=topic, ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        try:
            cli.disconnect()
        except Exception:  # noqa: BLE001
            pass
        return b.err("mqtt.publish %s: %s" % (topic, e))


def mqtt_subscribe(ep, secret, params):
    params = params or {}
    topic = (params.get("topic") or "").strip()
    if not topic:
        return b.err("mqtt.subscribe: нужен params.topic")
    try:
        wait_s = int(params.get("wait", 10))
    except (TypeError, ValueError):
        wait_s = 10
    wait_s = max(1, min(wait_s, 120))
    host, port = _mqtt_addr(ep)
    cli, err = _mqtt_client(ep, secret)
    if err:
        return err
    import time as _t
    got = []

    def _on_msg(_c, _u, msg):
        try:
            got.append({"topic": msg.topic,
                        "payload": msg.payload.decode("utf-8", "replace")})
        except Exception:  # noqa: BLE001
            pass

    t0 = b.now_ms()
    try:
        cli.on_message = _on_msg
        cli.connect(host, port, keepalive=30)
        cli.subscribe(topic)
        cli.loop_start()
        _t.sleep(wait_s)
        cli.loop_stop()
        cli.disconnect()
        return b.ok(topic=topic, messages=got[:50],
                    count=len(got), ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        try:
            cli.loop_stop()
            cli.disconnect()
        except Exception:  # noqa: BLE001
            pass
        return b.err("mqtt.subscribe %s: %s" % (topic, e))


# --- ws (мини-клиент, текстовые фреймы) ---

def _ws_key():
    import base64 as _b64
    import os as _o
    return _b64.b64encode(_o.urandom(16)).decode()


def _ws_connect(ep, secret, params):
    url = ((params or {}).get("url") or b.addr(ep, "base_url")
           or "").strip()
    if not url.startswith(("ws://", "wss://")):
        return None, b.err("ws: нужен ws:// или wss:// url")
    secure = url.startswith("wss://")
    rest = url[5:] if secure else url[4:]
    hostport, _, path = rest.partition("/")
    path = "/" + path
    if ":" in hostport:
        host, _, port = hostport.partition(":")
        port = int(port)
    else:
        host, port = hostport, 443 if secure else 80
    key = _ws_key()
    req = ("GET %s HTTP/1.1\r\nHost: %s\r\nUpgrade: websocket\r\n"
           "Connection: Upgrade\r\nSec-WebSocket-Key: %s\r\n"
           "Sec-WebSocket-Version: 13\r\n" % (path, host, key))
    token = (b.opt(ep, "auth") or "").strip().lower()
    if token == "bearer" and secret:
        req += "Authorization: Bearer %s\r\n" % secret
    req += "\r\n"
    t0 = b.now_ms()
    try:
        s = _socket.create_connection((host, port), timeout=b.timeout(ep))
        if secure:
            ctx = _ssl.create_default_context()
            if b.opt(ep, "verify", True) is False:
                ctx.check_hostname = False
                ctx.verify_mode = _ssl.CERT_NONE
            s = ctx.wrap_socket(s, server_hostname=host)
        s.sendall(req.encode())
        resp = b""
        while b"\r\n\r\n" not in resp:
            chunk = s.recv(4096)
            if not chunk:
                break
            resp += chunk
        if b" 101 " not in resp.split(b"\r\n", 1)[0]:
            s.close()
            return None, b.err("ws: handshake fail: %s" %
                               resp[:120].decode("utf-8", "replace"))
        return (s, b.now_ms() - t0), None
    except Exception as e:  # noqa: BLE001
        return None, b.err("ws.connect: %s" % e)


def _ws_send_text(s, text):
    import os as _o
    import struct as _st
    data = text.encode("utf-8")
    mask = _o.urandom(4)
    hdr = bytes([0x81])
    ln = len(data)
    if ln < 126:
        hdr += _st.pack("!B", 0x80 | ln)
    elif ln < 65536:
        hdr += _st.pack("!BH", 0x80 | 126, ln)
    else:
        hdr += _st.pack("!BQ", 0x80 | 127, ln)
    masked = bytes(c ^ mask[i % 4] for i, c in enumerate(data))
    s.sendall(hdr + mask + masked)


def _ws_recv(s, timeout_s):
    import struct as _st
    s.settimeout(timeout_s)
    try:
        hdr = s.recv(2)
        if len(hdr) < 2:
            return None
        ln = hdr[1] & 0x7F
        if ln == 126:
            ln = _st.unpack("!H", s.recv(2))[0]
        elif ln == 127:
            ln = _st.unpack("!Q", s.recv(8))[0]
        data = b""
        while len(data) < ln:
            chunk = s.recv(ln - len(data))
            if not chunk:
                break
            data += chunk
        opcode = hdr[0] & 0x0F
        if opcode == 0x8:
            return None
        return data.decode("utf-8", "replace")
    except Exception:  # noqa: BLE001
        return None


def ws_roundtrip(ep, secret, params):
    """Отправить текст, собрать до N ответов (wait_s)."""
    params = params or {}
    text = (params.get("send") or "")
    if not isinstance(text, str):
        return b.err("ws.roundtrip: нужен params.send (текст)")
    try:
        wait_s = int(params.get("wait", 5))
    except (TypeError, ValueError):
        wait_s = 5
    wait_s = max(1, min(wait_s, 60))
    (conn, ms0), err = _ws_connect(ep, secret, params)
    if err:
        return err
    s = conn
    import time as _t
    t0 = b.now_ms()
    try:
        _ws_send_text(s, text)
        got = []
        deadline = _t.time() + wait_s
        while _t.time() < deadline and len(got) < 20:
            msg = _ws_recv(s, max(0.5, deadline - _t.time()))
            if msg is None:
                break
            got.append(msg[:4000])
        s.close()
        return b.ok(messages=got, count=len(got), connect_ms=ms0,
                    ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        try:
            s.close()
        except Exception:  # noqa: BLE001
            pass
        return b.err("ws.roundtrip: %s" % e)


OPS = {
    "mqtt.publish": mqtt_publish,
    "mqtt.subscribe": mqtt_subscribe,
    "ws.roundtrip": ws_roundtrip,
}


def health(ep, secret):
    kind = (ep.get("kind") or "").strip()
    if kind == "mqtt":
        host, port = _mqtt_addr(ep)
        cli, err = _mqtt_client(ep, secret)
        if err:
            return err
        t0 = b.now_ms()
        try:
            cli.connect(host, port, keepalive=10)
            cli.disconnect()
            return b.ok(ms=b.now_ms() - t0)
        except Exception as e:  # noqa: BLE001
            return b.err("mqtt health: %s" % e)
    if kind == "ws":
        (conn, ms0), err = _ws_connect(ep, secret, {})
        if err:
            return err
        try:
            conn.close()
        except Exception:  # noqa: BLE001
            pass
        return b.ok(ms=ms0)
    return b.err("msg: нет health для kind=%s" % kind)

"""outbound.drivers.net — L3-L4: tcp/udp/icmp/dns/wol. Только stdlib."""
import socket
import struct
import time

from . import base as b

KINDS = ("tcp", "udp", "icmp", "dns", "wol")


def _target(ep):
    host = (b.addr(ep, "host") or "").strip()
    try:
        port = int(b.addr(ep, "port") or 0)
    except (TypeError, ValueError):
        port = 0
    return host, port


# --- tcp ---

def tcp_open(ep, secret, params):
    host, port = _target(ep)
    if not host or not port:
        return b.err("tcp.open: нужны address.host и address.port")
    t0 = b.now_ms()
    try:
        s = socket.create_connection((host, port),
                                     timeout=b.timeout(ep))
        s.close()
        return b.ok(ms=b.now_ms() - t0, host=host, port=port)
    except Exception as e:  # noqa: BLE001
        return b.err("tcp %s:%s: %s" % (host, port, e),
                     ms=b.now_ms() - t0)


def tcp_banner(ep, secret, params):
    """Открыть, прочитать до N байт (баннер), закрыть."""
    host, port = _target(ep)
    try:
        n = int((params or {}).get("bytes", 256))
    except (TypeError, ValueError):
        n = 256
    n = max(1, min(n, 4096))
    t0 = b.now_ms()
    try:
        s = socket.create_connection((host, port), timeout=b.timeout(ep))
        s.settimeout(b.timeout(ep))
        data = s.recv(n)
        s.close()
        return b.ok(banner=data.decode("utf-8", "replace"),
                    ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        return b.err("tcp.banner %s:%s: %s" % (host, port, e))


# --- udp ---

def udp_send(ep, secret, params):
    host, port = _target(ep)
    payload = ((params or {}).get("data") or "")
    if isinstance(payload, str):
        payload = payload.encode("utf-8")
    t0 = b.now_ms()
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.settimeout(b.timeout(ep))
        s.sendto(bytes(payload), (host, port))
        s.close()
        return b.ok(sent=len(bytes(payload)), ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        return b.err("udp %s:%s: %s" % (host, port, e))


# --- icmp (через системный ping, без root) ---

def icmp_ping(ep, secret, params):
    host, _ = _target(ep)
    if not host:
        return b.err("icmp.ping: нужен address.host")
    try:
        count = int((params or {}).get("count", 2))
    except (TypeError, ValueError):
        count = 2
    count = max(1, min(count, 5))
    t0 = b.now_ms()
    rc, out, err = b.run(["ping", "-c", str(count), "-W", "2", host],
                         timeout_s=b.timeout(ep) + 5)
    if rc == 0:
        return b.ok(ms=b.now_ms() - t0, detail=_ping_summary(out))
    return b.err("ping %s: %s" % (host, (err or out or "rc=%s" % rc)[:200]),
                 ms=b.now_ms() - t0)


def _ping_summary(out):
    for line in (out or "").splitlines():
        if "min/avg/max" in line:
            return line.strip()[:200]
    return ""


# --- dns (резолв + запрос записей по UDP) ---

def dns_resolve(ep, secret, params):
    name = ((params or {}).get("name") or b.addr(ep, "host") or "").strip()
    if not name:
        return b.err("dns.resolve: нужны params.name или address.host")
    t0 = b.now_ms()
    try:
        infos = socket.getaddrinfo(name, None, socket.AF_UNSPEC,
                                   socket.SOCK_STREAM)
        addrs = sorted({i[4][0] for i in infos})
        return b.ok(name=name, addresses=addrs, ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        return b.err("dns %s: %s" % (name, e))


def dns_query(ep, secret, params):
    """Запрос записи типа A/AAAA/MX/TXT к address.host (DNS-сервер)."""
    server = (b.addr(ep, "host") or "77.88.8.8").strip()
    try:
        port = int(b.addr(ep, "port") or 53)
    except (TypeError, ValueError):
        port = 53
    name = ((params or {}).get("name") or "").strip()
    rtype = ((params or {}).get("type") or "A").strip().upper()
    if not name:
        return b.err("dns.query: нужен params.name")
    pkt = _dns_packet(name, rtype)
    if pkt is None:
        return b.err("dns.query: тип %s не поддерживается" % rtype)
    t0 = b.now_ms()
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.settimeout(b.timeout(ep))
        s.sendto(pkt, (server, port))
        data, _ = s.recvfrom(2048)
        s.close()
        return b.ok(name=name, type=rtype, answers=_dns_parse(data),
                    ms=b.now_ms() - t0)
    except Exception as e:  # noqa: BLE001
        return b.err("dns.query %s %s: %s" % (name, rtype, e))


_TYPEMAP = {"A": 1, "AAAA": 28, "MX": 15, "TXT": 16, "CNAME": 5}


def _dns_packet(name, rtype):
    import random as _r
    qtype = _TYPEMAP.get(rtype)
    if not qtype:
        return None
    hdr = struct.pack(">HHHHHH", _r.getrandbits(16), 0x0100, 1, 0, 0, 0)
    qname = b"".join(bytes([len(p)]) + p.encode("ascii", "ignore")
                     for p in name.split(".") if p) + b"\x00"
    return hdr + qname + struct.pack(">HH", qtype, 1)


def _dns_parse(data):
    """Минимум: вытащить A/AAAA/CNAME/TXT из answer-секции."""
    out = []
    try:
        ancount = struct.unpack(">H", data[6:8])[0]
        pos = 12
        while data[pos] != 0:  # пропуск question
            pos += data[pos] + 1
        pos += 5
        for _ in range(min(ancount, 10)):
            while pos < len(data) and data[pos] >= 0xC0:
                pos += 2
                break
            else:
                while pos < len(data) and data[pos] != 0:
                    pos += data[pos] + 1
                pos += 1
            if pos + 10 > len(data):
                break
            rtype, _, _, rdlen = struct.unpack(">HHIH", data[pos:pos + 10])
            pos += 10
            rdata = data[pos:pos + rdlen]
            pos += rdlen
            if rtype == 1 and rdlen == 4:
                out.append(socket.inet_ntoa(rdata))
            elif rtype == 28 and rdlen == 16:
                out.append(socket.inet_ntop(socket.AF_INET6, rdata))
            elif rtype in (5,):
                out.append("cname")
            elif rtype == 16:
                out.append(rdata[1:].decode("utf-8", "replace"))
    except Exception:  # noqa: BLE001
        pass
    return out


# --- wol ---

def wol_send(ep, secret, params):
    mac = ((params or {}).get("mac") or b.addr(ep, "mac") or "")
    mac = "".join(c for c in mac if c.isalnum()).lower()
    if len(mac) != 12:
        return b.err("wol.send: нужен MAC (params.mac или address.mac)")
    bcast = (b.addr(ep, "broadcast") or "255.255.255.255").strip()
    try:
        port = int(b.addr(ep, "port") or 9)
    except (TypeError, ValueError):
        port = 9
    pkt = b"\xff" * 6 + bytes.fromhex(mac) * 16
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
        s.sendto(pkt, (bcast, port))
        s.close()
        return b.ok(mac=mac, via="%s:%s" % (bcast, port))
    except Exception as e:  # noqa: BLE001
        return b.err("wol: %s" % e)


OPS = {
    "tcp.open": tcp_open,
    "tcp.banner": tcp_banner,
    "udp.send": udp_send,
    "icmp.ping": icmp_ping,
    "dns.resolve": dns_resolve,
    "dns.query": dns_query,
    "wol.send": wol_send,
}


def health(ep, secret):
    kind = (ep.get("kind") or "").strip()
    if kind == "tcp":
        return tcp_open(ep, secret, {})
    if kind in ("udp", "wol"):
        host, _ = _target(ep)
        return icmp_ping({"address": {"host": host or "127.0.0.1"},
                          "opts": ep.get("opts")}, secret, {"count": 1})
    if kind == "dns":
        return dns_query(ep, secret, {"name": "ya.ru", "type": "A"})
    if kind == "icmp":
        return icmp_ping(ep, secret, {"count": 2})
    return b.err("net: нет health для kind=%s" % kind)