265 lines
8.3 KiB
Python
265 lines
8.3 KiB
Python
#!/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/<path:filename>')
|
|
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)
|