#!/usr/bin/env python3
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
#
# #589 — autolearn bad trackers/actors. Builds a HIGH-CONFIDENCE block list
# that ad_ghost consults (in addition to its static ad-host regex), from:
#   1. threat-intel domain IOCs (threatfox malicious C2/malware domains) ;
#   2. cross-site OPERATOR-GRADE / data-broker tracker domains
#      (social_host_meta.opgrade_vendor) seen on >= MIN_SITES sites.
# Deliberately conservative — a plain cross-site CDN (fonts, shared assets)
# is NOT learned, and ANTI-BOT vendors are NOT learned either : a site's own
# WAF (Datadome/PerimeterX) sits in the 1st-party path, so blocking it would
# break the site. So live R3 users don't get legit sites broken. Run hourly
# by secubox-toolbox-autolearn.timer ; output read by ad_ghost (cached).
from __future__ import annotations

import json
import os
import sqlite3
import sys
import time

DB = os.environ.get("SECUBOX_AUTOLEARN_DB", "/var/lib/secubox/toolbox/toolbox.db")
OUT = os.environ.get("SECUBOX_AUTOLEARN_OUT",
                     "/var/lib/secubox/toolbox/learned-trackers.txt")
PURE_OUT = os.environ.get("SECUBOX_AUTOLEARN_PURE_OUT",
                          "/var/lib/secubox/toolbox/pure-trackers.txt")
SPLICE_LEARNED_OUT = os.environ.get(
    "SECUBOX_SPLICE_LEARNED_OUT",
    "/var/lib/secubox/toolbox/splice-learned.txt")
SPLICE_MIN_HITS = int(os.environ.get("SECUBOX_SPLICE_MIN_HITS", "20"))
SPLICE_MAX = 2000
MIN_SITES = 2          # cross-site threshold for operator-grade trackers
MAX_ENTRIES = 8000
# #656 — ad-candidate promotion. #685 hardening: require >= 2 distinct sites
# (was 1 — a single-site host got hard-blocked, e.g. www.google.com → broke
# reCAPTCHA/consent on euronews). Env-overridable.
AD_MIN_SITES = int(os.environ.get("SECUBOX_AD_MIN_SITES", "2"))
AD_ALLOWLIST = os.environ.get("SECUBOX_AD_ALLOWLIST",
                              "/var/lib/secubox/toolbox/ad-allowlist.txt")

# #685 — NEVER-LEARN guard: registrables that host FUNCTIONAL content (CDNs,
# fonts, captcha, auth, OS/payment services). The learner must NEVER 204 these —
# blocking them breaks sites (www.google.com reCAPTCHA/consent broke euronews).
# Checked against the host AND its registrable; existing entries are also pruned.
_NEVER_LEARN_SEED = {
    "google.com", "gstatic.com", "googleapis.com", "googleusercontent.com",
    "googlevideo.com", "ytimg.com", "ggpht.com", "youtube.com", "recaptcha.net",
    "apple.com", "icloud.com", "mzstatic.com", "cdn-apple.com", "cloudflare.com",
    "jsdelivr.net", "jquery.com", "bootstrapcdn.com", "unpkg.com", "cdnjs.com",
    "akamaihd.net", "akamai.net", "fastly.net", "edgekey.net", "edgesuite.net",
    "microsoft.com", "office.com", "live.com", "windows.net", "azureedge.net",
    "msftauth.net", "paypal.com", "paypalobjects.com", "stripe.com",
}
NEVER_LEARN = _NEVER_LEARN_SEED | {
    d.strip().lower()
    for d in os.environ.get("SECUBOX_NEVER_LEARN", "").split(",") if d.strip()
}


def _never_learn(host: str) -> bool:
    h = (host or "").lower().strip(".")
    return bool(h) and (h in NEVER_LEARN or (registrable(h) or h) in NEVER_LEARN)
COOKIE_XSITE_TOP_N = int(os.environ.get("SECUBOX_COOKIE_XSITE_TOP_N", "5"))

