#!/usr/bin/env python3 """ Vector Zulu — KoreID Adapter API Port 8766 on Charlie. Provides status, queue monitoring, and manual event injection. """ import os, json, time, logging from datetime import datetime, timezone from flask import Flask, jsonify, request, abort import redis import psycopg2 from koreid import KoreIDGenerator logging.basicConfig(level=logging.INFO, format='%(asctime)s [ADAPTER-API] %(message)s') app = Flask(__name__) QUEUE_KEY = 'korechain:outbound_queue' DEAD_LETTER = 'korechain:dead_letter' def get_redis(): return redis.Redis( host=os.environ.get('REDIS_HOST', 'localhost'), port=6379, decode_responses=True ) def get_pg(): return psycopg2.connect( host=os.environ.get('POSTGRES_HOST', 'localhost'), dbname=os.environ.get('POSTGRES_DB', 'vz_oracle'), user=os.environ.get('POSTGRES_USER', 'vz_oracle'), password=os.environ.get('POSTGRES_PASSWORD', ''), connect_timeout=10 ) @app.route('/health') def health(): try: r = get_redis() r.ping() queue_len = r.llen(QUEUE_KEY) retry_len = r.zcard('korechain:retry_queue') dead_len = r.llen(DEAD_LETTER) last_block = r.get('eth_listener:last_block') return jsonify({ 'status': 'ok', 'queue_depth': queue_len, 'retry_queue': retry_len, 'dead_letter': dead_len, 'last_eth_block': last_block, 'timestamp': datetime.now(timezone.utc).isoformat(), }) except Exception as e: return jsonify({'status': 'error', 'error': str(e)}), 500 @app.route('/queue') def queue_status(): try: r = get_redis() queue_items = r.lrange(QUEUE_KEY, 0, 9) # last 10 return jsonify({ 'queue_depth': r.llen(QUEUE_KEY), 'recent': [json.loads(i) for i in queue_items], }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/events') def recent_events(): try: pg = get_pg() with pg.cursor() as cur: cur.execute(""" SELECT koreid, guid, event_type, token_symbol, tx_hash, amount, status, posted_at FROM kore_outbound_events ORDER BY posted_at DESC LIMIT 50 """) rows = cur.fetchall() pg.close() return jsonify({'events': [{ 'koreid': r[0], 'guid': r[1], 'event_type': r[2], 'token': r[3], 'tx_hash': r[4], 'amount': float(r[5]) if r[5] else 0, 'status': r[6], 'posted_at': r[7].isoformat() if r[7] else None, } for r in rows]}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/generate', methods=['POST']) def generate_koreid(): """Manually generate a KoreID for testing.""" data = request.get_json() if not data: abort(400) try: gen = KoreIDGenerator() result = gen.generate( type_token=data.get('event_type', 'EVT'), domain_token=data.get('domain', 'ANCHOR'), payload={'currency': data.get('currency', 'USD'), 'network': data.get('network', 'ETHEREUM')}, ) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/inject', methods=['POST']) def inject_event(): """ Manually inject an Ethereum event into the queue. Useful for testing or replaying missed events. """ data = request.get_json() if not data: abort(400) try: r = get_redis() gen = KoreIDGenerator() kid = gen.for_ethereum_anchor( original_koreid=data.get('linked_koreid', 'GENESIS-07'), tx_hash=data['tx_hash'], block_number=int(data.get('block_number', 0)), event_type=data['event_type'], ) payload = { 'koreid': kid['koreid'], 'guid': kid['guid'], 'step': data['event_type'], 'currency': data['currency'], 'amount': float(data['amount']), 'token_symbol': data['token_symbol'], 'ethereum_tx': data['tx_hash'], 'linked_to': data.get('linked_koreid', 'GENESIS-07'), 'anchored_at': __import__('datetime').datetime.utcnow().isoformat(), } r.lpush(QUEUE_KEY, json.dumps(payload)) return jsonify({'queued': True, 'koreid': kid['koreid']}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/dead-letter') def dead_letter(): try: r = get_redis() items = r.lrange(DEAD_LETTER, 0, 19) return jsonify({ 'count': r.llen(DEAD_LETTER), 'items': [json.loads(i) for i in items], }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/dead-letter/retry', methods=['POST']) def retry_dead_letter(): """Move all dead letter items back to main queue for retry.""" try: r = get_redis() count = 0 while True: item = r.rpop(DEAD_LETTER) if not item: break payload = json.loads(item) payload['_retries'] = 0 # reset retry count r.lpush(QUEUE_KEY, json.dumps(payload)) count += 1 return jsonify({'requeued': count}) except Exception as e: return jsonify({'error': str(e)}), 500 # ── Public file serving ─────────────────────────────────────── # Serves deployment artifacts for KoreNet/Corné to pull directly # Basic auth gated — credentials in environment variables def check_public_auth(): """Basic auth for public file serving.""" from flask import request, Response import os, hmac, hashlib auth = request.authorization expected_user = os.environ.get('PUBLIC_USER', 'korenet') expected_pass = os.environ.get('PUBLIC_PASS', '') if not expected_pass: return True # No password configured — open access if not auth: return False # Constant-time comparison user_ok = hmac.compare_digest(auth.username.encode(), expected_user.encode()) pass_ok = hmac.compare_digest(auth.password.encode(), expected_pass.encode()) return user_ok and pass_ok def auth_required(): from flask import Response return Response( 'Authentication required.', 401, {'WWW-Authenticate': 'Basic realm="VZ Public Files"'} ) PUBLIC_DIR = os.path.join(os.path.dirname(__file__), 'public') os.makedirs(PUBLIC_DIR, exist_ok=True) @app.route('/public/') def serve_public(filename): """Serve public deployment files — basic auth required.""" if not check_public_auth(): return auth_required() import re # Sanitise path — no directory traversal if '..' in filename or filename.startswith('/'): abort(400) filepath = os.path.join(PUBLIC_DIR, filename) if not os.path.exists(filepath): abort(404) with open(filepath, 'rb') as f: data = f.read() # Detect content type if filename.endswith('.yaml') or filename.endswith('.yml'): ct = 'text/yaml' elif filename.endswith('.sh'): ct = 'text/x-sh' elif filename.endswith('.json'): ct = 'application/json' else: ct = 'text/plain' from flask import Response return Response(data, mimetype=ct) @app.route('/public') def list_public(): """List available public files — basic auth required.""" if not check_public_auth(): return auth_required() files = [] if os.path.exists(PUBLIC_DIR): for f in sorted(os.listdir(PUBLIC_DIR)): fp = os.path.join(PUBLIC_DIR, f) files.append({ 'name': f, 'size': os.path.getsize(fp), 'url': f'http://63.245.138.38:8766/public/{f}', 'modified': datetime.fromtimestamp( os.path.getmtime(fp), tz=timezone.utc).isoformat() }) return jsonify({'files': files, 'count': len(files)}) if __name__ == '__main__': app.run(host='0.0.0.0', port=8766, debug=False)