# Antigravity CLI — auth internals

Detail behind the headless-auth flow in SKILL.md. `agy` is v1.0.0
(released at I/O 2026); behaviour below was verified empirically.

## Install notes

`curl -fsSL https://antigravity.google/cli/install.sh | bash` is a clean
installer: it fetches a per-platform manifest, downloads a ~52 MB tarball,
**verifies a SHA-512 checksum**, and expands a ~183 MB Go binary to
`~/.local/bin/agy`. It is idempotent, takes ~3 s, and `agy` self-updates
afterward. Distribution is install-script only — the `antigravity-cli`
package on npm is an unrelated squatter; ignore it.

## The 30-second print-mode timeout

`agy -p` (print mode), with no stored token, prints the
`accounts.google.com` OAuth URL and waits **30 seconds** for a pasted
authorization code, then aborts. The 30 s is hardcoded in
`printmode.waitForAuth`; `--print-timeout` governs the *response* wait
(default 5m), not the auth wait. There is no flag to extend it.

`agy -i` (interactive TUI) has **no auth timeout** — it waits
indefinitely. That is why the broker drives `-i`, not `-p`.

## Driving the TUI

The TUI queries the terminal for capabilities (DECRQM for modes 2026/2027,
primary device attributes, cursor position) and will not render until the
terminal answers. A plain pipe never answers, so the TUI hangs. The broker
runs `agy -i` under a real pty and writes back synthetic replies
(`capability_replies()` in `scripts/agy_auth_broker.py`), after which the
TUI renders normally: a login menu, then the OAuth URL, then an
authorization-code input field.

The login menu offers "1. Google OAuth" and "2. Use a Google Cloud
project". The broker auto-selects option 1.

## Token storage: keyring vs file

`agy` defaults to OS-keyring token storage, which on Linux needs D-Bus
(`dbus-launch`). Where D-Bus is absent the keyring write fails silently:

```
keyringAuth: failed to persist token to keyring:
  exec: "dbus-launch": executable file not found in $PATH
consumerOAuth: authentication completed successfully
```

Auth *succeeds* but the token is never persisted — it dies with the
process. `agy` switches to **file-based storage when it detects an SSH
session** (logs `Using file-based token storage because SSH session
detected`). Exporting `SSH_CONNECTION` / `SSH_CLIENT` / `SSH_TTY` triggers
that path. The broker sets them on the `agy` child automatically; export
them yourself before any later `agy -p` call too.

## The token file

Path: `~/.gemini/antigravity-cli/antigravity-oauth-token`. JSON:

```json
{
  "token": {
    "access_token": "ya29....",
    "token_type": "Bearer",
    "refresh_token": "1//04....",
    "expiry": "2026-05-20T17:19:27.000000000Z"
  },
  "auth_method": "consumer"
}
```

The `refresh_token` is the durable credential — `agy` refreshes the access
token from it automatically. To reuse auth on a fresh container, write this
file back (with SSH env set) instead of re-running the OAuth dance. A stale
`access_token`/`expiry` is fine; `agy` refreshes on use.

OAuth scopes requested: `cloud-platform`, `userinfo.email`,
`userinfo.profile`, `cclog`, `experimentsandconfigs`, `openid`. The request
uses `access_type=offline` so a refresh token is issued.

## Ephemeral containers

Cloud sandboxes (e.g. Claude Code on the Web) pause on session inactivity;
a pause kills the live `agy` process the broker is holding. The OAuth dance
must finish before that — in practice within a few minutes. If the process
dies mid-auth the authorization code is bound (via PKCE) to the dead
process and cannot be reused; relaunch the broker for a fresh URL.

The robust pattern is therefore: authenticate once, **save the token
file**, and re-plant it on each fresh container.

## State layout

`agy` reuses the Gemini CLI config tree:

```
~/.gemini/config/mcp_config.json                   # shared MCP server config
~/.gemini/config/projects/<uuid>.json               # per-project metadata
~/.gemini/antigravity-cli/antigravity-oauth-token    # the credential
~/.gemini/antigravity-cli/conversations/             # history (--conversation)
~/.gemini/antigravity-cli/{brain,knowledge}/
```

`agy` also drops a `.antigravitycli/` symlink dir in the working repo —
gitignore it.

## Network

| Host | Phase |
|---|---|
| `antigravity.google`, `antigravity-cli-auto-updater-*.run.app` | install, self-update |
| `storage.googleapis.com` | install payload |
| `accounts.google.com`, `oauth2.googleapis.com` | OAuth |
| `www.googleapis.com` | userinfo |
| `cloudcode-pa.googleapis.com` | runtime agent calls |

