didericis/bot-bottle
1
0
Fork 0
Code Issues 36 Pull Requests 3 Actions Packages Releases Wiki Activity

feat: add bb login command for console host registration

Browse Source 
Starts a device-authorization flow against a bot-bottle console, polls
until the operator approves, then writes access + refresh tokens to
$BOT_BOTTLE_ROOT/console.json. Console URL is read from --console-url
flag or BB_CONSOLE_URL env var.

Part of didericis/bot-bottle-platform#1
This commit is contained in:
didericis (claude)
2026-07-20 23:21:38 +00:00
committed by didericis
parent efd413c1ba
commit 4998a5ec6a
3 changed files with 304 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot_bottle/cli/__init__.py
+4 -1
View File
@@ -19,6 +19,7 @@ from .commit import cmd_commit
from .edit import cmd_edit
from .info import cmd_info
from .init import cmd_init
from .login import cmd_login
from .resume import cmd_resume
from .start import cmd_start
from .supervise import cmd_supervise
@@ -33,6 +34,7 @@ COMMANDS = {
"info": cmd_info,
"init": cmd_init,
"list": cmd_list,
"login": cmd_login,
"resume": cmd_resume,
"start": cmd_start,
"supervise": cmd_supervise,
@@ -43,7 +45,7 @@ COMMANDS = {
# the host (TAP pool, /dev/kvm, firecracker) and never opens the store, so
# gating it on the schema breaks preflight on a fresh CI runner where stdin
# isn't a TTY and the migration prompt can't be answered.
NO_MIGRATION_COMMANDS = frozenset({"backend"})
NO_MIGRATION_COMMANDS = frozenset({"backend", "login"})
def usage() -> None:
@@ -56,6 +58,7 @@ def usage() -> None:
sys.stderr.write(" info print env, skills, and prompt details for a named agent\n")
sys.stderr.write(" init interactively create a new agent and add it to bot-bottle.json\n")
sys.stderr.write(" list list available agents or active containers\n")
sys.stderr.write(" login register this host with a bot-bottle console\n")
sys.stderr.write(
" resume re-launch a bottle by its identity "
"(continues state from PRD 0016)\n"
bot_bottle/cli/login.py
+153
View File
@@ -0,0 +1,153 @@
"""bb login — register this host with a bot-bottle console.
Opens a device-authorization flow against the target console, waits for the
operator to approve, then writes access and refresh tokens to
~/.bot-bottle/console.json (or $BOT_BOTTLE_ROOT/console.json).
Usage:
bb login [--console-url URL] [--label LABEL]
Flags:
--console-url URL Target console URL (overrides BB_CONSOLE_URL env var)
--label LABEL Host label shown in the console (default: hostname)
"""
from __future__ import annotations
import json
import os
import socket
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
from ..paths import bot_bottle_root
_CONSOLE_URL_ENV = "BB_CONSOLE_URL"
_POLL_SLEEP = 2 # seconds between polls; matches console's poll_interval default
def _usage() -> None:
sys.stderr.write(
"usage: bb login [--console-url URL] [--label LABEL]\n"
"\n"
"Options:\n"
" --console-url URL Console base URL (or BB_CONSOLE_URL env var)\n"
" --label LABEL Host label shown in the console (default: hostname)\n"
)
def _flag(argv: list[str], name: str) -> str | None:
for i, arg in enumerate(argv):
if arg == name and i + 1 < len(argv):
return argv[i + 1]
if arg.startswith(f"{name}="):
return arg[len(name) + 1:]
return None
def _post(url: str, payload: dict) -> dict:
data = json.dumps(payload).encode()
req = urllib.request.Request(
url, data=data, headers={"Content-Type": "application/json"}
)
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read())
def _get(url: str) -> tuple[int, dict]:
req = urllib.request.Request(url)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return resp.status, json.loads(resp.read())
except urllib.error.HTTPError as e:
return e.code, {}
def _save_credentials(
console_url: str, host_id: str, access_token: str, refresh_token: str
) -> Path:
path = bot_bottle_root() / "console.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(
{
"url": console_url,
"host_id": host_id,
"access_token": access_token,
"refresh_token": refresh_token,
},
indent=2,
)
+ "\n"
)
path.chmod(0o600)
return path
def cmd_login(argv: list[str]) -> int:
if "--help" in argv or "-h" in argv:
_usage()
return 0
console_url = _flag(argv, "--console-url") or os.environ.get(_CONSOLE_URL_ENV)
if not console_url:
sys.stderr.write(
"bb login: --console-url or BB_CONSOLE_URL is required\n"
)
return 1
console_url = console_url.rstrip("/")
label = _flag(argv, "--label") or socket.gethostname()
try:
resp = _post(f"{console_url}/api/v1/hosts/authorize", {"label": label})
except Exception as exc:
sys.stderr.write(f"bb login: failed to start authorization: {exc}\n")
return 1
device_code = resp["device_code"]
user_code = resp["user_code"]
expires_in = resp.get("expires_in", 300)
sys.stderr.write(
f"\nOpen this URL in your browser to authorize this host:\n\n"
f" {console_url}/authorize?code={user_code}\n\n"
f"Waiting for approval"
)
deadline = time.monotonic() + expires_in
while time.monotonic() < deadline:
sys.stderr.write(".")
sys.stderr.flush()
time.sleep(_POLL_SLEEP)
try:
code, result = _get(
f"{console_url}/api/v1/hosts/authorize/{device_code}"
)
except Exception:
continue
if code == 410:
break
st = result.get("status")
if st == "approved":
sys.stderr.write("\n\nApproved.\n")
path = _save_credentials(
console_url,
result["host_id"],
result["access_token"],
result["refresh_token"],
)
sys.stderr.write(f"Credentials saved to {path}\n")
return 0
if st == "denied":
sys.stderr.write("\n\nDenied by operator.\n")
return 1
sys.stderr.write("\n\nAuthorization timed out.\n")
return 1
tests/unit/test_cli_login.py
+147
View File
@@ -0,0 +1,147 @@
"""Unit tests for bb login command."""
from __future__ import annotations
import json
import os
import tempfile
import unittest
from unittest.mock import MagicMock, patch
class TestFlagParsing(unittest.TestCase):
def test_console_url_flag(self) -> None:
from bot_bottle.cli.login import _flag
self.assertEqual(_flag(["--console-url", "http://x"], "--console-url"), "http://x")
def test_console_url_equals_form(self) -> None:
from bot_bottle.cli.login import _flag
self.assertEqual(_flag(["--console-url=http://x"], "--console-url"), "http://x")
def test_label_flag(self) -> None:
from bot_bottle.cli.login import _flag
self.assertEqual(_flag(["--label", "my-mac"], "--label"), "my-mac")
def test_missing_flag_returns_none(self) -> None:
from bot_bottle.cli.login import _flag
self.assertIsNone(_flag([], "--console-url"))
class TestSaveCredentials(unittest.TestCase):
def test_writes_json_and_sets_perms(self) -> None:
from bot_bottle.cli.login import _save_credentials
with tempfile.TemporaryDirectory() as tmp:
with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
path = _save_credentials("http://c", "hid", "at", "rt")
self.assertTrue(path.exists())
data = json.loads(path.read_text())
self.assertEqual(data["url"], "http://c")
self.assertEqual(data["host_id"], "hid")
self.assertEqual(data["access_token"], "at")
self.assertEqual(data["refresh_token"], "rt")
self.assertEqual(oct(path.stat().st_mode & 0o777), oct(0o600))
class TestCmdLoginMissingUrl(unittest.TestCase):
def test_returns_1_without_url(self) -> None:
from bot_bottle.cli.login import cmd_login
with patch.dict(os.environ, {}, clear=True):
os.environ.pop("BB_CONSOLE_URL", None)
result = cmd_login([])
self.assertEqual(result, 1)
def test_reads_env_var(self) -> None:
"""Exits 1 (network error) not because of missing URL when env var is set."""
from bot_bottle.cli.login import cmd_login
def _fail_post(url, payload):
raise OSError("connection refused")
with tempfile.TemporaryDirectory() as tmp:
with patch.dict(os.environ, {"BB_CONSOLE_URL": "http://localhost:9999", "BOT_BOTTLE_ROOT": tmp}):
with patch("bot_bottle.cli.login._post", side_effect=_fail_post):
result = cmd_login([])
self.assertEqual(result, 1)
class TestCmdLoginFlow(unittest.TestCase):
def _run_with_mocks(self, poll_responses: list[dict], tmp: str) -> int:
from bot_bottle.cli.login import cmd_login
start_resp = {
"device_code": "dc123",
"user_code": "ABC-DEF",
"expires_in": 300,
"poll_interval": 0,
}
poll_iter = iter(poll_responses)
def _fake_post(url, payload):
return start_resp
def _fake_get(url):
try:
resp = next(poll_iter)
except StopIteration:
return 200, {"status": "pending"}
return 200, resp
with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
with patch("bot_bottle.cli.login._post", side_effect=_fake_post):
with patch("bot_bottle.cli.login._get", side_effect=_fake_get):
with patch("bot_bottle.cli.login._POLL_SLEEP", 0):
return cmd_login(["--console-url", "http://console"])
def test_approved_flow_returns_0(self) -> None:
approved = {
"status": "approved",
"host_id": "hid",
"access_token": "at",
"refresh_token": "rt",
}
with tempfile.TemporaryDirectory() as tmp:
result = self._run_with_mocks(
[{"status": "pending"}, approved], tmp
)
self.assertEqual(result, 0)
with open(os.path.join(tmp, "console.json")) as f:
creds = json.loads(f.read())
self.assertEqual(creds["host_id"], "hid")
def test_denied_flow_returns_1(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
result = self._run_with_mocks([{"status": "denied"}], tmp)
self.assertEqual(result, 1)
def test_timeout_returns_1(self) -> None:
from bot_bottle.cli.login import cmd_login
start_resp = {
"device_code": "dc",
"user_code": "ZZZ-ZZZ",
"expires_in": 0, # already expired
"poll_interval": 0,
}
with tempfile.TemporaryDirectory() as tmp:
with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
with patch("bot_bottle.cli.login._post", return_value=start_resp):
with patch("bot_bottle.cli.login._POLL_SLEEP", 0):
result = cmd_login(["--console-url", "http://console"])
self.assertEqual(result, 1)
class TestDispatcherRegistration(unittest.TestCase):
def test_login_in_commands(self) -> None:
from bot_bottle.cli import COMMANDS
self.assertIn("login", COMMANDS)
def test_login_in_no_migration(self) -> None:
from bot_bottle.cli import NO_MIGRATION_COMMANDS
self.assertIn("login", NO_MIGRATION_COMMANDS)
if __name__ == "__main__":
unittest.main()