454 lines
18 KiB
Python
454 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Vector Zulu — Portal Feed Poller v3.0
|
|
Polls KoreNet payment/settlement/deal feeds for new events
|
|
and executes the full payment mirror lifecycle.
|
|
|
|
Key changes from v2:
|
|
- JWT-only auth (no client cert — ssl_verify_client optional confirmed)
|
|
- Primary endpoint: korechain.korenet.cloud
|
|
- Fallback: fnb.korenet.cloud
|
|
- LICK signing on all mutating calls via lick.py
|
|
- Correct swagger-confirmed API paths
|
|
- Payment mirror via /api/v1/payment-mirror/*
|
|
- Compliance via /api/Compliance/sanctions/ofac and /api/v1/compliance/ofac/*
|
|
- KoreID validation via benjii-prime.korenet.cloud (JWT auth)
|
|
- UETR passthrough — never mint new unless feed genuinely lacks one
|
|
"""
|
|
|
|
import os, json, time, uuid, logging, requests, redis, psycopg2, urllib3
|
|
from datetime import datetime, timezone
|
|
from koreid import KoreIDGenerator, KoreIDParser, KoreType, KoreDomain, GENESIS_ROOT
|
|
from eventbus import EventBusClient, Severity, EventType
|
|
from lick import build_headers, canonicalize
|
|
|
|
urllib3.disable_warnings()
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s [POLLER-V3] %(levelname)s %(message)s')
|
|
log = logging.getLogger(__name__)
|
|
|
|
# ── Config ────────────────────────────────────────────────────────────────────
|
|
|
|
API_PRIMARY = os.environ.get('KORENET_API_URL', 'https://korechain.korenet.cloud')
|
|
API_FALLBACK = os.environ.get('KORENET_API_FALLBACK', 'https://fnb.korenet.cloud')
|
|
RAILS_URL = os.environ.get('VECTOR_ZULU_RAILS_URL', 'https://vector.korenet.cloud')
|
|
BENJII_URL = 'https://benjii-prime.korenet.cloud'
|
|
POLL_INTERVAL = int(os.environ.get('POLL_INTERVAL', '30'))
|
|
TENANT_ID = 'vector-zulu'
|
|
VZ_VAULTS = ['VZ-USD-VAULT', 'VZ-USDT-VAULT', 'VZ-BTC-VAULT']
|
|
VENUES = ['cryptohost', 'koreexchange', 'uniswap']
|
|
|
|
|
|
def load_jwt() -> str:
|
|
jwt = os.environ.get('VECTOR_ZULU_JWT', '')
|
|
if jwt:
|
|
return jwt
|
|
for path in ['/app/.secrets', '/home/constellationnode/.korenet/vector-zulu/.secrets',
|
|
os.path.expanduser('~/.korenet/vector-zulu/.secrets')]:
|
|
if os.path.exists(path):
|
|
with open(path) as f:
|
|
for line in f:
|
|
if 'VECTOR_ZULU_JWT=' in line:
|
|
val = line.split('VECTOR_ZULU_JWT=', 1)[1].strip().strip('"').strip("'")
|
|
if val:
|
|
return val
|
|
return ''
|
|
|
|
|
|
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
|
|
)
|
|
|
|
|
|
def _session(jwt: str) -> requests.Session:
|
|
"""JWT-only session — no client cert required."""
|
|
s = requests.Session()
|
|
s.verify = False
|
|
s.headers.update({
|
|
'Authorization': f'Bearer {jwt}',
|
|
'X-Tenant-Id': TENANT_ID,
|
|
'Content-Type': 'application/json',
|
|
})
|
|
return s
|
|
|
|
|
|
def _get(jwt: str, path: str, base: str = None) -> dict:
|
|
for url_base in ([base] if base else [API_PRIMARY, API_FALLBACK, RAILS_URL]):
|
|
try:
|
|
resp = _session(jwt).get(f'{url_base}{path}', timeout=15)
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
except Exception as e:
|
|
log.debug(f"GET {path} from {url_base}: {e}")
|
|
return {}
|
|
|
|
|
|
def _post(jwt: str, path: str, body: dict, lick_sign: bool = True) -> dict:
|
|
headers = build_headers(body) if lick_sign else {}
|
|
for url_base in [API_PRIMARY, API_FALLBACK]:
|
|
try:
|
|
s = _session(jwt)
|
|
if headers:
|
|
s.headers.update(headers)
|
|
resp = s.post(f'{url_base}{path}', json=body, timeout=30)
|
|
if resp.status_code in (200, 201, 202):
|
|
return resp.json() if resp.content else {'ok': True}
|
|
log.warning(f"POST {path}: HTTP {resp.status_code} — {resp.text[:200]}")
|
|
except Exception as e:
|
|
log.error(f"POST {path} via {url_base}: {e}")
|
|
return {'error': f'All endpoints failed for {path}'}
|
|
|
|
|
|
# ── Feed polling ──────────────────────────────────────────────────────────────
|
|
|
|
def poll_feed(jwt: str, path: str) -> list:
|
|
"""Poll a KoreNet feed endpoint for new events."""
|
|
data = _get(jwt, path)
|
|
for key in ['payments', 'settlements', 'deals', 'items', 'events']:
|
|
if key in data and data[key]:
|
|
return data[key]
|
|
# Some feeds return list directly
|
|
if isinstance(data, list):
|
|
return data
|
|
return []
|
|
|
|
|
|
def is_processed(r, event_id: str) -> bool:
|
|
return r.exists(f'mirror:processed:{event_id}') == 1
|
|
|
|
|
|
def mark_processed(r, event_id: str):
|
|
r.setex(f'mirror:processed:{event_id}', 86400, '1')
|
|
|
|
|
|
# ── UETR ──────────────────────────────────────────────────────────────────────
|
|
|
|
def validate_uetr(uetr: str) -> str:
|
|
"""Validate UETR is UUID v4. Passthrough from feed — only generate if genuinely missing."""
|
|
if not uetr:
|
|
fresh = str(uuid.uuid4())
|
|
log.warning(f"Feed missing UETR — generated fresh: {fresh}")
|
|
return fresh
|
|
try:
|
|
return str(uuid.UUID(uetr, version=4)).lower()
|
|
except ValueError:
|
|
fresh = str(uuid.uuid4())
|
|
log.warning(f"Invalid UETR {uetr} — generated fresh: {fresh}")
|
|
return fresh
|
|
|
|
|
|
# ── KoreID validation ─────────────────────────────────────────────────────────
|
|
|
|
def validate_inbound_koreid(koreid: str, jwt: str, eb: EventBusClient, uetr: str) -> bool:
|
|
"""
|
|
Validate inbound KoreID from feed via benjii-prime.korenet.cloud.
|
|
Phase-4: must be genesis_anchored=True and integrity_verified=True.
|
|
"""
|
|
parser = KoreIDParser()
|
|
if not parser.is_valid_format(koreid):
|
|
log.error(f"Invalid KoreID format: {koreid}")
|
|
eb.emit_koreid_invalid(koreid, 'invalid_format', uetr)
|
|
return False
|
|
|
|
try:
|
|
# JWT auth on benjii-prime too
|
|
s = _session(jwt)
|
|
resp = s.post(f'{BENJII_URL}/api/v1/koreid/verify-lineage',
|
|
json={'kore_id': koreid, 'parent': GENESIS_ROOT}, timeout=15)
|
|
if resp.status_code != 200:
|
|
log.warning(f"benjii-prime returned {resp.status_code} for {koreid}")
|
|
# Don't block on API error — log and continue
|
|
return True
|
|
|
|
result = resp.json()
|
|
phase4 = (result.get('integrity_verified') is True and
|
|
result.get('genesis_anchored') is True)
|
|
if not phase4:
|
|
reason = result.get('reason', 'not genesis anchored')
|
|
log.error(f"KoreID {koreid} not Phase-4 compliant: {reason}")
|
|
eb.emit_koreid_invalid(koreid, reason, uetr)
|
|
return False
|
|
|
|
log.info(f"KoreID {koreid} validated — genesis anchored ✓")
|
|
return True
|
|
|
|
except Exception as e:
|
|
log.error(f"KoreID validation error: {e}")
|
|
return True # Fail-open on network error to avoid blocking entire pipeline
|
|
|
|
|
|
# ── Sanctions & compliance ────────────────────────────────────────────────────
|
|
|
|
def sanctions_check(jwt: str, entity: str, jurisdiction: str = 'GLOBAL') -> bool:
|
|
"""
|
|
Run OFAC/UN/EU sanctions check.
|
|
Returns True if CLEAR, False if HIT.
|
|
Fail-open on error (log and continue).
|
|
"""
|
|
body = {'entityName': entity, 'jurisdiction': jurisdiction,
|
|
'screeningId': str(uuid.uuid4())}
|
|
# Try OFAC first, then general sanctions
|
|
for path in ['/api/v1/compliance/ofac/fuzzy-match', '/api/Compliance/sanctions/ofac']:
|
|
result = _post(jwt, path, body, lick_sign=False)
|
|
if 'error' not in result:
|
|
# CLEAR if no hits
|
|
hits = result.get('hits', result.get('sanctionsHits', []))
|
|
if hits:
|
|
log.error(f"SANCTIONS HIT — {entity}: {hits}")
|
|
return False
|
|
log.info(f"Sanctions clear: {entity}")
|
|
return True
|
|
log.warning(f"Sanctions check unavailable for {entity} — continuing")
|
|
return True
|
|
|
|
|
|
# ── Mirror lifecycle ──────────────────────────────────────────────────────────
|
|
|
|
def mirror_mint(jwt: str, uetr: str, asset: str, amount: float,
|
|
rail_ref: str, vault: str) -> dict:
|
|
log.info(f"MINT {amount} {asset} UETR:{uetr}")
|
|
body = {
|
|
'uetr': uetr, 'customer': TENANT_ID, 'asset': asset,
|
|
'amount': amount, 'railRef': rail_ref, 'destinationVault': vault,
|
|
}
|
|
return _post(jwt, '/api/v1/payment-mirror/mint', body)
|
|
|
|
|
|
def mirror_settle(jwt: str, uetr: str) -> dict:
|
|
log.info(f"SETTLE UETR:{uetr}")
|
|
body = {'uetr': uetr, 'settledAt': datetime.now(timezone.utc).isoformat()}
|
|
return _post(jwt, '/api/v1/payment-mirror/settle', body)
|
|
|
|
|
|
def mirror_redeem(jwt: str, uetr: str, amount: float) -> dict:
|
|
log.info(f"REDEEM {amount} UETR:{uetr}")
|
|
body = {'uetr': uetr, 'amount': amount}
|
|
return _post(jwt, '/api/v1/payment-mirror/redeem', body)
|
|
|
|
|
|
def mirror_burn_sell(jwt: str, uetr: str, amount: float) -> dict:
|
|
for venue in VENUES:
|
|
log.info(f"BURN-SELL via {venue} UETR:{uetr}")
|
|
body = {'uetr': uetr, 'tokenAmount': amount, 'preferredVenue': venue}
|
|
result = _post(jwt, '/api/v1/payment-mirror/burn-sell', body)
|
|
if 'error' not in result:
|
|
return result
|
|
log.warning(f"Venue {venue} failed")
|
|
return {'error': 'All venues failed'}
|
|
|
|
|
|
def mirror_close(jwt: str, uetr: str) -> dict:
|
|
log.info(f"CLOSE UETR:{uetr}")
|
|
return _post(jwt, f'/api/v1/payment-mirror/{uetr}/close', {})
|
|
|
|
|
|
# ── Ethereum anchoring ────────────────────────────────────────────────────────
|
|
|
|
def queue_anchor(r, koreid: str, linked_to: str, uetr: str,
|
|
step: str, amount: float, currency: str, gen: KoreIDGenerator):
|
|
"""
|
|
Queue an Ethereum anchor confirmation for the KoreChain poster.
|
|
Generates anchor KoreID linked to the original.
|
|
"""
|
|
import hashlib
|
|
# Placeholder tx hash until real Ethereum tx is submitted
|
|
placeholder_tx = f"0x{hashlib.sha256(f'{koreid}{step}{uetr}'.encode()).hexdigest()}"
|
|
|
|
anchor_kid = gen.for_ethereum_anchor(
|
|
original_koreid=linked_to,
|
|
tx_hash=placeholder_tx,
|
|
block_number=0,
|
|
event_type=step
|
|
)
|
|
|
|
payload = {
|
|
'koreid': anchor_kid['koreid'],
|
|
'guid': anchor_kid['guid'],
|
|
'linked_to': linked_to,
|
|
'uetr': uetr,
|
|
'ethereum_tx': placeholder_tx,
|
|
'step': step,
|
|
'amount': amount,
|
|
'currency': currency,
|
|
'anchored_at': datetime.now(timezone.utc).isoformat(),
|
|
'genesis_root': GENESIS_ROOT,
|
|
}
|
|
r.lpush('korechain:outbound_queue', json.dumps(payload))
|
|
log.info(f"Queued anchor: {anchor_kid['koreid']} → {linked_to}")
|
|
return anchor_kid['koreid'], placeholder_tx
|
|
|
|
|
|
def trigger_oracle_update():
|
|
"""Trigger out-of-cycle oracle transmit when VZ vault balance changes."""
|
|
import subprocess
|
|
log.info("Triggering immediate oracle transmit")
|
|
try:
|
|
subprocess.Popen(
|
|
['node', '/opt/eth-deploy-v21/manual_transmit.js'],
|
|
stdout=open('/home/constellationnode/vz-transmit.log', 'a'),
|
|
stderr=subprocess.STDOUT
|
|
)
|
|
except Exception as e:
|
|
log.error(f"Oracle trigger failed: {e}")
|
|
|
|
|
|
# ── Main event processor ──────────────────────────────────────────────────────
|
|
|
|
def process_event(event: dict, r, pg, jwt: str,
|
|
gen: KoreIDGenerator, eb: EventBusClient):
|
|
"""Process a single payment/deal/settlement event through full mirror lifecycle."""
|
|
|
|
event_id = (event.get('id') or event.get('dealId') or
|
|
event.get('paymentId') or event.get('koreId', ''))
|
|
koreid = event.get('koreId') or event.get('kore_id', '')
|
|
amount = float(event.get('amount', 0))
|
|
currency = event.get('currency', 'USD').upper()
|
|
entity = event.get('counterparty') or event.get('customer', 'UNKNOWN')
|
|
rail_ref = (event.get('railRef') or event.get('reference') or
|
|
event.get('swiftRef') or f'REF-{event_id}')
|
|
vault = f"VZ-{currency}-VAULT" if f"VZ-{currency}-VAULT" in VZ_VAULTS else 'VZ-USD-VAULT'
|
|
uetr = validate_uetr(
|
|
event.get('uetr') or event.get('railRef') or event.get('swiftRef', '')
|
|
)
|
|
|
|
if not event_id or is_processed(r, event_id):
|
|
return
|
|
|
|
log.info(f"Processing: {event_id} | {currency} {amount:,.2f} | UETR:{uetr}")
|
|
|
|
try:
|
|
# Step 0a: Validate inbound KoreID
|
|
if koreid and not validate_inbound_koreid(koreid, jwt, eb, uetr):
|
|
log.error(f"Rejecting {event_id} — invalid KoreID")
|
|
mark_processed(r, event_id)
|
|
return
|
|
|
|
# Step 0b: Sanctions check
|
|
if not sanctions_check(jwt, entity):
|
|
eb.emit_sanctions_hit(entity, 'GLOBAL', uetr)
|
|
mark_processed(r, event_id)
|
|
return
|
|
|
|
parent = koreid or GENESIS_ROOT
|
|
|
|
# Step 1: Mint
|
|
mint_result = mirror_mint(jwt, uetr, currency, amount, rail_ref, vault)
|
|
if 'error' in mint_result:
|
|
log.error(f"MINT failed: {mint_result.get('error')}")
|
|
eb.emit_mirror_failed('mint', uetr, str(mint_result.get('error')))
|
|
return
|
|
|
|
mint_kid = gen.for_mirror_mint(uetr, amount, currency, vault, parent=parent)
|
|
anchor_kid, tx = queue_anchor(r, mint_kid['koreid'], mint_kid['koreid'],
|
|
uetr, 'MINT', amount, currency, gen)
|
|
eb.emit_mirror_success('mint', uetr, mint_kid['koreid'], amount, currency)
|
|
log.info(f"MINT ✓ {mint_kid['koreid']}")
|
|
|
|
# Step 2: Settle
|
|
mirror_settle(jwt, uetr)
|
|
settle_kid = gen.for_mirror_settle(uetr, parent_koreid=mint_kid['koreid'])
|
|
queue_anchor(r, settle_kid['koreid'], settle_kid['koreid'],
|
|
uetr, 'SETTLE', amount, currency, gen)
|
|
eb.emit_mirror_success('settle', uetr, settle_kid['koreid'], amount, currency)
|
|
log.info(f"SETTLE ✓ {settle_kid['koreid']}")
|
|
|
|
# Step 3: Redeem
|
|
mirror_redeem(jwt, uetr, amount)
|
|
redeem_kid = gen.for_mirror_redeem(uetr, amount, parent_koreid=settle_kid['koreid'])
|
|
queue_anchor(r, redeem_kid['koreid'], redeem_kid['koreid'],
|
|
uetr, 'REDEEM', amount, currency, gen)
|
|
eb.emit_mirror_success('redeem', uetr, redeem_kid['koreid'], amount, currency)
|
|
log.info(f"REDEEM ✓ {redeem_kid['koreid']}")
|
|
|
|
# Trigger oracle update after vault balance changes
|
|
if vault in VZ_VAULTS:
|
|
trigger_oracle_update()
|
|
|
|
# Step 4: Burn-sell
|
|
burn_result = mirror_burn_sell(jwt, uetr, amount)
|
|
venue = burn_result.get('venue', burn_result.get('preferredVenue', 'cryptohost'))
|
|
burn_kid = gen.for_mirror_burn(uetr, amount, venue,
|
|
parent_koreid=redeem_kid['koreid'])
|
|
queue_anchor(r, burn_kid['koreid'], burn_kid['koreid'],
|
|
uetr, 'BURN', amount, currency, gen)
|
|
eb.emit_mirror_success('burn', uetr, burn_kid['koreid'], amount, currency)
|
|
log.info(f"BURN ✓ {burn_kid['koreid']} via {venue}")
|
|
|
|
# Step 5: Close
|
|
mirror_close(jwt, uetr)
|
|
close_kid = gen.for_mirror_close(uetr, parent_koreid=burn_kid['koreid'])
|
|
queue_anchor(r, close_kid['koreid'], close_kid['koreid'],
|
|
uetr, 'CLOSE', amount, currency, gen)
|
|
eb.emit_mirror_success('close', uetr, close_kid['koreid'], amount, currency)
|
|
log.info(f"CLOSE ✓ {close_kid['koreid']}")
|
|
|
|
mark_processed(r, event_id)
|
|
log.info(f"✓ COMPLETE: {event_id} | {currency} {amount:,.2f} | UETR:{uetr}")
|
|
|
|
except Exception as e:
|
|
log.error(f"Processing error {event_id}: {e}", exc_info=True)
|
|
eb.emit(EventType.ENDPOINT_DOWN, {'event_id': event_id, 'error': str(e)}, Severity.HIGH)
|
|
|
|
|
|
# ── Main loop ─────────────────────────────────────────────────────────────────
|
|
|
|
FEED_PATHS = [
|
|
'/api/v1/tenants/portal/payments/feed',
|
|
'/api/v1/tenants/portal/settlements/feed',
|
|
'/api/v1/tenants/deals/feed',
|
|
'/api/v1/payment/pending',
|
|
'/api/v1/payments/pending',
|
|
]
|
|
|
|
|
|
def run():
|
|
log.info("VZ Portal Feed Poller v3.0 starting...")
|
|
log.info(f"Primary: {API_PRIMARY} | Fallback: {API_FALLBACK} | Poll: {POLL_INTERVAL}s")
|
|
|
|
r = get_redis()
|
|
pg = get_pg()
|
|
gen = KoreIDGenerator()
|
|
eb = EventBusClient()
|
|
|
|
while True:
|
|
jwt = load_jwt()
|
|
if not jwt:
|
|
log.error("No JWT available — waiting 60s")
|
|
time.sleep(60)
|
|
continue
|
|
|
|
all_events = []
|
|
for path in FEED_PATHS:
|
|
events = poll_feed(jwt, path)
|
|
all_events.extend(events)
|
|
|
|
# Deduplicate by event ID
|
|
seen = set()
|
|
unique_events = []
|
|
for ev in all_events:
|
|
eid = ev.get('id') or ev.get('dealId') or ev.get('paymentId') or ev.get('koreId', '')
|
|
if eid and eid not in seen:
|
|
seen.add(eid)
|
|
unique_events.append(ev)
|
|
|
|
if unique_events:
|
|
log.info(f"Found {len(unique_events)} unique events")
|
|
for event in unique_events:
|
|
process_event(event, r, pg, jwt, gen, eb)
|
|
else:
|
|
log.debug("No new events")
|
|
|
|
time.sleep(POLL_INTERVAL)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run()
|