sys.path.insert(0, os.environ.get("SECUBOX_TOOLBOX_LIB", "/usr/lib/secubox/toolbox"))
try:
    from secubox_toolbox import learn as _learn
except Exception:       # pragma: no cover - degraded mode
    _learn = None

try:
    from secubox_toolbox import ip_dns as _ip_dns
except Exception:       # pragma: no cover
    _ip_dns = None

UNBOUND_BLOCK_CONF = os.environ.get(
    "SECUBOX_UNBOUND_BLOCK_CONF",
    "/etc/unbound/unbound.conf.d/97-secubox-antitrack-block.conf")

_2L = ("co.uk", "com.au", "co.jp", "co.nz", "com.br", "co.za", "gouv.fr")


def registrable(host: str):
    host = (host or "").split(":")[0].lower().strip(".")
    if not host or host.replace(".", "").isdigit() or ":" in host:
        return None
    p = host.split(".")
    if len(p) <= 2:
        return host
    last2 = ".".join(p[-2:])
    return ".".join(p[-3:]) if (last2 in _2L and len(p) >= 3) else last2


def _dns_feed(pure_hosts) -> int:
    """Write the unbound NXDOMAIN drop-in for pure trackers + reload, gated by
    privacy_enforce && privacy_dns_feed. Returns the zone count written, or -1
    if the gate is off / unavailable (drop-in left untouched)."""
    if _ip_dns is None:
        return -1
    try:
        from secubox_toolbox.filters import get_filters
        f = get_filters()
    except Exception:
        return -1
    if not (f.get("privacy_enforce") and f.get("privacy_dns_feed")):
        return -1
    try:
        lines = _ip_dns.unbound_block_lines(pure_hosts)
        tmp = UNBOUND_BLOCK_CONF + ".tmp"
        with open(tmp, "w", encoding="utf-8") as fh:
            fh.write("\n".join(lines) + "\n")
        os.replace(tmp, UNBOUND_BLOCK_CONF)
    except Exception as e:
        sys.stderr.write(f"autolearn: dns drop-in write failed: {e}\n")
        return -1
    if os.environ.get("SECUBOX_UNBOUND_RELOAD", "1") != "0":
        import subprocess
        try:
            subprocess.run(["unbound-control", "reload"], timeout=10,
                           capture_output=True)
        except Exception:
            try:
                subprocess.run(["systemctl", "reload", "unbound"], timeout=10,
                               capture_output=True)
            except Exception as e:
                sys.stderr.write(f"autolearn: unbound reload failed: {e}\n")
    return sum(1 for l in lines if "local-zone:" in l)


def _splice_feed() -> int:
    """Promote hosts that NEVER served text/html over >= SPLICE_MIN_HITS
    observations into the learned-splice file (registrable-folded, capped).
    Gated: skip when tls_splice == 'off'. Returns count written, or -1 if gated."""
    try:
        from secubox_toolbox.filters import get_filters
        if get_filters().get("tls_splice", "observe") == "off":
            return -1
    except Exception:
        pass
    try:
        con = sqlite3.connect(DB, timeout=5)
        rows = con.execute(
            "SELECT host FROM splice_host_obs WHERE hits >= ? AND html_hits = 0",
            (SPLICE_MIN_HITS,)).fetchall()
        con.close()
    except Exception as e:
        sys.stderr.write(f"autolearn: splice query failed: {e}\n")
        return -1
    hosts = sorted({(r[0] or "").lower().strip(".") for r in rows if r[0]})[:SPLICE_MAX]
    try:
        os.makedirs(os.path.dirname(SPLICE_LEARNED_OUT), exist_ok=True)
        tmp = SPLICE_LEARNED_OUT + ".tmp"
        with open(tmp, "w", encoding="utf-8") as fh:
            fh.write("\n".join(hosts) + ("\n" if hosts else ""))
        os.replace(tmp, SPLICE_LEARNED_OUT)
    except Exception as e:
        sys.stderr.write(f"autolearn: splice write failed: {e}\n")
        return -1
    return len(hosts)


