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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184"""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("/settings") as response: return response.read().decode()
# exclusive: restarts the daemon to migrate global credential files @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()) anthropic = creds["accounts"]["anthropic"] anthropic = anthropic[0] if isinstance(anthropic, list) else anthropic self.assertEqual(anthropic["access"], "old-access") self.assertEqual(creds["mcp"]["docs"], {"bearerToken": "old-token"}) accounts = creds["accounts"]["openai"] accounts = accounts if isinstance(accounts, list) else [accounts] key = next(value for value in accounts if value["apiKey"] == "fixture-key") self.assertNotIn(app.profile, creds["providers"]) 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.assertEqual( config["providers"][app.profile]["accountId"], key["_albedo_account_id"] ) 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"}) with app.api("/server") as response: notices = json.load(response)["notices"] migrated = " ".join( notice["message"] for notice in notices if notice["kind"] == "migration" ) for name in ("auth.json", "mcp-credentials.json", "config.json"): self.assertIn(name, migrated) self.assertNotIn("old-access", migrated) self.assertNotIn("old-token", migrated) # 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): # No legacy config key: clients store and rotate only this fixture's # credential, without depending on startup migration of config.json. app = Albedo( self.provider, providers={ f"fixture-{self.provider.route}": { "baseUrl": self.provider.url, "model": "fixture-model", "protocol": "chat_completions", } }, ) app.__enter__() self.addCleanup(app.__exit__, None, None, None) server = app.profile
def patch(group, body): route = f"/settings?group={group}" with app.api(route) as response: json.load(response) revision = response.headers["ETag"] with app.api( route, body, method="PATCH", headers={ "If-Match": revision, "Content-Type": "application/merge-patch+json", }, ) as response: return json.load(response)["resource"]["value"]
patch("providers", {"profiles": {app.profile: {"api_key": "fixture-key"}}}) self.assertEqual(self.authorization(app), "Bearer fixture-key") patch("providers", {"profiles": {app.profile: {"api_key": "rotated-key"}}}) self.assertEqual(self.authorization(app), "Bearer rotated-key") value = patch( "mcp", { "definitions": { server: { "enabled": False, "transport": "http", "url": "http://127.0.0.1:1/mcp", "secrets": { "bearer_token": "secret-token", "headers": {"X-Team": "secret-team"}, }, } } }, ) self.assertEqual( value["definitions"][server]["secret_presence"], {"bearer_token": True, "headers": ["X-Team"], "environment": []}, ) value = patch( "mcp", { "definitions": { server: { "secrets": { "bearer_token": None, "environment": {"TOKEN": "secret-env"}, } } } }, ) self.assertEqual( value["definitions"][server]["secret_presence"], {"bearer_token": False, "headers": ["X-Team"], "environment": ["TOKEN"]}, ) summary = self.summary(app) self.assertFalse( any( secret in summary for secret in ( "rotated-key", "secret-token", "secret-team", "secret-env", "fixture-key", ) ) ) patch("mcp", {"definitions": {server: None}}) with (app.home / "creds.json").open() as source: self.assertNotIn(server, json.load(source).get("mcp", {}))
def backups(self, app): return list((app.home / "backups").glob("*-before-creds-*"))