## Provisioning options, worst to best

1. **Broker per container** — `agy_auth_broker.py` + one prompt human
   OAuth dance. Repeats on every fresh container.
2. **Persist the token file** — dance once, save
   `antigravity-oauth-token`, re-plant on boot. No further dances; ships a
   personal credential, so keep it out of git and logs.
3. **GCP service account (ADC)** — the binary references
   `GOOGLE_APPLICATION_CREDENTIALS` / `GOOGLE_CLOUD_PROJECT`; the
   `cloud-platform` scope and enterprise "connect your GCP project" path
   imply a fully non-interactive option. Unverified, but the correct
   answer for unattended orchestration — no keyring, no personal token.

## Troubleshooting

- **`agy -p` keeps prompting for auth** — the token file is missing, or
  `agy` used the keyring path. Confirm the SSH env vars are exported in the
  current shell and `~/.gemini/antigravity-cli/antigravity-oauth-token`
  exists.
- **Broker captures no URL** — inspect `/tmp/agybroker/status` and
  `/tmp/agybroker/log`; usually `agy` is not installed or not on `PATH`.
- **Process died mid-auth** — the container idle-paused; relaunch the
  broker for a fresh URL and complete the login faster.

#!/usr/bin/env python3
"""Headless OAuth broker for the Antigravity CLI (`agy`).

`agy -p` (print mode) gives the OAuth prompt only 30 s — too short for a
human-in-the-loop login. `agy -i` (interactive TUI) has no auth timeout but
must be driven through a pseudo-terminal. This broker spawns `agy -i` under
a pty, answers the terminal capability queries so the TUI renders,
auto-selects "Google OAuth", scrapes the auth URL, and feeds back an
authorization code.

It also exports fake SSH env vars so `agy` uses file-based token storage
(`~/.gemini/antigravity-cli/antigravity-oauth-token`) instead of the OS
keyring — the container has no D-Bus, so the keyring path fails silently.

Usage:
    python3 agy_auth_broker.py &              # launch; spawns agy, captures URL
    cat /tmp/agybroker/url                    # -> give this URL to a human
    # human consents in a browser, copies the authorization code
    printf '<code>' > /tmp/agybroker/code     # broker feeds it to agy

After a successful auth `agy` writes the token file itself; subsequent
`agy -p` calls (with the same SSH env set) run non-interactively.

State directory (default /tmp/agybroker): url, code, status, log.
The whole flow must finish before the CCotw container idle-pauses (a few
minutes) — a paused container kills the live `agy` process.
"""
import fcntl
import os
import pty
import re
import select
import struct
import sys
import termios
import time

OAUTH_URL_RE = re.compile(rb"https://accounts\.google\.com/o/oauth2/auth[^\s\x1b\"']+")


def extract_oauth_url(buf):
    """Return the Google OAuth URL found in a pty output buffer, or None.

    The match deliberately stops at whitespace, ESC (0x1b) and quotes, so a
    URL surrounded by TUI escape sequences is still recovered intact."""
    m = OAUTH_URL_RE.search(buf)
    return m.group(0) if m else None


def capability_replies(data):
    """Bytes to write back so a TUI's terminal capability queries are
    answered and it stops waiting. Returns b'' when `data` has no queries."""
    out = b""
    for mode_n in (2026, 2027, 1016, 1049):
        if (b"\x1b[?%d$p" % mode_n) in data:
            out += b"\x1b[?%d;2$y" % mode_n
    if b"\x1b[c" in data or b"\x1b[>c" in data or b"\x1b[>0c" in data:
        out += b"\x1b[?62;1;6c"
    if b"\x1b[6n" in data:
        out += b"\x1b[1;1R"
    return out