def _load_ad_allowlist() -> set:
    """host + registrable entries from the ad allowlist (comment-stripped)."""
    allow: set = set()
    try:
        if os.path.exists(AD_ALLOWLIST):
            with open(AD_ALLOWLIST, encoding="utf-8") as fh:
                for ln in fh:
                    ln = ln.split("#", 1)[0].strip().lower()
                    if not ln:
                        continue
                    allow.add(ln)
                    reg = registrable(ln)
                    if reg:
                        allow.add(reg)
    except Exception as e:
        sys.stderr.write(f"autolearn: ad allowlist read failed: {e}\n")
    return allow


def _ad_feed() -> int:
    """Promote ad-candidate hosts (seen on >= AD_MIN_SITES distinct sites) into
    the learned-trackers blocklist. Allowlist always wins. MERGES into the
    existing learned-trackers.txt (never overwrites). Gated on filters
    'ad_learn'. Returns count promoted, or -1 if gated/unavailable. Best-effort:
    never raises."""
    try:
        from secubox_toolbox.filters import get_filters
        if not get_filters().get("ad_learn", True):
            return -1
    except Exception:
        pass
    try:
        con = sqlite3.connect(DB, timeout=5)
        rows = con.execute(
            "SELECT host FROM ad_candidates GROUP BY host "
            "HAVING COUNT(DISTINCT site) >= ?", (AD_MIN_SITES,)).fetchall()
        con.close()
    except Exception as e:
        sys.stderr.write(f"autolearn: ad query failed: {e}\n")
        return -1
    allow = _load_ad_allowlist()
    # #658 — never promote the appliance's own domains (the learner once
    # self-promoted secubox.in). Hard default + env-overridable.
    self_doms = {d.strip().lower() for d in
                 os.environ.get("SECUBOX_SELF_DOMAINS", "secubox.in").split(",")
                 if d.strip()}
    promoted: set = set()
    for r in rows:
        h = (r[0] or "").lower().strip(".")
        if not h:
            continue
        reg = registrable(h) or h
        if h in allow or reg in allow:
            continue
        if reg in self_doms or any(h == d or h.endswith("." + d) for d in self_doms):
            continue
        # #685 — never hard-block functional infra (CDN/fonts/captcha/auth).
        if _never_learn(h):
            continue
        # #658 — promote the EXACT host, NOT the registrable: blocking a tracker
        # subdomain (analytics.tiktok.com) must never block the parent site
        # (tiktok.com). Dedicated ad hosts are already registrable-level.
        promoted.add(h)
    # MERGE with existing learned-trackers.txt (union, dedup, cap). #685: also
    # PRUNE any existing never-learn / allowlisted entries already on disk, so a
    # previously mis-learned host (e.g. www.google.com) is cleaned on the next run.
    existing: set = set()
    try:
        if os.path.exists(OUT):
            with open(OUT, encoding="utf-8") as fh:
                for ln in fh:
                    ln = ln.strip()
                    if ln:
                        existing.add(ln)
    except Exception as e:
        sys.stderr.write(f"autolearn: ad merge read failed: {e}\n")
    pruned = {e for e in existing
              if _never_learn(e) or e in allow or (registrable(e) or e) in allow}
    if not promoted and not pruned:
        return 0
    merged = sorted((existing - pruned) | promoted)[:MAX_ENTRIES]
    try:
        os.makedirs(os.path.dirname(OUT), exist_ok=True)
        tmp = OUT + ".tmp"
        with open(tmp, "w", encoding="utf-8") as fh:
            fh.write("\n".join(merged) + ("\n" if merged else ""))
        os.replace(tmp, OUT)
    except Exception as e:
        sys.stderr.write(f"autolearn: ad write failed: {e}\n")
        return -1
    return len(promoted)


