diff --git a/README.md b/README.md index 73b5af4..b2045b9 100644 --- a/README.md +++ b/README.md @@ -126,6 +126,64 @@ Config files are stored in `/etc/cryptolabs-proxy/`: └── ssl/ # SSL certificates ``` +The image ships a bootstrap Nginx configuration. Service lifecycle commands +regenerate the configuration above and require it to be mounted at +`/etc/nginx/nginx.conf` in the running proxy container. This is how optional +Vast Price Manager routing is enabled and removed; the CLI refuses to claim a +route change when the container is still using the bundled configuration. +The bootstrap configuration intentionally contains no Vast Price Manager route. +Before an installer enables or disables VPM, it must first create the registry +configuration, render `/etc/cryptolabs-proxy/nginx.conf`, bind-mount that file +as `/etc/nginx/nginx.conf`, and verify the proxy is using it. The subsequent +`cryptolabs-proxy register` or `unregister` command then validates and reloads +that mounted configuration atomically with the registry update. + +### Existing custom proxy configuration + +Do not use the generic CLI alone to migrate a proxy that already runs a custom +unmounted `nginx.conf`: it regenerates the full configuration. Use the reviewed +`scripts/vpm-custom-config-migrate.py` helper first. Its read-only `plan` mode +derives a migration ID from the source container and config hash. Its explicit +`apply` stores root-only backups, mounts a byte-preserved custom baseline with +only the canonical managed blocks added, and can perform a proxy-only +`rollback`. Custom-config mode then permits only normal VPM +`register`/`unregister`; it rejects changes to every other service. Its managed +Fleet auth deny block remains installed in every HTTP server block after VPM is unregistered, so the +always-running internal VPM session and reauthentication endpoints cannot fall +through a preserved public /auth/ route. Rollback restores the +original stored baseline exactly. + +### Vast Price Manager Fleet sign-in + +### Vast Price Manager prerequisite + +Before a new Vast Price Manager route can be added, the local Vast.ai exporter +must be running and report at least one connected account. Complete setup at +`/vastai/`. Existing VPM installations remain available for maintenance if the +exporter later loses connectivity; they are never automatically disabled. + +The proxy checks the exporter from inside its container using its internal +management token, which must be provisioned by the exporter setup. It records +only whether setup is ready and the count of connected accounts; API keys, +account names, balances, and tokens are not returned. + +When Vast Price Manager runs in Fleet mode, it uses the existing Fleet +`fleet_session` cookie and does not create a second VPM account or login. VPM +introspects the current Fleet session on the Docker network at +`http://cryptolabs-proxy:8081/auth/vast-price-manager/session`; the public +Nginx configuration explicitly returns `404` for that endpoint and its +reauthentication companion, including their trailing-slash variants. + +The internal session response is limited to the current enabled Fleet admin's +username, role, and two purpose-separated HMAC values derived from the signed +cookie: a subject and a CSRF token. It never returns the cookie, password, +password hash, or signing key. The proxy rereads the user record for every +session, reauthentication, and VPM proxy authorization check, so a disabled, +deleted, demoted, or password-change-required account loses VPM access +immediately. Sensitive VPM confirmation calls +`POST /auth/vast-price-manager/reauth` internally with the Fleet password and +the returned CSRF token; it uses the normal Fleet password throttle. + User authentication data is stored in `/data/auth/`: ``` diff --git a/landing-page/index.html b/landing-page/index.html index 243ab70..7052e9d 100644 --- a/landing-page/index.html +++ b/landing-page/index.html @@ -839,6 +839,16 @@

⚙️ System Updates

