"""title: CoreCI Broker author: coreci description: Single-file credential mediation broker for CoreCI agent tools. version: 0.1.0 requirements: [] licence: MIT """ import base64 import datetime import errno import hashlib import hmac import http.cookies import json import os import secrets import sqlite3 import threading import time import urllib.error import urllib.parse import urllib.request import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer try: import bcrypt except ImportError: bcrypt = None ENV_PATH_DEFAULT = ".env" ENV_KEYS = ( "BROKER_ENV_PATH", "BROKER_BIND", "BROKER_ADMIN_USER", "BROKER_ADMIN_PASS", "BROKER_INSTANCE_PEPPER", "BROKER_TOKEN_TTL", "BROKER_SESSION_TTL", "BROKER_DB_PATH", "BROKER_AUDIT_LOG_PATH", "BROKER_UI_ROOT", "BROKER_STATIC_ROOT", "BROKER_AGENT_STATE_BACKEND", "BROKER_CREDS_SOURCE", "BROKER_CREDS_FILE", "BROKER_OWUI_BASE_URL", "BROKER_OWUI_API_KEY", "BROKER_MAX_AGENTS_PER_APP", ) ENV_DEFAULTS = { "BROKER_BIND": "127.0.0.1:8765", "BROKER_ADMIN_USER": "admin", "BROKER_TOKEN_TTL": "86400", "BROKER_SESSION_TTL": "300", "BROKER_DB_PATH": "broker.db", "BROKER_AUDIT_LOG_PATH": "broker.audit.log", "BROKER_UI_ROOT": "coreci_broker_ui", "BROKER_STATIC_ROOT": ".", "BROKER_AGENT_STATE_BACKEND": "internal", "BROKER_CREDS_SOURCE": "file", "BROKER_CREDS_FILE": "data/creds.json", "BROKER_MAX_AGENTS_PER_APP": "50", } AUDIT_ACTIONS = ( "register", "read", "denied", "login", "logout", "add_app", "remove_app", "add_user", "remove_user", ) AUDIT_RESULTS = ("ok", "denied", "error") BCRYPT_PREFIXES = ("$2a$", "$2b$", "$2y$") SCRYPT_PREFIX = "scrypt$" SCRYPT_SALT_BYTES = 16 SCRYPT_N = 2 ** 14 SCRYPT_R = 8 SCRYPT_P = 1 STATIC_PREFIXES = ( ("ui", "ui_root"), ("static", "static_root"), ) STATIC_CONTENT_TYPES = { ".html": "text/html; charset=utf-8", ".css": "text/css; charset=utf-8", ".js": "application/javascript; charset=utf-8", ".py": "text/x-python; charset=utf-8", ".txt": "text/plain; charset=utf-8", ".json": "application/json; charset=utf-8", } MAX_BODY_BYTES = 1024 * 1024 MAX_STATE_BYTES = 256 * 1024 SCHEMA_SQL = """ CREATE TABLE IF NOT EXISTS broker_config ( key TEXT PRIMARY KEY, value TEXT ); CREATE TABLE IF NOT EXISTS app_registrations ( uuid TEXT PRIMARY KEY, name TEXT, status TEXT DEFAULT 'active', created_at TEXT, revoked_at TEXT ); CREATE TABLE IF NOT EXISTS user_allowlist ( user_id TEXT PRIMARY KEY, email TEXT, status TEXT DEFAULT 'active', created_at TEXT, revoked_at TEXT ); CREATE TABLE IF NOT EXISTS agent_registrations ( agent_id TEXT, app_registration_uuid TEXT, user_id TEXT, tags TEXT, token_hash TEXT, token_expires_at TEXT, first_registered_at TEXT, last_seen_at TEXT, PRIMARY KEY (agent_id, app_registration_uuid) ); CREATE INDEX IF NOT EXISTS idx_agent_registrations_app_uuid ON agent_registrations (app_registration_uuid); CREATE TABLE IF NOT EXISTS admin_sessions ( session_id TEXT PRIMARY KEY, created_at TEXT, expires_at TEXT ); CREATE TABLE IF NOT EXISTS agent_state ( agent_id TEXT, app_registration_uuid TEXT, state TEXT, updated_at TEXT, PRIMARY KEY (agent_id, app_registration_uuid) ); """ def _parse_env_file(path): entries = [] if not os.path.isfile(path): return entries with open(path, "r", encoding="utf-8") as handle: for raw in handle: line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = (part.strip() for part in line.partition("=")) if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'": value = value[1:-1] if key: entries.append((key, value)) return entries def _append_env_entries(path, entries): needs_newline = os.path.exists(path) and os.path.getsize(path) > 0 if needs_newline: with open(path, "rb") as handle: handle.seek(-1, os.SEEK_END) needs_newline = handle.read(1) != b"\n" fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) try: os.fchmod(fd, 0o600) except BaseException: os.close(fd) raise with os.fdopen(fd, "a", encoding="utf-8") as handle: if needs_newline: handle.write("\n") for key, value in entries: handle.write(f"{key}={value}\n") def _generate_pepper(): return base64.b64encode(secrets.token_bytes(32)).decode("ascii") def _generate_admin_pass(): return secrets.token_urlsafe(32) def _utc_now_iso(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def hash_token(token): token_bytes = token.encode("utf-8") if bcrypt is not None: return bcrypt.hashpw(token_bytes, bcrypt.gensalt()).decode("ascii") salt = secrets.token_bytes(SCRYPT_SALT_BYTES) digest = hashlib.scrypt(token_bytes, salt=salt, n=SCRYPT_N, r=SCRYPT_R, p=SCRYPT_P) encoded = base64.urlsafe_b64encode return f"{SCRYPT_PREFIX}{encoded(salt).decode('ascii')}${encoded(digest).decode('ascii')}" def verify_token(token, stored): if not isinstance(stored, str) or not stored: return False token_bytes = token.encode("utf-8") if stored.startswith(BCRYPT_PREFIXES): try: return bcrypt is not None and bcrypt.checkpw(token_bytes, stored.encode("ascii")) except (ValueError, TypeError): return False if stored.startswith(SCRYPT_PREFIX): fields = stored[len(SCRYPT_PREFIX):].split("$") if len(fields) != 2: return False try: salt = base64.urlsafe_b64decode(fields[0].encode("ascii")) expected = base64.urlsafe_b64decode(fields[1].encode("ascii")) actual = hashlib.scrypt(token_bytes, salt=salt, n=SCRYPT_N, r=SCRYPT_R, p=SCRYPT_P) except (ValueError, TypeError): return False return hmac.compare_digest(actual, expected) return False def derive_token(pepper, app_uuid, user_id, agent_id, tags): material = f"{pepper}:{app_uuid}:{user_id}:{agent_id}:{','.join(sorted(tags))}" return base64.b64encode(hashlib.sha256(material.encode("utf-8")).digest()).decode("ascii") def load_config(env=None): if env is None: env = os.environ env_path = str(env.get("BROKER_ENV_PATH") or ENV_PATH_DEFAULT) file_values = dict(_parse_env_file(env_path)) generated = [] if "BROKER_INSTANCE_PEPPER" not in env and "BROKER_INSTANCE_PEPPER" not in file_values: generated.append(("BROKER_INSTANCE_PEPPER", _generate_pepper())) if "BROKER_ADMIN_PASS" not in env and "BROKER_ADMIN_PASS" not in file_values: generated.append(("BROKER_ADMIN_PASS", _generate_admin_pass())) if generated: _append_env_entries(env_path, generated) file_values.update(generated) config = {"BROKER_ENV_PATH": env_path} for key in ENV_KEYS: if key == "BROKER_ENV_PATH": continue config[key] = str(env.get(key) or file_values.get(key) or ENV_DEFAULTS.get(key, "")) for key, value in generated: if key == "BROKER_ADMIN_PASS": print(f"BROKER_ADMIN_PASS={value}", flush=True) return config class Store: def __init__(self, path): self._lock = threading.Lock() fd = os.open(path, os.O_CREAT | os.O_WRONLY, 0o600) try: os.fchmod(fd, 0o600) finally: os.close(fd) self.conn = sqlite3.connect(path, check_same_thread=False) self.conn.executescript(SCHEMA_SQL) self._run( "DELETE FROM admin_sessions WHERE expires_at IS NULL OR expires_at <= ?", (_utc_now_iso(),), ) def _run(self, sql, params=()): with self._lock: rows = self.conn.execute(sql, params).fetchall() self.conn.commit() return rows def set_config(self, key, value): self._run("INSERT OR REPLACE INTO broker_config VALUES (?, ?)", (key, str(value))) def get_config(self, key): rows = self._run("SELECT value FROM broker_config WHERE key = ?", (key,)) return rows[0][0] if rows else None def create_admin_session(self, session_id, created_at, expires_at): self._run( "INSERT OR REPLACE INTO admin_sessions VALUES (?, ?, ?)", (session_id, created_at, expires_at), ) def get_admin_session(self, session_id): rows = self._run("SELECT * FROM admin_sessions WHERE session_id = ?", (session_id,)) return rows[0] if rows else None def delete_admin_session(self, session_id): self._run("DELETE FROM admin_sessions WHERE session_id = ?", (session_id,)) def list_rows(self, table): return self._run("SELECT * FROM %s ORDER BY created_at" % table) def get_row(self, table, key, value): rows = self._run("SELECT * FROM %s WHERE %s = ?" % (table, key), (value,)) return rows[0] if rows else None def add_row(self, table, value, field): self._run( "INSERT INTO %s VALUES (?, ?, 'active', ?, NULL)" % table, (value, field, _utc_now_iso()), ) def set_row_status(self, table, key, value, status): self._run( "UPDATE %s SET status = ?, revoked_at = ? WHERE %s = ?" % (table, key), (status, _utc_now_iso() if status == "revoked" else None, value), ) def get_agent(self, agent_id, app_uuid): rows = self._run( "SELECT * FROM agent_registrations WHERE agent_id = ? AND app_registration_uuid = ?", (agent_id, app_uuid), ) return rows[0] if rows else None def agents_by_uuid(self, app_uuid): return self._run( "SELECT * FROM agent_registrations WHERE app_registration_uuid = ?", (app_uuid,) ) def save_agent(self, agent_id, app_uuid, user_id, tags, token_hash, expires_at, now): self._run( "INSERT INTO agent_registrations VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (agent_id, app_uuid, user_id, tags, token_hash, expires_at, now, now), ) def update_agent(self, agent_id, app_uuid, user_id, tags, token_hash, expires_at, now): self._run( "UPDATE agent_registrations SET user_id = ?, tags = ?, token_hash = ?, " "token_expires_at = ?, last_seen_at = ? WHERE agent_id = ? AND app_registration_uuid = ?", (user_id, tags, token_hash, expires_at, now, agent_id, app_uuid), ) def touch_agent(self, agent_id, app_uuid, now): self._run( "UPDATE agent_registrations SET last_seen_at = ? WHERE agent_id = ? AND app_registration_uuid = ?", (now, agent_id, app_uuid), ) def get_agent_state(self, agent_id, app_uuid): rows = self._run("SELECT state FROM agent_state WHERE agent_id = ? AND app_registration_uuid = ?", (agent_id, app_uuid)) return json.loads(rows[0][0]) if rows else None def save_agent_state(self, agent_id, app_uuid, state): self._run("INSERT OR REPLACE INTO agent_state VALUES (?, ?, ?, ?)", (agent_id, app_uuid, state, _utc_now_iso())) def close(self): with self._lock: self.conn.close() class InternalAgentStateBackend: def __init__(self, store): self.store = store def get_state(self, agent_id, app_uuid): return self.store.get_agent_state(agent_id, app_uuid) def put_state(self, agent_id, app_uuid, state): self.store.save_agent_state(agent_id, app_uuid, json.dumps(state)) class OwuiAgentStateBackend: """Agent-state documents over HTTP: GET/PUT {base}/api/v1/broker-agent-state/{agent_id}.""" def __init__(self, base_url, api_key): self.base_url = str(base_url or "").rstrip("/") self.api_key = str(api_key or "") def _request(self, method, agent_id, payload=None): url = self.base_url + "/api/v1/broker-agent-state/" + urllib.parse.quote(agent_id, safe="") data = None if payload is None else json.dumps(payload).encode("utf-8") request = urllib.request.Request( url, data=data, method=method, headers={"Authorization": "Bearer %s" % self.api_key, "Content-Type": "application/json"}, ) with urllib.request.urlopen(request, timeout=30) as response: return response.read() def get_state(self, agent_id, app_uuid): try: raw = self._request("GET", agent_id) except urllib.error.HTTPError as error: if error.code == 404: return None raise return json.loads(raw.decode("utf-8")) if raw else None def put_state(self, agent_id, app_uuid, state): self._request("PUT", agent_id, state) def _build_agent_state_backend(config, store): if str(config.get("BROKER_AGENT_STATE_BACKEND") or "") == "owui": return OwuiAgentStateBackend( config.get("BROKER_OWUI_BASE_URL"), config.get("BROKER_OWUI_API_KEY") ) return InternalAgentStateBackend(store) class _SecretNotFound(Exception): pass class _CredsSourceError(Exception): pass def _secret_response(path, data, updated_at=None): serialized = json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8") return {"path": path, "data": data, "version": hashlib.sha256(serialized).hexdigest()[:8], "updated_at": updated_at} class FileCredsSource: def __init__(self, path): self.path = str(path or "data/creds.json") def get_secret(self, path): try: with open(self.path, "r", encoding="utf-8") as handle: document = json.load(handle) except FileNotFoundError: raise _SecretNotFound(path) except (OSError, ValueError) as error: raise _CredsSourceError(f"creds file unusable: {error}") if not isinstance(document, dict): raise _CredsSourceError("creds file must map paths to secret objects") if path not in document: raise _SecretNotFound(path) return _secret_response(path, document[path]) class OwuiApiCredsSource: """Secrets from the `creds-coreci-coo` KB via the OWUI HTTP API; a KB file's name is its secret path (e.g. `coreci/proxmox`), its content the secret JSON.""" KB_NAME = "creds-coreci-coo" def __init__(self, base_url, api_key): self.base_url = str(base_url or "").rstrip("/") self.api_key = str(api_key or "") self._file_map = None def _fetch_json(self, url_path): try: url = self.base_url + url_path request = urllib.request.Request(url, headers={"Authorization": "Bearer %s" % self.api_key}) with urllib.request.urlopen(request, timeout=30) as response: raw = response.read() except urllib.error.HTTPError as error: if error.code == 404: raise _SecretNotFound(url_path) raise _CredsSourceError(f"creds source returned HTTP {error.code} for {url_path}") except (OSError, ValueError) as error: raise _CredsSourceError(f"creds source unavailable: {error}") try: body = json.loads(raw.decode("utf-8")) except ValueError: raise _CredsSourceError(f"creds source returned invalid JSON for {url_path}") if not isinstance(body, dict): raise _CredsSourceError(f"creds source returned a non-object for {url_path}") return body def _kb_id(self): for page in range(1, 1001): body = self._fetch_json("/api/v1/knowledge/?page=%d" % page) items = body.get("items") or [] for item in items: if item.get("name") == self.KB_NAME: return item.get("id") total = body.get("total") if not items or (isinstance(total, int) and page * 30 >= total): raise _SecretNotFound(self.KB_NAME) raise _CredsSourceError("creds source pagination did not terminate") def _load_file_map(self): kb_id = self._kb_id() body = self._fetch_json("/api/v1/knowledge/%s" % urllib.parse.quote(str(kb_id), safe="")) file_map = {} for meta in body.get("files") or []: name, file_id = meta.get("name"), meta.get("id") if isinstance(name, str) and name and file_id is not None: file_map[name] = (str(file_id), meta.get("updated_at")) return file_map def get_secret(self, path): if self._file_map is None or path not in self._file_map: self._file_map = self._load_file_map() if path not in self._file_map: raise _SecretNotFound(path) file_id, updated_at = self._file_map[path] body = self._fetch_json("/api/v1/files/%s/data/content" % urllib.parse.quote(file_id, safe="")) content = body.get("content") if not isinstance(content, str): raise _CredsSourceError(f"creds file for {path} returned no content") try: data = json.loads(content) except ValueError: raise _CredsSourceError(f"creds file for {path} is invalid JSON") return _secret_response(path, data, updated_at) def _build_creds_source(config): source = str(config.get("BROKER_CREDS_SOURCE") or "file") if source == "file": return FileCredsSource(config.get("BROKER_CREDS_FILE")) if source == "owui": return OwuiApiCredsSource(config.get("BROKER_OWUI_BASE_URL"), config.get("BROKER_OWUI_API_KEY")) raise ValueError(f"BROKER_CREDS_SOURCE must be file or owui, got {source!r}") class Audit: def __init__(self, path): self.path = path self._lock, self._handle = threading.Lock(), None fd = os.open(path, os.O_CREAT | os.O_WRONLY, 0o600) try: os.fchmod(fd, 0o600) finally: os.close(fd) def write(self, action, path="", result="ok", user_id=None, app_registration_uuid=None, agent_id=None): if action not in AUDIT_ACTIONS: raise ValueError(f"unknown audit action: {action!r}") if result not in AUDIT_RESULTS: raise ValueError(f"unknown audit result: {result!r}") entry = { "ts": _utc_now_iso(), "user_id": user_id, "app_registration_uuid": app_registration_uuid, "agent_id": agent_id, "action": action, "path": path, "result": result, } line = json.dumps(entry, separators=(",", ":")) + "\n" with self._lock: if self._handle is None: self._handle = open(self.path, "a", encoding="utf-8") self._handle.write(line) self._handle.flush() os.fsync(self._handle.fileno()) def close(self): with self._lock: if self._handle is not None: self._handle.close() self._handle = None class Broker: def __init__(self, config, store, audit): self.config, self.store, self.audit = config, store, audit self.ui_root = config.get("BROKER_UI_ROOT") or "coreci_broker_ui" self.static_root = config.get("BROKER_STATIC_ROOT") or "." self.token_cache = {} self.token_cache_lock = threading.Lock() self.agent_state_backend = _build_agent_state_backend(config, store) self.creds_source = _build_creds_source(config) self.router = Router() self.router.add("GET", "/healthz", _healthz) self.router.add("POST", "/api/broker/auth/login", _auth_login) self.router.add("POST", "/api/broker/auth/logout", _auth_logout) self.router.add("GET", "/api/broker/auth/whoami", _auth_whoami) self.router.add("GET", "/api/broker/app_registrations", _apps_list) self.router.add("POST", "/api/broker/app_registrations", _apps_add) self.router.add("DELETE", "/api/broker/app_registrations/", _apps_revoke) self.router.add("POST", "/api/broker/app_registrations//reinstate", _apps_reinstate) self.router.add("GET", "/api/broker/user_allowlist", _users_list) self.router.add("POST", "/api/broker/user_allowlist", _users_add) self.router.add("DELETE", "/api/broker/user_allowlist/", _users_revoke) self.router.add("POST", "/api/broker/user_allowlist//reinstate", _users_reinstate) self.router.add("GET", "/api/broker/audit", _audit_query) self.router.add("POST", "/api/broker/agents/register", _agents_register) self.router.add("GET", "/api/broker/agents/", _require_app(_agent_meta)) self.router.add("GET", "/api/broker/agents//state", _require_app(_agent_state_get)) self.router.add("PUT", "/api/broker/agents//state", _require_app(_agent_state_put)) self.router.add("GET", "/api/broker/secrets/", _require_app(_secrets_get)) def clear_token_cache(self): with self.token_cache_lock: self.token_cache.clear() def boot(env=None): config = load_config(env) store = Store(config["BROKER_DB_PATH"]) store.set_config("BROKER_INSTANCE_PEPPER", config["BROKER_INSTANCE_PEPPER"]) audit = Audit(config["BROKER_AUDIT_LOG_PATH"]) return Broker(config, store, audit) class _BadRequest(Exception): pass def _healthz(request): request.send_json(200, {"ok": True}) def _check_admin_credentials(broker, username, password): if not isinstance(username, str) or not isinstance(password, str): return False config = broker.config user = str(config.get("BROKER_ADMIN_USER") or "").encode("utf-8") expected = str(config.get("BROKER_ADMIN_PASS") or "").encode("utf-8") return (hmac.compare_digest(username.encode("utf-8"), user) and hmac.compare_digest(password.encode("utf-8"), expected)) def _basic_credentials(request): header = request.headers.get("Authorization") or "" if header[:6].lower() != "basic ": return "", "" try: decoded = base64.b64decode(header[6:].strip(), validate=True).decode("utf-8") except ValueError: return "", "" username, sep, password = decoded.partition(":") return (username, password) if sep else ("", "") def _request_session(request): jar = http.cookies.SimpleCookie() try: jar.load(request.headers.get("Cookie") or "") except http.cookies.CookieError: return None morsel = jar.get("broker_session") if morsel is None or not morsel.value: return None row = request.broker.store.get_admin_session(morsel.value) if row is not None and row[2] > _utc_now_iso(): return row return None def _require_admin(request): if _request_session(request) is not None: return True if not request.headers.get("Authorization"): request.send_error_json(401, "unauthorized") return False username, password = _basic_credentials(request) if _check_admin_credentials(request.broker, username, password): return True request.send_error_json(401, "invalid credentials") return False def _auth_login(request): broker = request.broker body = request.body if isinstance(request.body, dict) else {} username = body.get("username") if not _check_admin_credentials(broker, username, body.get("password")): broker.audit.write("login", path="/api/broker/auth/login", result="denied") request.send_error_json(401, "invalid credentials") return session_id = secrets.token_urlsafe(32) raw_ttl = str(broker.config.get("BROKER_SESSION_TTL") or "") seconds = int(raw_ttl) if raw_ttl.isdigit() else 300 now = datetime.datetime.now(datetime.timezone.utc) expires = (now + datetime.timedelta(seconds=seconds)).isoformat() broker.store.create_admin_session(session_id, now.isoformat(), expires) broker.audit.write("login", path="/api/broker/auth/login", result="ok") cookie = "broker_session=%s; Path=/; HttpOnly; SameSite=Lax" % session_id request.send_json(200, {"ok": True, "username": username}, [("Set-Cookie", cookie)]) def _auth_logout(request): broker = request.broker if not _require_admin(request): return row = _request_session(request) if row is not None: broker.store.delete_admin_session(row[0]) broker.audit.write("logout", path="/api/broker/auth/logout", result="ok") request.send_json(200, {"ok": True}) def _auth_whoami(request): if not _require_admin(request): return row = _request_session(request) payload = {"username": request.broker.config.get("BROKER_ADMIN_USER") or "", "created_at": row[1] if row else None, "expires_at": row[2] if row else None} request.send_json(200, payload) def _list_rows(request, table, keys, field): if not _require_admin(request): return rows = request.broker.store.list_rows(table) request.send_json(200, {field: [dict(zip(keys, row)) for row in rows]}) def _apps_list(request): _list_rows(request, "app_registrations", ("uuid", "name", "status", "created_at", "revoked_at"), "registrations") def _users_list(request): _list_rows(request, "user_allowlist", ("user_id", "email", "status", "created_at", "revoked_at"), "allowlist") def _canonical_uuid(value): try: canonical = str(uuid.UUID(value)) except (ValueError, TypeError, AttributeError): return None return canonical if canonical == value else None def _required_string(value): return value if isinstance(value, str) and value else None def _has_control_chars(value): return any(ord(ch) < 32 or ord(ch) == 127 for ch in value) def _valid_id(value): if not _required_string(value): return None if value in (".", ".."): return None if ":" in value or "/" in value or "\\" in value or _has_control_chars(value): return None return value def _add_row(request, table, key, field, action, validate): if not _require_admin(request): return broker = request.broker body = request.body if isinstance(request.body, dict) else {} value = validate(body.get(key)) if value is None: request.send_error_json(400, "invalid " + key) return if broker.store.get_row(table, key, value): request.send_error_json(409, "duplicate " + key) return optional = body.get(field) broker.store.add_row(table, value, optional if isinstance(optional, str) else None) broker.audit.write(action, path="/api/broker/%s/%s" % (table, value)) request.send_json(201, {key: value, "status": "active"}) def _apps_add(request): _add_row(request, "app_registrations", "uuid", "name", "add_app", _canonical_uuid) def _users_add(request): _add_row(request, "user_allowlist", "user_id", "email", "add_user", _valid_id) def _set_row_status(request, table, key, status, action): if not _require_admin(request): return broker = request.broker value = request.path_params[key] row = broker.store.get_row(table, key, value) if row is None: request.send_error_json(404, "not found") return if row[2] != status: broker.store.set_row_status(table, key, value, status) broker.audit.write(action, path="/api/broker/%s/%s" % (table, value)) request.send_json(200, {key: value, "status": status}) def _apps_revoke(request): _set_row_status(request, "app_registrations", "uuid", "revoked", "remove_app") def _apps_reinstate(request): _set_row_status(request, "app_registrations", "uuid", "active", "add_app") def _users_revoke(request): _set_row_status(request, "user_allowlist", "user_id", "revoked", "remove_user") def _users_reinstate(request): _set_row_status(request, "user_allowlist", "user_id", "active", "add_user") def _parse_app_header(header): parts = (header or "").split() if len(parts) != 3 or parts[0] != "App": return None, None app_uuid = _canonical_uuid(parts[1]) return (app_uuid, parts[2]) if app_uuid is not None else (None, None) def _app_auth_context(row): return { "agent_id": row[0], "user_id": row[2], "app_registration_uuid": row[1], "token_expires_at": row[5], } def _verify_app_token(broker, header): app_uuid, token = _parse_app_header(header) if app_uuid is None: return None now = _utc_now_iso() key = (app_uuid, token) with broker.token_cache_lock: cached = broker.token_cache.get(key) if cached is not None: row = broker.store.get_agent(cached["agent_id"], app_uuid) if row is not None and row[5] and row[5] > now: return _app_auth_context(row) with broker.token_cache_lock: broker.token_cache.pop(key, None) return None for row in broker.store.agents_by_uuid(app_uuid): if row[5] and row[5] > now and verify_token(token, row[4]): context = _app_auth_context(row) with broker.token_cache_lock: broker.token_cache[key] = context return context return None def _app_token_valid(broker, header): return _verify_app_token(broker, header) is not None def _audit_app_denied(request): parts = (request.headers.get("Authorization") or "").split() if len(parts) < 2 or parts[0] != "App": return request.broker.audit.write( "denied", path=request.path, result="denied", app_registration_uuid=_canonical_uuid(parts[1]), ) def _require_app(handler): def wrapped(request): context = _verify_app_token(request.broker, request.headers.get("Authorization")) if context is None: _audit_app_denied(request) request.send_error_json(401, "invalid app credentials") return request.app_auth = context request.broker.store.touch_agent( context["agent_id"], context["app_registration_uuid"], _utc_now_iso() ) handler(request) return wrapped def _agents_register(request): broker = request.broker body = request.body if isinstance(request.body, dict) else {} agent_id = body.get("agent_id") app_uuid = body.get("app_registration_uuid") user_id = body.get("user_id") tags = body.get("tags") strings = [v for v in (agent_id, app_uuid, user_id) if isinstance(v, str) and v] if len(strings) != 3 or not isinstance(tags, list) or not all(isinstance(v, str) for v in tags): request.send_error_json(400, "invalid registration") return if _valid_id(agent_id) is None: request.send_error_json(400, "invalid agent_id") return if any(":" in tag or _has_control_chars(tag) for tag in tags): request.send_error_json(400, "invalid tags") return path = "/api/broker/agents/register" identity = {"user_id": user_id, "app_registration_uuid": app_uuid, "agent_id": agent_id} app_row = broker.store.get_row("app_registrations", "uuid", app_uuid) user_row = broker.store.get_row("user_allowlist", "user_id", user_id) if app_row is None or app_row[2] != "active" or user_row is None or user_row[2] != "active": broker.audit.write("register", path=path, result="denied", **identity) request.send_error_json(403, "registration not allowed") return max_raw = str(broker.config.get("BROKER_MAX_AGENTS_PER_APP") or "") max_agents = int(max_raw) if max_raw.isdigit() else 50 if len(broker.store.agents_by_uuid(app_uuid)) >= max_agents: request.send_error_json(400, "too many agents for this app") return pepper = broker.store.get_config("BROKER_INSTANCE_PEPPER") or broker.config["BROKER_INSTANCE_PEPPER"] token = derive_token(pepper, app_uuid, user_id, agent_id, tags) raw_ttl = str(broker.config.get("BROKER_TOKEN_TTL") or "") ttl = int(raw_ttl) if raw_ttl.isdigit() else 86400 now = datetime.datetime.now(datetime.timezone.utc) expires = (now + datetime.timedelta(seconds=ttl)).isoformat() tags_json = json.dumps(sorted(tags)) row = broker.store.get_agent(agent_id, app_uuid) created = row is None if created: try: broker.store.save_agent(agent_id, app_uuid, user_id, tags_json, hash_token(token), expires, now.isoformat()) except sqlite3.IntegrityError: row = broker.store.get_agent(agent_id, app_uuid) if row is None: raise created = False if not created: if row[2] == user_id and row[3] == tags_json and row[5] and row[5] > now.isoformat(): expires = row[5] broker.store.touch_agent(agent_id, app_uuid, now.isoformat()) else: broker.store.update_agent(agent_id, app_uuid, user_id, tags_json, hash_token(token), expires, now.isoformat()) with broker.token_cache_lock: for cached_key in [k for k in broker.token_cache if k[0] == app_uuid]: broker.token_cache.pop(cached_key, None) payload = {"token": token, "token_expires_at": expires, "created": created} if created or not _app_token_valid(broker, request.headers.get("Authorization")): payload["pepper"] = pepper broker.audit.write("register", path=path, **identity) request.send_json(201 if created else 200, payload) def _agent_identity_row(request): auth = request.app_auth row = request.broker.store.get_agent( request.path_params["agent_id"], auth["app_registration_uuid"] ) if row is None: request.send_error_json(404, "not found") return None if row[0] != auth["agent_id"] or row[2] != auth["user_id"]: request.send_error_json(403, "agent identity mismatch") return None return row def _agent_meta(request): row = _agent_identity_row(request) if row is None: return request.send_json(200, { "agent_id": row[0], "app_registration_uuid": row[1], "user_id": row[2], "tags": json.loads(row[3]), "token_expires_at": row[5], "first_registered_at": row[6], "last_seen_at": row[7], }) def _agent_state_get(request): row = _agent_identity_row(request) if row is None: return state = request.broker.agent_state_backend.get_state(row[0], row[1]) if state is None: request.send_error_json(404, "not found") return request.send_json(200, state) def _agent_state_put(request): row = _agent_identity_row(request) if row is None: return body = request.body if not isinstance(body, dict): request.send_error_json(400, "state must be a JSON object") return if len(json.dumps(body).encode("utf-8")) > MAX_STATE_BYTES: request.send_error_json(400, "state too large") return request.broker.agent_state_backend.put_state(row[0], row[1], body) request.send_json(200, {"ok": True}) def _secret_path_unsafe(path): return (path.startswith("/") or "\\" in path or "\x00" in path or ".." in path.split("/")) def _secrets_denied(request, path, identity): request.broker.audit.write("read", path=path, result="denied", **identity) request.send_error_json(403, "forbidden") def _secrets_get(request): broker = request.broker auth = request.app_auth path = urllib.parse.unquote(request.path_params["path"]) identity = {"user_id": auth["user_id"], "app_registration_uuid": auth["app_registration_uuid"], "agent_id": auth["agent_id"]} if _secret_path_unsafe(path): _secrets_denied(request, path, identity) return if path != "coreci/" + auth["agent_id"]: _secrets_denied(request, path, identity) return try: secret = broker.creds_source.get_secret(path) except _SecretNotFound: broker.audit.write("read", path=path, result="error", **identity) request.send_error_json(404, "not found") except _CredsSourceError: broker.audit.write("read", path=path, result="error", **identity) request.send_error_json(500, "creds source unavailable") else: broker.audit.write("read", path=path, **identity) request.send_json(200, secret) def _parse_iso_ts(value): try: parsed = datetime.datetime.fromisoformat(value.replace("Z", "+00:00")) except (ValueError, AttributeError): return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=datetime.timezone.utc) return parsed def _audit_entry_matches(entry, query, since, until): for field in ("user_id", "app_registration_uuid", "action"): if query.get(field) and entry.get(field) != query[field]: return False if since is None and until is None: return True parsed = _parse_iso_ts(entry.get("ts") or "") if parsed is None: return False if since is not None and parsed <= since: return False if until is not None and parsed > until: return False return True def _audit_query(request): if not _require_admin(request): return query = request.query since = until = None if query.get("since"): since = _parse_iso_ts(query["since"]) if since is None: request.send_error_json(400, "invalid since") return if query.get("until"): until = _parse_iso_ts(query["until"]) if until is None: request.send_error_json(400, "invalid until") return try: limit = int(query.get("limit") or "") except ValueError: limit = 100 limit = min(max(limit, 1), 1000) entries = [] with open(request.broker.audit.path, "r", encoding="utf-8") as handle: for raw in handle: if len(entries) >= limit: break line = raw.strip() if not line: continue try: entry = json.loads(line) except ValueError: continue if not isinstance(entry, dict): continue if not _audit_entry_matches(entry, query, since, until): continue entries.append(entry) next_since = entries[-1].get("ts") if len(entries) >= limit else None request.send_json(200, {"entries": entries, "next_since": next_since}) def _parse_query(query): values = urllib.parse.parse_qs(query, keep_blank_values=True) return {key: value[0] for key, value in values.items()} def _match_pattern(pattern, path): pattern_parts = pattern.split("/") path_parts = path.split("/") if len(pattern_parts) > len(path_parts): return None params = {} for index, expected in enumerate(pattern_parts): if not (len(expected) > 2 and expected.startswith("<") and expected.endswith(">")): if expected != path_parts[index]: return None continue name = expected[1:-1] if name.endswith("..."): if index + 1 != len(pattern_parts) or not path_parts[index:]: return None params[name[:-3]] = "/".join(path_parts[index:]) return params if not path_parts[index]: return None params[name] = path_parts[index] if len(pattern_parts) != len(path_parts): return None return params def _static_content_type(name): extension = os.path.splitext(name)[1].lower() return STATIC_CONTENT_TYPES.get(extension, "application/octet-stream") def _broker_protected_paths(broker): protected = set() for key in ("BROKER_ENV_PATH", "BROKER_DB_PATH", "BROKER_AUDIT_LOG_PATH", "BROKER_CREDS_FILE"): value = broker.config.get(key) or "" if not value: continue base = os.path.realpath(value) protected.add(base) for suffix in ("-journal", "-wal", "-shm"): protected.add(base + suffix) return protected def _resolve_static_file(root, rest, protected=()): if "\x00" in rest or "\\" in rest or rest.startswith("/"): return None parts = rest.split("/") if ".." in parts or any(part.startswith(".") for part in parts): return None stripped = rest.strip("/") names = [] if stripped: names.append(stripped) if "." not in os.path.basename(stripped): names.append(stripped + ".html") else: names.append("index.html") root_real = os.path.realpath(root) for name in names: target = os.path.realpath(os.path.join(root_real, name)) if not target.startswith(root_real + os.sep): continue if target in protected: continue if os.path.isfile(target): return target return None class Router: def __init__(self): self._routes = [] def add(self, method, pattern, handler): self._routes.append((method.upper(), pattern, handler)) def resolve(self, method, path): for route_method, pattern, handler in self._routes: if route_method != method: continue params = _match_pattern(pattern, path) if params is not None: return handler, params return None, None class BrokerRequestHandler(BaseHTTPRequestHandler): broker = None timeout = 30 responded = False path_params = None query = None body = None def log_message(self, format, *args): pass def do_GET(self): self._dispatch("GET") def do_POST(self): self._dispatch("POST") def do_PUT(self): self._dispatch("PUT") def do_DELETE(self): self._dispatch("DELETE") def send_json(self, status, payload, headers=()): if self.responded: return self.responded = True data = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") for key, value in headers: self.send_header(key, value) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def send_error_json(self, status, message): self.send_json(status, {"error": status, "message": message}) def _reject_oversized_body(self): header = self.headers.get("Content-Length") if not header: return try: length = int(header) except ValueError: return if length > MAX_BODY_BYTES: raise _BadRequest("body too large") def _read_body(self): header = self.headers.get("Content-Length") or "0" try: length = int(header) except ValueError: raise _BadRequest("invalid Content-Length") if length > MAX_BODY_BYTES: raise _BadRequest("body too large") if length <= 0: return b"" return self.rfile.read(length) def _dispatch(self, method): self.path_params, self.query, self.body = {}, {}, None try: self._reject_oversized_body() parsed = urllib.parse.urlsplit(self.path) path = urllib.parse.unquote(parsed.path) self.query = _parse_query(parsed.query) raw = self._read_body() if method in ("POST", "PUT") else b"" if self._serve_static(method, path): return handler, params = self.broker.router.resolve(method, path) if handler is None: self.send_error_json(404, "not found") return if raw: try: self.body = json.loads(raw) except (ValueError, RecursionError): self.send_error_json(400, "invalid JSON body") return self.path_params = params handler(self) except _BadRequest as error: self.send_error_json(400, str(error)) except Exception: if not self.responded: self.send_error_json(500, "internal error") def _serve_static(self, method, path): if method != "GET": return False for prefix, attr in STATIC_PREFIXES: if path.startswith("/" + prefix + "/"): self._send_static_file(getattr(self.broker, attr), path[len(prefix) + 2:]) return True return False def _send_static_file(self, root, rest): target = _resolve_static_file( root, rest, _broker_protected_paths(self.broker) ) if target is None: self.send_error_json(404, "not found") return with open(target, "rb") as handle: data = handle.read() self.send_response(200) self.send_header("Content-Type", _static_content_type(target)) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def serve(host, port, broker): handler = type("BoundHandler", (BrokerRequestHandler,), {"broker": broker}) server = ThreadingHTTPServer((host, port), handler) server.broker = broker thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server _server_lock = threading.Lock() _started_server = None def _parse_bind(bind): host, _, port = str(bind).rpartition(":") if not port.isdigit(): raise ValueError(f"BROKER_BIND must be host:port, got {bind!r}") return host or "127.0.0.1", int(port) def _start_guarded_server(env=None): global _started_server with _server_lock: if _started_server is not None: return _started_server broker = boot(env) host, port = _parse_bind(broker.config["BROKER_BIND"]) _started_server = serve(host, port, broker) return _started_server def _reset_server_guard(): global _started_server with _server_lock: server = _started_server _started_server = None if server is None: return None server.shutdown() server.server_close() server.broker.clear_token_cache() server.broker.audit.close() server.broker.store.close() return server class Tools: class Valves: pass def __init__(self): self.valves = self.Valves() try: self._server = _start_guarded_server() except OSError as error: if error.errno != errno.EADDRINUSE: raise raise RuntimeError( f"broker bind failed ({error}): a broker from previously loaded " "module code is still serving the configured bind; restart " "Open WebUI to load the updated broker" ) from error def status(self) -> str: """Report the broker server state as a one-line string.""" host, port = self._server.server_address[:2] return f"coreci-broker serving on http://{host}:{port}" def run_standalone(env=None): broker = boot(env) host, port = _parse_bind(broker.config["BROKER_BIND"]) server = serve(host, port, broker) bound_host, bound_port = server.server_address[:2] print(f"coreci-broker listening on http://{bound_host}:{bound_port}", flush=True) try: while True: time.sleep(1) except KeyboardInterrupt: pass finally: server.shutdown() server.server_close() broker.clear_token_cache() broker.audit.close() broker.store.close() print("coreci-broker shut down", flush=True) if __name__ == "__main__": run_standalone()