box/bin/hatch_menu/tabs/permissions.py

"""Permissions tab: defaults radios, website modes, protocols, advanced.

Contracts live here (single copy): toggles.py references these
constants for addressing. Drills open sub-pages, read, and come back
via the back affordance with tab re-entry fallback. All flows are
ws-level; sessions and error shaping live in toggles.py.
"""
import re
import time

from approvals import cdp_evaluate
from hatch_menu import controls, dialog
from hatch_menu.mouse import escape, real_click

TAB = "Permissions"
ROOT_MARK = "Manage permissions"

CONNECTOR_HEADING = "Connector defaults"
WEB_HEADING = "Web access defaults"
DEFAULT_VALUES = ("auto_allow", "always_ask")

WEBSITE_MODES = ("Allow", "Ask", "Deny")

ADV_LABELS = {"transparent_proxy": "Transparent proxy",
              "tls_interception": "TLS interception",
              "sni_mismatch_rejection": "SNI mismatch rejection"}

# Row titles (first line of each protocol row) pinned live 2026-10-06:
# network primitives on every node checked; MCP titles kept
# defensively in case they appear on other plans/accounts.
PROTOCOL_SLUGS = {"Outbound SSH": "outbound-ssh",
                  "Outgoing email (SMTP)": "smtp",
                  "Email mailbox access (IMAP, POP3)": "imap-pop3",
                  "Database connections": "database",
                  "File transfer (FTP)": "ftp",
                  "External DNS lookups": "dns",
                  "Other TCP connections": "other-tcp",
                  "Other UDP traffic": "other-udp",
                  "Model Context Protocol servers (SSE)": "mcp-sse",
                  "Model Context Protocol servers (Streamable HTTP)":
                  "mcp-streamable",
                  "Agent Skills endpoints": "agent-skills",
                  "MCP Apps (UI extensions)": "mcp-apps",
                  "MCP remote OAuth": "mcp-oauth"}

JS_WEBSITES = """(() => {
  const d = document.querySelector('[role="dialog"]');
  if (!d) return null;
  const out = [];
  for (const b of d.querySelectorAll('button')) {
    const m = (b.getAttribute('aria-label') || '').match(
      /^Change permission mode for (.+),\\s*(Allow|Ask|Deny)$/i);
    if (!m) continue;
    const r = b.getBoundingClientRect();
    out.push({host: m[1].trim(), mode: m[2],
      x: r.x + r.width / 2, y: r.y + r.height / 2});
  }
  return out;
})()"""

JS_MODE_MENU = """(() => {
  return Array.from(document.querySelectorAll('[role="menuitem"]'))
    .map(m => ({text: (m.innerText || '').trim()}));
})()"""

JS_CLICK_MODE = """((mode) => {
  const m = Array.from(document.querySelectorAll('[role="menuitem"]'))
    .find(el => (el.innerText || '').trim() === mode);
  if (!m) return 'NO_MATCH';
  m.click();
  return 'CLICKED';
})('%s')"""

JS_PROTO_ROWS = """(() => {
  const d = document.querySelector('[role="dialog"]');
  if (!d) return null;
  return Array.from(d.querySelectorAll('[role="switch"]')).map(s => {
    let el = s.parentElement, title = '', depth = 0;
    while (el && el !== d && depth < 6) {
      const t = (el.innerText || '').trim().split('\\n')[0] || '';
      if (t) { title = t.slice(0, 80); break; }
      el = el.parentElement;
      depth += 1;
    }
    const r = s.getBoundingClientRect();
    return {title: title,
            checked: s.getAttribute('aria-checked') === 'true',
            x: r.x + r.width / 2, y: r.y + r.height / 2};
  });
})()"""


def _eval(ws, js, timeout=8.0):
    try:
        return cdp_evaluate(ws, js, timeout=timeout)
    except Exception:
        return None


def _stable_rows(ws, js, retries=4, pause=1.5):
    """Repeat a row read until two consecutive reads agree.

    Guards mid-animation partial DOM (innerText shifts while the
    sub-page slides in). Returns the agreed list, or None.
    """
    last = "sentinel"
    for _ in range(retries):
        rows = _eval(ws, js)
        if isinstance(rows, list) and rows == last:
            return rows
        last = rows if isinstance(rows, list) else "sentinel"
        time.sleep(pause)
    return last if isinstance(last, list) else None


def _slug(title):
    """Protocol slug: registry hit, else slugified, else None."""
    if title in PROTOCOL_SLUGS:
        return PROTOCOL_SLUGS[title]
    clean = re.sub(r"[^a-z0-9]+", "-",
                   title.strip().lower()).strip("-")
    return clean or None


