Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116"""Every secret lives in creds.json, and only the daemon reads or writes it.
A boot moves secrets still kept in auth.json, mcp-credentials.json or aconfig.json profile into creds.json; clients then change them through thedaemon, which never answers with a secret. Both are only visible through areal daemon: the move happens at boot, and the keys prove themselves in therequests the provider receives."""
import jsonimport statimport unittest
from harness import Albedo, Provider, exclusive, text
class CredentialsTest(unittest.TestCase): def setUp(self): self.provider = Provider(lambda _request: text("ok")) self.addCleanup(self.provider.close)
def app_for(self, prepare=None): app = Albedo(self.provider, prepare=prepare) app.__enter__() self.addCleanup(app.__exit__, None, None, None) return app
def authorization(self, app): """The Authorization header of a turn on the active profile.""" start = len(self.provider.requests) session = app.session() app.prompt(session, "which key").close() app.idle(session) return self.provider.requests[start:][-1]["authorization"]
def summary(self, app): with app.api("/auth/credentials") as response: return response.read().decode()
@exclusive def test_a_boot_moves_every_secret_into_creds_json(self): app = self.app_for() # Written while the daemon runs, so the restart below is what moves them. for name, content in { "auth.json": {"anthropic": [{"type": "oauth", "access": "old-access"}]}, "mcp-credentials.json": {"servers": {"docs": {"bearerToken": "old-token"}}}, }.items(): (app.home / name).write_text(json.dumps(content)) (app.home / name).chmod(0o600) config = json.loads((app.home / "config.json").read_text()) config["providers"][app.profile]["apiKey"] = "fixture-key" (app.home / "config.json").write_text(json.dumps(config)) app.restart() creds = json.loads((app.home / "creds.json").read_text()) self.assertEqual(creds["accounts"]["anthropic"][0]["access"], "old-access") self.assertEqual(creds["mcp"]["docs"], {"bearerToken": "old-token"}) self.assertEqual(creds["providers"][app.profile], {"apiKey": "fixture-key"}) self.assertEqual(stat.S_IMODE((app.home / "creds.json").stat().st_mode), 0o600) config = json.loads((app.home / "config.json").read_text()) self.assertNotIn("apiKey", config["providers"][app.profile]) self.assertFalse((app.home / "auth.json").exists()) self.assertFalse((app.home / "mcp-credentials.json").exists()) backups = {path.name.split("-before-creds-")[0] for path in self.backups(app)} self.assertEqual(backups, {"auth.json", "mcp-credentials.json", "config.json"}) # The first client to ask hears about the move; later ones do not. for expected in (["auth.json", "mcp-credentials.json", "config.json"], []): with app.api("/auth/credentials/migration", {}) as response: self.assertEqual(json.load(response)["moved"], expected) # Not even their owner reads them without a chmod first. self.assertEqual( {stat.S_IMODE(path.stat().st_mode) for path in self.backups(app)}, {0} ) # The moved key still reaches the provider. self.assertEqual(self.authorization(app), "Bearer fixture-key")
@exclusive def test_clients_change_secrets_without_reading_them(self): # The daemon booted after the fixture wrote its profile, so the key # is already in creds.json, where a client rotates it. app = self.app_for() app.api( f"/auth/credentials/providers/{app.profile}", {"apiKey": "rotated-key"}, method="PUT", ).close() self.assertEqual(self.authorization(app), "Bearer rotated-key")
def patch(body): with app.api( "/auth/credentials/mcp/docs", body, method="PATCH" ) as response: return json.load(response)["undo"]
patch({"bearerToken": "secret-token", "headers": {"X-Team": "secret-team"}}) undo = patch({"bearerToken": None, "env": {"TOKEN": "secret-env"}}) summary = self.summary(app) self.assertIn(app.profile, json.loads(summary)["providers"]) self.assertEqual( json.loads(summary)["mcp"]["docs"], {"bearerToken": False, "headers": ["X-Team"], "env": ["TOKEN"]}, ) app.api("/auth/credentials/mcp/docs/undo", {"token": undo}).close() self.assertEqual( json.loads(self.summary(app))["mcp"]["docs"], {"bearerToken": True, "headers": ["X-Team"], "env": []}, ) self.assertFalse( any( secret in summary for secret in ("rotated-key", "secret-", "fixture-key") ) )
def backups(self, app): return list((app.home / "backups").glob("*-before-creds-*"))