Grok Bot

"""grok_install / grok_install_status — install Grok Bot on this PC, no admin, no login.

Grok Bot's own update feed is public:

    GET https://api2.cursor.sh/updates/api/update/<platform>/sand/<curVer>/<id>/stable
      -> {"version", "url", "sha256hash"?}   (url on downloads.cursor.com)

platform is win32-x64-user / win32-arm64-user / darwin-arm64 / darwin-x64 / linux-x64.
This is the same request the installed app makes to update itself (see
buildUpdateRequestUrl in its app.asar), so asking it with curVer 0.0.1 returns
the current full installer.

Install targets are per-user and match grok_detect's candidate paths:
  Windows  NSIS Setup.exe /S              -> %LOCALAPPDATA%\\Programs\\Grok Bot
  macOS    zip with Grok Bot.app          -> ~/Applications/Grok Bot.app
  Linux    AppImage (sha256 from feed)    -> ~/.local/bin/grok-bot

The download is ~120 MB, longer than ab's 60 s verb wait, so grok_install runs
as a background job and grok_install_status reports progress. The installer is
never launched unless its signature (Windows Authenticode "Anysphere") or its
feed sha256 checks out. Sign-in stays a human step (grok_login).
"""

from __future__ import annotations

import hashlib
import json
import os
import platform
import shutil
import stat
import subprocess
import sys
import tempfile
import threading
import time
import urllib.request
import zipfile
from pathlib import Path

from grok_detect import CREATE_NO_WINDOW, IS_LINUX, IS_MACOS, IS_WINDOWS, detect_host

FEED = "https://api2.cursor.sh/updates/api/update/{platform}/sand/0.0.1/0/stable"
SIGNERS = ("Anysphere, Inc.", "Anysphere")
UA = "adom-grok-bridge"

_lock = threading.Lock()
_job: dict = {"state": "idle"}


def _platform_key() -> str:
    arm = platform.machine().lower() in ("arm64", "aarch64")
    if IS_WINDOWS:
        return "win32-arm64-user" if arm else "win32-x64-user"
    if IS_MACOS:
        return "darwin-arm64" if arm else "darwin-x64"
    return "linux-x64"


def _feed() -> dict:
    req = urllib.request.Request(FEED.format(platform=_platform_key()), headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=30) as r:
        data = json.loads(r.read().decode("utf-8"))
    if not data.get("url"):
        raise RuntimeError(f"update feed returned no url: {data}")
    return {
        "version": data.get("version") or data.get("name"),
        "url": data["url"],
        "sha256": data.get("sha256hash"),
    }


def _set(**kw) -> None:
    with _lock:
        _job.update(kw)


def _download(url: str, dest: Path) -> str:
    h = hashlib.sha256()
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=60) as r, open(dest, "wb") as f:
        total = int(r.headers.get("Content-Length") or 0)
        done = 0
        _set(bytesTotal=total)
        while True:
            chunk = r.read(1 << 20)
            if not chunk:
                break
            f.write(chunk)
            h.update(chunk)
            done += len(chunk)
            _set(bytesDone=done)
    return h.hexdigest()


def _authenticode(path: Path) -> dict:
    ps = (
        "$s = Get-AuthenticodeSignature -LiteralPath $env:GROK_SETUP; "
        "[pscustomobject]@{status=[string]$s.Status; "
        "subject=[string]$s.SignerCertificate.Subject} | ConvertTo-Json -Compress"
    )
    env = dict(os.environ, GROK_SETUP=str(path))
    out = subprocess.run(
        ["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", ps],
        capture_output=True, text=True, timeout=60, env=env, creationflags=CREATE_NO_WINDOW,
    )
    try:
        return json.loads(out.stdout.strip() or "{}")
    except json.JSONDecodeError:
        return {"status": "Unknown", "subject": "", "stderr": out.stderr.strip()[:300]}


def _install_windows(setup: Path) -> None:
    sig = _authenticode(setup)
    _set(signature=sig)
    if sig.get("status") != "Valid" or not any(f"CN={s}" in (sig.get("subject") or "") or
                                                f'CN="{s}"' in (sig.get("subject") or "")
                                                for s in SIGNERS):
        raise RuntimeError(f"installer signature rejected: {sig}")
    _set(state="installing")
    # NSIS per-user installer: /S is silent, no UAC, installs to %LOCALAPPDATA%\Programs.
    p = subprocess.run([str(setup), "/S"], timeout=600, creationflags=CREATE_NO_WINDOW)
    _set(installerExit=p.returncode)


def _install_macos(zpath: Path) -> None:
    _set(state="installing")
    apps = Path.home() / "Applications"
    apps.mkdir(parents=True, exist_ok=True)
    tmp = Path(tempfile.mkdtemp(prefix="grokbot-"))
    # ditto keeps the code signature and symlinks that zipfile would break.
    subprocess.run(["ditto", "-x", "-k", str(zpath), str(tmp)], check=True, timeout=600)
    app = next(tmp.glob("*.app"), None)
    if app is None:
        raise RuntimeError("no .app inside the macOS zip")
    dest = apps / "Grok Bot.app"
    if dest.exists():
        shutil.rmtree(dest)
    shutil.move(str(app), str(dest))
    subprocess.run(["xattr", "-dr", "com.apple.quarantine", str(dest)], timeout=60)


