"""Read-only checks. No shell, privilege escalation or system mutations."""

from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import ipaddress
import json
import os
import platform
import re
from pathlib import Path
import shutil
import subprocess
import urllib.error
import urllib.request

from .i18n import translator
from . import __version__


@dataclass
class Result:
    id: str
    status: str
    title: str
    detail: str
    evidence: str = ""
    advice: str = ""


@dataclass
class Command:
    code: int | None
    output: str = ""
    reason: str = ""


# Only these fixed command arguments may reach subprocess.
ALLOWED = {
    ("ip", "-j", "address", "show"),
    ("ip", "-j", "-4", "route", "show", "default"),
    ("ip", "-j", "-6", "route", "show", "default"),
    ("ss", "-H", "-lntu"),
    ("sshd", "-T"),
    ("ufw", "status"),
    ("nft", "list", "ruleset"),
    ("iptables", "-S"),
    ("systemctl", "is-active", "sing-box.service"),
    ("systemctl", "is-enabled", "sing-box.service"),
}
SYSTEM_PATH = "/usr/sbin:/usr/bin:/sbin:/bin"
OUTPUT_LIMIT = 131072


class Host:
    def __init__(self, lang="zh"):
        self.lang = lang

    def system(self):
        return platform.system(), platform.machine()

    def read(self, path):
        try:
            return Path(path).read_text(encoding="utf-8")
        except (OSError, UnicodeError):
            return None

    def disk_free(self):
        return shutil.disk_usage("/").free

    def disk_total(self):
        return shutil.disk_usage("/").total

    def cpu_count(self):
        return os.cpu_count()

    def kernel(self):
        return platform.release()

    def ssh_connection(self):
        return os.environ.get("SSH_CONNECTION", "")

    def detect_public_ip(self):
        from .worker import enabled
        if enabled():
            from .net import request_json
            data,state=request_json('https://api.ipify.org?format=json')
            try:
                address=ipaddress.ip_address(data.get('ip')) if state=='ok' else None
                return str(address) if address and address.is_global else None
            except (ValueError,TypeError,AttributeError):return None
        # Explicit opt-in; no environment proxies or redirects.
        class NoRedirect(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, req, fp, code, msg, headers, newurl):
                return None
        try:
            opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())
            with opener.open('https://api64.ipify.org', timeout=5) as response:
                payload = response.read(129)
            if len(payload) > 128:
                return None
            address = ipaddress.ip_address(payload.decode('ascii').strip())
            return str(address) if address.is_global else None
        except (urllib.error.URLError, OSError, UnicodeError, ValueError):
            return None

    def run(self, *args):
        t = translator(self.lang)
        if args not in ALLOWED:
            raise ValueError("Command is not allowlisted")
        # Ignore the caller's PATH, especially when a user chooses to run as root.
        executable = shutil.which(args[0], path=SYSTEM_PATH)
        if executable is None and args[0] in {"sshd", "ufw", "nft", "iptables"}:
            for directory in ("/usr/sbin", "/sbin"):
                candidate = Path(directory) / args[0]
                if candidate.is_file():
                    executable = str(candidate)
                    break
        if executable is None:
            return Command(None, reason=t("命令不可用"))
        try:
            proc = subprocess.run(
                [executable, *args[1:]], capture_output=True, text=True,
                encoding="utf-8", errors="replace", timeout=8, check=False,
                env={"PATH": SYSTEM_PATH, "LC_ALL": "C"},
            )
            # Never return stderr: it can contain unrelated host details.
            if len(proc.stdout) > OUTPUT_LIMIT:
                return Command(None, reason=t("输出超过读取上限，不能作为完整检查结果"))
            return Command(proc.returncode, proc.stdout,
                           t("读取命令失败，可能是权限、配置或服务状态问题") if proc.returncode else "")
        except subprocess.TimeoutExpired:
            return Command(None, reason=t("读取超时（8 秒）"))
        except OSError:
            return Command(None, reason=t("无法执行读取命令"))


def parse_os(text):
    values = {}
    for line in (text or "").splitlines():
        if "=" in line and not line.startswith("#"):
            key, value = line.split("=", 1)
            values[key] = value.strip().strip('"').strip("'")
    return values


