384 lines
12 KiB
Python
384 lines
12 KiB
Python
|
|
"""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"}
|
||
|
|
|
||
|
|
PROTOCOL_SLUGS = {"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;
|
||
|
|
return Array.from(d.querySelectorAll('button')).filter(b =>
|
||
|
|
['allow', 'ask', 'deny'].includes((b.getAttribute('aria-label') || '')
|
||
|
|
.trim().toLowerCase())).map(b => {
|
||
|
|
let el = b.parentElement, host = '', depth = 0;
|
||
|
|
while (el && el !== d && depth < 6) {
|
||
|
|
const t = (el.innerText || '').trim().split('\\n')[0] || '';
|
||
|
|
if (t && t.includes('.') && t.length < 120) { host = t; break; }
|
||
|
|
el = el.parentElement;
|
||
|
|
depth += 1;
|
||
|
|
}
|
||
|
|
return {host: host, mode: (b.getAttribute('aria-label') || '').trim(),
|
||
|
|
x: b.getBoundingClientRect().x + b.getBoundingClientRect().width / 2,
|
||
|
|
y: b.getBoundingClientRect().y + b.getBoundingClientRect().height / 2};
|
||
|
|
});
|
||
|
|
})()"""
|
||
|
|
|
||
|
|
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_MODE_RECT = """((mode) => {
|
||
|
|
const m = Array.from(document.querySelectorAll('[role="menuitem"]'))
|
||
|
|
.find(el => (el.innerText || '').trim() === mode);
|
||
|
|
if (!m) return null;
|
||
|
|
const r = m.getBoundingClientRect();
|
||
|
|
return {x: r.x + r.width / 2, y: r.y + r.height / 2};
|
||
|
|
})('%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, label = '', depth = 0;
|
||
|
|
while (el && el !== d && depth < 6) {
|
||
|
|
const t = (el.innerText || '').trim().replace(/\\s+/g, ' ');
|
||
|
|
if (t && t.length < 250) { label = t; break; }
|
||
|
|
el = el.parentElement;
|
||
|
|
depth += 1;
|
||
|
|
}
|
||
|
|
const r = s.getBoundingClientRect();
|
||
|
|
return {label: label.slice(0, 120),
|
||
|
|
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 _slug(label):
|
||
|
|
"""Protocol slug: registry hit, else slugified, else None."""
|
||
|
|
if label in PROTOCOL_SLUGS:
|
||
|
|
return PROTOCOL_SLUGS[label]
|
||
|
|
clean = re.sub(r"[^a-z0-9]+", "-",
|
||
|
|
label.strip().lower()).strip("-")
|
||
|
|
return clean or None
|
||
|
|
|
||
|
|
|
||
|
|
def resolve_protocol(name):
|
||
|
|
"""Slug or label fragment -> row label, None when unresolvable."""
|
||
|
|
if not isinstance(name, str) or not name.strip():
|
||
|
|
return None
|
||
|
|
want = name.strip().lower()
|
||
|
|
for label, slug in PROTOCOL_SLUGS.items():
|
||
|
|
if want == slug or want == label.lower():
|
||
|
|
return label
|
||
|
|
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
|
||
|
|
time.sleep(0.6)
|
||
|
|
return _eval(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": 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. Verify + readback. Bool."""
|
||
|
|
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
|
||
|
|
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) or []
|
||
|
|
texts = [(i.get("text") or "") for i in items]
|
||
|
|
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
|
||
|
|
time.sleep(0.6)
|
||
|
|
rows = _eval(ws, JS_WEBSITES) or []
|
||
|
|
cur = next(((r.get("mode")) for r in rows
|
||
|
|
if (r.get("host") or "").lower() == host.lower()),
|
||
|
|
None)
|
||
|
|
if cur == mode:
|
||
|
|
_back_to_root(ws)
|
||
|
|
return True
|
||
|
|
rect = _eval(ws, JS_MODE_RECT % mode)
|
||
|
|
if rect and "x" in rect:
|
||
|
|
try:
|
||
|
|
real_click(ws, rect["x"], rect["y"])
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
time.sleep(0.6)
|
||
|
|
rows = _eval(ws, JS_WEBSITES) or []
|
||
|
|
cur = next(((r.get("mode")) for r in rows
|
||
|
|
if (r.get("host") or "").lower() == host.lower()),
|
||
|
|
None)
|
||
|
|
if 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
|
||
|
|
time.sleep(0.6)
|
||
|
|
return _eval(ws, JS_PROTO_ROWS)
|
||
|
|
|
||
|
|
|
||
|
|
def protocols(ws):
|
||
|
|
"""[{slug, label, on}] (back at root afterwards)."""
|
||
|
|
rows = _protocols_raw(ws)
|
||
|
|
if rows is None:
|
||
|
|
return []
|
||
|
|
out = [{"slug": _slug(r.get("label", "")),
|
||
|
|
"label": r.get("label", ""),
|
||
|
|
"on": "on" if r.get("checked") else "off"} for r in rows]
|
||
|
|
_back_to_root(ws)
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def protocol_state(ws, label):
|
||
|
|
"""on/off for one protocol row label, None when absent."""
|
||
|
|
for row in protocols(ws):
|
||
|
|
if row.get("label") == label:
|
||
|
|
return row.get("on")
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def set_protocol(ws, label, 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("label") == label), 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
|
||
|
|
time.sleep(0.6)
|
||
|
|
rows = _eval(ws, JS_PROTO_ROWS) or []
|
||
|
|
cur = next((r for r in rows if r.get("label") == label), None)
|
||
|
|
ok = cur is not None and bool(cur.get("checked")) == want
|
||
|
|
_back_to_root(ws)
|
||
|
|
return ok
|
||
|
|
|
||
|
|
|
||
|
|
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)}
|