amitis55/vz-kore-adapter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update portal_feed_poller.py — correct KoreID format + EventBus + v2 poller

Browse Source
This commit is contained in:
JA
2026-07-09 03:42:42 +00:00
parent 50218c28d5
commit ca3783f348
1 changed files with 183 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files

363
portal_feed_poller.py
View File

@@ -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-<period> 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
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,
'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()
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")