""" Pinokio registry check-in endpoint. When a user clicks "Check in" in Pinokio, it opens: http://localhost:/registry/checkin?return=&app=®istry=&...#token= The #token is in the URL hash — it never reaches Flask. Client-side JS reads it, calls POST /checkpoints/snapshot with X-Registry-Token header, which posts the token + current git hash to api.pinokio.co, then redirects back to beta.pinokio.co. """ import json import os import subprocess import time import requests from flask import Blueprint, Response, jsonify, request registry_bp = Blueprint('registry', __name__) @registry_bp.route('/registry/checkin') def registry_checkin(): return_url = request.args.get('return', '') repo = request.args.get('repo', 'OpenVoiceUI') app_slug = request.args.get('app', '') registry = request.args.get('registry', 'https://api.pinokio.co') # Sanitize return_url — only allow http/https redirects if return_url and not return_url.startswith(('http://', 'https://')): return_url = '' # Quick health check try: from services.health import health_checker readiness = health_checker.readiness() is_healthy = readiness.healthy status_msg = readiness.message uptime = int(time.time() - health_checker.start_time) uptime_str = f"{uptime // 3600}h {(uptime % 3600) // 60}m {uptime % 60}s" except Exception: is_healthy = True status_msg = "Server is running" uptime_str = "unknown" status_icon = "✔" if is_healthy else "⚠" status_color = "#4ade80" if is_healthy else "#fbbf24" # HTML-escape user-controlled params before interpolating into the page # (SEC-4 reflected XSS: `repo` was interpolated raw into the HTML below). import markupsafe repo_safe = markupsafe.escape(repo) # Pass values to JS safely via JSON encoding. Escape '<' so a `repo` # containing '' can't break out of the inline script tag. js_config = json.dumps({ 'returnUrl': return_url, 'registry': registry, 'appSlug': app_slug, 'repo': repo, }).replace('<', '\\u003c') html = f''' OpenVoiceUI — Install Verified

{status_icon}

OpenVoiceUI is Running

Install verified and healthy

Server {status_msg}

Uptime {uptime_str}

Repository {repo_safe}

Completing check-in…

