"""嵌入式 SFTP 服务，与 HTTP API 共用 uploads/ 目录；同时承载反向隧道的 SSH 转发。

启动方式：
    python -m app.services.sftp_server
"""

from __future__ import annotations

import asyncio
import logging
import os
import sys
from pathlib import Path

import asyncssh
import bcrypt

from ..config import PROJECT_ROOT, get_settings

logger = logging.getLogger("sftp")


class ZikaiSFTPServer(asyncssh.SFTPServer):
    """会话期把客户端 chroot 到 upload_root。"""

    def __init__(self, chan, *, upload_root: Path) -> None:
        super().__init__(chan, chroot=str(upload_root).encode())
        try:
            self._username = chan.get_extra_info("username") or "unknown"
        except Exception as exc:  # pragma: no cover
            self._username = "unknown"
            logger.debug("读取 SFTP 会话用户名失败: %s", exc)
        logger.info("SFTP 会话开始 user=%s chroot=%s", self._username, upload_root)

    def exit(self) -> None:
        logger.info("SFTP 会话结束 user=%s", self._username)


def _tunnel_dao():
    """惰性构造一个 TunnelSessionDAO（避免 import 时的副作用）。"""
    from ..database import get_session_local
    from ..dao.tunnel_session_dao import TunnelSessionDAO
    return TunnelSessionDAO(get_session_local()())


def _close_tunnel_dao(dao) -> None:
    try:
        dao.db.close()
    except Exception as exc:  # pragma: no cover
        logger.debug("关闭隧道 DAO 会话失败: %s", exc)


class ZikaiSSHServer(asyncssh.SSHServer):
    """支持密码（bcrypt）与公钥两种鉴权；允许反向隧道 user 的 remote forwarding。"""

    def __init__(self, settings, authorized_keys: asyncssh.SSHAuthorizedKeys | None) -> None:
        self._settings = settings
        self._authorized_keys = authorized_keys
        self._conn: asyncssh.SSHServerConnection | None = None
        self._username: str | None = None  # 认证成功后填入
        self._peer_ip: str = ""

    def connection_made(self, conn: asyncssh.SSHServerConnection) -> None:  # type: ignore[override]
        self._conn = conn
        peer = conn.get_extra_info("peername")
        self._peer_ip = peer[0] if isinstance(peer, tuple) and peer else ""

    def begin_auth(self, username: str) -> bool:
        # 始终要求鉴权；用户不存在时所有方法会失败 → 干净的 permission denied
        return True

    # 密码

    def password_auth_supported(self) -> bool:
        return True

    def validate_password(self, username: str, password: str) -> bool:
        user = self._find_sftp_user(username) or self._find_tunnel_user(username)
        if user is None or not getattr(user, "password_hash", "") \
                or user.password_hash == "CHANGE_ME_BCRYPT_HASH":
            return False
        try:
            ok = bcrypt.checkpw(password.encode(), user.password_hash.encode())
        except (ValueError, TypeError):
            ok = False
        if ok:
            self._username = username
        logger.info("SSH 密码鉴权 %s user=%s", "OK" if ok else "FAIL", username)
        return ok

    # 公钥

    def public_key_auth_supported(self) -> bool:
        return self._authorized_keys is not None

    def validate_public_key(self, username: str, key: asyncssh.SSHKey) -> bool:
        if self._authorized_keys is None:
            return False
        if self._find_sftp_user(username) is None and self._find_tunnel_user(username) is None:
            return False
        addr = self._peer_ip
        try:
            # asyncssh 命中返回 dict（可能为空），未命中返回 None
            result = self._authorized_keys.validate(key, client_host=addr, client_addr=addr)
        except Exception as exc:  # pragma: no cover
            logger.warning("公钥校验异常 user=%s: %s", username, exc)
            result = None
        ok = result is not None
        if ok:
            self._username = username
        logger.info("SSH 公钥鉴权 %s user=%s", "OK" if ok else "FAIL", username)
        return ok

    # 反向隧道：remote port-forward 请求

    def server_requested(self, listen_host: str, listen_port: int) -> bool:
        """客户端请求在本地（server 侧）绑端口做反向转发时回调。

        仅允许已认证的 tunnel user 绑定其配置中预定的 tunnel_port；记一条 active
        隧道会话到 DB。其他情况拒绝。
        """
        if not self._settings.tunnel.enabled:
            logger.warning("拒绝 remote forward：tunnel 未启用 user=%s", self._username)
            return False
        if not self._username:
            logger.warning("拒绝 remote forward：未认证")
            return False
        tunnel_user = self._find_tunnel_user(self._username)
        if tunnel_user is None:
            logger.warning("拒绝 remote forward：%s 不是 tunnel user", self._username)
            return False
        if listen_port != tunnel_user.tunnel_port:
            logger.warning(
                "拒绝 remote forward：user=%s 端口 %d 不等于配置 %d",
                self._username, listen_port, tunnel_user.tunnel_port,
            )
            return False

        dao = _tunnel_dao()
        try:
            from ..services.tunnel_service import TunnelService
            TunnelService(dao).register(
                user_name=self._username,
                user_ip=self._peer_ip,
                tunnel_port=listen_port,
                local_port=tunnel_user.local_port,
            )
        except Exception as exc:  # pragma: no cover
            _close_tunnel_dao(dao)
            logger.error("登记隧道会话失败：%s", exc)
            return False
        _close_tunnel_dao(dao)
        logger.info(
            "允许 remote forward user=%s listen=%s:%d -> local_port=%d",
            self._username, listen_host, listen_port, tunnel_user.local_port,
        )
        return True

    def connection_lost(self, exc: Exception | None) -> None:  # type: ignore[override]
        """SSH 连接断开时清理该 user 的活跃隧道会话。"""
        if self._username and self._settings.tunnel.enabled:
            dao = _tunnel_dao()
            try:
                from ..services.tunnel_service import TunnelService
                TunnelService(dao).close(self._username)
            except Exception as e:  # pragma: no cover
                logger.warning("清理隧道会话失败 user=%s: %s", self._username, e)
            _close_tunnel_dao(dao)
        if exc:
            logger.info("SSH 连接异常断开 user=%s: %s", self._username, exc)
        else:
            logger.info("SSH 连接关闭 user=%s", self._username)

    # 内部

    def _find_sftp_user(self, username: str):
        return next((u for u in self._settings.sftp.users if u.username == username), None)

    def _find_tunnel_user(self, username: str):
        return self._settings.tunnel.find_user(username)