def port_rows(text):
    """Parse ss -H -lntu without relying on process ownership visibility."""
    rows = []
    malformed = False
    for line in text.splitlines():
        fields = line.split()
        if len(fields) < 6 or fields[0] not in {"tcp", "udp"}:
            malformed = True
            continue
        endpoint = fields[4]
        try:
            port = int(endpoint.rsplit(":", 1)[1])
            if not 1 <= port <= 65535:
                raise ValueError("Invalid port")
        except (ValueError, IndexError):
            malformed = True
            continue
        rows.append((fields[0], endpoint, port))
    return rows, malformed


def ubuntu_version(value):
    """Parse VERSION_ID numerically; never infer Ubuntu from ID_LIKE."""
    if not re.fullmatch(r"[0-9]{2}\.(04|10)", value or ""):
        return None
    return tuple(int(part) for part in value.split("."))


def route_check(host, t):
    evidence, valid, failed = [], 0, False
    for family in ("-4", "-6"):
        result = host.run("ip", "-j", family, "route", "show", "default")
        try:
            if result.code != 0:
                raise ValueError
            routes = json.loads(result.output)
            if not isinstance(routes, list):
                raise ValueError
            for route in routes:
                if not isinstance(route, dict) or route.get("dst") not in ("default", "0.0.0.0/0", "::/0"):
                    raise ValueError
                if route.get("type", "unicast") != "unicast":
                    continue
                dev = route.get("dev")
                gateway = route.get("gateway")
                if not isinstance(dev, str) or not dev or (gateway is not None and not isinstance(gateway, str)):
                    raise ValueError
                if gateway is not None:
                    address = ipaddress.ip_address(gateway)
                    if address.version != (4 if family == "-4" else 6):
                        raise ValueError
                evidence.append(f"IPv{family[-1]}: {dev}" + (f" via {gateway}" if gateway else ""))
                valid += 1
        except (ValueError, TypeError, KeyError):
            failed = True
            evidence.append(f"IPv{family[-1]}: " + t("路由读取失败或格式无法解析"))
    return Result("default_route", "pass" if valid and not failed else "unknown", t("默认路由"),
                  t("已读取默认路由") if valid and not failed else
                  t("默认路由信息不完整或未找到，无法确认网络出口"), "\n".join(evidence),
                  t("仅检查当前网络命名空间的默认路由，不覆盖策略路由。"))


def dns_check(host, t):
    content = host.read("/etc/resolv.conf")
    if content is None:
        return Result("dns_config", "unknown", t("DNS 配置"), t("无法读取 DNS 配置"))
    addresses, invalid = [], False
    for line in content.splitlines():
        line = line.split("#", 1)[0].split(";", 1)[0].strip()
        fields = line.split()
        if fields and fields[0] == "nameserver":
            if len(fields) != 2:
                invalid = True
                continue
            try:
                address = ipaddress.ip_address(fields[1])
                if address.is_unspecified or address.is_multicast:
                    raise ValueError
                addresses.append(address)
            except ValueError:
                invalid = True
    evidence = "\n".join(str(addr) for addr in addresses)
    stub = any(addr.is_loopback for addr in addresses)
    status = "unknown" if invalid or not addresses or stub else "pass"
    detail = (t("DNS 配置不完整或没有有效的服务器地址") if invalid or not addresses else
              t("DNS 指向本机解析器") if stub else
              t("已读取 DNS 服务器配置"))
    return Result("dns_config", status, t("DNS 配置"), detail, evidence,
                  t("仅检查 resolv.conf，不主动发送 DNS 请求；实际解析路径可能不同。"))


