box/bin/hatch_menu/toggles.py

"""Toggle registry + sessions over controls.py and tabs/*.

Every settable toggle has ONE address; contracts live in the tab
modules (single-copy per part), addressing here. Set flows read back
through a fresh session and never partially report success.

Caller errors (unknown node/toggle/value) raise MenuError before any
CDP traffic. Transport failures return {"ok": False, ...}.
"""
from approvals import VALID_NODES, get_cdp_ws
from hatch_menu import controls, dialog
from hatch_menu.mouse import close, escape
from hatch_menu.tabs import TAB_MODULES

_PERM = TAB_MODULES["Permissions"]
_DC = TAB_MODULES["Data controls"]
_GEN = TAB_MODULES["General"]


class MenuError(ValueError):
    """Caller error: unknown node, toggle, or value. Raised before CDP."""


TOGGLES = {
    "permissions.connector_defaults": {
        "tab": "Permissions", "kind": "radio-heading",
        "heading": _PERM.CONNECTOR_HEADING,
        "values": _PERM.DEFAULT_VALUES},
    "permissions.web_access": {
        "tab": "Permissions", "kind": "radio-heading",
        "heading": _PERM.WEB_HEADING,
        "values": _PERM.DEFAULT_VALUES},
    "permissions.advanced.transparent_proxy": {
        "tab": "Permissions", "kind": "adv-switch",
        "label": _PERM.ADV_LABELS["transparent_proxy"],
        "values": ("on", "off")},
    "permissions.advanced.tls_interception": {
        "tab": "Permissions", "kind": "adv-switch",
        "label": _PERM.ADV_LABELS["tls_interception"],
        "values": ("on", "off")},
    "permissions.advanced.sni_mismatch_rejection": {
        "tab": "Permissions", "kind": "adv-switch",
        "label": _PERM.ADV_LABELS["sni_mismatch_rejection"],
        "values": ("on", "off")},
    "data_controls.ai_improvement": {
        "tab": "Data controls", "kind": "switch",
        "values": ("on", "off")},
    "general.theme": {
        "tab": "General", "kind": "radio-aria",
        "values": _GEN.THEME_VALUES},
}

WEBSITE_PREFIX = "permissions.websites:"
PROTOCOL_PREFIX = "permissions.protocols:"


def _check_node(node):
    if node not in VALID_NODES:
        raise MenuError("unknown node: %r (valid: %s)"
                        % (node, ", ".join(VALID_NODES)))


def normalize_onoff(value):
    """on/off/true/false/1/0/yes/no -> bool. None when invalid."""
    if isinstance(value, bool):
        return value
    if not isinstance(value, str):
        return None
    v = value.strip().lower()
    if v in ("on", "true", "1", "yes"):
        return True
    if v in ("off", "false", "0", "no"):
        return False
    return None


def resolve_toggle(name):
    """Resolve a toggle address to its spec. Raises MenuError."""
    if not isinstance(name, str) or not name:
        raise MenuError("toggle name must be a non-empty string")
    if name in TOGGLES:
        spec = dict(TOGGLES[name])
        spec["name"] = name
        return spec
    if name.startswith(WEBSITE_PREFIX) and len(name) > len(WEBSITE_PREFIX):
        return {"name": name, "tab": "Permissions", "kind": "website",
                "host": name[len(WEBSITE_PREFIX):],
                "values": _PERM.WEBSITE_MODES}
    if name.startswith(PROTOCOL_PREFIX) and len(name) > len(PROTOCOL_PREFIX):
        label = _PERM.resolve_protocol(name[len(PROTOCOL_PREFIX):])
        if label is None:
            raise MenuError("unknown protocol: %r (see list_toggles)"
                            % (name,))
        return {"name": name, "tab": "Permissions", "kind": "protocol",
                "label": label, "values": ("on", "off")}
    raise MenuError("unknown toggle: %r (see list_toggles)" % (name,))


