From ca3783f348cff415b621a5f3ae27ea1920cedf56 Mon Sep 17 00:00:00 2001 From: JA Date: Thu, 9 Jul 2026 03:42:42 +0000 Subject: [PATCH] =?UTF-8?q?Update=20portal=5Ffeed=5Fpoller.py=20=E2=80=94?= =?UTF-8?q?=20correct=20KoreID=20format=20+=20EventBus=20+=20v2=20poller?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- portal_feed_poller.py | 371 +++++++++++++++++++++--------------------- 1 file changed, 183 insertions(+), 188 deletions(-) diff --git a/portal_feed_poller.py b/portal_feed_poller.py index c2113c3..e3862b2 100644 --- a/portal_feed_poller.py +++ b/portal_feed_poller.py @@ -1,43 +1,36 @@ #!/usr/bin/env python3 """ -Vector Zulu — Payment Mirror Adapter -Polls KoreNet portal feeds for new payment events and executes the full -payment mirror lifecycle via api.korenet.cloud. +Vector Zulu — Portal Feed Poller v2.0 +Polls KoreNet portal feeds for new payment/settlement/deal events +and executes the full payment mirror lifecycle. -Full lifecycle per VZ runbook: - INCOMING SWIFT MT103/PACS.008 - → MINT mirror tokens - → SETTLE T+0 - → REDEEM fiat - → BURN-SELL via venue routing - → CLOSE tranche - → ANCHOR to Ethereum (ISO 20022 proof) - → POST confirmation KoreID back to KoreChain - -Dependencies: - - vz_mtls_client.py (Bryan's official client) - - koreid.py (KoreID generator) - - Redis (event deduplication) - - Postgres (audit trail) +Key facts from VZ-Welcome-Pack FINAL v1.0 (2026-07-08): +- UETR comes FROM the payment feed (SWIFT gpi UUID v4) — never mint a new one +- Auth-code format: AUTH-VZ-MINT- e.g. AUTH-VZ-MINT-2026Q2 +- Lifecycle keyed by SAME UETR: mint → settle → redeem → burn-sell → close +- Every step LICK-signed +- KoreID validated via benjii-prime.korenet.cloud/api/v1/koreid/verify-lineage +- EventBus topic: integration-partners (VZ's own scoped topic) """ import os import json import time +import uuid import logging -import hashlib import subprocess import requests import redis import psycopg2 import urllib3 from datetime import datetime, timezone -from koreid import KoreIDGenerator, wrap_ethereum_event +from koreid_new import KoreIDGenerator, KoreIDValidator, KoreIDParser, KoreType, KoreDomain, GENESIS_ROOT +from eventbus import EventBusClient, Severity, EventType urllib3.disable_warnings() logging.basicConfig( level=logging.INFO, - format='%(asctime)s [MIRROR] %(levelname)s %(message)s' + format='%(asctime)s [POLLER-V2] %(levelname)s %(message)s' ) log = logging.getLogger(__name__) @@ -48,13 +41,10 @@ KEY = os.environ.get('VECTOR_ZULU_KEY', '/certs/vector-zulu-client.key RAILS_URL = os.environ.get('VECTOR_ZULU_RAILS_URL', 'https://vector.korenet.cloud') API_URL = os.environ.get('VECTOR_ZULU_API_URL', 'https://api.korenet.cloud') POLL_INTERVAL = int(os.environ.get('POLL_INTERVAL', '30')) - -# Venue routing priority (per runbook) -VENUES = ['cryptohost', 'koreexchange', 'uniswap'] - -# VZ vault IDs -VZ_VAULTS = ['VZ-USD-VAULT', 'VZ-USDT-VAULT', 'VZ-BTC-VAULT'] - +AUTH_CODE = os.environ.get('VZ_MINT_AUTH_CODE', 'AUTH-VZ-MINT-2026Q2') +VZ_TENANT = 'vector-zulu' +VZ_VAULTS = ['VZ-USD-VAULT', 'VZ-USDT-VAULT', 'VZ-BTC-VAULT'] +VENUES = ['cryptohost', 'koreexchange', 'uniswap'] # ── Infrastructure ──────────────────────────────────────────────────────────── @@ -64,7 +54,6 @@ def get_redis(): port=6379, decode_responses=True ) - def get_pg(): return psycopg2.connect( host=os.environ.get('POSTGRES_HOST', 'localhost'), @@ -74,17 +63,15 @@ def get_pg(): connect_timeout=10 ) - -def session(): +def rails_session(): s = requests.Session() s.cert = (CERT, KEY) s.verify = False - s.headers = {'X-Tenant-Id': 'vector-zulu'} + s.headers = {'X-Tenant-Id': VZ_TENANT} return s - def run_tool(*args) -> dict: - """Run vz_mtls_client.py with given args, return parsed JSON.""" + """Run vz_mtls_client.py, return parsed JSON.""" cmd = ['python3', 'tools/vz_mtls_client.py'] + list(args) try: result = subprocess.run( @@ -92,24 +79,19 @@ def run_tool(*args) -> dict: cwd=os.path.expanduser('~/vector-zulu'), timeout=30 ) - # Strip [OK]/[FAIL] prefix line - lines = result.stdout.strip().split('\n') - json_lines = [l for l in lines if not l.startswith('[OK]') and not l.startswith('[FAIL]')] - return json.loads('\n'.join(json_lines)) + lines = [l for l in result.stdout.strip().split('\n') + if not l.startswith('[OK]') and not l.startswith('[FAIL]')] + return json.loads('\n'.join(lines)) if lines else {} except Exception as e: - log.error(f"Tool error: {e}") return {'error': str(e)} - # ── Feed polling ────────────────────────────────────────────────────────────── def poll_feed(path: str) -> list: - """Poll a KoreNet portal feed, return new items.""" try: - resp = session().get(f'{RAILS_URL}/{path}', timeout=15) + resp = rails_session().get(f'{RAILS_URL}/{path}', timeout=15) if resp.status_code == 200: data = resp.json() - # Return whichever list field is present for key in ['payments', 'settlements', 'deals']: if key in data and data[key]: return data[key] @@ -117,266 +99,279 @@ def poll_feed(path: str) -> list: log.error(f"Feed poll error {path}: {e}") return [] - def is_processed(r, event_id: str) -> bool: - """Check Redis deduplication cache.""" return r.exists(f'mirror:processed:{event_id}') == 1 - def mark_processed(r, event_id: str): - """Mark event as processed in Redis (24h TTL).""" r.setex(f'mirror:processed:{event_id}', 86400, '1') +# ── UETR validation ─────────────────────────────────────────────────────────── -# ── Compliance checks ───────────────────────────────────────────────────────── +def validate_uetr(uetr: str) -> str: + """ + Validate UETR is a proper UUID v4. + Per welcome pack: UETR comes from feed, VZ passes through. + Generate fresh UUID v4 ONLY if feed genuinely lacks one. + """ + if not uetr: + fresh = str(uuid.uuid4()) + log.warning(f"Feed missing UETR — generated fresh: {fresh}") + return fresh + try: + val = uuid.UUID(uetr, version=4) + return str(val).lower() + except ValueError: + fresh = str(uuid.uuid4()) + log.warning(f"Invalid UETR {uetr} — generated fresh: {fresh}") + return fresh -def sanctions_check(entity: str, jurisdiction: str = 'ZA') -> bool: +# ── KoreID validation ───────────────────────────────────────────────────────── + +def validate_inbound_koreid(koreid: str, validator: KoreIDValidator, + eb: EventBusClient, uetr: str) -> bool: """ - Run sanctions screening before any mirror operation. - Returns True if CLEAR, False if HIT. + Validate inbound KoreID from KoreNet feed. + Phase-4: must be genesis_anchored=True to proceed. """ - result = run_tool('oracle', 'sanctions', - '--entity', entity, - '--jurisdiction', jurisdiction) - status = result.get('status', 'UNKNOWN') - if status == 'CLEAR': - log.info(f"Sanctions CLEAR: {entity}") - return True - else: - log.warning(f"Sanctions HIT: {entity} — {status}") + 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 + result = validator.verify_lineage(koreid) + if not result: + log.error(f"KoreID validation failed (no response): {koreid}") + eb.emit_koreid_invalid(koreid, 'validation_timeout', uetr) + return False -def entity_enrich(entity: str, entity_type: str = 'BENEFICIARY', - jurisdiction: str = 'ZA') -> dict: - """Enrich entity data before mirror mint.""" - return run_tool('oracle', 'enrich', - '--entity', entity, - '--type', entity_type, - '--jurisdiction', jurisdiction) + if not result.get('phase4_compliant'): + reason = result.get('reason', 'unknown') + log.error(f"KoreID not Phase-4 compliant: {koreid} — {reason}") + eb.emit_koreid_invalid(koreid, reason, uetr) + return False + log.info(f"KoreID validated: {koreid} — genesis anchored ✓") + return True -# ── Payment mirror lifecycle ────────────────────────────────────────────────── +# ── Mirror lifecycle ────────────────────────────────────────────────────────── -def mirror_mint(uetr: str, customer: str, asset: str, - amount: float, rail_ref: str, vault: str) -> dict: - """ - Step 1: Mint mirror tokens against incoming SWIFT payment. - """ - log.info(f"MINT {amount} {asset} — UETR: {uetr}") +def mirror_mint(uetr: str, asset: str, amount: float, + rail_ref: str, vault: str) -> dict: + log.info(f"MINT {amount} {asset} UETR:{uetr}") return run_tool('mirror', 'mint', '--uetr', uetr, - '--customer', customer, + '--customer', VZ_TENANT, '--asset', asset, '--amount', str(amount), '--rail-ref', rail_ref, '--vault', vault) - def mirror_settle(uetr: str) -> dict: - """Step 2: Record T+0 settlement.""" - log.info(f"SETTLE — UETR: {uetr}") - settled_at = datetime.now(timezone.utc).isoformat() + log.info(f"SETTLE UETR:{uetr}") return run_tool('mirror', 'settle', '--uetr', uetr, - '--settled-at', settled_at) - + '--settled-at', datetime.now(timezone.utc).isoformat()) def mirror_redeem(uetr: str, amount: float) -> dict: - """Step 3: Redeem fiat from VZ vault.""" - log.info(f"REDEEM {amount} — UETR: {uetr}") + log.info(f"REDEEM {amount} UETR:{uetr}") return run_tool('mirror', 'redeem', '--uetr', uetr, '--amount', str(amount)) - def mirror_burn_sell(uetr: str, amount: float) -> dict: - """ - Step 4: Burn-sell via venue routing. - Tries CryptoHost → KoreExchange → Uniswap per runbook. - """ for venue in VENUES: - log.info(f"BURN-SELL via {venue} — UETR: {uetr}") + log.info(f"BURN-SELL via {venue} UETR:{uetr}") result = run_tool('mirror', 'burn-sell', '--uetr', uetr, '--amount', str(amount), '--venue', venue) if 'error' not in result: - log.info(f"Burn-sell success via {venue}") return result - log.warning(f"Venue {venue} failed, trying next") + log.warning(f"Venue {venue} failed") return {'error': 'All venues failed'} - def mirror_close(uetr: str) -> dict: - """Step 5: Close the tranche.""" - log.info(f"CLOSE — UETR: {uetr}") + log.info(f"CLOSE UETR:{uetr}") return run_tool('mirror', 'close', '--uetr', uetr) - # ── Ethereum anchoring ──────────────────────────────────────────────────────── -def anchor_to_ethereum(kore_id: str, uetr: str, amount: float, - currency: str, event_type: str) -> str: +def anchor_to_ethereum(koreid: str, uetr: str, amount: float, + currency: str, step: str, gen: KoreIDGenerator) -> str: """ - Anchor KoreNet event to Ethereum mainnet. - Returns Ethereum TX hash. - TODO: Deploy KoreAnchor.sol contract and call it here. - For now, stores in oracle transmit log. + Anchor KoreNet event to Ethereum. + Generates anchor KoreID linked to original. + TODO: Call KoreAnchor.sol once deployed. """ - payload = { - 'koreId': kore_id, - 'uetr': uetr, - 'amount': amount, - 'currency': currency, - 'eventType': event_type, - 'timestamp': datetime.now(timezone.utc).isoformat(), - 'hash': hashlib.sha256( - json.dumps({'koreId': kore_id, 'uetr': uetr, 'amount': amount}, - sort_keys=True).encode() - ).hexdigest() - } - log.info(f"Anchoring to Ethereum: {kore_id} — {event_type}") + import hashlib + anchor_kid = gen.for_ethereum_anchor( + original_koreid=koreid, + tx_hash=f"pending_{uetr}_{step}", + block_number=0, + event_type=step + ) + log.info(f"Anchor KoreID: {anchor_kid['koreid']} → {koreid}") # TODO: Call KoreAnchor.sol contract - # tx_hash = kore_anchor_contract.anchor(payload) - # return tx_hash - return f"0x{payload['hash'][:64]}" # placeholder until contract deployed - + # tx = anchor_contract.anchor(koreid, anchor_kid['hash'], amount) + placeholder_hash = f"0x{hashlib.sha256(f'{koreid}{step}'.encode()).hexdigest()}" + return placeholder_hash def trigger_oracle_update(): """Immediately trigger oracle transmit when VZ vault balance changes.""" - log.info("Triggering immediate oracle update") + log.info("Triggering immediate oracle transmit") try: - import subprocess subprocess.Popen( ['node', '/opt/eth-deploy-v21/manual_transmit.js'], stdout=open('/home/nigel/vz-transmit.log', 'a'), stderr=subprocess.STDOUT ) except Exception as e: - log.error(f"Oracle update failed: {e}") + log.error(f"Oracle trigger failed: {e}") +# ── Post confirmation to KoreChain ──────────────────────────────────────────── -# ── KoreID confirmation ─────────────────────────────────────────────────────── - -def post_confirmation(original_kore_id: str, tx_hash: str, - uetr: str, event_type: str, pg) -> dict: +def post_confirmation(original_koreid: str, tx_hash: str, uetr: str, + step: str, gen: KoreIDGenerator, r) -> str: """ Post Ethereum anchor confirmation back to KoreChain. - Generates a confirmation KoreID linked to the original. - TODO: POST to korechain.korenet.cloud/api/v1/anchor once G2 cert trusted. + Queued in Redis until korechain.korenet.cloud is accessible. """ - gen = KoreIDGenerator(pg) - payload = gen.generate( - event_type=f'{event_type}_CONFIRM', - currency='USD', - current_network='ETHEREUM', - parent_network='KORECHAIN', + anchor_kid = gen.for_ethereum_anchor( + original_koreid=original_koreid, + tx_hash=tx_hash, + block_number=0, + event_type=step ) - confirmation = { - 'koreid': payload['koreid'], - 'linked_to': original_kore_id, - 'uetr': uetr, + 'koreid': anchor_kid['koreid'], + 'linked_to': original_koreid, + 'uetr': uetr, 'ethereum_tx': tx_hash, - 'event_type': f'{event_type}_CONFIRM', + 'step': step, 'anchored_at': datetime.now(timezone.utc).isoformat(), - 'proof': { - 'hash': tx_hash, - 'network': 'ETHEREUM_MAINNET', - } + 'genesis_root': GENESIS_ROOT, } - - log.info(f"Confirmation KoreID: {payload['koreid']} → {original_kore_id}") - - # TODO: POST to korechain.korenet.cloud once G2 cert trusted - # session().post('https://korechain.korenet.cloud/api/v1/anchor', json=confirmation) - - # Queue for when endpoint is available - r = get_redis() r.lpush('korechain:outbound_queue', json.dumps(confirmation)) + log.info(f"Queued confirmation: {anchor_kid['koreid']}") + return anchor_kid['koreid'] - return confirmation +# ── Main event processor ────────────────────────────────────────────────────── +def process_event(event: dict, r, pg, + gen: KoreIDGenerator, + validator: KoreIDValidator, + eb: EventBusClient): + """Process a single payment/deal/settlement event.""" -# ── Main processing ─────────────────────────────────────────────────────────── - -def process_payment_event(event: dict, r, pg): - """ - Process a single payment/deal event from the KoreNet feed. - Executes full mirror lifecycle. - """ - event_id = event.get('id') or event.get('dealId') or event.get('uetr', '') - kore_id = event.get('koreId') or event.get('kore_id', '') + event_id = event.get('id') or event.get('dealId') or event.get('paymentId', '') + koreid = event.get('koreId') or event.get('kore_id', '') amount = float(event.get('amount', 0)) currency = event.get('currency', 'USD') entity = event.get('counterparty') or event.get('customer', 'UNKNOWN') - uetr = event.get('uetr') or event.get('railRef') or event_id rail_ref = event.get('railRef') or event.get('reference', f'REF-{event_id}') vault = f"VZ-{currency}-VAULT" if f"VZ-{currency}-VAULT" in VZ_VAULTS else 'VZ-USD-VAULT' + # UETR — comes from feed per welcome pack + 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: {event_id} | {currency} {amount:,.2f} | KoreID: {kore_id}") + log.info(f"Processing: {event_id} | {currency} {amount:,.2f} | UETR: {uetr}") + + correlation_id = str(uuid.uuid4()) try: - # Step 0: Compliance - if not sanctions_check(entity): - log.error(f"SANCTIONS HIT — {entity} — rejecting {event_id}") + # Step 0a: Validate inbound KoreID + if koreid and not validate_inbound_koreid(koreid, validator, eb, uetr): + log.error(f"Rejecting {event_id} — invalid KoreID") mark_processed(r, event_id) return - # Step 1: Mint - mint_result = mirror_mint(uetr, 'vector-zulu', currency, amount, rail_ref, vault) - if 'error' in mint_result: - log.error(f"Mint failed: {mint_result}") + # Step 0b: Sanctions check + sanctions = run_tool('oracle', 'sanctions', + '--entity', entity, + '--jurisdiction', 'ZA') + if sanctions.get('status') != 'CLEAR': + log.error(f"SANCTIONS HIT — {entity}") + eb.emit_sanctions_hit(entity, 'ZA', uetr) + mark_processed(r, event_id) return - # Anchor MINT to Ethereum - tx_hash = anchor_to_ethereum(kore_id, uetr, amount, currency, 'MINT') - post_confirmation(kore_id, tx_hash, uetr, 'MINT', pg) + # Step 0c: Entity enrichment + run_tool('oracle', 'enrich', '--entity', entity, + '--type', 'BENEFICIARY', '--jurisdiction', 'ZA') + + # Step 1: Mint + mint_result = mirror_mint(uetr, currency, amount, rail_ref, vault) + if 'error' in mint_result: + eb.emit_mirror_failed('mint', uetr, str(mint_result.get('error'))) + return + + mint_kid = gen.for_mirror_mint(uetr, amount, currency, vault, + parent=koreid or GENESIS_ROOT) + tx = anchor_to_ethereum(mint_kid['koreid'], uetr, amount, currency, 'MINT', gen) + post_confirmation(mint_kid['koreid'], tx, uetr, 'MINT', gen, r) + eb.emit_mirror_success('mint', uetr, mint_kid['koreid'], amount, currency) # Step 2: Settle settle_result = mirror_settle(uetr) - tx_hash = anchor_to_ethereum(kore_id, uetr, amount, currency, 'SETTLE') - post_confirmation(kore_id, tx_hash, uetr, 'SETTLE', pg) + settle_kid = gen.for_mirror_settle(uetr, parent_koreid=mint_kid['koreid']) + tx = anchor_to_ethereum(settle_kid['koreid'], uetr, amount, currency, 'SETTLE', gen) + post_confirmation(settle_kid['koreid'], tx, uetr, 'SETTLE', gen, r) + eb.emit_mirror_success('settle', uetr, settle_kid['koreid'], amount, currency) # Step 3: Redeem redeem_result = mirror_redeem(uetr, amount) - tx_hash = anchor_to_ethereum(kore_id, uetr, amount, currency, 'REDEEM') - post_confirmation(kore_id, tx_hash, uetr, 'REDEEM', pg) + redeem_kid = gen.for_mirror_redeem(uetr, amount, parent_koreid=settle_kid['koreid']) + tx = anchor_to_ethereum(redeem_kid['koreid'], uetr, amount, currency, 'REDEEM', gen) + post_confirmation(redeem_kid['koreid'], tx, uetr, 'REDEEM', gen, r) + eb.emit_mirror_success('redeem', uetr, redeem_kid['koreid'], amount, currency) - # Trigger oracle update if VZ vault affected + # Oracle update if VZ vault affected if vault in VZ_VAULTS: trigger_oracle_update() # Step 4: Burn-sell burn_result = mirror_burn_sell(uetr, amount) - tx_hash = anchor_to_ethereum(kore_id, uetr, amount, currency, 'BURN_SELL') - post_confirmation(kore_id, tx_hash, uetr, 'BURN_SELL', pg) + venue = burn_result.get('venue', 'unknown') + burn_kid = gen.for_mirror_burn(uetr, amount, venue, + parent_koreid=redeem_kid['koreid']) + tx = anchor_to_ethereum(burn_kid['koreid'], uetr, amount, currency, 'BURN', gen) + post_confirmation(burn_kid['koreid'], tx, uetr, 'BURN', gen, r) + eb.emit_mirror_success('burn', uetr, burn_kid['koreid'], amount, currency) # Step 5: Close close_result = mirror_close(uetr) - tx_hash = anchor_to_ethereum(kore_id, uetr, amount, currency, 'CLOSE') - post_confirmation(kore_id, tx_hash, uetr, 'CLOSE', pg) + close_kid = gen.for_mirror_close(uetr, parent_koreid=burn_kid['koreid']) + tx = anchor_to_ethereum(close_kid['koreid'], uetr, amount, currency, 'CLOSE', gen) + post_confirmation(close_kid['koreid'], tx, uetr, 'CLOSE', gen, r) + eb.emit_mirror_success('close', uetr, close_kid['koreid'], amount, currency) mark_processed(r, event_id) - log.info(f"✓ Complete: {event_id} | {currency} {amount:,.2f}") + log.info(f"✓ Complete: {event_id} | {currency} {amount:,.2f} | UETR: {uetr}") except Exception as e: log.error(f"Processing error {event_id}: {e}") + eb.emit(EventType.ENDPOINT_DOWN, + {'event_id': event_id, 'error': str(e)}, + Severity.HIGH) def run(): - log.info("Vector Zulu Payment Mirror Adapter starting...") + log.info("VZ Portal Feed Poller v2.0 starting...") log.info(f"Rails: {RAILS_URL} | API: {API_URL} | Poll: {POLL_INTERVAL}s") - r = get_redis() - pg = get_pg() + r = get_redis() + pg = get_pg() + gen = KoreIDGenerator() + validator = KoreIDValidator(CERT, KEY) + eb = EventBusClient() while True: - # Poll all three feeds payments = poll_feed('api/v1/tenants/portal/payments/feed') settlements = poll_feed('api/v1/tenants/portal/settlements/feed') deals = poll_feed('api/v1/tenants/deals/feed') @@ -384,9 +379,9 @@ def run(): all_events = payments + settlements + deals if all_events: - log.info(f"Found {len(all_events)} events to process") + log.info(f"Found {len(all_events)} events") for event in all_events: - process_payment_event(event, r, pg) + process_event(event, r, pg, gen, validator, eb) else: log.debug("No new events")