Cli Bridge

"""Shared helpers for the cli-bridge skill scripts.

Mirror the conventions of skills/workroom/scripts/_common.py but slimmer —
cli-bridge needs loopback access to clawd's /api/keys plus authed access
to sc-chatroom's /cli-keys for the short-code exchange.
"""
from __future__ import annotations

import os
import sys
from typing import Optional

import httpx

# ---------------------------------------------------------------------------
# Env
# ---------------------------------------------------------------------------

USER_ID = os.environ.get("USER_ID", "").strip()
CLAWD_BASE_URL = os.environ.get("CLAWD_BASE_URL", "http://127.0.0.1:8000").rstrip("/")

# Public URL of the sc-chatroom gateway. The bundle hands this to the CLI
# as the dial target — it MUST be reachable from the user's laptop, not
# the .internal one.
CHATROOM_PUBLIC_URL = os.environ.get(
    "CHATROOM_PUBLIC_URL", "https://workroom.iamstarchild.com",
).rstrip("/")

# Internal URL the skill uses for its own server-side calls (POST /cli-keys).
# Defaults to .internal so the call stays inside Fly; falls back to public.
CHATROOM_SERVER_URL = os.environ.get(
    "CHATROOM_SERVER_URL", "http://sc-chatroom.internal:8080",
).rstrip("/")

# Fly machine id for this clawd — replayed by sc-chatroom as the
# `fly-force-instance-id` header so the proxy lands on this exact machine.
CONTAINER_ID = (
    os.environ.get("CONTAINER_ID")
    or os.environ.get("FLY_MACHINE_ID")
    or ""
).strip()

CLI_BRIDGE_SCOPE = "chat:bridge:cli"


def require_env() -> None:
    if not USER_ID:
        die("USER_ID env var is not set")
    if not CONTAINER_ID:
        die(
            "no CONTAINER_ID/FLY_MACHINE_ID — sc-chatroom needs the Fly machine "
            "id so /agent/chat/stream can route to this exact clawd. Set "
            "FLY_MACHINE_ID (Fly auto-injects this on every machine) or CONTAINER_ID."
        )


def get_user_jwt() -> str:
    """Return the JWT clawd uses to identify itself to other Starchild
    internal services. Same logic as the chatroom skill: prefer USER_JWT
    env override, fall back to CONTAINER_JWT (the long-lived token Fly
    injects into every clawd container)."""
    override = os.environ.get("USER_JWT", "").strip()
    if override:
        return override
    container = os.environ.get("CONTAINER_JWT", "").strip()
    if container:
        return container
    die(
        "no identity JWT available — expected CONTAINER_JWT env (production) "
        "or USER_JWT env (dev)."
    )
    return ""  # unreachable


# ---------------------------------------------------------------------------
# UI
# ---------------------------------------------------------------------------

def die(msg: str, code: int = 1) -> None:
    print(f"error: {msg}", file=sys.stderr)
    sys.exit(code)


def info(msg: str) -> None:
    print(msg)


# ---------------------------------------------------------------------------
# HTTP
# ---------------------------------------------------------------------------

def clawd_call(method: str, path: str, **kwargs) -> httpx.Response:
    """Loopback to this agent's own clawd — middleware recognizes 127.0.0.1
    and treats it as `auth_type=internal`, so no bearer required."""
    url = CLAWD_BASE_URL + path
    with httpx.Client(timeout=10.0) as c:
        return c.request(method, url, **kwargs)


def chatroom_call(method: str, path: str, *, user_jwt: Optional[str] = None,
                  **kwargs) -> httpx.Response:
    """Authed call to sc-chatroom server (Fly-internal). Bearer is the
    container JWT unless overridden — sc-chatroom's auth.py treats it as
    a starchild user identity."""
    headers = dict(kwargs.pop("headers", {}))
    if "authorization" not in {k.lower() for k in headers}:
        jwt = user_jwt or get_user_jwt()
        headers["Authorization"] = f"Bearer {jwt}"
    url = CHATROOM_SERVER_URL + path
    with httpx.Client(timeout=15.0) as c:
        return c.request(method, url, headers=headers, **kwargs)