def _install_linux(appimage: Path) -> None:
    _set(state="installing")
    dest = Path.home() / ".local" / "bin" / "grok-bot"
    dest.parent.mkdir(parents=True, exist_ok=True)
    shutil.move(str(appimage), str(dest))
    dest.chmod(dest.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)


def _run(feed: dict) -> None:
    work = Path(tempfile.mkdtemp(prefix="grokbot-dl-"))
    name = feed["url"].rsplit("/", 1)[-1] or "grokbot-setup"
    dest = work / name
    try:
        _set(state="downloading", file=name)
        digest = _download(feed["url"], dest)
        _set(sha256=digest)
        if feed.get("sha256") and digest.lower() != feed["sha256"].lower():
            raise RuntimeError(f"sha256 mismatch: got {digest}, feed says {feed['sha256']}")
        if IS_WINDOWS:
            _install_windows(dest)
        elif IS_MACOS:
            _install_macos(dest)
        else:
            if not feed.get("sha256"):
                raise RuntimeError("linux feed had no sha256; refusing to install an unverified AppImage")
            _install_linux(dest)
        # NSIS returns before every file settles on some boxes; give detection a moment.
        for _ in range(30):
            if detect_host().get("hostAppInstalled") and not detect_host().get("hostAppFallback"):
                break
            time.sleep(1)
        info = detect_host()
        ok = bool(info.get("hostAppInstalled")) and not info.get("hostAppFallback")
        _set(state="done" if ok else "failed", installed=ok, paths=info.get("paths") or [],
             finishedAt=time.time(),
             error="" if ok else "installer finished but Grok Bot was not detected")
    except Exception as e:  # noqa: BLE001 - every failure must land in the job state
        _set(state="failed", error=str(e), finishedAt=time.time())
    finally:
        shutil.rmtree(work, ignore_errors=True)


def _snapshot() -> dict:
    with _lock:
        j = dict(_job)
    total, done = j.get("bytesTotal") or 0, j.get("bytesDone") or 0
    if total:
        j["percent"] = round(100.0 * done / total, 1)
    return j


def handle_install(args: dict) -> dict:
    info = detect_host()
    force = bool(args.get("force"))
    if info.get("hostAppInstalled") and not info.get("hostAppFallback") and not force:
        return {
            "success": True,
            "alreadyInstalled": True,
            "paths": info.get("paths") or [],
            "_hint": "Grok Bot is already installed. Next: grok_open {} (background), then grok_login if it is not signed in. Pass force:true to reinstall over it.",
            "_next": "grok_open {}",
        }
    with _lock:
        busy = _job.get("state") in ("resolving", "downloading", "installing")
    if busy:
        out = _snapshot()
        out.update(success=True, alreadyRunning=True,
                   _hint="An install is already running. Poll grok_install_status about every 10 s; do not start another.",
                   _next="grok_install_status {}")
        return out
    try:
        _set(state="resolving", startedAt=time.time(), error="", bytesDone=0, bytesTotal=0,
             platform=_platform_key(), signature=None, installed=False)
        feed = _feed()
    except Exception as e:  # noqa: BLE001
        _set(state="failed", error=f"update feed: {e}")
        return {
            "success": False,
            "errorCode": "feed_unreachable",
            "error": str(e),
            "_hint": "Could not read Grok Bot's update feed (api2.cursor.sh). Check this PC's internet, then retry grok_install once.",
            "_next": "grok_install {}",
        }
    _set(version=feed["version"], url=feed["url"])
    threading.Thread(target=_run, args=(feed,), daemon=True, name="grok-install").start()
    return {
        "success": True,
        "started": True,
        "version": feed["version"],
        "platform": _platform_key(),
        "url": feed["url"],
        "_hint": (
            f"Installing Grok Bot {feed['version']} in the background from its public update feed "
            "(~120 MB, per-user, no admin prompt, no window). The installer only runs after its "
            "signature (Windows: Authenticode Anysphere) or sha256 (Linux) checks out. "
            "Poll grok_install_status about every 10 s until state is done or failed."
        ),
        "_next": "grok_install_status {}",
        "related": ["grok_install_status", "grok_readiness", "grok_open", "grok_login"],
    }


def handle_install_status(_args: dict) -> dict:
    j = _snapshot()
    state = j.get("state")
    hints = {
        "idle": "No install has run since the bridge started. grok_readiness says whether Grok Bot is present; grok_install installs it.",
        "resolving": "Asking the update feed for the current installer.",
        "downloading": "Downloading. Poll again in ~10 s.",
        "installing": "Running the silent per-user installer. Poll again in ~10 s.",
        "done": "Installed. Next: grok_open {} (background), grok_login (presses Sign in; the account picks its provider on xAI's page), then grok_webhook_ensure so this PC can fire webhooks (bindings are per PC).",
        "failed": "Install failed; read error. A signature or sha256 rejection means the download was not Anysphere's: do not retry blindly, report it.",
    }
    j["success"] = state != "failed"
    j["_hint"] = hints.get(state, "Unknown install state.")
    j["_next"] = {"done": "grok_open {}", "failed": "grok_install {}"}.get(state, "grok_install_status {}")
    if state == "failed":
        j["errorCode"] = "install_failed"
    return j