""" Member Portal — membership middleware for the Inference Cooperative. Reconciles three systems: 1. Open Collective — who is a paying member (billing) 2. Cloudron — who can log in (identity/SSO) 3. LiteLLM — who can use the models, and how much (inference) Three responsibilities: A. Webhook handler — listen for Open Collective membership events B. Key injector — read the member's email from a header, inject their LiteLLM key, and forward the request to the gateway C. Admin endpoints — health, status, manual reconciliation """ import os import json import logging import re import sqlite3 import secrets import smtplib from datetime import datetime, timedelta, timezone from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart # --- Key encryption at rest (inlined from keycrypto.py; see repo history) --- # Member chat keys are encrypted with AES-256-GCM using a key-encryption key # (KEK) from the PORTAL_KEK env var. Plaintext exists only in process memory # when a key is served. Format: "enc1:" + base64(nonce[12] || ct || tag). # Values without the prefix are legacy plaintext (pre-migration) and pass # through on decrypt. Fail-closed: nothing new is stored plaintext when # PORTAL_KEK is unset, and encrypted values cannot be served without it. import base64 as _b64 from cryptography.hazmat.primitives.ciphers.aead import AESGCM as _AESGCM _KEYENC_PREFIX = "enc1:" _KEYENC_AAD = b"member-chat-key-v1" def _kek() -> bytes: raw = os.environ.get("PORTAL_KEK", "") if not raw: raise RuntimeError( "PORTAL_KEK is not set - refusing to encrypt/decrypt member keys" ) key = _b64.b64decode(raw) if len(key) != 32: raise RuntimeError("PORTAL_KEK must decode to exactly 32 bytes (base64)") return key def encrypt_key(plain: str | None) -> str | None: """Encrypt a member key. Already-encrypted and empty values pass through.""" if not plain or plain.startswith(_KEYENC_PREFIX): return plain nonce = os.urandom(12) ct = _AESGCM(_kek()).encrypt(nonce, plain.encode(), _KEYENC_AAD) return _KEYENC_PREFIX + _b64.b64encode(nonce + ct).decode() def decrypt_key(stored: str | None) -> str | None: """Decrypt a member key. Legacy plaintext rows pass through unchanged.""" if not stored: return stored if not stored.startswith(_KEYENC_PREFIX): return stored # pre-migration plaintext blob = _b64.b64decode(stored[len(_KEYENC_PREFIX):]) return _AESGCM(_kek()).decrypt(blob[:12], blob[12:], _KEYENC_AAD).decode() def migrate_plaintext_keys(conn) -> int: """Encrypt plaintext key_token rows. Idempotent; returns rows changed.""" if not os.environ.get("PORTAL_KEK"): logging.getLogger("uvicorn.error").critical( "PORTAL_KEK unset - member keys remain PLAINTEXT at rest; " "set PORTAL_KEK to a base64 32-byte key and restart" ) return 0 rows = conn.execute( "SELECT email, key_token FROM members " "WHERE key_token != '' AND key_token IS NOT NULL " "AND key_token NOT LIKE 'enc1:%'" ).fetchall() changed = 0 for email, kt in rows: enc = encrypt_key(kt) if enc != kt: conn.execute( "UPDATE members SET key_token = ? WHERE email = ?", (enc, email) ) changed += 1 return changed # --- end key encryption --- import httpx import jwt import asyncio from fastapi import FastAPI, Request, Response, HTTPException from fastapi.responses import JSONResponse, StreamingResponse from slowapi import Limiter from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded logging.basicConfig(level=logging.INFO) logger = logging.getLogger("member-portal") # SECURITY (ops/security#2): OC's generic webhooks cannot send headers, so the shared # secret must travel in the URL path — which means uvicorn writes it into the access # log on every delivery, and the log is read by more things than the secret is. The # token cannot be kept out of the URL; it can be kept out of the retention. Filters # sit on the access logger and on the root handler so both uvicorn's line and any # string the app itself logs get scrubbed. _WEBHOOK_TOKEN_RE = re.compile(r"(/webhook/opencollective/)[^\s\"'?&]+") def _redact_webhook_path(value): if isinstance(value, str) and "/webhook/opencollective/" in value: return _WEBHOOK_TOKEN_RE.sub(r"\1", value) return value class _RedactWebhookPath(logging.Filter): """Replace the webhook token in any logged request path with .""" def filter(self, record: logging.LogRecord) -> bool: try: record.msg = _redact_webhook_path(record.msg) if isinstance(record.args, tuple): record.args = tuple(_redact_webhook_path(a) for a in record.args) elif isinstance(record.args, dict): record.args = {k: _redact_webhook_path(v) for k, v in record.args.items()} except Exception: pass # a logging filter must never break logging return True logging.getLogger("uvicorn.access").addFilter(_RedactWebhookPath()) for _handler in logging.getLogger().handlers: _handler.addFilter(_RedactWebhookPath()) app = FastAPI(title="Inference Cooperative Member Portal") # --- Rate limiting (control plane only, not the member inference path) --- # The data plane (chat completions via /v1/*) is capped by each member's # LiteLLM budget, NOT by rate — a member streaming a long conversation must # never be throttled. Rate limiting here covers the control-plane surfaces: # admin endpoints, the broker, the webhook, and the unauthenticated model list. limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter @app.exception_handler(RateLimitExceeded) async def _rate_limit_handler(request: Request, exc: RateLimitExceeded): return JSONResponse( status_code=429, content={"detail": "Too many requests. Slow down."}, ) # --- Configuration (from environment) --- CLOUDRON_API = os.environ.get("CLOUDRON_API_ORIGIN", "https://my.inference.coop") CLOUDRON_TOKEN = os.environ.get("CLOUDRON_TOKEN", "") LITELLM_BASE = os.environ.get("LITELLM_BASE", "https://gateway.inference.coop") LITELLM_MASTER_KEY = os.environ.get("LITELLM_MASTER_KEY", "") # Shared secret that the chat frontend uses to authenticate requests to the # portal. LibreChat sends it as an X-Portal-Secret header; OpenWebUI signs the # user's email as a JWT (see FORWARD_USER_INFO_HEADER_JWT_SECRET below). PORTAL_SECRET = os.environ.get("PORTAL_SECRET", "") # OpenWebUI can sign the forwarded user identity as a JWT using this shared # secret. When set, the portal verifies the JWT (HS256) to confirm the email # genuinely came from OpenWebUI (behind Cloudron SSO) rather than a spoofed # header. This is stronger than the plain X-Portal-Secret header because the # email itself is tamper-proof. OWUI_JWT_SECRET = os.environ.get("OWUI_JWT_SECRET", "") # Open WebUI admin API key — used at provisioning time to add the new member to # Open WebUI's `members` group, which gates read access to the shared # "Inference Cooperative Docs" RAG collection. Without this, a new member's # first chat session has no docs grounding (the nightly refresh would only fix # it hours later). Machine-to-machine only; never logged. OWUI_API_KEY = os.environ.get("OWUI_API_KEY", "") OWUI_BASE = os.environ.get("OWUI_BASE", "http://chat:8080") # internal Cloudron network OWUI_MEMBERS_GROUP_ID = os.environ.get("OWUI_MEMBERS_GROUP_ID", "eb348e1a-059a-4a8f-8920-0f73942f6272") # Open WebUI knowledge collection id for the co-op docs RAG. Attached to every # model in /v1/models so answers are grounded by automatic RAG injection (and # the model's knowledge tools are scoped to the docs rather than browsing all # knowledge bases). Env-driven; empty disables the attachment. DOCS_KNOWLEDGE_ID = os.environ.get("DOCS_KNOWLEDGE_ID", "946e18ef-c1bc-4c57-b23e-7f5fc5b66e51") # Shared secret that the member dashboard uses to authenticate broker requests. # The dashboard is Cloudron-SSO-gated and passes the authenticated member's # email; the portal trusts that email only because it carries this secret. Kept # separate from PORTAL_SECRET so the two can be rotated independently. BROKER_SECRET = os.environ.get("BROKER_SECRET", "") # Secret token required in the Open Collective webhook URL path. Open # Collective's generic webhooks are not HMAC-signed, so a secret in the URL # is the standard way to authenticate them. WEBHOOK_TOKEN = os.environ.get("WEBHOOK_TOKEN", "") # SECURITY (2026-10-27 member disclosure): the /admin/* token is SEPARATE from # WEBHOOK_TOKEN. WEBHOOK_TOKEN must sit in the Open Collective webhook URL (OC's # generic webhooks cannot send headers), so anything that learns that URL — an OC # admin, OC's logs, this app's own access logs, a Referer — used to gain portal # admin. ADMIN_TOKEN is only ever sent as a header and never appears in a URL. # Cutover COMPLETE (2026-10-28): ADMIN_TOKEN_LEGACY=0, so the legacy token no # longer authenticates here (verified: ADMIN_TOKEN -> 200, WEBHOOK_TOKEN -> 403). # ADMIN_TOKEN was minted new on 2026-10-27 and WEBHOOK_TOKEN was rotated the same # day, so NO rotation is outstanding — do not "finish" this by rotating again. # Callers must read ADMIN_TOKEN and fail loudly; never fall back to WEBHOOK_TOKEN, # which 403s and would turn a missing variable into a mid-run failure. ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "") ADMIN_TOKEN_LEGACY = os.environ.get("ADMIN_TOKEN_LEGACY", "1") != "0" OC_GRAPHQL_URL = "https://opencollective.com/api/graphql/v2" OC_COLLECTIVE_SLUG = os.environ.get("OC_COLLECTIVE_SLUG", "inference-cooperative") # Cache of order idV2 -> tip-free contribution amount in cents, resolved via # OC's PUBLIC GraphQL API (no token needed; our personal token lacks the # orders scope but unauthenticated reads of public orders work). _order_amount_cache: dict[str, int] = {} async def get_order_amount(order_id_v2: str | None) -> int | None: """Look up an order's contribution amount (EXCLUDING the platform tip). The webhook's `totalAmount` includes the buyer's optional OC tip, which the co-op never receives — credits must be granted on the tip-free amount. Returns cents, or None if unresolvable. """ if not order_id_v2: return None if order_id_v2 in _order_amount_cache: return _order_amount_cache[order_id_v2] try: query = ('{ order(order: {id: "%s"}) { amount { valueInCents } ' 'platformTipAmount { valueInCents } } }' % order_id_v2) async with httpx.AsyncClient(timeout=15.0) as client: r = await client.post(OC_GRAPHQL_URL, headers={"Content-Type": "application/json", "User-Agent": "Mozilla/5.0 (X11; Linux x86_64)"}, json={"query": query}) o = r.json().get("data", {}).get("order") or {} cents = (o.get("amount") or {}).get("valueInCents") if cents is not None: _order_amount_cache[order_id_v2] = int(cents) return int(cents) except Exception as e: logger.warning("Order amount lookup failed for %s: %s", order_id_v2, e) return None # Personal token for the "Inference Co-op Bot" account, which is an admin of # the collective. Authenticated as an admin, the GraphQL API exposes member # emails (which are hidden from anonymous access). We use this to match a # member by email — the stable identifier that works even for guest # contributors (who have no Open Collective account and thus no usable slug). OC_PERSONAL_TOKEN = os.environ.get("OC_PERSONAL_TOKEN", "") # Monthly credit budget (in USD of tokens) for all members. # Single sliding-scale tier: everyone gets the same $15/month in credits, # regardless of their $10/15/20 contribution. Governance decision (Loomio). MEMBER_BUDGET = float(os.environ.get("MEMBER_BUDGET", "15.0")) # Days to wait before re-sending the welcome/invite email to a member who was # provisioned but never completed account setup (inviteAccepted=False). The # nightly sweep re-invites such members, throttled by this interval so they # aren't emailed every single night. REINVITE_AFTER_DAYS = float(os.environ.get("REINVITE_AFTER_DAYS", "3")) # Models every member key/team may access. # Deliberately REMOVED: members get access to every model the gateway serves. # The gateway's `model_list` is the single source of truth for what's available; # teams and keys carry no per-model restriction (empty = "all models allowed" in # LiteLLM). Adding a provider is a gateway-config change only — no team/key # migration. (Kept as a documented empty value so the intent is explicit.) MEMBER_MODELS: list[str] = [] # The "members" group in Cloudron (group-based access control). # Members are assigned to this group, which grants access to the chat app. MEMBERS_GROUP_ID = os.environ.get("MEMBERS_GROUP_ID", "") # The "inactive" group in Cloudron. Lapsed members are moved here (instead of # being deleted) so they lose access but can reclaim their data on reactivation. INACTIVE_GROUP_ID = os.environ.get("INACTIVE_GROUP_ID", "") # Manually-provisioned members (comma-separated emails) who are NOT in the # Open Collective member list. Used for the free "founder" tier, which OC # cannot process as a $0 recurring subscription. These members are exempt from # the sweep's deactivation (they have no OC membership to lapse). MANUAL_MEMBERS = { e.strip().lower() for e in os.environ.get("MANUAL_MEMBERS", "").split(",") if e.strip() } # Loomio integration (forum.inference.coop). The portal reconciles the Loomio # group to the current active-member list via the User API (/api/b2). LOOMIO_BASE = os.environ.get("LOOMIO_BASE", "https://forum.inference.coop") LOOMIO_API_KEY = os.environ.get("LOOMIO_API_KEY", "") # Server API key (B3) — used to fix auto-generated Loomio usernames after # invites. Loomio's b2/memberships API only takes emails; new users get a # mangled auto-handle (email local-part + random suffix). The B3 users API # lets us rename them to the member's Cloudron username. LOOMIO_B3_KEY = os.environ.get("LOOMIO_B3_KEY", "") # Matrix homeserver (self-hosted Synapse at matrix.inference.coop). The bot # account (admin) invites every active member to the co-op Space on sync. # Members authenticate via Cloudron SSO; their homeserver MXID is # @:inference.coop. Federated MXIDs (external servers) # can also be invited directly. MATRIX_HOMESERVER = os.environ.get("MATRIX_HOMESERVER", "https://matrix.inference.coop") MATRIX_BOT_TOKEN = os.environ.get("MATRIX_BOT_TOKEN", "") MATRIX_SPACE_ID = os.environ.get("MATRIX_SPACE_ID", "") MATRIX_GENERAL_ROOM = os.environ.get("MATRIX_GENERAL_ROOM", "") # Additional member-facing rooms that every member gets invited to (comma-separated # room IDs). Restricted rooms are NOT discoverable over federation until a member # has joined (the space hierarchy omits them for remote servers), so new rooms need # an explicit invite pass — either here or a one-off script. MATRIX_EXTRA_ROOMS = [r.strip() for r in os.environ.get("MATRIX_EXTRA_ROOMS", "").split(",") if r.strip()] LOOMIO_GROUP_ID = os.environ.get("LOOMIO_GROUP_ID", "") # Operator accounts that must always remain in the Loomio group (never removed # by the remove_absent reconciliation), comma-separated. LOOMIO_ALWAYS_KEEP = [ e.strip() for e in os.environ.get("LOOMIO_ALWAYS_KEEP", "info@inference.coop,bot@inference.coop").split(",") if e.strip() ] # Persistent store (SQLite) for email → LiteLLM key token mapping. # Lives in /app/data (Cloudron localstorage addon persists this). DB_PATH = os.environ.get("DB_PATH", "/app/data/members.db") # --- Email (Cloudron sendmail addon) --- # The sendmail addon exports SMTP credentials as CLOUDRON_MAIL_* env vars. SMTP_SERVER = os.environ.get("CLOUDRON_MAIL_SMTP_SERVER", "") SMTP_PORT = int(os.environ.get("CLOUDRON_MAIL_SMTP_PORT", "2525")) SMTP_USERNAME = os.environ.get("CLOUDRON_MAIL_SMTP_USERNAME", "") SMTP_PASSWORD = os.environ.get("CLOUDRON_MAIL_SMTP_PASSWORD", "") MAIL_FROM = os.environ.get("CLOUDRON_MAIL_FROM", "info@inference.coop") MAIL_FROM_NAME = os.environ.get("CLOUDRON_MAIL_FROM_DISPLAY_NAME", "Inference Cooperative") # Reply-to address so member replies land at the co-op inbox. MAIL_REPLY_TO = os.environ.get("MAIL_REPLY_TO", "info@inference.coop") # Public base URL for the portal (used in email links). PORTAL_BASE = os.environ.get("PORTAL_BASE", "https://portal.inference.coop") # Listmonk newsletter integration. The portal keeps the "Members" list in sync: # subscribe on provision, unsubscribe on deactivation. Auth uses listmonk's # native API-user token scheme (Authorization: token :). LISTMONK_BASE = os.environ.get("LISTMONK_BASE") or "https://newsletter.inference.coop" # NOTE: `or` (not default-arg) — an env var set to EMPTY must fall back to the # default. An empty LISTMONK_API_USER made every sync 403 silently (the token # alone is not a valid listmonk credential) while listmonk_sync's best-effort # error handling swallowed it — that is how stale accounts stayed subscribed. LISTMONK_API_USER = os.environ.get("LISTMONK_API_USER") or "portal-sync" LISTMONK_API_TOKEN = os.environ.get("LISTMONK_API_TOKEN", "") LISTMONK_LIST_ID = int(os.environ.get("LISTMONK_LIST_ID", "3")) # "Members" list # Direct read access to LiteLLM's Postgres (for usage aggregations). # READ-ONLY purpose: never write to LiteLLM's tables — its HTTP API remains the # only interface for writes/management. Falls back to the HTTP path when unset. LITELLM_DB_HOST = os.environ.get("LITELLM_DB_HOST", "") LITELLM_DB_PORT = int(os.environ.get("LITELLM_DB_PORT", "5432")) LITELLM_DB_USER = os.environ.get("LITELLM_DB_USER", "") LITELLM_DB_PASSWORD = os.environ.get("LITELLM_DB_PASSWORD", "") LITELLM_DB_NAME = os.environ.get("LITELLM_DB_NAME", "") def litellm_db_ready() -> bool: """True when LiteLLM DB read credentials are configured.""" return bool(LITELLM_DB_HOST and LITELLM_DB_USER and LITELLM_DB_NAME) def _pg_connect(): """Open a short-lived connection to LiteLLM's Postgres (read-only use).""" import psycopg2 return psycopg2.connect( host=LITELLM_DB_HOST, port=LITELLM_DB_PORT, user=LITELLM_DB_USER, password=LITELLM_DB_PASSWORD, dbname=LITELLM_DB_NAME, connect_timeout=10, sslmode="prefer", ) def aggregate_model_usage(days: int = 0, team_id: str | None = None) -> dict: """Aggregate LiteLLM spend logs per model, straight from Postgres. Returns {"models": [...], "totals": {...}} in the same shape the HTTP path produced, so callers don't change. `days` bounds the window (0 = all time); `team_id` restricts to one member's team (used by the dashboard path). Prefers the `model_group` column, which carries our provider/model alias (e.g. greenpt/glm-5.3-flash); falls back to `model` for pre-alias rows — matching the logic the HTTP path used, expressed in SQL. """ if not litellm_db_ready(): raise RuntimeError("LiteLLM DB credentials not configured") where = [] params: list = [] if days and days > 0: from datetime import datetime, timedelta, timezone cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).strftime( "%Y-%m-%d %H:%M:%S" ) where.append("\"startTime\" >= %s") params.append(cutoff) if team_id: where.append("team_id = %s") params.append(team_id) where_sql = ("WHERE " + " AND ".join(where)) if where else "" sql = f""" SELECT COALESCE(NULLIF(model_group, ''), NULLIF(model, ''), 'unknown') AS model, COALESCE(SUM(spend), 0) AS spend, COALESCE(SUM(total_tokens), 0) AS tokens, COUNT(*) AS requests FROM "LiteLLM_SpendLogs" {where_sql} GROUP BY 1 ORDER BY requests DESC """ conn = _pg_connect() try: with conn.cursor() as cur: cur.execute(sql, params) rows = cur.fetchall() finally: conn.close() models = [ {"model": r[0], "spend": float(r[1] or 0.0), "tokens": int(r[2] or 0), "requests": int(r[3] or 0)} for r in rows ] return { "models": models, "totals": { "total_spend": sum(m["spend"] for m in models), "total_tokens": sum(m["tokens"] for m in models), "total_requests": sum(m["requests"] for m in models), }, } def get_team_budget_from_db(team_id: str) -> dict | None: """Fetch a team's spend/budget/reset directly from LiteLLM's Postgres. Same data as the /team/list HTTP path (LiteLLM computes `spend`), without downloading the whole team list. Returns None when the team is unknown. """ if not litellm_db_ready(): return None sql = """ SELECT max_budget, spend, budget_reset_at FROM "LiteLLM_TeamTable" WHERE team_id = %s LIMIT 1 """ conn = _pg_connect() try: with conn.cursor() as cur: cur.execute(sql, (team_id,)) row = cur.fetchone() finally: conn.close() if not row: return None return { "max_budget": float(row[0]) if row[0] is not None else None, "spend": float(row[1] or 0.0), "budget_reset_at": row[2], } async def listmonk_sync(email: str, subscribe: bool) -> None: """Subscribe or unsubscribe a member to the newsletter list. Best-effort: never raises, so a newsletter outage can't break provisioning. """ if not LISTMONK_API_TOKEN: logger.warning("Listmonk not configured; skipping sync for %s", email) return auth = {"Authorization": f"token {LISTMONK_API_USER}:{LISTMONK_API_TOKEN}"} try: async with httpx.AsyncClient(timeout=10.0) as client: if subscribe: r = await client.post( f"{LISTMONK_BASE}/api/subscribers", headers=auth, json={ "email": email, "name": "", "status": "enabled", "lists": [LISTMONK_LIST_ID], "preconfirm_subscriptions": True, }, ) else: # Look up the subscriber id by email, then blocklist by id. # (listmonk's blocklist endpoint takes an id, not an email.) q = await client.get( f"{LISTMONK_BASE}/api/subscribers", headers=auth, params={"query": f"subscribers.email='{email}'"}, ) if q.status_code != 200: logger.warning("Listmonk lookup for %s failed: %s", email, q.status_code) return results = q.json().get("data", {}).get("results", []) if not results: return # not subscribed; nothing to do sid = results[0]["id"] r = await client.put( f"{LISTMONK_BASE}/api/subscribers/{sid}/blocklist", headers=auth, ) if r.status_code not in (200, 201): logger.warning( "Listmonk sync %s for %s failed: %s", "subscribe" if subscribe else "unsubscribe", email, r.status_code, ) except Exception as e: logger.warning("Listmonk sync error for %s: %s", email, e) async def listmonk_reconcile() -> dict: """Reconcile the Members newsletter list against the active membership. The newsletter list mirrors the portal's ACTIVE members. Subscribers who are not active members fall into two classes, and only one is acted on: - emails that exist in the members table but are inactive, and emails that are not members at all, are taken off the list (the subscriber record stays as a suppression marker). A DELETE would destroy the subscriber's consent history, and a stray test account is indistinguishable from a real person who subscribed outside the portal, so nobody is purged automatically; - OPERATOR_EMAILS — infrastructure accounts, which the panel surfaces under "admins" and excludes from member counts — are left exactly as they are. They are neither members nor stale. The Loomio reconciliation applies the same rule through LOOMIO_ALWAYS_KEEP. Best-effort: never raises, so a newsletter outage can't break the sweep. """ if not LISTMONK_API_TOKEN: logger.warning("Listmonk not configured; skipping newsletter reconcile") return {"removed": 0, "emails": [], "error": "not configured"} auth = {"Authorization": f"token {LISTMONK_API_USER}:{LISTMONK_API_TOKEN}"} conn = get_db() active = { row[0].lower() for row in conn.execute("SELECT email FROM members WHERE active = 1") } member_emails = { row[0].lower() for row in conn.execute("SELECT email FROM members") } conn.close() try: async with httpx.AsyncClient(timeout=20.0) as client: # Fetch the full subscriber set for the list (paginated). subs, page = [], 1 while True: r = await client.get( f"{LISTMONK_BASE}/api/subscribers", headers=auth, params={"list_id": LISTMONK_LIST_ID, "page": page, "per_page": 100}, ) if r.status_code != 200: logger.warning( "Listmonk reconcile: subscriber fetch failed: %s", r.status_code ) return {"removed": 0, "emails": [], "error": f"fetch {r.status_code}"} data = r.json().get("data", {}) results = data.get("results", []) subs += results total = data.get("total") # Stop on an empty or short page, or once the reported count is # reached. The short-page test means a missing `total` cannot # truncate the reconcile at page 1. if ( not results or len(results) < 100 or (total is not None and len(subs) >= total) ): break page += 1 removed, emails = 0, [] by_reason = {"inactive member": 0, "not a member": 0} # Operator accounts are neither members nor stale; never touch their # subscription (LOOMIO_ALWAYS_KEEP is the same rule for the forum). operators_kept = sum( 1 for s in subs if (s.get("email") or "").lower() in OPERATOR_EMAILS ) for s in subs: email = (s.get("email") or "").lower() if not email or email in active or email in OPERATOR_EMAILS: continue sid = s.get("id") # Both classes leave the list the same way: unsubscribe, keep the # record. Nothing is deleted. reason = "inactive member" if email in member_emails else "not a member" # listmonk validates the PUT body's email field: omitting it is a # 400 "Invalid email." and the subscriber stays on the list (which # is why the inactive-member branch never actually worked). r = await client.put( f"{LISTMONK_BASE}/api/subscribers/{sid}", headers=auth, json={"email": s.get("email"), "lists": []}, ) if r.status_code in (200, 201): removed += 1 by_reason[reason] += 1 emails.append(email) else: logger.warning( "Listmonk reconcile: unsubscribe %s failed: %s", email, r.status_code ) if removed: # Counts only — subscriber addresses don't belong in the log. logger.info( "Listmonk reconcile: removed %d stale subscribers (%s)", removed, ", ".join(f"{k}: {v}" for k, v in by_reason.items() if v), ) if operators_kept: logger.info( "Listmonk reconcile: left %d operator address(es) on the list", operators_kept, ) return {"removed": removed, "emails": emails, "operators_kept": operators_kept} except Exception as e: logger.warning("Listmonk reconcile error: %s", e) return {"removed": 0, "emails": [], "error": str(e)} def _verify_portal_secret(request: Request) -> None: """Reject requests that didn't come through LibreChat (shared secret).""" if not PORTAL_SECRET: # If no secret is configured, refuse to inject keys (fail closed). raise HTTPException(503, "Portal secret not configured") provided = request.headers.get("x-portal-secret", "") if not secrets.compare_digest(provided, PORTAL_SECRET): raise HTTPException(401, "Invalid portal secret") def _verify_admin_token(request: Request) -> None: """Authorize a /admin/* caller (SECURITY 2026-10-27). Admin auth is a HEADER, never a URL token: the webhook URL token lives in places we do not control, and it used to unlock all of /admin/*. """ provided = request.headers.get("x-admin-token", "") if ADMIN_TOKEN and secrets.compare_digest(provided, ADMIN_TOKEN): return if ADMIN_TOKEN_LEGACY and WEBHOOK_TOKEN and secrets.compare_digest(provided, WEBHOOK_TOKEN): logger.warning( "admin auth: caller used the legacy WEBHOOK_TOKEN — move it to" " ADMIN_TOKEN, then set ADMIN_TOKEN_LEGACY=0" ) return raise HTTPException(403, "Forbidden") def get_db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.execute( "CREATE TABLE IF NOT EXISTS members (" "email TEXT PRIMARY KEY, " "key_token TEXT, " "cloudron_user_id TEXT, " "slug TEXT, " "active INTEGER DEFAULT 1" ")" ) # Daily metrics snapshot, written by /app/code/metrics_snapshot.py from the # scheduler and read by the admin panel + dashboard. Cumulative columns # (members_*, newsletter_subs, matrix_space_members) are counts AS OF that # day; spend_usd/requests are THAT day's traffic. A NULL column means the # source could not be read that day — charts must show a gap, never a zero. conn.execute( "CREATE TABLE IF NOT EXISTS metrics_daily (" "day TEXT PRIMARY KEY, " "members_active INTEGER, " "members_total INTEGER, " "members_new INTEGER, " "newsletter_subs INTEGER, " "matrix_space_members INTEGER, " "spend_usd REAL, " "requests INTEGER, " "out_tokens INTEGER, " "energy_wh REAL, " "co2e_g REAL, " "tokens_measured INTEGER, " "tokens_estimated INTEGER, " "factor_revision TEXT, " "recorded_at TEXT" ")" ) # CREATE IF NOT EXISTS never adds columns to an existing table, so migrate: # impact columns arrived after the table did (see co-op/support#26), and the # measured-vs-estimated provenance columns after those. All additive: existing # rows keep NULL, and a NULL split reports as unknown — never as "measured". for _col, _decl in ( ("out_tokens", "INTEGER"), ("energy_wh", "REAL"), ("co2e_g", "REAL"), ("tokens_measured", "INTEGER"), ("tokens_estimated", "INTEGER"), ("factor_revision", "TEXT"), ): try: conn.execute(f"ALTER TABLE metrics_daily ADD COLUMN {_col} {_decl}") except sqlite3.OperationalError: pass # already present # Member-managed API keys (created via the broker). Each row is one API key # under the member's team, tracked so we can list/revoke by name without # exposing the chat key. `sk_token` is the full sk- key (revealed once). conn.execute( "CREATE TABLE IF NOT EXISTS member_api_keys (" "email TEXT NOT NULL, " "name TEXT NOT NULL, " "sk_token TEXT NOT NULL, " "hash TEXT NOT NULL, " "created_at TEXT DEFAULT (datetime('now')), " "PRIMARY KEY (email, name)" ")" ) # Migration: add slug column if the members table predates it. cols = [r[1] for r in conn.execute("PRAGMA table_info(members)").fetchall()] if "slug" not in cols: conn.execute("ALTER TABLE members ADD COLUMN slug TEXT") # Migration: add balance column (USD) — the member's current credit # allowance. Defaults to MEMBER_BUDGET; can be topped up by payments. # This is the flexible counter to the fixed $15/month allowance. if "balance" not in cols: conn.execute("ALTER TABLE members ADD COLUMN balance REAL") # Migration: welcome_sent_at — timestamp of the last welcome/invite email, # used to throttle re-invites of members who never activated. if "welcome_sent_at" not in cols: conn.execute("ALTER TABLE members ADD COLUMN welcome_sent_at TEXT") # Migration: matrix_mxid — an OPTIONAL extra Matrix account a member wants # invited to the co-op Space (e.g. their existing federated account on # another homeserver). Empty = only the local @username:inference.coop # account is invited. At most one extra MXID per member. if "matrix_mxid" not in cols: conn.execute("ALTER TABLE members ADD COLUMN matrix_mxid TEXT") # Migration: credit_balance — purchased, NON-EXPIRING credits (USD). # Distinct from `balance` (the monthly allowance, which resets): credits # carry over indefinitely and are spent only after the allowance is # exhausted within a cycle. if "credit_balance" not in cols: conn.execute("ALTER TABLE members ADD COLUMN credit_balance REAL NOT NULL DEFAULT 0") # Migration: org_slug — for ORGANIZATION-backed memberships. OC orgs never # expose an email to our integration, so the org's designated contact is # linked manually (POST /admin/org-link): org_slug holds the OC org slug # whose Membership-tier dues keep this account active. The nightly sweep # keeps the account active while the org is a BACKER on the Membership # tier and deactivates it when the org's dues lapse. if "org_slug" not in cols: conn.execute("ALTER TABLE members ADD COLUMN org_slug TEXT") # Credit ledger: audit trail for every credit movement (purchases via the # OC webhook, admin adjustments, spends drawn against credits). Money needs # an audit trail. conn.execute( "CREATE TABLE IF NOT EXISTS credit_ledger (" "id INTEGER PRIMARY KEY AUTOINCREMENT, " "email TEXT NOT NULL, " "delta REAL NOT NULL, " "reason TEXT NOT NULL, " "reference TEXT, " "created_at TEXT DEFAULT (datetime('now'))" ")" ) # Held credit packs: purchases by NON-members. Money received but credits # not granted (a membership is required to use credits). Each hold is # either refunded manually (OC dashboard, 30-day window) or granted # automatically when the buyer becomes a member. conn.execute( "CREATE TABLE IF NOT EXISTS held_credits (" "id INTEGER PRIMARY KEY AUTOINCREMENT, " "slug TEXT NOT NULL, " "email TEXT, " "name TEXT, " "amount REAL NOT NULL, " "oc_reference TEXT, " "status TEXT NOT NULL DEFAULT 'held', " "created_at TEXT DEFAULT (datetime('now')), " "resolved_at TEXT" ")" ) # Member-managed API keys: add a fingerprint (last-4 of the key, shown in # lists) and purge the stored plaintext. Member API keys are revealed ONCE # (at creation) and never again — the industry standard. Revocation uses the # LiteLLM hash, not the plaintext, so the portal has no reason to keep it. # (The `members.key_token` chat key is a different table and IS retained — # encrypted at rest (AES-256-GCM), decrypted in memory only when served.) ak_cols = [r[1] for r in conn.execute("PRAGMA table_info(member_api_keys)").fetchall()] if "fingerprint" not in ak_cols: conn.execute("ALTER TABLE member_api_keys ADD COLUMN fingerprint TEXT") # Backfill fingerprints for any key still holding plaintext (fingerprint = # last-4 of the key), then purge the plaintext. Idempotent — runs every # time, affects only rows where plaintext is still present. conn.execute( "UPDATE member_api_keys SET fingerprint = substr(sk_token, -4) " "WHERE (fingerprint IS NULL OR fingerprint = '') AND sk_token != ''" ) conn.execute("UPDATE member_api_keys SET sk_token = '' WHERE sk_token != ''") # Encrypt member chat keys at rest (AES-256-GCM, KEK from PORTAL_KEK env). # Idempotent: only touches rows still holding plaintext. Skipped with a # CRITICAL log if PORTAL_KEK is unset — see the inline key-encryption module. migrate_plaintext_keys(conn) # Commit the DML above — ALTER TABLE auto-commits, but the UPDATEs open an # implicit transaction that would otherwise roll back when get_db() closes. conn.commit() return conn def get_member_balance(email: str) -> float: """Return the member's current balance (USD), defaulting to MEMBER_BUDGET. The balance is the flexible allowance — it starts at MEMBER_BUDGET but can be adjusted (topped up by payments) without changing code. """ conn = get_db() row = conn.execute( "SELECT balance FROM members WHERE email = ?", (email,) ).fetchone() conn.close() if row and row[0] is not None: return float(row[0]) return MEMBER_BUDGET def set_member_balance(email: str, balance: float) -> None: """Set the member's balance (USD). Used by top-up / billing adjustments.""" conn = get_db() conn.execute( "UPDATE members SET balance = ? WHERE email = ?", (balance, email) ) conn.commit() conn.close() # --- Credits: purchased, non-expiring usage dollars --- def get_credit_balance(email: str) -> float: """Return the member's non-expiring credit balance (USD, default 0).""" conn = get_db() row = conn.execute( "SELECT credit_balance FROM members WHERE email = ?", (email,) ).fetchone() conn.close() if row and row[0] is not None: return float(row[0]) return 0.0 def add_credits(email: str, delta: float, reason: str, reference: str | None = None) -> float: """Adjust a member's credit balance and record the movement in the ledger. Returns the new balance. Negative deltas are allowed (draws against credits); the balance is clamped at >= 0. """ conn = get_db() conn.execute( "UPDATE members SET credit_balance = MAX(0, COALESCE(credit_balance, 0) + ?) " "WHERE email = ?", (delta, email), ) conn.execute( "INSERT INTO credit_ledger (email, delta, reason, reference) VALUES (?, ?, ?, ?)", (email, delta, reason, reference), ) row = conn.execute( "SELECT credit_balance FROM members WHERE email = ?", (email,) ).fetchone() conn.commit() conn.close() return float(row[0]) if row and row[0] is not None else 0.0 def compute_max_budget(allowance: float, credit_balance: float, team_spend: float | None) -> dict: """Compute the effective LiteLLM max_budget from the two buckets. Model: the member has a monthly ALLOWANCE (renews on each membership payment) plus a credit balance that adds spendable headroom ON TOP of the allowance. The total spendable cap is simply `allowance + credit_balance` — credits never shrink the allowance; they extend it. `credit_balance` is the TOTAL purchased (the ledger's lifetime sum); the amount still available this cycle is `credit_balance - credits_used`. `team_spend` is the member's spend so far this cycle (from LiteLLM). It is used only to split the spend between allowance vs credits for display, NOT to reduce the cap. Returns {"max_budget", "allowance_used", "credits_used", "credits_remaining"} for dashboards. """ max_budget = allowance + credit_balance if team_spend is None: return { "max_budget": max_budget, "allowance_used": 0.0, "credits_used": 0.0, "credits_remaining": credit_balance, } allowance_used = min(max(team_spend, 0.0), allowance) credits_used = max(0.0, team_spend - allowance) return { "max_budget": max_budget, "allowance_used": allowance_used, "credits_used": credits_used, "credits_remaining": max(0.0, credit_balance - credits_used), } async def fetch_all_member_nodes() -> list[dict]: """Every member node of the collective, via OC's offset pagination. The members connection has NO cursor/pageInfo pagination (older-style API) and caps at 100 per page: an unpaginated query silently hides everyone past the first page — the same failure class as Cloudron's 25-per-page user list. Raises on ANY page failure so callers can apply their own fail-safes; never returns a partial list (a partial member list would make the sweep deactivate everyone missing from it). """ if not OC_PERSONAL_TOKEN: raise RuntimeError("OC_PERSONAL_TOKEN not set; cannot fetch members") nodes: list[dict] = [] offset = 0 async with httpx.AsyncClient() as client: for _ in range(20): # hard cap: 20 pages × 100 = 2000 members query = ( '{ collective(slug: "%s") { members(limit: 100, offset: %d) { ' "totalCount nodes { role tier { name } account { name slug type ... on Individual { email } } } } } }" % (OC_COLLECTIVE_SLUG, offset) ) r = await client.post( OC_GRAPHQL_URL, headers={"Personal-Token": OC_PERSONAL_TOKEN}, json={"query": query}, ) if r.status_code != 200: raise RuntimeError(f"OC member fetch failed: HTTP {r.status_code}") members = ((r.json().get("data") or {}).get("collective") or {}).get("members") or {} total = members.get("totalCount") or 0 page = members.get("nodes") or [] nodes.extend(page) offset += len(page) if not page or offset >= total: return nodes raise RuntimeError("OC member pagination exceeded 20 pages") _MEMBER_SET_MIN_RATIO = 0.5 def _member_set_verdict(fetched: int, local_active: int) -> bool: """Is this member list plausible enough to act on? (member-portal#2) Pure decision so it can be tested without a database. Guards the case a raise-only fail-safe cannot see: a fetch that SUCCEEDS while returning nothing, or far less than usual. A narrowed token scope, an expired credential with partial access, or a renamed Membership tier on Open Collective all produce that, and the sweep's response to it is to deactivate the membership rather than to fail. An empty collective is legitimate only when we have no active members ourselves. """ if local_active <= 0: return True # Strictly above: a halving of the membership is implausible enough, and refusing # costs a loud error (no action taken) where allowing it costs the membership. return fetched > 0 and fetched > local_active * _MEMBER_SET_MIN_RATIO def _member_set_guard(fetched: int, label: str) -> None: """Raise when the fetched member list is implausible, so callers fail safe.""" try: conn = get_db() local_active = conn.execute( "SELECT COUNT(*) FROM members WHERE active = 1" ).fetchone()[0] conn.close() except Exception as e: # fresh install / unreadable DB: nothing to compare against logger.warning("member-set guard: cannot read the local member count (%s); skipping", e) return if _member_set_verdict(fetched, local_active): logger.info("member-set guard OK (%s): fetched %d, %d active locally", label, fetched, local_active) return raise RuntimeError( f"implausible member list from Open Collective ({label}): fetched {fetched} while " f"{local_active} members are active locally. Refusing to reconcile — the sweep " "fail-safe keeps members active instead. Check the OC token's scopes and expiry, " "and that the Membership tier still exists (fetch_members filters on its name)." ) async def fetch_members() -> list[dict]: """Return the full active-member list: email, name, slug, role, tier. Uses the bot's personal token (admin) so emails are exposed. This is the authoritative source of truth for reconciliation (provisioning missing members, deactivating lapsed ones). Membership is tied to the "Membership" TIER — a one-time donation grants BACKER role on Open Collective but is NOT a membership. So this returns only users whose tier name contains "membership" (case-insensitive) or who have no tier (legacy: contributors who predate the tier system). """ if not OC_PERSONAL_TOKEN: logger.warning("OC_PERSONAL_TOKEN not set; cannot fetch members") return [] # Paginated fetch (raises on failure → caller fail-safes apply: the sweep # skips deactivation entirely when we can't get the full member list). nodes = await fetch_all_member_nodes() result: list[dict] = [] for n in nodes: role = n.get("role", "") acct = n.get("account", {}) or {} email = (acct.get("email") or "").strip().lower() tier = ((n.get("tier") or {}).get("name") or "").strip() if not email or role not in ("BACKER", "ADMIN"): continue # Admins (the co-op itself) always count. BACKERs must be on the # Membership tier; no-tier BACKERs are legacy contributors from before # tiers existed — grandfathered as members. if role == "ADMIN" or "membership" in tier.lower() or tier == "": result.append({ "email": email, "name": acct.get("name") or email.split("@")[0], "slug": acct.get("slug") or "", "role": role, "tier": tier, }) # Guard before ANY caller acts on this: the sweep deactivates members missing # from the list, so an implausible list must fail rather than propagate. _member_set_guard(len(result), "oc members query") return result async def fetch_org_backers() -> dict[str, dict]: """Organization-backed memberships on the Membership tier. OC Organizations never expose an email to our integration (the `... on Individual { email }` fragment doesn't match them, and OC does not return org emails even to collective admins), so these members are LINKED to portal accounts via members.org_slug (POST /admin/org-link). Keyed by org slug: {slug: {name, role, tier}}. """ if not OC_PERSONAL_TOKEN: logger.warning("OC_PERSONAL_TOKEN not set; cannot fetch org members") raise RuntimeError("OC_PERSONAL_TOKEN not set") # Paginated fetch; RAISES on failure so the sweep's fail-safe keeps linked # accounts active rather than deactivating them on a fetch error. nodes = await fetch_all_member_nodes() result: dict[str, dict] = {} for n in nodes: role = n.get("role", "") acct = n.get("account", {}) or {} if (acct.get("type") or "") != "ORGANIZATION": continue if role not in ("BACKER", "ADMIN"): continue tier = ((n.get("tier") or {}).get("name") or "").strip() if role != "ADMIN" and "membership" not in tier.lower() and tier != "": continue slug = acct.get("slug") or "" if slug: result[slug] = { "name": acct.get("name") or slug, "role": role, "tier": tier, } return result def send_email(to: str, subject: str, text_body: str, html_body: str | None = None) -> bool: """Send an email via Cloudron's sendmail addon (SMTP). Returns True on success, False on failure (logged, not raised — email is best-effort and must never break the provisioning flow). """ if not SMTP_SERVER: logger.warning("SMTP not configured; skipping email to %s", to) return False try: msg = MIMEMultipart("alternative") msg["From"] = f"{MAIL_FROM_NAME} <{MAIL_FROM}>" msg["To"] = to msg["Subject"] = subject msg["Reply-To"] = MAIL_REPLY_TO msg.attach(MIMEText(text_body, "plain", "utf-8")) if html_body: msg.attach(MIMEText(html_body, "html", "utf-8")) with smtplib.SMTP(SMTP_SERVER, SMTP_PORT, timeout=15) as s: if SMTP_USERNAME: s.login(SMTP_USERNAME, SMTP_PASSWORD) s.sendmail(MAIL_FROM, [to], msg.as_string()) logger.info("Sent email to %s: %s", to, subject) return True except Exception as e: logger.warning("Failed to send email to %s: %s", to, e) return False def welcome_email(name: str, setup_url: str) -> tuple[str, str, str]: """Build the welcome email (subject, text, html) for a new member. `setup_url` is the Cloudron account-setup link (from invite_link), which lets the member create their account directly — no OAuth step needed. The HTML uses a centered, card-style layout matching the landing page (cream background, forest-green button, Figtree-like system fonts). Email HTML is table-based for client compatibility; no external assets. """ subject = "Welcome to the Inference Cooperative — finish your setup" text = ( f"Hi {name},\n\n" "Thanks for joining the Inference Cooperative!\n\n" "To finish setting up your account and start using private AI chat, " "click the link below to create your account:\n\n" f"{setup_url}\n\n" "This takes about a minute.\n\n" "Questions? Reply to this email or write to info@inference.coop.\n\n" "— The Inference Cooperative\n" ) html = f"""

