feat(cli): dimos login — device-code cloud auth

dimos/cli/cloud.py

@@ -0,0 +1,158 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Dimensional cloud auth: `dimos login` / `dimos logout` / `dimos whoami`.
+
+Device-code flow (RFC 8628 shaped) against login.dimensional.org — built for robots:
+no browser or clipboard needed on this machine. The CLI prints an 8-character code,
+you approve it from any signed-in browser (laptop, phone), and the minted API key is
+stored in the system keyring, falling back to a 0600 file (`CREDENTIALS_PATH`) on
+headless machines with no keyring backend. `DIMOS_API_KEY` (via GlobalConfig)
+overrides any stored login.
+"""
+
+import json
+import os
+import socket
+import time
+from types import ModuleType
+from typing import Any, cast
+import urllib.error
+import urllib.parse
+import urllib.request
+
+import typer
+
+from dimos.constants import CREDENTIALS_PATH
+from dimos.core.global_config import global_config
+
+_KEYRING_SERVICE = "dimos-cloud"
+_KEYRING_USER = "default"
+
+
+def _base() -> str:
+    return global_config.dimos_cloud_url.rstrip("/")
+
+
+def _post(path: str, **params: str | int) -> dict[str, Any]:
+    url = f"{_base()}{path}?" + urllib.parse.urlencode(params)
+    with urllib.request.urlopen(urllib.request.Request(url, method="POST")) as r:
+        return cast("dict[str, Any]", json.load(r))
+
+
+def _keyring() -> ModuleType | None:
+    """The OS keyring, or None on machines without a usable backend (headless robots)."""
+    try:
+        import keyring
+
+        keyring.get_password(_KEYRING_SERVICE, "probe")
+        return keyring
+    except Exception:
+        return None
+
+
+def _store(creds: dict[str, str]) -> str:
+    """Persist credentials; returns a human-readable location for the login message."""
+    blob = json.dumps(creds)
+    if kr := _keyring():
+        kr.set_password(_KEYRING_SERVICE, _KEYRING_USER, blob)
+        return "system keyring"
+    CREDENTIALS_PATH.parent.mkdir(parents=True, exist_ok=True)
+    # No keyring backend (typical on robots): owner-only file, the same convention
+    # gh / aws / kubectl use for exactly this situation.
+    fd = os.open(CREDENTIALS_PATH, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    with os.fdopen(fd, "w") as f:
+        f.write(blob)
+    return str(CREDENTIALS_PATH)
+
+
+def _load() -> dict[str, str] | None:
+    if kr := _keyring():
+        if blob := kr.get_password(_KEYRING_SERVICE, _KEYRING_USER):
+            return cast("dict[str, str]", json.loads(blob))
+    try:
+        return cast("dict[str, str]", json.loads(CREDENTIALS_PATH.read_text()))
+    except (OSError, ValueError):
+        return None
+
+
+def _forget() -> bool:
+    found = False
+    if kr := _keyring():
+        if kr.get_password(_KEYRING_SERVICE, _KEYRING_USER):
+            kr.delete_password(_KEYRING_SERVICE, _KEYRING_USER)
+            found = True
+    if CREDENTIALS_PATH.exists():
+        CREDENTIALS_PATH.unlink()
+        found = True
+    return found
+
+
+def api_key() -> str | None:
+    """The credential for cloud calls: DIMOS_API_KEY first, then the stored login."""
+    if global_config.dimos_api_key:
+        return global_config.dimos_api_key
+    creds = _load()
+    return creds.get("api_key") if creds else None
+
+
+def login() -> None:
+    """Sign this machine in to Dimensional cloud."""
+    d = _post("/auth/device", label=socket.gethostname())
+    typer.echo(f"\n  Open        {d['verification_uri']}")
+    typer.echo(f"  Enter code  {d['user_code']}\n")
+    deadline = time.time() + d["expires_in"]
+    while time.time() < deadline:
+        time.sleep(d["interval"])
+        r = _post("/auth/token", device_code=d["device_code"])
+        if r["status"] == "ok":
+            where = _store({"api_key": r["api_key"], "email": r["email"]})
+            typer.echo(f"Logged in as {r['email']} (key {r['key_id']}…, stored in {where})")
+            return
+        if r["status"] in ("denied", "expired"):
+            typer.echo(f"Login {r['status']}.", err=True)
+            raise typer.Exit(1)
+    typer.echo("Login timed out.", err=True)
+    raise typer.Exit(1)
+
+
+def logout() -> None:
+    """Forget the stored key. Revoke it fully at login.dimensional.org/keys."""
+    if _forget():
+        typer.echo(f"Logged out. Revoke the key at {_base()}/keys.")
+    else:
+        typer.echo("Not logged in.")
+
+
+def whoami() -> None:
+    """Show which account this machine's key belongs to."""
+    key = api_key()
+    if not key:
+        typer.echo("Not logged in — run `dimos login`.", err=True)
+        raise typer.Exit(1)
+    req = urllib.request.Request(
+        f"{_base()}/auth/whoami", headers={"Authorization": f"Bearer {key}"}
+    )
+    try:
+        with urllib.request.urlopen(req) as r:
+            who = json.load(r)
+    except urllib.error.HTTPError as e:
+        typer.echo(
+            "Key invalid or revoked — run `dimos login`."
+            if e.code == 401
+            else f"Cloud error: {e.code}",
+            err=True,
+        )
+        raise typer.Exit(1) from e
+    typer.echo(f"{who['email']} (scopes: {who['scopes']})")

