#!/usr/bin/env python3
"""Sonde volontaire Trinity ; Python standard, cible fixe et configuration privée.

Usage : python trinity-probe.py --config trinity-probe.json --once
        python trinity-probe.py --config trinity-probe.json --interval
"""
from __future__ import annotations

import argparse
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import hmac
import http.client
from http.client import HTTPException
import ipaddress
import json
import math
import os
from pathlib import Path
import queue
import re
import socket
import sqlite3
import ssl
import stat
import threading
import time
from urllib.parse import urlsplit
import uuid


BASE_URL = "https://trinity.julienpiron.fr"
ALLOWED_BASE_URLS = (BASE_URL, "https://julienpiron.fr/trinity", "https://julienpiron.fr/trinity-1f")
ALLOWED_HOSTS = tuple(urlsplit(base).hostname for base in ALLOWED_BASE_URLS)
HOST = urlsplit(BASE_URL).hostname
TARGET = BASE_URL + "/probe-target.json"
REPORT_URL = BASE_URL + "/api/stations/report"
CONFIG_SCHEMA = "trinity.probe.config/1"
PROTOCOL = "trinity.station.report/1"
RECEIPT_PROTOCOL = "trinity.station.receipt/1"
REPORT_DIRECTION = "trinity.station.report/1"
RESPONSE_DIRECTION = "trinity.station.response/1"
INTERVAL_SECONDS = 300
TIMEOUT = 10
MAX_BODY = 8192
SIGNATURE_HEADER = "X-Trinity-Station-Signature"
_DNS_SLOTS = threading.BoundedSemaphore(4)


def encoded(value):
    return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")


def signature(key, body, direction):
    return hmac.new(key, direction.encode("ascii") + b"\n" + body, hashlib.sha256).hexdigest()