# #662 — cross-site-reuse promotion. A tracker_domain seen issuing cookies on
# >= SOCIAL_MIN_SITES DISTINCT src_site (across peers, recent window) is a
# BEHAVIOURALLY-confirmed cross-site tracker (the social graph), independent of
# the ad-path heuristic. Promote it into learned-trackers.txt so the engine
# blocks (204) + smogs it. Conservative + reuses the SAME allowlist/self guard as
# _ad_feed (NEVER promote allowlisted or self domains). De-dups against OUT.
SOCIAL_MIN_SITES = int(os.environ.get("SECUBOX_SOCIAL_MIN_SITES", "3"))
SOCIAL_WINDOW_HOURS = int(os.environ.get("SECUBOX_SOCIAL_WINDOW_HOURS", "168"))


def _social_feed() -> int:
    """Promote cross-site cookie-reuse trackers (social_edges) into the learned
    blocklist. A tracker_domain linking >= SOCIAL_MIN_SITES distinct src_site in
    the last SOCIAL_WINDOW_HOURS is promoted. Allowlist + self domains excluded
    (reused guard). MERGES into OUT (never overwrites). Returns count promoted, or
    -1 if unavailable (e.g. no social_edges table). Best-effort: never raises."""
    cutoff = int(time.time()) - SOCIAL_WINDOW_HOURS * 3600
    try:
        con = sqlite3.connect(DB, timeout=5)
        rows = con.execute(
            "SELECT tracker_domain, COUNT(DISTINCT src_site) AS sites "
            "FROM social_edges WHERE ts >= ? "
            "GROUP BY tracker_domain", (cutoff,)).fetchall()
        con.close()
    except Exception as e:
        sys.stderr.write(f"autolearn: social query failed: {e}\n")
        return -1
    # Fold to registrable and aggregate the distinct-site count per eTLD+1 (two
    # tracker subdomains of the same registrable jointly meet the threshold).
    by_reg: dict[str, set] = {}
    try:
        scon = sqlite3.connect(DB, timeout=5)
        for td, _sites in rows:
            reg = registrable(td)
            if not reg:
                continue
            ss = by_reg.setdefault(reg, set())
            for (s,) in scon.execute(
                "SELECT DISTINCT src_site FROM social_edges "
                "WHERE ts >= ? AND tracker_domain = ?", (cutoff, td)):
                if s:
                    ss.add(s)
        scon.close()
    except Exception as e:
        sys.stderr.write(f"autolearn: social fold failed: {e}\n")
        return -1

    allow = _load_ad_allowlist()
    self_doms = {d.strip().lower() for d in
                 os.environ.get("SECUBOX_SELF_DOMAINS", "secubox.in").split(",")
                 if d.strip()}
    promoted: set = set()
    for reg, sites in by_reg.items():
        if len(sites) < SOCIAL_MIN_SITES:
            continue
        if reg in allow:
            continue
        if reg in self_doms or any(reg == d or reg.endswith("." + d) for d in self_doms):
            continue
        promoted.add(reg)
    if not promoted:
        return 0
    existing: set = set()
    try:
        if os.path.exists(OUT):
            with open(OUT, encoding="utf-8") as fh:
                for ln in fh:
                    ln = ln.strip()
                    if ln:
                        existing.add(ln)
    except Exception as e:
        sys.stderr.write(f"autolearn: social merge read failed: {e}\n")
    new = promoted - existing
    merged = sorted(existing | promoted)[:MAX_ENTRIES]
    try:
        os.makedirs(os.path.dirname(OUT), exist_ok=True)
        tmp = OUT + ".tmp"
        with open(tmp, "w", encoding="utf-8") as fh:
            fh.write("\n".join(merged) + ("\n" if merged else ""))
        os.replace(tmp, OUT)
    except Exception as e:
        sys.stderr.write(f"autolearn: social write failed: {e}\n")
        return -1
    return len(new)


