feat: Hermes prompt callback endpoint + API key management

- Add /api/prompt/<rid> callback endpoint with dual auth:
  per-request signed URL token + Bearer API key.
- Add HERMES_API_KEY config and runtime key helpers in app.py.
- Update admin request page to copy request details + callback URL.
- Add API key management to admin settings: regenerate, mask, show-once.
- Update .env.example, docker-compose.yml, README, REVIEW docs.
This commit is contained in:
Troll (Hermes Agent) 2026-08-03 17:16:56 +00:00
parent e46893e0b4
commit 6a60aa686c
8 changed files with 242 additions and 35 deletions

154
app.py
View file

@ -45,8 +45,12 @@ from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import hmac
import hashlib
import secrets
# Flask and related imports
from flask import Flask, request, render_template, redirect, url_for, flash, session, send_from_directory, abort, current_app, send_file
from flask import Flask, request, render_template, redirect, url_for, flash, session, send_from_directory, abort, current_app, send_file, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from werkzeug.utils import secure_filename
@ -263,6 +267,82 @@ def get_refresh_seconds():
val = 10
return val if val in (10, 20, 30) else 10
def get_hermes_api_key():
"""
Return the effective Hermes API key.
Runtime settings override the environment variable so the key can be
regenerated from /admin/settings without redeploying.
"""
cfg = load_booth_settings()
return decrypt_value(cfg.get('hermes_api_key', '')) or current_app.config.get('HERMES_API_KEY', '')
def set_hermes_api_key(key):
"""Persist a new Hermes API key (encrypted) to runtime settings."""
cfg = load_booth_settings()
cfg['hermes_api_key'] = encrypt_value(key)
save_booth_settings(cfg)
def generate_hermes_api_key():
"""Generate a new random API key for Hermes callback authentication."""
return secrets.token_urlsafe(32)
def mask_api_key(key):
"""Return a masked version of the API key showing only the last 6 characters."""
if not key:
return 'Not set'
if len(key) <= 6:
return '*' * len(key)
return '*' * (len(key) - 6) + key[-6:]
def sign_prompt_callback(rid, expires_at=None):
"""
Create a signed callback token for a specific request ID.
The signature is HMAC-SHA256 over "rid:expires_at" using APP_SECRET_KEY.
Returns a URL-safe token string.
"""
secret = current_app.config['SECRET_KEY'].encode()
if expires_at is None:
# Default expiry: 1 hour.
expires_at = int(time.time()) + 3600
payload = f"{rid}:{expires_at}"
sig = hmac.new(secret, payload.encode(), hashlib.sha256).hexdigest()[:16]
return f"{rid}:{expires_at}:{sig}"
def verify_prompt_callback(token):
"""
Verify a signed callback token. Returns (rid, ok) tuple.
The token must match the HMAC signature and not be expired.
"""
if not token:
return None, False
parts = token.split(':')
if len(parts) != 3:
return None, False
try:
rid = int(parts[0])
expires_at = int(parts[1])
except ValueError:
return None, False
if int(time.time()) > expires_at:
return None, False
expected = sign_prompt_callback(rid, expires_at)
if not hmac.compare_digest(expected, token):
return None, False
return rid, True
def build_prompt_callback_url(rid):
"""Build the full callback URL an operator pastes into Hermes for a request."""
token = sign_prompt_callback(rid)
return f"{current_app.config['PUBLIC_BASE_URL']}/api/prompt/{rid}?token={token}"
def send_email(to, subject, body, attachments=None, inline_images=None):
"""Send an email using the configured or runtime SMTP settings."""
cfg = get_email_config()
@ -383,6 +463,59 @@ def request_form():
return render_template('request.html', form=None)
@app.route('/api/prompt/<int:rid>', methods=['POST'])
def api_update_prompt(rid):
"""
Hermes callback endpoint.
Accepts a JSON POST with generated Suno prompt fields and updates the
matching request. Two layers of auth:
1) A per-request signed callback token in the query string.
2) A Hermes API key in the Authorization header (Bearer).
Only records in 'pending' status can be updated. On success, status is set
to 'prompt_ready'.
"""
# Layer 1: verify the signed callback URL token.
callback_token = request.args.get('token', '').strip()
token_rid, token_ok = verify_prompt_callback(callback_token)
if not token_ok or token_rid != rid:
abort(401)
# Layer 2: verify the Hermes API key from the Authorization header.
expected_key = get_hermes_api_key()
if not expected_key:
abort(500, description='Hermes API key is not configured')
auth_header = request.headers.get('Authorization', '').strip()
if not auth_header.startswith('Bearer '):
abort(401)
provided_key = auth_header[7:].strip()
if not hmac.compare_digest(expected_key, provided_key):
abort(401)
req = get_request_by_id(rid)
if not req:
abort(404)
if req['status'] != 'pending':
abort(409, description='Request is no longer pending')
data = request.get_json(silent=True) or {}
# Optional email verification to make sure clipboard matches record.
provided_email = data.get('email', '').strip().lower()
if provided_email and provided_email != req['email'].lower():
abort(400, description='Email mismatch')
update_request(rid,
suno_title=data.get('suno_title', '').strip(),
suno_style=data.get('suno_style', '').strip(),
suno_lyrics=data.get('suno_lyrics', '').strip(),
status='prompt_ready'
)
return jsonify({'ok': True, 'request_id': rid, 'status': 'prompt_ready'}), 200
@app.route('/thanks/<int:rid>')
def thanks(rid):
"""Confirmation page shown after a customer submits a request."""
@ -692,7 +825,7 @@ def admin_request(rid):
return redirect(url_for('admin_request', rid=rid))
return render_template('admin/request.html', req=req, statuses=STATUS_LABELS, file_exists=file_exists, basename=basename, extra_files=extra_files)
return render_template('admin/request.html', req=req, statuses=STATUS_LABELS, file_exists=file_exists, basename=basename, extra_files=extra_files, callback_url=build_prompt_callback_url(rid))
@app.route('/admin/request/<int:rid>/delete', methods=['POST'])
@ -765,6 +898,12 @@ def admin_settings():
'smtp_pass_set': bool(runtime_settings.get('smtp_pass', '')),
}
# Hermes API key state for the settings page.
hermes_key = get_hermes_api_key()
hermes_key_masked = mask_api_key(hermes_key)
hermes_key_set = bool(hermes_key)
hermes_key_just_generated = session.pop('hermes_key_just_generated', None)
# Compute upload folder stats.
total_upload_size = 0
upload_file_count = 0
@ -948,6 +1087,14 @@ def admin_settings():
flash(f'Database restore failed: {e}', 'error')
return redirect(url_for('admin_settings'))
elif action == 'regenerate_hermes_key':
# Generate a new Hermes API key, persist it, and show it exactly once.
new_key = generate_hermes_api_key()
set_hermes_api_key(new_key)
session['hermes_key_just_generated'] = new_key
flash('Hermes API key regenerated. Copy it now — it will not be shown again.', 'success')
return redirect(url_for('admin_settings'))
return render_template(
'admin/settings.html',
health=health,
@ -965,6 +1112,9 @@ def admin_settings():
booth_open=booth_open,
email_form=email_form,
metadata=runtime_settings,
hermes_key_masked=hermes_key_masked,
hermes_key_set=hermes_key_set,
hermes_key_just_generated=hermes_key_just_generated,
)