MISF-Planner-Download/scripts/diag_login.py

#!/usr/bin/env python3
"""
Diagnostic: instrumented Microsoft login for the Planner HTTP download.

Mirrors ``_playwright_login_and_capture_token`` in onedrive_planner_http.py but
records everything that function is blind to:

  * every request the browser context makes (host, path, whether it carries a
    Bearer token)                                   -> diag/requests.txt
  * page URL / title / visible text after each step -> diag/NN_<step>.txt
  * screenshot + full HTML after each step          -> diag/NN_<step>.png/.html
  * the SPA's localStorage keys and any MSAL access
    tokens sitting in it                            -> diag/msal.json

Run:  .venv\\Scripts\\python scripts\\diag_login.py [--headless] [--wait 180]
"""
from __future__ import annotations

import argparse
import json
import sys
import time
from datetime import datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
SRC = ROOT / "src"
for p in (str(ROOT), str(SRC)):
    if p not in sys.path:
        sys.path.insert(0, p)

import pyotp  # noqa: E402
from playwright.sync_api import sync_playwright  # noqa: E402

from config import read_app_config  # noqa: E402

DIAG = ROOT / "diag" / datetime.now().strftime("%Y%m%d_%H%M%S")


class _Recorder:
    """Records every request; flags the ones carrying an Authorization header."""

    def __init__(self) -> None:
        self.rows: list[tuple[float, str, bool, str]] = []
        self.t0 = time.monotonic()
        self.bearer_hosts: dict[str, int] = {}

    def on_request(self, request) -> None:
        auth = request.headers.get("authorization", "")
        has_bearer = auth.startswith("Bearer ")
        url = request.url
        self.rows.append((time.monotonic() - self.t0, url, has_bearer, request.method))
        if has_bearer:
            host = url.split("/")[2] if "//" in url else url
            self.bearer_hosts[host] = self.bearer_hosts.get(host, 0) + 1

    def dump(self, path: Path) -> None:
        lines = [
            f"{t:7.1f}s  {'BEARER' if b else '      '}  {m:6s}  {u}"
            for t, u, b, m in self.rows
        ]
        path.write_text("\n".join(lines), encoding="utf-8", errors="replace")


_step_no = 0


def snapshot(page, label: str) -> None:
    """Dump URL, title, visible text, HTML and a screenshot for one step."""
    global _step_no
    _step_no += 1
    stem = DIAG / f"{_step_no:02d}_{label}"
    try:
        url, title = page.url, page.title()
    except Exception as exc:  # page closed / navigating
        url, title = f"<unavailable: {exc}>", ""
    text = ""
    try:
        text = page.inner_text("body")[:4000]
    except Exception as exc:
        text = f"<no body text: {exc}>"
    stem.with_suffix(".txt").write_text(
        f"URL   : {url}\nTITLE : {title}\n\n--- visible text ---\n{text}\n",
        encoding="utf-8",
        errors="replace",
    )
    try:
        stem.with_suffix(".html").write_text(page.content(), encoding="utf-8", errors="replace")
    except Exception:
        pass
    try:
        page.screenshot(path=str(stem.with_suffix(".png")), full_page=True)
    except Exception:
        pass
    print(f"\n[{_step_no:02d}] {label}")
    print(f"     url   : {url}")
    print(f"     title : {title}")
    first = " | ".join(ln.strip() for ln in text.splitlines() if ln.strip())[:300]
    print(f"     text  : {first}")