def _canon_mode(mode):
    """Canonical Allow/Ask/Deny (case-insensitive); passthrough else."""
    for m in WEBSITE_MODES:
        if (mode or "").lower() == m.lower():
            return m
    return mode


def resolve_protocol(name):
    """Slug/title -> row title, None when unresolvable.

    Exact slug or title first; then a unique case-insensitive
    substring over titles+slugs (so 'ssh' finds Outbound SSH).
    """
    if not isinstance(name, str) or not name.strip():
        return None
    want = name.strip().lower()
    for title, slug in PROTOCOL_SLUGS.items():
        if want == slug or want == title.lower():
            return title
    hits = [t for t, s in PROTOCOL_SLUGS.items()
            if want in t.lower() or want in s]
    if len(hits) == 1:
        return hits[0]
    return None


def _back_to_root(ws):
    """Back affordance, else tab re-entry; verify root text."""
    dialog.go_back(ws)
    if ROOT_MARK in (dialog.dialog_text(ws) or ""):
        return True
    if not dialog.goto_tab(ws, TAB):
        return False
    return ROOT_MARK in (dialog.dialog_text(ws) or "")


def defaults(ws):
    """Connector + web default values (each value or None)."""
    if not dialog.goto_tab(ws, TAB):
        return {"connector_defaults": None, "web_access": None}
    heads = {CONNECTOR_HEADING.lower(): "connector_defaults",
             WEB_HEADING.lower(): "web_access"}
    vals = {CONNECTOR_HEADING.lower(): [], WEB_HEADING.lower(): []}
    for r in controls.list_radios(ws):
        h = (r.get("heading") or "").lower()
        if h in vals and r.get("checked"):
            vals[h].append(r.get("value"))
    out = {}
    for h, key in heads.items():
        out[key] = vals[h][0] if len(vals[h]) == 1 else None
    return out


def set_default(ws, which, value):
    """Set one defaults radio. Bool."""
    if not dialog.goto_tab(ws, TAB):
        return False
    heading = CONNECTOR_HEADING if which == "connector_defaults" \
        else WEB_HEADING
    return controls.set_radio_by_heading(ws, heading, value)


def ensure_advanced(ws):
    """Expand Advanced network settings when collapsed. Bool."""
    if not dialog.goto_tab(ws, TAB):
        return False
    labels = [s.get("label", "") for s in controls.list_switches(ws)]
    if any("Transparent proxy" in lab for lab in labels):
        return True
    if not dialog.click_row(ws, "Advanced network settings", TAB):
        return False
    time.sleep(0.6)
    labels = [s.get("label", "") for s in controls.list_switches(ws)]
    return any("Transparent proxy" in lab for lab in labels)


def advanced(ws):
    """Advanced switch states {key: on/off/None}."""
    if not ensure_advanced(ws):
        return {k: None for k in ADV_LABELS}
    out = {}
    for key, label in ADV_LABELS.items():
        state = None
        for s in controls.list_switches(ws):
            if label.lower() in (s.get("label") or "").lower():
                state = bool(s.get("checked"))
                break
        out[key] = ("on" if state else "off") if state is not None \
            else None
    return out


def set_advanced(ws, key, on):
    """Set one advanced switch. Bool."""
    if key not in ADV_LABELS or not ensure_advanced(ws):
        return False
    return controls.set_switch(ws, ADV_LABELS[key], on)


def _websites_raw(ws):
    """Drill into Websites; rows or None (stays on sub-page)."""
    if not dialog.click_row(ws, "Websites", TAB):
        return None
    return _stable_rows(ws, JS_WEBSITES)


def websites(ws):
    """[{host, mode}] (back at root afterwards)."""
    rows = _websites_raw(ws)
    if rows is None:
        return []
    out = [{"host": r.get("host"), "mode": _canon_mode(r.get("mode"))}
           for r in rows]
    _back_to_root(ws)
    return out


def website_mode(ws, host):
    """Mode for one host, or None when absent/unreadable."""
    for row in websites(ws):
        if (row.get("host") or "").lower() == host.lower():
            return row.get("mode")
    return None


