remotedesk v0.6.0 [extra]

"""remotedesk.creds — шифрованное хранилище учётных данных.

Файл data/remotedesk/credentials.json.enc — AES-GCM (pycryptodome).
Ключ: CRED_KEY из .env модуля (base64 32 байта) либо автогенерация
в data/remotedesk/.credkey (0600). Секреты наружу не отдаём:
list() — только метаданные, секрет — через secret() внутри процесса.

Модель: {id, name, username, kind (password|key), note, owners, updated_at}
+ secret (только внутри).
"""

import base64
import json
import os
import re
import time

_VERSION = 1

_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")


def _read_text(path):
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    except OSError:
        return ""


def _write_text(path, text):
    d = os.path.dirname(os.path.abspath(path))
    os.makedirs(d, exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(text)
    try:
        os.chmod(tmp, 0o600)
    except OSError:
        pass
    os.replace(tmp, path)
    try:
        os.chmod(path, 0o600)
    except OSError:
        pass


def resolve_key(config=None, data_dir="", credkey_path=""):
    """Вернуть 32-байтный ключ. Источники: CRED_KEY (.env) или файл .credkey."""
    cfg = config or {}
    raw = str(cfg.get("CRED_KEY") or "").strip()
    if raw:
        # base64 (44 символа) либо hex (64) либо сырая строка -> sha256.
        try:
            b = base64.b64decode(raw, validate=False)
            if len(b) == 32:
                return b
        except Exception:
            pass
        try:
            b = bytes.fromhex(raw)
            if len(b) == 32:
                return b
        except Exception:
            pass
        if len(raw) >= 16:
            import hashlib as _h
            return _h.sha256(raw.encode()).digest()
    # Файловый ключ (автогенерация один раз).
    kp = credkey_path or (os.path.join(data_dir or ".", "remotedesk",
                                       ".credkey") if data_dir else "")
    if kp:
        txt = _read_text(kp).strip() if os.path.isfile(kp) else ""
        if txt:
            try:
                b = base64.b64decode(txt, validate=False)
                if len(b) == 32:
                    return b
            except Exception:
                pass
        import secrets as _s
        key = _s.token_bytes(32)
        try:
            _write_text(kp, base64.b64encode(key).decode())
        except OSError:
            pass
        return key
    import secrets as _s
    return _s.token_bytes(32)


def _aes():
    try:
        from Crypto.Cipher import AES as _AES
    except ImportError:
        raise ValueError("нет пакета pycryptodome "
                         "(нужен для шифрования кредов remotedesk)")
    return _AES


def encrypt_blob(key, plain: bytes) -> str:
    import secrets as _s
    aes = _aes()
    nonce = _s.token_bytes(12)
    cipher = aes.new(key, aes.MODE_GCM, nonce=nonce)
    ct, tag = cipher.encrypt_and_digest(plain)
    return json.dumps({
        "v": 1,
        "nonce": base64.b64encode(nonce).decode(),
        "ct": base64.b64encode(ct).decode(),
        "tag": base64.b64encode(tag).decode(),
    })


def decrypt_blob(key, blob: str) -> bytes:
    aes = _aes()
    try:
        obj = json.loads(blob)
        nonce = base64.b64decode(obj["nonce"])
        ct = base64.b64decode(obj["ct"])
        tag = base64.b64decode(obj["tag"])
    except Exception:
        raise ValueError("файл кредов повреждён")
    try:
        cipher = aes.new(key, aes.MODE_GCM, nonce=nonce)
        return cipher.decrypt_and_verify(ct, tag)
    except Exception:
        raise ValueError("неверный CRED_KEY или файл повреждён")


class CredStore:
    def __init__(self, path, key):
        self.path = path
        self.key = key

    def load(self):
        """Все записи WITH секретами (только внутри процесса)."""
        if not os.path.isfile(self.path):
            return {}
        blob = _read_text(self.path)
        if not blob.strip():
            return {}
        raw = decrypt_blob(self.key, blob)
        try:
            data = json.loads(raw.decode("utf-8"))
        except Exception:
            raise ValueError("файл кредов повреждён")
        creds = data.get("creds") if isinstance(data, dict) else None
        return creds if isinstance(creds, dict) else {}

    def save(self, creds):
        plain = json.dumps({"version": _VERSION, "creds": creds or {}},
                           ensure_ascii=False).encode("utf-8")
        _write_text(self.path, encrypt_blob(self.key, plain))

    def public(self, rec):
        return {
            "id": rec.get("id", ""),
            "name": rec.get("name", ""),
            "username": rec.get("username", ""),
            "kind": rec.get("kind", "password"),
            "note": rec.get("note", ""),
            "owners": list(rec.get("owners") or []),
            "updated_at": rec.get("updated_at", ""),
            "has_secret": bool(rec.get("secret")),
        }

    def list(self):
        try:
            creds = self.load()
        except ValueError:
            return []
        return [self.public(v) for _, v in sorted(creds.items())]

    def validate(self, fields, partial=False):
        if not isinstance(fields, dict):
            return None, "нужен объект"
        norm = {}
        if not partial or "name" in fields:
            name = str(fields.get("name") or "").strip()
            if not name or len(name) > 128:
                return None, "нужно name (1-128 символов)"
            norm["name"] = name
        if "username" in fields or not partial:
            username = str(fields.get("username") or "").strip()
            if len(username) > 128:
                return None, "username: до 128 символов"
            norm["username"] = username
        if "kind" in fields or not partial:
            kind = str(fields.get("kind") or "password").strip().lower()
            if kind not in ("password", "key"):
                return None, "kind: password|key"
            norm["kind"] = kind
        if "secret" in fields or not partial:
            secret = fields.get("secret")
            secret = "" if secret is None else str(secret)
            if len(secret) > 8192:
                return None, "secret: до 8 КБ"
            norm["secret"] = secret
        if "note" in fields:
            note = str(fields.get("note") or "").strip()
            if len(note) > 256:
                return None, "note: до 256 символов"
            norm["note"] = note
        if "owners" in fields:
            owners = fields.get("owners")
            if owners is None:
                norm["owners"] = []
            elif isinstance(owners, list) and all(
                    isinstance(x, str) for x in owners):
                norm["owners"] = owners[:64]
            else:
                return None, "owners: список строк"
        return norm, ""

    def add(self, fields, actor=""):
        norm, err = self.validate(fields)
        if err:
            raise ValueError(err)
        import uuid as _uuid
        cid = _uuid.uuid4().hex[:12]
        if not _ID_RE.match(cid):
            raise ValueError("внутренняя ошибка id")
        rec = {
            "id": cid,
            "owners": [actor] if actor else [],
            "created_by": actor or "",
            "updated_at": time.strftime("%Y-%m-%d %H:%M:%S",
                                        time.localtime()),
        }
        rec.update(norm)
        creds = self.load()
        creds[cid] = rec
        self.save(creds)
        return {"ok": True, "id": cid}

    def update(self, cid, patch):
        if not cid or not _ID_RE.match(str(cid)):
            raise ValueError("плохой id")
        norm, err = self.validate(patch or {}, partial=True)
        if err:
            raise ValueError(err)
        creds = self.load()
        rec = creds.get(cid)
        if not isinstance(rec, dict):
            raise ValueError("кред не найден: %s" % cid)
        rec.update(norm)
        rec["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S",
                                          time.localtime())
        creds[cid] = rec
        self.save(creds)
        return {"ok": True}

    def delete(self, cid):
        if not cid or not _ID_RE.match(str(cid)):
            raise ValueError("плохой id")
        creds = self.load()
        if cid not in creds:
            raise ValueError("кред не найден: %s" % cid)
        del creds[cid]
        self.save(creds)
        return {"ok": True}

    def get(self, cid):
        creds = self.load()
        rec = creds.get(cid)
        return rec if isinstance(rec, dict) else None

"""remotedesk.Module — соединения, креды, SFTP (API + аудит)."""
import logging
import os
import socket

from core.base_module import BaseModule

logger = logging.getLogger(__name__)

# Пул SFTP-клиентов: (actor, cid) -> {client, at}. Один SSH-коннект на
# (uid, conn) вместо нового на каждую файловую операцию. Секреты НЕ
# храним — только живой client; TTL чистит протухшие. Закрытие — через
# _ssh_pool_drop (ошибка канала) или janitor по TTL.
# Ключ пула включает login-отпечаток (cred_id/username): повторный вызов
# с чужим кредом/ручным секретом НЕ должен получить чужое соединение.
import threading as _th

_SSH_POOL = {}
_SSH_POOL_LOCK = _th.Lock()
_SSH_POOL_TTL = 300
_SSH_POOL_MAX_LIFE = 1800


def _login_fp(login):
    """Отпечаток login-оверрайда для ключа пула (без секрета)."""
    if not isinstance(login, dict) or not login:
        return ""
    parts = []
    for k in ("cred_id", "username"):
        v = login.get(k)
        if v is not None and str(v) != "":
            parts.append("%s=%s" % (k, str(v)[:256]))
    return "|".join(parts)


def _pool_key(actor, cid, login):
    return (actor, cid, _login_fp(login))


def _ssh_pool_get(actor, cid, login=None, pin=""):
    import time as _t
    key = _pool_key(actor, cid, login)
    with _SSH_POOL_LOCK:
        rec = _SSH_POOL.get(key)
        if not rec:
            return None
        now = _t.time()
        if (now - rec.get("at", 0) > _SSH_POOL_TTL
                or now - rec.get("born", 0) > _SSH_POOL_MAX_LIFE
                or rec.get("pin", "") != (pin or "")):
            rec = _SSH_POOL.pop(key, None)
        else:
            try:
                tr = rec["client"].get_transport()
                alive = tr is not None and tr.is_active()
            except Exception:  # noqa: BLE001
                alive = False
            if not alive:
                rec = _SSH_POOL.pop(key, None)
            else:
                rec["at"] = now
                return rec["client"]
    if rec is not None:
        try:
            rec["client"].close()
        except Exception:  # noqa: BLE001
            pass
    return None


def _ssh_pool_put(actor, cid, client, login=None, pin=""):
    import time as _t
    key = _pool_key(actor, cid, login)
    with _SSH_POOL_LOCK:
        old = _SSH_POOL.pop(key, None)
        _SSH_POOL[key] = {"client": client, "at": _t.time(),
                          "born": _t.time(), "pin": pin or ""}
    if old is not None:
        try:
            old["client"].close()
        except Exception:  # noqa: BLE001
            pass


def _ssh_pool_touch(actor, cid, login=None, ok=True):
    import time as _t
    key = _pool_key(actor, cid, login)
    with _SSH_POOL_LOCK:
        rec = _SSH_POOL.get(key)
        # Touch только на успехе: умирающий клиент не продлеваем.
        if rec and ok:
            rec["at"] = _t.time()


def _ssh_pool_drop(actor, cid, login=None):
    """Выкинуть клиент из пула. Вернуть True если был."""
    key = _pool_key(actor, cid, login)
    with _SSH_POOL_LOCK:
        rec = _SSH_POOL.pop(key, None)
    if rec is None:
        return False
    try:
        rec["client"].close()
    except Exception:  # noqa: BLE001
        pass
    return True


def _ssh_pool_janitor():
    import time as _t
    now = _t.time()
    dead = []
    with _SSH_POOL_LOCK:
        for key, r in list(_SSH_POOL.items()):
            if (now - r.get("at", 0) > _SSH_POOL_TTL
                    or now - r.get("born", 0) > _SSH_POOL_MAX_LIFE):
                dead.append(key)
        recs = [_SSH_POOL.pop(k, None) for k in dead]
    for rec in recs:
        if rec is None:
            continue
        try:
            rec["client"].close()
        except Exception:  # noqa: BLE001
            pass


class Module(BaseModule):
    def setup(self, ctx) -> None:
        from . import store as _store
        from . import creds as _creds

        mod_dir = os.path.dirname(os.path.abspath(__file__))
        self.config = ctx.module_config("remotedesk", mod_dir)
        self.ctx = ctx

        def _f(key, default):
            return (self.config.get(key, "") or "").strip() or default

        from core.paths import abs_data
        self.conn_file = abs_data(
            ctx.data_dir, _f("CONNECTIONS_FILE", ""),
            os.path.join(ctx.data_dir, "remotedesk", "connections.json"))
        self.creds_file = abs_data(
            ctx.data_dir, _f("CREDENTIALS_FILE", ""),
            os.path.join(ctx.data_dir, "remotedesk",
                         "credentials.json.enc"))
        self.credkey_file = abs_data(
            ctx.data_dir, _f("CREDKEY_FILE", ""),
            os.path.join(ctx.data_dir, "remotedesk", ".credkey"))
        try:
            self.timeout = max(2, min(60, int(
                _f("CONNECT_TIMEOUT", "10") or 10)))
        except (TypeError, ValueError):
            self.timeout = 10
        self.store = _store.Store(self.conn_file)
        try:
            _key = _creds.resolve_key(self.config, ctx.data_dir,
                                      self.credkey_file)
        except ValueError:
            _key = None
        self._cred_key = _key
        self.credstore = _creds.CredStore(self.creds_file, _key) \
            if _key else None

        api = ctx.api.register
        api("remotedesk", "conn_list", self.api_list)
        api("remotedesk", "conn_add", self.api_add)
        api("remotedesk", "conn_update", self.api_update)
        api("remotedesk", "conn_delete", self.api_delete)
        api("remotedesk", "conn_test", self.api_test)
        api("remotedesk", "conn_secret", self.api_secret)
        api("remotedesk", "auth", self.api_auth)
        api("remotedesk", "cred_list", self.api_cred_list)
        api("remotedesk", "cred_add", self.api_cred_add)
        api("remotedesk", "cred_update", self.api_cred_update)
        api("remotedesk", "cred_delete", self.api_cred_delete)
        api("remotedesk", "conn_pin_save", self.api_pin_save)
        api("remotedesk", "files_list", self.api_files_list)
        api("remotedesk", "files_read", self.api_files_read)
        api("remotedesk", "files_write", self.api_files_write)
        api("remotedesk", "files_mkdir", self.api_files_mkdir)
        api("remotedesk", "files_delete", self.api_files_delete)
        api("remotedesk", "files_rename", self.api_files_rename)
        api("remotedesk", "files_chmod", self.api_files_chmod)
        api("remotedesk", "files_chown", self.api_files_chown)
        api("remotedesk", "files_zip", self.api_files_zip)

    def _audit(self, action, target, actor="", detail="", ok=True):
        try:
            self.ctx.api.call("audit.record", actor or "?", action,
                              target, detail, ok)
        except Exception:  # noqa: BLE001
            pass

    def api_list(self):
        return {"ok": True,
                "connections": self.store.list()}

    def api_add(self, fields, actor=""):
        try:
            out = self.store.add(fields or {}, actor or "")
        except ValueError as e:
            self._audit_rec("remotedesk.conn_add", "", actor, str(e), False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.conn_add", out["id"], actor)
        return out

    def api_update(self, cid, patch, actor=""):
        try:
            out = self.store.update(cid, patch or {})
        except ValueError as e:
            self._audit_rec("remotedesk.conn_update", cid or "",
                            actor, str(e), False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.conn_update", cid, actor)
        return out

    def api_delete(self, cid, actor=""):
        try:
            out = self.store.delete(cid)
        except ValueError as e:
            self._audit_rec("remotedesk.conn_delete", cid or "",
                            actor, str(e), False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.conn_delete", cid, actor)
        return out

    def api_test(self, cid):
        """TCP-пробник host:port (без секретов, только reachability)."""
        conns = self.store.load()
        rec = conns.get(cid)
        if not isinstance(rec, dict):
            raise ValueError("соединение не найдено: %s" % cid)
        host = rec.get("host", "")
        port = int(rec.get("port") or 0)
        import time as _time
        t0 = _time.time()
        try:
            with socket.create_connection((host, port),
                                          timeout=self.timeout):
                pass
        except Exception as e:  # noqa: BLE001
            return {"ok": False,
                    "ms": int((_time.time() - t0) * 1000),
                    "error": str(e)[:200]}
        return {"ok": True, "ms": int((_time.time() - t0) * 1000)}

    def _audit_rec(self, action, target, actor="", detail="", ok=True):
        self._audit(action, target, actor, detail, ok)


    def api_secret(self, cid):
        """Секрет соединения: cred_ref (шифр. хранилище) -> auth_ref (.env).

        Только внутри процесса (WS/SFTP). Наружу не отдавать.
        """
        rec = self.store.load().get(cid)
        if not isinstance(rec, dict):
            raise ValueError("соединение не найдено: %s" % cid)
        # 1) привязанный кред из шифрованного хранилища.
        cred_ref = (rec.get("cred_ref") or "").strip()
        if cred_ref and self.credstore is not None:
            try:
                cred = self.credstore.get(cred_ref)
            except ValueError:
                cred = None
            if isinstance(cred, dict) and cred.get("secret"):
                return cred.get("secret") or ""
        # 2) legacy: ключ секрета в .env модуля.
        ref = (rec.get("auth_ref") or "").strip()
        if not ref:
            return ""
        try:
            val = self.config.get(ref, "") if self.config else ""
        except Exception:  # noqa: BLE001
            val = ""
        return val or ""

    def api_login(self, cid):
        """(username, secret): дефолтный логин соединения.

        username — из params.username, секрет — cred_ref/auth_ref.
        Ручной ввод поверх — на уровне WS (login override).
        """
        rec = self.store.load().get(cid)
        if not isinstance(rec, dict):
            raise ValueError("соединение не найдено: %s" % cid)
        params = rec.get("params") or {}
        user = params.get("username", "") or params.get("user", "")
        cred_ref = (rec.get("cred_ref") or "").strip()
        if cred_ref and self.credstore is not None:
            try:
                cred = self.credstore.get(cred_ref)
            except ValueError:
                cred = None
            if isinstance(cred, dict):
                if not user:
                    user = cred.get("username", "")
                if cred.get("secret"):
                    return user, cred.get("secret") or ""
        return user, self.api_secret(cid)

    def api_auth(self, cid, login=None):
        """Разрешить логин соединения с оверрайдом из веба.

        login: {cred_id?, username?, secret?/password?} — ручной ввод
        поверх сохранённого. Возвращает (rec, username, secret).
        Секреты — только внутри процесса, не логировать.
        """
        rec = self.store.load().get(cid)
        if not isinstance(rec, dict):
            raise ValueError("соединение не найдено: %s" % cid)
        user, secret = self.api_login(cid)
        login = login if isinstance(login, dict) else {}
        # Явный кред из списка (вместо привязанного к соединению).
        cred_id = str(login.get("cred_id") or "").strip()
        if cred_id and self.credstore is not None:
            try:
                cred = self.credstore.get(cred_id)
            except ValueError:
                cred = None
            if not isinstance(cred, dict):
                raise ValueError("кред не найден: %s" % cred_id)
            if str(login.get("username") or "").strip():
                user = str(login.get("username") or "").strip()
            elif cred.get("username"):
                user = cred.get("username")
            if cred.get("secret"):
                secret = cred.get("secret")
        # Ручной ввод — высший приоритет.
        if str(login.get("username") or "").strip() and not cred_id:
            user = str(login.get("username") or "").strip()
        manual = login.get("secret", login.get("password", None))
        if manual is not None and str(manual) != "":
            secret = str(manual)
            if len(secret) > 8192:
                raise ValueError("секрет: до 8 КБ")
        return rec, user, secret or ""

    def api_pin_save(self, cid, field, pin, actor=""):
        """Сохранить TOFU-пин (host_key_pin / cert_pin) в params соединения.

        Чтобы подтверждение host-key/сертификата спрашивалось один раз,
        а SFTP (он ходит тем же пином) не падал следом.
        """
        if field not in ("host_key_pin", "cert_pin"):
            raise ValueError("плохое поле пина")
        pin = (pin or "").strip()
        if not pin or len(pin) > 256:
            raise ValueError("плохой пин")
        conns = self.store.load()
        rec = conns.get(cid)
        if not isinstance(rec, dict):
            raise ValueError("соединение не найдено: %s" % cid)
        params = dict(rec.get("params") or {})
        if params.get(field) == pin:
            return {"ok": True}
        params[field] = pin
        self.store.update(cid, {"params": params})
        self._audit_rec("remotedesk.pin_save", cid, actor, field)
        return {"ok": True}

    # --- креды (шифрованное хранилище) ---

    def _creds_ok(self):
        if self.credstore is None:
            raise ValueError("хранилище кредов недоступно "
                             "(нет pycryptodome)")
        return self.credstore

    def api_cred_list(self):
        return {"ok": True,
                "creds": self._creds_ok().list()}

    def api_cred_add(self, fields, actor=""):
        try:
            out = self._creds_ok().add(fields or {}, actor or "")
        except ValueError as e:
            self._audit_rec("remotedesk.cred_add", "", actor, str(e),
                            False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.cred_add", out["id"], actor)
        return out

    def api_cred_update(self, cid, patch, actor=""):
        try:
            out = self._creds_ok().update(cid, patch or {})
        except ValueError as e:
            self._audit_rec("remotedesk.cred_update", cid or "",
                            actor, str(e), False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.cred_update", cid, actor)
        return out

    def api_cred_delete(self, cid, actor=""):
        # Не даём удалить кред, пока на него ссылаются соединения.
        try:
            conns = self.store.load()
            used = sorted(
                c.get("name") or cid2 for cid2, c in conns.items()
                if isinstance(c, dict) and
                (c.get("cred_ref") or "").strip() == cid)
            if used:
                raise ValueError("кред используется: %s" % ", ".join(used))
            out = self._creds_ok().delete(cid)
        except ValueError as e:
            self._audit_rec("remotedesk.cred_delete", cid or "",
                            actor, str(e), False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.cred_delete", cid, actor)
        return out

    def _resolve_ssh(self, cid, actor="", login=None, reuse=False):
        """(rec, client) для SFTP-операции.

        reuse=False (дефолт) — свежий клиент, закрыть вызывающему
        (как раньше). reuse=True — клиент из пула, НЕ закрывать:
        один SSH/SFTP-коннект на (uid, conn).
        """
        from . import ssh_client as _ssh

        try:
            rec, user, secret = self.api_auth(cid, login)
        except ValueError:
            raise
        if rec.get("protocol") != "ssh":
            raise ValueError("файлы — только ssh-соединения")
        owners = rec.get("owners") or []
        if owners and (actor or "") not in owners:
            self._audit_rec("remotedesk.files_denied", cid, actor, "", False)
            raise ValueError("нет доступа (owners)")
        params = dict(rec.get("params") or {})
        if not (params.get("host_key_pin") or "").strip():
            raise ValueError("нет host-key пина: сначала подключитесь "
                             "терминалом и подтвердите ключ")
        try:
            max_mb = float(self.config.get("FILE_MAX_MB", "10") or 10)
        except (TypeError, ValueError):
            max_mb = 10
        max_bytes = max(1, min(100, int(max_mb))) * 1048576
        _ssh_pool_janitor()
        pin = (params.get("host_key_pin") or "").strip()
        if reuse:
            pooled = _ssh_pool_get(actor, cid, login, pin)
            if pooled is not None:
                return rec, pooled, max_bytes
        client = _ssh.connect(rec.get("host", ""),
                              int(rec.get("port") or 22),
                              user, secret, params, self.timeout)
        if reuse:
            _ssh_pool_put(actor, cid, client, login, pin)
        return rec, client, max_bytes

    def _sftp_call(self, cid, actor, login, fn_name, *args):
        """SFTP-операция через пул (клиент НЕ закрываем).

        При смерти транспорта — drop + один ретрай свежим коннектом.
        Смерть определяем состоянием транспорта (is_active), а не
        префиксом строки (транспортные ошибки середины операции
        завёрнуты как 'sftp list:/stat:/...' и по строке не ловятся).
        Возвращает (out, rec).
        """
        from . import ssh_client as _ssh

        fn = getattr(_ssh, fn_name)
        rec, client, max_bytes = self._resolve_ssh(cid, actor, login,
                                                   reuse=True)
        ok = False
        try:
            try:
                out = self._call_sftp_fn(fn, client, max_bytes, *args)
            except TypeError:
                out = fn(client, *args)
            ok = True
            return out, rec
        except ValueError as e:
            if self._transport_dead(client) and _ssh_pool_drop(
                    actor, cid, login):
                rec2, client2, mb2 = self._resolve_ssh(cid, actor, login,
                                                       reuse=True)
                try:
                    try:
                        out2 = self._call_sftp_fn(fn, client2, mb2,
                                                  *args)
                    except TypeError:
                        out2 = fn(client2, *args)
                    ok = True
                    return out2, rec2
                except ValueError as e2:
                    raise ValueError(str(e2))
            raise ValueError(str(e))
        finally:
            _ssh_pool_touch(actor, cid, login, ok=ok)

    @staticmethod
    def _transport_dead(client):
        try:
            tr = client.get_transport() if client is not None else None
            return tr is None or not tr.is_active()
        except Exception:  # noqa: BLE001
            return True

    @staticmethod
    def _call_sftp_fn(fn, client, max_bytes, *args):
        import inspect as _insp
        try:
            params = _insp.signature(fn).parameters
        except (TypeError, ValueError):
            params = {}
        if "max_bytes" in params:
            return fn(client, *args, max_bytes=max_bytes)
        return fn(client, *args)

    def api_files_list(self, cid, path="", actor="", login=None):
        try:
            out, rec = self._sftp_call(cid, actor, login, "sftp_list",
                                       path or ".")
        except ValueError as e:
            raise ValueError(str(e))
        self._audit_rec("remotedesk.files_list", "%s:%s" % (cid, path),
                        actor)
        return {"ok": True, "conn": rec.get("name", cid), **out}

    def api_files_read(self, cid, path, actor="", login=None):
        try:
            out, _ = self._sftp_call(cid, actor, login, "sftp_read", path)
        except ValueError as e:
            self._audit_rec("remotedesk.files_read", "%s:%s" % (
                cid, path), actor, str(e)[:200], False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.files_read", "%s:%s" % (cid, path),
                        actor)
        return {"ok": True, **out}

    def api_files_write(self, cid, path, text="", actor="", login=None,
                        binary=False):
        if not isinstance(text, (str, bytes)):
            raise ValueError("плохой text: нужен str/bytes")
        if binary:
            # Бинарная заливка: фронт шлёт base64 строкой.
            import base64 as _b64
            try:
                raw_in = text.encode() if isinstance(text, str) else text
                payload = _b64.b64decode(raw_in, validate=True)
            except Exception:  # noqa: BLE001
                raise ValueError("плохой base64")
        else:
            if not isinstance(text, str):
                raise ValueError("плохой text: нужен str")
            payload = text
        try:
            out, _ = self._sftp_call(cid, actor, login, "sftp_write",
                                     path, payload)
        except ValueError as e:
            self._audit_rec("remotedesk.files_write", path or "",
                            actor, str(e)[:200], False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.files_write", "%s:%s" % (cid, path),
                        actor)
        return out

    def api_files_mkdir(self, cid, path, actor="", login=None):
        try:
            out, _ = self._sftp_call(cid, actor, login, "sftp_mkdir", path)
        except ValueError as e:
            raise ValueError(str(e))
        self._audit_rec("remotedesk.files_mkdir", "%s:%s" % (cid, path),
                        actor)
        return out

    def api_files_delete(self, cid, path, recursive=False, actor="",
                         login=None):
        try:
            if recursive:
                out, _ = self._sftp_call(cid, actor, login, "sftp_rmtree",
                                         path)
            else:
                out, _ = self._sftp_call(cid, actor, login,
                                         "sftp_delete", path)
        except ValueError as e:
            self._audit_rec("remotedesk.files_delete", path or "",
                            actor, str(e)[:200], False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.files_delete", "%s:%s" % (cid, path),
                        actor)
        return out

    def api_files_rename(self, cid, src, dst, actor="", login=None):
        try:
            out, _ = self._sftp_call(cid, actor, login, "sftp_rename",
                                     src, dst)
        except ValueError as e:
            self._audit_rec("remotedesk.files_rename", src or "",
                            actor, str(e)[:200], False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.files_rename", "%s:%s->%s" % (
            cid, src, dst), actor)
        return out

    def api_files_chmod(self, cid, path, mode, actor="", login=None):
        try:
            out, _ = self._sftp_call(cid, actor, login, "sftp_chmod",
                                     path, mode)
        except ValueError as e:
            self._audit_rec("remotedesk.files_chmod", path or "",
                            actor, str(e)[:200], False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.files_chmod", "%s:%s=%s" % (
            cid, path, mode), actor)
        return out

    def api_files_chown(self, cid, path, uid=None, gid=None, actor="",
                        login=None):
        try:
            out, _ = self._sftp_call(cid, actor, login, "sftp_chown",
                                     path, uid, gid)
        except ValueError as e:
            self._audit_rec("remotedesk.files_chown", path or "",
                            actor, str(e)[:200], False)
            raise ValueError(str(e))
        self._audit_rec("remotedesk.files_chown", "%s:%s=%s/%s" % (
            cid, path, uid, gid), actor)
        return out

    def api_files_zip(self, cid, path, actor="", login=None):
        import base64 as _b64
        try:
            out, _ = self._sftp_call(cid, actor, login, "sftp_zip_tree",
                                     path)
        except ValueError as e:
            self._audit_rec("remotedesk.files_zip", "%s:%s" % (
                cid, path), actor, str(e)[:200], False)
            raise ValueError(str(e))
        data = out.pop("data", b"")
        self._audit_rec("remotedesk.files_zip", "%s:%s (%d Б)" % (
            cid, path, len(data)), actor)
        return {"ok": True,
                "text": _b64.b64encode(data).decode(),
                "binary": True, **out}

    def health(self) -> dict:
        try:
            n = len(self.store.load())
        except Exception:  # noqa: BLE001
            n = -1
        try:
            from . import session as _sess
            sess = _sess.stats()
        except Exception:  # noqa: BLE001
            sess = {}
        return {"ok": True, "module": self.name, "connections": n,
                **sess}

"""remotedesk.rdp package."""

"""remotedesk.rdp.client — высокоуровневый RDP-клиент (фазы MS-RDPBCGR).

Фазы: negotiate (tpkt) -> [tls|credssp] -> MCS Connect -> channel join ->
Security Exchange (legacy rdp) -> Capabilities -> Bitmap loop.
Кадр — общий Frame из rfb.frame (переиспользуем PNG-патчи).
"""
import logging
import struct
import threading
import time

logger = logging.getLogger(__name__)


class RdpSession:
    def __init__(self, conn, info, frame, user_id, io_channel,
                 bpp=32, selected="rdp", clip_channel=None):
        self.conn = conn
        self.info = info
        self.frame = frame
        self.user_id = user_id
        self.io_channel = io_channel
        self.clip_channel = clip_channel  # id канала cliprdr или None
        self.bpp = bpp
        self.selected = selected
        self.palette = None  # палитра 8bit от сервера (список r,g,b)
        self._stop = threading.Event()
        self._lock = threading.Lock()  # чтение (loop)
        self._send_lock = threading.Lock()  # запись (ввод, отдельно:
        # сокет полнодуплексный, RC4-потоки направлений независимы —
        # ввод не должен ждать 15-секундного блокирующего read)
        self._thread = None
        self.last_update = time.time()
        self.updates = 0
        self._clip = None  # ClipChannel (логика CLIPRDR, без сокета)
        self._clip_reasm = bytearray()  # сборка фрагментов канала

    @property
    def width(self):
        return self.frame.width

    @property
    def height(self):
        return self.frame.height

    def start(self, fps=10):
        import threading as _th
        self._fps = max(1, min(15, int(fps or 10)))
        self._thread = _th.Thread(target=self._loop, daemon=True)
        self._thread.start()

    def stop(self):
        self._stop.set()
        try:
            self.conn.close()
        except Exception:  # noqa: BLE001
            pass

    def _send_input(self, payload):
        from . import mcs as _mcs
        from . import security as _sec
        crypto = getattr(self, "crypto", None)
        if crypto is not None:
            # Legacy: input PDU шифруются как share (флаги 0x0808,
            # БЕЗ SEC_LOGON_INFO — иначе сервер рвёт соединение).
            payload = _sec.wrap_share(crypto, payload)
        with self._send_lock:
            _mcs.write_mcs(self.conn,
                           _mcs.send_data(self.user_id, self.io_channel,
                                           payload))

    def _send_channel(self, clip_pdu):
        """Clipboard PDU кусками: length — длина данных без 8-байт шапки,
        FIRST/LAST по краям, каждый кусок шифруем как share."""
        import struct as _s
        from . import clip as _clip
        from . import mcs as _mcs
        from . import security as _sec
        from . import tpkt as _t
        if not self.clip_channel:
            raise _t.RdpError("нет канала cliprdr")
        crypto = getattr(self, "crypto", None)
        data = bytes(clip_pdu)
        total = len(data)
        frags = [data[i:i + _clip.CHAN_CHUNK]
                 for i in range(0, max(len(data), 1), _clip.CHAN_CHUNK)]
        with self._send_lock:
            for i, frag in enumerate(frags):
                flags = _clip.CHAN_FLAG_SHOW_PROTOCOL
                if i == 0:
                    flags |= _clip.CHAN_FLAG_FIRST
                if i == len(frags) - 1:
                    flags |= _clip.CHAN_FLAG_LAST
                chunk = _s.pack("<LL", total, flags) + frag
                if crypto is not None:
                    chunk = _sec.wrap_share(crypto, chunk)
                _mcs.write_mcs(self.conn,
                               _mcs.send_data(self.user_id,
                                              self.clip_channel, chunk))

    def _clip_handshake(self):
        """CLIPRDR CAPS + MONITOR_READY после финализации (как FreeRDP).

        Порядок: сначала читаем спонтанные CAPS+READY сервера (2-3с),
        потом шлём свои — иначе сервер не шлёт DATA_REQUEST.
        Ошибки не роняют сессию: без канала — paste_unicode.
        """
        from . import clip as _clip
        if not self.clip_channel:
            return
        try:
            if self._clip is None:
                self._clip = _clip.ClipChannel()
            self._clip_drain_spont()
            for pdu in self._clip.connect_pdus():
                self._send_channel(pdu)
        except Exception as e:  # noqa: BLE001
            logger.warning("remotedesk rdp clip handshake: %s",
                           str(e)[:100])

    def _clip_drain_spont(self, seconds=3.0):
        """Прочитать спонтанные CAPS+READY сервера до своих CAPS.

        Чужие PDU роутятся в штатные обработчики (_on_fast/_on_mcs),
        а не выкидываются: начальный пейнт приходит в этом же окне.
        Ошибки глотаем — drain не роняет сессию.
        """
        import time as _time
        from . import tpkt as _t
        c = self.conn
        if not hasattr(c, "read_pdu"):
            return  # тестовый коннект без чтения — сразу к своим CAPS
        try:
            old_timeout = c.sock.gettimeout()
        except Exception:  # noqa: BLE001
            old_timeout = None
        try:
            c.sock.settimeout(2)
        except Exception:  # noqa: BLE001
            pass
        t0 = _time.time()
        try:
            while _time.time() - t0 < seconds:
                try:
                    kind, pdu = c.read_pdu()
                except _t.RdpTimeout:
                    continue
                except _t.RdpError:
                    break
                if kind != "slow":
                    try:
                        self._on_fast(bytes(pdu))
                    except Exception:  # noqa: BLE001
                        pass
                    continue
                try:
                    # _on_mcs сам демультиплексирует: clip-канал -> _on_clip
                    # (спонтанные CAPS+READY как раньше), share -> кадр.
                    self._on_mcs(bytes(pdu))
                except Exception:  # noqa: BLE001
                    pass
        finally:
            try:
                if old_timeout is not None:
                    c.sock.settimeout(old_timeout)
            except Exception:  # noqa: BLE001
                pass

    def clip_offer(self, text):
        """Объявить исходящий текст (FORMAT_LIST). Вернуть True если
        ушло по каналу, False — нет канала (вызывающий падает в
        paste_unicode эмуляцию). Пустое/None — False без отправки."""
        if not text or not self.clip_channel:
            return False
        try:
            if self._clip is None:
                from . import clip as _clip
                self._clip = _clip.ClipChannel()
            for pdu in self._clip.offer(text):
                self._send_channel(pdu)
            return True
        except Exception as e:  # noqa: BLE001
            logger.warning("remotedesk rdp clip offer: %s", str(e)[:100])
            return False

    def clip_take_inbox(self):
        """Забрать входящие тексты сервера (для WS-sender)."""
        clip = getattr(self, "_clip", None)
        if clip is None:
            return []
        try:
            return clip.take_inbox()
        except Exception:  # noqa: BLE001
            return []

    def key(self, scancode, down=True, extended=False):
        from . import orders as _o
        ev = _o.slowpath_key(scancode, down, extended)
        self._send_input(_o.client_input(self._share_id(), ev))

    def combo(self, events):
        """Атомарное комбо: готовые input events (slowpath_key/mouse)
        одной INPUT PDU — сервер применяет строго по порядку.

        Одиночные key/pointer шлют по одному событию на PDU; пачка
        из 6 отдельных key через WS приходила рваной (rate-limit ел
        середину — CAD не работал). Комбо неделимо.
        """
        from . import orders as _o
        evs = list(events)
        if evs:
            self._send_input(_o.client_input(self._share_id(), *evs))

    def pointer(self, x, y, buttons=0, wheel=0):
        from . import orders as _o
        x = max(0, min(int(x), self.frame.width - 1))
        y = max(0, min(int(y), self.frame.height - 1))
        self._ptr_pos = (x, y)
        events = []
        if wheel:
            events.append(_o.slowpath_mouse(x, y, 0, wheel))
        else:
            # UI шлёт абсолютное состояние кнопок — диффим с прошлым
            # и шлём press/release переходами (RDP нужны PTRFLAGS_DOWN).
            prev = getattr(self, "_btn_state", 0)
            changed = (int(buttons) ^ prev) & 0x07
            for bit, mask in ((0, 1), (1, 2), (2, 4)):
                if changed & mask:
                    events.append(_o.slowpath_mouse(
                        x, y, mask, 0, down=bool(int(buttons) & mask)))
            self._btn_state = int(buttons) & 0x07
            if not events:
                events.append(_o.slowpath_mouse(x, y, 0, 0))  # MOVE
        for ev in events:
            self._send_input(_o.client_input(self._share_id(), ev))

    def paste_unicode(self, pairs):
        """Вставка текста: [[flags, codepoint]] unicode-событиями.

        Формат пар отличим от scancode-ключей [domCode, down]:
        первое поле — число-флаг, не строка кода.
        """
        from . import orders as _o
        evs = []
        for item in list(pairs or [])[:512]:
            try:
                flags, cp = (list(item) + [0, 0])[:2]
            except Exception:  # noqa: BLE001
                continue
            try:
                cp = int(cp)
                rel = bool(int(flags) & 0x8000)
            except (TypeError, ValueError):
                continue
            # Astral (>0xFFFF): surrogate pair двумя событиями.
            if cp > 0xFFFF:
                cp -= 0x10000
                evs.append(_o.slowpath_unicode(0xD800 + (cp >> 10),
                                              rel))
                evs.append(_o.slowpath_unicode(0xDC00 + (cp & 0x3FF),
                                              rel))
            else:
                evs.append(_o.slowpath_unicode(cp & 0xFFFF, rel))
        if evs:
            self._send_input(_o.client_input(self._share_id(), *evs))

    @property
    def pointer_pos(self):
        """Последняя позиция мыши (для курсор-оверлея фронта)."""
        return getattr(self, "_ptr_pos", (0, 0))

    def _share_id(self):
        return getattr(self, "_share_id_val", 0x03EA)

    def sync(self, toggle_flags=0):
        """SYNC event (RDP_INPUT_SYNCHRONIZE), шлём первым."""
        from . import orders as _o
        self._send_input(_o.client_input(
            self._share_id(), _o.slowpath_sync(toggle_flags)))

    def _loop(self):
        from . import tpkt as _t

        c = self.conn
        try:
            while not self._stop.is_set():
                try:
                    with self._lock:
                        kind, pdu = c.read_pdu()
                except _t.RdpTimeout:
                    # Сервер молчит между апдейтами — продолжаем ждать.
                    continue
                except _t.RdpError as e:
                    logger.info("remotedesk rdp: конец: %s", e)
                    break
                try:
                    if kind == "fast":
                        self._on_fast(pdu)
                    else:
                        self._on_mcs(pdu)
                except _t.RdpError as e:
                    logger.warning("remotedesk rdp pdu: %s", e)
        except Exception as e:  # noqa: BLE001
            logger.info("remotedesk rdp loop: %s: %s",
                        type(e).__name__, str(e)[:150])

    def _on_fast(self, pdu):
        """Fast-Path Output: updates -> rects во фрейм."""
        from . import fastpath as _fp
        if getattr(self, "_fp", None) is None:
            self._fp = _fp.FastPathParser(self.bpp)
        rects, palette = self._fp.feed(pdu)
        if palette is not None:
            self.palette = palette
        self._apply_rects(rects)

    def _apply_rects(self, rects):
        """Положить rects [(x,y,w,h,rgb888)] во фрейм, посчитать update.

        Клиппинг + проверка длины: иначе порча фреймбуфера
        со сдвигом строк. Битые rects дропаем молча.
        """
        import time as _time
        if not rects:
            return
        fw, fh = self.frame.width, self.frame.height
        for (x, y, w, h, rgb) in rects:
            try:
                x, y, w, h = int(x), int(y), int(w), int(h)
            except (TypeError, ValueError):
                continue
            if w <= 0 or h <= 0:
                continue
            if len(rgb) != w * h * 3:
                continue
            # Клиппинг по фрейму (сервер иногда шлёт за край).
            x0, y0 = max(0, x), max(0, y)
            x1, y1 = min(fw, x + w), min(fh, y + h)
            if x1 <= x0 or y1 <= y0:
                continue
            if x0 == x and y0 == y and x1 == x + w and y1 == y + h:
                loc = rgb
            else:
                rows = []
                for r in range(y0 - y, y1 - y):
                    rows.append(rgb[r * w * 3 + (x0 - x) * 3:
                                    r * w * 3 + (x1 - x) * 3])
                loc = b"".join(rows)
            cw, chh = x1 - x0, y1 - y0
            for j in range(chh):
                off = ((y0 + j) * fw + x0) * 3
                row = loc[j * cw * 3:(j + 1) * cw * 3]
                self.frame.fb[off:off + cw * 3] = row[:cw * 3]
            self.frame.apply([(x0, y0, cw, chh)])
        self.last_update = _time.time()
        self.updates += 1

    def _on_mcs(self, mcs):
        from . import orders as _o
        from . import tpkt as _t
        # MCS SendData: клиентский Request 0x64, серверный Indication 0x68.
        if not mcs or mcs[0] not in (0x64, 0x68):
            return
        channel, body = _strip_send_data_ex(mcs)
        if body is None:
            return
        crypto = getattr(self, "crypto", None)
        if crypto is not None:
            body = _decrypt_share(crypto, body)
        if body is None:
            return
        # Виртуальный канал cliprdr — не share PDU: разбираем отдельно
        # (иначе share-парсер молча выкинет как pdu_type != 7).
        if (self.clip_channel is not None
                and channel == self.clip_channel):
            try:
                self._on_clip(body)
            except _t.RdpError as e:
                logger.warning("remotedesk rdp clip: %s", e)
            return
        if len(body) < 18:
            return
        # Структурный разбор (без поиска маркеров): shareControlHeader(6:
        # totalLength + pduType + pduSource) — ждём Data PDU (pduType 7),
        # затем sharedDataHeader(12) с pduType2. Update PDU: pduType2 = 2.
        import struct as _s
        pdu_type = _s.unpack("<H", body[2:4])[0] & 0x0F
        if pdu_type != 7:
            return
        pdu_type2 = body[14]
        # type 9 (PALETTE) — обновляем палитру 8bit, кадра нет.
        # Проверка ДО фильтра pdu_type2==2 (иначе недостижимо).
        if pdu_type2 == 9:
            try:
                self.palette = _o.parse_palette(body[18:])
            except _t.RdpError:
                pass
            return
        if pdu_type2 != 2:
            return
        # body[18:] = TS_UPDATE_PDU (updateType 1=bitmap, 6=orders).
        # uncompressedLength check мягкий (сервер иногда врёт на +-байты).
        try:
            rects = _o.parse_update_pdu(body[18:], self.bpp,
                                        self.palette)
        except _t.RdpError:
            return
        self._apply_rects(rects)

    def _on_clip(self, data):
        """Канальные данные cliprdr: сборка фрагментов -> ClipChannel.

        CHANNEL_PDU_HEADER: length(4, вся длина С шапкой) + flags(4).
        FIRST сбрасывает сборку, LAST отдаёт целое на разбор (в целом
        может быть несколько PDU подряд — разбираем циклом).
        """
        import struct as _s
        from . import clip as _clip
        from . import tpkt as _t
        if len(data) < 8:
            raise _t.RdpError("короткий channel pdu")
        total, flags = _s.unpack("<LL", data[:8])
        # total = длина данных БЕЗ шапки (как шлём сами); пустой кусок
        # (total=0) допустим, кап — как у inbox.
        if total > 4 * 1048576 + 8:
            raise _t.RdpError("плохая длина channel pdu: %d" % total)
        frag = data[8:]
        if flags & _clip.CHAN_FLAG_FIRST:
            self._clip_reasm = bytearray()
        self._clip_reasm += frag
        if len(self._clip_reasm) > 4 * 1048576 + 8:
            self._clip_reasm = bytearray()
            raise _t.RdpError("переполнение сборки clip")
        if not (flags & _clip.CHAN_FLAG_LAST):
            return
        blob = bytes(self._clip_reasm)
        self._clip_reasm = bytearray()
        if self._clip is None:
            self._clip = _clip.ClipChannel()
        pos = 0
        while pos + 8 <= len(blob):
            _t2, _f2, ln = _s.unpack("<HHL", blob[pos:pos + 8])
            if ln > _clip.IN_TEXT_MAX + 8 or pos + 8 + ln > len(blob):
                break
            pdu = blob[pos:pos + 8 + ln]
            pos += 8 + ln
            try:
                for resp in self._clip.on_pdu(pdu):
                    self._send_channel(resp)
            except _t.RdpError as e:
                logger.warning("remotedesk rdp clip pdu: %s", e)
                break


def connect(host, port=3389, username="", password="", domain="",
            mode="any", width=1024, height=768, bpp=32, timeout=10,
            cert_pin="", ignore_cert=False, fps=10, compat="modern"):
    """Полное подключение. Вернуть RdpSession (поток уже запущен).

    compat: "modern" (RDP 10.7 Core, 40|128bit) или "legacy" (RDP 5.0,
    только 40bit, 8bit — для Win XP/2003 и старых терминалов).
    """
    from . import mcs as _mcs
    from . import security as _sec
    from . import tpkt as _t
    from ..rfb import frame as _frame

    if compat == "legacy":
        # XP: только legacy rdp security, без NLA/TLS попыток.
        mode = "rdp"
        bpp = 8
    conn, selected = _t.negotiate(host, port, mode, username, timeout)
    logger.info("remotedesk rdp: negotiate %s:%s -> %s (mode=%s)",
                host, port, selected, mode)
    try:
        if selected in ("tls", "nla", "nla-ext"):
            der = conn.start_tls(host)
            logger.info("remotedesk rdp: TLS ok (%s)",
                        selected)
            if not ignore_cert and not cert_pin:
                # TOFU: fingerprint показать вызывающему через исключение.
                fp = "sha256:" + _cert_sha256(der)
                conn.close()
                from ..rfb import proto as _rp  # reuse NeedPin-идеи
                raise CertPinNeeded(fp)
            if cert_pin:
                fp = "sha256:" + _cert_sha256(der)
                if cert_pin != fp:
                    conn.close()
                    raise _t.RdpError(
                        "сертификат НЕ совпал с пином "
                        "(сервер дал %s)" % fp)
            if selected in ("nla", "nla-ext"):
                try:
                    _sec.credssp_handshake(
                        conn, username, password, domain,
                        early_auth=(selected == "nla-ext"))
                except _t.RdpError as e:
                    raise _t.RdpError("CredSSP (%s): %s" % (selected, e))
                except ValueError as e:
                    raise _t.RdpError("CredSSP (%s): %s" % (selected, e))
                logger.info("remotedesk rdp: CredSSP ok (%s)", selected)
        # MCS Connect Initial.
        ci = _mcs.build_connect_initial(username, width, height, bpp,
                                        compat, selected)
        with _mcs_lock(conn):
            _mcs.write_mcs(conn, ci)
            resp = conn.read_mcs()
        srv = _mcs.parse_connect_response(resp)
        # Erect + Attach.
        with _mcs_lock(conn):
            _mcs.write_mcs(conn, _mcs.erect_domain())
            _mcs.write_mcs(conn, _mcs.attach_user())
            confirm = conn.read_mcs()
        user_id = _parse_attach_confirm(confirm)
        io_ch = int(srv.get("io_channel") or 1003)
        # Канал cliprdr — первый из заявленных в NET (порядок SC_NET =
        # порядок дефов; channels[0] наш, если сервер его принял).
        clip_ch = None
        srv_channels = [int(c) for c in (srv.get("channels") or [])]
        if srv_channels:
            clip_ch = srv_channels[0]
        # Join: user + io + clip (если есть).
        with _mcs_lock(conn):
            for ch in [user_id, io_ch] + ([clip_ch] if clip_ch else []):
                _mcs.write_mcs(conn, _mcs.channel_join(user_id, ch))
                conn.read_mcs()  # join confirm (проверка мягкая)
        # Legacy rdp security: Security Exchange + Client Info.
        # NLA/TLS: Client Info открытым текстом до Demand Active.
        crypto = None
        if selected == "rdp":
            crypto = _legacy_security_exchange(
                conn, user_id, io_ch, srv, username, password, domain)
        else:
            _send_client_info_plain(conn, user_id, io_ch, username,
                                    password, domain)
        # Confirm Active (capabilities минимум) -> Demand Active.
        # Возвращает shareId сервера — нужен для input PDU.
        share_id = _confirm_active(conn, user_id, io_ch, crypto, width,
                                   height)
        fr = _frame.Frame(width, height)
        sess = RdpSession(conn, {"width": fr.width, "height": fr.height,
                                 "selected": selected},
                          fr, user_id, io_ch, bpp, selected, clip_ch)
        sess.crypto = crypto
        sess._share_id_val = share_id
        try:
            sess._clip_handshake()  # CLIPRDR CAPS + READY (не роняет)
        except Exception:  # noqa: BLE001
            pass
        sess.start(fps)
        return sess
    except Exception:
        try:
            conn.close()
        except Exception:  # noqa: BLE001
            pass
        raise


class CertPinNeeded(Exception):
    """Сертификат сервера неизвестен — нужен TOFU-пин (как NeedPin в SSH)."""

    def __init__(self, fp):
        super().__init__("нужен cert пин: %s" % fp)
        self.fp = fp


def _cert_sha256(der):
    import hashlib as _h
    import base64 as _b
    if not der:
        return _b.b64encode(b"\x00" * 32).decode().rstrip("=")
    return _b.b64encode(_h.sha256(der).digest()).decode().rstrip("=")


def _mcs_lock(conn):
    return conn._lock if hasattr(conn, "_lock") else _DummyLock()


class _DummyLock:
    def __enter__(self):
        return self

    def __exit__(self, *a):
        return False


def _parse_attach_confirm(payload):
    """MCS Attach User Confirm -> user channel id (PER, см. mcs.py)."""
    from . import mcs as _mcs
    return _mcs.parse_attach_confirm(payload)


def _legacy_security_exchange(conn, user_id, io_ch, srv, username,
                              password, domain=""):
    """Legacy: Security Exchange (RSA) -> Client Info (RC4).

    Возвращает RdpCrypto (ключи сессии) для шифра дальнейшего трафика.
    Сервер отвечает License Error (игнорим) и Demand Active.
    """
    import os as _os
    from . import mcs as _mcs
    from . import security as _sec
    from . import tpkt as _t

    cert = srv.get("cert") or b""
    if not cert or not (srv.get("server_random") or b""):
        raise _t.RdpError("сервер не прислал сертификат (legacy rdp)")
    rnd = _os.urandom(32)
    exch = _sec.build_security_exchange(cert, rnd)
    _mcs.write_mcs(conn, _mcs.send_data(user_id, io_ch, exch))
    crypto = _sec.RdpCrypto(rnd, srv["server_random"],
                            int(srv.get("enc_method") or 2))
    info = _sec.build_client_info(username, password, domain)
    _mcs.write_mcs(conn, _mcs.send_data(
        user_id, io_ch, _sec.wrap_encrypted(crypto, info)))
    return crypto


def _send_client_info_plain(conn, user_id, io_ch, username, password,
                            domain=""):
    """Client Info без шифра (NLA/TLS): SEC_INFO_PKT + TS_INFO_PACKET.

    Канал уже защищён TLS (CredSSP внутри), шифровать нечего — шлём
    открытые флаги 0x40 + build_client_info. crypto для сессии нет.
    """
    import struct as _s
    from . import mcs as _mcs
    from . import security as _sec
    info = _sec.build_client_info(username, password, domain)
    _mcs.write_mcs(conn, _mcs.send_data(
        user_id, io_ch, _s.pack("<L", 0x40) + info))


def _confirm_active(conn, user_id, io_ch, crypto=None, width=1024,
                    height=768):
    """Confirm Active PDU (capabilities минимум) + ждать Demand Active.

    Confirm Active: shareId(4) + originator(2) + lenSourceDesc(2) +
    lenCombinedCaps(2) + sourceDesc + caps. Шлём пустые General(8B) +
    Bitmap(28B) + Order(88B, минимум: orderSupport[32]=0 кроме DSTBLT?) —
    сервер обязан откатиться на Bitmap Updates. Перед Demand Active сервер
    шлёт SERVER_LICENSE_REQUEST — отвечаем CLIENT_LICENSE_INFO один раз
    (иначе сервер молчит). Затем читаем до Demand Active (pduType2=1)
    и отвечаем Confirm Active.
    """
    import struct as _s
    import time as _time
    from . import mcs as _mcs
    from . import security as _sec
    from . import tpkt as _t

    conn.sock.settimeout(15)
    t0 = _time.time()
    demand = None
    lic_replied = False
    while _time.time() - t0 < 15:
        try:
            raw = conn.read_mcs()
        except _t.RdpError:
            break
        # MCS Send Data Indication: снимаем, ищем share PDU.
        body = _strip_send_data(raw)
        if body is None:
            continue
        # License PDU сервера (SERVER_LICENSE_REQUEST / LICENSE_ERROR):
        # flags u32 с битом 0x80, без ENCRYPT. На запрос отвечаем один раз
        # CLIENT_LICENSE_INFO — иначе сервер молчит (нет Demand Active).
        if len(body) >= 4:
            lic_flags = _s.unpack("<L", body[:4])[0]
            if (lic_flags & 0x0080) and not (lic_flags & 0x0008):
                if not lic_replied and crypto is not None:
                    lic = _sec.wrap_license(
                        crypto, _sec.build_license_info())
                    _mcs.write_mcs(
                        conn, _mcs.send_data(user_id, io_ch, lic))
                    lic_replied = True
                continue
        if crypto is not None:
            body = _decrypt_share(crypto, body)
        if body is None:
            continue
        # Demand Active PDU (MS-RDPBCGR 2.2.8.1.1.1.2): после SEC-заголовка
        # идёт shareControlHeader(6: totalLength + pduType + pduSource) +
        # shareId(4) + caps. Demand Active: pduType = 0x11 (offset 2).
        # (pduType2 есть только у Data PDU — здесь его нет.)
        if len(body) >= 4 and body[2] == 0x11:
            demand = body
            break
        # License Error PDU (pduType=0xFF?) — пропускаем.
    if demand is None:
        raise _t.RdpError("нет Demand Active от сервера")
    share_id = _s.unpack("<L", demand[6:10])[0]  # shareId сервера
    caps = _build_caps_min(width, height)
    # TS_CONFIRM_ACTIVE_PDU (MS-RDPBCGR 2.2.1.13.2.2): shareControlHeader(6:
    # totalLength + pduType=ConfirmActive(3)+version + pduSource) +
    # shareId(4) + originator(2) + lenSourceDesc(2) + lenCombinedCaps(2) +
    # sourceDesc + caps.
    # pduSource = io-канал (не user_id), originator = 0x03EA,
    # lenCombinedCaps — длина caps без sourceDesc.
    sdesc = _confirm_source_desc()
    confirm_body = (_s.pack("<LHHH", share_id, 0x03EA, len(sdesc),
                            len(caps))
                    + sdesc + caps)
    confirm = (_s.pack("<HHH", 6 + len(confirm_body), 0x10 | 3, io_ch)
               + confirm_body)
    full = confirm
    if crypto is not None:
        full = _sec.wrap_share(crypto, full)
    _mcs.write_mcs(conn, _mcs.send_data(user_id, io_ch, full))
    # Synchronize -> Control(4) -> Control(1) -> Font List, шифруем share.
    from . import orders as _o
    for pdu in (_o.client_synchronize(share_id, user_id),
                _o.client_control(share_id, 4),
                _o.client_control(share_id, 1),
                _o.client_font_list(share_id)):
        out = _sec.wrap_share(crypto, pdu) if crypto is not None else pdu
        _mcs.write_mcs(conn, _mcs.send_data(user_id, io_ch, out))
    return share_id


def _strip_send_data(raw):
    """Снять MCS SendData Indication. Вернуть userData или None."""
    _, body = _strip_send_data_ex(raw)
    return body


def _strip_send_data_ex(raw):
    """Снять MCS SendData Indication. Вернуть (channel_id, userData).

    channel_id нужен демультиплексору виртуальных каналов (cliprdr):
    share идёт с io-канала, clip — со своего. Старый _strip_send_data —
    обёртка, сигнатуру не меняем (зовёт _confirm_active).
    """
    import struct as _s
    # Клиент шлёт Request (0x64), сервер отвечает Indication (0x68).
    if not raw or raw[0] not in (0x64, 0x68):
        return None, None
    # 64 | initiator(2) | channel(2) | 70 | per_len | data.
    if len(raw) < 7:
        return None, None
    channel = _s.unpack(">H", raw[3:5])[0]
    p = 1 + 2 + 2 + 1
    b0 = raw[p]
    if b0 < 128:
        return channel, raw[p + 1:]
    return channel, raw[p + 2:]


def _decrypt_share(crypto, body):
    """Снять SEC_ENCRYPT с share PDU. Вернуть открытый body или None."""
    import struct as _s
    if len(body) < 12:
        return None
    flags = _s.unpack("<L", body[:4])[0]
    if not (flags & 0x0008):
        return body  # не шифровано
    sig, enc = body[4:12], body[12:]
    dec = crypto.decrypt(enc)
    return dec


def _build_caps_min(width=1024, height=768):
    """Combined Capabilities: General + Bitmap + Order + BitmapCacheHostSupport.

    Минимальный рабочий набор {General, Bitmap, Order, HostSupport}:
    без сета 0x0D или с нулевым Order сервер рвёт финализацию.
    Width/height сессии подставляются в Bitmap, остальное — константы.
    """
    import struct as _s
    # General (type 1, len 24): osMajor=4, osMinor=7, ver=0x0200,
    # extraFlags=0x0415, refreshRect=1, suppressOutput=1.
    gen = bytes.fromhex("01001800") + bytes.fromhex(
        "0400070000020000000015040000000000000101")
    # Bitmap (type 2, len 28): 32bpp, resize=1, compression=1,
    # drawingFlags=0x0E, multiRect=1; width/height — свои.
    bmp = (_s.pack("<HH", 2, 28) + _s.pack(
        "<HHHH", 32, 1, 1, 1) + _s.pack("<HH", width, height)
        + bytes.fromhex("000001000100000e01000000"))
    # Order (type 3, len 88): эталон целиком (orderSupport с флагами
    # DSTBLT/PATBLT/SCRBLT/MEMBLT/LINE/MEM3BLT/TRIANGLES/...).
    ordcaps = bytes.fromhex(
        "03005800000000000000000000000000000000000000000001001400"
        "000001000000aa000101010000000000010000000000000000000100"
        "00000100000000000000000000000400000000000084030000000000"
        "e9fd0000")
    # BitmapCacheHostSupport (type 0x0D, len 88): эталон целиком
    # (body 84: cacheFlags=0x13D, bmjhCacheHandle=0x409, bmjhCacheCount=4,
    #  остальное нули).
    hostsup = bytes.fromhex(
        "0d0058003d0100000904000004000000000000000c00000000000000"
        "00000000000000000000000000000000000000000000000000000000"
        "00000000000000000000000000000000000000000000000000000000"
        "00000000")
    caps = gen + bmp + ordcaps + hostsup
    return _s.pack("<HH", 4, len(caps) + 4) + caps


def _confirm_source_desc():
    """Source descriptor Confirm Active."""
    return b"FREERDP\x00"