dimos/cli/dimos.py

@@ -50,6 +50,7 @@
 
 from dimos.agents.mcp.mcp_adapter import McpAdapter, McpError
 from dimos.cli.cache import app as cache_app
+from dimos.cli.cloud import login as cloud_login, logout as cloud_logout, whoami as cloud_whoami
 from dimos.cli.hardware_cli import app as hardware_app
 from dimos.cli.shell import shell
 from dimos.constants import CONFIG_DIR, LOG_DIR
@@ -178,6 +179,9 @@ def callback(**kwargs) -> None:  # type: ignore[no-untyped-def]
 main.add_typer(piper_app, name="piper")
 main.command()(shell)
 main.add_typer(cache_app, name="cache")
+main.command("login")(cloud_login)
+main.command("logout")(cloud_logout)
+main.command("whoami")(cloud_whoami)
 
 
 def _with_relay_bridge(blueprint: Blueprint) -> Blueprint:

dimos/cli/test_cloud.py

@@ -0,0 +1,160 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import io
+import json
+from pathlib import Path
+import time
+from typing import Any
+import urllib.error
+import urllib.request
+
+import pytest
+import typer
+
+from dimos.cli import cloud
+from dimos.core.global_config import global_config
+
+
+class FakeKeyring:
+    """Dict-backed stand-in for the OS keyring."""
+
+    def __init__(self) -> None:
+        self.store: dict[tuple[str, str], str] = {}
+
+    def get_password(self, service: str, user: str) -> str | None:
+        return self.store.get((service, user))
+
+    def set_password(self, service: str, user: str, value: str) -> None:
+        self.store[(service, user)] = value
+
+    def delete_password(self, service: str, user: str) -> None:
+        del self.store[(service, user)]
+
+
+@pytest.fixture
+def filestore(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
+    """Force the headless path: no keyring backend, credentials in a temp file."""
+    monkeypatch.setattr(cloud, "_keyring", lambda: None)
+    cred = tmp_path / "credentials.json"
+    monkeypatch.setattr(cloud, "CREDENTIALS_PATH", cred)
+    monkeypatch.setattr(global_config, "dimos_api_key", None)
+    return cred
+
+
+def _responses(*rs: dict[str, Any]) -> Any:
+    it = iter(rs)
+    return lambda path, **kw: next(it)
+
+
+def test_login_stores_key_owner_only(monkeypatch: pytest.MonkeyPatch, filestore: Path) -> None:
+    monkeypatch.setattr(time, "sleep", lambda s: None)
+    monkeypatch.setattr(
+        cloud,
+        "_post",
+        _responses(
+            {
+                "device_code": "dc",
+                "user_code": "AAAA-BBBB",
+                "verification_uri": "u",
+                "interval": 5,
+                "expires_in": 900,
+            },
+            {"status": "authorization_pending"},
+            {"status": "ok", "api_key": "dimos_sk_x", "key_id": "dimos_sk_x", "email": "e@x"},
+        ),
+    )
+
+    cloud.login()
+
+    assert json.loads(filestore.read_text())["api_key"] == "dimos_sk_x"
+    assert oct(filestore.stat().st_mode)[-3:] == "600"
+    assert cloud.api_key() == "dimos_sk_x"
+
+    cloud.logout()
+    assert not filestore.exists() and cloud.api_key() is None
+
+
+def test_login_denied_exits(monkeypatch: pytest.MonkeyPatch, filestore: Path) -> None:
+    monkeypatch.setattr(time, "sleep", lambda s: None)
+    monkeypatch.setattr(
+        cloud,
+        "_post",
+        _responses(
+            {
+                "device_code": "dc",
+                "user_code": "AAAA-BBBB",
+                "verification_uri": "u",
+                "interval": 5,
+                "expires_in": 900,
+            },
+            {"status": "denied"},
+        ),
+    )
+    with pytest.raises(typer.Exit):
+        cloud.login()
+    assert not filestore.exists()
+
+
+def test_global_config_key_overrides_stored(
+    monkeypatch: pytest.MonkeyPatch, filestore: Path
+) -> None:
+    filestore.write_text(json.dumps({"api_key": "dimos_sk_stored"}))
+    monkeypatch.setattr(global_config, "dimos_api_key", "dimos_sk_env")
+    assert cloud.api_key() == "dimos_sk_env"
+
+
+def test_keyring_store_load_forget(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    kr = FakeKeyring()
+    monkeypatch.setattr(cloud, "_keyring", lambda: kr)
+    monkeypatch.setattr(cloud, "CREDENTIALS_PATH", tmp_path / "never-written.json")
+    monkeypatch.setattr(global_config, "dimos_api_key", None)
+
+    assert cloud._store({"api_key": "dimos_sk_kr", "email": "e@x"}) == "system keyring"
+    assert not (tmp_path / "never-written.json").exists()  # keyring won; no file
+    assert cloud.api_key() == "dimos_sk_kr"
+    assert cloud._forget() is True
+    assert cloud.api_key() is None and cloud._forget() is False
+
+
+def test_whoami_displays_account(
+    monkeypatch: pytest.MonkeyPatch, filestore: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    filestore.write_text(json.dumps({"api_key": "dimos_sk_w"}))
+    seen: dict[str, str] = {}
+
+    def fake_urlopen(req: Any) -> io.BytesIO:
+        seen["auth"] = req.get_header("Authorization")
+        return io.BytesIO(json.dumps({"email": "e@x", "scopes": "data"}).encode())
+
+    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
+    cloud.whoami()
+    assert seen["auth"] == "Bearer dimos_sk_w"
+    assert "e@x (scopes: data)" in capsys.readouterr().out
+
+
+def test_whoami_revoked_key_exits(monkeypatch: pytest.MonkeyPatch, filestore: Path) -> None:
+    filestore.write_text(json.dumps({"api_key": "dimos_sk_dead"}))
+
+    def fake_urlopen(req: Any) -> io.BytesIO:
+        raise urllib.error.HTTPError(req.full_url, 401, "unauthorized", {}, None)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
+    with pytest.raises(typer.Exit):
+        cloud.whoami()
+
+
+def test_whoami_not_logged_in_exits(filestore: Path) -> None:
+    with pytest.raises(typer.Exit):
+        cloud.whoami()

dimos/constants.py

@@ -38,6 +38,8 @@
     LOG_DIR = STATE_DIR / "logs"
     RECORDINGS_DIR = STATE_DIR / "recordings"
 
+CREDENTIALS_PATH = CONFIG_DIR / "dimos" / "credentials.json"
+
 """
 Constants for shared memory
 Usually, auto-detection for size would be preferred. Sadly, though, channels are made

dimos/core/global_config.py

@@ -102,6 +102,8 @@ class GlobalConfig(BaseSettings):
     dimsim_headless: bool = True
     local_relay: bool = False
     relay_url: str | None = None
+    dimos_cloud_url: str = "https://login.dimensional.org"
+    dimos_api_key: str | None = None
 
     model_config = SettingsConfigDict(
         env_file=".env",

pyproject.toml

@@ -136,6 +136,7 @@ dependencies = [
     "textual-serve>=1.1.1,<2",
     "terminaltexteffects==0.12.2",
     "typer>=0.19.2,<1",
+    "keyring>=25,<26",
     "ipython",
     "plotext==5.3.2",
     # Used for calculating the occupancy map.