"""Fake monky-tenancy agent endpoint + fake OpenBao, as real HTTP servers on 127.0.0.1. They implement exactly the shapes of record (ADR-0028 + Gate 1 v2): the lease is a `{login_jwt, ttl_s, mount, role, vault}` deploy grant; OpenBao's `auth/jwt-tenancy/login` accepts any grant listed in `bao.grants` and pins the token's meta to that grant's env_id.""" from __future__ import annotations import io import json import tarfile import threading import time import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from monky_deployd.bundle import bundle_sha ENV = "env-qa-02" COMPOSE = """services: see: image: harbor.tikali.net/monky/see-backend:${SEE_TAG:-develop} environment: POSTGRES_PASSWORD: ${POSTGRES_PASSWORD} GEMINI_API_KEY: ${GEMINI_API_KEY} SEE_ADMIN_TOKEN: ${SEE_ADMIN_TOKEN} healthcheck: test: ["CMD", "true"] """ def make_manifest(env=ENV, versions=None, prefix="monky"): versions = versions or {} entries = [ { "var": "GEMINI_API_KEY", "path": f"{prefix}/{env}/see/gemini_api_key", "kind": "supplied", "version": versions.get("gemini_api_key"), }, { "var": "POSTGRES_PASSWORD", "path": f"{prefix}/{env}/see/pg_password", "kind": "generated", "version": versions.get("pg_password", 1), }, { "var": "SEE_ADMIN_TOKEN", "path": f"{prefix}/{env}/see/admin_token", "kind": "generated", "version": versions.get("admin_token", 1), }, ] return {"vault_mode": "openbao", "entries": entries} def env_template(manifest): lines = ["# .env template — PLACEHOLDERS ONLY", ""] for e in manifest["entries"]: lines += [f"# [{e['kind']}] openbao: {e['path']}", f"{e['var']}=${{{e['var']}}}"] return "\n".join(lines) + "\n" def make_files(env=ENV, *, compose=COMPOSE, manifest=None, meta=None): manifest = manifest or make_manifest(env) files = { "docker-compose.yml": compose, ".env.template": env_template(manifest), "secrets.manifest.json": json.dumps(manifest, indent=2, sort_keys=True) + "\n", } bundle_json = { "env_id": env, "tier": env.split("-")[1] if env.startswith("env-") else "dev", "site": "cbs", "bundle": "see", "target_class": "docker-host", "renderer": "compose", "images_policy": "develop", "secrets_provider": "openbao", "agent": {}, "files": sorted([*files, "bundle.json"]), } if meta: bundle_json.update(meta) files["bundle.json"] = json.dumps(bundle_json, indent=2, sort_keys=True) + "\n" return files def tar_bytes(files: dict[str, str]) -> bytes: buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tar: for name in sorted(files): data = files[name].encode() info = tarfile.TarInfo(name=name) info.size = len(data) info.mode = 0o644 tar.addfile(info, io.BytesIO(data)) return buf.getvalue() class _Server(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True class _Handler(BaseHTTPRequestHandler): server_version = "fake/0" protocol_version = "HTTP/1.1" def log_message(self, *a): # quiet pass def _body(self): n = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(n) if n else b"" return json.loads(raw) if raw else {} def _send(self, status, body=None, *, raw=None, headers=None, ctype="application/json"): data = raw if raw is not None else (json.dumps(body).encode() if body is not None else b"") self.send_response(status) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(data))) for k, v in (headers or {}).items(): self.send_header(k, v) self.end_headers() self.wfile.write(data) def do_GET(self): self.server.app.handle(self, "GET", self.path, {}) def do_POST(self): self.server.app.handle(self, "POST", self.path, self._body()) class FakeBao: """auth/jwt-tenancy/login, token lookup-self/renew-self/revoke-self, KV-v2 reads.""" def __init__(self, env=ENV): self.env = env self.grants: dict[str, str] = {} # jwt -> env_id self.tokens: dict[str, dict] = {} self.kv: dict[str, list[dict]] = {} # "/see/" -> versions [ {value} ] self.calls: list[tuple] = [] self.login_fail = False self.token_ttl = 86400 self.max_ttl = 30 * 86400 self.mount = "jwt-tenancy" self.role = "see-env" self.kv_mount = "monky" self.seed_secrets(env) def seed_secrets(self, env): self.kv[f"{env}/see/gemini_api_key"] = [ {"value": "AIzaSy-FAKE-GEMINI-KEY-0001"}, {"value": "AIzaSy-FAKE-GEMINI-KEY-0002"}, ] self.kv[f"{env}/see/pg_password"] = [{"value": 'pg-s3cret-"quoted"$dollar'}] self.kv[f"{env}/see/admin_token"] = [{"value": "adm1n-t0ken-abcdef0123456789"}] def grant(self, env=None, jti=None) -> str: jwt = "eyJhbGciOiJFUzI1NiJ9." + uuid.uuid4().hex + uuid.uuid4().hex + "." + uuid.uuid4().hex + uuid.uuid4().hex self.grants[jwt] = json.dumps({"env_id": env or self.env, "jti": jti or uuid.uuid4().hex}) return jwt def mint(self, env=None, *, age_s=0, ttl=None) -> str: tok = "hvs." + uuid.uuid4().hex + uuid.uuid4().hex[:8] self.tokens[tok] = { "accessor": "acc-" + uuid.uuid4().hex[:12], "env_id": env or self.env, "grant_jti": uuid.uuid4().hex, "creation_time": int(time.time()) - age_s, "ttl": ttl if ttl is not None else self.token_ttl, "renewable": True, "revoked": False, } return tok def _auth(self, h): tok = h.headers.get("X-Vault-Token") t = self.tokens.get(tok) if not t or t["revoked"]: return None, tok return t, tok def handle(self, h, method, path, body): self.calls.append((method, path.split("?")[0])) if method == "POST" and path == f"/v1/auth/{self.mount}/login": if self.login_fail: return h._send(400, {"errors": ["error validating token: expired"]}) if body.get("role") != self.role: return h._send(400, {"errors": [f"role {body.get('role')!r} could not be found"]}) meta = self.grants.pop(body.get("jwt", ""), None) # single-use grant if meta is None: return h._send(400, {"errors": ["error validating token: unknown or already used grant"]}) m = json.loads(meta) tok = self.mint(m["env_id"]) self.tokens[tok]["grant_jti"] = m["jti"] t = self.tokens[tok] return h._send( 200, { "auth": { "client_token": tok, "accessor": t["accessor"], "lease_duration": t["ttl"], "renewable": True, "token_policies": ["see-env"], "metadata": {"env_id": m["env_id"], "grant_jti": m["jti"], "role": self.role}, } }, ) t, tok = self._auth(h) if t is None: return h._send(403, {"errors": ["permission denied"]}) if method == "GET" and path == "/v1/auth/token/lookup-self": return h._send( 200, { "data": { "accessor": t["accessor"], "creation_time": t["creation_time"], "ttl": t["ttl"], "renewable": t["renewable"], "explicit_max_ttl": self.max_ttl, "meta": {"env_id": t["env_id"], "grant_jti": t["grant_jti"]}, } }, ) if method == "POST" and path == "/v1/auth/token/renew-self": age = int(time.time()) - t["creation_time"] t["ttl"] = max(0, min(self.token_ttl, self.max_ttl - age)) return h._send(200, {"auth": {"client_token": tok, "lease_duration": t["ttl"], "renewable": True}}) if method == "POST" and path == "/v1/auth/token/revoke-self": t["revoked"] = True return h._send(204) if method == "GET" and path.startswith(f"/v1/{self.kv_mount}/data/"): p, _, q = path.partition("?") key = p[len(f"/v1/{self.kv_mount}/data/") :] env = key.split("/")[0] if env != t["env_id"]: return h._send(403, {"errors": ["1 error occurred:\n\t* permission denied\n\n"]}) versions = self.kv.get(key) if not versions: return h._send(404, {"errors": []}) want = None for part in q.split("&"): if part.startswith("version="): want = int(part.split("=", 1)[1]) idx = (want or len(versions)) - 1 if idx < 0 or idx >= len(versions): return h._send(404, {"errors": []}) return h._send(200, {"data": {"data": versions[idx], "metadata": {"version": idx + 1}}}) return h._send(404, {"errors": [f"no handler for {method} {path}"]}) class FakeTenancy: """The agent entrypoint. `bao` validates bearers (a token for another env -> AGENT_ENV_MISMATCH).""" def __init__(self, bao: FakeBao, env=ENV): self.bao = bao self.env = env self.files = make_files(env) self.desired_sha = bundle_sha(self.files) self.action = "apply" self.purge_volumes = False self.checkins: list[dict] = [] self.reports: list[dict] = [] self.leases: list[dict] = [] self.lease_shape = "jwt" # or "approle" to simulate the un-migrated tenancy self.lease_limit = 5 self.superseded_jtis: set[str] = set() self.down = False # network down -> connection refused simulated by test stopping server self.kv_mount = "monky" def set_files(self, files): self.files = files self.desired_sha = bundle_sha(files) def _principal(self, h): auth = h.headers.get("Authorization", "") tok = auth.split(" ", 1)[1] if auth.lower().startswith("bearer ") else None t = self.bao.tokens.get(tok or "") if not t or t["revoked"]: return None, ("AGENT_UNAUTHENTICATED", "token unknown, expired or revoked") if t["grant_jti"] in self.superseded_jtis: return None, ("AGENT_UNAUTHENTICATED", "grant superseded") return t, None def _err(self, h, status, code, detail, headers=None): return h._send(status, {"code": code, "detail": detail}, headers=headers) def handle(self, h, method, path, body): t, err = self._principal(h) if err: return self._err(h, 401, *err) env_in = body.get("env_id") if body else None if path.startswith("/v1/agent/bundle/"): env_in = path.split("/")[4] if env_in and env_in != t["env_id"]: return self._err(h, 403, "AGENT_ENV_MISMATCH", f"token is pinned to {t['env_id']}, not {env_in}") if method == "POST" and path == "/v1/agent/checkin": self.checkins.append(body) action = self.action if action == "apply" and body.get("applied_sha") == self.desired_sha: action = "none" return h._send( 200, { "env_id": self.env, "desired_sha": None if self.action == "down" else self.desired_sha, "action": action, "purge_volumes": self.purge_volumes if action == "down" else False, "bundle_url": f"/v1/agent/bundle/{self.env}/{self.desired_sha}", "checkin_interval_s": 60, "vault": { "addr": "https://bao.cbs.tikali.net:8200", "mount": self.kv_mount, "prefix": f"{self.env}/see", }, }, ) if method == "GET" and path.startswith("/v1/agent/bundle/"): sha = path.rsplit("/", 1)[1] if sha != self.desired_sha: return self._err(h, 404, "BUNDLE_NOT_FOUND", f"no published bundle {sha}") return h._send( 200, raw=tar_bytes(self.files), ctype="application/x-tar", headers={"Cache-Control": "no-store", "X-Bundle-Sha": sha}, ) if method == "POST" and path == "/v1/agent/lease": if len(self.leases) >= self.lease_limit: return self._err(h, 429, "LEASE_RATE_LIMITED", "5 leases per hour", headers={"Retry-After": "600"}) self.leases.append(body) if self.lease_shape == "approle": return h._send( 200, { "env_id": self.env, "wrapping_token": "hvs.wrap-legacy", "wrap_ttl_s": 3600, "role_id": "role-id-see-env", "vault": {}, }, ) jwt = self.bao.grant(self.env) return h._send( 200, { "env_id": self.env, "login_jwt": jwt, "ttl_s": 3600, "mount": self.bao.mount, "role": self.bao.role, # tenancy AgentLeaseOut: `addr` is top-level; KV mount/prefix come from checkin "addr": "https://bao.cbs.tikali.net:8200", }, headers={"Cache-Control": "no-store"}, ) if method == "POST" and path == "/v1/agent/report": self.reports.append(body) if body.get("result") not in ("applied", "failed", "down"): return self._err(h, 400, "BAD_REQUEST", "unknown result") return h._send(200, {"env_id": self.env, "accepted": True, "result": body["result"]}) return self._err(h, 404, "NOT_FOUND", path) def serve(app): srv = _Server(("127.0.0.1", 0), _Handler) srv.app = app th = threading.Thread(target=srv.serve_forever, daemon=True) th.start() return srv, srv.server_address[1]