def _load_authorized_keys(path: Path) -> asyncssh.SSHAuthorizedKeys | None:
    if not path.exists() or path.stat().st_size == 0:
        return None
    try:
        return asyncssh.read_authorized_keys(str(path))
    except Exception as exc:  # pragma: no cover
        logger.warning("解析 %s 失败：%s", path, exc)
        return None


def _ensure_host_key(path: Path) -> None:
    if path.exists():
        return
    path.parent.mkdir(parents=True, exist_ok=True)
    logger.info("生成 ed25519 SFTP 主机密钥：%s", path)
    key = asyncssh.generate_private_key("ssh-ed25519")
    key.write_private_key(str(path))
    key.write_public_key(str(path) + ".pub")
    os.chmod(path, 0o600)


async def _run() -> None:
    settings = get_settings()
    if not settings.sftp.enabled:
        logger.info("SFTP 在配置中已禁用，进程退出。")
        return

    upload_root = settings.resolved_upload_dir()
    upload_root.mkdir(parents=True, exist_ok=True)
    # SFTP 客户端的暂存目录（chroot 内）。
    (upload_root / "incoming").mkdir(parents=True, exist_ok=True)

    host_key_path = (PROJECT_ROOT / settings.sftp.host_key_path).resolve()
    _ensure_host_key(host_key_path)

    authorized_keys = _load_authorized_keys(
        (PROJECT_ROOT / settings.sftp.authorized_keys_path).resolve()
    )

    def sftp_factory(chan) -> ZikaiSFTPServer:
        return ZikaiSFTPServer(chan, upload_root=upload_root)

    logger.info(
        "启动 SFTP 服务 %s:%d (root=%s)",
        settings.sftp.host, settings.sftp.port, upload_root,
    )
    await asyncssh.create_server(
        lambda: ZikaiSSHServer(settings, authorized_keys),
        settings.sftp.host,
        settings.sftp.port,
        server_host_keys=[str(host_key_path)],
        sftp_factory=sftp_factory,
        allow_scp=False,
    )
    await asyncio.Event().wait()


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
        stream=sys.stdout,
    )
    try:
        asyncio.run(_run())
    except (KeyboardInterrupt, SystemExit):
        logger.info("SFTP 服务退出。")


if __name__ == "__main__":
    main()