exporterName: 'vastai', keyPlaceholder: 'Your Vast.ai API Key' }, + 'vast-price-manager': { + displayName: 'Vast Price Manager', + icon: '💰', + description: 'Manage Vast pricing with your existing Fleet login.', + path: '/vast-pricing/', + productUrl: 'https://github.com/cryptolabsza/vast-price-manager', + isCryptoLabs: true, + installNote: 'Install and manage this service from the Vast setup in Server Manager.', + lifecycleManager: 'dc-overview' + }, 'dc-watchdog': { displayName: 'DC Watchdog', icon: '', @@ -900,6 +910,7 @@

⚙️ System Updates

for (const [key, service] of Object.entries(DEFAULT_SERVICES)) { const serviceHealth = health[key] || {}; const isRunning = serviceHealth.running === true; + const isUnconfigured = serviceHealth.state === 'unconfigured'; const isExternalUrl = service.externalUrl ? true : false; // Determine if this service belongs in "Services" or "Available Products" @@ -912,7 +923,7 @@

⚙️ System Updates

if (isRunning || (isExternalUrl && isConfigured && !service.needsApiKey)) { runningCount++; - runningHtml += renderServiceCard(key, service, isExternalUrl ? 'external' : 'running', health); + runningHtml += renderServiceCard(key, service, isExternalUrl ? 'external' : (isUnconfigured ? 'unconfigured' : 'running'), health); } else { availableHtml += renderServiceCard(key, service, isRunning ? 'running' : 'not-installed', health); } @@ -947,6 +958,10 @@

⚙️ System Updates

statusClass = 'running'; statusText = 'Running'; statusDot = 'running'; + } else if (status === 'unconfigured') { + statusClass = 'not-installed'; + statusText = 'Setup Required'; + statusDot = 'not-installed'; } else { statusClass = 'not-installed'; statusText = 'Not Installed'; @@ -956,7 +971,7 @@

⚙️ System Updates

let actions = ''; let installSection = ''; - if (status === 'running' || status === 'external') { + if (status === 'running' || status === 'unconfigured' || status === 'external') { // Active service - show Open button const openUrl = service.externalUrl || service.path; const target = service.externalUrl ? ' target="_blank"' : ''; @@ -987,7 +1002,28 @@

⚙️ System Updates

} } else { // Local service not installed - if (service.deployable) { + const vpmPrerequisite = key === 'vast-price-manager' + ? health['vast-price-manager']?.prerequisite + : null; + if (key === 'vast-price-manager' && !vpmPrerequisite?.configured) { + installSection = ` +
+

Requires Vast.ai setup with a connected account.

+
+ `; + actions = ` + Set up Vast.ai + `; + } else if (key === 'vast-price-manager') { + installSection = ` +
+

Available to install from the Vast.ai setup in Server Manager.

+
+ `; + actions = ` + Learn More + `; + } else if (service.deployable) { // Deployable exporter — check if Prometheus/Grafana are running first const prometheusUp = health['prometheus']?.running === true; const grafanaUp = health['grafana']?.running === true; @@ -1240,32 +1276,20 @@

${service.displayName}

versionText = 'Not running'; } - // --- Update logic --- - // Simple: compare the Docker image TAG against target. - // Target "main" means containers should be on ":latest" tag. - // Target "dev" means containers should be on ":dev" tag. - const targetBranch = settings.target_branch || 'main'; - const currentTag = info.tag || ''; - const expectedTag = targetBranch === 'dev' ? 'dev' : 'latest'; - const tagMatches = currentTag === expectedTag || currentTag === targetBranch; - let updateBtn = ''; - if (!info.running) { - const isExporter = name === 'vastai-exporter' || name === 'runpod-exporter'; - const prometheusUp = serviceHealth['prometheus']?.running === true; - const grafanaUp = serviceHealth['grafana']?.running === true; - - if (isExporter && (!prometheusUp || !grafanaUp)) { - updateBtn = 'Needs dc-overview'; - } else { - updateBtn = ``; - } - } else if (!tagMatches) { - // Tag mismatch (e.g. running :dev but target is main/latest) - updateBtn = ``; + if (info.lifecycle_manager) { + updateBtn = `Managed by Server Manager`; + } else if (info.pinned) { + updateBtn = 'Pinned image · managed separately'; + } else if (!info.running) { + updateBtn = 'Use service setup to start'; + } else if (info.update_available === true) { + updateBtn = ``; updatableCount++; + } else if (info.update_available === false) { + updateBtn = '✓ Downloaded image running'; } else { - updateBtn = '✓ Up to date'; + updateBtn = 'Check for updates first'; } // Self badge for proxy @@ -1311,7 +1335,7 @@

${service.displayName}

updateAllBtn.style.cursor = 'pointer'; } else { updateAllBtn.disabled = true; - updateAllBtn.title = 'All services are up to date or not running'; + updateAllBtn.title = 'Check for updates; pinned and stopped services are preserved'; updateAllBtn.style.opacity = '0.5'; updateAllBtn.style.cursor = 'not-allowed'; } @@ -1336,92 +1360,91 @@

${service.displayName}

}); if (!response.ok) throw new Error('Failed to save'); showUpdateStatus('info', `Target branch set to: ${branch}`); + loadVersions(); } catch (e) { showUpdateStatus('error', `Failed to save branch setting: ${e.message}`); } } - async function checkForUpdates() { - showUpdateStatus('info', '🔍 Checking for updates...'); - const branch = document.getElementById('targetBranch').value; + let activeUpdateJob = null; + let updatePollTimer = null; + const UPDATE_JOB_KEY = 'fleet-update-job'; + + function rememberUpdateJob(id) { try { - const response = await fetch('/api/pull', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ service: 'all', branch }) - }); - const data = await response.json(); - if (data.success) { - showUpdateStatus('success', '✓ Images pulled. Restart containers to apply updates.'); - loadVersions(); - } else { - showUpdateStatus('error', 'Failed to pull updates'); - } - } catch (e) { - showUpdateStatus('error', `Error: ${e.message}`); - } + if (id) localStorage.setItem(UPDATE_JOB_KEY, id); + else localStorage.removeItem(UPDATE_JOB_KEY); + } catch (e) { /* Storage may be disabled; this page can still poll. */ } } - async function updateService(serviceName) { - const branch = document.getElementById('targetBranch').value; - showUpdateStatus('info', `⬆️ Updating ${serviceName}...`); + async function pollUpdateJob(id, failures = 0) { + activeUpdateJob = id; + clearTimeout(updatePollTimer); try { - const response = await fetch('/api/update', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ service: serviceName, branch }) - }); - const data = await response.json(); - if (data.success) { - const result = data.results[serviceName] || {}; - if (result.message === 'self-update-required') { - showUpdateStatus('info', '🔄 Proxy is updating... Page will reload shortly.'); - setTimeout(() => location.reload(), 5000); - } else { - showUpdateStatus('success', `✓ ${serviceName} updated: ${result.message}`); - } - setTimeout(loadVersions, 2000); - } else { - showUpdateStatus('error', `Failed to update ${serviceName}`); + const response = await fetch(`/api/update-status?id=${encodeURIComponent(id)}`); + if (!response.ok) throw new Error('Update status unavailable'); + const job = await response.json(); + if (job.state === 'queued' || job.state === 'running') { + showUpdateStatus('info', 'Update operation in progress. Services restart only after their images are ready.'); + updatePollTimer = setTimeout(() => pollUpdateJob(id), 2000); + return; } + const details = Object.entries(job.results || {}).map(([name, result]) => + `${name}: ${result.message || result.state}`).join(' · '); + showUpdateStatus(job.success === true ? 'success' : 'error', + details || job.message || `Update operation ${job.state}`); + activeUpdateJob = null; + rememberUpdateJob(null); + loadVersions(); + loadServices(); } catch (e) { - showUpdateStatus('error', `Error: ${e.message}`); + if (failures >= 60) { + showUpdateStatus('error', 'Update status is unavailable. Reload to reconnect before starting another update.'); + return; + } + showUpdateStatus('info', 'Reconnecting to update status. The operation may still be running.'); + updatePollTimer = setTimeout(() => pollUpdateJob(id, failures + 1), 3000); } } - async function updateAll() { - if (!confirm('Update all running services to the target branch? This may cause brief downtime. Non-running services will not be started.')) return; - + async function startUpdateJob(action, service) { + if (activeUpdateJob) { + showUpdateStatus('info', 'An update operation is already in progress.'); + return; + } const branch = document.getElementById('targetBranch').value; - showUpdateStatus('info', '⬆️ Updating all services...'); + showUpdateStatus('info', action === 'pull' ? 'Checking images…' : 'Preparing update…'); try { - const response = await fetch('/api/update', { + const response = await fetch(action === 'pull' ? '/api/pull' : '/api/update', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ service: 'all', branch }) + body: JSON.stringify({ service, branch }) }); const data = await response.json(); - if (data.success) { - let msg = 'Updated: '; - for (const [name, result] of Object.entries(data.results || {})) { - msg += `${name}(${result.success ? '✓' : '✗'}) `; - } - if (data.results['cryptolabs-proxy']?.message?.includes('self-update')) { - showUpdateStatus('info', '🔄 Proxy is updating... Page will reload shortly.'); - setTimeout(() => location.reload(), 5000); - } else { - showUpdateStatus('success', msg); - } - setTimeout(loadVersions, 2000); - setTimeout(loadServices, 2000); + if (data.job && (response.ok || response.status === 409)) { + rememberUpdateJob(data.job.id); + await pollUpdateJob(data.job.id); } else { - showUpdateStatus('error', 'Update failed'); + showUpdateStatus('error', data.error || 'Could not start update operation.'); } } catch (e) { - showUpdateStatus('error', `Error: ${e.message}`); + showUpdateStatus('error', 'Could not confirm whether the update started. Refresh status before retrying.'); } } + async function checkForUpdates() { + return startUpdateJob('pull', 'all'); + } + + async function updateService(serviceName) { + return startUpdateJob('update', serviceName); + } + + async function updateAll() { + if (!confirm('Update eligible running services? Services will briefly restart. Pinned images and stopped services are preserved.')) return; + return startUpdateJob('update', 'all'); + } + function showUpdateStatus(type, message) { // Remove any existing status const existingStatus = document.querySelector('.update-status'); @@ -1695,6 +1718,10 @@

${service.displayName}

loadServices(); loadBuildInfo(); loadVersions(); // This calls loadWatchdogStatus() at the end + try { + const pendingUpdate = localStorage.getItem(UPDATE_JOB_KEY); + if (pendingUpdate) pollUpdateJob(pendingUpdate); + } catch (e) { /* Browser storage is optional. */ } // Auto-refresh every 30 seconds setInterval(loadServices, 30000); diff --git a/pyproject.toml b/pyproject.toml index 114e9e0..4fa32da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "cryptolabs-proxy" -version = "1.1.7" +version = "1.1.9" description = "Unified reverse proxy and fleet management landing page for CryptoLabs products" readme = "README.md" license = {text = "MIT"} diff --git a/scripts/health-api.py b/scripts/health-api.py index 222b3b4..ed269c0 100644 --- a/scripts/health-api.py +++ b/scripts/health-api.py @@ -6,18 +6,32 @@ import json import subprocess +import time import http.server import socketserver import threading import os import re +import sys +from urllib.error import HTTPError, URLError from urllib.parse import urlparse, parse_qs +from urllib.request import Request, urlopen from pathlib import Path +sys.path.insert(0, '/app/src') +from cryptolabs_proxy.vpm_prerequisite import get_vpm_prerequisite +from cryptolabs_proxy.updates import ( + SERVICES, UpdateBusy, UpdateError, job_status, + submit_job as submit_update_job, update_status, +) + PORT = 8080 BUILD_INFO_FILE = '/app/BUILD_INFO' SETTINGS_FILE = '/data/auth/update-settings.json' SHARED_CONFIG_FILE = '/data/auth/shared-config.json' +VPM_READY_URL = 'http://vast-price-manager:8088/readyz' +VPM_PREREQUISITE_CACHE_TTL_SECONDS = 30 +_VPM_PREREQUISITE_CACHE = {'expires_at': 0, 'value': None} # Internal Docker network subnet - only allow requests from this range INTERNAL_NETWORK = '172.30.' @@ -34,16 +48,6 @@ INTERNAL_CONFIG_KEYS = {'watchdog_api_key'} NEVER_EXPOSE_KEYS = {'fleet_admin_pass', 'fleet_admin_user', 'auth_secret'} -# Services to check (Docker containers) -SERVICES = { - 'cryptolabs-proxy': {'container': 'cryptolabs-proxy', 'port': 8080, 'image': 'ghcr.io/cryptolabsza/cryptolabs-proxy', 'self': True}, - 'ipmi-monitor': {'container': 'ipmi-monitor', 'port': 5000, 'image': 'ghcr.io/cryptolabsza/ipmi-monitor'}, - 'dc-overview': {'container': 'dc-overview', 'port': 5001, 'image': 'ghcr.io/cryptolabsza/dc-overview'}, - 'grafana': {'container': 'grafana', 'port': 3000, 'image': 'grafana/grafana'}, - 'prometheus': {'container': 'prometheus', 'port': 9090, 'image': 'prom/prometheus'}, - 'vastai-exporter': {'container': 'vastai-exporter', 'port': 8622, 'image': 'ghcr.io/cryptolabsza/vastai-exporter'}, - 'runpod-exporter': {'container': 'runpod-exporter', 'port': 8623, 'image': 'ghcr.io/cryptolabsza/runpod-exporter'}, -} def load_update_settings(): @@ -231,6 +235,67 @@ def check_container_running(container_name): return False +def get_vast_price_manager_allowed_host(): + """Read only VPM's non-secret public-host setting from Docker inspect.""" + try: + result = subprocess.run( + ['docker', 'inspect', '--format', '{{range .Config.Env}}{{println .}}{{end}}', 'vast-price-manager'], + capture_output=True, text=True, timeout=5, + ) + if result.returncode != 0: + return None + for env_line in result.stdout.splitlines(): + if env_line.startswith('VPM_ALLOWED_HOSTS='): + host = env_line.split('=', 1)[1].split(',', 1)[0].strip() + return host or None + except (OSError, subprocess.SubprocessError): + pass + return None + + +def get_vast_price_manager_docker_health(): + """Return Docker's VPM healthcheck result without treating running as healthy.""" + try: + result = subprocess.run( + ['docker', 'inspect', 'vast-price-manager'], capture_output=True, text=True, timeout=5, + ) + if result.returncode != 0: + return None + data = json.loads(result.stdout) + return data[0].get('State', {}).get('Health', {}).get('Status') if data else None + except (OSError, subprocess.SubprocessError, json.JSONDecodeError, IndexError): + return None + + +def get_vast_price_manager_readiness(): + """Report VPM account setup separately from Docker liveness. + + VPM's Docker healthcheck uses ``/healthz``. Its ``/readyz`` endpoint is + surfaced for Fleet status only, so an unconfigured account never creates a + liveness restart loop. + """ + allowed_host = get_vast_price_manager_allowed_host() + if not allowed_host: + return 'unavailable' + try: + request = Request(VPM_READY_URL, headers={'Host': allowed_host}) + with urlopen(request, timeout=2) as response: + return 'ready' if response.status == 200 else 'unavailable' + except HTTPError as error: + return 'unconfigured' if error.code == 503 else 'unavailable' + except (URLError, OSError, TimeoutError): + return 'unavailable' + + +def service_action_error(service_name): + """Return an error when lifecycle is owned outside generic proxy updates.""" + config = SERVICES.get(service_name, {}) + if config.get('update_supported') is False: + display_name = 'Vast Price Manager' if service_name == 'vast-price-manager' else service_name + return f"{display_name} lifecycle is managed by {config['lifecycle_manager']}." + return None + + def get_container_version(container_name): """Get version info for a container from its image, labels, and environment variables.""" try: @@ -317,218 +382,6 @@ def get_container_version(container_name): return None -def pull_image(image_name, tag='latest'): - """Pull a Docker image.""" - full_image = f"{image_name}:{tag}" - try: - result = subprocess.run( - ['docker', 'pull', full_image], - capture_output=True, text=True, timeout=300 - ) - return result.returncode == 0, result.stdout + result.stderr - except Exception as e: - return False, str(e) - - -def get_container_config(container_name): - """Get the configuration of a running container for restart.""" - try: - result = subprocess.run( - ['docker', 'inspect', container_name], - capture_output=True, text=True, timeout=10 - ) - if result.returncode != 0: - return None - - data = json.loads(result.stdout) - if not data: - return None - - container = data[0] - config = container.get('Config', {}) - host_config = container.get('HostConfig', {}) - network_settings = container.get('NetworkSettings', {}) - - # Get network names - networks = list(network_settings.get('Networks', {}).keys()) - - # Get port bindings - port_bindings = host_config.get('PortBindings', {}) - - # Get volume bindings - binds = host_config.get('Binds', []) or [] - - # Get environment variables - env_vars = config.get('Env', []) or [] - - # Get restart policy - restart_policy = host_config.get('RestartPolicy', {}).get('Name', 'unless-stopped') - - return { - 'networks': networks, - 'port_bindings': port_bindings, - 'binds': binds, - 'env_vars': env_vars, - 'restart_policy': restart_policy, - } - except Exception as e: - print(f"Error getting container config for {container_name}: {e}") - return None - - -def restart_container(container_name, image_with_tag, container_config): - """Restart a container with the same configuration but new image.""" - if not container_config: - return False, "No container configuration available" - - try: - # Build docker run command - cmd = ['docker', 'run', '-d', '--name', container_name] - - # Restart policy - cmd.extend(['--restart', container_config.get('restart_policy', 'unless-stopped')]) - - # Networks - docker run only supports one --network, use first non-bridge network - # Additional networks will be connected after container starts - networks = container_config.get('networks', ['cryptolabs']) - non_bridge_networks = [n for n in networks if n and n != 'bridge'] - primary_network = non_bridge_networks[0] if non_bridge_networks else None - additional_networks = non_bridge_networks[1:] if len(non_bridge_networks) > 1 else [] - - if primary_network: - cmd.extend(['--network', primary_network]) - - # Port bindings - for container_port, host_bindings in container_config.get('port_bindings', {}).items(): - if host_bindings: - for binding in host_bindings: - host_port = binding.get('HostPort', '') - host_ip = binding.get('HostIp', '') - if host_ip: - cmd.extend(['-p', f"{host_ip}:{host_port}:{container_port.split('/')[0]}"]) - else: - cmd.extend(['-p', f"{host_port}:{container_port.split('/')[0]}"]) - - # Volume bindings - for bind in container_config.get('binds', []): - cmd.extend(['-v', bind]) - - # Environment variables (filter out build-time vars that we'll update) - for env in container_config.get('env_vars', []): - # Skip PATH and other system vars, keep user-defined ones - if env.startswith('PATH=') or env.startswith('HOME='): - continue - cmd.extend(['-e', env]) - - # Image - cmd.append(image_with_tag) - - result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) - if result.returncode != 0: - return False, f"Failed to start container: {result.stderr}" - - # Connect to additional networks (docker run only supports one --network) - for network in additional_networks: - try: - subprocess.run(['docker', 'network', 'connect', network, container_name], - capture_output=True, timeout=10) - except Exception as e: - print(f"Warning: Failed to connect {container_name} to network {network}: {e}") - - return True, "Container restarted successfully" - except Exception as e: - return False, f"Error restarting container: {e}" - - -def update_container(container_name, service_config, target_branch='main'): - """Update a container to a new image version.""" - image = service_config.get('image', '') - if not image: - return False, "No image configured for service" - - is_self = service_config.get('self', False) - tag = target_branch if target_branch in ['dev', 'main'] else 'latest' - - # Default to latest for all images - if not tag or tag == 'main': - tag = 'latest' - - full_image = f"{image}:{tag}" - - # Pull new image first - success, output = pull_image(image, tag) - if not success: - return False, f"Failed to pull image: {output}" - - if is_self: - # For self-update, we need special handling - # Create a script that will restart the container after we exit - return True, "self-update-required" - - # Get current container configuration before stopping - container_config = get_container_config(container_name) - - # Stop and remove old container - try: - subprocess.run(['docker', 'stop', container_name], capture_output=True, timeout=30) - subprocess.run(['docker', 'rm', container_name], capture_output=True, timeout=10) - except: - pass - - # Restart container with new image - if container_config: - success, msg = restart_container(container_name, full_image, container_config) - if success: - return True, f"Updated to {tag} and restarted" - else: - return False, f"Image pulled but restart failed: {msg}" - else: - # Fallback: container wasn't running or couldn't get config - return True, f"Image pulled. Container needs manual restart (was not running)." - - -def trigger_self_update(target_branch='main'): - """Trigger self-update for the proxy container.""" - # Pull the new image - image = 'ghcr.io/cryptolabsza/cryptolabs-proxy' - tag = target_branch if target_branch in ['dev', 'main'] else 'latest' - - success, output = pull_image(image, tag) - if not success: - return False, f"Failed to pull image: {output}" - - # Create a restart script that runs after the API responds - # This uses docker to restart the container from outside - script = f"""#!/bin/bash -sleep 2 -docker stop cryptolabs-proxy -docker rm cryptolabs-proxy -# The container should be recreated by docker-compose or systemd -# For safety, try to start it using the same command pattern -docker run -d --name cryptolabs-proxy \\ - --restart unless-stopped \\ - --network cryptolabs \\ - -v /var/run/docker.sock:/var/run/docker.sock \\ - -v /data/auth:/data/auth \\ - -p 80:80 -p 443:443 \\ - {image}:{tag} -""" - - # Write script and execute in background - script_path = '/tmp/proxy-update.sh' - try: - with open(script_path, 'w') as f: - f.write(script) - os.chmod(script_path, 0o755) - subprocess.Popen(['/bin/bash', script_path], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - start_new_session=True) - return True, "Self-update initiated. Proxy will restart in a few seconds." - except Exception as e: - return False, f"Failed to initiate self-update: {e}" - - def get_all_service_status(include_versions=False): """Get status of all services.""" status = {} @@ -546,6 +399,22 @@ def get_all_service_status(include_versions=False): 'image': config.get('image', ''), 'self': config.get('self', False), } + + if config.get('lifecycle_manager'): + service_info['lifecycle_manager'] = config['lifecycle_manager'] + service_info['update_supported'] = config.get('update_supported', True) + + if name == 'vast-price-manager': + readiness = get_vast_price_manager_readiness() if running else 'not-installed' + docker_health = get_vast_price_manager_docker_health() if running else None + service_info.update({ + 'healthy': docker_health == 'healthy', + 'docker_health': docker_health, + 'readiness': readiness, + 'configured': readiness == 'ready', + 'state': 'running' if readiness == 'ready' else readiness, + 'prerequisite': get_vpm_prerequisite_for_display(), + }) if include_versions and running: version_info = get_container_version(container) @@ -560,6 +429,18 @@ def get_all_service_status(include_versions=False): return status +def get_vpm_prerequisite_for_display(): + """Cache the sanitized prerequisite only for status-page polling.""" + now = time.monotonic() + cached = _VPM_PREREQUISITE_CACHE + if cached['value'] is not None and now < cached['expires_at']: + return cached['value'] + value = get_vpm_prerequisite() + _VPM_PREREQUISITE_CACHE['value'] = value + _VPM_PREREQUISITE_CACHE['expires_at'] = now + VPM_PREREQUISITE_CACHE_TTL_SECONDS + return value + + def get_all_versions(): """Get version info for all services.""" versions = {} @@ -575,6 +456,9 @@ def get_all_versions(): 'image': config.get('image', ''), 'self': config.get('self', False), } + if config.get('lifecycle_manager'): + version_info['lifecycle_manager'] = config['lifecycle_manager'] + version_info['update_supported'] = config.get('update_supported', True) if running: v = get_container_version(container) @@ -596,6 +480,8 @@ def get_all_versions(): }) versions[name] = version_info + if running and config.get('update_supported') is not False: + version_info.update(update_status(name, settings.get('branch', 'main'))) # Add configured branch versions['_settings'] = { @@ -669,6 +555,12 @@ def do_GET(self): elif path == '/api/update-settings': settings = load_update_settings() self.send_json(settings) + + elif path == '/api/update-status': + try: + self.send_json(job_status(query.get('id', [''])[0])) + except UpdateError as error: + self.send_json({'error': str(error)}, 404) # ---- Internal Config API (fleet services only) ---- # Security: 4 layers of protection @@ -753,70 +645,21 @@ def do_POST(self): else: self.send_json({'error': 'Failed to save settings'}, 500) - elif path == '/api/update': - # Trigger update for one or all services - service = data.get('service', 'all') - target_branch = data.get('branch', load_update_settings().get('branch', 'main')) - - results = {} - - if service == 'all': - # Update all services - for name, config in SERVICES.items(): - if config.get('self'): - # Handle self-update last - continue - tag = 'dev' if target_branch == 'dev' else 'latest' - success, msg = update_container(name, config, tag) - results[name] = {'success': success, 'message': msg} - - # Handle proxy self-update last (if requested) - if 'cryptolabs-proxy' in SERVICES: - success, msg = trigger_self_update(target_branch) - results['cryptolabs-proxy'] = {'success': success, 'message': msg} - - elif service == 'cryptolabs-proxy': - # Self-update - success, msg = trigger_self_update(target_branch) - results[service] = {'success': success, 'message': msg} - - elif service in SERVICES: - config = SERVICES[service] - tag = 'dev' if target_branch == 'dev' else 'latest' - success, msg = update_container(service, config, tag) - results[service] = {'success': success, 'message': msg} - - else: - self.send_json({'error': f'Unknown service: {service}'}, 400) - return - - self.send_json({'success': True, 'results': results}) - - elif path == '/api/pull': - # Just pull images without restarting - service = data.get('service', 'all') - target_branch = data.get('branch', load_update_settings().get('branch', 'main')) - - results = {} - services_to_pull = [service] if service != 'all' else list(SERVICES.keys()) - - for name in services_to_pull: - if name not in SERVICES: - results[name] = {'success': False, 'message': 'Unknown service'} - continue - - config = SERVICES[name] - tag = 'dev' if target_branch == 'dev' else 'latest' - - image = config.get('image', '') - if image: - success, msg = pull_image(image, tag) - results[name] = {'success': success, 'message': msg[:200] if len(msg) > 200 else msg} - else: - results[name] = {'success': False, 'message': 'No image configured'} - - self.send_json({'success': True, 'results': results}) - + elif path in ('/api/update', '/api/pull'): + try: + job = submit_update_job( + data.get('service', 'all'), + data.get('branch', load_update_settings().get('branch', 'main')), + 'pull' if path == '/api/pull' else 'update', + ) + self.send_json({'success': True, 'job': job}, 202) + except UpdateBusy as error: + self.send_json({'success': False, 'error': str(error), 'job': error.job}, 409) + except UpdateError as error: + self.send_json({'success': False, 'error': str(error)}, 400) + except Exception: + self.send_json({'success': False, 'error': 'Updater unavailable; no update confirmed.'}, 503) + # ---- Internal Config API: SET values (always requires token) ---- elif path == '/internal/api/config': if not is_internal_request(self.client_address): diff --git a/scripts/vpm-custom-config-migrate.py b/scripts/vpm-custom-config-migrate.py new file mode 100644 index 0000000..c5a701e --- /dev/null +++ b/scripts/vpm-custom-config-migrate.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +"""Run the reviewed VPM custom-proxy migration helper on its target host.""" + +import argparse +import json +from pathlib import Path +import sys + +from cryptolabs_proxy.migration import PROXY_READY_TIMEOUT, CustomConfigMigrator, MigrationError + + +def parser(): + command = argparse.ArgumentParser(description="VPM custom proxy migration (no default apply)") + command.add_argument("--config-dir", type=Path, default=Path("/etc/cryptolabs-proxy")) + command.add_argument("--backup-root", type=Path, default=Path("/etc/cryptolabs-proxy/migrations")) + command.add_argument("--transport-timeout", type=int, default=15, help="per Docker/HTTP transport timeout in seconds") + command.add_argument( + "--readiness-timeout", + type=int, + default=PROXY_READY_TIMEOUT, + help="Docker health and local auth readiness deadline in seconds", + ) + subcommands = command.add_subparsers(dest="action", required=True) + + plan = subcommands.add_parser("plan", help="read-only sanitized migration plan") + plan.add_argument("--container", default="cryptolabs-proxy") + plan.add_argument("--proxy-image", required=True) + + apply = subcommands.add_parser("apply", help="explicit proxy-only switch with rollback") + apply.add_argument("--container", default="cryptolabs-proxy") + apply.add_argument("--proxy-image", required=True) + apply.add_argument("--migration-id", required=True) + apply.add_argument("--enable-vpm", action="store_true", help="write the canonical VPM registry entry and route") + + rollback = subcommands.add_parser("rollback", help="restore only the named proxy migration") + rollback.add_argument("--container", default="cryptolabs-proxy") + rollback.add_argument("--migration-id", required=True) + return command + + +def main(): + args = parser().parse_args() + migrator = CustomConfigMigrator( + args.config_dir, + args.backup_root, + timeout=args.transport_timeout, + readiness_timeout=args.readiness_timeout, + ) + try: + if args.action == "plan": + print(json.dumps(migrator.plan(args.container, args.proxy_image).sanitized_manifest(), sort_keys=True)) + elif args.action == "apply": + outcome = migrator.apply(args.container, args.proxy_image, args.migration_id, args.enable_vpm) + print(json.dumps({"migration_id": args.migration_id, "state": outcome.state, "detail": outcome.detail})) + if outcome.state == "candidate_retained": + # The candidate is serving, but its fallback did not meet the + # required health contract. Surface that operator action is + # required without letting generic exception cleanup rewrite + # the live managed configuration. + return 2 + else: + outcome = migrator.rollback(args.container, args.migration_id) + print(json.dumps({"migration_id": args.migration_id, "state": outcome.state, "detail": outcome.detail})) + except MigrationError as error: + print(f"migration failed: {error}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/cryptolabs_proxy/__init__.py b/src/cryptolabs_proxy/__init__.py index 8b91ef4..e92123b 100644 --- a/src/cryptolabs_proxy/__init__.py +++ b/src/cryptolabs_proxy/__init__.py @@ -1,6 +1,6 @@ """CryptoLabs Proxy - Unified reverse proxy for CryptoLabs products.""" -__version__ = "1.1.7" +__version__ = "1.1.9" # Export programmatic setup API from .setup import ( diff --git a/src/cryptolabs_proxy/auth.py b/src/cryptolabs_proxy/auth.py index def0707..daf704d 100644 --- a/src/cryptolabs_proxy/auth.py +++ b/src/cryptolabs_proxy/auth.py @@ -1121,6 +1121,55 @@ def decorated(*args, **kwargs): return f(*args, **kwargs) return decorated + def current_vast_price_manager_user(): + """Return the live Fleet administrator behind this signed session. + + Flask has already verified ``fleet_session`` before exposing ``session``. + The database lookup is deliberately repeated for every VPM request so a + role change, disablement, deletion, or forced password change takes + effect immediately instead of trusting values cached in the cookie. + """ + if not session.get('logged_in') or session.get('require_password_change'): + return None, 401 + + username = session.get('username') + if not isinstance(username, str) or not username: + return None, 401 + + user = get_user(username) + if not user or not user.get('enabled', True) or user.get('role') != 'admin': + return None, 403 + if user.get('require_password_change', False): + return None, 401 + return user, 200 + + def vast_price_manager_session_payload(): + """Build the bounded VPM authority response from the signed cookie.""" + user, status = current_vast_price_manager_user() + if not user: + return None, status + + signed_cookie = request.cookies.get(app.config['SESSION_COOKIE_NAME']) + if not signed_cookie: + return None, 401 + + cookie_bytes = signed_cookie.encode('utf-8') + + def derive(purpose: bytes) -> str: + return hmac.new( + AUTH_SECRET_KEY.encode('utf-8'), purpose + cookie_bytes, hashlib.sha256 + ).hexdigest() + + return { + 'authenticated': True, + 'username': user['username'], + 'role': 'admin', + # Purpose separation prevents either stable value from standing in + # for the other while avoiding disclosure of the signed cookie. + 'subject': derive(b'cryptolabs/vpm/subject/v1:'), + 'csrf_token': derive(b'cryptolabs/vpm/csrf/v1:'), + }, 200 + # ========================================================================= # ROUTES # ========================================================================= @@ -1442,6 +1491,69 @@ def get_headers(): return response return '', 401 + + @app.route('/auth/vast-price-manager/authorize') + def authorize_vast_price_manager(): + """Authorize the Fleet administrator-only VPM proxy route. + + This endpoint is used only by Nginx's internal ``auth_request`` + subrequest. It reads the signed-in Flask session and deliberately + ignores incoming role headers, which a client could forge. + """ + _, status = current_vast_price_manager_user() + if status != 200: + return '', status + return '', 204 + + @app.route('/auth/vast-price-manager/session') + def vast_price_manager_session(): + """Expose current Fleet authority only to VPM's Docker-network client.""" + payload, status = vast_price_manager_session_payload() + if not payload: + return '', status + return jsonify(payload) + + @app.route('/auth/vast-price-manager/reauth', methods=['POST']) + def reauthenticate_vast_price_manager(): + """Recheck the current Fleet password for one sensitive VPM action.""" + payload, status = vast_price_manager_session_payload() + if not payload: + return '', status + + data = request.get_json(silent=True) + if not isinstance(data, dict): + return '', 400 + password = data.get('password') + csrf_token = data.get('csrf_token') + if ( + not isinstance(password, str) + or not isinstance(csrf_token, str) + or len(password) > 4096 + or len(csrf_token) != 64 + ): + return '', 400 + if not hmac.compare_digest(csrf_token, payload['csrf_token']): + return '', 403 + + verified = verify_user(payload['username'], password) + if not verified: + # ``verify_user`` records failures and applies the existing Fleet + # lockout policy. Surface a rate limit only once it is observable. + current = get_user(payload['username']) + try: + locked = current and current.get('locked_until') and ( + datetime.fromisoformat(current['locked_until']) > datetime.utcnow() + ) + except (TypeError, ValueError): + locked = False + return '', 429 if locked else 401 + + # Re-read after password verification so a concurrent state change + # cannot turn this endpoint into a stale authority grant. + payload, status = vast_price_manager_session_payload() + if not payload: + return '', status + return jsonify(payload) # API endpoints for programmatic access @app.route('/auth/api/users', methods=['GET']) diff --git a/src/cryptolabs_proxy/cli.py b/src/cryptolabs_proxy/cli.py index 23a47a7..5d16a2c 100644 --- a/src/cryptolabs_proxy/cli.py +++ b/src/cryptolabs_proxy/cli.py @@ -1,6 +1,9 @@ """CryptoLabs Proxy CLI - Setup and manage the unified reverse proxy.""" import click +import copy +from contextlib import contextmanager +import fcntl import os import subprocess import sys @@ -14,10 +17,23 @@ from . import __version__ from .config import CONFIG_DIR, get_jinja_env, generate_nginx_config, generate_docker_compose -from .services import ServiceRegistry +from .custom_config import ( + CustomConfigError, + custom_config_settings, + ensure_vpm_only_change, + load_verified_baseline, + render_managed_vpm_config, +) +from .services import DEFAULT_SERVICES, ServiceRegistry +from .vpm_prerequisite import get_vpm_prerequisite console = Console() +# A registry mutation can invoke the proxy container, but each command is +# bounded so a stalled Docker daemon cannot indefinitely block other lifecycle +# operations waiting for the registry lock. +NGINX_COMMAND_TIMEOUT = 15 + custom_style = questionary.Style([ ('qmark', 'fg:cyan bold'), ('question', 'bold'), @@ -90,6 +106,140 @@ def get_local_ip() -> str: return "127.0.0.1" +def validate_nginx_config(): + """Validate the mounted proxy configuration before asking Nginx to reload.""" + try: + result = subprocess.run( + ["docker", "exec", "cryptolabs-proxy", "nginx", "-t"], + capture_output=True, + text=True, + timeout=NGINX_COMMAND_TIMEOUT, + ) + return result.returncode == 0, result.stderr or result.stdout + except (OSError, subprocess.SubprocessError) as error: + return False, str(error) + + +def reload_nginx_config(): + """Reload Nginx and return its output to the caller.""" + try: + result = subprocess.run( + ["docker", "exec", "cryptolabs-proxy", "nginx", "-s", "reload"], + capture_output=True, + text=True, + timeout=NGINX_COMMAND_TIMEOUT, + ) + return result.returncode == 0, result.stderr or result.stdout + except (OSError, subprocess.SubprocessError) as error: + return False, str(error) + + +def proxy_uses_generated_config(config_path: Path) -> bool: + """Confirm the active container sees exactly the generated host config. + + The image's bundled nginx.conf is a bootstrap configuration. Lifecycle + registration is supported only for deployments that mount the generated + config into /etc/nginx/nginx.conf; otherwise a reload would not change the + active routes. + """ + try: + result = subprocess.run( + ["docker", "exec", "cryptolabs-proxy", "cat", "/etc/nginx/nginx.conf"], + capture_output=True, + text=True, + timeout=NGINX_COMMAND_TIMEOUT, + ) + return result.returncode == 0 and config_path.read_text() == result.stdout + except (OSError, subprocess.SubprocessError): + return False + + +@contextmanager +def registry_lock(config_dir: Path): + """Serialize lifecycle changes to one generated proxy configuration. + + The lock deliberately covers registry loading through validation, reload, + and rollback. Docker calls made during that interval use a timeout so a + competing install or disable action is never held behind an unbounded + command. + """ + config_dir.mkdir(parents=True, exist_ok=True) + lock_path = config_dir / ".registry.lock" + with lock_path.open("a+") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +def _snapshot_files(paths): + return {path: path.read_bytes() if path.exists() else None for path in paths} + + +def _restore_files(snapshot): + for path, content in snapshot.items(): + if content is None: + path.unlink(missing_ok=True) + else: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + + +def apply_registry_change(registry: ServiceRegistry, services: dict): + """Persist, validate, and reload a registry mutation with file rollback. + + The route remains active until Nginx successfully reloads. If validation or + reload fails, registry/settings/config files are restored; a failed reload + also triggers an old-config reload attempt to make the active state match + the restored files. + """ + nginx_path = registry.config_dir / "nginx.conf" + snapshot = _snapshot_files([registry.services_file, registry.config_file, nginx_path]) + try: + if custom_config_settings(registry.config): + ensure_vpm_only_change(registry.services, services) + baseline = load_verified_baseline(registry.config) + rendered_config = render_managed_vpm_config( + baseline, + enabled="vast-price-manager" in services, + ) + else: + rendered_config = None + registry.services = services + registry.save() + if rendered_config is None: + generate_nginx_config( + registry.config_dir, + domain=registry.config.get("domain", get_local_ip()), + letsencrypt=registry.config.get("letsencrypt", False), + services=registry.services, + ) + else: + nginx_path.write_bytes(rendered_config) + if not proxy_uses_generated_config(nginx_path): + raise RuntimeError( + "Proxy is not using the generated /etc/cryptolabs-proxy/nginx.conf; " + "route change was not applied." + ) + valid, validation_output = validate_nginx_config() + if not valid: + raise RuntimeError(f"Nginx configuration validation failed: {validation_output.strip()}") + reloaded, reload_output = reload_nginx_config() + if reloaded: + return True, "" + + _restore_files(snapshot) + rollback_ok, rollback_output = reload_nginx_config() + detail = f"Nginx reload failed: {reload_output.strip()}" + if not rollback_ok: + detail += f"; rollback reload also failed: {rollback_output.strip()}" + return False, detail + except (OSError, RuntimeError, CustomConfigError) as error: + _restore_files(snapshot) + return False, str(error) + + # Docker network subnet for UFW rules DOCKER_NETWORK_SUBNET = "172.30.0.0/16" @@ -385,36 +535,63 @@ def setup(): def register(service_name, container_name, path, port, display_name, icon, description): """Register a service with the proxy.""" check_root() - - registry = ServiceRegistry(CONFIG_DIR) - - if path is None: - path = f"/{service_name}/" - - registry.add_service( - name=service_name, - container_name=container_name, - path=path, - port=port, - display_name=display_name or service_name.replace("-", " ").title(), - icon=icon, - description=description - ) - - # Regenerate nginx config - generate_nginx_config( - CONFIG_DIR, - domain=registry.config.get("domain", get_local_ip()), - letsencrypt=registry.config.get("letsencrypt", False), - services=registry.services - ) - - # Reload nginx - subprocess.run(["docker", "exec", "cryptolabs-proxy", "nginx", "-s", "reload"], capture_output=True) - + with registry_lock(CONFIG_DIR): + registry = ServiceRegistry(CONFIG_DIR) + + # Only the first registration creates VPM access. Existing instances + # remain manageable when the exporter later becomes unavailable. + if service_name == "vast-price-manager" and service_name not in registry.services: + prerequisite = get_vpm_prerequisite() + if not prerequisite["configured"]: + raise click.ClickException( + "Requires Vast.ai setup with at least one connected account. " + "Open /vastai/ to finish setup." + ) + + if path is None: + path = f"/{service_name}/" + services = copy.deepcopy(registry.services) + service = copy.deepcopy(services.get(service_name, {})) + defaults = DEFAULT_SERVICES.get(service_name, {}) + service.update({ + "container_name": container_name, + "path": path, + "port": port, + "display_name": display_name or defaults.get("display_name") or service_name.replace("-", " ").title(), + "icon": icon if icon != "🔧" else defaults.get("icon", icon), + "description": description or defaults.get("description", ""), + }) + # VPM registration is intentionally narrow: the route fragment has one + # fixed container/path/port contract, so retain every canonical field + # even when the installer invokes the generic CLI command. + if service_name == "vast-price-manager": + service = copy.deepcopy(defaults) + services[service_name] = service + applied, message = apply_registry_change(registry, services) + if not applied: + raise click.ClickException(message) console.print(f"[green]✓[/green] Registered {service_name} at {path}") +@main.command() +@click.argument("service_name") +def unregister(service_name): + """Remove a service route while preserving unrelated proxy settings.""" + check_root() + with registry_lock(CONFIG_DIR): + registry = ServiceRegistry(CONFIG_DIR) + if service_name not in registry.services: + console.print(f"[yellow]•[/yellow] {service_name} is not registered") + return + + services = copy.deepcopy(registry.services) + del services[service_name] + applied, message = apply_registry_change(registry, services) + if not applied: + raise click.ClickException(message) + console.print(f"[green]✓[/green] Unregistered {service_name}") + + @main.command() def status(): """Show status of all services.""" diff --git a/src/cryptolabs_proxy/config.py b/src/cryptolabs_proxy/config.py index f95e1f5..9447f04 100644 --- a/src/cryptolabs_proxy/config.py +++ b/src/cryptolabs_proxy/config.py @@ -31,6 +31,16 @@ def generate_nginx_config(config_dir: Path, domain: str, letsencrypt: bool = Fal (config_dir / "nginx.conf").write_text(content) +def render_vpm_fragment() -> str: + """Render the exact VPM route shared by generated and custom configs.""" + return get_jinja_env().get_template("vast-price-manager.conf.j2").render() + + +def render_vpm_internal_auth_blocks() -> str: + """Render public-Nginx blocks for Docker-only VPM Fleet auth endpoints.""" + return get_jinja_env().get_template("vast-price-manager-auth-blocks.conf.j2").render() + + def generate_docker_compose(config_dir: Path, domain: str = None, use_letsencrypt: bool = False): """Generate docker-compose.yml from template.""" env = get_jinja_env() diff --git a/src/cryptolabs_proxy/custom_config.py b/src/cryptolabs_proxy/custom_config.py new file mode 100644 index 0000000..271313e --- /dev/null +++ b/src/cryptolabs_proxy/custom_config.py @@ -0,0 +1,203 @@ +"""VPM-only management for a preserved, custom Nginx baseline.""" + +from __future__ import annotations + +import hashlib +import re +from pathlib import Path + +from .config import render_vpm_fragment, render_vpm_internal_auth_blocks + + +CUSTOM_CONFIG_KEY = "custom_config" +CUSTOM_CONFIG_MODE = "vpm-managed" +MANAGED_BEGIN = b"# BEGIN CRYPTOLABS MANAGED VPM" +MANAGED_END = b"# END CRYPTOLABS MANAGED VPM" +AUTH_DENIES_BEGIN = b"# BEGIN CRYPTOLABS MANAGED INTERNAL AUTH DENIES" +AUTH_DENIES_END = b"# END CRYPTOLABS MANAGED INTERNAL AUTH DENIES" +_ANCHOR = re.compile(rb"^[ \t]*location[ \t]+@login_redirect[ \t]*\{") +_HTTP = re.compile(rb"^[ \t]*http[ \t]*\{") +_SERVER = re.compile(rb"^[ \t]*server[ \t]*\{") + + +class CustomConfigError(RuntimeError): + """The preserved baseline cannot safely receive a VPM-only mutation.""" + + +def configure_custom_config_mode(config: dict, baseline_path: Path) -> None: + """Store the immutable baseline identity in existing registry settings.""" + baseline = baseline_path.read_bytes() + config[CUSTOM_CONFIG_KEY] = { + "mode": CUSTOM_CONFIG_MODE, + "baseline_path": str(baseline_path), + "baseline_sha256": hashlib.sha256(baseline).hexdigest(), + } + + +def custom_config_settings(config: dict) -> dict | None: + settings = config.get(CUSTOM_CONFIG_KEY) + if settings and settings.get("mode") == CUSTOM_CONFIG_MODE: + return settings + return None + + +def load_verified_baseline(config: dict) -> bytes: + """Read the root-owned baseline only when its configured hash still matches.""" + settings = custom_config_settings(config) + if not settings: + raise CustomConfigError("custom-config mode is not configured") + path = Path(settings["baseline_path"]) + baseline = path.read_bytes() + actual = hashlib.sha256(baseline).hexdigest() + if actual != settings.get("baseline_sha256"): + raise CustomConfigError("custom-config baseline SHA-256 changed; refusing mutation") + return baseline + + +def _strip_comments_and_strings(line: bytes) -> bytes: + """Keep braces structural while ignoring simple quoted Nginx directive text.""" + result = bytearray() + quote = None + escaped = False + for char in line: + if quote: + if escaped: + escaped = False + elif char == ord("\\"): + escaped = True + elif char == quote: + quote = None + continue + if char in (ord("'"), ord('"')): + quote = char + elif char == ord("#"): + break + else: + result.append(char) + return bytes(result) + + +def _structural_brace_events(line: bytes) -> list[tuple[int, int]]: + """Return brace positions that are outside quoted strings and comments.""" + events = [] + quote = None + escaped = False + for index, char in enumerate(line): + if quote: + if escaped: + escaped = False + elif char == ord("\\"): + escaped = True + elif char == quote: + quote = None + continue + if char in (ord("'"), ord('"')): + quote = char + elif char == ord("#"): + break + elif char in (ord("{"), ord("}")): + events.append((index, char)) + return events + + +def _anchor_end_offset(baseline: bytes) -> int: + lines = baseline.splitlines(keepends=True) + anchors = [index for index, line in enumerate(lines) if _ANCHOR.match(line)] + if len(anchors) != 1: + raise CustomConfigError("custom-config baseline must contain exactly one location @login_redirect anchor") + + anchor = anchors[0] + depth = 0 + for line in lines[:anchor]: + structural = _strip_comments_and_strings(line) + depth += structural.count(b"{") - structural.count(b"}") + target_depth = depth + offset = sum(len(line) for line in lines[:anchor]) + for line in lines[anchor:]: + structural = _strip_comments_and_strings(line) + depth += structural.count(b"{") - structural.count(b"}") + offset += len(line) + if depth == target_depth: + return offset + raise CustomConfigError("custom-config login redirect anchor has unbalanced Nginx braces") + + +def _http_server_end_offsets(baseline: bytes) -> list[int]: + """Locate every HTTP server closing brace without parsing quoted braces.""" + depth = 0 + http_depth = None + server_parents = [] + offsets = [] + offset = 0 + + for line in baseline.splitlines(keepends=True): + events = _structural_brace_events(line) + http_open = None + server_open = None + if http_depth is None and _HTTP.match(line): + http_open = next((index for index, char in events if char == ord("{")), None) + elif http_depth is not None and depth == http_depth + 1 and _SERVER.match(line): + server_open = next((index for index, char in events if char == ord("{")), None) + + for index, char in events: + if char == ord("{"): + if index == http_open: + http_depth = depth + if index == server_open: + server_parents.append(depth) + depth += 1 + else: + depth -= 1 + if server_parents and depth == server_parents[-1]: + server_parents.pop() + offsets.append(offset + index) + if http_depth is not None and depth == http_depth: + http_depth = None + offset += len(line) + + if http_depth is not None or server_parents: + raise CustomConfigError("custom-config HTTP server blocks have unbalanced Nginx braces") + if not offsets: + raise CustomConfigError("custom-config baseline must contain at least one HTTP server block") + return offsets + + +def render_managed_vpm_config(baseline: bytes, enabled: bool) -> bytes: + """Splice auth denies and, when enabled, the VPM route after the anchor. + + The auth server is always installed, so its VPM-only authority endpoints + must remain private even after VPM itself is unregistered. The source + baseline is never changed: each render starts from its recorded bytes, + which keeps a migration rollback byte-exact. + """ + if any(marker in baseline for marker in ( + MANAGED_BEGIN, MANAGED_END, AUTH_DENIES_BEGIN, AUTH_DENIES_END, + )): + raise CustomConfigError("custom-config baseline already contains a managed VPM delimiter") + if b"upstream auth_server" not in baseline or b"location @service_unavailable" not in baseline: + raise CustomConfigError("custom-config baseline lacks required Fleet auth or service-unavailable handlers") + anchor_offset = _anchor_end_offset(baseline) + server_end_offsets = _http_server_end_offsets(baseline) + auth_denies = render_vpm_internal_auth_blocks().encode().rstrip() + auth_block = b"\n" + AUTH_DENIES_BEGIN + b"\n" + auth_denies + b"\n" + AUTH_DENIES_END + b"\n" + insertions = [(offset, auth_block) for offset in server_end_offsets] + if enabled: + fragment = render_vpm_fragment().encode().rstrip() + vpm_block = MANAGED_BEGIN + b"\n" + fragment + b"\n" + MANAGED_END + b"\n" + insertions.append((anchor_offset, vpm_block)) + + rendered = baseline + for offset, block in sorted(insertions, reverse=True): + rendered = rendered[:offset] + block + rendered[offset:] + return rendered + + +def ensure_vpm_only_change(previous: dict, requested: dict) -> None: + """Custom mode never rewrites routes for a service other than VPM.""" + changed = { + name + for name in set(previous) | set(requested) + if previous.get(name) != requested.get(name) + } + if changed - {"vast-price-manager"}: + raise CustomConfigError("custom-config mode permits only vast-price-manager lifecycle mutations") diff --git a/src/cryptolabs_proxy/migration.py b/src/cryptolabs_proxy/migration.py new file mode 100644 index 0000000..66065fa --- /dev/null +++ b/src/cryptolabs_proxy/migration.py @@ -0,0 +1,652 @@ +"""Dry-run planning and bounded state helpers for custom VPM proxy migration. + +This module intentionally does not contact Docker at import time. The calling +host helper owns Docker execution; these functions build deterministic plans +and create payloads without printing inspected configuration or environment. +""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass +import hashlib +import http.client +import json +import os +from pathlib import Path +import re +import socket +import subprocess +import time +from typing import Any +from urllib.parse import quote + +from .custom_config import configure_custom_config_mode, render_managed_vpm_config +from .vpm_prerequisite import get_vpm_prerequisite + + +_LOCAL_IMAGE_ID = re.compile(r"^sha256:[0-9a-f]{64}$") + + +class MigrationError(RuntimeError): + """A custom-config migration precondition or compensated switch failed.""" + + +@dataclass(frozen=True) +class MigrationOutcome: + """The serving state reached by a completed compensating switch.""" + + state: str + detail: str = "" + + @classmethod + def applied(cls) -> "MigrationOutcome": + return cls("applied") + + @classmethod + def candidate_retained(cls, detail: str) -> "MigrationOutcome": + return cls("candidate_retained", detail) + + +# Keep this in sync with the image HEALTHCHECK in Dockerfile. A candidate can +# remain in Docker's `starting` state until the start period and a scheduled +# health check have elapsed; the Engine request timeout is deliberately a +# separate, short transport bound. +DOCKER_HEALTH_START_PERIOD = 10 +DOCKER_HEALTH_INTERVAL = 30 +DOCKER_HEALTH_TIMEOUT = 10 +DOCKER_HEALTH_RETRIES = 3 +PROXY_READY_TIMEOUT = ( + DOCKER_HEALTH_START_PERIOD + + DOCKER_HEALTH_INTERVAL * DOCKER_HEALTH_RETRIES + + DOCKER_HEALTH_TIMEOUT + + 10 # bounded local auth-probe allowance +) + + +@dataclass(frozen=True) +class MigrationPlan: + container_id: str + baseline_sha256: str + image: str + migration_id: str + + @classmethod + def from_source(cls, container_id: str, baseline: bytes, image: str) -> "MigrationPlan": + baseline_sha256 = hashlib.sha256(baseline).hexdigest() + material = f"{container_id}:{baseline_sha256}:{image}".encode() + return cls(container_id, baseline_sha256, image, hashlib.sha256(material).hexdigest()[:20]) + + def sanitized_manifest(self) -> dict[str, str]: + return { + "migration_id": self.migration_id, + "container_id": self.container_id, + "baseline_sha256": self.baseline_sha256, + "image": self.image, + } + + +def _mount_request(mount: dict[str, Any]) -> dict[str, Any]: + """Convert inspect mount data to the Docker create API representation.""" + mount_type = mount["Type"] + if mount_type not in ("bind", "volume"): + raise MigrationError(f"unsupported mount type {mount_type}; refusing to drop its options") + source = mount.get("Name") if mount_type == "volume" else mount.get("Source") + if not source: + raise MigrationError(f"cannot preserve {mount_type} mount without a source") + result = { + "Type": mount_type, + "Source": source, + "Target": mount["Destination"], + "ReadOnly": not mount.get("RW", True), + } + if mount_type == "bind" and mount.get("Propagation"): + result["BindOptions"] = {"Propagation": mount["Propagation"]} + return result + + +def endpoint_configurations(inspect: dict[str, Any]) -> dict[str, dict[str, Any]]: + """Return only user-configured endpoint fields accepted by Docker create.""" + source_networks = inspect.get("NetworkSettings", {}).get("Networks", {}) + # Inspect also reports Engine-assigned endpoint IDs, IP addresses, and + # gateways. They are not valid create inputs; retain every user-configured + # endpoint setting while allowing Docker to allocate those runtime fields. + endpoint_fields = ("Aliases", "Links", "IPAMConfig", "MacAddress", "DriverOpts", "GwPriority") + networks = { + name: {field: deepcopy(endpoint[field]) for field in endpoint_fields if field in endpoint} + for name, endpoint in source_networks.items() + } + if not networks: + raise MigrationError("source proxy has no inspectable network endpoints") + return networks + + +def build_recreate_request(inspect: dict[str, Any], image: str, candidate_config_path: str) -> dict[str, Any]: + """Clone inspected Engine settings and add only the read-only config bind.""" + config = deepcopy(inspect["Config"]) + host_config = deepcopy(inspect["HostConfig"]) + config["Image"] = image + mounts = [_mount_request(mount) for mount in inspect.get("Mounts", [])] + if any(mount["Target"] == "/etc/nginx/nginx.conf" for mount in mounts): + raise MigrationError("source proxy already has an nginx.conf mount") + mounts.append({ + "Type": "bind", + "Source": candidate_config_path, + "Target": "/etc/nginx/nginx.conf", + "ReadOnly": True, + }) + # Docker accepts Mounts as the declarative form. Removing Binds prevents a + # duplicate mount while every source mount is reconstructed above. + host_config.pop("Binds", None) + host_config["Mounts"] = mounts + networks = endpoint_configurations(inspect) + return { + "Config": config, + "HostConfig": host_config, + "NetworkingConfig": {"EndpointsConfig": networks}, + } + + +def prepare_custom_registry(registry, baseline_path: Path, enable_vpm: bool) -> bytes: + """Configure custom mode and return the candidate config without reloading.""" + configure_custom_config_mode(registry.config, baseline_path) + if enable_vpm: + from .services import DEFAULT_SERVICES + registry.services["vast-price-manager"] = deepcopy(DEFAULT_SERVICES["vast-price-manager"]) + else: + registry.services.pop("vast-price-manager", None) + return render_managed_vpm_config(baseline_path.read_bytes(), enable_vpm) + + +class SwitchController: + """Compensating swap sequence with an injected Engine adapter for testing.""" + + def __init__(self, engine, health_check, rollback_health_check=None): + self.engine = engine + self.health_check = health_check + self.rollback_health_check = rollback_health_check or health_check + + def apply( + self, + source_id: str, + migration_id: str, + request: dict[str, Any], + candidate_name: str | None = None, + ) -> MigrationOutcome: + candidate_name = candidate_name or source_id + rollback_name = f"{candidate_name}.rollback-{migration_id}" + held_candidate_name = f"{candidate_name}.candidate-{migration_id}" + endpoints = request["NetworkingConfig"]["EndpointsConfig"] + candidate_created = False + renamed = False + stopped = False + detached_source_networks: list[str] = [] + try: + self.engine.stop(source_id) + stopped = True + for network in endpoints: + self.engine.disconnect(source_id, network) + detached_source_networks.append(network) + self.engine.rename(source_id, rollback_name) + renamed = True + self.engine.create(candidate_name, request) + candidate_created = True + self.engine.start(candidate_name) + if not self.health_check(candidate_name): + raise MigrationError("candidate proxy health check failed") + except Exception as error: + try: + if candidate_created: + self.engine.stop(candidate_name, ignore_missing=True) + for network in endpoints: + self.engine.disconnect(candidate_name, network) + self.engine.rename(candidate_name, held_candidate_name) + if renamed: + self.engine.rename(rollback_name, candidate_name) + for network, endpoint in endpoints.items(): + self.engine.connect(candidate_name, network, endpoint) + self.engine.start(candidate_name) + if not self.rollback_health_check(candidate_name): + if candidate_created: + self.engine.stop(candidate_name, ignore_missing=True) + for network in endpoints: + self.engine.disconnect(candidate_name, network) + self.engine.rename(candidate_name, rollback_name) + self.engine.rename(held_candidate_name, candidate_name) + for network, endpoint in endpoints.items(): + self.engine.connect(candidate_name, network, endpoint) + self.engine.start(candidate_name) + if not self.health_check(candidate_name): + raise MigrationError("candidate and restored proxy health checks both failed") + return MigrationOutcome.candidate_retained( + "source restoration health check failed; recovered candidate is serving" + ) + raise MigrationError("restored proxy health check failed") + if candidate_created: + self.engine.remove(held_candidate_name, ignore_missing=True) + elif stopped: + # A failed disconnect or rename leaves the source under + # its original name but stopped. Restore only endpoints + # whose detach request completed, preserving static IPs + # and aliases before restarting it. + for network in detached_source_networks: + self.engine.connect(source_id, network, endpoints[network]) + self.engine.start(source_id) + except Exception as rollback_error: + raise MigrationError( + f"custom proxy migration failed: {error}; rollback also failed: {rollback_error}" + ) from rollback_error + if isinstance(error, MigrationError): + raise + raise MigrationError(f"custom proxy migration failed: {error}") from error + return MigrationOutcome.applied() + + +def wait_for_proxy_ready(engine, container: str, auth_probe, timeout: int, interval: float = 0.25) -> bool: + """Wait through Docker health startup and require anonymous auth rejection.""" + deadline = time.monotonic() + timeout + while True: + state = engine.inspect(container).get("State", {}) + health = state.get("Health", {}).get("Status") + if not state.get("Running") or health == "unhealthy": + return False + if health == "healthy": + try: + return bool(auth_probe()) + except OSError: + return False + if time.monotonic() >= deadline: + return False + time.sleep(interval) + + +def wait_for_container_healthy(engine, container: str, timeout: int, interval: float = 0.25) -> bool: + """Bounded health wait used when restoring the pre-VPM proxy image.""" + deadline = time.monotonic() + timeout + while True: + state = engine.inspect(container).get("State", {}) + health = state.get("Health", {}).get("Status") + if not state.get("Running") or health == "unhealthy": + return False + if health == "healthy": + return True + if time.monotonic() >= deadline: + return False + time.sleep(interval) + + +class _UnixHTTPConnection(http.client.HTTPConnection): + """Small standard-library Docker Engine API client over its Unix socket.""" + + def __init__(self, socket_path: str, timeout: int = 15): + super().__init__("localhost", timeout=timeout) + self.socket_path = socket_path + + def connect(self): + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.settimeout(self.timeout) + self.sock.connect(self.socket_path) + + +class DockerEngine: + """Bounded Engine actions that keep inspect/config values out of stdout.""" + + def __init__(self, socket_path: str = "/var/run/docker.sock", timeout: int = 15): + self.socket_path = socket_path + self.timeout = timeout + + def _request(self, method: str, path: str, payload: dict | None = None, allowed=(200, 201, 204)): + connection = _UnixHTTPConnection(self.socket_path, self.timeout) + body = json.dumps(payload).encode() if payload is not None else None + headers = {"Content-Type": "application/json"} if body else {} + try: + connection.request(method, path, body=body, headers=headers) + response = connection.getresponse() + content = response.read() + except OSError as error: + raise MigrationError(f"Docker Engine request failed: {error}") from error + finally: + connection.close() + if response.status not in allowed: + # Engine failures can include sensitive Config data; retain only + # the status, never echo its body into command output. + raise MigrationError(f"Docker Engine {method} {path} returned HTTP {response.status}") + return content + + def inspect(self, container: str) -> dict[str, Any]: + return json.loads(self._request("GET", f"/containers/{quote(container, safe='')}/json")) + + def stop(self, container: str, ignore_missing: bool = False): + allowed = (204, 304, 404) if ignore_missing else (204, 304) + # Leave the Unix-socket read timeout enough headroom to receive the + # Engine response after its graceful-stop period. + grace = max(1, self.timeout - 2) + self._request("POST", f"/containers/{quote(container, safe='')}/stop?t={grace}", allowed=allowed) + + def rename(self, old: str, new: str): + self._request("POST", f"/containers/{quote(old, safe='')}/rename?name={quote(new, safe='')}", allowed=(204,)) + + def create(self, name: str, request: dict[str, Any]): + payload = deepcopy(request["Config"]) + payload["HostConfig"] = request["HostConfig"] + payload["NetworkingConfig"] = request["NetworkingConfig"] + self._request("POST", f"/containers/create?name={quote(name, safe='')}", payload, allowed=(201,)) + + def start(self, container: str): + self._request("POST", f"/containers/{quote(container, safe='')}/start", allowed=(204, 304)) + + def remove(self, container: str, ignore_missing: bool = False): + allowed = (204, 404) if ignore_missing else (204,) + self._request("DELETE", f"/containers/{quote(container, safe='')}?force=1", allowed=allowed) + + def disconnect(self, container: str, network: str): + self._request( + "POST", + f"/networks/{quote(network, safe='')}/disconnect", + {"Container": container, "Force": True}, + allowed=(200,), + ) + + def connect(self, container: str, network: str, endpoint: dict[str, Any]): + self._request( + "POST", + f"/networks/{quote(network, safe='')}/connect", + {"Container": container, "EndpointConfig": endpoint}, + allowed=(200,), + ) + + def healthy(self, container: str) -> bool: + state = self.inspect(container).get("State", {}) + health = state.get("Health", {}).get("Status") + return state.get("Running") and health in (None, "healthy") + + +class CustomConfigMigrator: + """Explicit plan/apply/rollback workflow for one custom proxy container.""" + + def __init__( + self, + config_dir: Path, + backup_root: Path, + engine: DockerEngine | None = None, + timeout: int = 15, + readiness_timeout: int = PROXY_READY_TIMEOUT, + ): + self.config_dir = config_dir + self.backup_root = backup_root + self.engine = engine or DockerEngine(timeout=timeout) + # `timeout` bounds each Unix socket/subprocess/HTTP operation. It is + # not the lifecycle deadline, because Docker's own health cadence is + # intentionally much longer. + self.timeout = timeout + if readiness_timeout < PROXY_READY_TIMEOUT: + raise MigrationError( + f"readiness timeout must be at least {PROXY_READY_TIMEOUT} seconds for the proxy healthcheck" + ) + self.readiness_timeout = readiness_timeout + + def _read_active_config(self, container: str) -> bytes: + try: + result = subprocess.run( + ["docker", "exec", container, "cat", "/etc/nginx/nginx.conf"], + capture_output=True, + timeout=self.timeout, + ) + except (OSError, subprocess.SubprocessError) as error: + raise MigrationError(f"cannot read active proxy config: {error}") from error + if result.returncode: + raise MigrationError("cannot read active proxy config") + return result.stdout + + def plan(self, container: str, image: str) -> MigrationPlan: + """Read only: return a deterministic, sanitized migration identity.""" + if "@sha256:" not in image and _LOCAL_IMAGE_ID.fullmatch(image) is None: + raise MigrationError("proxy image must be an immutable sha256 digest") + inspect = self.engine.inspect(container) + return MigrationPlan.from_source(inspect["Id"], self._read_active_config(container), image) + + def _backup_path(self, migration_id: str) -> Path: + return self.backup_root / migration_id + + @staticmethod + def _write_private(path: Path, content: bytes): + path.write_bytes(content) + path.chmod(0o600) + + def _validate_candidate(self, image: str, candidate: Path, inspect: dict[str, Any], migration_id: str): + if _LOCAL_IMAGE_ID.fullmatch(image) is not None: + # Docker otherwise treats an absent sha256: argument as a + # repository/tag candidate for `docker run`. Verify the exact + # locally loaded content ID before creating any validator. + try: + local = subprocess.run( + ["docker", "image", "inspect", "--format", "{{.Id}}", image], + capture_output=True, + text=True, + timeout=self.timeout, + ) + except (OSError, subprocess.SubprocessError) as error: + raise MigrationError(f"offline image missing: exact local immutable image ID is unavailable ({error})") from error + if local.returncode != 0 or local.stdout.strip() != image: + raise MigrationError("offline image missing: exact local immutable image ID is unavailable") + ssl_mount = next((m for m in inspect.get("Mounts", []) if m.get("Destination") == "/etc/nginx/ssl"), None) + if not ssl_mount: + raise MigrationError("source proxy has no /etc/nginx/ssl mount for candidate validation") + validation_name = f"cryptolabs-vpm-validate-{migration_id}"[:63] + command = [ + "docker", "run", "--name", validation_name, + "--label", f"cryptolabs.migration.validation={migration_id}", + "--network", "none", "--entrypoint", "nginx", + "-v", f"{candidate}:/etc/nginx/nginx.conf:ro", + "-v", f"{ssl_mount['Source']}:/etc/nginx/ssl:ro", + image, "-t", + ] + try: + result = subprocess.run(command, capture_output=True, timeout=self.timeout) + except (OSError, subprocess.SubprocessError) as error: + raise MigrationError(f"candidate Nginx validation could not run: {error}") from error + finally: + # The image entrypoint may ignore argv or outlive a killed Docker + # client. Explicitly remove only our unique labeled container. + subprocess.run( + ["docker", "rm", "-f", validation_name], + capture_output=True, + timeout=self.timeout, + ) + if result.returncode: + raise MigrationError("candidate Nginx validation failed") + + def _anonymous_auth_is_rejected(self) -> bool: + """The replacement must expose the new endpoint without a session.""" + connection = http.client.HTTPConnection("127.0.0.1", 80, timeout=self.timeout) + try: + connection.request("GET", "/auth/vast-price-manager/authorize") + return connection.getresponse().status == 401 + except OSError: + return False + finally: + connection.close() + + def apply(self, container: str, image: str, migration_id: str, enable_vpm: bool) -> MigrationOutcome: + """Snapshot privately, validate, then perform a compensating proxy-only swap.""" + if os.geteuid() != 0: + raise MigrationError("custom proxy migration must run as root") + from .cli import registry_lock + from .services import ServiceRegistry + with registry_lock(self.config_dir): + registry = ServiceRegistry(self.config_dir) + # Existing registered VPM services remain maintainable when their + # exporter later goes down. A first custom enable needs fresh proof + # before backups, candidate files, or Docker changes are created. + if enable_vpm and registry.get_service("vast-price-manager") is None: + prerequisite = get_vpm_prerequisite() + if not prerequisite["configured"]: + raise MigrationError( + "Requires Vast.ai setup with at least one connected account. " + "Open /vastai/ to finish setup." + ) + # Rebuild the read-only plan while holding the same lock that + # serializes later VPM register/unregister writes. + plan = self.plan(container, image) + if plan.migration_id != migration_id: + raise MigrationError("migration ID does not match the current source container/config/image") + backup = self._backup_path(migration_id) + if backup.exists(): + raise MigrationError("migration backup already exists") + self.backup_root.mkdir(parents=True, mode=0o700) + self.backup_root.chmod(0o700) + backup.mkdir(parents=True, mode=0o700) + backup.chmod(0o700) + inspect = self.engine.inspect(container) + baseline = self._read_active_config(container) + current = MigrationPlan.from_source(inspect["Id"], baseline, image) + if current != plan: + raise MigrationError("source container ID or config changed before switch") + self._write_private(backup / "baseline.nginx.conf", baseline) + self._write_private(backup / "inspect.json", json.dumps(inspect, sort_keys=True).encode()) + previous = _snapshot_registry_files(registry) + _write_registry_backup(backup, previous) + candidate = self.config_dir / "nginx.conf" + previous_candidate = candidate.read_bytes() if candidate.exists() else None + try: + candidate_content = prepare_custom_registry(registry, backup / "baseline.nginx.conf", enable_vpm) + self._write_private(candidate, candidate_content) + registry.save() + self._validate_candidate(image, candidate, inspect, migration_id) + request = build_recreate_request(inspect, image, str(candidate)) + outcome = SwitchController( + self.engine, + lambda name: wait_for_proxy_ready( + self.engine, + name, + self._anonymous_auth_is_rejected, + self.readiness_timeout, + ), + rollback_health_check=lambda name: wait_for_container_healthy( + self.engine, + name, + self.readiness_timeout, + ), + ).apply(plan.container_id, migration_id, request, candidate_name=container) + except Exception: + _restore_registry_files(previous) + if previous_candidate is None: + candidate.unlink(missing_ok=True) + else: + self._write_private(candidate, previous_candidate) + raise + # A retained candidate is the live recovery path. Marker writing + # is diagnostic only and must never re-enter generic cleanup that + # would unlink/revert the configuration it is serving. + if outcome.state == "candidate_retained": + try: + self._write_private( + backup / "outcome.json", + json.dumps({"state": outcome.state, "detail": outcome.detail}, sort_keys=True).encode(), + ) + except OSError: + return MigrationOutcome.candidate_retained( + f"{outcome.detail}; recovery marker could not be written" + ) + return outcome + + def rollback(self, container: str, migration_id: str) -> MigrationOutcome: + """Restore only this proxy without discarding its healthy replacement.""" + if os.geteuid() != 0: + raise MigrationError("custom proxy rollback must run as root") + from .cli import registry_lock + # Match lifecycle register/unregister: the same lock covers runtime + # changes and the registry/config restore so no command observes or + # persists a split state. + with registry_lock(self.config_dir): + return self._rollback_locked(container, migration_id) + + def _rollback_locked(self, container: str, migration_id: str) -> MigrationOutcome: + backup = self._backup_path(migration_id) + if not backup.is_dir(): + raise MigrationError("migration backup does not exist") + rollback_name = f"{container}.rollback-{migration_id}" + held_candidate = f"{container}.candidate-{migration_id}" + source = json.loads((backup / "inspect.json").read_text()) + restored = self.engine.inspect(rollback_name) + if restored.get("Id") != source.get("Id"): + raise MigrationError("rollback source does not match the private backup") + # Prove the replacement works before it is stopped or renamed. + if not wait_for_proxy_ready(self.engine, container, self._anonymous_auth_is_rejected, self.readiness_timeout): + raise MigrationError("current replacement is not healthy; refusing to remove its recovery path") + source_endpoints = endpoint_configurations(source) + candidate_endpoints = endpoint_configurations(self.engine.inspect(container)) + candidate_held = False + source_promoted = False + detached_candidate_networks: list[str] = [] + try: + self.engine.stop(container) + for network in candidate_endpoints: + self.engine.disconnect(container, network) + detached_candidate_networks.append(network) + self.engine.rename(container, held_candidate) + candidate_held = True + self.engine.rename(rollback_name, container) + source_promoted = True + for network, endpoint in source_endpoints.items(): + self.engine.connect(container, network, endpoint) + self.engine.start(container) + if not wait_for_container_healthy(self.engine, container, self.readiness_timeout): + raise MigrationError("restored proxy health check failed") + except Exception as error: + try: + if source_promoted: + self.engine.stop(container, ignore_missing=True) + for network in source_endpoints: + self.engine.disconnect(container, network) + self.engine.rename(container, rollback_name) + if candidate_held: + self.engine.rename(held_candidate, container) + for network, endpoint in candidate_endpoints.items(): + self.engine.connect(container, network, endpoint) + self.engine.start(container) + if not wait_for_proxy_ready( + self.engine, + container, + self._anonymous_auth_is_rejected, + self.readiness_timeout, + ): + raise MigrationError("candidate recovery health check failed") + else: + # Candidate was never renamed, so restore exactly the + # networks detached before the failed rename/disconnect. + for network in detached_candidate_networks: + self.engine.connect(container, network, candidate_endpoints[network]) + self.engine.start(container) + except Exception as recovery_error: + raise MigrationError( + f"rollback failed: {error}; candidate recovery also failed: {recovery_error}" + ) from recovery_error + raise MigrationError("rollback failed; healthy replacement was restored") from error + self.engine.remove(held_candidate, ignore_missing=True) + _restore_registry_files(_read_registry_backup(backup)) + return MigrationOutcome("rolled_back") + + +def _snapshot_registry_files(registry) -> dict[Path, bytes | None]: + paths = [registry.services_file, registry.config_file, registry.config_dir / "nginx.conf"] + return {path: path.read_bytes() if path.exists() else None for path in paths} + + +def _write_registry_backup(backup: Path, snapshot: dict[Path, bytes | None]): + encoded = {str(path): content.decode("latin1") if content is not None else None for path, content in snapshot.items()} + CustomConfigMigrator._write_private(backup / "registry-snapshot.json", json.dumps(encoded).encode()) + + +def _read_registry_backup(backup: Path) -> dict[Path, bytes | None]: + encoded = json.loads((backup / "registry-snapshot.json").read_text()) + return {Path(path): content.encode("latin1") if content is not None else None for path, content in encoded.items()} + + +def _restore_registry_files(snapshot: dict[Path, bytes | None]): + for path, content in snapshot.items(): + if content is None: + path.unlink(missing_ok=True) + else: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) diff --git a/src/cryptolabs_proxy/services.py b/src/cryptolabs_proxy/services.py index 72d406f..4922913 100644 --- a/src/cryptolabs_proxy/services.py +++ b/src/cryptolabs_proxy/services.py @@ -71,6 +71,18 @@ "icon": "💎", "description": "Prometheus exporter for Vast.ai metrics", }, + "vast-price-manager": { + "container_name": "vast-price-manager", + "path": "/vast-pricing/", + "port": 8088, + "display_name": "Vast Price Manager", + "icon": "💰", + "description": "Optional Vast pricing management service with its own secure login.", + "product_url": "https://github.com/cryptolabsza/vast-price-manager", + "admin_only": True, + "lifecycle_manager": "dc-overview", + "optional": True, + }, "runpod-exporter": { "container_name": "runpod-exporter", "path": "/runpod-metrics/", diff --git a/src/cryptolabs_proxy/templates/nginx.conf.j2 b/src/cryptolabs_proxy/templates/nginx.conf.j2 index 496d1e0..fa001fc 100644 --- a/src/cryptolabs_proxy/templates/nginx.conf.j2 +++ b/src/cryptolabs_proxy/templates/nginx.conf.j2 @@ -158,8 +158,16 @@ http { return 302 /auth/login?next=$request_uri; } + {% include "vast-price-manager-auth-blocks.conf.j2" %} + + {% if 'vast-price-manager' in services %} + {% include "vast-price-manager.conf.j2" %} + {% endif %} + {% for name, service in services.items() %} - {% if not service.path %} + {% if name == 'vast-price-manager' %} + # Vast Price Manager uses its dedicated admin-only route above. + {% elif not service.path %} # {{ service.get('display_name', name) }} - external service, no local proxy route {% else %} # {{ service.get('display_name', name) }} - uses variable for runtime DNS resolution diff --git a/src/cryptolabs_proxy/templates/vast-price-manager-auth-blocks.conf.j2 b/src/cryptolabs_proxy/templates/vast-price-manager-auth-blocks.conf.j2 new file mode 100644 index 0000000..5c4653c --- /dev/null +++ b/src/cryptolabs_proxy/templates/vast-price-manager-auth-blocks.conf.j2 @@ -0,0 +1,7 @@ + # VPM calls these endpoints only over the Docker network on port 8081. + # Exact locations keep the generic public /auth/ forwarding route from + # exposing the session authority or password reauthentication API. + location = /auth/vast-price-manager/session { return 404; } + location = /auth/vast-price-manager/session/ { return 404; } + location = /auth/vast-price-manager/reauth { return 404; } + location = /auth/vast-price-manager/reauth/ { return 404; } diff --git a/src/cryptolabs_proxy/templates/vast-price-manager.conf.j2 b/src/cryptolabs_proxy/templates/vast-price-manager.conf.j2 new file mode 100644 index 0000000..bee6989 --- /dev/null +++ b/src/cryptolabs_proxy/templates/vast-price-manager.conf.j2 @@ -0,0 +1,43 @@ + # Vast Price Manager has a dedicated admin-only route. It strips the + # prefix once and explicitly preserves the remainder and query string. + location = /_vast_pricing_admin { + internal; + proxy_pass http://127.0.0.1:8081/auth/vast-price-manager/authorize; + proxy_pass_request_body off; + proxy_set_header Content-Length ""; + proxy_set_header Cookie $http_cookie; + proxy_set_header X-Original-URI $request_uri; + } + + location = /vast-pricing { + return 308 /vast-pricing/; + } + + location ~ ^/vast-pricing(?/.*)$ { + auth_request /_vast_pricing_admin; + error_page 401 = @login_redirect; + + # Runtime Docker DNS resolution permits startup before this opt-in + # service has been installed. + set $upstream_vast_price_manager vast-price-manager; + proxy_pass http://$upstream_vast_price_manager:8088$vpm_upstream_path$is_args$args; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Host $host; + proxy_set_header X-Forwarded-Proto $scheme; + # VPM directly introspects the signed Fleet session on its + # configured internal origin. Never pass client identity headers. + proxy_set_header X-Fleet-Auth-User ""; + proxy_set_header X-Fleet-Auth-Role ""; + proxy_set_header X-Fleet-Auth-Token ""; + proxy_set_header X-Fleet-Authenticated ""; + # Preserve the browser Fleet cookie for VPM's direct introspection. + proxy_set_header Cookie $http_cookie; + # Auth-request failures still use the redirect above. Do not + # intercept VPM's own 401 reauthentication result, which must + # reach its caller as an error rather than a Fleet login redirect. + proxy_intercept_errors off; + error_page 502 503 504 = @service_unavailable; + } diff --git a/src/cryptolabs_proxy/updates.py b/src/cryptolabs_proxy/updates.py new file mode 100644 index 0000000..a5f50ff --- /dev/null +++ b/src/cryptolabs_proxy/updates.py @@ -0,0 +1,450 @@ +"""Serialized fleet image updates executed outside the serving proxy container. + +Job files contain only operation state, never container environments or Docker +error bodies. Existing containers remain available as rollback artifacts. +""" +from __future__ import annotations + +import fcntl +import json +import os +import re +import subprocess +import time +import uuid +from contextlib import contextmanager +from copy import deepcopy +from pathlib import Path +from urllib.parse import quote +from urllib.request import urlopen + +from .migration import DockerEngine, MigrationError, _mount_request, endpoint_configurations + + +SERVICES = { + 'cryptolabs-proxy': {'container': 'cryptolabs-proxy', 'port': 8080, 'image': 'ghcr.io/cryptolabsza/cryptolabs-proxy', 'self': True}, + 'ipmi-monitor': {'container': 'ipmi-monitor', 'port': 5000, 'image': 'ghcr.io/cryptolabsza/ipmi-monitor'}, + 'dc-overview': {'container': 'dc-overview', 'port': 5001, 'image': 'ghcr.io/cryptolabsza/dc-overview'}, + 'grafana': {'container': 'grafana', 'port': 3000, 'image': 'grafana/grafana'}, + 'prometheus': {'container': 'prometheus', 'port': 9090, 'image': 'prom/prometheus'}, + 'vastai-exporter': {'container': 'vastai-exporter', 'port': 8622, 'image': 'ghcr.io/cryptolabsza/vastai-exporter'}, + 'vast-price-manager': {'container': 'vast-price-manager', 'port': 8088, 'image': '', + 'lifecycle_manager': 'dc-overview', 'update_supported': False}, + 'runpod-exporter': {'container': 'runpod-exporter', 'port': 8623, 'image': 'ghcr.io/cryptolabsza/runpod-exporter'}, +} +JOBS_DIR = Path('/data/auth/update-jobs') +CHANNEL_LABEL = 'io.cryptolabs.update.channel-image' +IMAGE_LABEL = 'io.cryptolabs.update.image-id' +ACTIVE_STATES = {'queued', 'running'} +BUILD_ENV = {'GIT_COMMIT', 'GIT_BRANCH', 'BUILD_TIME', 'APP_VERSION', 'HOSTNAME'} +# Docker 29 no longer accepts old Engine API clients. 1.45 is supported by +# Docker 29 (minimum 1.44) and retains the endpoint fields used below. +DOCKER_API_VERSION = 'v1.45' + + +class UpdateError(RuntimeError): + """Public, non-sensitive updater failure.""" + + +class UpdateBusy(UpdateError): + def __init__(self, job): + super().__init__('An update is already in progress.') + self.job = job + + +class UpdateEngine(DockerEngine): + def _request(self, method, path, payload=None, allowed=(200, 201, 204)): + """Use an Engine API version accepted by the Docker 29 fleet host.""" + if not path.startswith(f'/{DOCKER_API_VERSION}/'): + path = f'/{DOCKER_API_VERSION}{path}' + return super()._request(method, path, payload, allowed) + + def image(self, reference): + return json.loads(self._request('GET', f'/images/{quote(reference, safe="")}/json')) + + +def target_for(name, source, branch): + """Only floating, first-party channels may follow the branch selector.""" + service = SERVICES[name] + if service.get('update_supported') is False: + return None + config = source.get('Config', {}) + reference = config.get('Image', '') + labels = config.get('Labels') or {} + if reference.startswith('sha256:') and labels.get(IMAGE_LABEL) == source.get('Image'): + reference = labels.get(CHANNEL_LABEL, reference) + base = service['image'] + if reference not in {base, f'{base}:latest', f'{base}:main', f'{base}:dev'}: + return None + tag = ('dev' if branch == 'dev' else 'latest') if base.startswith('ghcr.io/cryptolabsza/') else reference.rsplit(':', 1)[-1] + if tag == base or tag == 'main': + tag = 'latest' + return f'{base}:{tag}' + + +def update_status(name, branch, *, engine=None): + engine = engine or UpdateEngine() + result = {'update_available': None, 'pinned': False} + try: + source = engine.inspect(SERVICES[name]['container']) + target = target_for(name, source, branch) + result.update({'running_image_id': source['Image'], 'target_image': target, 'pinned': target is None}) + if target: + candidate = engine.image(target) + result.update({'target_image_id': candidate['Id'], 'update_available': candidate['Id'] != source['Image']}) + except Exception: + pass # Unknown image identity is never presented as up-to-date. + return result + + +def _env(values): + return dict(value.split('=', 1) for value in (values or []) if '=' in value) + + +def build_update_request(source, old_image, new_image): + """Preserve deploy overrides while allowing target image defaults to change.""" + config = deepcopy(source['Config']) + old_defaults, new_defaults = old_image.get('Config') or {}, new_image.get('Config') or {} + env = _env(new_defaults.get('Env')) + old_env = _env(old_defaults.get('Env')) + for key, value in _env(config.get('Env')).items(): + if key not in BUILD_ENV and (key not in old_env or old_env[key] != value): + env[key] = value + config['Env'] = [f'{key}={value}' for key, value in env.items()] + labels = deepcopy(new_defaults.get('Labels') or {}) + old_labels = old_defaults.get('Labels') or {} + for key, value in (config.get('Labels') or {}).items(): + if not key.startswith('org.opencontainers.image.') and (key not in old_labels or old_labels[key] != value): + labels[key] = value + config['Labels'] = labels + for key in ('Cmd', 'Entrypoint', 'User', 'WorkingDir', 'Healthcheck', 'StopSignal', 'Shell', 'ExposedPorts', 'Volumes'): + if key in old_defaults and config.get(key) == old_defaults[key]: + if key in new_defaults: + config[key] = deepcopy(new_defaults[key]) + else: + config.pop(key, None) + if config.get('Hostname') in (source.get('Id'), source.get('Id', '')[:12]): + config.pop('Hostname', None) + config['Image'] = new_image['Id'] + host = deepcopy(source['HostConfig']) + if host.get('AutoRemove') or host.get('VolumesFrom'): + raise UpdateError('Automatic removal or inherited volumes require a managed update.') + mounts = deepcopy(host.get('Mounts') or []) + targets = {mount.get('Target') for mount in mounts} + targets.update((host.get('Tmpfs') or {}).keys()) + for binding in host.get('Binds') or []: + parts = binding.split(':') + if len(parts) < 2: + raise UpdateError('Cannot preserve an ambiguous mount.') + targets.add(parts[1]) + for mount in source.get('Mounts', []): + if mount['Destination'] not in targets: + mounts.append(_mount_request(mount)) + host['Mounts'] = mounts + endpoints = endpoint_configurations(source) + generated_aliases = {source.get('Id'), source.get('Id', '')[:12]} + for endpoint in endpoints.values(): + if endpoint.get('Aliases'): + endpoint['Aliases'] = [alias for alias in endpoint['Aliases'] if alias not in generated_aliases] + return {'Config': config, 'HostConfig': host, 'NetworkingConfig': {'EndpointsConfig': endpoints}} + + +def _probe(source, name): + """Fallback application probe for containers without Docker healthchecks.""" + service = SERVICES[name] + addresses = [endpoint.get('IPAddress') for endpoint in source.get('NetworkSettings', {}).get('Networks', {}).values()] + if source.get('HostConfig', {}).get('NetworkMode') == 'host': + addresses = ['127.0.0.1'] + paths = {'cryptolabs-proxy': '/api/health', 'dc-overview': '/api/health', + 'ipmi-monitor': '/health', 'grafana': '/api/health', 'prometheus': '/-/ready'} + path = paths.get(name, '/metrics') + if name == 'prometheus': + for argument in source.get('Config', {}).get('Cmd') or []: + if argument.startswith('--web.route-prefix='): + path = argument.split('=', 1)[1].rstrip('/') + '/-/ready' + for address in addresses: + if address: + try: + with urlopen(f'http://{address}:{service["port"]}{path}', timeout=3) as response: + if response.status == 200: + return True + except Exception: + pass + return False + + +def wait_ready(engine, name, image_id, timeout=150): + deadline = time.monotonic() + timeout + while True: + source = engine.inspect(name) + state = source.get('State', {}) + health = state.get('Health', {}).get('Status') + if source.get('Image') != image_id or not state.get('Running') or health == 'unhealthy': + return False + if health == 'healthy' or (health is None and _probe(source, name)): + return True + if time.monotonic() >= deadline: + return False + time.sleep(1) + + +def replace_container(engine, name, image, job_id, *, timeout=150, expected_id=None): + """Swap without destroying the previous runtime; prove recovery on failure.""" + backup = f'{name}.rollback-{job_id}' + detached = [] + started_stop = renamed = created = False + source = None + try: + source = engine.inspect(name) + if expected_id and source['Id'] != expected_id: + raise UpdateError('Container changed during update preparation.') + if not source.get('State', {}).get('Running'): + raise UpdateError('Service is stopped; use its installer to start it.') + candidate = engine.image(image) + if candidate['Id'] == source['Image']: + return {'success': True, 'state': 'unchanged', 'image_id': source['Image'], 'message': 'Already running the selected image.'} + request = build_update_request(source, engine.image(source['Image']), candidate) + if not image.startswith('sha256:'): + request['Config']['Labels'].update({CHANNEL_LABEL: image, IMAGE_LABEL: candidate['Id']}) + endpoints = request['NetworkingConfig']['EndpointsConfig'] + started_stop = True + engine.stop(name) + for network in endpoints: + engine.disconnect(name, network) + detached.append(network) + engine.rename(name, backup) + renamed = True + engine.create(name, request) + created = True + engine.start(name) + if not wait_ready(engine, name, candidate['Id'], timeout): + raise UpdateError('Replacement failed application health or image verification.') + return {'success': True, 'state': 'completed', 'image_id': candidate['Id'], + 'rollback_container': backup, 'message': 'Updated and health verified.'} + except Exception as error: + message = str(error) if isinstance(error, UpdateError) else 'Container update failed.' + if not started_stop: + return {'success': False, 'state': 'failed', 'message': message} + try: + if created: + engine.remove(name, ignore_missing=True) + if renamed: + engine.rename(backup, name) + for network in detached: + engine.connect(name, network, endpoints[network]) + engine.start(name) + if not wait_ready(engine, name, source['Image'], timeout): + raise UpdateError('Original service did not recover.') + return {'success': False, 'state': 'rolled_back', 'message': message + ' Original service restored and health verified.'} + except Exception: + return {'success': False, 'state': 'recovery_required', 'message': message + ' Automatic recovery was not verified; operator recovery required.', 'rollback_container': backup if renamed else name} + + +def _job_path(directory, job_id): + if not re.fullmatch('[0-9a-f]{32}', job_id): + raise UpdateError('Invalid update job identifier.') + return Path(directory) / f'{job_id}.json' + + +def _write_json(path, value): + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + temporary = path.with_suffix('.tmp-' + uuid.uuid4().hex) + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(descriptor, 'w') as stream: + json.dump(value, stream) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + + +def write_job(directory, job): + _write_json(_job_path(directory, job['id']), job) + + +def read_job(directory, job_id): + try: + return json.loads(_job_path(directory, job_id).read_text()) + except FileNotFoundError: + raise UpdateError('Update job not found.') from None + + +def job_status(job_id, *, directory=JOBS_DIR, engine=None): + job = read_job(directory, job_id) + if job.get('state') in ACTIVE_STATES: + engine = engine or UpdateEngine() + try: + running = engine.inspect(job['helper'])['State']['Running'] + except MigrationError as error: + if not str(error).endswith('HTTP 404'): + return job + running = False + except Exception: + return job + if not running: + job = read_job(directory, job_id) + if job.get('state') in ACTIVE_STATES: + job.update(state='interrupted', success=False, + message='Update helper stopped before completion. Inspect services before retrying.') + write_job(directory, job) + return job + + +@contextmanager +def _lock(directory, name): + Path(directory).mkdir(mode=0o700, parents=True, exist_ok=True) + descriptor = os.open(Path(directory) / name, os.O_RDWR | os.O_CREAT, 0o600) + with os.fdopen(descriptor, 'w') as stream: + fcntl.flock(stream, fcntl.LOCK_EX) + yield + + +def helper_request(source, directory, job_id): + """The helper uses the current immutable proxy image, outside its PID space.""" + parent = next((mount for mount in sorted(source.get('Mounts', []), key=lambda item: len(item['Destination']), reverse=True) + if str(directory).startswith(mount['Destination'].rstrip('/') + '/')), None) + if not parent or not parent.get('RW', True): + raise UpdateError('Updater needs a writable persistent /data/auth mount.') + mount = _mount_request(parent) + return { + 'Config': {'Image': source['Image'], 'Entrypoint': ['python3'], + 'Cmd': ['-m', 'cryptolabs_proxy.updates', 'worker', job_id, str(directory)], + 'Env': ['PYTHONPATH=/app/src'], 'Healthcheck': {'Test': ['NONE']}, + 'Labels': {'io.cryptolabs.update.worker': job_id, 'com.centurylinklabs.watchtower.enable': 'false'}}, + 'HostConfig': {'NetworkMode': 'host', 'RestartPolicy': {'Name': 'no'}, + 'Mounts': [mount, {'Type': 'bind', 'Source': '/var/run/docker.sock', 'Target': '/var/run/docker.sock'}], + 'LogConfig': {'Type': 'json-file', 'Config': {'max-size': '1m', 'max-file': '1'}}}, + 'NetworkingConfig': {}, + } + + +def submit_job(service, branch, action, *, directory=JOBS_DIR, engine=None): + if branch not in ('main', 'dev') or action not in ('pull', 'update'): + raise UpdateError('Invalid update channel or action.') + if service != 'all' and (service not in SERVICES or SERVICES[service].get('update_supported') is False): + raise UpdateError('Service is not managed by this updater.') + engine = engine or UpdateEngine() + with _lock(directory, 'dispatch.lock'): + active = Path(directory) / 'active.json' + if active.exists(): + previous = read_job(directory, json.loads(active.read_text())['id']) + if previous.get('state') in ACTIVE_STATES: + try: + running = engine.inspect(previous['helper'])['State']['Running'] + except MigrationError as error: + if not str(error).endswith('HTTP 404'): + # An inaccessible Engine is not proof that a worker stopped. + raise UpdateBusy(previous) from None + running = False + except Exception: + # An inaccessible Engine is not proof that a worker stopped. + raise UpdateBusy(previous) from None + if running: + raise UpdateBusy(previous) + previous.update(state='interrupted', success=False, + message='Update helper exited before recording completion. Inspect service status before retrying.') + write_job(directory, previous) + job_id = uuid.uuid4().hex + helper = 'cryptolabs-update-' + job_id + request = helper_request(engine.inspect('cryptolabs-proxy'), Path(directory), job_id) + job = {'id': job_id, 'helper': helper, 'service': service, 'branch': branch, + 'action': action, 'state': 'queued', 'results': {}, 'created_at': time.time()} + write_job(directory, job) + _write_json(active, {'id': job_id}) + try: + engine.create(helper, request) + engine.start(helper) + except Exception: + job.update(state='failed', success=False, message='Could not launch update helper.') + write_job(directory, job) + raise UpdateError(job['message']) from None + return job + + +def pull_target(target): + try: + result = subprocess.run(['docker', 'pull', target], capture_output=True, timeout=300) + return result.returncode == 0 + except (OSError, subprocess.SubprocessError): + return False + + +def run_job(directory, job_id, *, engine=None): + engine = engine or UpdateEngine() + with _lock(directory, 'operation.lock'): + job = read_job(directory, job_id) + if job.get('state') not in (None, 'queued'): + return job # A completed/interrupted operation is never replayed. + job.update(state='running', results={}) + write_job(directory, job) + names = list(SERVICES) if job['service'] == 'all' else [job['service']] + names.sort(key=lambda name: name == 'cryptolabs-proxy') + for name in names: + if SERVICES[name].get('update_supported') is False: + continue + try: + source = engine.inspect(name) + target = target_for(name, source, job['branch']) + if not source.get('State', {}).get('Running') or not target: + result = {'success': job['service'] == 'all', 'state': 'skipped', + 'message': 'Stopped or pinned service; configuration preserved.'} + elif not pull_target(target): + result = {'success': False, 'state': 'failed', 'message': 'Image pull failed; service unchanged.'} + elif job['action'] == 'pull': + image = engine.image(target) + result = {'success': True, 'state': 'checked', 'image_id': image['Id'], + 'update_available': image['Id'] != source['Image'], 'message': 'Image checked; running service unchanged.'} + else: + result = replace_container(engine, name, target, job_id, expected_id=source['Id']) + except Exception as error: + missing = isinstance(error, MigrationError) and str(error).endswith('HTTP 404') + if missing and job['service'] == 'all': + result = {'success': True, 'state': 'skipped', 'message': 'Service is not installed.'} + else: + result = {'success': False, 'state': 'failed', 'message': 'Could not inspect or prepare service; no update started.'} + job['results'][name] = result + write_job(directory, job) + if result.get('state') == 'recovery_required': + break + success = all(result['success'] for result in job['results'].values()) + job.update(state='completed' if success else 'failed', success=success, finished_at=time.time()) + write_job(directory, job) + return job + + +def main(argv=None): + """Supported JSON interface for host fleet CLI clients.""" + import argparse + parser = argparse.ArgumentParser(description='Fleet image updater') + commands = parser.add_subparsers(dest='command', required=True) + submit = commands.add_parser('submit') + submit.add_argument('--branch', choices=('main', 'dev'), default='main') + submit.add_argument('--service', default='all', choices=['all', *SERVICES]) + submit.add_argument('--action', choices=('update', 'pull'), default='update') + status = commands.add_parser('status') + status.add_argument('job_id') + worker = commands.add_parser('worker') + worker.add_argument('job_id') + worker.add_argument('directory', type=Path) + args = parser.parse_args(argv) + try: + if args.command == 'submit': + result = submit_job(args.service, args.branch, args.action) + elif args.command == 'status': + result = job_status(args.job_id) + else: + result = run_job(args.directory, args.job_id) + print(json.dumps(result)) + return 0 + except UpdateBusy as error: + print(json.dumps({'error': str(error), 'job': error.job})) + return 2 + except UpdateError as error: + print(json.dumps({'error': str(error)})) + return 1 + except Exception: + print(json.dumps({'error': 'Updater unavailable; no completion confirmed.'})) + return 1 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/src/cryptolabs_proxy/vpm_prerequisite.py b/src/cryptolabs_proxy/vpm_prerequisite.py new file mode 100644 index 0000000..d684cb7 --- /dev/null +++ b/src/cryptolabs_proxy/vpm_prerequisite.py @@ -0,0 +1,100 @@ +"""Fresh, secret-safe readiness check for enabling Vast Price Manager.""" + +import json +import subprocess + + +EXPORTER_CONTAINER = "vastai-exporter" +PROBE_TIMEOUT_SECONDS = 15 + +# The command intentionally executes inside the exporter. That is the only +# place the management token is read; stdout is limited to a connected count. +_EXPORTER_PROBE = r""" +import json +import os +from urllib.error import HTTPError, URLError +from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener + +token = os.environ.get("MGMT_TOKEN") +if not token: + raise SystemExit(1) +request = Request( + "http://127.0.0.1:8622/api/accounts", + headers={"X-Mgmt-Token": token}, +) +try: + class NoRedirect(HTTPRedirectHandler): + def redirect_request(self, request, fp, code, msg, headers, newurl): + return None + + # Ignore inherited HTTP(S)_PROXY values and reject redirects: the token is + # valid only for the exporter's loopback management API. + opener = build_opener(ProxyHandler({}), NoRedirect()) + with opener.open(request, timeout=5) as response: + if response.status != 200: + raise SystemExit(1) + payload = json.load(response) +except (HTTPError, URLError, OSError, ValueError): + raise SystemExit(1) + +accounts = payload.get("accounts") if isinstance(payload, dict) else None +if not isinstance(accounts, list): + raise SystemExit(1) +count = sum( + 1 for account in accounts + if isinstance(account, dict) and account.get("status") == "connected" +) +print(json.dumps({"connected_account_count": count})) +""" + + +def _result(configured: bool, reason: str, count: int = 0) -> dict: + """Return the sole public prerequisite representation.""" + return { + "configured": configured, + "reason": reason, + "connected_account_count": count, + } + + +def get_vpm_prerequisite() -> dict: + """Check exporter availability and connected accounts without exposing secrets. + + This function is deliberately uncached. Lifecycle callers use its current + result immediately before their first VPM registration or custom enable. + """ + try: + running = subprocess.run( + ["docker", "inspect", "--format", "{{.State.Running}}", EXPORTER_CONTAINER], + capture_output=True, + text=True, + timeout=PROBE_TIMEOUT_SECONDS, + ) + except (OSError, subprocess.SubprocessError): + return _result(False, "exporter-not-running") + + if running.returncode != 0 or running.stdout.strip().lower() != "true": + return _result(False, "exporter-not-running") + + try: + probe = subprocess.run( + ["docker", "exec", EXPORTER_CONTAINER, "python3", "-c", _EXPORTER_PROBE], + capture_output=True, + text=True, + timeout=PROBE_TIMEOUT_SECONDS, + ) + except (OSError, subprocess.SubprocessError): + return _result(False, "unavailable") + + if probe.returncode != 0: + return _result(False, "unavailable") + try: + payload = json.loads(probe.stdout) + count = payload["connected_account_count"] + except (TypeError, ValueError, KeyError): + return _result(False, "unavailable") + if isinstance(count, bool) or not isinstance(count, int) or count < 0: + return _result(False, "unavailable") + if count == 0: + return _result(False, "no-connected-account") + return _result(True, "ready", count) diff --git a/tests/test_offline_image.py b/tests/test_offline_image.py new file mode 100644 index 0000000..cd1f7d5 --- /dev/null +++ b/tests/test_offline_image.py @@ -0,0 +1,96 @@ +"""Offline immutable local image-ID coverage for custom proxy migration.""" + +from pathlib import Path + +import pytest + +from cryptolabs_proxy.migration import ( + CustomConfigMigrator, + MigrationError, + build_recreate_request, +) + + +LOCAL_ID = "sha256:" + "b" * 64 + + +def test_migration_plan_accepts_full_canonical_local_image_id(monkeypatch, tmp_path: Path): + class Engine: + def inspect(self, _container): + return {"Id": "source-id"} + + migrator = CustomConfigMigrator(tmp_path, tmp_path / "backups", engine=Engine()) + monkeypatch.setattr(migrator, "_read_active_config", lambda _container: b"baseline") + + assert migrator.plan("cryptolabs-proxy", LOCAL_ID).image == LOCAL_ID + + +@pytest.mark.parametrize("image", ["sha256:" + "a" * 63, "sha256:" + "A" * 64]) +def test_migration_plan_rejects_truncated_or_noncanonical_local_image_ids(image, tmp_path: Path): + migrator = CustomConfigMigrator(tmp_path, tmp_path / "backups", engine=object()) + + with pytest.raises(MigrationError, match="immutable"): + migrator.plan("cryptolabs-proxy", image) + + +def test_candidate_validation_and_create_request_keep_exact_local_image_id(monkeypatch, tmp_path: Path): + commands = [] + + class Result: + returncode = 0 + + def run(command, **_kwargs): + commands.append(command) + if command[:4] == ["docker", "image", "inspect", "--format"]: + return type("InspectResult", (), {"returncode": 0, "stdout": f"{LOCAL_ID}\n", "stderr": ""})() + return Result() + + monkeypatch.setattr("cryptolabs_proxy.migration.subprocess.run", run) + candidate = tmp_path / "nginx.conf" + candidate.write_text("events {}") + inspect = {"Mounts": [{"Destination": "/etc/nginx/ssl", "Source": "/ssl"}]} + + CustomConfigMigrator(tmp_path, tmp_path / "backups")._validate_candidate( + LOCAL_ID, candidate, inspect, "offline-id" + ) + request = build_recreate_request( + {"Config": {}, "HostConfig": {}, "Mounts": [], "NetworkSettings": {"Networks": {"cryptolabs": {}}}}, + LOCAL_ID, + str(candidate), + ) + + assert commands[0] == ["docker", "image", "inspect", "--format", "{{.Id}}", LOCAL_ID] + assert LOCAL_ID in commands[1] + assert request["Config"]["Image"] == LOCAL_ID + + +@pytest.mark.parametrize( + "inspect_result", + [ + (1, "", "No such image"), + (0, "sha256:" + "c" * 64 + "\n", ""), + ], +) +def test_local_image_id_missing_or_mismatched_rejects_before_candidate_validator_runs(monkeypatch, tmp_path: Path, inspect_result): + commands = [] + returncode, stdout, stderr = inspect_result + + class Result: + def __init__(self, returncode, stdout="", stderr=""): + self.returncode, self.stdout, self.stderr = returncode, stdout, stderr + + def run(command, **_kwargs): + commands.append(command) + return Result(returncode, stdout, stderr) + + monkeypatch.setattr("cryptolabs_proxy.migration.subprocess.run", run) + candidate = tmp_path / "nginx.conf" + candidate.write_text("events {}") + inspect = {"Mounts": [{"Destination": "/etc/nginx/ssl", "Source": "/ssl"}]} + + with pytest.raises(MigrationError, match="offline image missing"): + CustomConfigMigrator(tmp_path, tmp_path / "backups")._validate_candidate( + LOCAL_ID, candidate, inspect, "offline-id" + ) + + assert commands == [["docker", "image", "inspect", "--format", "{{.Id}}", LOCAL_ID]] diff --git a/tests/test_routes.py b/tests/test_routes.py index 3f88a23..9812ba4 100644 --- a/tests/test_routes.py +++ b/tests/test_routes.py @@ -1,6 +1,9 @@ """Tests for Flask routes: login/logout, user management, settings, API endpoints.""" +import hashlib +import hmac import json +from datetime import timedelta import pytest @@ -111,6 +114,157 @@ def test_logout_clears_session(self, logged_in_admin): assert 'logged_in' not in sess +# --------------------------------------------------------------------------- +# Vast Price Manager Fleet session authority +# --------------------------------------------------------------------------- + +class TestVastPriceManagerFleetSession: + def _login(self, client, admin_user): + response = client.post('/auth/login', data={ + 'username': admin_user['username'], + 'password': admin_user['password'], + }) + assert response.status_code == 302 + + def _vpm_session(self, client): + """Model VPM's direct call without forwarding Fleet's Set-Cookie back.""" + incoming_cookie = client.get_cookie('fleet_session') + response = client.get('/auth/vast-price-manager/session') + if incoming_cookie: + client.set_cookie('fleet_session', incoming_cookie.value) + return response + + def test_session_uses_real_fleet_cookie_and_returns_only_contract_fields(self, client, admin_user): + import cryptolabs_proxy.auth as auth + + self._login(client, admin_user) + signed_cookie = client.get_cookie('fleet_session').value.encode('utf-8') + + response = self._vpm_session(client) + + assert response.status_code == 200 + assert response.get_json() == { + 'authenticated': True, + 'username': admin_user['username'], + 'role': 'admin', + 'subject': response.get_json()['subject'], + 'csrf_token': response.get_json()['csrf_token'], + } + payload = response.get_json() + assert len(payload['subject']) == 64 + assert len(payload['csrf_token']) == 64 + assert payload['subject'] != payload['csrf_token'] + assert all(character in '0123456789abcdef' for character in payload['subject']) + assert all(character in '0123456789abcdef' for character in payload['csrf_token']) + assert payload['subject'] == hmac.new( + auth.AUTH_SECRET_KEY.encode('utf-8'), + b'cryptolabs/vpm/subject/v1:' + signed_cookie, + hashlib.sha256, + ).hexdigest() + assert payload['csrf_token'] == hmac.new( + auth.AUTH_SECRET_KEY.encode('utf-8'), + b'cryptolabs/vpm/csrf/v1:' + signed_cookie, + hashlib.sha256, + ).hexdigest() + assert b'fleet_session=' not in response.data + assert b'password' not in response.data + + def test_session_rechecks_current_user_state_instead_of_cached_session_role(self, client, admin_user): + import cryptolabs_proxy.auth as auth + + self._login(client, admin_user) + auth.update_user(admin_user['username'], role='readonly') + + assert client.get('/auth/vast-price-manager/session').status_code == 403 + assert client.get('/auth/vast-price-manager/authorize').status_code == 403 + + @pytest.mark.parametrize('change', [ + lambda auth, username: auth.update_user(username, enabled=False), + lambda auth, username: auth.update_user(username, require_password_change=True), + lambda auth, username: auth.delete_user(username), + ]) + def test_session_rejects_disabled_forced_change_or_deleted_user(self, client, admin_user, change): + import cryptolabs_proxy.auth as auth + + # Keep a second admin so the delete operation is valid. + auth.create_user('other-admin', 'otherpass', role='admin') + self._login(client, admin_user) + change(auth, admin_user['username']) + + expected = 401 if auth.get_user(admin_user['username']) and auth.get_user(admin_user['username']).get('require_password_change') else 403 + assert client.get('/auth/vast-price-manager/session').status_code == expected + + def test_session_requires_a_signed_logged_in_fleet_session(self, client, admin_user): + assert client.get('/auth/vast-price-manager/session').status_code == 401 + client.set_cookie('fleet_session', 'forged-session') + assert client.get('/auth/vast-price-manager/session').status_code == 401 + + def test_session_rejects_an_expired_signed_fleet_session(self, client, admin_user): + self._login(client, admin_user) + client.application.config['PERMANENT_SESSION_LIFETIME'] = timedelta(seconds=-1) + + assert client.get('/auth/vast-price-manager/session').status_code == 401 + + def test_reauth_validates_csrf_and_current_fleet_password(self, client, admin_user): + self._login(client, admin_user) + session = self._vpm_session(client).get_json() + + invalid_csrf = client.post('/auth/vast-price-manager/reauth', json={ + 'password': admin_user['password'], 'csrf_token': '0' * 64, + }) + assert invalid_csrf.status_code == 403 + + wrong_password = client.post('/auth/vast-price-manager/reauth', json={ + 'password': 'incorrect', 'csrf_token': session['csrf_token'], + }) + assert wrong_password.status_code == 401 + + session = self._vpm_session(client).get_json() + valid = client.post('/auth/vast-price-manager/reauth', json={ + 'password': admin_user['password'], 'csrf_token': session['csrf_token'], + }) + assert valid.status_code == 200 + assert valid.get_json() == session + + def test_reauth_rejects_unbounded_or_non_json_input(self, client, admin_user): + self._login(client, admin_user) + session = client.get('/auth/vast-price-manager/session').get_json() + + assert client.post('/auth/vast-price-manager/reauth', data='not-json').status_code == 400 + assert client.post('/auth/vast-price-manager/reauth', json={ + 'password': 'x' * 4097, 'csrf_token': session['csrf_token'], + }).status_code == 400 + + def test_reauth_reports_detectable_password_throttling(self, client, admin_user): + import cryptolabs_proxy.auth as auth + + auth.save_settings({'max_login_attempts': 3, 'lockout_duration_minutes': 15}) + self._login(client, admin_user) + for _ in range(2): + csrf_token = self._vpm_session(client).get_json()['csrf_token'] + assert client.post('/auth/vast-price-manager/reauth', json={ + 'password': 'incorrect', 'csrf_token': csrf_token, + }).status_code == 401 + + csrf_token = self._vpm_session(client).get_json()['csrf_token'] + limited = client.post('/auth/vast-price-manager/reauth', json={ + 'password': 'incorrect', 'csrf_token': csrf_token, + }) + assert limited.status_code == 429 + + def test_reauth_rechecks_live_user_state(self, client, admin_user): + import cryptolabs_proxy.auth as auth + + self._login(client, admin_user) + csrf_token = self._vpm_session(client).get_json()['csrf_token'] + auth.update_user(admin_user['username'], enabled=False) + + response = client.post('/auth/vast-price-manager/reauth', json={ + 'password': admin_user['password'], 'csrf_token': csrf_token, + }) + assert response.status_code == 403 + + # --------------------------------------------------------------------------- # Change password route # --------------------------------------------------------------------------- diff --git a/tests/test_updates.py b/tests/test_updates.py new file mode 100644 index 0000000..b49cfb3 --- /dev/null +++ b/tests/test_updates.py @@ -0,0 +1,358 @@ +"""Updater failures must not destroy existing runtime state or conceal outcomes.""" +import importlib.util +import json +import subprocess +from copy import deepcopy +from pathlib import Path + +import pytest + +from cryptolabs_proxy import updates + + +def source(): + return { + 'Id': 'a' * 64, 'Name': '/prometheus', 'Image': 'sha256:old', + 'Config': { + 'Image': 'prom/prometheus:latest', 'Env': ['GIT_COMMIT=old', 'SECRET=keep', 'PATH=/old'], + 'Labels': {'org.opencontainers.image.revision': 'old', 'custom': 'keep'}, + 'Cmd': ['--web.route-prefix=/prometheus/', '--storage.tsdb.retention.time=90d'], + 'Entrypoint': ['/custom-entrypoint'], 'User': '123', + 'Healthcheck': {'Test': ['CMD', 'probe'], 'Interval': 10}, + }, + 'HostConfig': {'Binds': [], 'RestartPolicy': {'Name': 'on-failure', 'MaximumRetryCount': 3}, + 'PortBindings': {'999/udp': [{'HostIp': '127.0.0.1', 'HostPort': '999'}]}, + 'LogConfig': {'Type': 'json-file', 'Config': {'max-size': '10m'}}}, + 'Mounts': [{'Type': 'volume', 'Name': 'metrics', 'Destination': '/prometheus', 'RW': False}, + {'Type': 'bind', 'Source': '/site/nginx.conf', 'Destination': '/etc/nginx/nginx.conf', 'RW': False}], + 'NetworkSettings': {'Networks': {'cryptolabs': { + 'Aliases': ['prometheus', 'a' * 12], 'IPAMConfig': {'IPv4Address': '172.30.0.10'}, + 'IPAddress': '172.30.0.10', 'EndpointID': 'runtime', + }}}, + 'State': {'Running': True, 'Health': {'Status': 'healthy'}}, + } + + +def test_clone_preserves_runtime_but_refreshes_image_defaults(): + original = source() + old = {'Config': {'Env': ['GIT_COMMIT=old', 'PATH=/old'], 'Cmd': ['default'], + 'Entrypoint': ['/image-entrypoint'], 'Labels': {'org.opencontainers.image.revision': 'old'}}} + new = {'Id': 'sha256:new', 'Config': {'Env': ['GIT_COMMIT=new', 'PATH=/new'], + 'Labels': {'org.opencontainers.image.revision': 'new'}, 'Cmd': ['new-default']}} + request = updates.build_update_request(original, old, new) + assert original == source() + assert request['Config']['Image'] == 'sha256:new' + assert request['Config']['Env'] == ['GIT_COMMIT=new', 'PATH=/new', 'SECRET=keep'] + assert request['Config']['Labels'] == {'org.opencontainers.image.revision': 'new', 'custom': 'keep'} + for key in ('Cmd', 'Entrypoint', 'User', 'Healthcheck'): + assert request['Config'][key] == original['Config'][key] + assert request['HostConfig']['RestartPolicy'] == original['HostConfig']['RestartPolicy'] + assert request['HostConfig']['PortBindings'] == original['HostConfig']['PortBindings'] + assert request['HostConfig']['Mounts'][0]['Source'] == 'metrics' + assert request['HostConfig']['Mounts'][0]['ReadOnly'] is True + endpoint = request['NetworkingConfig']['EndpointsConfig']['cryptolabs'] + assert endpoint['IPAMConfig']['IPv4Address'] == '172.30.0.10' + assert endpoint['Aliases'] == ['prometheus'] + assert 'EndpointID' not in endpoint + + +def test_image_defaults_change_without_losing_custom_overrides(): + original = source() + original['Config']['Cmd'] = ['old-default'] + old = {'Config': {'Cmd': ['old-default']}} + new = {'Id': 'sha256:new', 'Config': {'Cmd': ['new-default']}} + assert updates.build_update_request(original, old, new)['Config']['Cmd'] == ['new-default'] + + +@pytest.mark.parametrize('reference', ['sha256:123', 'repo@sha256:123', 'prom/prometheus:v3.2', 'custom/prometheus:latest']) +def test_pinned_and_custom_images_are_never_silently_retargeted(reference): + current = source() + current['Config']['Image'] = reference + assert updates.target_for('prometheus', current, 'dev') is None + + +def test_channel_only_changes_first_party_images(): + assert updates.target_for('prometheus', source(), 'dev') == 'prom/prometheus:latest' + current = source() + current['Config']['Image'] = 'ghcr.io/cryptolabsza/dc-overview:latest' + assert updates.target_for('dc-overview', current, 'dev').endswith(':dev') + assert updates.target_for('vast-price-manager', current, 'main') is None + + +class Engine: + def __init__(self, failure=None): + self.current = source() + self.containers = {'prometheus': self.current} + self.calls = [] + self.failure = failure + + def inspect(self, name): + if self.failure == 'inspect': + raise RuntimeError('private detail') + if name not in self.containers: + raise updates.MigrationError('Docker Engine GET returned HTTP 404') + return deepcopy(self.containers[name]) + + def image(self, image): + return {'Id': 'sha256:old' if image == 'sha256:old' else 'sha256:new', 'Config': {}} + + def stop(self, name, **kwargs): + self.calls.append(('stop', name)) + self.containers[name]['State']['Running'] = False + + def disconnect(self, name, network): + self.calls.append(('disconnect', name, network)) + + def connect(self, name, network, endpoint): + self.calls.append(('connect', name, network)) + + def rename(self, name, new): + self.calls.append(('rename', name, new)) + self.containers[new] = self.containers.pop(name) + + def create(self, name, request): + self.calls.append(('create', name)) + if self.failure == 'create': + raise RuntimeError('private detail') + self.containers[name] = deepcopy(source()) + self.containers[name]['Config'] = request['Config'] + self.containers[name]['Image'] = request['Config']['Image'] + + def start(self, name): + self.calls.append(('start', name)) + current = self.containers[name] + current['State']['Running'] = True + if self.failure == 'health' and current['Image'] == 'sha256:new': + current['State']['Health']['Status'] = 'unhealthy' + + def remove(self, name, **kwargs): + self.calls.append(('remove', name)) + self.containers.pop(name, None) + + +@pytest.mark.parametrize('failure', ['create', 'health']) +def test_failed_update_restores_original_and_reports_failure(failure): + engine = Engine(failure) + result = updates.replace_container(engine, 'prometheus', 'sha256:new', 'job', timeout=0) + assert result['success'] is False + assert result['state'] == 'rolled_back' + assert engine.inspect('prometheus')['Image'] == 'sha256:old' + assert engine.inspect('prometheus')['State']['Running'] is True + assert 'private detail' not in json.dumps(result) + + +def test_inspection_failure_never_stops_original(): + engine = Engine('inspect') + result = updates.replace_container(engine, 'prometheus', 'sha256:new', 'job', timeout=0) + assert result['success'] is False + assert not engine.calls + + +def test_success_retains_old_container_and_verifies_new_identity(): + engine = Engine() + result = updates.replace_container(engine, 'prometheus', 'sha256:new', 'job', timeout=0) + assert result['state'] == 'completed' + assert result['image_id'] == 'sha256:new' + assert engine.inspect('prometheus.rollback-job')['Image'] == 'sha256:old' + assert not engine.inspect('prometheus.rollback-job')['State']['Running'] + + +def test_same_tag_image_identity_detects_update_and_unknown_is_not_current(): + engine = Engine() + status = updates.update_status('prometheus', 'main', engine=engine) + assert status['update_available'] is True + engine.image = lambda image: {'Id': 'sha256:old'} + assert updates.update_status('prometheus', 'main', engine=engine)['update_available'] is False + engine.image = lambda image: (_ for _ in ()).throw(RuntimeError()) + assert updates.update_status('prometheus', 'main', engine=engine)['update_available'] is None + + +def test_worker_does_not_touch_stopped_pinned_or_vpm(monkeypatch, tmp_path): + engine = Engine() + engine.current['State']['Running'] = False + monkeypatch.setattr(updates, 'SERVICES', {'prometheus': updates.SERVICES['prometheus'], + 'vast-price-manager': updates.SERVICES['vast-price-manager']}) + job = {'id': 'a' * 32, 'service': 'all', 'branch': 'main', 'action': 'update'} + updates.write_job(tmp_path, job) + result = updates.run_job(tmp_path, job['id'], engine=engine) + assert not engine.calls + assert result['results']['prometheus']['state'] == 'skipped' + assert 'vast-price-manager' not in result['results'] + + +def test_helper_runs_independently_with_persistent_state_and_no_copied_secrets(): + current = source() + current['Mounts'].append({'Type': 'volume', 'Name': 'fleet-data', 'Destination': '/data', 'RW': True}) + request = updates.helper_request(current, Path('/data/auth/update-jobs'), 'a' * 32) + assert request['Config']['Image'] == 'sha256:old' + assert request['Config']['Entrypoint'] == ['python3'] + assert request['Config']['Cmd'][:2] == ['-m', 'cryptolabs_proxy.updates'] + assert request['HostConfig']['NetworkMode'] == 'host' + assert request['Config']['Env'] == ['PYTHONPATH=/app/src'] + assert any(m.get('Source') == 'fleet-data' for m in request['HostConfig']['Mounts']) + assert request['Config']['Healthcheck'] == {'Test': ['NONE']} + + +def test_update_api_queues_job_instead_of_mutating_docker(monkeypatch): + path = Path(__file__).parents[1] / 'scripts' / 'health-api.py' + spec = importlib.util.spec_from_file_location('health_api_updates', path) + api = importlib.util.module_from_spec(spec) + spec.loader.exec_module(api) + import io + body = json.dumps({'service': 'prometheus', 'branch': 'main'}).encode() + handler = object.__new__(api.HealthHandler) + handler.path = '/api/update' + handler.headers = {'Content-Length': str(len(body))} + handler.rfile = io.BytesIO(body) + replies = [] + handler.send_json = lambda result, status=200: replies.append((result, status)) + monkeypatch.setattr(api, 'submit_update_job', lambda *args, **kwargs: {'id': 'job', 'state': 'queued'}) + handler.do_POST() + assert replies == [({'success': True, 'job': {'id': 'job', 'state': 'queued'}}, 202)] + + +def test_cli_submit_and_status_return_machine_readable_jobs(monkeypatch, capsys): + monkeypatch.setattr(updates, 'submit_job', lambda *args, **kwargs: {'id': 'a' * 32, 'state': 'queued'}) + assert updates.main(['submit', '--branch', 'dev']) == 0 + assert json.loads(capsys.readouterr().out)['state'] == 'queued' + monkeypatch.setattr(updates, 'job_status', lambda *args, **kwargs: {'id': 'a' * 32, 'state': 'failed', 'success': False}) + assert updates.main(['status', 'a' * 32]) == 0 + assert json.loads(capsys.readouterr().out)['success'] is False + + +def test_bulk_missing_service_is_skipped_and_partial_failure_is_reported(monkeypatch, tmp_path): + engine = Engine() + monkeypatch.setattr(updates, 'pull_target', lambda target: False) + monkeypatch.setattr(updates, 'SERVICES', {name: updates.SERVICES[name] for name in ['prometheus', 'runpod-exporter']}) + job = {'id': 'a' * 32, 'service': 'all', 'branch': 'main', 'action': 'update'} + updates.write_job(tmp_path, job) + result = updates.run_job(tmp_path, job['id'], engine=engine) + assert result['success'] is False + assert result['results']['prometheus']['success'] is False + assert result['results']['runpod-exporter']['state'] == 'skipped' + assert not engine.calls + + +def test_new_managed_image_is_not_misclassified_as_pinned(): + engine = Engine() + result = updates.replace_container(engine, 'prometheus', 'prom/prometheus:latest', 'job', timeout=0) + assert result['success'] is True + assert updates.target_for('prometheus', engine.inspect('prometheus'), 'main') == 'prom/prometheus:latest' + + +def test_no_healthcheck_requires_real_application_probe(monkeypatch): + engine = Engine() + engine.current['State'].pop('Health') + monkeypatch.setattr(updates, '_probe', lambda *args: False) + assert updates.wait_ready(engine, 'prometheus', 'sha256:old', timeout=0) is False + + +def test_duplicate_submission_does_not_launch_second_helper(monkeypatch, tmp_path): + engine = Engine() + job = {'id': 'a' * 32, 'state': 'running', 'helper': 'prometheus'} + updates.write_job(tmp_path, job) + (tmp_path / 'active.json').write_text(json.dumps({'id': job['id']})) + with pytest.raises(updates.UpdateBusy): + updates.submit_job('all', 'main', 'update', directory=tmp_path, engine=engine) + assert not engine.calls + + +def test_missing_helper_is_recorded_as_interrupted_not_left_active(tmp_path): + job = {'id': 'a' * 32, 'state': 'running', 'helper': 'missing-helper'} + updates.write_job(tmp_path, job) + result = updates.job_status(job['id'], directory=tmp_path, engine=Engine()) + assert result['state'] == 'interrupted' + assert result['success'] is False + assert 'stopped before completion' in result['message'] + assert updates.read_job(tmp_path, job['id'])['state'] == 'interrupted' + + +def test_submit_recovers_a_missing_previous_helper(tmp_path): + engine = Engine() + engine.current['Mounts'].append({ + 'Type': 'volume', 'Name': 'fleet-data', 'Destination': str(tmp_path.parent), 'RW': True, + }) + engine.containers['cryptolabs-proxy'] = engine.current + previous = {'id': 'a' * 32, 'state': 'running', 'helper': 'missing-helper'} + updates.write_job(tmp_path, previous) + (tmp_path / 'active.json').write_text(json.dumps({'id': previous['id']})) + + submitted = updates.submit_job('all', 'main', 'pull', directory=tmp_path, engine=engine) + + assert submitted['id'] != previous['id'] + assert updates.read_job(tmp_path, previous['id'])['state'] == 'interrupted' + assert ('create', submitted['helper']) in engine.calls + + +def test_submit_keeps_job_busy_when_docker_cannot_confirm_helper_state(tmp_path): + previous = {'id': 'a' * 32, 'state': 'running', 'helper': 'missing-helper'} + updates.write_job(tmp_path, previous) + (tmp_path / 'active.json').write_text(json.dumps({'id': previous['id']})) + + with pytest.raises(updates.UpdateBusy): + updates.submit_job('all', 'main', 'pull', directory=tmp_path, engine=Engine('inspect')) + + +def test_update_engine_uses_docker_29_compatible_api(monkeypatch): + calls = [] + + def request(self, method, path, payload=None, allowed=(200, 201, 204)): + calls.append((method, path, payload, allowed)) + return b'{}' + + monkeypatch.setattr(updates.DockerEngine, '_request', request) + updates.UpdateEngine()._request('GET', '/containers/example/json') + assert calls[0][1] == '/v1.45/containers/example/json' + + +def landing_script(): + html = (Path(__file__).parents[1] / 'landing-page/index.html').read_text() + return html.rsplit('