def _session(node):
    _check_node(node)
    try:
        ws, _ = get_cdp_ws(node)
    except Exception as e:
        raise MenuError("CDP unreachable for %s: %s" % (node, e))
    return ws


def _normalize_value(spec, value):
    kind = spec["kind"]
    if kind == "website":
        if not isinstance(value, str):
            raise MenuError("mode must be one of %s"
                            % (spec["values"],))
        for m in spec["values"]:
            if value.strip().lower() == m.lower():
                return m
        raise MenuError("mode must be one of %s (got %r)"
                        % (spec["values"], value))
    if kind in ("switch", "adv-switch", "protocol"):
        b = normalize_onoff(value)
        if b is None:
            raise MenuError("value must be on/off (got %r)" % (value,))
        return "on" if b else "off"
    if kind == "radio-heading":
        if value not in spec["values"]:
            raise MenuError("%s must be one of %s (got %r)"
                            % (spec["name"], spec["values"], value))
        return value
    if kind == "radio-aria":
        if not isinstance(value, str):
            raise MenuError("theme must be one of %s" % (spec["values"],))
        v = value.strip().lower()
        if v in spec["values"]:
            return v
        raise MenuError("theme must be one of %s (got %r)"
                        % (spec["values"], value))
    raise MenuError("cannot set kind %r" % (kind,))


def get_toggle(node, name):
    """Read one toggle. Returns {"ok", "node", "toggle", "value"}."""
    spec = resolve_toggle(name)
    _check_node(node)
    try:
        ws = _session(node)
    except MenuError as e:
        return {"ok": False, "node": node, "toggle": name,
                "error": str(e)}
    try:
        if not dialog.open_settings(ws):
            return {"ok": False, "node": node, "toggle": name,
                    "error": "settings dialog did not open"}
        if not dialog.goto_tab(ws, spec["tab"]):
            return {"ok": False, "node": node, "toggle": name,
                    "error": "tab did not open: %s" % spec["tab"]}
        kind = spec["kind"]
        if kind == "radio-heading":
            vals = {r["value"]: r["checked"]
                    for r in controls.list_radios(ws)
                    if (r.get("heading") or "").lower()
                    == spec["heading"].lower()}
            on = [v for v, c in vals.items() if c]
            value = on[0] if len(on) == 1 else None
        elif kind == "radio-aria":
            vals = [(r.get("value"), r.get("checked"))
                    for r in controls.list_radios(ws)]
            on = [v for v, c in vals if c]
            value = on[0] if len(on) == 1 else None
        elif kind == "switch":
            state = _DC.ai_improvement(ws)
            value = ("on" if state else "off") if state is not None \
                else None
        elif kind == "adv-switch":
            _PERM.ensure_advanced(ws)
            state = controls.switch_state(ws, spec["label"])
            value = ("on" if state else "off") if state is not None \
                else None
        elif kind == "website":
            value = _PERM.website_mode(ws, spec["host"])
        elif kind == "protocol":
            value = _PERM.protocol_state(ws, spec["label"])
        else:
            value = None
        if value is None:
            if kind == "website":
                return {"ok": False, "node": node, "toggle": name,
                        "error": "host not in Websites list (effective: "
                        "permissions.web_access default)"}
            return {"ok": False, "node": node, "toggle": name,
                    "error": "toggle not readable (site changed?)"}
        return {"ok": True, "node": node, "toggle": name, "value": value}
    except Exception as e:
        return {"ok": False, "node": node, "toggle": name,
                "error": "%s: %s" % (type(e).__name__, e)}
    finally:
        escape(ws)
        close(ws)