def ssh_port_check(host, ssh, port, t):
    configured, malformed = [], False
    if ssh.code == 0:
        for line in ssh.output.splitlines():
            fields = line.split()
            if fields and fields[0] == "port":
                try:
                    if len(fields) != 2 or not 1 <= int(fields[1]) <= 65535:
                        raise ValueError
                    configured.append(int(fields[1]))
                except ValueError:
                    malformed = True
    try:
        session = host.ssh_connection()
    except AttributeError:
        session = ""
    session_port = None
    if session:
        try:
            fields = session.split()
            if len(fields) != 4:
                raise ValueError
            ipaddress.ip_address(fields[0]); ipaddress.ip_address(fields[2])
            if not 1 <= int(fields[1]) <= 65535 or not 1 <= int(fields[3]) <= 65535:
                raise ValueError
            session_port = int(fields[3])
        except ValueError:
            session_port = None
    evidence = []
    if configured:
        evidence.append(t("sshd 默认配置端口：{ports}", ports=", ".join(map(str, sorted(set(configured))))))
    if session_port is not None:
        evidence.append(t("当前 SSH 会话本地端口：{port}", port=session_port))
    matched = port in configured or port == session_port
    detail = t("目标端口与 SSH 已知端口冲突；不要用于代理") if matched else t("未能排除与运行中 SSH 端口冲突")
    if malformed:
        evidence.append(t("SSH 端口输出不完整或无效"))
    return Result("ssh_port_safety", "warn" if matched else "unknown", t("SSH 端口保护"), detail, "\n".join(evidence), "")


