#!/usr/bin/env python3 import argparse import sys from urllib.parse import urlparse from cluster_summary import summarize_grafana_cluster from common import ( DEFAULT_CHROME_PROFILE, discover_grafana_dashboard, grafana_session, is_perimeter81_host, parse_grafana_link, probe_grafana_session, write_json, ) def main() -> int: parser = argparse.ArgumentParser( description="Collect normalized access evidence for an online Grafana link" ) parser.add_argument("link", help="Grafana dashboard link or Grafana base URL") parser.add_argument("--dashboard-uid", default="", help="Grafana dashboard uid when link is only the Grafana base URL") parser.add_argument("--profile", default=DEFAULT_CHROME_PROFILE, help="Chrome profile name for reusing Grafana login cookies") parser.add_argument("--datasource-id", default="1", help="Grafana datasource proxy id, for example 1") parser.add_argument("--from", dest="from_ts", default="", help="Optional unix timestamp override for the analysis window start") parser.add_argument("--to", dest="to_ts", default="", help="Optional unix timestamp override for the analysis window end") args = parser.parse_args() grafana_context = parse_grafana_link(args.link, dashboard_uid=args.dashboard_uid or None) if args.from_ts: grafana_context["from"] = int(args.from_ts) grafana_context["from_ms"] = int(args.from_ts) * 1000 if args.to_ts: grafana_context["to"] = int(args.to_ts) grafana_context["to_ms"] = int(args.to_ts) * 1000 base_url = grafana_context["base_url"] is_p81 = is_perimeter81_host(urlparse(base_url).hostname) if base_url else False access_warnings: list[str] = [] # Create session from Chrome DB cookies (with P81 headers auto-added) session = grafana_session( base_url, profile_name=args.profile, path_prefix=str(grafana_context.get("path_prefix") or ""), ) # For P81 hosts, probe session validity before expensive queries. # P81 uses short-lived session cookies; probe detects expired sessions early # by hitting /api/org without redirects. Expired sessions redirect to login page (302). if is_p81: cookies_loaded = getattr(session, "grafana_cookies_loaded", False) probe = probe_grafana_session(session, base_url) if not probe["ok"]: print("", file=sys.stderr) if not cookies_loaded: print("P81 Grafana session failed: no cookies loaded from Chrome DB.", file=sys.stderr) print("", file=sys.stderr) print(" Possible causes:", file=sys.stderr) print(" - Chrome cookies for this host not found in any profile", file=sys.stderr) print(" - macOS Keychain access denied (pycookiecheat cannot decrypt cookies)", file=sys.stderr) print(" - Chrome is not the browser used to access this Grafana", file=sys.stderr) else: print("P81 Grafana session invalid: cookies loaded but session expired server-side.", file=sys.stderr) print(f" Probe detail: {probe.get('detail', 'unknown')}", file=sys.stderr) print("", file=sys.stderr) print(" To fix:", file=sys.stderr) print(" 1. Open " + base_url + " in Chrome (P81 VPN must be connected)", file=sys.stderr) print(" 2. Complete the P81 login flow in your browser", file=sys.stderr) print(" 3. After the page loads, refresh it once", file=sys.stderr) print(" 4. Re-run this script", file=sys.stderr) if not cookies_loaded: print(" 5. If still failing, check that pycookiecheat has Keychain access", file=sys.stderr) print(" (System Settings > Privacy & Security > Full Disk Access)", file=sys.stderr) print("", file=sys.stderr) # Exit early — all subsequent API calls will fail without valid P81 session write_json({ "source": { "kind": "grafana_link", "link": args.link, "base_url": base_url, "dashboard_uid": grafana_context.get("dashboard_uid"), "window": { "from": grafana_context.get("from"), "to": grafana_context.get("to"), }, "ready": False, }, "grafana_context": grafana_context, "access_warnings": [ ("P81 cookies not found in Chrome DB. Check Keychain access and Chrome profile." if not cookies_loaded else "P81 session expired server-side. Re-authenticate in Chrome and re-run.") ], "cluster_summary": None, "dashboard_discovery": None, }) return 1 # Dashboard discovery discovery = None if "/d/" in str(grafana_context.get("path") or "") or str(args.dashboard_uid or "").strip(): try: discovery = discover_grafana_dashboard( args.link, dashboard_uid=args.dashboard_uid or None, profile_name=args.profile, session=session, ) except Exception as exc: import traceback print(f"Dashboard discovery failed: {exc}", file=sys.stderr) traceback.print_exc(file=sys.stderr) access_warnings.append(f"Dashboard discovery failed: {exc}") # Cluster summary (Prometheus queries) cluster_summary = summarize_grafana_cluster( session, datasource_id=args.datasource_id, time_ts=grafana_context.get("to"), ) write_json({ "source": { "kind": "grafana_link", "link": args.link, "base_url": base_url, "dashboard_uid": grafana_context.get("dashboard_uid"), "window": { "from": grafana_context.get("from"), "to": grafana_context.get("to"), }, "ready": True, }, "grafana_context": grafana_context, "access_warnings": access_warnings, "cluster_summary": cluster_summary, "dashboard_discovery": discovery, }) return 0 if __name__ == "__main__": raise SystemExit(main())