''' return Response(html, content_type='text/html') def _get_git_sha(repo_url: str) -> str: """Get the real git commit SHA — try local git first, then GitHub API.""" app_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # 1. Local git (works in dev/host installs) try: return subprocess.check_output( ['git', 'rev-parse', 'HEAD'], cwd=app_root, stderr=subprocess.DEVNULL ).decode().strip() except Exception: pass # 2. GIT_COMMIT env var baked in at Docker build time env_sha = os.getenv('GIT_COMMIT', '').strip() if env_sha: return env_sha # 3. GIT_HASH file written during Docker build hash_file = os.path.join(app_root, 'GIT_HASH') if os.path.exists(hash_file): return open(hash_file).read().strip() # 4. Fetch from GitHub API — repo_url like https://github.com/OWNER/REPO try: parts = repo_url.rstrip('/').rstrip('.git').split('/') if len(parts) >= 2: owner, repo = parts[-2], parts[-1] resp = requests.get( f'https://api.github.com/repos/{owner}/{repo}/commits/main', headers={'Accept': 'application/vnd.github.sha'}, timeout=8, ) if resp.ok: return resp.text.strip() except Exception: pass return '' def _compute_checkpoint_hash(repo_url: str, git_sha: str) -> str: """ Compute the Pinokio checkpoint hash exactly as pinokiod does. Format: "sha256:" + SHA256(stableStringify({version, root, repos[]})) stableStringify = JSON with keys sorted alphabetically at every level. """ import hashlib def stable_stringify(obj) -> str: if isinstance(obj, dict): pairs = ','.join( f'{json.dumps(k)}:{stable_stringify(v)}' for k, v in sorted(obj.items()) ) return '{' + pairs + '}' if isinstance(obj, list): return '[' + ','.join(stable_stringify(v) for v in obj) + ']' return json.dumps(obj) # canonicalRepoUrl() in pinokiod strips protocol, then lowercases the full URL canonical_url = repo_url.strip().rstrip('/') if canonical_url.lower().endswith('.git'): canonical_url = canonical_url[:-4] canonical_url = canonical_url.lower() canonical = { 'version': 1, 'root': canonical_url, 'repos': [{'commit': git_sha, 'path': '.', 'repo': canonical_url}], } serialized = stable_stringify(canonical) digest = hashlib.sha256(serialized.encode()).hexdigest() return f'sha256:{digest}' @registry_bp.route('/checkpoints/snapshot', methods=['POST']) def checkpoints_snapshot(): """ Called by the /registry/checkin page JS. Creates a git snapshot of the app and publishes it to api.pinokio.co. """ token = request.headers.get('X-Registry-Token', '').strip() registry = request.args.get('registry', 'https://api.pinokio.co').rstrip('/') publish = request.args.get('publish', '0') == '1' repo_url = request.args.get('repo', 'https://github.com/MCERQUA/OpenVoiceUI') app_slug = request.args.get('app', 'github-com-mcerqua-openvoiceui') if not token: return jsonify({'ok': False, 'error': 'missing_token'}), 400 # Get real git SHA, then compute the proper Pinokio checkpoint hash git_sha = _get_git_sha(repo_url) if not git_sha: return jsonify({'ok': False, 'error': 'snapshot_failed', 'detail': 'Could not determine git commit SHA'}), 500 checkpoint_hash = _compute_checkpoint_hash(repo_url, git_sha) created = {'hash': checkpoint_hash} if not publish: return jsonify({'ok': True, 'created': created}) # Publish to Pinokio registry with the exact body format pinokiod uses. # The checkpoint URLs must be lowercased to match the registry's canonicalRepoUrl(). import platform as _platform canon_url = repo_url.strip().rstrip('/') if canon_url.lower().endswith('.git'): canon_url = canon_url[:-4] canon_url = canon_url.lower() post_body = { 'hash': checkpoint_hash, 'visibility': 'public', 'checkpoint': { 'version': 1, 'root': canon_url, 'repos': [{'commit': git_sha, 'path': '.', 'repo': canon_url}], }, 'system': { 'platform': _platform.system().lower(), 'arch': _platform.machine(), }, } try: resp = requests.post( f'{registry}/checkpoints', headers={ 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json', }, json=post_body, timeout=15, ) except requests.RequestException as e: return jsonify({'ok': False, 'error': 'publish_failed', 'detail': str(e)}), 502 if not resp.ok: try: detail = resp.json() except Exception: detail = resp.text[:500] import logging logging.getLogger(__name__).warning( f'Pinokio registry publish failed: HTTP {resp.status_code} — {detail} ' f'(our_sha={git_sha[:12]}, our_hash={checkpoint_hash})' ) return jsonify({'ok': False, 'error': 'publish_failed', 'detail': detail, 'status': resp.status_code, '_debug': {'sha': git_sha, 'hash': checkpoint_hash, 'canon_url': canon_url}}), 502 try: pub_data = resp.json() except Exception: pub_data = {} return jsonify({ 'ok': True, 'created': created, 'publish': {'ok': True, 'hash': checkpoint_hash, **pub_data}, }) @registry_bp.route('/registry/debug') def registry_debug(): """ Debug endpoint — visit http://localhost:/registry/debug to see what SHA and checkpoint hash this server would compute for the registry check-in. Compare with what Pinokio computed locally. """ repo_url = request.args.get('repo', 'https://github.com/MCERQUA/OpenVoiceUI') import platform as _platform git_sha = _get_git_sha(repo_url) canon_url = repo_url.strip().rstrip('/') if canon_url.lower().endswith('.git'): canon_url = canon_url[:-4] canon_url = canon_url.lower() checkpoint_hash = _compute_checkpoint_hash(repo_url, git_sha) if git_sha else '' import json as _json # Show what the POST body would look like post_body = { 'hash': checkpoint_hash, 'visibility': 'public', 'checkpoint': { 'version': 1, 'root': canon_url, 'repos': [{'commit': git_sha, 'path': '.', 'repo': canon_url}], }, 'system': { 'platform': _platform.system().lower(), 'arch': _platform.machine(), }, } import os as _os app_root = _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))) hash_file = _os.path.join(app_root, 'GIT_HASH') git_hash_file_contents = '' if _os.path.exists(hash_file): git_hash_file_contents = open(hash_file).read().strip() return jsonify({ 'git_sha': git_sha, 'git_sha_source': 'git' if not git_sha else ( 'env' if _os.getenv('GIT_COMMIT', '').strip() == git_sha else ( 'file' if git_hash_file_contents == git_sha else 'github_api' ) ), 'git_hash_file': git_hash_file_contents, 'canonical_url': canon_url, 'checkpoint_hash': checkpoint_hash, 'post_body': post_body, })