def main():
    state_dir = os.environ.get("AGY_BROKER_DIR", "/tmp/agybroker")
    os.makedirs(state_dir, exist_ok=True)
    log_p, url_p, code_p, status_p = (
        f"{state_dir}/{n}" for n in ("log", "url", "code", "status"))
    for f in (log_p, url_p, code_p, status_p):
        try:
            os.remove(f)
        except FileNotFoundError:
            pass

    prompt = sys.argv[1] if len(sys.argv) > 1 else "reply with the single word OK"
    agy = os.path.expanduser("~/.local/bin/agy")

    def status(s):
        open(status_p, "w").write(s + "\n")

    status("starting")
    pid, fd = pty.fork()
    if pid == 0:
        os.environ["TERM"] = "xterm-256color"
        os.environ["PATH"] = os.path.expanduser("~/.local/bin") + ":" + os.environ.get("PATH", "")
        # SSH env => agy uses file-based token storage, not the D-Bus keyring
        os.environ.setdefault("SSH_CONNECTION", "203.0.113.1 50000 203.0.113.2 22")
        os.environ.setdefault("SSH_CLIENT", "203.0.113.1 50000 22")
        os.environ.setdefault("SSH_TTY", "/dev/pts/0")
        os.execv(agy, [agy, "-i", prompt])
        os._exit(1)

    # wide pty so a ~400-char URL is never line-wrapped
    fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", 50, 1000, 0, 0))

    buf = b""
    log = open(log_p, "wb")
    url_done = code_sent = menu_done = False
    start = time.time()
    status("running")
    while True:
        r, _, _ = select.select([fd], [], [], 0.2)
        if r:
            try:
                data = os.read(fd, 8192)
            except OSError:
                data = b""
            if not data:
                break
            buf += data
            log.write(data)
            log.flush()
            reply = capability_replies(data)
            if reply:
                os.write(fd, reply)
            # menu: "Google OAuth" is option 1, already highlighted -> Enter
            if not menu_done and b"Select login method" in buf:
                time.sleep(0.4)
                os.write(fd, b"\r")
                menu_done = True
                status("oauth-selected")
            if not url_done:
                url = extract_oauth_url(buf)
                if url:
                    open(url_p, "wb").write(url)
                    url_done = True
                    status("url-ready")
        if not code_sent and os.path.exists(code_p):
            code = open(code_p).read().strip()
            if code:
                os.write(fd, code.encode() + b"\r")
                code_sent = True
                status("code-sent")
        if time.time() - start > 1200:
            break
    status("exited")


if __name__ == "__main__":
    main()

"""Unit tests for scripts/agy_auth_broker.py pure helpers.

Covers OAuth-URL extraction and terminal-capability-query answering. The
pty/subprocess machinery in main() is not unit-tested — it requires a live
`agy` binary and a real OAuth TUI.
"""
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import agy_auth_broker as broker

_URL = (b"https://accounts.google.com/o/oauth2/auth?access_type=offline"
        b"&client_id=123.apps.googleusercontent.com&code_challenge=abc"
        b"&redirect_uri=https%3A%2F%2Fantigravity.google%2Foauth-callback"
        b"&response_type=code&scope=openid&state=xyz")


def test_extract_url_clean_buffer():
    assert broker.extract_oauth_url(b"prefix\n" + _URL + b"\nsuffix") == _URL


def test_extract_url_none_when_absent():
    assert broker.extract_oauth_url(b"no url here, just text") is None
    assert broker.extract_oauth_url(b"") is None


def test_extract_url_amid_escape_codes():
    # the TUI wraps the URL in cursor-positioning escape sequences
    noisy = b"\x1b[2J\x1b[1;1H" + _URL + b"\x1b[0m\x1b[2;1H"
    assert broker.extract_oauth_url(noisy) == _URL


def test_extract_url_stops_at_whitespace():
    assert broker.extract_oauth_url(_URL + b" trailing words") == _URL


def test_extract_url_ignores_non_oauth_google_urls():
    assert broker.extract_oauth_url(b"see https://accounts.google.com/signin") is None


def test_capability_reply_decrqm_2026():
    assert broker.capability_replies(b"\x1b[?2026$p") == b"\x1b[?2026;2$y"


def test_capability_reply_primary_da():
    assert broker.capability_replies(b"\x1b[c") == b"\x1b[?62;1;6c"


def test_capability_reply_cursor_position_request():
    assert broker.capability_replies(b"\x1b[6n") == b"\x1b[1;1R"


def test_capability_reply_empty_for_plain_output():
    assert broker.capability_replies(b"just normal TUI text, no queries") == b""


def test_capability_reply_concatenates_multiple_queries():
    out = broker.capability_replies(b"\x1b[?2026$p\x1b[?2027$p\x1b[6n")
    assert b"\x1b[?2026;2$y" in out
    assert b"\x1b[?2027;2$y" in out
    assert b"\x1b[1;1R" in out


if __name__ == "__main__":
    fns = sorted(n for n in dir() if n.startswith("test_"))
    for n in fns:
        globals()[n]()
        print(f"  ok  {n}")
    print(f"{len(fns)} passed")