def set_toggle(node, name, value):
    """Set one toggle with readback. Never partially reports success."""
    spec = resolve_toggle(name)
    want = _normalize_value(spec, value)
    _check_node(node)
    try:
        ws = _session(node)
    except MenuError as e:
        return {"ok": False, "node": node, "toggle": name,
                "error": str(e)}
    try:
        if not dialog.open_settings(ws):
            return {"ok": False, "node": node, "toggle": name,
                    "error": "settings dialog did not open"}
        if not dialog.goto_tab(ws, spec["tab"]):
            return {"ok": False, "node": node, "toggle": name,
                    "error": "tab did not open: %s" % spec["tab"]}
        kind = spec["kind"]
        if kind == "radio-heading":
            ok = controls.set_radio_by_heading(ws, spec["heading"], want)
        elif kind == "radio-aria":
            ok = controls.set_radio_by_aria(ws, want)
        elif kind == "switch":
            ok = _DC.set_ai_improvement(ws, want == "on")
        elif kind == "adv-switch":
            _PERM.ensure_advanced(ws)
            ok = controls.set_switch(ws, spec["label"], want == "on")
        elif kind == "website":
            ok = _PERM.set_website_mode(ws, spec["host"], want)
        elif kind == "protocol":
            ok = _PERM.set_protocol(ws, spec["label"], want == "on")
        else:
            ok = False
        # The fresh-session readback is the source of truth: switch
        # commits can land slowly or on dialog close, after the
        # in-flow verify had its chance.
        readback = get_toggle(node, name)
        if readback.get("ok") and readback.get("value") == want:
            out = {"ok": True, "node": node, "toggle": name,
                   "value": want}
            if not ok:
                out["readback_only"] = True
            return out
        return {"ok": False, "node": node, "toggle": name,
                "error": "readback mismatch (want %r, got %r)"
                % (want, readback.get("value"))}
    except Exception as e:
        return {"ok": False, "node": node, "toggle": name,
                "error": "%s: %s" % (type(e).__name__, e)}
    finally:
        escape(ws)
        close(ws)


def list_toggles(node, tab=None):
    """All toggle states, optionally filtered to one tab."""
    _check_node(node)
    if tab is not None and tab not in TAB_MODULES:
        raise MenuError("unknown tab: %r (valid: %s)"
                        % (tab, sorted(TAB_MODULES)))
    want = [tab] if tab else ["Permissions", "Data controls", "General"]
    try:
        ws = _session(node)
    except MenuError as e:
        return {"ok": False, "node": node, "error": str(e)}
    try:
        if not dialog.open_settings(ws):
            return {"ok": False, "node": node,
                    "error": "settings dialog did not open"}
        out = {}
        if "Permissions" in want:
            dialog.goto_tab(ws, "Permissions")
            out.update(_PERM.all_toggles(ws))
        if "Data controls" in want:
            state = _DC.ai_improvement(ws)
            out["data_controls.ai_improvement"] = \
                ("on" if state else "off") if state is not None else None
        if "General" in want:
            dialog.goto_tab(ws, "General")
            out["general.theme"] = _GEN.theme(ws)
        return {"ok": True, "node": node, "toggles": out}
    except Exception as e:
        return {"ok": False, "node": node,
                "error": "%s: %s" % (type(e).__name__, e)}
    finally:
        escape(ws)
        close(ws)


def describe_tab(node, tab):
    """Full inventory of one tab (debugging/patching aid)."""
    _check_node(node)
    if tab not in TAB_MODULES:
        raise MenuError("unknown tab: %r (valid: %s)"
                        % (tab, sorted(TAB_MODULES)))
    try:
        ws = _session(node)
    except MenuError as e:
        return {"ok": False, "node": node, "tab": tab, "error": str(e)}
    try:
        if not dialog.open_settings(ws):
            return {"ok": False, "node": node, "tab": tab,
                    "error": "settings dialog did not open"}
        if not dialog.goto_tab(ws, tab):
            return {"ok": False, "node": node, "tab": tab,
                    "error": "tab did not open"}
        return {"ok": True, "node": node, "tab": tab,
                "inventory": TAB_MODULES[tab].describe(ws)}
    except Exception as e:
        return {"ok": False, "node": node, "tab": tab,
                "error": "%s: %s" % (type(e).__name__, e)}
    finally:
        escape(ws)
        close(ws)