def set_website_mode(ws, host, mode):
    """Set one host mode via the mode chooser. Bool.

    One-way for Ask/Deny: the override row leaves the allowed list
    (no add UI), so removal verifies by absence. No-op when already
    there; absent hosts fail (nothing to click).
    """
    if mode not in WEBSITE_MODES:
        return False
    rows = _websites_raw(ws)
    if rows is None:
        return False
    target = next((r for r in rows
                   if (r.get("host") or "").lower() == host.lower()),
                  None)
    if target is None:
        _back_to_root(ws)
        return False
    if _canon_mode(target.get("mode")) == mode:
        _back_to_root(ws)
        return True
    try:
        real_click(ws, target["x"], target["y"])
    except Exception:
        _back_to_root(ws)
        return False
    time.sleep(0.8)
    items = _eval(ws, JS_MODE_MENU)
    texts = [(i.get("text") or "") for i in items] \
        if isinstance(items, list) else []
    if mode not in texts:
        escape(ws)
        _back_to_root(ws)
        return False
    if _eval(ws, JS_CLICK_MODE % mode) != "CLICKED":
        escape(ws)
        _back_to_root(ws)
        return False
    for _ in range(5):
        time.sleep(2.0)
        rows = _eval(ws, JS_WEBSITES)
        if not isinstance(rows, list):
            continue
        cur = next((_canon_mode(r.get("mode")) for r in rows
                    if (r.get("host") or "").lower() == host.lower()),
                   None)
        if mode in ("Ask", "Deny"):
            if cur is None:
                _back_to_root(ws)
                return True
        elif cur == mode:
            _back_to_root(ws)
            return True
    escape(ws)
    _back_to_root(ws)
    return False


def _protocols_raw(ws):
    """Drill into protocols; rows or None (stays on sub-page)."""
    if not dialog.click_row(ws, "Direct network protocols", TAB):
        return None
    return _stable_rows(ws, JS_PROTO_ROWS)


def protocols(ws):
    """[{slug, title, on}] (back at root afterwards)."""
    rows = _protocols_raw(ws)
    if rows is None:
        return []
    out = [{"slug": _slug(r.get("title", "")),
            "title": r.get("title", ""),
            "on": "on" if r.get("checked") else "off"} for r in rows]
    _back_to_root(ws)
    return out


def protocol_state(ws, title):
    """on/off for one protocol row title, None when absent."""
    for row in protocols(ws):
        if row.get("title") == title:
            return row.get("on")
    return None


def set_protocol(ws, title, on):
    """Set one protocol switch in place; readback before returning."""
    rows = _protocols_raw(ws)
    if rows is None:
        return False
    target = next((r for r in rows if r.get("title") == title), None)
    if target is None:
        _back_to_root(ws)
        return False
    want = bool(on)
    if bool(target.get("checked")) == want:
        _back_to_root(ws)
        return True
    try:
        real_click(ws, target["x"], target["y"])
    except Exception:
        _back_to_root(ws)
        return False
    for _ in range(8):
        time.sleep(2.0)
        rows = _eval(ws, JS_PROTO_ROWS)
        if not isinstance(rows, list):
            continue
        cur = next((r for r in rows if r.get("title") == title), None)
        if cur is not None and bool(cur.get("checked")) == want:
            _back_to_root(ws)
            return True
    _back_to_root(ws)
    # In-dialog verify missed (slow commit or commit-on-close); the
    # toggles-level fresh readback is the source of truth.
    return False


def manage_counts(ws):
    """Manageable-row summary counts (name -> count|None)."""
    if not dialog.goto_tab(ws, TAB):
        return {}
    text = dialog.dialog_text(ws) or ""
    out = {}
    for name in ("Websites", "Connectors", "Scheduled tasks",
                 "Direct network protocols"):
        m = re.search(re.escape(name) + r"\s*(\d+)", text)
        out[name] = int(m.group(1)) if m else None
    return out


def scheduled_tasks(ws):
    """[{name, cadence}] (back at root afterwards; empty when none)."""
    if not dialog.click_row(ws, "Scheduled tasks", TAB):
        return []
    time.sleep(0.6)
    text = dialog.dialog_text(ws) or ""
    _back_to_root(ws)
    rows = []
    lines = [line.strip() for line in text.splitlines()
             if line.strip()]
    for i, line in enumerate(lines):
        if re.search(r"\b(daily|weekly|hourly|every|min)\b", line,
                     re.IGNORECASE) and i > 0:
            rows.append({"name": lines[i - 1], "cadence": line})
    return rows


def all_toggles(ws):
    """Flat map of every settable Permissions toggle (for list)."""
    out = {}
    defs = defaults(ws)
    out["permissions.connector_defaults"] = defs.get("connector_defaults")
    out["permissions.web_access"] = defs.get("web_access")
    adv = advanced(ws)
    for key, val in adv.items():
        out["permissions.advanced." + key] = val
    for row in protocols(ws):
        if row.get("slug"):
            out["permissions.protocols:" + row["slug"]] = row.get("on")
    for row in websites(ws):
        if row.get("host"):
            out["permissions.websites:" + row["host"]] = row.get("mode")
    return out


def describe(ws):
    """Full Permissions inventory: defaults, counts, adv, rows."""
    return {"defaults": defaults(ws), "counts": manage_counts(ws),
            "advanced": advanced(ws), "websites": websites(ws),
            "protocols": protocols(ws),
            "scheduled_tasks": scheduled_tasks(ws)}