#!/usr/bin/env python3
"""cli-list [--include-revoked]

List the CLI bundles this user has minted on sc-chatroom (sc_… short
codes). Each row: code, label, created, expires, last_used, use_count.

The underlying AKM secrets are NOT shown — sc-chatroom holds them server-
side and never returns them. To see the AKM rows themselves, use the
chatroom skill's `list_room_keys` style introspection on clawd directly.
"""
from __future__ import annotations

import argparse
import datetime
import sys

import _common as C


def _ts(v) -> str:
    if not v:
        return "-"
    return datetime.datetime.fromtimestamp(v).isoformat(timespec="minutes")


def _parse(argv: list[str]) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="cli-list",
        description="Show CLI short-code bundles minted on sc-chatroom.",
    )
    p.add_argument(
        "--include-revoked", action="store_true",
        help="Also show codes the user has already revoked.",
    )
    return p.parse_args(argv[1:])


def main(argv: list[str]) -> int:
    args = _parse(argv)
    C.require_env()

    qs = "?include_revoked=true" if args.include_revoked else ""
    r = C.chatroom_call("GET", f"/cli-keys{qs}")
    if r.status_code != 200:
        C.die(f"sc-chatroom GET /cli-keys returned {r.status_code}: {r.text}")
    keys = r.json().get("keys", [])
    if not keys:
        C.info("no CLI bundles on sc-chatroom for this user")
        if not args.include_revoked:
            C.info("  (re-run with --include-revoked to see revoked ones)")
        return 0

    hdr = f"{'CODE':<14} {'ISSUED':<18} {'EXPIRES':<18} {'USES':<6} LABEL"
    C.info(hdr)
    C.info("-" * len(hdr))
    for k in keys:
        flags = " ✗revoked" if k.get("revoked") else ""
        C.info(
            f"{k.get('code', '?'):<14} "
            f"{_ts(k.get('created_at')):<18} "
            f"{_ts(k.get('expires_at')):<18} "
            f"{str(k.get('use_count') or 0):<6} "
            f"{k.get('label') or ''}{flags}"
        )
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))

#!/usr/bin/env python3
"""cli-login --label "<name>" [--ttl-days N]

Mint a fresh AKM key with `scope=chat:bridge:cli` on the local clawd, then
register it with sc-chatroom in exchange for a short opaque code
(``sc_<8>``). The bundle handed to the user contains only the short code
— never the AKM secret, never the Fly machine id.

Bundle layout (matches tools/starchild/internal/identity/identity.go):

    {
      "d":   <sc-chatroom public URL>,
      "c":   ""                          ← deprecated; resolved server-side
      "k":   "sc_xxxxxxxx",              ← short code, server-resolves to AKM
      "s":   "chat:bridge:cli",
      "exp": <unix expiry>,
      "l":   <user-supplied label>
    }

Revoking the bundle is now: ``cli-revoke <sc_…>`` (kills the short code,
AKM stays alive for direct use); or ``cli-revoke --akm <prefix>`` to also
nuke the underlying AKM on clawd.
"""
from __future__ import annotations

import argparse
import base64
import json
import sys

import _common as C


DEFAULT_TTL_DAYS = 90
MAX_TTL_DAYS = 365
DEFAULT_RATE_LIMIT = {"per_minute": 30}


def _parse(argv: list[str]) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="cli-login",
        description="Mint a CLI bundle for the starchild binary.",
    )
    p.add_argument(
        "--label", required=True,
        help='Human reminder for this key (e.g. "my laptop", "codex-vm").',
    )
    p.add_argument(
        "--ttl-days", type=int, default=DEFAULT_TTL_DAYS,
        help=f"Days until the AKM key expires (default {DEFAULT_TTL_DAYS}, "
             f"max {MAX_TTL_DAYS}).",
    )
    p.add_argument(
        "--enable-shell", action="store_true",
        help="Grant the `shell` capability so `starchild agent-shell` can run "
             "commands on the user's machine. OFF by default: a plain bundle is "
             "a chat bridge only, never local RCE. Only pass this when the user "
             "explicitly asks for local shell access.",
    )
    p.add_argument(
        "--enable-files", action="store_true",
        help="Grant the `files` capability so the agent can read/write files on "
             "the user's machine via agent-shell. OFF by default. Independent of "
             "--enable-shell. Transfers are path-gated on the laptop (dedicated "
             "dir + policy); only pass this when the user wants file transfer.",
    )
    return p.parse_args(argv[1:])