Welcome, {name}

Thanks for joining the Inference Cooperative.

Finish setting up your account to start using private AI chat.

Finish setup

This takes about a minute.
Questions? Reply to this email or write to info@inference.coop.
The Inference Cooperative · inference.coop
By creating your account, you agree to our Terms of Service and Privacy Policy.
""" return subject, text, html async def get_active_member_emails() -> list[str]: """Return the list of active member emails from the portal's own database. The local `members` table is the authoritative source of member emails (populated during provisioning). Used for the Loomio sync. """ conn = get_db() rows = conn.execute( "SELECT email FROM members WHERE active = 1" ).fetchall() conn.close() return [r[0] for r in rows] def get_member_matrix_mxid(email: str) -> str: """Return the member's optional extra Matrix MXID, or "" if unset.""" conn = get_db() row = conn.execute( "SELECT matrix_mxid FROM members WHERE email = ?", (email,) ).fetchone() conn.close() return (row[0] or "").strip() if row else "" def set_member_matrix_mxid(email: str, mxid: str) -> None: """Set (or clear) the member's optional extra Matrix MXID.""" conn = get_db() conn.execute( "UPDATE members SET matrix_mxid = ? WHERE email = ?", (mxid.strip() if mxid else "", email), ) conn.commit() conn.close() def _loomio_safe_username(raw) -> str | None: """Sanitize a candidate into a Loomio-legal username. Loomio accepts lowercase letters, numbers, underscores and INTERNAL hyphens only, so a dot-containing candidate — an OAuth uid or an email local part such as "joshua.z.tan" — must be normalized rather than passed through. Passing it through is what made this fix fail with HTTP 422 on every sweep. """ u = re.sub(r"[^a-z0-9_-]", "_", (raw or "").lower()).strip("-_") return u if len(u) >= 2 else None def _loomio_is_autogenerated(current: str, email: str) -> bool: """Is this handle one Loomio generated for us? Loomio derives a handle from the email local part with non-alphanumerics stripped, plus a random suffix ("joshuaztan" + "xjan5rzb8j27" for joshua.z.tan@...). Anything that is not that shape — "amarino", say — was chosen by a person, and renaming it would be an unrequested member-visible change. Only auto-generated handles are in scope for this fix. """ local = re.sub(r"[^a-z0-9]", "", (email or "").split("@")[0].lower()) cur = (current or "").lower() return bool(local) and cur.startswith(local) and len(cur) > len(local) async def loomio_fix_usernames(client: httpx.AsyncClient) -> dict: """Rename Loomio users with auto-generated handles to their Cloudron username. Loomio's b2/memberships invite creates the account with a mangled handle (email local-part + random suffix, e.g. danishipleydsdk4ibdihka). After first SSO login the user carries an oauth identity whose uid IS their Cloudron username — prefer that; fall back to the email local-part. Uses the server (b3) API. Returns a {fixed, skipped} summary. """ H = {"Authorization": f"Bearer {LOOMIO_B3_KEY}", "Content-Type": "application/json"} r = await client.get(f"{LOOMIO_BASE}/api/b3/users", headers=H) r.raise_for_status() users = r.json().get("users", []) fixed, skipped = 0, 0 for u in users: # Only Loomio's own auto-generated handles are in scope: a name a human # chose stays, even when it differs from the Cloudron username. if not _loomio_is_autogenerated(u.get("username") or "", u.get("email") or ""): skipped += 1 continue target = None for ident in u.get("identities") or []: if ident.get("identity_type") == "oauth" and ident.get("uid"): target = _loomio_safe_username(ident["uid"]) if not target: # uid unusable even after sanitizing — fall back to the email local part target = _loomio_safe_username((u.get("email") or "").split("@")[0]) break if not target: target = _loomio_safe_username((u.get("email") or "").split("@")[0]) if not target or u.get("username") == target: skipped += 1 continue rr = await client.patch( f"{LOOMIO_BASE}/api/b3/users/{u['id']}", headers=H, json={"username": target}, ) if rr.status_code == 200: fixed += 1 else: # log the Loomio user id, never the member's email logger.warning("Loomio username fix for user %s failed: %s %s", u.get("id"), rr.status_code, rr.text[:120]) if fixed: logger.info("Loomio usernames fixed: %s", fixed) return {"fixed": fixed, "skipped": skipped} async def sync_matrix_space() -> dict: """Invite every active member to the co-op Matrix Space (and General room). Members log in via Cloudron SSO, so their local MXID is @:inference.coop. We resolve the Cloudron username for each active member email and invite them if not already in the space. Best-effort: missing config just skips; failures log and continue. """ if not (MATRIX_BOT_TOKEN and MATRIX_SPACE_ID): logger.info("Matrix space sync skipped (not configured)") return {"skipped": True} H = {"Authorization": f"Bearer {MATRIX_BOT_TOKEN}", "Content-Type": "application/json"} async def invite(mxid: str, room: str) -> bool: import asyncio as _aio for attempt in range(3): r = await client.post( f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/{room}/invite", headers=H, json={"user_id": mxid}, ) if r.status_code == 200: return True if r.status_code == 429: wait = (r.json().get("retry_after_ms", 1000) / 1000) + 0.2 await _aio.sleep(wait) continue # 403 with 'already in room' is fine if "already" in r.text: return True logger.warning("Matrix invite %s -> %s failed: %s %s", mxid, room, r.status_code, r.text[:120]) return False return False async with httpx.AsyncClient(timeout=30) as client: # current members of the space (to avoid duplicate invites) r = await client.get( f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/{MATRIX_SPACE_ID}/joined_members", headers=H) joined = set(r.json().get("joined", {}).keys()) if r.status_code == 200 else set() # also pending invites r = await client.get( f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/{MATRIX_SPACE_ID}/members?membership=invite", headers=H) invited = {m["state_key"] for m in r.json().get("chunk", [])} if r.status_code == 200 else set() # One identity per member: when a member sets a dashboard extra MXID, # their auto-generated local account (@username:inference.coop) should # NOT also sit in the Space. Track the desired set, then retire the # local accounts of members whose dashboard MXID replaced them. invited_count = 0 async for email, mxid in matrix_member_mxids(client): if not mxid or mxid in joined or mxid in invited: continue await invite(mxid, MATRIX_SPACE_ID) if MATRIX_GENERAL_ROOM: await invite(mxid, MATRIX_GENERAL_ROOM) for room in MATRIX_EXTRA_ROOMS: await invite(mxid, room) invited_count += 1 if invited_count: logger.info("Matrix space invites sent: %s", invited_count) # Retire local accounts: for each member with a dashboard-set external # MXID, kick their @username:inference.coop account if it's currently # in the Space. Only ever touches accounts derived from that member's # own Cloudron username — never other members' or federated accounts. conn = get_db() rows = conn.execute( "SELECT email, matrix_mxid FROM members WHERE active = 1 AND matrix_mxid IS NOT NULL AND matrix_mxid != ''" ).fetchall() conn.close() users: dict[str, str] = {} page = 1 async with httpx.AsyncClient(timeout=30) as uc: while True: r = await uc.get( f"{CLOUDRON_API}/api/v1/users", params={"page": page}, headers=cloudron_headers(), ) if r.status_code != 200: break batch = r.json().get("users", []) for u in batch: if u.get("email") and u.get("username"): users[u["email"].lower()] = u["username"] if len(batch) < 25: break page += 1 retired = 0 for email, ext_mxid in rows: username = users.get(email.lower()) if not username: continue local_mxid = f"@{username}:inference.coop" if local_mxid == ext_mxid: continue # dashboard MXID IS the local account — nothing to retire # Retire BOTH joined and invite-state local accounts: a stale invite # leaves the local account visible in the user directory and # accept-able from a lingering client notification, which defeats # the one-identity rule. Revoke invites via state-event membership # (kick 403s on invite-state users). import urllib.parse # local import: only needed on this retire path for room in [MATRIX_SPACE_ID] + ([MATRIX_GENERAL_ROOM] if MATRIX_GENERAL_ROOM else []): enc_room = urllib.parse.quote(room, safe="") if local_mxid in joined: await client.post( f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/{enc_room}/kick", headers=H, json={"user_id": local_mxid, "reason": f"This co-op account is superseded by {ext_mxid} (set on your dashboard)"}, ) retired += 1 logger.info("Matrix sync: retired local account %s (member uses %s)", local_mxid, ext_mxid) elif local_mxid in invited: await client.put( f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/{enc_room}/state/m.room.member/{urllib.parse.quote(local_mxid, safe='')}", headers=H, json={"membership": "leave", "reason": f"This co-op account is superseded by {ext_mxid} (set on your dashboard)"}, ) retired += 1 logger.info("Matrix sync: revoked stale invite for local account %s (member uses %s)", local_mxid, ext_mxid) if retired: logger.info("Matrix sync: retired %s local account(s)/invite(s)", retired) return {"invited": invited_count} async def matrix_member_mxids(client: httpx.AsyncClient): """Yield (email, mxid) for each active member. Local members log in via Cloudron SSO and Synapse names them by Cloudron username → @:inference.coop. Emails that don't resolve to a Cloudron user (e.g. federated/external members) are skipped here but can be added via the MATRIX_EXTRA_MXIDS env (comma list of full MXIDs). """ extra = [m.strip() for m in os.environ.get("MATRIX_EXTRA_MXIDS", "").split(",") if m.strip()] seen: set[str] = set() # Cloudron paginates /api/v1/users at 25/page — walk every page. users: dict[str, str] = {} page = 1 while True: r = await client.get( f"{CLOUDRON_API}/api/v1/users", params={"page": page}, headers=cloudron_headers(), ) if r.status_code != 200: logger.warning("Cloudron users lookup page %s failed: %s", page, r.status_code) break batch = r.json().get("users", []) for u in batch: if u.get("email") and u.get("username"): users[u["email"].lower()] = u["username"] if len(batch) < 25: break page += 1 for email in await get_active_member_emails(): # A member's own extra MXID (set on the dashboard) takes precedence — # that's the account they want in the Space. Otherwise use their local # @username:inference.coop account. own = get_member_matrix_mxid(email) if own: if own not in seen: seen.add(own) yield email, own continue username = users.get(email.lower()) if username: mxid = f"@{username}:inference.coop" seen.add(mxid) yield email, mxid for mxid in extra: if mxid not in seen: yield mxid, mxid async def sync_loomio_memberships() -> None: """Reconcile the Loomio group to the current active-member list. Uses POST /api/b2/memberships with remove_absent=1, which adds new members and removes anyone not in the list. Operator accounts (LOOMIO_ALWAYS_KEEP) are always included so they are never removed. Only members whose Cloudron account is already set up (inviteAccepted) are invited. Loomio's b2/memberships endpoint has no "silent invite" flag — it unconditionally emails a membership_created notification — so inviting a not-yet-activated member sends them a Loomio invite email before they've even created their account. Filtering to activated members prevents that; unactivated members are simply picked up on a later sync once they finish account setup. """ if not LOOMIO_API_KEY or not LOOMIO_GROUP_ID: logger.warning("Loomio not configured; skipping membership sync") return emails = await get_active_member_emails() # Always keep operator accounts in the group. for keep in LOOMIO_ALWAYS_KEEP: if keep not in emails: emails.append(keep) # Only invite members who have completed Cloudron account setup. The # operator accounts (LOOMIO_ALWAYS_KEEP) bypass this filter — they must # stay in the group regardless of inviteAccepted (the bootstrap owner # account shows inviteAccepted: false forever even though it's functional). emails = await filter_loomio_activated(emails) async with httpx.AsyncClient() as client: r = await client.post( f"{LOOMIO_BASE}/api/b2/memberships", headers={"Authorization": f"Bearer {LOOMIO_API_KEY}"}, json={"group_id": int(LOOMIO_GROUP_ID), "emails": emails, "remove_absent": 1}, ) if r.status_code != 200: logger.warning("Loomio membership sync failed: %s %s", r.status_code, r.text[:200]) return result = r.json() logger.info( "Loomio sync: added=%s removed=%s", result.get("added_emails", []), result.get("removed_emails", []), ) # Fix auto-generated usernames (email+randomsuffix) → Cloudron username. # Runs after every sync so newly invited members get corrected within a day # (or immediately when the sync follows provisioning). Best-effort: no B3 # key or API failure just logs and moves on. if LOOMIO_B3_KEY: try: await loomio_fix_usernames(client) except Exception as e: logger.warning("Loomio username fix pass failed: %s", e) async def filter_loomio_activated(emails: list[str]) -> list[str]: """Return the subset of `emails` whose Cloudron account is set up, plus any operator accounts (LOOMIO_ALWAYS_KEEP), which must always pass through. Fetches the Cloudron user list once and builds an inviteAccepted set keyed by email. Members who are provisioned but not yet activated are dropped so Loomio doesn't email them an invite before they've set up their account. """ activated: set[str] = set() try: async with httpx.AsyncClient(timeout=15.0) as client: r = await client.get( f"{CLOUDRON_API}/api/v1/users?per_page=100", headers=cloudron_headers(), ) r.raise_for_status() for u in r.json().get("users", []): if u.get("inviteAccepted") and u.get("email"): activated.add(u["email"]) except Exception as e: # If we can't determine activation state, fail open (invite everyone) # rather than silently dropping members from Loomio. logger.warning("filter_loomio_activated: cloudron users fetch failed: %s", e) return emails keep = set(LOOMIO_ALWAYS_KEEP) return [e for e in emails if e in keep or e in activated] def store_member(email: str, key_token: str, cloudron_user_id: str, slug: str = "") -> None: conn = get_db() # Preserve an existing slug if the caller passed none — re-provisioning an # already-provisioned member must not wipe their OC slug (the deactivation # webhook maps slugs back to members). if not slug: row = conn.execute("SELECT slug FROM members WHERE email = ?", (email,)).fetchone() if row and row[0]: slug = row[0] conn.execute( "INSERT INTO members (email, key_token, cloudron_user_id, slug, active) " "VALUES (?, ?, ?, ?, 1) " "ON CONFLICT(email) DO UPDATE SET key_token=excluded.key_token, " "cloudron_user_id=excluded.cloudron_user_id, slug=excluded.slug, active=1", (email, encrypt_key(key_token), cloudron_user_id, slug), ) conn.commit() conn.close() def get_member_key(email: str) -> str | None: conn = get_db() row = conn.execute( "SELECT key_token FROM members WHERE email = ? AND active = 1", (email,) ).fetchone() conn.close() return decrypt_key(row[0]) if row else None def get_stored_key(email: str) -> str | None: """Return the stored LiteLLM key token regardless of active status. Unlike get_member_key (which filters active=1), this returns the key even for a lapsed member being reactivated, so provisioning can reuse it rather than mint a duplicate alias (LiteLLM rejects a duplicate `member:` alias with 400). """ conn = get_db() row = conn.execute( "SELECT key_token FROM members WHERE email = ?", (email,) ).fetchone() conn.close() return (decrypt_key(row[0]) if row and row[0] else None) def touch_welcome_sent(email: str) -> None: """Record that a welcome/invite email was just sent for this member.""" conn = get_db() conn.execute( "UPDATE members SET welcome_sent_at = datetime('now') WHERE email = ?", (email,), ) conn.commit() conn.close() def get_member_by_slug(slug: str) -> dict | None: """Look up a member (email, key_token, cloudron_user_id) by their OC slug.""" conn = get_db() row = conn.execute( "SELECT email, key_token, cloudron_user_id FROM members WHERE slug = ?", (slug,) ).fetchone() conn.close() if not row: return None return {"email": row[0], "key_token": decrypt_key(row[1]), "cloudron_user_id": row[2]} def deactivate_member(email: str) -> str | None: """Mark member inactive and return their DECRYPTED key token (for deletion).""" conn = get_db() row = conn.execute( "SELECT key_token FROM members WHERE email = ?", (email,) ).fetchone() conn.execute("UPDATE members SET active = 0 WHERE email = ?", (email,)) conn.commit() conn.close() return decrypt_key(row[0]) if row else None # --- Helpers --- def cloudron_headers() -> dict: return {"Authorization": f"Bearer {CLOUDRON_TOKEN}"} async def cloudron_create_user(email: str, name: str) -> str: """Create (or return existing) Cloudron user, assigned to the members group. Verified user shape (from live API): {id, username, email, fallbackEmail, displayName, role, active, groupIds} Roles: "owner", "admin", "user". Group assignment is a SEPARATE call: PUT /api/v1/users/:userId/groups with body {"groupIds": [...]}. """ async with httpx.AsyncClient() as client: # Check if user exists r = await client.get( f"{CLOUDRON_API}/api/v1/users?per_page=100", headers=cloudron_headers(), ) r.raise_for_status() for u in r.json().get("users", []): if u.get("email") == email: # If the existing user has no display name (e.g. created with a # placeholder), fill it in. Otherwise leave it alone — the user # may have set their own name during account setup. if not u.get("displayName"): await cloudron_update_profile(u["id"], email, name) return u["id"] # Create user (role "user" = regular member). Only email is required; # Cloudron lets the user choose their own username during account setup. r = await client.post( f"{CLOUDRON_API}/api/v1/users", headers=cloudron_headers(), json={ "email": email, "displayName": name, "role": "user", "active": True, }, ) r.raise_for_status() return r.json()["id"] async def cloudron_set_group(user_id: str, group_id: str | None = None) -> None: """Assign a user to a group (defaults to the members group).""" gid = group_id or MEMBERS_GROUP_ID if not gid: logger.warning("No group id; skipping group assignment") return async with httpx.AsyncClient() as client: r = await client.put( f"{CLOUDRON_API}/api/v1/users/{user_id}/groups", headers=cloudron_headers(), json={"groupIds": [gid]}, ) r.raise_for_status() async def cloudron_set_active(user_id: str, active: bool) -> None: async with httpx.AsyncClient() as client: r = await client.put( f"{CLOUDRON_API}/api/v1/users/{user_id}/active", headers=cloudron_headers(), json={"active": active}, ) r.raise_for_status() async def owui_add_to_members_group(cloudron_user_id: str) -> bool: """Add a (new) member to Open WebUI's `members` group by their Cloudron uid. Open WebUI authenticates via Cloudron SSO and stores users keyed by the SSO `sub` — which is the Cloudron user id (uid-…). The shared docs-RAG knowledge collection is granted read to the `members` group, so group membership is what makes the docs visible. Called at provisioning time so a new member's FIRST chat session already has the docs (the nightly refresh sync is too slow for that). Best-effort: any failure logs a warning and returns False — provisioning must not fail over this; the nightly refresh self-heals within a day. """ if not OWUI_API_KEY: logger.warning("OWUI_API_KEY not set; skipping Open WebUI group add for %s", cloudron_user_id) return False try: async with httpx.AsyncClient(timeout=15.0) as client: # find the OWUI user (SSO `sub` = Cloudron uid); if they've never # logged in, the user row may not exist yet — the nightly sync # covers it, and group-add by uid also works after first login. r = await client.get( f"{OWUI_BASE}/api/v1/users/", headers={"Authorization": f"Bearer {OWUI_API_KEY}"}, params={"query": cloudron_user_id}, ) if r.status_code != 200: logger.warning("OWUI user lookup failed: %s %s", r.status_code, r.text[:150]) return False payload = r.json() or {} users = payload.get("users", []) if isinstance(payload, dict) else (payload or []) owui_uid = None for u in users: if u.get("id") == cloudron_user_id: owui_uid = u["id"] break if not owui_uid: # user hasn't logged in yet — they'll be added to the group by # the nightly sync after first login. Not an error. logger.info("OWUI: no user row yet for %s (not logged in); nightly sync will add", cloudron_user_id) return False r = await client.post( f"{OWUI_BASE}/api/v1/groups/id/{OWUI_MEMBERS_GROUP_ID}/users/add", headers={"Authorization": f"Bearer {OWUI_API_KEY}"}, json={"user_ids": [owui_uid]}, ) if r.status_code == 200: logger.info("OWUI: added %s to members group", cloudron_user_id) return True # 400 "already in group" is fine logger.warning("OWUI group add for %s: %s %s", cloudron_user_id, r.status_code, r.text[:150]) return False except Exception as e: logger.warning("OWUI group add failed for %s: %s", cloudron_user_id, e) return False async def cloudron_update_profile(user_id: str, email: str, display_name: str, fallback_email: str = "") -> None: """Update a user's profile (display name, email) after creation. Uses POST /api/v1/users/:userId/profile (the Cloudron "Update user" route). Needed because cloudron_create_user() only sets displayName at creation and returns early if the user already exists — so corrections require this. """ async with httpx.AsyncClient() as client: r = await client.post( f"{CLOUDRON_API}/api/v1/users/{user_id}/profile", headers=cloudron_headers(), json={"email": email, "displayName": display_name, "fallbackEmail": fallback_email}, ) r.raise_for_status() async def cloudron_get_invite_link(user_id: str) -> str: """Return the account-setup link WITHOUT sending Cloudron's own email. We use this so we can send our own branded email (with the setup link embedded) instead of Cloudron's default invite email. """ async with httpx.AsyncClient() as client: r = await client.get( f"{CLOUDRON_API}/api/v1/users/{user_id}/invite_link", headers=cloudron_headers(), ) r.raise_for_status() return r.json().get("inviteLink", "") async def cloudron_is_activated(user_id: str) -> bool: """Return True if the user has completed account setup (inviteAccepted). `inviteAccepted` flips true once the member clicks their setup link and chooses a username/password. Until then they're provisioned but not able to log in — the state we re-invite for. """ async with httpx.AsyncClient() as client: r = await client.get( f"{CLOUDRON_API}/api/v1/users?per_page=100", headers=cloudron_headers(), ) r.raise_for_status() for u in r.json().get("users", []): if u.get("id") == user_id: return bool(u.get("inviteAccepted")) return False async def provision_member(email: str, name: str, slug: str = "") -> str: """Provision a member end-to-end and send our own welcome email. Creates the Cloudron user (members group), issues a LiteLLM key under the member's team (budget = the member's current balance, not a hardcoded $15), stores the mapping, and sends our branded welcome email containing the Cloudron account-setup link (instead of Cloudron's default invite email). Idempotent: if the member already has a stored LiteLLM key, reuse it rather than minting a new one (LiteLLM rejects a duplicate `member:` alias with 400 — the bug that previously made re-provisioning of existing members fail). Re-sending this way also re-sends the welcome email, so it doubles as the manual "re-invite" path. """ user_id = await cloudron_create_user(email, name) await cloudron_set_group(user_id) await cloudron_set_active(user_id, True) balance = get_member_balance(email) team_id = await litellm_get_or_create_team(email, balance) # Pre-provision the member into Open WebUI's docs-RAG access group so their # FIRST chat session has the co-op docs. Best-effort (most new members # haven't logged in yet, so the OWUI user row doesn't exist — the nightly # refresh catches them after first login). try: await owui_add_to_members_group(user_id) except Exception as e: logger.warning("OWUI members-group add skipped for %s: %s", email, e) # Reuse an existing key if we already have one (the sk- value is stored in # our DB at creation time and is the authoritative token for this member's # chat key). Only mint a fresh key when there is none. key_token = get_stored_key(email) if not key_token: key_token = await litellm_create_key(email, balance, team_id) store_member(email, key_token, user_id, slug) # Send our own welcome email with the account-setup link (no OAuth step). setup_url = await cloudron_get_invite_link(user_id) subject, text, html = welcome_email(name, setup_url) send_email(email, subject, text, html) touch_welcome_sent(email) # Subscribe to the member newsletter (single opt-in; members consented by joining). await listmonk_sync(email, subscribe=True) # Auto-grant any HELD credit packs for this member (they pre-purchased # credits before joining — apply them now as part of provisioning). granted = 0.0 try: conn = get_db() rows = conn.execute( "SELECT id, amount FROM held_credits " "WHERE status = 'held' AND (LOWER(email) = ? OR LOWER(slug) = ?)", (email.lower(), slug.lower() if slug else ""), ).fetchall() if rows: hold_ids = [r[0] for r in rows] granted = sum(float(r[1] or 0.0) for r in rows) conn.execute( "UPDATE held_credits SET status = 'granted', resolved_at = datetime('now') " "WHERE id IN (%s)" % ",".join("?" * len(hold_ids)), hold_ids, ) conn.commit() conn.close() if granted > 0: new_balance = add_credits( email, granted, reason="held credit pack(s) granted on membership", reference=f"held:{','.join(str(r[0]) for r in rows)}", ) await sync_team_budget_from_balance(email) logger.info("Granted %.2f held credits to new member %s (balance %s)", granted, email, new_balance) except Exception as e: logger.warning("Held-credit grant for %s failed: %s", email, e) await sync_loomio_memberships() await sync_matrix_space() logger.info("Provisioned member %s (user_id=%s)", email, user_id) return user_id async def litellm_get_or_create_team(email: str, budget: float) -> str: """Return the team_id for a member's budget team, creating it if needed. Each member gets a LiteLLM "team" (a budget container) with a $15/30d budget. The team holds the chat key + any member-managed API keys, so all spend draws from the one pool. Idempotent: looks up the existing team by alias before creating. Note that /team/new is NOT idempotent (it creates a new team each call), so we must check /team/list first. """ alias = f"member:{email}" # /team/list has grown slow as usage accumulates; the httpx default 5s # read timeout was too short (root cause of dashboard 500s on 2026-09-26). async with httpx.AsyncClient(timeout=httpx.Timeout(connect=15.0, read=60.0, write=15.0, pool=15.0)) as client: r = await client.get( f"{LITELLM_BASE}/team/list", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, ) r.raise_for_status() for t in r.json(): if t.get("team_alias") == alias: return t["team_id"] r = await client.post( f"{LITELLM_BASE}/team/new", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, json={ "team_alias": alias, "max_budget": budget, "budget_duration": "30d", "models": MEMBER_MODELS, }, ) r.raise_for_status() return r.json()["team_id"] async def litellm_update_team_budget(team_id: str, budget: float) -> None: """Set a team's max_budget (the enforced spend cap). This is the primitive behind "payment → add to balance": a top-up adjusts the team's max_budget, which every key under the team (chat + API) draws from. Keys inherit the team budget (their own max_budget is ignored when a team_id is set), so the team is the single source of truth for allowance. """ async with httpx.AsyncClient() as client: r = await client.post( f"{LITELLM_BASE}/team/update", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, json={"team_id": team_id, "max_budget": budget}, ) r.raise_for_status() async def litellm_pin_reset_date(team_id: str, reset_at: datetime) -> None: """Pin a team's allowance reset to an exact moment (next OC payment date). LiteLLM's /team/update silently DROPS budget_reset_at (verified v1.74.0), so this writes the authoritative DB directly. Budgets reset to zero spend at the pinned instant; the pinned date is honored across restarts. """ def _write() -> None: conn = _pg_connect() try: cur = conn.cursor() cur.execute( 'UPDATE "LiteLLM_TeamTable" SET budget_reset_at = %s WHERE team_id = %s', (reset_at, team_id), ) conn.commit() finally: conn.close() await asyncio.get_event_loop().run_in_executor(None, _write) logger.info("Pinned budget reset for %s to %s", team_id, reset_at.isoformat()) def next_reset_from_now() -> datetime: """The member's next allowance reset: 30 days from now, on the hour.""" return datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0) + timedelta(days=30) async def sync_team_budget_from_balance(email: str) -> float: """Apply the member's effective budget to their LiteLLM team. Effective budget = monthly allowance + remaining non-expiring credits (credits drawn only after the allowance is exhausted this cycle). Reads the member's live spend from LiteLLM to determine how much credit has already been consumed in the current cycle. Returns the applied budget. """ balance = get_member_balance(email) credits = get_credit_balance(email) team_id = await litellm_get_or_create_team(email, balance) # Live spend this cycle (DB path preferred; LiteLLM computes the reset). team = get_team_budget_from_db(team_id) or {} team_spend = team.get("spend") eff = compute_max_budget(balance, credits, team_spend) await litellm_update_team_budget(team_id, eff["max_budget"]) return eff["max_budget"] async def litellm_create_key(email: str, budget: float, team_id: str) -> str: """Create a fresh LiteLLM virtual key under the member's team. Returns the ACTUAL sk- key (only revealed at generation time). We always create a new key rather than reusing — the sk- value is unrecoverable from a stored hash, so "reuse" would store a useless hash (a bug that previously broke chat for three members). NOTE: the key does NOT set its own max_budget — it inherits the team's budget (LiteLLM ignores a key's max_budget when team_id is set anyway). The team is the single source of truth for allowance, so a balance change (top-up) updates the team once and every key follows. The `budget` arg is retained only for the team-creation fallback in the migration path. """ alias = f"member:{email}" async with httpx.AsyncClient() as client: r = await client.post( f"{LITELLM_BASE}/key/generate", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, json={ "key_alias": alias, "team_id": team_id, "models": MEMBER_MODELS, }, ) r.raise_for_status() return r.json().get("key", "") async def litellm_disable_key(key_token: str) -> None: """Delete a member's LiteLLM key (on payment lapse). /key/delete accepts either the sk- key or its SHA256 hash, so this works for both current keys and legacy hash-keyed members. """ if not key_token: return async with httpx.AsyncClient() as client: await client.post( f"{LITELLM_BASE}/key/delete", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, json={"keys": [key_token]}, ) async def litellm_team_id_for_email(email: str) -> str | None: """The team_id of a member's budget team, or None if they have none.""" async with httpx.AsyncClient(timeout=httpx.Timeout(connect=15.0, read=60.0, write=15.0, pool=15.0)) as client: r = await client.get(f"{LITELLM_BASE}/team/list", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}) r.raise_for_status() for t in r.json(): if t.get("team_alias") == f"member:{email}": return t["team_id"] return None async def revoke_member_litellm_access(email: str) -> bool: """Cut a deactivated member's gateway access (SECURITY 2026-10-27). Zeroing the team budget blocks every key under it: member-managed keys carry no max_budget of their own (verified live 2026-10-27), so they inherit the team's. Deleting the key instead would break reactivation, which deliberately reuses the stored sk- token — and get_or_create_team does NOT update the budget of an existing team, so reactivation paths must re-sync explicitly. Best-effort by design: a failure here must not stop deactivation from completing, but it must be loud, because the member keeps access until it succeeds. """ try: team_id = await litellm_team_id_for_email(email) if not team_id: logger.info("Revoke: %s has no gateway team; nothing to zero", email) return True await litellm_update_team_budget(team_id, 0.0) logger.info("Revoke: %s gateway budget zeroed (team %s)", email, team_id) return True except Exception as e: logger.error("Revoke: FAILED to zero gateway budget for %s: %s", email, e) try: send_email("info@inference.coop", "Deactivation did not revoke gateway access", f"{email} was deactivated but their LiteLLM team budget could not be zeroed:\n\n{e}\n\n" "Their existing API keys may still work. Fix by hand: set the team's max_budget to 0.\n") except Exception: pass return False # --- Broker: member-managed API keys (scoped to the member's team) --- def _api_key_alias(email: str, name: str) -> str: return f"api:{email}:{name}" async def broker_create_api_key(email: str, name: str) -> dict: """Create a member-managed API key under the member's team. Returns {name, sk_token, created_at}. The key draws from the member's team budget and is tracked in member_api_keys so it can be listed/revoked by name. """ if not name or not re.fullmatch(r"[A-Za-z0-9._-]{1,64}", name): raise HTTPException(400, "Invalid key name (1-64 chars, alphanumeric/._-)") balance = get_member_balance(email) team_id = await litellm_get_or_create_team(email, balance) alias = _api_key_alias(email, name) async with httpx.AsyncClient() as client: # Rotation support: LiteLLM requires unique key aliases, so a re-create # with an existing (email, name) must retire the old key first. Without # this, rotating a key from the dashboard 400s ("alias already exists"). conn = get_db() row = conn.execute( "SELECT hash FROM member_api_keys WHERE email = ? AND name = ?", (email, name), ).fetchone() conn.close() if row and row[0]: old_hash = row[0] try: del_res = await client.post( f"{LITELLM_BASE}/key/delete", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, json={"keys": [old_hash]}, ) if del_res.status_code == 200: logger.info("Rotated API key %s for %s (old key retired)", name, email) else: logger.warning( "Old key retire failed for %s/%s: %s %s", email, name, del_res.status_code, del_res.text[:200], ) except Exception as exc: # noqa: BLE001 - rotation is best-effort logger.warning("Old key retire errored for %s/%s: %s", email, name, exc) r = await client.post( f"{LITELLM_BASE}/key/generate", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, json={ "key_alias": alias, "team_id": team_id, "models": MEMBER_MODELS, }, ) r.raise_for_status() data = r.json() sk_token = data.get("key", "") h = data.get("token") or data.get("token_id") or "" # Reveal the key ONCE (in the response). Persist only a fingerprint (last-4) # and the LiteLLM hash — the plaintext is NOT stored. Revocation uses the # hash, so the full key never needs to be retained. fingerprint = sk_token[-4:] if sk_token else "" conn = get_db() conn.execute( "INSERT INTO member_api_keys (email, name, sk_token, hash, fingerprint) VALUES (?, ?, ?, ?, ?) " "ON CONFLICT(email, name) DO UPDATE SET sk_token='', hash=excluded.hash, fingerprint=excluded.fingerprint, " "created_at=datetime('now')", (email, name, "", h, fingerprint), ) conn.commit() row = conn.execute( "SELECT created_at FROM member_api_keys WHERE email = ? AND name = ?", (email, name), ).fetchone() conn.close() return {"name": name, "sk_token": sk_token, "created_at": row[0] if row else None} async def broker_list_api_keys(email: str) -> list[dict]: conn = get_db() rows = conn.execute( "SELECT name, fingerprint, created_at FROM member_api_keys WHERE email = ? ORDER BY created_at DESC", (email,), ).fetchall() conn.close() return [ {"name": r[0], "fingerprint": r[1] or "", "created_at": r[2]} for r in rows ] async def broker_revoke_api_key(email: str, name: str) -> None: conn = get_db() row = conn.execute( "SELECT hash FROM member_api_keys WHERE email = ? AND name = ?", (email, name) ).fetchone() if not row: conn.close() raise HTTPException(404, "Key not found") conn.execute("DELETE FROM member_api_keys WHERE email = ? AND name = ?", (email, name)) conn.commit() conn.close() # Delete the underlying LiteLLM key by its hash. await litellm_disable_key(row[0]) # --- A. Open Collective webhook --- @app.post("/webhook/opencollective/{token}") @limiter.limit("30/minute") async def opencollective_webhook(request: Request, token: str): """Handle Open Collective membership events. Authenticated by a secret token in the URL path (Open Collective's generic webhooks are not HMAC-signed, so a secret URL is the standard way to authenticate them). """ if not WEBHOOK_TOKEN or not secrets.compare_digest(token, WEBHOOK_TOKEN): raise HTTPException(401, "Invalid webhook token") payload = await request.json() event_type = payload.get("type", "") data = payload.get("data", {}) # The webhook does NOT include email (Open Collective strips it for # privacy). We get name + slug, then look up the email via the admin token. # # Slug/name live in different places depending on the event type: # - order.processed / transaction.created → data.fromCollective.{slug,name} # - collective.member.created → data.member.memberCollective.{slug,name} from_collective = data.get("fromCollective", {}) member_collective = (data.get("member", {}) or {}).get("memberCollective", {}) slug = from_collective.get("slug") or member_collective.get("slug", "") name = from_collective.get("name") or member_collective.get("name", "Member") logger.info("Open Collective event: %s (name=%s, slug=%s)", event_type, name, slug) # Open Collective webhook events (verified): # - "order.processed" → fires on EVERY payment (incl. monthly recurring) # - "new member" → fires on FIRST contribution only # - payload has "firstPayment" boolean to distinguish new vs recurring # - "collective.transaction.created" is DEPRECATED (being removed) if event_type in ("order.processed", "new.member", "collective.member.created"): # New member → provision immediately and send our own welcome email. # The webhook strips email, so look it up by slug via the admin token. # Only provision on firstPayment (order.processed also fires on monthly # renewals, which should NOT re-send the welcome email). Default to # False so a missing field never triggers a spurious re-provision; the # nightly sweep backstops any missed first payment. first_payment = data.get("firstPayment", False) # Membership provision: only on payments to the "Membership" tier. A # one-time donation (no tier, or any other tier) grants BACKER role on # OC but is NOT a membership — don't provision it. # # Tier detection: the webhook payload carries only `data.order.TierId` # (numeric) — NO tier name. The order description embeds it # ("Financial contribution to Inference Cooperative (Credit pack)"). order = data.get("order") or {} order_desc = order.get("description") or "" m = re.search(r"\(([^)]+)\)\s*$", order_desc.strip()) tier_name = m.group(1).strip() if m else "" amount_cents = data.get("amount") or 0 if not amount_cents: amount_cents = order.get("totalAmount") or 0 # Credit-pack detection: one-time contributions to a "Credit pack" # tier route to the member's NON-EXPIRING credit balance instead of # provisioning/allowance. Detection: the OC tier name contains # "credit" (case-insensitive). Recurring subscriptions never route # here (a credit pack is a one-time purchase). if "credit" in tier_name.lower() and slug and event_type == "order.processed": members = await fetch_members() email = next((m["email"] for m in members if m["slug"] == slug), None) # The tier is VARIABLE-amount — no fixed face value. Credits are # granted on the contribution amount EXCLUDING the OC platform # tip (totalAmount - tip = amount), which we resolve via OC's # public order API. Fallback: webhook totalAmount (slightly # overgrants if a tip was included; logged). tip_free_cents = await get_order_amount(order.get("idV2")) if tip_free_cents is None: # SECURITY (2026-10-27): do NOT fall back to the payload's own # amount. This route is authenticated only by a URL token, so a # forged POST naming its own totalAmount used to mint credits for # any slug. Treat the payload as a hint; require a resolvable OC # order. A forged id cannot resolve; a transient OC failure holds # for manual reconciliation instead of granting blind. claimed = round(float(amount_cents) / 100.0, 2) logger.warning( "Credit pack REJECTED: OC could not resolve order %r (slug=%s, payload claimed $%.2f)" " — no credits granted", order.get("idV2"), slug, claimed, ) try: send_email( "info@inference.coop", "Credit pack could not be verified — no credits granted", f"A credit-pack webhook arrived for {slug} claiming ${claimed:.2f}, but\n" f"Open Collective's order API could not resolve order id {order.get('idV2')!r}.\n\n" "No credits were granted: the payload's amount is not trusted. If this was a\n" "real purchase, confirm the order in Open Collective and grant the credits\n" "by hand; if it was not, this was an attempt to mint credits.\n", ) except Exception as e: logger.warning("Credit pack rejection: operator email failed: %s", e) return JSONResponse({"status": "unverified_order_held", "slug": slug}) amount = round(tip_free_cents / 100.0, 2) charged = round(float(amount_cents) / 100.0, 2) if charged > amount: logger.info( "Credit pack: charged $%.2f including tip; granting tip-free $%.2f", charged, amount, ) oc_ref = f"oc:{slug}:{data.get('id', '')}" if email: # Member purchase → credits land immediately. new_balance = add_credits( email, amount, reason="credit pack purchase (Open Collective)", reference=oc_ref, ) # Push the enlarged budget immediately. await sync_team_budget_from_balance(email) logger.info("Credit pack: +%s credits for %s (balance %s)", amount, email, new_balance) return JSONResponse({ "status": "credits_added", "email": email, "credits_added": amount, "credit_balance": new_balance, }) # Non-member purchase → HOLD. Record it, email the buyer with # their two options (refund or join-and-apply). Credits are NOT # granted: a membership is required to use them. buyer_name = name or slug buyer_email = "" # The webhook payload strips emails; look up the payer via the # admin token (bot account sees contributor emails). try: # Paginated member fetch (raises on failure → caught below) for n in await fetch_all_member_nodes(): if (n.get("account") or {}).get("slug") == slug: buyer_email = ((n.get("account") or {}).get("email") or "").strip() buyer_name = (n.get("account") or {}).get("name") or buyer_name break except Exception as e: logger.warning("Credit-pack hold: payer lookup failed for %s: %s", slug, e) conn = get_db() conn.execute( "INSERT INTO held_credits (slug, email, name, amount, oc_reference) " "VALUES (?, ?, ?, ?, ?)", (slug, buyer_email or None, buyer_name, amount, oc_ref), ) conn.commit() conn.close() logger.warning( "Credit pack HELD from non-member %s (%.2f credits); buyer notified", slug, amount, ) if buyer_email: send_email( buyer_email, "Inference Cooperative — about your credit pack", f"Hi {buyer_name},\n\n" f"Thank you for your ${amount:.2f} credit pack purchase — we've received it.\n\n" "One thing to flag: credit packs are for members, and our records\n" "don't show you as one yet. You have two options:\n\n" "1. Join the co-op ($10-20/month, sliding scale):\n" " https://opencollective.com/inference-cooperative/contribute\n" f" Once you're a member, your ${amount:.2f} in credits will be\n" " applied to your account automatically.\n\n" "2. Request a refund:\n" " Reply to this email or write to info@inference.coop and\n" " we'll refund you in full.\n\n" "Sorry for the friction — memberships keep the co-op\n" "member-governed, which is rather the point of the place.\n\n" "— The Inference Cooperative\n" "https://inference.coop", ) else: logger.warning("Credit-pack hold: no buyer email found for %s; no notification sent", slug) return JSONResponse({ "status": "held", "note": "credit pack purchased without membership; hold recorded and buyer notified", }) if slug and "membership" in tier_name.lower() and event_type == "order.processed": # Every Membership-tier payment (first or recurring) re-pins the # member's allowance reset to payment + 30 days — the reset # follows the member's actual billing rhythm, not a global # calendar (all members landing on Oct 1 otherwise). members = await fetch_members() member = next((m for m in members if m["slug"] == slug), None) if member: if first_payment: try: await provision_member(member["email"], member["name"], member["slug"]) # A lapsed member's team budget was zeroed on deactivation and # get_or_create_team does not restore it — re-apply explicitly. await sync_team_budget_from_balance(member["email"]) except Exception as e: logger.warning("Webhook: failed to provision %s: %s", member["email"], e) try: team_id = await litellm_get_or_create_team( member["email"], get_member_balance(member["email"])) await litellm_pin_reset_date(team_id, next_reset_from_now()) except Exception as e: logger.warning("Webhook: failed to pin reset date for %s: %s", member["email"], e) elif slug: # Organization-backed membership: OC orgs never expose an email # to our integration, so fetch_members can't see them. Linked # orgs (members.org_slug) get their account re-activated on # dues; unlinked orgs ping the operator to link them. conn = get_db() row = conn.execute( "SELECT email, org_slug FROM members WHERE org_slug = ?", (slug,) ).fetchone() conn.close() if row: org_email, org_slug_col = row conn = get_db() conn.execute("UPDATE members SET active = 1 WHERE email = ?", (org_email,)) conn.commit() conn.close() logger.info("Webhook: org dues for %s — linked account %s re-activated", slug, org_email) if first_payment: try: await provision_member(org_email, name, slug) await sync_team_budget_from_balance(org_email) except Exception as e: logger.warning("Webhook: org re-provision failed for %s: %s", org_email, e) else: logger.warning( "Webhook: Membership payment from ORGANIZATION %s (%s) — " "no linked account; add via POST /admin/org-link", name, slug, ) send_email( "info@inference.coop", "Action needed: organizational membership without a linked account", f"Open Collective reports a Membership-tier payment from the\n" f"organization \"{name}\" (slug: {slug}).\n\n" "Organizations don't expose an email to our integration, so the\n" "membership needs to be linked to a portal account manually:\n\n" " POST /admin/org-link\n" " {\"org_slug\": \"" + slug + "\", \"email\": \"\"}\n\n" "Until linked, the organization cannot onboard.\n\n" "— The portal", ) elif slug and first_payment and event_type == "order.processed": # One-time donation (no tier or non-membership tier): NOT a # membership. Log it so we can see it, but don't provision. logger.info("One-time donation from slug %s (tier=%r) — not a membership, not provisioning", slug, tier_name) return JSONResponse({"status": "provisioned", "name": name, "slug": slug}) if event_type in ("collective.member.deleted", "collective.transaction.deleted"): # Lapsed member → move to the inactive group (lose access, keep data). # The webhook gives us the slug; we map it to the member's email/user. if slug: member = get_member_by_slug(slug) if member: # Move to inactive group (revokes chat access, keeps account + data) if INACTIVE_GROUP_ID and member.get("cloudron_user_id"): await cloudron_set_group(member["cloudron_user_id"], INACTIVE_GROUP_ID) # Mark inactive in our DB (key injector will refuse requests) deactivate_member(member["email"]) await revoke_member_litellm_access(member["email"]) await listmonk_sync(member["email"], subscribe=False) await sync_loomio_memberships() await sync_matrix_space() logger.info("Deactivated member %s (slug=%s)", member["email"], slug) return JSONResponse({"status": "deactivated", "email": member["email"]}) return JSONResponse({"status": "deactivated", "note": "no matching member"}) return JSONResponse({"status": "ignored", "type": event_type}) # --- B. Key injector (proxy) --- # --- Model picker: value-based naming --- # The chat model picker displays each model's `name` field. We set it to a # friendly name prefixed with an emoji for the provider's value tier, so the # flat picker list clusters by what members care about (privacy / green / # openness). The `id` field is left untouched — it remains the canonical key # used for actual API calls. Same emoji as the docs (models.md). # Emoji marks the FEATURE tier, not the provider: 🌱 = green/renewable-powered # (GreenPT and Regolo both), 🔒 = TEE-architected private (Tinfoil). MODEL_VALUE_EMOJI = { "tinfoil": "🔒", # architecturally private (TEE) "greenpt": "🌱", # 100% renewable energy "regolo": "🌱", # EU-hosted, zero data retention, 100% renewables } MODEL_FRIENDLY_NAMES = { "greenpt/deepseek-v4.1-flash": "DeepSeek V4.1", "tinfoil/deepseek-v4-1-flash": "DeepSeek V4.1", "tinfoil/glm-5.3": "GLM-5.3", "greenpt/glm-5.3-flash": "GLM-5.3 Flash", "greenpt/glm-5.3": "GLM-5.3", "tinfoil/gpt-oss-120b": "GPT-OSS 120B", "regolo/apertus-70b": "Apertus 70B", "regolo/qwen3.5-122b": "Qwen 3.5 122B", } @app.get("/v1/models") @limiter.limit("60/minute") async def list_models(request: Request): """List models (same for everyone — no per-user auth needed). Annotates each model with OpenWebUI metadata so web search is enabled by default (defaultFeatureIds includes 'web_search'). Without this, OpenWebUI leaves web search off for these models and members must toggle it manually. """ async with httpx.AsyncClient() as client: r = await client.get( f"{LITELLM_BASE}/v1/models", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, ) if r.status_code != 200: return Response( content=r.content, status_code=r.status_code, headers={"content-type": r.headers.get("content-type", "application/json")}, ) payload = r.json() # Chat-list filter: exclude audio-only models (STT/TTS) and embedding # models. They are registered and callable via their own routes # (/v1/audio/* and /v1/embeddings) for API users, but they are not chat # models — listing them in Open WebUI would just confuse members. The # gateway's model_info carries `mode` for these; fall back to a name # check for safety. audio_modes = {"audio_transcription", "audio_speech"} audio_name_hints = ("whisper", "voxtral", "tts") embedding_name_hints = ("embed",) # Image-generation models are used through Open WebUI's image feature # (Admin > Images), not as chat models — listing them in the picker # would fail on chat completions. image_modes = {"image_generation"} def is_audio_model(m) -> bool: mode = ((m.get("info") or {}).get("meta") or {}).get("mode") or "" if mode in audio_modes: return True mid = (m.get("id") or "").lower() return any(h in mid for h in audio_name_hints) def is_embedding_model(m) -> bool: mode = ((m.get("info") or {}).get("meta") or {}).get("mode") or "" if mode == "embedding": return True mid = (m.get("id") or "").lower() return any(h in mid for h in embedding_name_hints) image_name_hints = ("qwen-image",) def is_image_model(m) -> bool: mode = ((m.get("info") or {}).get("meta") or {}).get("mode") or "" if mode in image_modes: return True mid = (m.get("id") or "").lower() return any(h in mid for h in image_name_hints) payload["data"] = [ m for m in payload.get("data", []) if not is_audio_model(m) and not is_embedding_model(m) and not is_image_model(m) ] for model in payload.get("data", []): # Friendly display name: emoji value-tier + short name. The `id` # stays canonical (Open WebUI keys lookups and requests on it). mid = model.get("id", "") provider = mid.split("/", 1)[0] if "/" in mid else "" emoji = MODEL_VALUE_EMOJI.get(provider, "") friendly = MODEL_FRIENDLY_NAMES.get( mid, mid.split("/", 1)[-1] if "/" in mid else mid ) model["name"] = f"{emoji} {friendly}".strip() model.setdefault("info", {}).setdefault("meta", {}) model["info"]["meta"].setdefault("capabilities", {})["web_search"] = True model["info"]["meta"].setdefault("defaultFeatureIds", []) if "web_search" not in model["info"]["meta"]["defaultFeatureIds"]: model["info"]["meta"]["defaultFeatureIds"].append("web_search") # Attach the co-op docs knowledge collection to every model. This # makes Open WebUI treat the docs as the model's own knowledge: # automatic RAG injection grounds answers without tool-calling, and # the knowledge toolset is scoped to the attached collection instead # of exposing browse-all-knowledge tools (the "explored # search_knowledge_files… view_knowledge_file" wandering that new # members saw from the getting-started prompt). Env-driven so the # collection id can be rotated without a code change. if DOCS_KNOWLEDGE_ID: model["info"]["meta"].setdefault("knowledge", []) kb_entry = {"id": DOCS_KNOWLEDGE_ID, "type": "collection"} if kb_entry not in model["info"]["meta"]["knowledge"]: model["info"]["meta"]["knowledge"].append(kb_entry) # SILENT AUTO-RAG (legacy function-calling mode), chosen 2026-10-09: # with native FC the model gets a query_knowledge_files tool and # calls it on every prompt — including off-topic ones — with # visible "let me check the docs" narration. Prompt steering # failed twice (soft + firm). Legacy mode instead auto-injects # relevant docs chunks into context on every message: silent, # no tool-calls, no announcements, grounding always on. Cost: # one embedding + vector search per message (~300ms, paid even # on off-topic messages). builtinTools.knowledge=false removes # the knowledge tool + tag so the model # can't narrate or fumble; the injected sources still ground # co-op answers. model.setdefault("params", {}) model["params"]["function_calling"] = "legacy" model["info"]["meta"]["builtinTools"] = {"knowledge": False} # Steering: sources arrive as plain context. Tell the model the # docs may be present but are co-op-only, so it never cites or # mentions them for off-topic answers. steering = ( "Some messages include excerpts from the Inference " "Cooperative's own documentation (membership, models, API, " "governance), added automatically as background context. " "Use those excerpts only when they are relevant to the " "user's actual question. For general questions, ignore them " "and answer from your own knowledge — do not mention the " "co-op, its documentation, or any knowledge search unless " "the user explicitly asks about the co-op." ) existing = model["params"].get("system") if existing: if "Inference Cooperative" not in existing: model["params"]["system"] = existing + "\n\n" + steering else: model["params"]["system"] = steering return JSONResponse(content=payload) @app.api_route("/v1/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"]) @limiter.limit("120/minute") async def inject_key(request: Request, path: str): """Read the member's email from a header, inject their key, forward to LiteLLM.""" # Verify the request came through the chat frontend, so a direct caller # can't spoof the user-email header and use another member's key. # # Two supported auth paths: # 1. OpenWebUI signs the user identity as a JWT (X-OpenWebUI-User-Jwt) # with a shared secret — the email is tamper-proof, so no separate # secret header is needed. # 2. LibreChat sends a plain X-User-Email header plus an X-Portal-Secret # shared-secret header. email = "" jwt_header = request.headers.get("x-openwebui-user-jwt", "") if jwt_header and OWUI_JWT_SECRET: try: claims = jwt.decode(jwt_header, OWUI_JWT_SECRET, algorithms=["HS256"]) email = claims.get("email", "") except jwt.PyJWTError: raise HTTPException(401, "Invalid user identity token") else: _verify_portal_secret(request) email = ( request.headers.get("x-user-email", "") or request.headers.get("x-openwebui-user-email", "") ) if not email: raise HTTPException(401, "No member identity (user-email header)") # Look up the member's key from the persistent store member_key = get_member_key(email) if not member_key: raise HTTPException(403, "No active membership key") # Forward the request to LiteLLM with the member's key, STREAMING the # response back token-by-token (or byte-by-byte) rather than buffering it. # Buffering collapsed the model's stream and handed OpenWebUI the full # answer in one blob, so members saw nothing until generation finished — # which made chat feel slow even though time-to-first-byte was fine. body = await request.body() headers = dict(request.headers) headers["authorization"] = f"Bearer {member_key}" headers.pop("host", None) headers.pop("content-length", None) headers.pop("accept-encoding", None) # let httpx handle decompression timeout = httpx.Timeout(connect=15.0, read=600.0, write=30.0, pool=15.0) # Send the request upstream, but don't buffer the body — use a stream. client = httpx.AsyncClient(timeout=timeout) upstream_req = client.build_request( method=request.method, url=f"{LITELLM_BASE}/v1/{path}", headers=headers, content=body, ) upstream = await client.send(upstream_req, stream=True) async def gen(): try: async for chunk in upstream.aiter_raw(): yield chunk finally: await upstream.aclose() await client.aclose() # Pass through the content-type (critical for SSE streaming) and status. return StreamingResponse( gen(), status_code=upstream.status_code, headers={ "content-type": upstream.headers.get("content-type", "application/json"), "cache-control": "no-cache", }, ) # --- D. Admin / health --- @app.get("/health") async def health(): return {"status": "ok"} @app.post("/admin/sync-loomio") @limiter.limit("20/minute") async def admin_sync_loomio(request: Request): """Manually trigger a Loomio membership sync. Authenticated by the X-Admin-Token header. Useful for reconciling pre-existing members (accounts created before the sync existed) or recovering from a missed webhook. Idempotent — safe to call repeatedly. """ _verify_admin_token(request) await sync_loomio_memberships() await sync_matrix_space() return {"status": "synced"} @app.post("/admin/provision") @limiter.limit("20/minute") async def admin_provision(request: Request): """Manually provision a member by email (full pipeline). Authenticated by the X-Admin-Token header. Runs the full provisioning pipeline: Cloudron user + members group + LiteLLM key + our welcome email + Loomio sync. Body: {"email": "...", "name": "..."}. Idempotent. """ _verify_admin_token(request) body = await request.json() email = (body.get("email") or "").strip().lower() name = body.get("name") or email.split("@")[0] if not email: raise HTTPException(400, "Missing email") user_id = await provision_member(email, name, "") return {"status": "provisioned", "email": email, "user_id": user_id} @app.post("/admin/org-link") @limiter.limit("20/minute") async def admin_org_link(request: Request): """Link an OC organizational membership to a portal account. OC Organizations never expose an email to our integration, so org-backed memberships are linked manually: body {"org_slug": "...", "email": "...", "name": "..."}. Creates (or reuses) the members row for the email with org_slug set, and provisions the full pipeline — Cloudron user, LiteLLM key, and the welcome email containing the Cloudron account-setup link, so the organization's contact can complete signup. The nightly sweep keeps the account active while the org holds the Membership tier and deactivates it when the org's dues lapse (dues monitoring, automatic). """ _verify_admin_token(request) body = await request.json() org_slug = (body.get("org_slug") or "").strip().lower() email = (body.get("email") or "").strip().lower() name = (body.get("name") or org_slug).strip() if not org_slug or not email or "@" not in email: raise HTTPException(400, "org_slug and a valid email are required") try: org = (await fetch_org_backers()).get(org_slug) except Exception as e: # OC unreachable: link without provisioning — the sweep provisions and # activates once OC is reachable again. logger.warning("org-link: OC status fetch failed for %s (linking anyway): %s", org_slug, e) org = None conn = get_db() existing = conn.execute("SELECT email FROM members WHERE org_slug = ?", (org_slug,)).fetchone() # slug = the org slug (this row represents the org's contact, not an OC # individual — provision_member() overwrites slug/cloudron_user_id/key for # emails that resolve to a real OC account). conn.execute( "INSERT INTO members (email, slug, active, org_slug) VALUES (?, ?, 1, ?) " "ON CONFLICT(email) DO UPDATE SET org_slug = excluded.org_slug, active = 1", (email, org_slug, org_slug), ) conn.commit() conn.close() provisioned = False if org: # only provision while the org's membership is active try: await provision_member(email, name, org_slug) provisioned = True except Exception as e: logger.warning("org-link: provision failed for %s: %s", email, e) return { "status": "linked", "org_slug": org_slug, "email": email, "org_active": org is not None, "tier": (org or {}).get("tier"), "provisioned": provisioned, "replaced_existing_link": bool(existing), "note": "the nightly sweep deactivates this account automatically if the org's dues lapse", } @app.get("/admin/org-links") @limiter.limit("30/minute") async def admin_org_links(request: Request): """List organization-linked accounts with their live OC dues status.""" _verify_admin_token(request) org_backers = await fetch_org_backers() conn = get_db() rows = conn.execute( "SELECT email, org_slug, active FROM members WHERE org_slug IS NOT NULL" ).fetchall() conn.close() links = [] for email, org_slug, active in rows: org = org_backers.get(org_slug) links.append({ "org_slug": org_slug, "org_name": (org or {}).get("name") or org_slug, "email": email, "account_active": bool(active), "dues_active": org is not None, "tier": (org or {}).get("tier"), }) return {"links": links} @app.post("/admin/set-balance") @limiter.limit("20/minute") async def admin_set_balance(request: Request): """Set (or add to) a member's credit balance and sync it to LiteLLM. The flexible alternative to a fixed $15/month allowance. A payment (from OC or another system) is recorded by adjusting the balance here, which pushes the new cap to the member's team (every chat/API key under it follows). Body: {"email": "...", "balance": 25.0} → set absolute balance {"email": "...", "add": 10.0} → add to current balance """ _verify_admin_token(request) body = await request.json() email = (body.get("email") or "").strip().lower() if not email: raise HTTPException(400, "Missing email") current = get_member_balance(email) if "add" in body: new_balance = current + float(body["add"]) elif "balance" in body: new_balance = float(body["balance"]) else: raise HTTPException(400, "Provide 'balance' or 'add'") if new_balance < 0: raise HTTPException(400, "Balance cannot be negative") set_member_balance(email, new_balance) applied = await sync_team_budget_from_balance(email) return {"status": "updated", "email": email, "balance": new_balance, "team_budget": applied} @app.post("/admin/set-credits") @limiter.limit("20/minute") async def admin_set_credits(request: Request): """Adjust a member's non-expiring credit balance (with ledger entry). Body: {"email": "...", "add": 10.0, "reason": "goodwill"} → add/subtract {"email": "...", "set": 0.0, "reason": "correction"} → set absolute """ _verify_admin_token(request) body = await request.json() email = (body.get("email") or "").strip().lower() if not email: raise HTTPException(400, "Missing email") reason = (body.get("reason") or "admin adjustment").strip() if "add" in body: delta = float(body["add"]) new_balance = add_credits(email, delta, reason=reason) elif "set" in body: target = float(body["set"]) current = get_credit_balance(email) new_balance = add_credits(email, target - current, reason=reason) else: raise HTTPException(400, "Provide 'add' or 'set'") applied = await sync_team_budget_from_balance(email) return { "status": "updated", "email": email, "credit_balance": new_balance, "team_budget": applied, } @app.get("/admin/credits/{email}") @limiter.limit("30/minute") async def admin_credit_ledger(email: str, request: Request): """Return a member's credit balance and full ledger (audit trail).""" _verify_admin_token(request) conn = get_db() rows = conn.execute( "SELECT delta, reason, reference, created_at FROM credit_ledger " "WHERE email = ? ORDER BY created_at DESC", (email,), ).fetchall() conn.close() return { "email": email, "credit_balance": get_credit_balance(email), "ledger": [ {"delta": r[0], "reason": r[1], "reference": r[2], "created_at": r[3]} for r in rows ], } @app.get("/admin/held-credits") @limiter.limit("30/minute") async def admin_held_credits(request: Request): """List held credit packs (non-member purchases awaiting refund or grant).""" _verify_admin_token(request) conn = get_db() rows = conn.execute( "SELECT id, slug, email, name, amount, oc_reference, status, created_at, resolved_at " "FROM held_credits ORDER BY created_at DESC" ).fetchall() conn.close() return { "held": [ {"id": r[0], "slug": r[1], "email": r[2], "name": r[3], "amount": r[4], "reference": r[5], "status": r[6], "created_at": r[7], "resolved_at": r[8]} for r in rows ] } @app.post("/admin/held-credits/{hold_id}/refund") @limiter.limit("20/minute") async def admin_held_refund(hold_id: int, request: Request): """Mark a held credit pack as refunded (the refund itself is done manually in the Open Collective dashboard — this records it and updates the hold).""" _verify_admin_token(request) conn = get_db() cur = conn.execute( "UPDATE held_credits SET status = 'refunded', resolved_at = datetime('now') " "WHERE id = ? AND status = 'held'", (hold_id,), ) conn.commit() updated = cur.rowcount conn.close() if not updated: raise HTTPException(404, "Held credit not found (or already resolved)") return {"status": "refunded", "id": hold_id} # Operator accounts — shown under a separate "admin" status in the panel and # excluded from the member counts. They're infrastructure accounts (the SSO # owner and the bot), not dues-paying members. OPERATOR_EMAILS = { e.strip().lower() for e in os.environ.get("OPERATOR_EMAILS", "info@inference.coop,bot@inference.coop").split(",") if e.strip() } @app.get("/admin/overview") @limiter.limit("30/minute") async def admin_overview(request: Request): """Return a co-op-wide overview for the admin dashboard. Authenticated by the X-Admin-Token header. Returns every member (active + inactive) with their balance, spend, remaining, and team reset date, plus co-op aggregates (total members, total spend). """ _verify_admin_token(request) conn = get_db() rows = conn.execute( "SELECT email, active, balance, cloudron_user_id, org_slug, matrix_mxid FROM members ORDER BY email" ).fetchall() conn.close() # Fetch Cloudron users once to map user_id → username + inviteAccepted. # username is set only after the member completes account setup (chooses a # username + password via the setup link); inviteAccepted tracks that same # step, so "activated" = inviteAccepted (username present). usernames = {} invite_accepted = {} try: async with httpx.AsyncClient(timeout=15.0) as client: r = await client.get( f"{CLOUDRON_API}/api/v1/users?per_page=100", headers={"Authorization": f"Bearer {CLOUDRON_TOKEN}"}, ) r.raise_for_status() for u in r.json().get("users", []): uid = u.get("id") if u.get("username"): usernames[uid] = u["username"] invite_accepted[uid] = bool(u.get("inviteAccepted")) except Exception as e: logger.warning("admin overview: cloudron users fetch failed: %s", e) # Fetch team spend/budget from LiteLLM once. teams = {} try: async with httpx.AsyncClient(timeout=15.0) as client: r = await client.get( f"{LITELLM_BASE}/team/list", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, ) r.raise_for_status() for t in r.json(): teams[t.get("team_alias")] = t except Exception as e: logger.warning("admin overview: team/list failed: %s", e) # All-time spend per member team, from SpendLogs (the team `spend` counter # is cycle-only and resets at each member's budget_reset_at — the panel # needs both to avoid the "month change wiped the numbers" confusion). all_time_spend: dict[str, float] = {} try: conn = _pg_connect() try: with conn.cursor() as cur: cur.execute( """ SELECT t.team_alias, COALESCE(SUM(s.spend), 0) FROM "LiteLLM_TeamTable" t LEFT JOIN "LiteLLM_SpendLogs" s ON s."team_id" = t.team_id WHERE t.team_alias LIKE 'member:%' GROUP BY 1 """ ) all_time_spend = {alias: float(v or 0) for alias, v in cur.fetchall()} finally: conn.close() except Exception as e: logger.warning("admin overview: all-time spend query failed: %s", e) members = [] admins = [] total_spend = 0.0 total_all_time = 0.0 for email, active, balance, user_id, org_slug, matrix_mxid in rows: team = teams.get(f"member:{email}", {}) spend = float(team.get("spend") or 0.0) member_all_time = float(all_time_spend.get(f"member:{email}", 0.0)) # Normalize floating-point residue (e.g. -5.55e-17 after a cycle reset) # so the panel/dashboards never render "-0.00". if abs(spend) < 1e-9: spend = 0.0 if abs(member_all_time) < 1e-9: member_all_time = 0.0 credits = get_credit_balance(email) eff = compute_max_budget( float(balance) if balance is not None else MEMBER_BUDGET, credits, spend, ) entry = { "email": email, "all_time_spend": round(member_all_time, 2), "username": usernames.get(user_id), "activated": invite_accepted.get(user_id, False), "active": bool(active), "manual": email.lower() in MANUAL_MEMBERS, "org": org_slug or None, "matrix_mxid": matrix_mxid or None, "balance": float(balance) if balance is not None else MEMBER_BUDGET, "credit_balance": credits, "credits_remaining": max(0.0, credits - eff["credits_used"]), "allowance_used": eff["allowance_used"], "credits_used": eff["credits_used"], "spend": spend, "remaining": max(eff["max_budget"] - spend, 0.0), "reset_at": team.get("budget_reset_at"), "cloudron_user_id": user_id, } if email.lower() in OPERATOR_EMAILS: # Operator accounts: surfaced under a separate "admin" category in # the panel, excluded from member counts and aggregates (their # usage is a collective expense, not member activity). entry["admin"] = True admins.append(entry) total_spend += spend total_all_time += member_all_time else: members.append(entry) total_spend += spend total_all_time += member_all_time # Bot and sandbox teams: real spend, but not members. Surfaced in their own # category so the co-op's own usage is accountable without inflating member # counts — the same reasoning as the operator accounts above. agents = [ { "alias": alias, # Display name: the panel's section header already says "Agents", so the # "agent:" prefix is noise there. `alias` keeps the full string. "name": alias.split(":", 1)[1] if ":" in alias else alias, "spend": float(t.get("spend") or 0.0), "budget": t.get("max_budget"), "team_members": len(t.get("members_with_roles") or []), } for alias, t in teams.items() # member:* are members; sandbox:* are throwaway test accounts rather than # agents of the co-op. Their spend stops being visible anywhere once # filtered here — accepted knowingly. if alias and not alias.startswith(("member:", "sandbox:")) ] agents.sort(key=lambda a: a["spend"], reverse=True) return { "members": members, "admins": admins, "agents": agents, "totals": { "members": len(members), "active": sum(1 for m in members if m["active"]), "total_spend": round(total_spend, 2), "total_all_time_spend": round(total_all_time, 2), }, } def _metrics_provenance(tokens_measured: int | None, tokens_estimated: int | None) -> str: """Classify a day's measured-vs-estimated split for the API. "measured" — every token rested on a directly measured factor. "estimated" — every token rested on a non-measured factor (borrowed SAME_FAMILY or the flat proxy). "mixed" — the day's total is a blend of both. "none" — no provenance recorded: either the day has no output tokens or the row predates these columns (both split columns NULL). Deliberately NOT "measured" — an unknown split must never read as a confident measurement. """ if tokens_measured is None and tokens_estimated is None: return "none" m = tokens_measured or 0 e = tokens_estimated or 0 if m and e: return "mixed" if m: return "measured" if e: return "estimated" return "none" def metrics_series(days: int = 90) -> list[dict]: """Daily metrics rows, oldest first. One query, two callers: the admin panel (via /admin/metrics/series) and the member dashboard (via /broker/metrics). Columns are cumulative as of that day except spend_usd/requests/out_tokens/energy_wh/co2e_g, which are that day only; NULL means the source could not be read that day. energy_wh and co2e_g are computed in metrics_snapshot.py from the day's output tokens against a factor table: measured for GreenPT models (their API returns an `impact` object), estimated for the rest. See co-op/support#26. Each row also carries its provenance: `tokens_measured`/`tokens_estimated` (the output-token split across measured vs non-measured factors), `factor_revision` (a short stable hash of the factor table that produced the row), and a derived `provenance` indicator — measured | mixed | estimated | none. Rows written before these columns exist have NULL splits and report `provenance: "none"`, never a confident "measured". See docs/energy-metrics.md. """ limit = max(1, min(int(days or 90), 730)) conn = get_db() try: rows = conn.execute( "SELECT day, members_active, members_total, members_new, newsletter_subs, " "matrix_space_members, spend_usd, requests, out_tokens, energy_wh, co2e_g, " "tokens_measured, tokens_estimated, factor_revision, recorded_at " "FROM metrics_daily ORDER BY day DESC LIMIT ?", (limit,), ).fetchall() finally: conn.close() cols = [ "day", "members_active", "members_total", "members_new", "newsletter_subs", "matrix_space_members", "spend_usd", "requests", "out_tokens", "energy_wh", "co2e_g", "tokens_measured", "tokens_estimated", "factor_revision", "recorded_at", ] series = [] for row in reversed(rows): rec = dict(zip(cols, row)) rec["provenance"] = _metrics_provenance( rec["tokens_measured"], rec["tokens_estimated"] ) series.append(rec) return series @app.get("/admin/metrics/series") @limiter.limit("30/minute") async def admin_metrics_series(request: Request, days: int = 90): """Daily metrics series for charts. Authenticated by the X-Admin-Token header. Returns oldest-first rows from `metrics_daily` for the last `days` days, for the admin panel and the member dashboard to chart from one source of truth. A NULL column means that source could not be read on that day: render a gap, not a zero. Cumulative columns (members_*, newsletter_subs, matrix_space_members) are counts as of that day; spend_usd/requests are that day's traffic only. Backfilled days carry a LOWER BOUND for members_total (members provisioned without a welcome email are absent from the history). Each row carries its energy provenance — `provenance` (measured | mixed | estimated | none) plus `tokens_measured`, `tokens_estimated` and `factor_revision`. A row written before those columns existed reports `provenance: "none"`, not "measured". See docs/energy-metrics.md. """ _verify_admin_token(request) series = metrics_series(days) return {"series": series, "count": len(series), "days": max(1, min(int(days or 90), 730))} @app.get("/admin/model-usage") @limiter.limit("30/minute") async def admin_model_usage(request: Request, days: int = 0): """Co-op-wide per-model usage from LiteLLM spend logs. Authenticated by the X-Admin-Token header. Aggregates spend-log rows (across ALL members) by model: total spend, tokens, and request count, sorted by spend descending. Optional ?days=N limits to the last N days (0 or omitted = all-time). No per-member breakdown — this is co-op-level so admins can see which models the membership actually uses. Reads go straight to LiteLLM's Postgres when configured (fast, bounded); otherwise falls back to the HTTP API path. """ _verify_admin_token(request) # Fast path: direct Postgres aggregation. if litellm_db_ready(): try: return aggregate_model_usage(days=days) except Exception as e: logger.warning("admin model-usage: DB path failed (%s); falling back to HTTP", e) # Fallback path: LiteLLM HTTP API. cutoff = None if days and days > 0: from datetime import datetime, timedelta, timezone cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).strftime( "%Y-%m-%dT%H:%M:%S" ) models = {} total_spend = 0.0 total_tokens = 0 total_requests = 0 try: async with httpx.AsyncClient(timeout=30.0) as client: r = await client.get( f"{LITELLM_BASE}/spend/logs", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, ) r.raise_for_status() rows = r.json() if isinstance(rows, dict): rows = rows.get("data", rows.get("logs", [])) for row in rows if isinstance(rows, list) else []: # Prefer model_group: it carries our provider/model alias (e.g. # greenpt/glm-5.3-flash, publicai/apertus-v1.5-8b) when present; the # raw `model` field holds messy upstream/internal names. Older # rows predating the alias convention have model_group empty, # so fall back to `model` for those. if cutoff: started = str(row.get("startTime") or "") if started and started < cutoff: continue model = row.get("model_group") or row.get("model") or "unknown" m = models.setdefault(model, {"spend": 0.0, "tokens": 0, "requests": 0}) m["spend"] += float(row.get("spend") or 0.0) m["tokens"] += int(row.get("total_tokens") or 0) m["requests"] += 1 total_spend += float(row.get("spend") or 0.0) total_tokens += int(row.get("total_tokens") or 0) total_requests += 1 except Exception as e: logger.warning("admin model-usage: spend/logs fetch failed: %s", e) model_list = [ {"model": name, **stats} for name, stats in sorted(models.items(), key=lambda kv: kv[1]["spend"], reverse=True) ] return { "models": model_list, "totals": { "total_spend": total_spend, "total_tokens": total_tokens, "total_requests": total_requests, }, } # --- Broker HTTP endpoints (called by the member dashboard) --- # The dashboard is Cloudron-SSO-gated; it authenticates the member and passes # their email with a shared secret. The portal trusts the email only because it # carries BROKER_SECRET. These endpoints expose ONLY the authenticated member's # own keys — never another member's, and never the chat key. def _verify_broker_secret(request: Request) -> str: """Return the authenticated member's email, or 401. The member dashboard sends X-Broker-Secret (shared secret) + X-Member-Email (or X-Member-User, a Cloudron username). Both must be present and the secret must match, or we refuse (fail closed). Cloudron's proxyAuth injects the USERNAME (e.g. "ntnsndr"), not the email. So when the dashboard sends a username (no "@"), we resolve it to an email via the Cloudron user list (we already hold a Cloudron admin token). """ if not BROKER_SECRET: raise HTTPException(503, "Broker secret not configured") provided = request.headers.get("x-broker-secret", "") if not secrets.compare_digest(provided, BROKER_SECRET): raise HTTPException(401, "Invalid broker secret") identity = ( request.headers.get("x-member-email") or request.headers.get("x-member-user") or "" ).strip() if not identity: raise HTTPException(401, "Missing member identity") email = resolve_member_email(identity) # The member must be active in our DB before they can manage keys. if not get_member_key(email): raise HTTPException(403, "No active membership") return email def resolve_member_email(identity: str) -> str: """Resolve a member identity (email, Cloudron username, or uid) to email. Cloudron's OIDC `sub` claim is the user ID (uid-…), and email-invited users have no username at all. So accept three identity shapes: - email ("@") → return as-is - uid-… → match on the user id - username → match on the username field (legacy accounts) """ identity = identity.strip() if "@" in identity: return identity.lower() # uid / username → email via Cloudron user list (one fetch covers both). import urllib.request req = urllib.request.Request( f"{CLOUDRON_API}/api/v1/users?per_page=100", headers={"Authorization": f"Bearer {CLOUDRON_TOKEN}"}, ) with urllib.request.urlopen(req, timeout=15) as r: data = json.loads(r.read().decode()) users = data.get("users", []) for u in users: if u.get("id") == identity: return (u.get("email") or "").lower() for u in users: if u.get("username") == identity: return (u.get("email") or "").lower() raise HTTPException(403, f"No Cloudron user for identity {identity!r}") @app.get("/broker/keys") @limiter.limit("30/minute") async def broker_list(request: Request): email = _verify_broker_secret(request) return {"keys": await broker_list_api_keys(email)} @app.post("/broker/keys") @limiter.limit("30/minute") async def broker_create(request: Request): email = _verify_broker_secret(request) body = await request.json() name = (body.get("name") or "").strip() key = await broker_create_api_key(email, name) return {"status": "created", **key} @app.delete("/broker/keys/{name}") @limiter.limit("30/minute") async def broker_revoke(name: str, request: Request): email = _verify_broker_secret(request) await broker_revoke_api_key(email, name) return {"status": "revoked", "name": name} # Matrix MXID (self-serve federated identity). A member can attach ONE extra # Matrix account (e.g. their existing @user:matrix.org) to be invited to the # co-op Space. Empty string clears it. Validation is loose on purpose — Synapse # will simply reject a malformed MXID on invite, and we log that. _MXID_RE = re.compile(r"^@[^\s:]+:[^\s]+\.[^\s]+$") @app.get("/broker/matrix-mxid") @limiter.limit("30/minute") async def broker_get_matrix_mxid(request: Request): email = _verify_broker_secret(request) return {"matrix_mxid": get_member_matrix_mxid(email), "local_mxid": local_mxid_for_email(email)} def local_mxid_for_email(email: str) -> str: """Resolve a member's local Matrix ID (@username:inference.coop), or '' if unknown.""" import urllib.request try: req = urllib.request.Request( f"{CLOUDRON_API}/api/v1/users?per_page=100", headers={"Authorization": f"Bearer {CLOUDRON_TOKEN}"}, ) with urllib.request.urlopen(req, timeout=15) as r: users = json.loads(r.read().decode()).get("users", []) for u in users: if (u.get("email") or "").lower() == email and u.get("username"): return f"@{u['username']}:inference.coop" except Exception as e: logger.warning("local_mxid_for_email lookup failed: %s", e) return "" @app.post("/broker/matrix-mxid") @limiter.limit("30/minute") async def broker_set_matrix_mxid(request: Request): email = _verify_broker_secret(request) body = await request.json() mxid = (body.get("matrix_mxid") or "").strip() if mxid and not _MXID_RE.match(mxid): return JSONResponse( {"error": "Enter a Matrix ID like @username:server.org"}, status_code=400 ) set_member_matrix_mxid(email, mxid) # Invite the new account into the Space immediately (best-effort; the daily # sync will also pick it up). Clear = no invite (just removes the mapping). invite_sent = True if mxid: try: await sync_matrix_space() except Exception as e: logger.warning("matrix-mxid: immediate invite failed: %s", e) invite_sent = False # invite_sent=False tells the dashboard the nightly sync will retry; the # mapping is saved either way, so the member gets a confident confirmation. return {"status": "saved", "matrix_mxid": mxid, "invite_sent": invite_sent} async def broker_get_usage(email: str, days: int = 0) -> dict: """Return the member's spend vs. balance, plus per-model and token detail. This is the read-only primitive the member dashboard uses (no master key on the client). Reads go straight to LiteLLM's Postgres when configured (fast, bounded); otherwise falls back to the HTTP API path. `days` bounds the per-model breakdown window (0 = all time); balance/spend always come from the team's live totals. """ team_id = await litellm_get_or_create_team(email, get_member_balance(email)) allowance = get_member_balance(email) credits = get_credit_balance(email) spend = 0.0 reset_at = None # Fast path: direct Postgres reads. if litellm_db_ready(): try: team = get_team_budget_from_db(team_id) if team: if team["max_budget"] is not None: balance = float(team["max_budget"]) spend = float(team["spend"] or 0.0) reset_at = team["budget_reset_at"] usage = aggregate_model_usage(days=days, team_id=team_id) # All-time spend for this member: total of their spend-log rows # (the team `spend` counter resets with their allowance period). all_time_spend = 0.0 try: conn = _pg_connect() try: with conn.cursor() as cur: cur.execute( 'SELECT COALESCE(SUM(spend), 0) FROM "LiteLLM_SpendLogs" WHERE team_id = %s', (team_id,), ) all_time_spend = float(cur.fetchone()[0] or 0.0) finally: conn.close() except Exception as e: logger.warning("broker_get_usage: all-time spend query failed: %s", e) model_list = [ {"model": m["model"], "spend": m["spend"], "tokens": m["tokens"], "calls": m["requests"]} for m in usage["models"] ] return { "email": email, "balance": balance, "allowance": allowance, "credits": credits, "credits_remaining": max(0.0, credits - max(0.0, spend - allowance)), "spend": spend, "all_time_spend": round(all_time_spend, 2), "remaining": max(balance - spend, 0.0), "total_tokens": usage["totals"]["total_tokens"], "reset_at": reset_at, "models": model_list, } except Exception as e: logger.warning("broker_get_usage: DB path failed (%s); falling back to HTTP", e) # Fallback path: LiteLLM HTTP API. # /team/list and /spend/logs have grown slow as usage accumulates; the # httpx default 5s read timeout was too short and made /broker/usage 500. client_timeout = httpx.Timeout(connect=15.0, read=60.0, write=15.0, pool=15.0) async with httpx.AsyncClient(timeout=client_timeout) as client: # 1. Team spend + budget r = await client.get( f"{LITELLM_BASE}/team/list", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, ) r.raise_for_status() for t in r.json(): if t.get("team_id") == team_id: balance = float(t.get("max_budget") or balance) spend = float(t.get("spend") or 0.0) reset_at = t.get("budget_reset_at") break # 2. Per-model token/spend breakdown from spend logs (filtered by team). # /spend/logs does not reliably filter by team_id server-side, so we # filter client-side on the returned rows. models = {} total_tokens = 0 try: r2 = await client.get( f"{LITELLM_BASE}/spend/logs", headers={"Authorization": f"Bearer {LITELLM_MASTER_KEY}"}, ) r2.raise_for_status() rows = r2.json() if isinstance(rows, dict): rows = rows.get("data", rows.get("logs", [])) for row in rows if isinstance(rows, list) else []: if row.get("team_id") != team_id: continue # Prefer model_group: it carries our provider/model alias (e.g. # greenpt/glm-5.3-flash) when present; the raw `model` field holds # messy upstream/internal names. Older rows predating the alias # convention have model_group empty, so fall back to `model`. model = row.get("model_group") or row.get("model") or "unknown" m = models.setdefault(model, {"spend": 0.0, "tokens": 0, "calls": 0}) m["spend"] += float(row.get("spend") or 0.0) m["tokens"] += int(row.get("total_tokens") or 0) m["calls"] += 1 total_tokens += int(row.get("total_tokens") or 0) except Exception as e: # spend logs are best-effort; don't fail the whole view logger.warning("spend/logs fetch failed: %s", e) # All-time spend from the same log rows (team counter is period-only). all_time_spend = round(sum(m["spend"] for m in models.values()), 2) # Sort models by spend descending (which models are draining the most). model_list = [ {"model": name, **stats} for name, stats in sorted(models.items(), key=lambda kv: kv[1]["spend"], reverse=True) ] return { "email": email, "balance": balance, "allowance": allowance, "credits": credits, "credits_remaining": max(0.0, credits - max(0.0, spend - allowance)), "spend": spend, "all_time_spend": all_time_spend, "remaining": max(balance - spend, 0.0), "total_tokens": total_tokens, "reset_at": reset_at, "models": model_list, } @app.get("/broker/usage") @limiter.limit("60/minute") async def broker_usage(request: Request, days: int = 0): email = _verify_broker_secret(request) return await broker_get_usage(email, days=days) @app.get("/broker/metrics") @limiter.limit("30/minute") async def broker_metrics(request: Request, days: int = 90): """Co-op metrics over time, for the member dashboard. The same series the admin panel reads. It comes through the broker because it is AGGREGATE and non-personal — co-op size, requests served, spend — the same class of social-proof data as /broker/model-leaderboard. No per-member figure passes through here, and members asked to see how the co-op is doing. Rows carry the energy provenance the admin series does: `provenance` (measured | mixed | estimated | none) plus `tokens_measured`, `tokens_estimated` and `factor_revision` — so a member-facing chart can say how much of a day's figure rested on a measurement. NULL splits (older rows) report `provenance: "none"`. See docs/energy-metrics.md. """ _verify_broker_secret(request) series = metrics_series(days) return {"series": series, "count": len(series), "days": max(1, min(int(days or 90), 730))} @app.get("/broker/model-leaderboard") @limiter.limit("30/minute") async def broker_model_leaderboard(request: Request, days: int = 0): """Co-op-wide model popularity for the member dashboard. Same aggregation as the admin model-usage view (spend/tokens/requests per model, across all members) but served through the broker so members can see which models the co-op uses most — social guidance for model choice. No per-member data; aggregate totals only. """ _verify_broker_secret(request) if litellm_db_ready(): try: usage = aggregate_model_usage(days=days) except Exception as e: logger.warning("broker leaderboard: DB path failed (%s)", e) usage = None if usage: return { "models": [ {"model": m["model"], "spend": m["spend"], "tokens": m["tokens"], "requests": m["requests"]} for m in usage["models"][:10] ] } # Fallback: empty (rare; the DB path is the norm) return {"models": []} async def reconcile_memberships() -> dict: """Reconcile the portal against the live Open Collective member list. - Provisions members who are BACKER/ADMIN on OC but not yet in our DB (missed webhook, or contributed before the email step existed). - Deactivates members who are in our DB but no longer BACKER/ADMIN (lapsed or cancelled membership). - Syncs Loomio to the resulting active-member set. Idempotent and safe to run repeatedly. Returns a summary of actions taken. """ # fetch_members() raises when OC is unreachable or the query fails; treat # that like an empty result so the documented fail-safe below actually # applies, instead of letting the exception escape and 500 the sweep. try: active = await fetch_members() except Exception as e: logger.warning( "Sweep: member fetch failed; skipping reconciliation (fail-safe): %s", e ) return { "status": "skipped", "reason": "member_fetch_failed", "provisioned": 0, "deactivated": 0, "reinvited": 0, "newsletter_removed": 0, } active_emails = {m["email"] for m in active} active_by_email = {m["email"]: m for m in active} # FAIL-SAFE: if the OC fetch returned nothing (token expired, OC outage, # or a query error), we must NOT deactivate everyone — that would be a # catastrophic false-positive. Skip reconciliation entirely in that case. if not active: logger.warning("Sweep: fetch_members returned empty; skipping reconciliation (fail-safe)") return {"status": "skipped", "reason": "empty_member_list", "provisioned": 0, "deactivated": 0} # Currently provisioned members (from our DB). conn = get_db() rows = conn.execute("SELECT email, cloudron_user_id FROM members WHERE active = 1").fetchall() conn.close() provisioned = {r[0]: r[1] for r in rows} provisioned_count = 0 deactivated_count = 0 # 0. Organization-backed memberships: dues paid by an OC Organization. # Linked accounts (members.org_slug) stay active while their org holds a # BACKER/ADMIN role on the Membership tier; lapsed orgs deactivate like # individuals. Fail-safe: if the org fetch fails, keep linked accounts # active (never deactivate on our own fetch failure). try: org_backers = await fetch_org_backers() org_fetch_ok = True except Exception as e: logger.warning("Sweep: org fetch failed (keeping linked accounts active): %s", e) org_backers = {} org_fetch_ok = False if org_backers or not org_fetch_ok: conn = get_db() org_rows = conn.execute( "SELECT email, org_slug FROM members WHERE org_slug IS NOT NULL" ).fetchall() conn.close() for org_email, org_slug in org_rows: org = org_backers.get(org_slug) if org_fetch_ok and not org: continue # org lapsed → falls through to deactivation below if org_email not in active_emails: active.append({ "email": org_email, "name": (org or {}).get("name") or org_slug, "slug": org_slug, "role": (org or {}).get("role", "BACKER"), "tier": ((org or {}).get("tier", "") or "") + " (org)", }) active_emails.add(org_email) logger.info("Sweep: org-linked member %s active via org %s", org_email, org_slug) # 1. Provision missing members. for m in active: email = m["email"] if email not in provisioned: try: await provision_member(email, m["name"], m["slug"]) provisioned_count += 1 logger.info("Sweep: provisioned %s", email) except Exception as e: logger.warning("Sweep: failed to provision %s: %s", email, e) # 2. Deactivate lapsed members (but never manual/founder members, who have # no OC membership to lapse). for email, user_id in provisioned.items(): if email not in active_emails and email not in MANUAL_MEMBERS: try: if INACTIVE_GROUP_ID and user_id: await cloudron_set_group(user_id, INACTIVE_GROUP_ID) deactivate_member(email) await revoke_member_litellm_access(email) await listmonk_sync(email, subscribe=False) deactivated_count += 1 logger.info("Sweep: deactivated %s", email) except Exception as e: logger.warning("Sweep: failed to deactivate %s: %s", email, e) # 2b. Re-invite never-activated members (provisioned but no account setup). # A member who was provisioned (e.g. webhook fired while the OC token was # missing the `email` scope, or the invite email went to spam) may sit # active-but-unactivated indefinitely. Re-send their welcome email, but # throttle by REINVITE_AFTER_DAYS so we don't spam them nightly. reinvited_count = 0 conn = get_db() reinvite_rows = conn.execute( "SELECT email, cloudron_user_id, welcome_sent_at FROM members WHERE active = 1" ).fetchall() conn.close() now = datetime.now(timezone.utc) for email, user_id, sent_at in reinvite_rows: if not user_id: continue # Skip if already activated. try: if await cloudron_is_activated(user_id): continue except Exception as e: logger.warning("Sweep: activation check failed for %s: %s", email, e) continue # Throttle: only re-invite if the last send was > REINVITE_AFTER_DAYS ago. if sent_at: try: last = datetime.fromisoformat(sent_at) # stored as UTC naive (datetime('now') is UTC); attach tz if last.tzinfo is None: last = last.replace(tzinfo=timezone.utc) if (now - last).total_seconds() < REINVITE_AFTER_DAYS * 86400: continue except Exception: pass # Re-send the welcome email (reuses the stored key — no duplicate alias). try: name = active_by_email.get(email, {}).get("name") or email.split("@")[0] await provision_member(email, name, "") reinvited_count += 1 logger.info("Sweep: re-invited %s (never activated)", email) except Exception as e: logger.warning("Sweep: failed to re-invite %s: %s", email, e) # 3. Sync Loomio. await sync_loomio_memberships() await sync_matrix_space() # 4. Reconcile the newsletter list against active membership (removes # deleted accounts, never-activated test accounts, lapsed members). newsletter = await listmonk_reconcile() return { "status": "reconciled", "provisioned": provisioned_count, "deactivated": deactivated_count, "reinvited": reinvited_count, "newsletter_removed": newsletter.get("removed", 0), } @app.post("/admin/sweep") @limiter.limit("20/minute") async def admin_sweep(request: Request): """Manually trigger a full membership reconciliation. Authenticated by the X-Admin-Token header. Idempotent — safe to call repeatedly. This is also what the Cloudron scheduler invokes on a cron schedule (see the sweep script). """ _verify_admin_token(request) result = await reconcile_memberships() # Budget notices ride the same cron the Cloudron scheduler already hits. They are OFF # unless BUDGET_NOTICES=on, and wrapped so a failure here can never break reconciliation. try: result["budget_notices"] = await budget_notice_sweep(send=(budget_notices_enabled() == "on")) except Exception as exc: # noqa: BLE001 result["budget_notices"] = {"error": str(exc)[:200]} return result @app.post("/admin/newsletter/reconcile") @limiter.limit("20/minute") async def admin_newsletter_reconcile(request: Request): """Remove stale subscribers from the Members newsletter list. The newsletter list must mirror ACTIVE members: deleted accounts, never-activated test accounts, and lapsed members are removed (see listmonk_reconcile for the exact rules). Also runs automatically as step 4 of the nightly membership sweep. Authenticated by the X-Admin-Token header. """ _verify_admin_token(request) return await listmonk_reconcile() def _verify_sync_token(request: Request) -> None: """Authenticate the LibreChat key-sync feed via X-Sync-Token. Dedicated machine secret (least privilege): it only ever exposes the members' own chat keys, never the admin API. The same secret is set on the LibreChat sandbox app (LIBRECHAT_SYNC_TOKEN in /app/data/env). """ expected = os.environ.get("LIBRECHAT_SYNC_TOKEN", "") if not expected: raise HTTPException(503, "Sync token not configured") provided = request.headers.get("x-sync-token", "") if not secrets.compare_digest(provided, expected): raise HTTPException(401, "Invalid sync token") @app.get("/admin/librechat/keys") @limiter.limit("30/minute") async def admin_librechat_keys(request: Request): """Key feed for the LibreChat sandbox's per-member key auto-load. Returns every ACTIVE member's portal-provisioned chat key so the sandbox can seed LibreChat's user-key store and members never see an API-key prompt (chat bills to the member's own LiteLLM team/budget). Auth is the dedicated X-Sync-Token header — NOT the admin token. """ _verify_sync_token(request) conn = get_db() rows = conn.execute( "SELECT email, key_token FROM members WHERE active = 1 AND key_token != ''" ).fetchall() conn.close() return { "members": [ {"email": email, "key": decrypt_key(key_token)} for email, key_token in rows ] } @app.get("/") async def index(): return {"service": "Inference Cooperative Member Portal", "version": "0.1.0"} # --------------------------------------------------------------------------- # # Budget notices — email a member as they approach their monthly limit. # # --------------------------------------------------------------------------- # # OFF by default: BUDGET_NOTICES = off | dryrun | on. Nothing is ever emailed # unless it is "on"; the admin endpoint below can preview at any time, including # while off. One notice per level per budget cycle, so a member hears from us # once when they cross 80% and once when they reach the cap — never on every # sweep. Spend and budget are read from LiteLLM's own Postgres, the same source # the dashboard and admin panel display, so the numbers cannot drift. # # CREDIT_PACK_URL is configuration because the website has no dedicated pack # link today; the Open Collective contribution page is what actually credits a # member (the webhook settles contributions into credits). # Generic on purpose: the balance can be short because the monthly allowance is spent, # because purchased credit is spent, or a mix — the letter says "balance" so it is true # either way. The trigger is proportional — 90% of the balance used — so it means the same # thing to a member on $15 and one on $35: the letter arrives when roughly a tenth is left, # on whichever sweep follows, rather than at an arbitrary absolute amount. BUDGET_NOTICE_AT = float(os.environ.get("BUDGET_NOTICE_AT", "0.90")) PINE_LOGO_SVG = """""" def budget_notices_enabled() -> str: """BUDGET_NOTICES: off (default) | dryrun | on.""" return (os.environ.get("BUDGET_NOTICES", "off") or "off").strip().lower() def credit_pack_url() -> str: return os.environ.get( "CREDIT_PACK_URL", "https://opencollective.com/inference-cooperative/contribute" ) def _ensure_budget_notices_table(conn) -> None: conn.execute( "CREATE TABLE IF NOT EXISTS budget_notices (" " email TEXT NOT NULL, level TEXT NOT NULL, cycle TEXT NOT NULL," " sent_at TEXT NOT NULL, detail TEXT," " PRIMARY KEY (email, level, cycle))" ) conn.commit() def budget_notice_email( name: str, spend: float, budget: float, remaining: float, pack_url: str ) -> tuple[str, str, str]: """Build the low-balance notice (subject, text, html). Deliberately generic about where the balance came from: a member may be short because the monthly allowance is spent, because credit they bought is spent, or a mix of both. The wording says "balance" so it is true either way, and it names no mechanism a member would have to reconcile against their own dashboard. Same shell as welcome_email(): cream background, white card, forest-green button, inline pine-tree mark, table-based layout, no external assets. """ subject = "Your Inference Cooperative balance is running low" heading = "Your balance is running low" lede = (f"You have ${remaining:.2f} of your ${budget:.2f} balance left. When it " "runs out, chat and API requests stop until you add more.") money = f"${remaining:.2f} remaining" text = ( f"Hi {name},\n\n" f"{lede}\n\n" f"{money}\n\n" "If you'd like to keep going, you can buy credit packs here:\n\n" f"{pack_url}\n\n" "This is the only notice of its kind you'll get until your balance goes back " "above the line and runs low again. Questions? Reply to this email or write to " "info@inference.coop.\n\n" "— The Inference Cooperative\n" ) html = f"""
{PINE_LOGO_SVG}

