feat(toolbox): cookie_xsite_detail aggregation over social_edges (ref #749)

Add _xsite_detail_from_conn() and cookie_xsite_detail() to social.py,
detecting (tracker_domain, cookie_id_hash) pairs reused across >=2 distinct
first-party sites. Mirrors aggregate() envelope. 7 tests green.

packages/secubox-toolbox/secubox_toolbox/social.py

@@ -1139,6 +1139,89 @@ def aggregate(hours: int = 24) -> Dict:
     return out
 
 
+def _xsite_detail_from_conn(conn, since: int, top_n: int) -> list:
+    """Pure cross-site tracker detail over a social_edges connection.
+
+    A (tracker_domain, cookie_id_hash) pair is cross-site when its cookie id is
+    observed on >= 2 DISTINCT valid src_sites (src_site not in '', 'null') within
+    the window (ts >= since). For every such pair, aggregate per REGISTRABLE
+    tracker domain (IP literals dropped). Ranked by client_count, then
+    site_count, then domain; capped to top_n.
+    """
+    rows = conn.execute(
+        "SELECT ts, client_mac_hash, src_site, tracker_domain, "
+        "       cookie_id_hash, consent_state "
+        "FROM social_edges "
+        "WHERE ts >= ? "
+        "  AND cookie_id_hash IS NOT NULL AND cookie_id_hash <> '' "
+        "  AND src_site NOT IN ('', 'null') "
+        "LIMIT 50000",
+        (since,),
+    ).fetchall()
+
+    # Pass 1: which (raw tracker_domain, cookie_id_hash) pairs are cross-site.
+    sites_per_pair: dict = {}
+    for r in rows:
+        key = (r["tracker_domain"], r["cookie_id_hash"])
+        sites_per_pair.setdefault(key, set()).add(r["src_site"])
+    xsite_pairs = {k for k, s in sites_per_pair.items() if len(s) >= 2}
+    if not xsite_pairs:
+        return []
+
+    # Pass 2: aggregate the cross-site rows per registrable tracker domain.
+    agg: dict = {}
+    for r in rows:
+        if (r["tracker_domain"], r["cookie_id_hash"]) not in xsite_pairs:
+            continue
+        dom = _registrable_domain(r["tracker_domain"])
+        if not dom or _is_ip(dom):
+            continue
+        e = agg.setdefault(dom, {
+            "tracker_domain": dom, "sites": set(), "clients": set(),
+            "cookies": set(), "pre_consent_hits": 0, "last_seen": 0,
+        })
+        e["sites"].add(r["src_site"])
+        e["clients"].add(r["client_mac_hash"])
+        e["cookies"].add(r["cookie_id_hash"])
+        if r["consent_state"] == "pre_consent":
+            e["pre_consent_hits"] += 1
+        if r["ts"] > e["last_seen"]:
+            e["last_seen"] = r["ts"]
+
+    out = [{
+        "tracker_domain": e["tracker_domain"],
+        "sites": sorted(e["sites"]),
+        "site_count": len(e["sites"]),
+        "client_count": len(e["clients"]),
+        "cookie_count": len(e["cookies"]),
+        "pre_consent_hits": e["pre_consent_hits"],
+        "last_seen": e["last_seen"],
+    } for e in agg.values()]
+    out.sort(key=lambda t: (-t["client_count"], -t["site_count"],
+                            t["tracker_domain"]))
+    return out[:max(0, top_n)]
+
+
+def cookie_xsite_detail(hours: int = 24, top_n: int = 50) -> Dict:
+    """Operator view of cross-site tracker cookies over social_edges.
+
+    Mirrors aggregate()'s envelope shape. JWT-gated in the API layer.
+    """
+    if hours < 1 or hours > 24 * 31:
+        hours = 24
+    if top_n < 1 or top_n > 500:
+        top_n = 50
+    now = int(time.time())
+    since = now - hours * 3600
+    out: Dict = {"window_hours": hours, "generated_at": now, "trackers": []}
+    try:
+        with _conn() as c:
+            out["trackers"] = _xsite_detail_from_conn(c, since, top_n)
+    except sqlite3.Error as e:
+        log.warning("cookie_xsite_detail: DB error, returning empty: %s", e)
+    return out
+
+
 def evidence(mac_hash: str, since_seconds: int = 86400) -> Dict:
     """Phase 11.C evidence helper — returns the legal-grade slice
     consumed by the bilingual PDF report.

packages/secubox-toolbox/tests/test_cookie_xsite_detail.py

@@ -0,0 +1,105 @@
+# SPDX-License-Identifier: LicenseRef-CMSD-1.0
+"""Tests for social.cookie_xsite_detail / _xsite_detail_from_conn (ref #749)."""
+import sqlite3
+from secubox_toolbox import social
+
+
+def _edges_db():
+    c = sqlite3.connect(":memory:")
+    c.row_factory = sqlite3.Row
+    c.executescript("""
+        CREATE TABLE social_edges (
+            ts INTEGER, client_mac_hash TEXT, src_site TEXT,
+            tracker_domain TEXT, cookie_id_hash TEXT, ja4_hash TEXT,
+            consent_state TEXT DEFAULT 'none_seen');
+    """)
+    return c
+
+
+def _add(c, ts, client, site, tracker, cid, consent="pre_consent"):
+    c.execute("INSERT INTO social_edges(ts,client_mac_hash,src_site,"
+              "tracker_domain,cookie_id_hash,ja4_hash,consent_state) "
+              "VALUES (?,?,?,?,?,'ja4',?)",
+              (ts, client, site, tracker, cid, consent))
+
+
+def test_crosssite_tracker_detected_with_detail():
+    c = _edges_db()
+    # same cookie id reused across 2 distinct sites -> cross-site
+    _add(c, 100, "m1", "news.example", "www.criteo.com", "CID1")
+    _add(c, 200, "m2", "shop.example2", "www.criteo.com", "CID1", consent="post_consent")
+    c.commit()
+    rows = social._xsite_detail_from_conn(c, since=0, top_n=50)
+    assert len(rows) == 1
+    t = rows[0]
+    assert t["tracker_domain"] == "criteo.com"
+    assert t["site_count"] == 2
+    assert sorted(t["sites"]) == ["news.example", "shop.example2"]
+    assert t["client_count"] == 2
+    assert t["cookie_count"] == 1
+    assert t["pre_consent_hits"] == 1
+    assert t["last_seen"] == 200
+
+
+def test_single_site_cookie_ignored():
+    c = _edges_db()
+    _add(c, 100, "m1", "news.example", "tracker.foo", "CID2")
+    _add(c, 110, "m1", "news.example", "tracker.foo", "CID2")
+    c.commit()
+    assert social._xsite_detail_from_conn(c, since=0, top_n=50) == []
+
+
+def test_null_and_empty_src_site_excluded():
+    c = _edges_db()
+    _add(c, 100, "m1", "null", "t.bar", "CID3")
+    _add(c, 110, "m1", "", "t.bar", "CID3")
+    _add(c, 120, "m1", "real.site", "t.bar", "CID3")
+    c.commit()
+    # only one VALID site remains for CID3 -> not cross-site
+    assert social._xsite_detail_from_conn(c, since=0, top_n=50) == []
+
+
+def test_window_filters_old_edges():
+    c = _edges_db()
+    _add(c, 100, "m1", "a.example", "t.win", "CIDW")
+    _add(c, 200, "m1", "b.example2", "t.win", "CIDW")
+    c.commit()
+    assert social._xsite_detail_from_conn(c, since=150, top_n=50) == []
+
+
+def test_ip_literal_tracker_dropped():
+    c = _edges_db()
+    _add(c, 100, "m1", "a.example", "192.0.2.5", "CIDIP")
+    _add(c, 200, "m1", "b.example2", "192.0.2.5", "CIDIP")
+    c.commit()
+    assert social._xsite_detail_from_conn(c, since=0, top_n=50) == []
+
+
+def test_ranking_and_top_n_cap():
+    c = _edges_db()
+    # tracker A: 2 clients ; tracker B: 1 client -> A ranks first
+    _add(c, 100, "m1", "s1.x", "a.trk", "A1"); _add(c, 110, "m2", "s2.x", "a.trk", "A1")
+    _add(c, 120, "m1", "s1.x", "b.trk", "B1"); _add(c, 130, "m1", "s2.x", "b.trk", "B1")
+    c.commit()
+    rows = social._xsite_detail_from_conn(c, since=0, top_n=1)
+    assert len(rows) == 1
+    assert rows[0]["tracker_domain"] == "a.trk"  # registrable of a.trk (_registrable_domain returns last two labels)
+
+
+def test_envelope_shape_via_conn(monkeypatch):
+    c = _edges_db()
+    _add(c, 100, "m1", "news.example", "www.criteo.com", "CID1")
+    _add(c, 200, "m2", "shop.example2", "www.criteo.com", "CID1")
+    c.commit()
+
+    class _Ctx:
+        def __enter__(self): return c
+        def __exit__(self, *a): return False
+
+    # Freeze time to 300 so since = 300 - 24*3600 < 0, letting ts=100/200 through.
+    monkeypatch.setattr(social.time, "time", lambda: 300)
+    monkeypatch.setattr(social, "_conn", lambda: _Ctx())
+    out = social.cookie_xsite_detail(hours=24, top_n=50)
+    assert out["window_hours"] == 24
+    assert isinstance(out["generated_at"], int)
+    assert out["trackers"][0]["tracker_domain"] == "criteo.com"