def dump_storage(page) -> None:
    """Dump localStorage keys and any MSAL access-token entries."""
    try:
        data = page.evaluate(
            """() => {
                const out = {};
                for (let i = 0; i < localStorage.length; i++) {
                    const k = localStorage.key(i);
                    out[k] = localStorage.getItem(k);
                }
                return out;
            }"""
        )
    except Exception as exc:
        print(f"     localStorage unreadable: {exc}")
        return

    summary = {}
    for key, raw in data.items():
        entry = {"len": len(raw or "")}
        if "accesstoken" in key.lower():
            try:
                parsed = json.loads(raw)
                entry["target"] = parsed.get("target")
                entry["realm"] = parsed.get("realm")
                entry["clientId"] = parsed.get("clientId")
                entry["expiresOn"] = parsed.get("expiresOn")
                entry["secret_len"] = len(parsed.get("secret", "") or "")
            except Exception:
                entry["parse"] = "failed"
        summary[key] = entry

    (DIAG / "msal.json").write_text(
        json.dumps(summary, indent=2), encoding="utf-8", errors="replace"
    )
    tokens = [k for k in summary if "accesstoken" in k.lower()]
    print(f"     localStorage keys: {len(summary)}, MSAL access-token entries: {len(tokens)}")
    for k in tokens:
        print(f"       - target={summary[k].get('target')!r} secret_len={summary[k].get('secret_len')}")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--headless", action="store_true", help="run headless (default: visible)")
    ap.add_argument("--wait", type=float, default=180, help="seconds to observe after login")
    args = ap.parse_args()

    DIAG.mkdir(parents=True, exist_ok=True)
    print(f"Diagnostics -> {DIAG}")

    cfg = read_app_config()

    proxy_cfg = None
    if cfg.proxy_ip and cfg.proxy_port:
        proxy_cfg = {"server": f"http://{cfg.proxy_ip}:{cfg.proxy_port}"}
        if cfg.proxy_username and cfg.proxy_password:
            proxy_cfg["username"] = cfg.proxy_username
            proxy_cfg["password"] = str(cfg.proxy_password)
    print(f"Proxy: {proxy_cfg}")

    rec = _Recorder()

    with sync_playwright() as pw:
        browser = pw.chromium.launch(headless=args.headless, proxy=proxy_cfg)
        ctx = browser.new_context(
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/124.0.0.0 Safari/537.36"
            )
        )
        ctx.on("request", rec.on_request)
        page = ctx.new_page()

        planner_url = cfg.planner_url or "https://planner.cloud.microsoft/"
        print(f"Navigating: {planner_url}")
        page.goto(planner_url, timeout=60_000)
        snapshot(page, "after_goto")

        # --- username ---
        try:
            page.wait_for_selector("input[type='email'], input[name='loginfmt']", timeout=30_000)
            page.fill("input[type='email'], input[name='loginfmt']", cfg.onedrive_username)
            page.click("input[type='submit'], button[type='submit']")
            snapshot(page, "after_username")
        except Exception as exc:
            print(f"     username step: {exc}")
            snapshot(page, "username_failed")

        # --- password ---
        try:
            page.wait_for_selector("input[type='password'], input[name='passwd']", timeout=20_000)
            page.fill("input[type='password'], input[name='passwd']", cfg.onedrive_password)
            page.click("input[type='submit'], button[type='submit']")
            time.sleep(3)
            snapshot(page, "after_password")
        except Exception as exc:
            print(f"     password step: {exc}")
            snapshot(page, "password_failed")

        # --- TOTP ---
        try:
            page.wait_for_selector(
                "input[name='otc'], input[autocomplete='one-time-code']", timeout=15_000
            )
            snapshot(page, "otp_prompt")
            otp = pyotp.TOTP(cfg.microsoft_totp_secret).now()
            page.fill("input[name='otc'], input[autocomplete='one-time-code']", otp)
            page.click("input[type='submit'], button[type='submit']")
            print(f"     TOTP submitted ({otp})")
            time.sleep(5)
            snapshot(page, "after_totp")
        except Exception as exc:
            print(f"     no TOTP prompt: {exc}")
            snapshot(page, "no_otp_prompt")

        # --- whatever comes next: DO NOT click anything, just observe ---
        for wait_s in (5, 10, 20):
            time.sleep(wait_s)
            snapshot(page, f"observe_t{wait_s}")

        # Report what the current page offers as clickable controls, so we can
        # see exactly how the "Stay signed in?" prompt is rendered now.
        try:
            controls = page.evaluate(
                """() => Array.from(
                    document.querySelectorAll('button, input[type=submit], input[type=button], a[role=button]')
                ).slice(0, 40).map(e => ({
                    tag: e.tagName, id: e.id, type: e.type || '',
                    value: e.value || '', text: (e.innerText || '').trim().slice(0, 60),
                    visible: !!(e.offsetWidth || e.offsetHeight)
                }))"""
            )
            (DIAG / "controls.json").write_text(
                json.dumps(controls, indent=2), encoding="utf-8", errors="replace"
            )
            print("\n     clickable controls on current page:")
            for c in controls:
                if c["visible"]:
                    print(f"       {c['tag']}#{c['id'] or '-'} type={c['type']!r} "
                          f"value={c['value']!r} text={c['text']!r}")
        except Exception as exc:
            print(f"     control scan failed: {exc}")

        dump_storage(page)

        # --- observe for the remaining budget, watching for a Graph bearer ---
        deadline = time.monotonic() + args.wait
        seen_graph = False
        while time.monotonic() < deadline:
            graph = [r for r in rec.rows if "graph.microsoft.com" in r[1] and r[2]]
            if graph and not seen_graph:
                seen_graph = True
                print(f"\n     *** first Graph bearer request at {graph[0][0]:.1f}s: {graph[0][1][:140]}")
            time.sleep(2)

        snapshot(page, "final")
        dump_storage(page)

        rec.dump(DIAG / "requests.txt")

        print("\n" + "=" * 70)
        print(f"Total requests recorded : {len(rec.rows)}")
        print("Hosts that received a Bearer token:")
        for host, n in sorted(rec.bearer_hosts.items(), key=lambda kv: -kv[1]):
            print(f"   {n:5d}  {host}")
        graph_urls = sorted({r[1].split("?")[0] for r in rec.rows if "graph.microsoft.com" in r[1]})
        print(f"\nDistinct graph.microsoft.com paths seen: {len(graph_urls)}")
        for u in graph_urls[:40]:
            print(f"   {u}")
        print("=" * 70)
        print(f"Artifacts: {DIAG}")

        browser.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())