{heading}

{lede}

{money}

Buy credit packs

This is the only notice of its kind you'll get until your balance goes back above the line and runs low again.
Questions? Reply to this email or write to info@inference.coop.
The Inference Cooperative · inference.coop
""" return subject, text, html async def budget_notice_sweep(send: bool = False) -> dict: """Find members whose balance has fallen below the low-water mark, and optionally email. `send=False` is a preview: it computes exactly who would be written to, with the numbers, and sends nothing. `send=True` emails and records the send — the caller is responsible for requiring BUDGET_NOTICES=on first. """ mode = budget_notices_enabled() if send and mode != "on": return {"refused": f"BUDGET_NOTICES is '{mode}', not 'on' — nothing sent"} conn = get_db() _ensure_budget_notices_table(conn) cols = {r[1] for r in conn.execute("PRAGMA table_info(members)")} # There is no name column in `members` — `slug` is the username (e.g. ntnsndr). Falling # back to the email local part produced "Hi n," for the operator, which is how this was # caught. Prefer the username; say "there" rather than inventing a name. ident = "slug" if "slug" in cols else None sel = "SELECT email" + (f", {ident}" if ident else "") + " FROM members WHERE active = 1" rows = conn.execute(sel).fetchall() pack = credit_pack_url() preview, sent, skipped = [], 0, 0 for row in rows: email = row[0] name = ((row[1] or "").strip() if len(row) > 1 else "") or "there" team_id = await litellm_team_id_for_email(email) if not team_id: skipped += 1 continue stats = get_team_budget_from_db(team_id) or {} budget = stats.get("max_budget") spend = float(stats.get("spend") or 0.0) if not budget or budget <= 0: skipped += 1 # no cap configured: nothing to warn about continue remaining = float(budget) - spend if spend < BUDGET_NOTICE_AT * float(budget): # Above the line — they added funds, or the allowance reset. Clear the latch so the # NEXT crossing sends a fresh notice. That is what makes it once each time, and it # is why this is a latch rather than a key on the budget cycle. conn.execute("DELETE FROM budget_notices WHERE email = ?", (email,)) conn.commit() continue if conn.execute("SELECT 1 FROM budget_notices WHERE email = ?", (email,)).fetchone(): skipped += 1 # already told them for this crossing continue subject, text_body, html_body = budget_notice_email(name, spend, budget, remaining, pack) entry = { "email": email, "remaining": round(remaining, 2), "spend": round(spend, 2), "budget": round(budget, 2), "resets": str(stats.get("budget_reset_at") or stats.get("reset_at") or "")[:19], } if send: ok = send_email(email, subject, text_body, html_body) entry["sent"] = bool(ok) if ok: conn.execute( "INSERT OR REPLACE INTO budget_notices" " (email, level, cycle, sent_at, detail) VALUES (?, 'low', ?, datetime('now'), ?)", (email, entry["resets"] or "crossing", f"remaining={remaining:.4f}"), ) conn.commit() sent += 1 else: entry["subject"] = subject preview.append(entry) conn.close() if send: return {"mode": mode, "sent": sent, "considered": len(rows), "skipped": skipped, "notices": preview} return {"mode": mode, "dry_run": True, "considered": len(rows), "skipped": skipped, "would_notify": len(preview), "notices": preview} async def budget_notice_test(dest: str) -> dict: """Email one sample low-balance letter to `dest`, using that member's real numbers. Deliberately does NOT write to budget_notices: a test must not consume the real once-per-crossing latch, or reviewing the copy would silence the actual notice. Works while BUDGET_NOTICES is off, so copy can be reviewed before arming anything. """ conn = get_db() row = conn.execute("SELECT slug FROM members WHERE email = ?", (dest,)).fetchone() name = ((row[0] or "").strip() if row else "") or "there" conn.close() budget, spend = 15.0, 13.34 # sample numbers if they aren't a member if row: team_id = await litellm_team_id_for_email(dest) stats = (get_team_budget_from_db(team_id) or {}) if team_id else {} if stats.get("max_budget"): budget = float(stats["max_budget"]) spend = float(stats.get("spend") or 0.0) remaining = budget - spend subject, text_body, html_body = budget_notice_email( name, spend, budget, remaining, credit_pack_url() ) ok = send_email(dest, subject, text_body, html_body) return { "mode": "test", "to": dest, "sent": bool(ok), "subject": subject, "rendered": {"name": name, "spend": round(spend, 2), "budget": round(budget, 2), "remaining": round(remaining, 2)}, "note": "no ledger row written — the real once-per-crossing latch is untouched", } @app.post("/admin/budget-notices") @limiter.limit("10/minute") async def admin_budget_notices(request: Request, mode: str = "dryrun", to: str | None = None): """Preview (default), send, or email a single sample. Authenticated by X-Admin-Token. `mode=dryrun` never emails; `mode=send` requires BUDGET_NOTICES=on and refuses otherwise; `mode=test&to=
` emails one sample using that member's real numbers so the copy can be read in a mail client first. A test works while the flag is off and leaves no trace in the ledger. """ _verify_admin_token(request) if mode == "test": dest = (to or "").strip() if not dest: return {"refused": "mode=test needs &to=
"} return await budget_notice_test(dest) if mode == "send": if budget_notices_enabled() != "on": return {"refused": f"BUDGET_NOTICES is '{budget_notices_enabled()}', not 'on'" " — set it on the portal app first"} return await budget_notice_sweep(send=True) return await budget_notice_sweep(send=False)