def inspect(host=None, port=8388, public_ip=None, lang="zh", detect_public_ip=False):
    t = translator(lang)
    host = host or Host(lang)
    results = []
    add = results.append
    system, arch = host.system()
    os_info = parse_os(host.read("/etc/os-release")) if system == "Linux" else {}
    version = ubuntu_version(os_info.get("VERSION_ID"))
    is_ubuntu = os_info.get("ID") == "ubuntu"
    supported = is_ubuntu and version is not None and version >= (22, 4)
    os_unknown = system == "Linux" and not os_info.get("ID")
    version_unknown = is_ubuntu and version is None
    add(Result("system", "pass" if supported else "unknown" if version_unknown or os_unknown else "warn", t("操作系统"),
               t("版本符合 Ubuntu 22.04 最低要求；实机兼容性按已测试环境确认") if supported else
               t("无法读取 Linux 发行版标识，不能确认系统要求") if os_unknown else
               t("无法解析 Ubuntu 版本，不能确认是否符合最低要求") if version_unknown else t("需要 Ubuntu 22.04 或更新版本"),
               f"{system}; {os_info.get('PRETTY_NAME', t('版本未知'))}; {arch}",
               t("支持范围：Ubuntu 22.04 及更新版本。") if supported else
               t("其他系统的结果仅供参考。")))
    add(Result("architecture", "pass" if arch.lower() in {"x86_64", "amd64", "aarch64", "arm64"} else "unknown",
               t("系统架构"), t("检测架构；本阶段不验证安装包兼容性"), arch))
    try:
        count = host.cpu_count()
    except (AttributeError, OSError):
        count = None
    cpuinfo = host.read('/proc/cpuinfo') if system == 'Linux' else None
    model = next((line.split(':', 1)[1].strip() for line in (cpuinfo or '').splitlines()
                  if line.split(':', 1)[0].strip() in ('model name', 'Hardware') and ':' in line), None)
    add(Result('cpu', 'pass' if count else 'unknown', t('CPU 信息'),
               t('系统可见逻辑 CPU：{count}', count=count) if count else t('逻辑 CPU 数量未知'),
               t('处理器型号：{model}', model=model) if model else t('处理器型号未知'),
               t('逻辑 CPU 数不一定等于物理核心数；配额限制可能不同。')))
    try:
        kernel = host.kernel()
    except (AttributeError, OSError):
        kernel = None
    add(Result('kernel', 'pass' if kernel else 'unknown', t('内核版本'), kernel or t('无法读取')))
    try:
        free = host.disk_free() / 1024**3
        add(Result("disk", "pass" if free >= 1 else "warn", t("根分区可用空间"),
                   t("可用 {free:.1f} GiB（体检提示阈值 1 GiB）", free=free)))
    except OSError:
        add(Result("disk", "unknown", t("根分区可用空间"), t("无法读取")))
    try:
        total, free = host.disk_total() / 1024**3, host.disk_free() / 1024**3
        add(Result('disk_capacity', 'pass', t('根分区容量'),
                   t('总容量 {total:.1f} GiB，可用 {free:.1f} GiB', total=total, free=free)))
    except (AttributeError, OSError):
        add(Result('disk_capacity', 'unknown', t('根分区容量'), t('无法读取')))
    if public_ip and detect_public_ip:
        raise ValueError("public_ip and detect_public_ip are mutually exclusive")
    if detect_public_ip:
        address = host.detect_public_ip()
        add(Result("public_ip", "unknown", t("公网 IP"),
                   t("已通过 ipify 查询出口地址；入站可达性未确认") if address else t("出口 IP 查询失败，无法确定公网地址"),
                   address or "", t("出口地址可能是 NAT 或代理地址。")))
    elif public_ip:
        addr = ipaddress.ip_address(public_ip)
        add(Result("public_ip", "unknown" if addr.is_global else "warn", t("公网 IP"),
                   t("用户提供的地址具备公网地址属性") if addr.is_global else t("提供的地址不是全局可路由公网地址"),
                   str(addr)))
    else:
        add(Result("public_ip", "unknown", t("公网 IP"), t("默认不向第三方查询出口 IP")))
    add(Result("cloud_firewall", "unknown", t("云防火墙与外部可达性"),
               t("本机读取不能确认云安全组、NAT 或外部网络路径")))
    if system != "Linux":
        for key, title in (("memory", t("内存")), ("interfaces", t("网络地址")),
                           ("default_route", t("默认路由")), ("dns_config", t("DNS 配置")),
                           ("ports", t("端口占用")), ("ssh", t("SSH 配置")),
                           ("ssh_port_safety", t("SSH 端口保护")),
                           ("firewall", t("本机防火墙")), ("service", t("sing-box 服务")),
                           ("service_startup", t("sing-box 开机启动"))):
            add(Result(key, "skip", title, t("此项需要 Linux；当前主机仅验证程序运行")))
        return report(results, port, lang)
    memory = host.read("/proc/meminfo")
    total_line = next((line for line in (memory or '').splitlines() if line.startswith('MemTotal:')), '')
    try:
        total_mib = int(total_line.split()[1]) / 1024
        add(Result('memory_total', 'pass', t('内存总量'), t('系统可见 {total:.0f} MiB', total=total_mib)))
    except (ValueError, IndexError):
        add(Result('memory_total', 'unknown', t('内存总量'), t('无法读取')))
    available = next((line for line in (memory or "").splitlines() if line.startswith("MemAvailable:")), "")
    try:
        mib = int(available.split()[1]) / 1024
        add(Result("memory", "pass" if mib >= 256 else "warn", t("可用内存"),
                   t("约 {mib:.0f} MiB（体检提示阈值 256 MiB）", mib=mib)))
    except (ValueError, IndexError):
        add(Result("memory", "unknown", t("可用内存"), t("无法读取 MemAvailable")))
    network = host.run("ip", "-j", "address", "show")
    try:
        interfaces = json.loads(network.output) if network.code == 0 else None
        addresses = [f"{item['ifname']}: {addr['local']}" for item in interfaces
                     for addr in item.get("addr_info", []) if addr.get("scope") != "host"]
        add(Result("interfaces", "pass" if addresses else "unknown", t("网络地址"),
                   t("仅表示本机接口地址"), "\n".join(addresses)))
    except (TypeError, ValueError, KeyError, AttributeError):
        add(Result("interfaces", "unknown", t("网络地址"), t(network.reason) or t("无法解析接口信息")))
    add(route_check(host, t))
    add(dns_check(host, t))
    ports = host.run("ss", "-H", "-lntu")
    rows, malformed = port_rows(ports.output)
    conflicts = [f"{proto} {endpoint}" for proto, endpoint, number in rows if number == port]
    add(Result("ports", "warn" if conflicts else "unknown" if ports.code != 0 or malformed else "pass",
               t("目标端口占用"), t("目标 TCP/UDP 端口 {port} 已有监听", port=port) if conflicts else
               t("无法完整确认端口占用") if ports.code != 0 or malformed else t("当前未发现 TCP/UDP 端口 {port} 的监听", port=port),
               "\n".join(conflicts), t("选择未占用端口；不会停止已有服务。") if conflicts else ""))
    ssh = host.run("sshd", "-T")
    add(ssh_port_check(host, ssh, port, t))
    # Only retain selected settings, never dump full ssh configuration.
    settings = dict(line.split(None, 1) for line in ssh.output.splitlines() if len(line.split(None, 1)) == 2)
    selected = {key: settings[key] for key in ("port", "permitrootlogin", "passwordauthentication", "pubkeyauthentication") if key in settings}
    risk = settings.get("permitrootlogin") == "yes" or settings.get("passwordauthentication") == "yes"
    ssh_complete = all(key in settings for key in ("permitrootlogin", "passwordauthentication"))
    add(Result("ssh", "unknown" if ssh.code != 0 else "warn" if risk else "unknown", t("SSH 配置"),
               (t(ssh.reason) or t("读取失败")) if ssh.code != 0 else t("允许 root 登录或密码认证") if risk else
               t("SSH 输出缺少关键字段，无法完成风险判断") if not ssh_complete else t("未发现这两项宽松设置"),
               "\n".join(line for line in ssh.output.splitlines() if line.split() and line.split()[0] in selected) if ssh.code == 0 else "",
               t("sshd -T 读取默认配置；权限不足、配置错误或 SSH 运行目录尚未创建均可能导致失败。结果可能与运行中服务或 Match 条件不同。")))
    firewall = []
    any_rules = False
    complete = True
    for args in (("ufw", "status"), ("nft", "list", "ruleset"), ("iptables", "-S")):
        cmd = host.run(*args)
        if cmd.code != 0:
            complete = False
            firewall.append(f"{args[0]}: {t(cmd.reason) or t('读取失败')}")
        else:
            meaningful = bool(cmd.output.strip()) and not cmd.output.strip().startswith("Status: inactive")
            any_rules |= meaningful
            firewall.append(f"{args[0]}: " + t("已读取；") + (t("存在状态或规则信息") if meaningful else t("未发现活动状态或规则")))
    add(Result("firewall", "unknown", t("本机防火墙"),
               t("发现防火墙状态或规则；未判断规则是否允许目标端口") if any_rules else t("没有足够证据确认防火墙保护状态"),
               "\n".join(firewall), t("部分工具不可用或权限不足。") if not complete else t("规则存在不等于配置安全。")))
    active = host.run("systemctl", "is-active", "sing-box.service")
    enabled = host.run("systemctl", "is-enabled", "sing-box.service")
    running = active.code == 0 and active.output.strip() == "active"
    failed = active.output.strip() == "failed"
    add(Result("service", "warn" if failed else "pass" if running else "unknown", t("sing-box 服务"),
               t("sing-box 服务处于 failed 状态，需要检查启动失败原因") if failed else
               t("服务处于 active") if running else t("未确认 sing-box 服务运行；新系统尚未安装属于正常情况"),
               f"active: {active.output.strip() or t(active.reason)}\nenabled: {enabled.output.strip() or t(enabled.reason)}"))
    startup = enabled.output.strip()
    startup_ok = enabled.code == 0 and startup == "enabled"
    startup_warn = startup in {"disabled", "masked", "enabled-runtime", "masked-runtime"}
    add(Result("service_startup", "pass" if startup_ok else "warn" if startup_warn else "unknown",
               t("sing-box 开机启动"), t("已配置持久开机启动") if startup_ok else
               t("未配置持久开机启动或服务被屏蔽，重启后可能不可用") if startup_warn else
               t("无法确认 sing-box 的持久开机启动状态"), startup or t(enabled.reason)))
    return report(results, port, lang)


def report(results, port, lang="zh"):
    document={"schema_version": 1, "language": lang, "tool_version": __version__,
            "generated_at": datetime.now(timezone.utc).isoformat(), "mode": "read-only",
            "target_port": port, "checks": [asdict(result) for result in results]}
    from .schema import normalize_report
    normalize_report(document)
    return document