def _encode_bundle(payload: dict) -> str:
    raw = json.dumps(payload, separators=(",", ":")).encode()
    return "starchild_" + base64.urlsafe_b64encode(raw).decode().rstrip("=")


def main(argv: list[str]) -> int:
    args = _parse(argv)
    label = args.label.strip()
    if not label:
        C.die("--label cannot be empty")
    ttl_days = args.ttl_days
    if ttl_days < 1:
        C.die("--ttl-days must be >= 1")
    if ttl_days > MAX_TTL_DAYS:
        C.die(f"--ttl-days exceeds max ({MAX_TTL_DAYS}); use a shorter TTL")
    C.require_env()
    ttl_seconds = ttl_days * 86400
    enable_shell = args.enable_shell
    enable_files = args.enable_files

    # 1. Mint the AKM key on local clawd. Capabilities are granted ONLY when
    # the matching --enable-* flag is passed — the AKM is the authoritative
    # capability source (clawd reads it on the /ws/cli-shell handshake and
    # refuses exec/file frames for connections lacking the capability, #264).
    # `shell` (run commands) and `files` (read/write files) are independent.
    # A default bundle is a chat bridge only: even if it leaks, it cannot drive
    # local_shell or file transfer.
    capabilities = []
    if enable_shell:
        capabilities.append("shell")
    if enable_files:
        capabilities.append("files")
    akm_body = {
        "scope": C.CLI_BRIDGE_SCOPE,
        "ttl_seconds": ttl_seconds,
        "label": label,
        "rate_limit": DEFAULT_RATE_LIMIT,
        "capabilities": capabilities,
    }
    r = C.clawd_call("POST", "/api/keys", json=akm_body)
    if r.status_code != 201:
        C.die(f"clawd POST /api/keys returned {r.status_code}: {r.text}")
    akm_resp = r.json()
    akm_secret = akm_resp["secret"]
    akm_prefix = akm_resp["key"]["prefix"]
    # clawd stores expires_at as REAL (services/akm.py); the Go CLI's
    # Bundle.ExpiresAt is int64 and rejects floats during JSON decode.
    expires_at = int(akm_resp["key"]["expires_at"])

    # 2. Register the AKM + container_id with sc-chatroom in exchange for
    # a short sc_… code. After this, the AKM secret never leaves the
    # Fly internal network in plaintext — bundle carries only the code.
    register_body = {
        "akm_key": akm_secret,
        "container_id": C.CONTAINER_ID,
        "label": label,
        "ttl_seconds": ttl_seconds,
    }
    rr = C.chatroom_call("POST", "/cli-keys", json=register_body)
    if rr.status_code != 201:
        # Roll back the AKM — don't leave a live secret nobody references.
        try:
            C.clawd_call("DELETE", f"/api/keys/{akm_prefix}")
        except Exception:
            pass
        C.die(
            f"sc-chatroom POST /cli-keys returned {rr.status_code}: {rr.text}"
            " (AKM rolled back)"
        )
    code = rr.json()["code"]

    # 3. Pack the bundle. No secret inside — only the short code, the
    # gateway URL, expiry, label, and the advertised capabilities (`x`).
    # `x` is informational for the laptop (the AKM in clawd is authoritative);
    # the agent-shell daemon reads it to decide whether to even offer shell.
    # Omitted entirely for a plain bridge bundle.
    payload = {
        "d": C.CHATROOM_PUBLIC_URL,
        "c": "",                      # routing target now resolved by the code
        "k": code,                    # bearer the CLI will send
        "s": C.CLI_BRIDGE_SCOPE,
        "exp": expires_at,
        "l": label,
    }
    if capabilities:
        payload["x"] = capabilities   # only present when shell was granted
    bundle = _encode_bundle(payload)

    C.info(f"  ✓ minted CLI key (akm_prefix {akm_prefix}, code {code}, "
           f"expires {expires_at})")
    if enable_shell:
        C.info("  ⚠ shell ENABLED — agent-shell on this bundle can run commands "
               "on the user's machine (gated by their local exec-policy).")
    if enable_files:
        C.info("  ⚠ files ENABLED — the agent can read/write files on the user's "
               "machine (gated by their local file-policy; dedicated transfer dir).")
    if not capabilities:
        C.info("  • chat bridge only (no shell, no files). Re-run with "
               "--enable-shell and/or --enable-files to allow local access.")
    C.info("")
    C.info("First time on this device? Grab the starchild binary "
           "(auto-detects your OS):")
    C.info(f"  {C.CHATROOM_PUBLIC_URL.rstrip('/')}/starchild")
    C.info("")
    C.info("Then pair the CLI by pasting this into your terminal:")
    C.info("")
    C.info(f"  starchild login {bundle}")
    C.info("")
    C.info(
        "The bundle no longer contains your AKM secret — sc-chatroom resolves "
        f"the short code ({code}) on each call. Revoke immediately if it leaks:"
    )
    C.info(f"  python3 skills/cli-bridge/scripts/cli_revoke.py {code}")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))