def utc(stamp):
    return datetime.fromtimestamp(stamp, timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")


def _unique(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("Champ JSON répété.")
        result[key] = value
    return result


def parse_json(raw):
    if not isinstance(raw, bytes) or not 0 < len(raw) <= MAX_BODY:
        raise ValueError("Réponse JSON hors limite.")
    try:
        return json.loads(raw.decode("utf-8"), object_pairs_hook=_unique, parse_constant=lambda _: (_ for _ in ()).throw(ValueError("Nombre invalide.")))
    except (UnicodeError, RecursionError, json.JSONDecodeError):
        raise ValueError("Réponse JSON invalide.") from None


@dataclass(frozen=True)
class Config:
    station_id: str
    key: bytes
    target: str = TARGET
    interval_seconds: int = INTERVAL_SECONDS
    api_base_url: str = BASE_URL + "/api"

    @classmethod
    def load(cls, path):
        path = Path(path)
        flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
        fd = os.open(path, flags)
        try:
            info = os.fstat(fd)
            if not stat.S_ISREG(info.st_mode):
                raise ValueError("Un fichier de configuration ordinaire est requis.")
            if os.name != "nt":
                if info.st_uid != os.getuid():
                    raise ValueError("La configuration doit appartenir à cet utilisateur.")
                os.fchmod(fd, 0o600)
            with os.fdopen(fd, "rb", closefd=False) as handle:
                data = parse_json(handle.read(MAX_BODY + 1))
        finally:
            os.close(fd)
        if not isinstance(data, dict) or set(data) != {"schema", "station_id", "secret", "target", "api_base_url", "interval_seconds"}:
            raise ValueError("Configuration de station invalide.")
        try:
            station_id = str(uuid.UUID(data["station_id"]))
            secret = data["secret"]
            if not isinstance(secret, str) or not re.fullmatch(r"[0-9a-fA-F]{64}", secret):
                raise ValueError
            key = bytes.fromhex(secret)
        except (ValueError, TypeError, AttributeError):
            raise ValueError("Identifiant ou secret invalide.") from None
        valid_pair = any(data["target"] == base + "/probe-target.json" and data["api_base_url"] == base + "/api" for base in ALLOWED_BASE_URLS)
        if data["schema"] != CONFIG_SCHEMA or data["station_id"] != station_id or not valid_pair or type(data["interval_seconds"]) is not int or data["interval_seconds"] != INTERVAL_SECONDS:
            raise ValueError("Cible Trinity et cadence serveur requises.")
        return cls(station_id, key, target=data["target"], api_base_url=data["api_base_url"])


class ClientState:
    """Réserve durablement un numéro avant toute mesure ou tentative d'envoi."""

    def __init__(self, path, config):
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        if path.is_symlink():
            raise ValueError("L'état privé ne peut pas être un lien symbolique.")
        self.db = sqlite3.connect(path, timeout=5)
        if os.name != "nt":
            path.chmod(0o600)
        self.db.execute("CREATE TABLE IF NOT EXISTS state(singleton INTEGER PRIMARY KEY CHECK(singleton=1),station_id TEXT NOT NULL,sequence INTEGER NOT NULL,next_due REAL NOT NULL,last_outcome TEXT NOT NULL)")
        with self.db:
            self.db.execute("INSERT OR IGNORE INTO state VALUES(1,?,0,0,'waiting')", (config.station_id,))
        if self.db.execute("SELECT station_id FROM state WHERE singleton=1").fetchone()[0] != config.station_id:
            self.db.close()
            raise ValueError("Ce fichier d'état appartient à une autre station.")

    def close(self):
        self.db.close()

    def reserve(self, now):
        if isinstance(now, bool) or not isinstance(now, (int, float)) or not math.isfinite(now) or now < 0:
            raise ValueError("Horloge invalide.")
        self.db.execute("BEGIN IMMEDIATE")
        try:
            sequence, next_due = self.db.execute("SELECT sequence,next_due FROM state WHERE singleton=1").fetchone()
            if now < next_due:
                self.db.rollback()
                return None
            sequence += 1
            if sequence > 2**53 - 1:
                raise ValueError("Séquence épuisée.")
            self.db.execute("UPDATE state SET sequence=?,next_due=?,last_outcome='unknown' WHERE singleton=1", (sequence, now + INTERVAL_SECONDS))
            self.db.commit()
            return sequence
        except BaseException:
            self.db.rollback()
            raise

    def finish(self, sequence, outcome):
        with self.db:
            self.db.execute("UPDATE state SET last_outcome=? WHERE singleton=1 AND sequence=?", (outcome, sequence))

    def next_due(self):
        return self.db.execute("SELECT next_due FROM state WHERE singleton=1").fetchone()[0]


@dataclass(frozen=True)
class HttpResponse:
    status: int
    body: bytes
    elapsed_ms: float


def _bounded_dns(host, port, family, timeout):
    if host not in ALLOWED_HOSTS or port != 443 or family not in (0, socket.AF_INET, socket.AF_INET6):
        raise ValueError("Résolution hors cible interdite.")
    if not _DNS_SLOTS.acquire(blocking=False):
        raise TimeoutError("Résolveur occupé.")
    results = queue.Queue(maxsize=1)

    def work():
        try:
            results.put((True, socket.getaddrinfo(host, 443, family, socket.SOCK_STREAM)))
        except OSError as exc:
            results.put((False, exc))
        finally:
            _DNS_SLOTS.release()

    worker = threading.Thread(target=work, daemon=True)
    worker.start()
    try:
        ok, value = results.get(timeout=timeout)
    except queue.Empty:
        raise TimeoutError("Délai DNS dépassé.") from None
    if not ok:
        raise value
    return value


def _fixed_connection(address, timeout=TIMEOUT, source_address=None):
    """Résolution propre à la connexion HTTPS ; aucune IP de rapport réutilisée."""
    if address[0] not in ALLOWED_HOSTS or address[1] != 443 or source_address is not None:
        raise ValueError("Connexion hors cible interdite.")
    started = time.monotonic()
    addresses = _bounded_dns(address[0], 443, 0, timeout)
    last_error = OSError("Aucune adresse publique disponible pour Trinity.")
    for family, socktype, proto, _, sockaddr in addresses[:16]:
        # Une entrée DNS vers un réseau local n'autorise pas de connexion locale.
        if not ipaddress.ip_address(sockaddr[0]).is_global:
            continue
        remaining = timeout - (time.monotonic() - started)
        if remaining <= 0:
            raise TimeoutError("Délai de connexion dépassé.")
        sock = socket.socket(family, socktype, proto)
        try:
            sock.settimeout(remaining)
            sock.connect(sockaddr)
            return sock
        except OSError as exc:
            last_error = exc
            sock.close()
    raise last_error


def https_request(method, url, *, body=None, headers=None, timeout=TIMEOUT):
    """Seules les deux routes constantes sont accessibles ; aucun redirect/proxy."""
    allowed = tuple(pair for base in ALLOWED_BASE_URLS for pair in (("GET", base + "/probe-target.json"), ("POST", base + "/api/stations/report")))
    if (method, url) not in allowed:
        raise ValueError("Destination de sonde interdite.")
    if headers is not None and (not isinstance(headers, dict) or set(headers) - {"Content-Type", SIGNATURE_HEADER}):
        raise ValueError("En-tête de sonde interdit.")
    parts = urlsplit(url)
    connection = http.client.HTTPSConnection(parts.hostname, port=443, timeout=timeout, context=ssl.create_default_context())
    connection._create_connection = _fixed_connection
    start = time.monotonic()

    def abort():
        sock = connection.sock
        if sock is not None:
            try:
                sock.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass
        connection.close()

    watchdog = threading.Timer(timeout, abort)
    watchdog.daemon = True
    watchdog.start()
    try:
        request_headers = {"Accept": "application/json", "Accept-Encoding": "identity", "User-Agent": "Trinity-Volunteer-Probe/1"}
        if headers:
            request_headers.update(headers)
        connection.request(method, parts.path, body=body, headers=request_headers)
        response = connection.getresponse()
        # Un Location n'est jamais lu ni suivi, y compris vers Trinity.
        if 300 <= response.status <= 399:
            return HttpResponse(response.status, b"", (time.monotonic() - start) * 1000)
        length = response.getheader("Content-Length")
        if length is not None and (not length.isdecimal() or int(length) > MAX_BODY):
            raise ValueError("Réponse trop volumineuse.")
        if response.getheader("Content-Encoding", "identity").lower() not in ("", "identity"):
            raise ValueError("Réponse compressée non acceptée.")
        chunks, count = [], 0
        while True:
            remaining = timeout - (time.monotonic() - start)
            if remaining <= 0:
                raise TimeoutError("Délai HTTPS dépassé.")
            if connection.sock is not None:
                connection.sock.settimeout(remaining)
            chunk = response.read(min(4096, MAX_BODY + 1 - count))
            if not chunk:
                break
            chunks.append(chunk)
            count += len(chunk)
            if count > MAX_BODY:
                raise ValueError("Réponse trop volumineuse.")
        return HttpResponse(response.status, b"".join(chunks), (time.monotonic() - start) * 1000)
    finally:
        watchdog.cancel()
        connection.close()


def resolve_fixed(host, family):
    if host not in ALLOWED_HOSTS or family not in (socket.AF_INET, socket.AF_INET6):
        raise ValueError("Résolution hors cible interdite.")
    return sorted({str(ipaddress.ip_address(item[4][0])) for item in _bounded_dns(host, 443, family, TIMEOUT)})[:16]


def measure(*, target=TARGET, transport=https_request, resolve=resolve_fixed):
    if target not in tuple(base + "/probe-target.json" for base in ALLOWED_BASE_URLS):
        raise ValueError("Cible de mesure interdite.")
    host = urlsplit(target).hostname
    start = time.monotonic()
    answers, failures = {}, 0
    for name, family, version in (("a", socket.AF_INET, 4), ("aaaa", socket.AF_INET6, 6)):
        try:
            values = resolve(host, family)
            if not isinstance(values, (list, tuple)) or len(values) > 16:
                raise ValueError
            addresses = [ipaddress.ip_address(value) for value in values]
            if any(value.version != version or "%" in str(value) for value in addresses):
                raise ValueError
            answers[name] = sorted({str(value) for value in addresses})
        except (OSError, ValueError, TypeError):
            answers[name] = []
            failures += 1
    dns = {**answers, "latency_ms": min(60_000, round((time.monotonic() - start) * 1000, 3)), "error": "resolve" if failures == 2 else "partial" if failures else None}
    http = {"ok": False, "status": None, "latency_ms": None, "body_sha256": None, "error": "transport"}
    try:
        response = transport("GET", target, timeout=TIMEOUT)
        if not isinstance(response.body, bytes) or len(response.body) > MAX_BODY:
            raise ValueError
        http.update(status=response.status, latency_ms=min(60_000, max(0, round(response.elapsed_ms, 3))), body_sha256=hashlib.sha256(response.body).hexdigest())
        if 300 <= response.status <= 399:
            http["error"] = "redirect"
        elif response.status != 200:
            http["error"] = "http"
        else:
            data = parse_json(response.body)
            if not isinstance(data, dict) or data.get("schema") != "trinity.probe-target/1" or data.get("ok") is not True:
                http["error"] = "target_schema"
            else:
                http.update(ok=True, error=None)
    except ssl.SSLError:
        http["error"] = "tls"
    except (TimeoutError, socket.timeout):
        http["error"] = "timeout"
    except socket.gaierror:
        http["error"] = "dns"
    except (ConnectionError, OSError, HTTPException):
        http["error"] = "connect"
    except ValueError:
        http["error"] = "body_limit"
    return http, dns


def verify_receipt(config, raw, body, sequence):
    data = parse_json(raw)
    keys = {"protocol", "station_id", "sequence", "received_at", "payload_sha256", "accepted", "signature"}
    if not isinstance(data, dict) or set(data) != keys:
        raise ValueError("Reçu invalide.")
    supplied = data.pop("signature")
    if not isinstance(supplied, str) or not re.fullmatch(r"[0-9a-f]{64}", supplied) or not hmac.compare_digest(supplied, signature(config.key, encoded(data), RESPONSE_DIRECTION)):
        raise ValueError("Signature de reçu invalide.")
    if data["protocol"] != RECEIPT_PROTOCOL or data["station_id"] != config.station_id or type(data["sequence"]) is not int or data["sequence"] != sequence or data["accepted"] is not True or data["payload_sha256"] != hashlib.sha256(body).hexdigest():
        raise ValueError("Reçu sans correspondance.")
    return data


def send_once(config, state, *, transport=https_request, resolve=resolve_fixed, now=time.time):
    sequence = state.reserve(now())
    if sequence is None:
        return {"outcome": "waiting"}
    http, dns = measure(target=config.target, transport=transport, resolve=resolve)
    body = encoded({"protocol": PROTOCOL, "station_id": config.station_id, "sequence": sequence, "sent_at": utc(now()), "target": config.target, "http": http, "dns": dns})
    outcome = "unknown"
    try:
        response = transport("POST", config.api_base_url + "/stations/report", body=body, headers={"Content-Type": "application/json", SIGNATURE_HEADER: signature(config.key, body, REPORT_DIRECTION)}, timeout=TIMEOUT)
        if response.status == 200:
            verify_receipt(config, response.body, body, sequence)
            outcome = "accepted"
        else:
            outcome = "rejected"
    except (ValueError, OSError, HTTPException):
        # Le serveur a pu accepter le rapport : aucun rejeu de cette séquence.
        pass
    state.finish(sequence, outcome)
    return {"outcome": outcome, "sequence": sequence}


def main(argv=None):
    parser = argparse.ArgumentParser(description="Sonde volontaire Trinity : une mesure toutes les cinq minutes, cible fixe.")
    parser.add_argument("--config", type=Path, required=True, help="Configuration privée téléchargée depuis Participer.")
    modes = parser.add_mutually_exclusive_group(required=True)
    modes.add_argument("--once", action="store_true", help="Une tentative ; la cadence persistante reste respectée.")
    modes.add_argument("--interval", action="store_true", help="Boucle à la cadence du serveur, sans valeur à fournir.")
    args = parser.parse_args(argv)
    state = None
    try:
        config = Config.load(args.config)
        state = ClientState(args.config.with_suffix(args.config.suffix + ".state.sqlite"), config)
        while True:
            result = send_once(config, state)
            print(json.dumps(result), flush=True)
            if args.once:
                return 0 if result["outcome"] in ("accepted", "waiting") else 1
            time.sleep(max(1, state.next_due() - time.time()))
    except KeyboardInterrupt:
        return 0
    except (OSError, ValueError, sqlite3.Error):
        print("Configuration, état privé ou connexion indisponible ; aucun secret n'est affiché.", flush=True)
        return 1
    finally:
        if state is not None:
            state.close()


if __name__ == "__main__":
    raise SystemExit(main())