def main() -> int:
    learned: set[str] = set()
    try:
        c = sqlite3.connect(DB, timeout=10)
        c.row_factory = sqlite3.Row
    except Exception as e:
        sys.stderr.write(f"autolearn: cannot open {DB}: {e}\n")
        return 0

    # 1) threat-intel malicious domains (high confidence).
    try:
        for r in c.execute("SELECT DISTINCT ioc FROM threat_intel WHERE type='domain'"):
            d = registrable(r["ioc"])
            if d:
                learned.add(d)
    except Exception:
        pass
    ti = len(learned)

    # 2) cross-site OPERATOR-GRADE / data-broker trackers ONLY. Anti-bot
    # vendors are deliberately excluded — they're frequently the visited
    # site's own WAF (in-path), so blocking them breaks the page.
    try:
        classified = set()
        for r in c.execute(
            "SELECT tracker_domain FROM social_host_meta "
            "WHERE opgrade_vendor IS NOT NULL"):
            d = registrable(r["tracker_domain"])
            if d:
                classified.add(d)
        # distinct 1st-party sites per registrable tracker domain
        sites: dict[str, set] = {}
        for r in c.execute("SELECT tracker_domain, sites_jsonl FROM social_nodes"):
            d = registrable(r["tracker_domain"])
            if not d or d not in classified:
                continue
            try:
                for s in json.loads(r["sites_jsonl"] or "[]"):
                    sites.setdefault(d, set()).add(s)
            except Exception:
                pass
        for d, ss in sites.items():
            if len(ss) >= MIN_SITES:
                learned.add(d)
    except Exception:
        pass

    # 3) NEW (#633): cross-site pre-consent cookie trackers, top-N capped.
    if _learn is not None:
        try:
            for d in _learn.cookie_xsite_trackers(c, top_n=COOKIE_XSITE_TOP_N):
                learned.add(d)
        except Exception:
            pass
    # compute pure-trackers (seed + conservative auto-promote) while DB open.
    pure: set = set(_learn.PURE_SEED) if _learn is not None else set()
    if _learn is not None:
        try:
            pure = _learn.pure_trackers(c, learned=learned, seed=_learn.PURE_SEED)
        except Exception:
            pass

    c.close()
    learned.discard(None)
    out = sorted(learned)[:MAX_ENTRIES]
    try:
        tmp = OUT + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            f.write("\n".join(out) + ("\n" if out else ""))
        os.replace(tmp, OUT)
    except Exception as e:
        sys.stderr.write(f"autolearn: write failed: {e}\n")
        return 0
    # Only (re)write pure-trackers.txt when the learn module is available; if
    # the import failed, leave the previous file intact rather than zeroing the
    # hard-block allowlist (which would silently disable seed hard-blocks).
    pout = sorted(x for x in pure if x)
    if _learn is not None:
        try:
            ptmp = PURE_OUT + ".tmp"
            with open(ptmp, "w", encoding="utf-8") as f:
                f.write("\n".join(pout) + ("\n" if pout else ""))
            os.replace(ptmp, PURE_OUT)
        except Exception as e:
            sys.stderr.write(f"autolearn: pure write failed: {e}\n")
    dns_zones = _dns_feed(pout)
    try:
        _n_splice = _splice_feed()
        sys.stderr.write(f"autolearn: {_n_splice} splice hosts learned\n")
    except Exception as e:
        sys.stderr.write(f"autolearn: splice feed error: {e}\n")
    try:
        _n_ad = _ad_feed()
        sys.stderr.write(f"autolearn: {_n_ad} ad-candidate hosts promoted\n")
    except Exception as e:
        sys.stderr.write(f"autolearn: ad feed error: {e}\n")
    try:
        _n_social = _social_feed()
        sys.stderr.write(f"autolearn: {_n_social} cross-site cookie trackers promoted\n")
    except Exception as e:
        sys.stderr.write(f"autolearn: social feed error: {e}\n")
    sys.stderr.write(
        f"autolearn: {len(out)} hosts learned ({ti} threat-intel + "
        f"{len(out) - ti} classified cross-site) @ {int(time.time())}"
        f" ; {len(pout)} pure ; dns_feed={dns_zones}\n")
    return 0


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