#!/usr/bin/env python3
"""
ganyant-agent — telemetry, intrusion signals and zero-knowledge backups.

  ganyant-agent run                       daemon (systemd / Windows service)
  ganyant-agent backup <job> [--manual]   run one backup now
  ganyant-agent restore <job> --at latest|"YYYY-MM-DD HH:MM" --to PATH
  ganyant-agent decrypt FILE.gya --out PATH

Snapshot format (.gya v1):
  b"GYA1" | salt[16] | opslimit u32 | memlimit u32 | secretstream header[24] | { u32 len | chunk }*
  plaintext = zstd(dump); chunk = XChaCha20-Poly1305 secretstream (libsodium), key = Argon2id(passphrase, salt).
  Integrity: BLAKE2b-256 over the whole ciphertext file, verified before upload and before decrypt.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import platform
import re
import shutil
import socket
import struct
import subprocess
import sys
import tempfile
import threading
import time
import tomllib
import urllib.request
import uuid
from collections import defaultdict, deque
from pathlib import Path
from typing import Any, BinaryIO, Iterator

VERSION = "1.0.0"
CONF = Path(os.environ.get("GANYANT_CONFIG", "/etc/ganyant/agent.toml" if os.name != "nt" else r"C:\ProgramData\ganyant\agent.toml"))
API = os.environ.get("GANYANT_API", "https://ganyant.com/api/agent")
CHUNK = 4 * 1024 * 1024
MAGIC = b"GYA1"


# ----------------------------------------------------------------------------- config / http
def load_conf() -> dict[str, Any]:
    with CONF.open("rb") as f:
        c = tomllib.load(f)
    c.setdefault("vault", {})
    c.setdefault("backup", {})
    return c


def secret(v: str | None) -> str | None:
    if v and v.startswith("env:"):
        return os.environ.get(v[4:])
    return v


def machine_id() -> str:
    for p in ("/etc/machine-id", "/var/lib/dbus/machine-id"):
        try:
            return Path(p).read_text().strip()
        except OSError:
            pass
    return str(uuid.getnode())


class Api:
    def __init__(self, key: str) -> None:
        self.key = key
        self.mid = machine_id()

    def post(self, action: str, body: Any, timeout: int = 20) -> dict[str, Any]:
        req = urllib.request.Request(f"{API}/{action}", data=json.dumps(body).encode(), method="POST", headers={
            "Authorization": f"Bearer {self.key}", "Content-Type": "application/json",
            "X-Ganyant-Machine": self.mid, "User-Agent": f"ganyant-agent/{VERSION}"})
        try:
            with urllib.request.urlopen(req, timeout=timeout) as r:
                return json.loads(r.read() or b"{}")
        except urllib.error.HTTPError as e:
            raise RuntimeError(f"{action}: HTTP {e.code} {e.read()[:200].decode(errors='ignore')}") from None


# ----------------------------------------------------------------------------- crypto
def _sodium():
    from nacl import bindings as b, pwhash
    return b, pwhash


def derive_key(passphrase: str, salt: bytes, ops: int, mem: int) -> bytes:
    _, pwhash = _sodium()
    return pwhash.argon2id.kdf(32, passphrase.encode(), salt, opslimit=ops, memlimit=mem)


def encrypt_stream(src: Iterator[bytes], out: BinaryIO, passphrase: str) -> None:
    b, pwhash = _sodium()
    salt = os.urandom(16)
    ops, mem = pwhash.argon2id.OPSLIMIT_MODERATE, pwhash.argon2id.MEMLIMIT_MODERATE
    key = derive_key(passphrase, salt, ops, mem)
    state = b.crypto_secretstream_xchacha20poly1305_state()
    header = b.crypto_secretstream_xchacha20poly1305_init_push(state, key)
    out.write(MAGIC + salt + struct.pack(">II", ops, mem) + header)
    pending: bytes | None = None
    for chunk in src:
        if pending is not None:
            c = b.crypto_secretstream_xchacha20poly1305_push(state, pending, None, 0)
            out.write(struct.pack(">I", len(c)) + c)
        pending = chunk
    c = b.crypto_secretstream_xchacha20poly1305_push(state, pending or b"", None, b.crypto_secretstream_xchacha20poly1305_TAG_FINAL)
    out.write(struct.pack(">I", len(c)) + c)


def decrypt_stream(inp: BinaryIO, passphrase: str) -> Iterator[bytes]:
    b, _ = _sodium()
    if inp.read(4) != MAGIC:
        raise ValueError("not a ganyant snapshot")
    salt = inp.read(16)
    ops, mem = struct.unpack(">II", inp.read(8))
    header = inp.read(24)
    key = derive_key(passphrase, salt, ops, mem)
    state = b.crypto_secretstream_xchacha20poly1305_state()
    b.crypto_secretstream_xchacha20poly1305_init_pull(state, header, key)
    while True:
        ln = inp.read(4)
        if not ln:
            raise ValueError("truncated snapshot")
        msg, tag = b.crypto_secretstream_xchacha20poly1305_pull(state, inp.read(struct.unpack(">I", ln)[0]), None)
        yield msg
        if tag == b.crypto_secretstream_xchacha20poly1305_TAG_FINAL:
            return


def blake2b_file(path: Path) -> str:
    h = hashlib.blake2b(digest_size=32)
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(CHUNK), b""):
            h.update(chunk)
    return h.hexdigest()


# ----------------------------------------------------------------------------- dumps
def dump_command(kind: str, spec: dict[str, Any]) -> list[str]:
    if kind == "postgres":
        return ["pg_dump", "--format=custom", "--no-owner", secret(spec.get("dsn")) or spec["database"]]
    if kind == "mysql":
        args = ["mysqldump", "--single-transaction", "--quick", "--routines", "--triggers"]
        if spec.get("defaults_file"):
            args.insert(1, f"--defaults-extra-file={spec['defaults_file']}")
        return args + [spec["database"]]
    if kind == "mongodb":
        return ["mongodump", f"--uri={secret(spec.get('uri')) or 'mongodb://127.0.0.1'}", "--archive"]
    if kind == "sqlite":
        tmp = Path(tempfile.gettempdir()) / f"gy-{uuid.uuid4().hex}.sqlite"
        subprocess.run(["sqlite3", spec["path"], f".backup '{tmp}'"], check=True)
        return ["sh", "-c", f"cat '{tmp}' && rm -f '{tmp}'"] if os.name != "nt" else ["cmd", "/c", f"type {tmp}"]
    if kind == "directory":
        return ["tar", "-C", str(Path(spec["path"]).parent), "-cf", "-", Path(spec["path"]).name]
    raise ValueError(f"unknown backup kind {kind}")


def compressed_dump(cmd: list[str], counter: list[int]) -> Iterator[bytes]:
    import zstandard
    nice = ["nice", "-n", "15", "ionice", "-c3"] if shutil.which("ionice") else []
    proc = subprocess.Popen(nice + cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    cctx = zstandard.ZstdCompressor(level=9, threads=-1)
    reader = cctx.stream_reader(_Counting(proc.stdout, counter))
    while chunk := reader.read(CHUNK):
        yield chunk
    if proc.wait() != 0:
        raise RuntimeError(f"{cmd[0]} falló: {proc.stderr.read()[:300].decode(errors='ignore')}")


class _Counting:
    def __init__(self, f: BinaryIO, counter: list[int]) -> None:
        self.f, self.c = f, counter

    def read(self, n: int = -1) -> bytes:
        d = self.f.read(n)
        self.c[0] += len(d)
        return d


def run_backup(api: Api, conf: dict[str, Any], job: dict[str, Any], manual: bool = False) -> None:
    spec = {**job.get("source", {}), **conf["backup"].get(job["name"], {})}
    passphrase = secret(conf["vault"].get("passphrase"))
    if not passphrase:
        raise RuntimeError("vault.passphrase no configurada en agent.toml")
    raw = [0]
    with tempfile.NamedTemporaryFile(prefix="gy-", suffix=".gya", delete=False) as tmp:
        path = Path(tmp.name)
        try:
            encrypt_stream(compressed_dump(dump_command(job["kind"], spec), raw), tmp, passphrase)
        except Exception as e:
            path.unlink(missing_ok=True)
            raise RuntimeError(f"dump/cifrado falló: {e}") from e
    try:
        digest = blake2b_file(path)
        size = path.stat().st_size
        salt_hex = path.open("rb").read(20)[4:].hex()
        p = api.post("presign", {"job_id": job["id"], "size_bytes": size, "original_bytes": raw[0], "kdf_salt": salt_hex, "manual": manual})
        try:
            with path.open("rb") as f:
                req = urllib.request.Request(p["upload_url"], data=f, method="PUT", headers={"Content-Length": str(size), "Content-Type": "application/octet-stream"})
                urllib.request.urlopen(req, timeout=3600).close()
            api.post("complete", {"snapshot_id": p["snapshot_id"], "ok": True, "blake2b": digest})
            log(f"backup {job['name']}: {raw[0]} → {size} bytes, blake2b {digest[:12]}…")
        except Exception as e:
            api.post("complete", {"snapshot_id": p["snapshot_id"], "ok": False, "error": str(e)[:400]})
            raise
    finally:
        path.unlink(missing_ok=True)


def restore(api: Api, conf: dict[str, Any], job_name: str, at: str, to: str) -> None:
    import zstandard
    snap = api.post("restore", {"job_name": job_name, "at": at})
    passphrase = secret(conf["vault"].get("passphrase")) or input("Frase de cifrado: ")
    with tempfile.NamedTemporaryFile(suffix=".gya", delete=False) as tmp:
        with urllib.request.urlopen(snap["download_url"], timeout=3600) as r:
            shutil.copyfileobj(r, tmp, CHUNK)
    path = Path(tmp.name)
    try:
        if blake2b_file(path) != snap["blake2b"]:
            raise RuntimeError("BLAKE2b no coincide: el snapshot fue alterado o está incompleto")
        _write_plain(path, Path(to), passphrase, zstandard)
        print(f"✓ BLAKE2b verificado · descifrado · escrito en {to}")
    finally:
        path.unlink(missing_ok=True)


def _write_plain(src: Path, dst: Path, passphrase: str, zstandard: Any) -> None:
    d = zstandard.ZstdDecompressor().decompressobj()
    with src.open("rb") as f, dst.open("wb") as out:
        for chunk in decrypt_stream(f, passphrase):
            out.write(d.decompress(chunk))


# ----------------------------------------------------------------------------- telemetry
def probe_lines(interval: int) -> Iterator[dict[str, Any]]:
    native = Path(__file__).with_name("ganyant-probe")
    if native.exists() and os.name != "nt":
        p = subprocess.Popen([str(native), str(interval)], stdout=subprocess.PIPE, text=True)
        for line in p.stdout:  # type: ignore[union-attr]
            yield json.loads(line)
    import psutil  # portable fallback (Windows / no compiler)
    last = psutil.net_io_counters()
    while True:
        cpu = psutil.cpu_percent(interval=interval)
        vm, sw, now = psutil.virtual_memory(), psutil.swap_memory(), psutil.net_io_counters()
        load = os.getloadavg() if hasattr(os, "getloadavg") else (0.0, 0.0, 0.0)
        disks = []
        for part in psutil.disk_partitions(all=False):
            try:
                u = psutil.disk_usage(part.mountpoint)
                disks.append({"mount": part.mountpoint, "used": u.used, "total": u.total})
            except OSError:
                pass
        yield {"cpu_pct": cpu, "mem_used_bytes": vm.total - vm.available, "mem_total_bytes": vm.total, "swap_used_bytes": sw.used,
               "swap_total_bytes": sw.total, "load1": load[0], "load5": load[1], "load15": load[2], "disks": disks,
               "net_rx_bps": int((now.bytes_recv - last.bytes_recv) / interval), "net_tx_bps": int((now.bytes_sent - last.bytes_sent) / interval)}
        last = now


def _event(api: Api, kind: str, severity: str, details: dict[str, Any], source_ip: str | None = None) -> None:
    try:
        api.post("events", [{"kind": kind, "severity": severity, "source_ip": source_ip, "details": details}])
    except Exception as e:  # noqa: BLE001
        log(f"events: {e}")


class Signals:
    """Local heuristics computed on the agent: traffic anomalies and disk-full prediction."""

    def __init__(self, api: Api) -> None:
        self.api = api
        self.rx: deque[float] = deque(maxlen=720)
        self.disk: dict[str, deque[tuple[float, float, float]]] = defaultdict(lambda: deque(maxlen=2000))
        self.last: dict[str, float] = {}

    def _cool(self, key: str, seconds: int) -> bool:
        now = time.time()
        if now - self.last.get(key, 0) < seconds:
            return False
        self.last[key] = now
        return True

    def feed(self, m: dict[str, Any]) -> None:
        rx = float(m.get("net_rx_bps") or 0)
        if len(self.rx) >= 60:
            mean = sum(self.rx) / len(self.rx)
            std = (sum((x - mean) ** 2 for x in self.rx) / len(self.rx)) ** 0.5
            if rx > mean + 4 * std and rx > 5_000_000 and self._cool("traffic", 1800):
                _event(self.api, "traffic_anomaly", "medium", {"rx_bps": int(rx), "baseline_bps": int(mean), "message": f"Tráfico entrante {rx/1e6:.1f} MB/s, {rx/max(mean,1):.0f}× lo normal"})
        self.rx.append(rx)
        now = time.time()
        for d in m.get("disks") or []:
            hist = self.disk[d["mount"]]
            hist.append((now, float(d["used"]), float(d["total"])))
            if len(hist) < 30 or now - hist[0][0] < 3 * 3600:
                continue
            (t0, u0, _), (t1, u1, total) = hist[0], hist[-1]
            rate = (u1 - u0) / max(t1 - t0, 1)  # bytes/s
            if rate <= 0:
                continue
            days = (total - u1) / rate / 86400
            if days < 7 and self._cool(f"disk:{d['mount']}", 86400):
                _event(self.api, "disk_predictive", "high" if days < 2 else "medium", {"mount": d["mount"], "days_to_full": round(days, 1), "message": f"{d['mount']} se llena en ~{days:.1f} días al ritmo actual"})


def telemetry_loop(api: Api, interval: int) -> None:
    batch: list[dict[str, Any]] = []
    flush_every = max(1, 15 // interval)
    signals = Signals(api)
    for m in probe_lines(interval):
        m["ts"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        signals.feed(m)
        batch.append(m)
        if len(batch) >= flush_every:
            try:
                api.post("metrics", batch)
                batch = []
            except Exception as e:  # noqa: BLE001
                log(f"metrics: {e}")
                batch = batch[-120:]


# ----------------------------------------------------------------------------- intrusion signals
SSH_FAIL = re.compile(r"(?:Failed password|Invalid user|authentication failure).*?(?:for (?:invalid user )?(\S+) )?from (\d+\.\d+\.\d+\.\d+|[0-9a-f:]+)")


def auth_lines() -> Iterator[str]:
    if shutil.which("journalctl"):
        p = subprocess.Popen(["journalctl", "-f", "-n0", "-o", "cat", "-u", "ssh", "-u", "sshd"], stdout=subprocess.PIPE, text=True)
        yield from p.stdout  # type: ignore[misc]
    for path in ("/var/log/auth.log", "/var/log/secure"):
        if Path(path).exists():
            with open(path, errors="ignore") as f:
                f.seek(0, 2)
                while True:
                    line = f.readline()
                    if not line:
                        time.sleep(1)
                        continue
                    yield line


def intrusion_loop(api: Api, threshold: int = 20, window: int = 300) -> None:
    hits: dict[str, deque[float]] = defaultdict(deque)
    users: dict[str, set[str]] = defaultdict(set)
    reported: dict[str, float] = {}
    for line in auth_lines():
        m = SSH_FAIL.search(line)
        if not m:
            continue
        user, ip = m.group(1), m.group(2)
        now = time.time()
        q = hits[ip]
        q.append(now)
        if user:
            users[ip].add(user)
        while q and q[0] < now - window:
            q.popleft()
        if len(q) >= threshold and now - reported.get(ip, 0) > 1800:
            reported[ip] = now
            try:
                api.post("events", [{"kind": "ssh_bruteforce", "severity": "high", "source_ip": ip,
                                      "details": {"attempts": len(q), "window_s": window, "users": sorted(users[ip])[:10]}}])
            except Exception as e:  # noqa: BLE001
                log(f"events: {e}")


def listening_ports() -> dict[str, str]:
    import psutil
    out: dict[str, str] = {}
    for c in psutil.net_connections(kind="inet"):
        if c.status != psutil.CONN_LISTEN or not c.laddr:
            continue
        ip = c.laddr.ip
        if ip.startswith("127.") or ip == "::1":
            continue
        name = ""
        try:
            name = psutil.Process(c.pid).name() if c.pid else ""
        except Exception:  # noqa: BLE001
            pass
        out[f"{c.laddr.port}/{'tcp' if c.type == socket.SOCK_STREAM else 'udp'}"] = name
    return out


def ports_loop(api: Api) -> None:
    try:
        known = listening_ports()
    except Exception as e:  # noqa: BLE001
        log(f"ports: {e}")
        return
    while True:
        time.sleep(60)
        try:
            now = listening_ports()
            for port, proc in now.items():
                if port not in known:
                    _event(api, "new_listening_port", "medium", {"port": port, "process": proc, "message": f"Nuevo puerto {port} en escucha ({proc or 'proceso desconocido'})"})
            known = now
        except Exception as e:  # noqa: BLE001
            log(f"ports: {e}")


def rdp_loop(api: Api, threshold: int = 20, window: int = 300) -> None:
    """Windows: counts failed logons (event 4625) per source IP."""
    reported: dict[str, float] = {}
    while True:
        try:
            q = f"*[System[(EventID=4625) and TimeCreated[timediff(@SystemTime) <= {window * 1000}]]]"
            out = subprocess.run(["wevtutil", "qe", "Security", f"/q:{q}", "/f:text", "/c:2000"], capture_output=True, text=True, timeout=30).stdout
            counts: dict[str, int] = defaultdict(int)
            users: dict[str, set[str]] = defaultdict(set)
            for block in out.split("Event["):
                ip = re.search(r"Source Network Address:\s+([0-9a-fA-F.:]+)", block)
                if not ip or ip.group(1) in ("-", "127.0.0.1", "::1"):
                    continue
                counts[ip.group(1)] += 1
                u = re.findall(r"Account Name:\s+(\S+)", block)
                if len(u) > 1:
                    users[ip.group(1)].add(u[1])
            for ip, n in counts.items():
                if n >= threshold and time.time() - reported.get(ip, 0) > 1800:
                    reported[ip] = time.time()
                    _event(api, "rdp_bruteforce", "high", {"attempts": n, "window_s": window, "users": sorted(users[ip])[:10]}, ip)
        except Exception as e:  # noqa: BLE001
            log(f"rdp: {e}")
        time.sleep(60)


# ----------------------------------------------------------------------------- daemon
def log(msg: str) -> None:
    print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}", flush=True)


def daemon() -> None:
    conf = load_conf()
    api = Api(secret(conf["agent"]["key"]) or "")
    info = api.post("register", {
        "machine_id": api.mid, "hostname": socket.gethostname(), "os": f"{platform.system()} {platform.release()}",
        "kernel": platform.version()[:120], "arch": platform.machine(), "agent_version": VERSION,
        "cpu_cores": os.cpu_count(), "mem_total_bytes": _mem_total()})
    log(f"registrado como {socket.gethostname()} (modo {info.get('mode')})")
    if info.get("mode") == "full" and info.get("telemetry_interval_s"):
        threading.Thread(target=telemetry_loop, args=(api, int(info["telemetry_interval_s"])), daemon=True).start()
        threading.Thread(target=ports_loop, args=(api,), daemon=True).start()
        if os.name == "nt":
            threading.Thread(target=rdp_loop, args=(api,), daemon=True).start()
        else:
            threading.Thread(target=intrusion_loop, args=(api,), daemon=True).start()
    running: set[str] = set()
    while True:
        try:
            cfg = api.post("config", {})
            for job in cfg.get("jobs", []):
                if job.get("due") and job["id"] not in running:
                    running.add(job["id"])
                    threading.Thread(target=_job_thread, args=(api, conf, job, running), daemon=True).start()
        except Exception as e:  # noqa: BLE001
            log(f"config: {e}")
        time.sleep(60)


def _job_thread(api: Api, conf: dict[str, Any], job: dict[str, Any], running: set[str]) -> None:
    try:
        run_backup(api, conf, job)
    except Exception as e:  # noqa: BLE001
        log(f"backup {job['name']} falló: {e}")
    finally:
        running.discard(job["id"])


def _mem_total() -> int | None:
    try:
        return int(re.search(r"MemTotal:\s+(\d+)", Path("/proc/meminfo").read_text()).group(1)) * 1024  # type: ignore[union-attr]
    except Exception:  # noqa: BLE001
        return None


def main() -> None:
    ap = argparse.ArgumentParser(prog="ganyant-agent")
    sub = ap.add_subparsers(dest="cmd", required=True)
    sub.add_parser("run")
    b = sub.add_parser("backup"); b.add_argument("job"); b.add_argument("--manual", action="store_true")
    r = sub.add_parser("restore"); r.add_argument("job"); r.add_argument("--at", default="latest"); r.add_argument("--to", required=True)
    d = sub.add_parser("decrypt"); d.add_argument("file"); d.add_argument("--out", required=True)
    sub.add_parser("version")
    a = ap.parse_args()
    if a.cmd == "version":
        print(VERSION); return
    if a.cmd == "decrypt":
        import getpass, zstandard
        _write_plain(Path(a.file), Path(a.out), getpass.getpass("Frase de cifrado: "), zstandard)
        print(f"✓ descifrado en {a.out}"); return
    conf = load_conf()
    api = Api(secret(conf["agent"]["key"]) or "")
    if a.cmd == "run":
        daemon()
    elif a.cmd == "backup":
        jobs = api.post("config", {}).get("jobs", [])
        job = next((j for j in jobs if j["name"] == a.job), None)
        if not job:
            sys.exit(f"no existe la tarea {a.job} para este servidor")
        run_backup(api, conf, job, manual=a.manual)
    elif a.cmd == "restore":
        restore(api, conf, a.job, a.at, a.to)


if __name__ == "__main__":
    main()