#!/usr/bin/env python3
"""cli-revoke <code|prefix>

Revoke a CLI bundle. Two flavors:

  cli-revoke sc_xxxxxxxx          # default — kills the short code in
                                  # sc-chatroom; underlying AKM stays alive
                                  # (other bundles tied to the same AKM
                                  # are unaffected).

  cli-revoke --akm sk_xxxxxxxx    # also kills the AKM on local clawd —
                                  # ALL bundles backed by it stop working.

Use the short-code form unless you specifically want to nuke the AKM too.
"""
from __future__ import annotations

import argparse
import sys

import _common as C


def _parse(argv: list[str]) -> argparse.Namespace:
    p = argparse.ArgumentParser(prog="cli-revoke")
    p.add_argument(
        "target",
        help="sc_… short code (default) or sk_… AKM prefix when used with --akm",
    )
    p.add_argument(
        "--akm", action="store_true",
        help="treat target as an AKM prefix and revoke the AKM directly on "
             "clawd (kills every bundle backed by it).",
    )
    return p.parse_args(argv[1:])


def main(argv: list[str]) -> int:
    args = _parse(argv)
    target = args.target.strip()
    if not target:
        C.die("target is empty")
    C.require_env()

    if args.akm:
        # AKM-level revoke: only kill if scope matches the cli-bridge scope,
        # so a fat-fingered chatroom AKM prefix can't accidentally take out
        # someone's room key.
        lookup = C.clawd_call("GET", "/api/keys?include_expired=true")
        if lookup.status_code == 200:
            match = next(
                (k for k in lookup.json().get("keys", []) if k.get("prefix") == target),
                None,
            )
            if match is None:
                C.die(f"no AKM key with prefix {target!r} on this clawd")
            if match.get("scope") != C.CLI_BRIDGE_SCOPE:
                C.die(
                    f"refusing to revoke: key {target} has scope "
                    f"{match.get('scope')!r}, not {C.CLI_BRIDGE_SCOPE!r}. "
                    "Use the chatroom skill's revoke commands for room keys."
                )
        r = C.clawd_call("DELETE", f"/api/keys/{target}")
        if r.status_code in (200, 204):
            C.info(f"  ✓ revoked AKM {target} (all bundles using it now dead)")
            return 0
        if r.status_code == 404:
            C.die(f"AKM {target} not found")
        C.die(f"clawd DELETE returned {r.status_code}: {r.text}")

    # Short-code revoke: this is the default path.
    if not target.startswith("sc_"):
        C.die(
            f"expected sc_… short code (got {target!r}); use --akm if you "
            "intended to revoke an AKM prefix"
        )
    r = C.chatroom_call("DELETE", f"/cli-keys/{target}")
    if r.status_code in (200, 204):
        C.info(f"  ✓ revoked CLI bundle {target} on sc-chatroom")
        C.info("    (underlying AKM left alive; use --akm to revoke that too)")
        return 0
    if r.status_code == 404:
        C.die(f"code {target} not found, already revoked, or not yours")
    C.die(f"sc-chatroom DELETE /cli-keys/{target} returned {r.status_code}: {r.text}")
    return 1  # unreachable


if __name__ == "__main__":
    sys.exit(main(sys.argv))