routes/vault.py
Rama main · réplica de lectura del repositorio original.
"""
Vault API — credential management, OAuth flows, sync, and testing.
Replaces the old /api/admin/ai-config endpoints with a dynamic,
plugin-aware credential system.
"""
import json
import logging
import os
from functools import wraps
from flask import Blueprint, current_app, g, jsonify, redirect, request
logger = logging.getLogger(__name__)
vault_bp = Blueprint('vault', __name__)
def _platform_admin_only(view):
"""SEC-069: the platform-setup surface edits .platform-oauth.env — the
PLATFORM's shared OAuth app credentials, consumed by every tenant's OVU.
The app-level gate (app.py _ADMIN_ONLY_PREFIXES → is_admin_user) admits
anyone in the TENANT's ADMIN_USER_IDS, i.e. the tenant owner. That is the
right gate for the owner's own vault, and the wrong gate here: a tenant
owner must not be able to rewrite platform-wide OAuth client secrets from
their own admin UI. This decorator adds the stricter test. It runs AFTER
app.py's gate, so an anonymous caller still gets 401 and a non-admin
tenant user still gets 403 admin_required exactly as before.
Local / self-hosted mode (no Clerk key): the whole app is open by design
and the operator IS the platform, so the decorator is a no-op there — the
same rule app.py applies to every other admin surface. The switch is read
from app.config['CLERK_AUTH_ENABLED'] (set by create_app from the SAME env
read as the app gate); an app that never set it falls back to the
import-time view, which fails closed when a Clerk key is present.
"""
@wraps(view)
def wrapped(*args, **kwargs):
import services.auth as _auth
gate_on = current_app.config.get('CLERK_AUTH_ENABLED', _auth._CLERK_CONFIGURED)
if gate_on:
user_id = getattr(g, 'clerk_user_id', None)
if not _auth.is_platform_admin(user_id):
logger.warning('Platform-setup denied: user_id=%s path=%s (tenant admin is not platform admin)',
user_id, request.path)
return jsonify({
'error': 'Platform OAuth setup is managed by the platform operator, not per tenant.',
'code': 'platform_admin_only',
}), 403
return view(*args, **kwargs)
return wrapped
def _get_username() -> str:
"""
Determine the current client username.
In multi-tenant JamBot, this comes from the DOMAIN env var (e.g., tenant-l.jam-bot.com → tenant-l).
In single-user mode, defaults to 'default'.
"""
domain = os.getenv('DOMAIN', '')
if domain and '.jam-bot.com' in domain:
return domain.split('.')[0]
return os.getenv('VAULT_USERNAME', 'default')
# ---------------------------------------------------------------------------
# Credential CRUD
# ---------------------------------------------------------------------------
@vault_bp.route('/api/vault/credentials', methods=['GET'])
def list_credentials():
"""
List all credentials with status — merged from platform catalog + installed plugins.
Grouped by credential group, includes has_value/masked_value/consumers.
"""
from services.vault import get_credentials_status, get_current_model_selection, get_available_models
username = _get_username()
creds = get_credentials_status(username)
models = get_available_models(username)
model_sel = get_current_model_selection(username)
# Group credentials
groups = {}
for c in creds:
g = c['group']
groups.setdefault(g, []).append(c)
# Stable group ordering
group_order = ['LLM Providers', 'Voice', 'Services', 'Platform', 'Connections', 'Custom']
ordered_groups = []
for g in group_order:
if g in groups:
ordered_groups.append({'name': g, 'credentials': groups.pop(g)})
# Remaining groups (from plugins, etc.)
for g, clist in groups.items():
ordered_groups.append({'name': g, 'credentials': clist})
return jsonify({
'groups': ordered_groups,
'models': models,
'model_selection': model_sel,
})
@vault_bp.route('/api/vault/credentials/<cred_id>', methods=['PUT'])
def update_credential(cred_id):
"""
Set/update a credential value.
Body for single-key: {"value": "sk-..."}
Body for multi-field: {"fields": {"login": "...", "password": "..."}}
Body for model selection: {"primary": "mx/Model", "fallback": "zai/glm-5-turbo"}
"""
from services.vault import (
set_credential, get_catalog_credential, get_merged_catalog,
set_model_selection, ensure_vault
)
username = _get_username()
ensure_vault(username)
data = request.get_json(silent=True) or {}
# Handle model selection update
if cred_id == '_model_selection':
result = set_model_selection(
username,
primary=data.get('primary'),
fallback=data.get('fallback'),
)
# set_model_selection now returns {'ok': bool, 'error': str} — honor a
# real write failure instead of always reporting success (ADMIN-BUG-1).
if isinstance(result, dict) and not result.get('ok'):
return jsonify({
'ok': False,
'message': result.get('error', 'Model selection update failed'),
}), 500
return jsonify({'ok': True, 'message': 'Model selection updated'})
value = data.get('value')
fields = data.get('fields')
# Don't accept masked values as updates
if value and value.startswith('***'):
return jsonify({'ok': True, 'message': 'No change (masked value)'})
if fields:
fields = {k: v for k, v in fields.items() if not str(v).startswith('***')}
if not fields:
return jsonify({'ok': True, 'message': 'No change (masked values)'})
# Validate-before-save for credentials the provider can check (Supabase).
# FAIL = the provider itself rejected the values: not saved unless the user
# explicitly forces it. CANNOT-TELL (provider unreachable) saves, marked
# unverified. Either way the verdict is stored with the values, so the card
# can never read "working" for a key the provider rejected.
validation = None
check = None
if fields:
check = _validate_before_save(username, cred_id, fields)
if check is not None:
if check['verdict'] == 'fail' and not data.get('force'):
return jsonify({
'ok': False,
'saved': False,
'verdict': 'fail',
'message': check['message'],
'checks': check.get('checks', []),
'can_force': True,
}), 422
fields = check['fields']
validation = check['record']
to_restart = set_credential(username, cred_id, value=value, fields=fields,
validation=validation)
extra = {}
if check is not None:
extra = {'verdict': check['verdict'], 'verdict_message': check['message'],
'note': check.get('note', '')}
if to_restart:
containers = sorted(to_restart)
msg = (
f'Saved. Agent will restart in ~6s to pick up the new value '
f'(affects: {", ".join(containers)}).'
)
return jsonify({
'ok': True,
'message': msg,
'restarting': containers,
'eta_seconds': 15,
**extra,
})
return jsonify({
'ok': True,
'message': f'Credential {cred_id} saved (no container restart needed)',
'restarting': [],
'eta_seconds': 0,
**extra,
})
def _validate_before_save(username: str, cred_id: str, fields: dict):
"""Run the credential's validator on (saved values + this save). Returns
None when the credential has no validator, else {verdict, message, checks,
fields (possibly normalized), record, note}."""
from services.vault import (get_catalog_credential, read_vault, _has_validator,
_resolved_fields, validate_credential_fields)
from services.connectors import validation_record
cat = get_catalog_credential(cred_id)
if not cat or not _has_validator(cat):
return None
vault_entry = read_vault(username).get('credentials', {}).get(cred_id, {}) or {}
merged = {**_resolved_fields(cat, vault_entry), **fields}
v = validate_credential_fields(cred_id, merged) or {}
fields = dict(fields)
# Save the URL the check actually used (dashboard link / trailing /rest/v1 removed)
if 'url' in fields and v.get('url') and v['url'] != fields['url']:
fields['url'] = v['url']
merged['url'] = v['url']
return {
'verdict': v.get('verdict', 'cannot_tell'),
'message': v.get('message', ''),
'checks': [{'field': c.get('field'), 'verdict': c.get('verdict'), 'status': c.get('status'),
'message': c.get('message')} for c in v.get('checks', [])],
'note': v.get('note', ''),
'fields': fields,
'record': validation_record(v, merged),
}
@vault_bp.route('/api/vault/credentials', methods=['POST'])
def add_credential():
"""
Add a custom credential not in any catalog.
Body: {"name": "My API", "env_var": "MY_API_KEY", "value": "...", "group": "Custom"}
"""
from services.vault import add_custom_credential, ensure_vault
username = _get_username()
ensure_vault(username)
data = request.get_json(silent=True) or {}
name = data.get('name', '').strip()
env_var = data.get('env_var', '').strip()
value = data.get('value', '').strip()
group = data.get('group', 'Custom')
if not name or not env_var:
return jsonify({'error': 'name and env_var are required'}), 400
cred_id = add_custom_credential(username, name, env_var, value, group)
return jsonify({'ok': True, 'id': cred_id, 'message': f'Custom credential "{name}" added'})
@vault_bp.route('/api/vault/credentials/<cred_id>', methods=['DELETE'])
def remove_credential(cred_id):
"""Delete a custom credential. Only works for user-added custom credentials."""
from services.vault import delete_credential, read_vault
username = _get_username()
vault = read_vault(username)
entry = vault.get('credentials', {}).get(cred_id, {})
if not entry.get('custom'):
return jsonify({'error': 'Can only delete custom credentials'}), 400
delete_credential(username, cred_id)
return jsonify({'ok': True, 'message': f'Credential {cred_id} removed'})
# ---------------------------------------------------------------------------
# Credential testing
# ---------------------------------------------------------------------------
@vault_bp.route('/api/vault/credentials/<cred_id>/test', methods=['POST'])
def test_credential_endpoint(cred_id):
"""Test a credential by hitting the provider's test endpoint."""
from services.vault import test_credential
username = _get_username()
result = test_credential(username, cred_id)
return jsonify(result)
# ---------------------------------------------------------------------------
# OAuth flows
# ---------------------------------------------------------------------------
@vault_bp.route('/api/vault/oauth/<cred_id>/connect', methods=['GET'])
def oauth_connect(cred_id):
"""
Start an OAuth flow. Returns the authorization URL to redirect the user to.
"""
from services.vault import build_oauth_url
username = _get_username()
domain = os.getenv('DOMAIN', 'localhost')
url = build_oauth_url(cred_id, username, domain)
if not url:
return jsonify({'error': 'OAuth not configured for this credential'}), 400
return jsonify({'url': url})
@vault_bp.route('/api/vault/oauth/callback/<provider>', methods=['GET'])
def oauth_callback(provider):
"""
OAuth callback — exchange auth code for tokens.
This endpoint is hit by the OAuth provider after user consent.
Returns HTML that closes the popup and notifies the parent window.
"""
from services.vault import exchange_oauth_code, consume_oauth_nonce
code = request.args.get('code', '')
state_raw = request.args.get('state', '{}')
error = request.args.get('error', '')
if error:
return _oauth_callback_html(False, f'OAuth error: {error}')
if not code:
return _oauth_callback_html(False, 'No authorization code received')
try:
state = json.loads(state_raw)
except json.JSONDecodeError:
state = {}
cred_id = state.get('cred_id', provider)
username = state.get('username', '')
if not username:
return _oauth_callback_html(False, 'Missing username in state')
# F-6 (2026-07-15): verify the single-use CSRF state nonce BEFORE exchanging the code.
# Without this an attacker could forge this callback and connect their own account
# into the victim's vault (OAuth login/connection CSRF).
nonce_entry = consume_oauth_nonce(username, cred_id, state.get('nonce', ''))
if nonce_entry is None:
return _oauth_callback_html(False, 'Invalid or expired OAuth state (CSRF check failed) — please retry the connection.')
# PKCE: the verifier was stored with the nonce at authorize time and never
# left this tenant, so a code delivered anywhere else is useless.
result = exchange_oauth_code(cred_id, code, username,
code_verifier=nonce_entry.get('cv', ''))
if result.get('ok'):
return _oauth_callback_html(True, result.get('account', 'Connected'))
else:
return _oauth_callback_html(False, result.get('error', 'Unknown error'))
@vault_bp.route('/api/vault/oauth/relay/<provider>', methods=['GET'])
def oauth_relay(provider):
"""Single platform callback for providers that accept one (or a few)
redirect URIs. Forwards the browser, with code + state untouched, to the
tenant named in state — only if that tenant's domain is on the
OAUTH_RELAY_ALLOWED_DOMAINS list. Public (like the callback): the provider
redirects here without a Clerk session. It holds no secret, reads no vault
and exchanges nothing; the tenant callback still checks its own single-use
nonce (F-6) and PKCE verifier.
"""
from services.connectors import relay_target
target, why = relay_target(provider, request.args)
if not target:
body, _status, headers = _oauth_callback_html(False, why)
return body, 400, headers
resp = redirect(target, code=302)
resp.headers['Cache-Control'] = 'no-store'
resp.headers['Referrer-Policy'] = 'no-referrer'
return resp
# ---------------------------------------------------------------------------
# One-click connectors
# ---------------------------------------------------------------------------
@vault_bp.route('/api/vault/connectors', methods=['GET'])
def list_connectors():
"""Card data for the Connectors view: every OAuth service, its connected
account, and — for connectors that fill another credential (Supabase) —
that credential's current verdict."""
from services.vault import get_credentials_status
from services.connectors import relay_base_url
username = _get_username()
creds = get_credentials_status(username)
by_id = {c['id']: c for c in creds}
out = []
for c in creds:
if c.get('type') != 'oauth2':
continue
item = {
'id': c['id'],
'name': c.get('name', c['id']),
'description': c.get('description', ''),
'unlocks': c.get('unlocks', '') or c.get('description', ''),
'icon': c.get('icon', ''),
'provider': (c.get('oauth') or {}).get('provider', ''),
'connected': bool(c.get('has_value')),
'account': c.get('masked_value', '') if c.get('has_value') else '',
'account_detail': c.get('account_detail'),
'platform_configured': bool(c.get('platform_configured')),
'fills': c.get('fills', ''),
}
target = by_id.get(c.get('fills') or '')
if target:
item['target'] = {
'id': target['id'],
'name': target.get('name', target['id']),
'has_value': bool(target.get('has_value')),
'validation': target.get('validation'),
'oauth_project': target.get('oauth_project'),
}
out.append(item)
# Connected first, then available, then not yet enabled; catalog order within.
out.sort(key=lambda x: (0 if x['connected'] else 1 if x['platform_configured'] else 2))
relay = relay_base_url()
from urllib.parse import urlparse
relay_origin = ''
if relay:
p = urlparse(relay)
relay_origin = f'{p.scheme}://{p.netloc}'
return jsonify({'connectors': out, 'relay_origin': relay_origin})
def _connector_error(exc):
return jsonify({'ok': False, 'code': exc.code, 'message': exc.message}), exc.status
@vault_bp.route('/api/vault/connectors/supabase/projects', methods=['GET'])
def supabase_projects():
"""Organizations + projects the Supabase connection can see, and whether
the currently saved Project URL is one of them."""
from services.connectors import ConnectorError, supabase_list_projects
try:
return jsonify(supabase_list_projects(_get_username()))
except ConnectorError as exc:
return _connector_error(exc)
@vault_bp.route('/api/vault/connectors/supabase/select', methods=['POST'])
def supabase_select():
"""Body {"ref": "<project ref>"}: fill the supabase credential (URL,
anon key, service role key) from that project via the Management API."""
from services.connectors import ConnectorError, supabase_select_project
data = request.get_json(silent=True) or {}
try:
result = supabase_select_project(_get_username(), data.get('ref', ''))
except ConnectorError as exc:
return _connector_error(exc)
result['eta_seconds'] = 15 if result.get('restarting') else 0
return jsonify(result)
@vault_bp.route('/api/vault/oauth/<cred_id>/disconnect', methods=['POST'])
def oauth_disconnect(cred_id):
"""Disconnect an OAuth connection."""
from services.vault import disconnect_oauth
username = _get_username()
disconnect_oauth(username, cred_id)
return jsonify({'ok': True, 'message': f'{cred_id} disconnected'})
@vault_bp.route('/api/vault/oauth/<cred_id>/status', methods=['GET'])
def oauth_status(cred_id):
"""Get the status of an OAuth connection."""
from services.vault import get_oauth_status
username = _get_username()
status = get_oauth_status(username, cred_id)
return jsonify(status)
# ---------------------------------------------------------------------------
# Sync
# ---------------------------------------------------------------------------
@vault_bp.route('/api/vault/sync', methods=['POST'])
def sync_all_endpoint():
"""Re-sync ALL credentials to ALL consumers."""
from services.vault import sync_all
username = _get_username()
sync_all(username)
return jsonify({'ok': True, 'message': 'All credentials synced'})
@vault_bp.route('/api/vault/sync/<cred_id>', methods=['POST'])
def sync_one_endpoint(cred_id):
"""Re-sync one credential to its consumers."""
from services.vault import sync_credential
username = _get_username()
sync_credential(username, cred_id)
return jsonify({'ok': True, 'message': f'{cred_id} synced'})
# ---------------------------------------------------------------------------
# Phase 1.5 Cycle C: Platform Setup
#
# These endpoints power the admin "Platform Setup" page where the operator
# of a JamBot instance configures OAuth provider credentials (Google,
# Facebook, etc.) once for all clients on the instance.
#
# The page is NOT role-gated — any admin can see it. The credentials live
# in /mnt/system/base/.platform-oauth.env (chmod 640 mike:openvoiceui).
# Saving via this endpoint writes the file and recreates the openvoiceui
# container so the new env vars are loaded.
# ---------------------------------------------------------------------------
# Provider definitions — stable identifiers that group catalog credentials.
# Each provider lists the env vars it needs and the catalog credentials it
# enables. This is the source of truth for the Platform Setup page.
_PLATFORM_PROVIDERS = [
{
'id': 'google',
'name': 'Google',
'description': 'Enables Google Business Profile, Google Analytics, Google Search Console, Gmail and other Google APIs.',
'console_url': 'https://console.cloud.google.com/apis/credentials',
'console_create_url': 'https://console.cloud.google.com/projectcreate',
'docs_url': 'https://developers.google.com/identity/protocols/oauth2',
'env_vars': {
'client_id': 'GOOGLE_OAUTH_CLIENT_ID',
'client_secret': 'GOOGLE_OAUTH_CLIENT_SECRET',
},
'enables_credentials': ['google_business', 'google_analytics', 'google_search_console'],
'setup_steps': [
'Create a Google Cloud project (free tier is fine).',
'Enable the APIs you want clients to use: Business Profile API, Analytics Data API, Search Console API.',
'Configure the OAuth consent screen (External, fill in app name + support email).',
'Create OAuth 2.0 Client ID → Web application.',
'Add the redirect URIs from this page to "Authorized redirect URIs".',
'Copy the Client ID and Client Secret below and click Save.',
],
},
{
'id': 'facebook',
'name': 'Facebook / Meta',
'description': 'Enables Facebook Pages and Instagram Business posting.',
'console_url': 'https://developers.facebook.com/apps',
'console_create_url': 'https://developers.facebook.com/apps/create/',
'docs_url': 'https://developers.facebook.com/docs/facebook-login/',
'env_vars': {
'client_id': 'FACEBOOK_OAUTH_APP_ID',
'client_secret': 'FACEBOOK_OAUTH_APP_SECRET',
},
'enables_credentials': ['facebook', 'instagram'],
'setup_steps': [
'Go to developers.facebook.com → Create App → Business type.',
'Add Products: Facebook Login, Instagram Basic Display.',
'Settings → Basic → copy App ID and App Secret.',
'Facebook Login → Settings → paste redirect URIs from this page into "Valid OAuth Redirect URIs".',
'Save the credentials below.',
],
},
{
'id': 'intuit',
'name': 'QuickBooks (Intuit)',
'description': 'Enables QuickBooks Online — invoices, payments, customers, accounting.',
'console_url': 'https://developer.intuit.com/app/developer/dashboard',
'console_create_url': 'https://developer.intuit.com/app/developer/dashboard',
'docs_url': 'https://developer.intuit.com/app/developer/qbo/docs/develop/authentication-and-authorization',
'env_vars': {
'client_id': 'INTUIT_OAUTH_CLIENT_ID',
'client_secret': 'INTUIT_OAUTH_CLIENT_SECRET',
},
'enables_credentials': ['quickbooks'],
'setup_steps': [
'Sign in at developer.intuit.com and create a new App.',
'Choose APIs your clients need (Accounting API at minimum).',
'Keys & OAuth tab → copy production Client ID and Client Secret.',
'Add the redirect URIs from this page to "Redirect URIs".',
'Save the credentials below.',
],
},
{
'id': 'linkedin',
'name': 'LinkedIn',
'description': 'Enables posting to LinkedIn pages and company updates.',
'console_url': 'https://www.linkedin.com/developers/apps',
'console_create_url': 'https://www.linkedin.com/developers/apps/new',
'docs_url': 'https://learn.microsoft.com/en-us/linkedin/shared/authentication/authorization-code-flow',
'env_vars': {
'client_id': 'LINKEDIN_OAUTH_CLIENT_ID',
'client_secret': 'LINKEDIN_OAUTH_CLIENT_SECRET',
},
'enables_credentials': [], # No catalog entry yet
'setup_steps': [
'Sign in to LinkedIn Developer Portal and create an app.',
'Fill in company info, logo, and terms-of-service URL.',
'Auth tab → copy Client ID and Client Secret.',
'Add the redirect URIs from this page.',
'Products tab → request Sign In with LinkedIn + Share on LinkedIn.',
],
},
# ── One-click connectors (config/connectors-catalog.json) ──────────────
{
'id': 'supabase',
'name': 'Supabase',
'description': 'Enables one-click "Connect Supabase": clients pick the project their app uses and the project URL and keys fill in automatically. No pasting.',
'console_url': 'https://supabase.com/dashboard/org/_/apps',
'console_create_url': 'https://supabase.com/dashboard/org/_/apps',
'docs_url': 'https://supabase.com/docs/guides/integrations/build-a-supabase-oauth-integration',
'env_vars': {
'client_id': 'SUPABASE_OAUTH_CLIENT_ID',
'client_secret': 'SUPABASE_OAUTH_CLIENT_SECRET',
},
'enables_credentials': ['supabase_connect'],
'scopes_note': 'Scopes are set on the app in Supabase, not sent in the URL. Tick exactly: Organizations read, Projects read, Secrets read.',
'setup_steps': [
'In the Supabase dashboard, open your organization settings, then OAuth Apps, then Add application.',
'Name it, set the website URL, and add the callback URL(s) from this page. Supabase accepts several; all must be https.',
'Scopes: Organizations read, Projects read, Secrets read. Nothing else is needed.',
'Confirm, then copy the Client ID and Client Secret below and click Save.',
],
},
{
'id': 'github',
'name': 'GitHub',
'description': 'Enables one-click "Connect GitHub" so a builder can read and push code in the client\'s repositories.',
'console_url': 'https://github.com/settings/developers',
'console_create_url': 'https://github.com/settings/applications/new',
'docs_url': 'https://docs.github.com/en/apps/oauth-apps/building-oauth-apps/authorizing-oauth-apps',
'env_vars': {
'client_id': 'GITHUB_OAUTH_CLIENT_ID',
'client_secret': 'GITHUB_OAUTH_CLIENT_SECRET',
},
'enables_credentials': ['github'],
'scopes_note': 'Requested when the client connects: repo, read:user.',
'setup_steps': [
'GitHub: Settings, Developer settings, OAuth Apps, New OAuth App (an organization can own it instead).',
'Homepage URL: your platform site.',
'Authorization callback URL: the URL from this page. GitHub allows up to 10 per app.',
'Leave "Expire user access tokens" on; tokens are refreshed automatically.',
'Generate a client secret, copy the Client ID and secret below, and click Save.',
],
},
{
'id': 'netlify',
'name': 'Netlify',
'description': 'Enables one-click "Connect Netlify" so a builder can manage the client\'s sites, deploys and environment variables.',
'console_url': 'https://app.netlify.com/user/applications',
'console_create_url': 'https://app.netlify.com/user/applications',
'docs_url': 'https://docs.netlify.com/api-and-cli-guides/api-guides/get-started-with-api/',
'env_vars': {
'client_id': 'NETLIFY_OAUTH_CLIENT_ID',
'client_secret': 'NETLIFY_OAUTH_CLIENT_SECRET',
},
'enables_credentials': ['netlify_connect'],
'scopes_note': 'Netlify has no scopes: a connected token can do anything the user can do on Netlify.',
'setup_steps': [
'Netlify: User settings, Applications, OAuth applications, New OAuth app.',
'Redirect URI: the URL from this page.',
'Copy the Client ID and Secret below and click Save.',
],
},
]
def _relay_providers() -> set:
"""OAuth providers whose catalog entries use the single-callback relay."""
from services.vault import get_platform_catalog
return {
(c.get('oauth') or {}).get('provider', '')
for c in get_platform_catalog()
if c.get('type') == 'oauth2' and (c.get('oauth') or {}).get('redirect') == 'relay'
}
def _redirect_note(provider_id: str) -> str:
from services.connectors import relay_base_url
if provider_id not in _relay_providers():
return ''
if relay_base_url():
return ('One URL for every client. The platform relay forwards each sign-in back to the '
'client that started it (allow-list: OAUTH_RELAY_ALLOWED_DOMAINS).')
return ('OAUTH_RELAY_BASE_URL is not set, so this lists one URL per client. GitHub allows 10 '
'callback URLs and Netlify one: set OAUTH_RELAY_BASE_URL + OAUTH_RELAY_ALLOWED_DOMAINS '
'in .platform-oauth.env to register a single URL instead.')
def _platform_oauth_env_path():
"""Return the path to .platform-oauth.env using the same resolution
logic as services/vault.py."""
from services.vault import _PLATFORM_OAUTH_ENV
return _PLATFORM_OAUTH_ENV
def _read_platform_oauth_env() -> dict:
"""Parse .platform-oauth.env into a dict. Empty/missing → {}."""
path = _platform_oauth_env_path()
result = {}
if not path.is_file():
return result
try:
for raw in path.read_text().splitlines():
line = raw.strip()
if not line or line.startswith('#') or '=' not in line:
continue
k, v = line.split('=', 1)
result[k.strip()] = v.strip().strip('"').strip("'")
except Exception as exc:
logger.error(f"Cannot read platform-oauth env: {exc}")
return result
class PlatformOAuthEnvWriteError(Exception):
"""Raised by _write_platform_oauth_env when the write cannot be made or
cannot be PROVEN to have landed. `code` is the machine-readable reason the
route returns to the caller."""
def __init__(self, code: str, message: str):
super().__init__(message)
self.code = code
self.message = message
def _platform_env_writable(path) -> bool:
"""True iff the process can write `path` (existing file) — os.access
reports False both for mode bits and for a read-only mount (EROFS), which
is exactly the SEC-069 `:ro` case."""
return os.access(path, os.W_OK)
def _write_platform_oauth_env(updates: dict) -> None:
"""Update .platform-oauth.env with the given key/value pairs.
Preserves comments, blank lines, and ordering. Atomic write.
Raises PlatformOAuthEnvWriteError (never returns False) when:
- the file (or, if it must be created, its directory) is read-only
→ code `platform_env_read_only`. SEC-069 mounts the file `:ro` inside
tenant containers; this must surface as an honest error, never as
"saved".
- the write ran but a re-read of the file does not show every requested
value → code `platform_env_write_failed`. The writer's own return
value is not the evidence; the file is.
A no-op write (values already present) is a success.
"""
from services.vault import _update_env_file
path = _platform_oauth_env_path()
if not path.is_file():
if not os.access(path.parent, os.W_OK):
raise PlatformOAuthEnvWriteError(
'platform_env_read_only',
f'{path} does not exist and its directory is read-only here; '
'create it on the host.')
# Create from template if missing
try:
template = path.parent / 'OpenVoiceUI' / '.platform-oauth.env.example'
if template.is_file():
path.write_text(template.read_text())
else:
# Minimal seed
seed = '# Platform OAuth credentials — see .platform-oauth.env.example for full template\n'
seed += '\n'.join(f'{k}=' for k in updates.keys()) + '\n'
path.write_text(seed)
except Exception as exc:
logger.error(f"Cannot create platform-oauth env: {exc}")
raise PlatformOAuthEnvWriteError('platform_env_write_failed',
f'Cannot create {path}: {exc}')
elif not _platform_env_writable(path):
logger.error(f"platform-oauth env is read-only in this container: {path}")
raise PlatformOAuthEnvWriteError(
'platform_env_read_only',
f'{path} is read-only in this container; platform OAuth credentials are set on the host.')
_update_env_file(path, updates)
# Verify from the file, not from the writer's return value (which is also
# False for "nothing changed").
landed = _read_platform_oauth_env() # strips surrounding quotes; compare like-for-like
missing = {k: v for k, v in updates.items()
if landed.get(k, '') != v.strip().strip('"').strip("'")}
if missing:
logger.error(f"platform-oauth env write did not land for keys: {sorted(missing)}")
raise PlatformOAuthEnvWriteError(
'platform_env_write_failed',
f'Write to {path} did not land for: {", ".join(sorted(missing))}.')
def _redirect_uris_for_provider(provider_id: str) -> list[str]:
"""Build the redirect URI list for a provider, one per live client.
Reads /mnt/system/base/registry.json if available, falls back to
listing /mnt/clients/<user> dirs.
Relay providers (catalog oauth.redirect == "relay") with
OAUTH_RELAY_BASE_URL set register exactly one URI: the relay."""
import os
from pathlib import Path
if provider_id in _relay_providers():
from services.connectors import relay_redirect_uri
relayed = relay_redirect_uri(provider_id)
if relayed:
return [relayed]
registry_path = Path('/mnt/system/base/registry.json')
domains = []
if registry_path.is_file():
try:
with open(registry_path) as f:
reg = json.load(f)
for u in reg.get('users', []):
d = u.get('domain', '').strip()
if d:
domains.append(d)
except Exception:
pass
if not domains:
clients_dir = Path('/mnt/clients')
if clients_dir.is_dir():
for child in sorted(clients_dir.iterdir()):
if child.is_dir():
domains.append(f'{child.name}.jam-bot.com')
return [f'https://{d}/api/vault/oauth/callback/{provider_id}' for d in sorted(domains)]
@vault_bp.route('/api/vault/platform-setup', methods=['GET'])
@_platform_admin_only
def platform_setup_get():
"""Return the current Platform Setup state — list of OAuth providers
with configuration status, env vars, and redirect URIs."""
env = _read_platform_oauth_env()
providers_out = []
for p in _PLATFORM_PROVIDERS:
cid_var = p['env_vars']['client_id']
csec_var = p['env_vars']['client_secret']
cid_val = env.get(cid_var, '').strip()
csec_val = env.get(csec_var, '').strip()
configured = bool(cid_val and csec_val)
# Mask secret for display, leave client_id visible
masked_secret = ''
if csec_val:
if len(csec_val) >= 12:
masked_secret = csec_val[:4] + '...' + csec_val[-4:]
else:
masked_secret = '***'
providers_out.append({
'id': p['id'],
'name': p['name'],
'description': p['description'],
'console_url': p['console_url'],
'console_create_url': p.get('console_create_url', ''),
'docs_url': p.get('docs_url', ''),
'env_vars': p['env_vars'],
'enables_credentials': p['enables_credentials'],
'setup_steps': p['setup_steps'],
'configured': configured,
'client_id': cid_val,
'client_secret_masked': masked_secret,
'has_secret': bool(csec_val),
'redirect_uris': _redirect_uris_for_provider(p['id']),
'redirect_note': _redirect_note(p['id']),
'scopes_note': p.get('scopes_note', ''),
})
return jsonify({
'providers': providers_out,
'env_file': str(_platform_oauth_env_path()),
})
@vault_bp.route('/api/vault/platform-setup/<provider_id>', methods=['PUT'])
@_platform_admin_only
def platform_setup_put(provider_id):
"""Update an OAuth provider's client_id and client_secret. Body:
{"client_id": "...", "client_secret": "..."}.
Empty string clears the value.
No container recreate needed: .platform-oauth.env is only consumed by the
openvoiceui Python process via os.environ. We update os.environ directly
in the running process so the new values are live immediately. The file
is also updated so the values persist across container restarts.
"""
import os as _os
p = next((x for x in _PLATFORM_PROVIDERS if x['id'] == provider_id), None)
if not p:
return jsonify({'error': f'Unknown provider {provider_id}'}), 400
data = request.get_json(silent=True) or {}
cid = data.get('client_id', '').strip()
csec = data.get('client_secret', '').strip()
# Reject masked values
if csec.startswith('***') or '...' in csec:
return jsonify({'ok': True, 'message': 'No change (masked secret)'})
updates = {
p['env_vars']['client_id']: cid,
p['env_vars']['client_secret']: csec,
}
try:
_write_platform_oauth_env(updates)
except PlatformOAuthEnvWriteError as exc:
# Loud, coded, and BEFORE os.environ is touched: a read-only file must
# never produce a value that is "live" in this process only.
return jsonify({'error': exc.message, 'code': exc.code}), 500
# Update os.environ in the running process so the new values are live
# immediately. No container restart, no debounce, no rename collision.
for k, v in updates.items():
if v:
_os.environ[k] = v
else:
_os.environ.pop(k, None)
return jsonify({
'ok': True,
'message': f'{p["name"]} OAuth credentials saved and live.',
'provider': provider_id,
'restarting': [],
'eta_seconds': 0,
})
@vault_bp.route('/api/vault/platform-setup/<provider_id>', methods=['DELETE'])
@_platform_admin_only
def platform_setup_delete(provider_id):
"""Clear an OAuth provider's credentials."""
import os as _os
p = next((x for x in _PLATFORM_PROVIDERS if x['id'] == provider_id), None)
if not p:
return jsonify({'error': f'Unknown provider {provider_id}'}), 400
updates = {
p['env_vars']['client_id']: '',
p['env_vars']['client_secret']: '',
}
try:
_write_platform_oauth_env(updates)
except PlatformOAuthEnvWriteError as exc:
return jsonify({'error': exc.message, 'code': exc.code}), 500
# Live update — no restart needed
for k in updates.keys():
_os.environ.pop(k, None)
return jsonify({'ok': True, 'message': f'{p["name"]} OAuth credentials cleared.'})
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _oauth_callback_html(success: bool, message: str) -> str:
"""Return HTML that closes the popup and notifies the parent window.
`message` can carry attacker-controlled text (the provider's `error` query
param on this public, unauthenticated callback) — it MUST be escaped for the
HTML context and JSON-encoded for the inline-script context (reflected XSS).
"""
import html as _html
status = 'success' if success else 'error'
safe_html_message = _html.escape(str(message))
# json.dumps gives a quoted JS string literal; also neutralise "</script>"
# sequences which would otherwise terminate the inline script block.
safe_js_message = json.dumps(str(message)).replace('</', '<\\/')
return f"""<!DOCTYPE html>
<html><head><title>OAuth {status}</title></head>
<body style="background:#1a1a2e;color:#e0e0e0;font-family:system-ui;display:flex;align-items:center;justify-content:center;height:100vh;margin:0">
<div style="text-align:center">
<h2>{'Connected' if success else 'Connection Failed'}</h2>
<p>{safe_html_message}</p>
<p style="color:#888">This window will close automatically...</p>
</div>
<script>
if (window.opener) {{
window.opener.postMessage({{type:'oauth_callback',status:'{status}',message:{safe_js_message}}}, '*');
setTimeout(function(){{ window.close(); }}, 2000);
}} else {{
// Opened as a tab (phones, popup blockers) or cut off from the opener by the
// provider's Cross-Origin-Opener-Policy: go back to the Connectors view.
setTimeout(function(){{ window.close(); location.replace('/admin#connectors'); }}, 2000);
}}
</script>
</body></html>""", 200, {'Content-Type': 'text/html'}