Update koreid.py — correct KoreID format + EventBus + v2 poller
This commit is contained in:
552
koreid.py
552
koreid.py
@@ -1,289 +1,353 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Vector Zulu — KoreID Generator
|
||||
Generates KoreID display numbers per the KoreID Display Generation Specification.
|
||||
Vector Zulu — KoreID Generator & Validator v2.0
|
||||
Correct format per VZ-Welcome-Pack FINAL v1.0 (2026-07-08)
|
||||
|
||||
Format: <S1>/<S2>-<S3>-<S4>-<S5>-<S6>
|
||||
Example: 061925a1/b231f4e5-3487-0a1b-03e8-001540009abc
|
||||
KoreID format: KID-<TYPE>-<DOMAIN>-<DATE>-<HASH>
|
||||
TYPE = record-type token (A-Z0-9, 2-8 chars) e.g. TXN, SLP, DOC, ACC, AGT
|
||||
DOMAIN = business-domain token (A-Z0-9, 2-10 chars) e.g. PAY, SETTLE, LEDGER, VAULT
|
||||
DATE = UTC date YYYYMMDD
|
||||
HASH = first 16 hex of sha256 over canonical payload + parent link
|
||||
|
||||
S1 (8) = DDMMYY (6-digit UTC date) + CT (2-hex ID-type)
|
||||
S2 (8) = PT (2-hex parent ID-type) + T (1-hex parent treasury) + PNNNNN (5-hex parent network)
|
||||
S3 (4) = CCC (3-hex ISO currency) + CK (1-hex XOR checksum)
|
||||
S4 (4) = First 4 hex of current 8-hex network code
|
||||
S5 (4) = Last 4 hex of current 8-hex network code
|
||||
S6 (12) = PPPP (4-hex parent monthly seq) + T (1-hex current treasury) + CCCC (4-hex current monthly seq) + RRR (3-hex nonce)
|
||||
GENESIS-07 is the public chain-root token (GENESIS_ROOT).
|
||||
Validation is by calling benjii-prime.korenet.cloud, not table lookup.
|
||||
"""
|
||||
|
||||
import os
|
||||
import uuid
|
||||
import hashlib
|
||||
import json
|
||||
import uuid
|
||||
import logging
|
||||
import requests
|
||||
import urllib3
|
||||
from datetime import datetime, timezone
|
||||
from functools import reduce
|
||||
from typing import Optional
|
||||
|
||||
urllib3.disable_warnings()
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# ── Lookup tables (placeholders — populate from Bryan's full spec) ────────────
|
||||
# ── KoreID type tokens ────────────────────────────────────────────────────────
|
||||
|
||||
# kore_id_type: maps event type names to 2-hex codes
|
||||
# TODO: replace with values from Bryan's kore_id_type SQL table
|
||||
ID_TYPES = {
|
||||
'MINT': 'a1', # SV token mint event
|
||||
'BURN': 'a2', # SV token burn/redemption event
|
||||
'TRANSFER': 'a3', # SV token transfer event
|
||||
'RESERVE': 'a4', # Reserve oracle update
|
||||
'SWAP': 'a5', # Uniswap swap event
|
||||
'ONRAMP': 'a6', # Fiat on-ramp
|
||||
'OFFRAMP': 'a7', # Fiat off-ramp
|
||||
'EXTERNAL': 'b1', # Generic external response
|
||||
}
|
||||
class KoreType:
|
||||
# Payment mirror
|
||||
TXN = 'TXN' # Transaction
|
||||
MINT = 'MINT' # Mirror token mint
|
||||
SETTLE = 'SLP' # Settlement
|
||||
REDEEM = 'RDM' # Redemption
|
||||
BURN = 'BURN' # Burn-sell
|
||||
CLOSE = 'CLO' # Close tranche
|
||||
# Oracle
|
||||
ORACLE = 'ORC' # Oracle attestation
|
||||
ANCHOR = 'ANC' # Ethereum anchor confirmation
|
||||
# Vault
|
||||
DEPOSIT = 'DEP' # Vault deposit
|
||||
WITHDRAW = 'WDR' # Vault withdrawal
|
||||
TRANSFER = 'TRF' # Vault transfer
|
||||
# Compliance
|
||||
SANCTION = 'SCN' # Sanctions check
|
||||
ENRICH = 'ENR' # Entity enrichment
|
||||
# Generic
|
||||
EVENT = 'EVT' # Generic event
|
||||
DOC = 'DOC' # Document
|
||||
CONFIRM = 'CNF' # Confirmation
|
||||
|
||||
# kore_treasury: maps treasury names to 1-hex codes
|
||||
# TODO: replace with values from Bryan's kore_treasury SQL table
|
||||
TREASURIES = {
|
||||
'VECTOR_ZULU': '1',
|
||||
'SOVEREIGN_GLOBAL': '2',
|
||||
'KORE_COLLECTIVE': '3',
|
||||
'MERIDIAN': '4',
|
||||
}
|
||||
class KoreDomain:
|
||||
PAY = 'PAY' # Payment
|
||||
SETTLE = 'SETTLE' # Settlement
|
||||
LEDGER = 'LEDGER' # Ledger
|
||||
VAULT = 'VAULT' # Vault operations
|
||||
ORACLE = 'ORACLE' # Oracle operations
|
||||
MIRROR = 'MIRROR' # Payment mirror
|
||||
ANCHOR = 'ANCHOR' # Ethereum anchoring
|
||||
COMPLY = 'COMPLY' # Compliance
|
||||
TOKEN = 'TOKEN' # Token operations
|
||||
AUDIT = 'AUDIT' # Audit trail
|
||||
|
||||
# kore_network: maps network names to 5-hex parent network codes
|
||||
# TODO: replace with values from Bryan's kore_network SQL table
|
||||
PARENT_NETWORKS = {
|
||||
'KORECHAIN': 'f4e50',
|
||||
'ETHEREUM': '0a1b0',
|
||||
'AMITIS': '03e80',
|
||||
'TRON': '00154',
|
||||
}
|
||||
GENESIS_ROOT = 'GENESIS-07'
|
||||
|
||||
# kore_network_current: maps network names to 8-hex current network codes
|
||||
# TODO: replace with values from Bryan's kore_network_current SQL table
|
||||
CURRENT_NETWORKS = {
|
||||
'ETHEREUM': '0a1b03e8',
|
||||
'AMITIS': '03e80015',
|
||||
'KORECHAIN': 'f4e5b231',
|
||||
'TRON': '001540ab',
|
||||
}
|
||||
|
||||
# kore_currency: maps ISO currency codes to 3-hex values
|
||||
# TODO: replace with values from Bryan's kore_currency SQL table
|
||||
CURRENCIES = {
|
||||
'USD': '001',
|
||||
'GBP': '002',
|
||||
'EUR': '003',
|
||||
'CHF': '004',
|
||||
'JPY': '005',
|
||||
'SGD': '006',
|
||||
'BTC': '010',
|
||||
'ETH': '011',
|
||||
'USDT': '012',
|
||||
'USDC': '013',
|
||||
'GPSC': '020',
|
||||
'AMTS': '030',
|
||||
}
|
||||
|
||||
# Parent ID type for VZ as a child of KoreChain
|
||||
PARENT_ID_TYPE = 'b2' # TODO: confirm with Bryan
|
||||
|
||||
# ── KoreID Generator ──────────────────────────────────────────────────────────
|
||||
|
||||
class KoreIDGenerator:
|
||||
"""
|
||||
Generates KoreID display numbers for events originating from Vector Zulu.
|
||||
Requires a sequence store (Postgres) for monthly sequence tracking.
|
||||
"""
|
||||
|
||||
def __init__(self, pg_conn=None):
|
||||
self.pg = pg_conn
|
||||
|
||||
def _xor_checksum(self, hex_string: str) -> str:
|
||||
"""XOR all bytes of a hex string, return 1-hex result."""
|
||||
bytes_list = [int(hex_string[i:i+2], 16) for i in range(0, len(hex_string), 2)]
|
||||
result = reduce(lambda a, b: a ^ b, bytes_list, 0)
|
||||
return format(result & 0xF, 'x')
|
||||
|
||||
def _get_sequence(self, mmyy: str, id_type: str, treasury: str,
|
||||
network: str, currency: str) -> tuple[int, int]:
|
||||
"""
|
||||
Get or create monthly sequence entry.
|
||||
Returns (parent_seq, current_seq) — increments current_seq.
|
||||
Falls back to random if no DB connection.
|
||||
"""
|
||||
if not self.pg:
|
||||
import random
|
||||
return (random.randint(0, 0xFFFF), random.randint(0, 0xFFFF))
|
||||
|
||||
try:
|
||||
with self.pg.cursor() as cur:
|
||||
cur.execute("""
|
||||
INSERT INTO kore_sequence (mmyy, ct, t, net, ccc, parent_seq, current_seq)
|
||||
VALUES (%s, %s, %s, %s, %s, 0, 1)
|
||||
ON CONFLICT (mmyy, ct, t, net, ccc)
|
||||
DO UPDATE SET current_seq = kore_sequence.current_seq + 1
|
||||
RETURNING parent_seq, current_seq
|
||||
""", (mmyy, id_type, treasury, network, currency))
|
||||
row = cur.fetchone()
|
||||
self.pg.commit()
|
||||
return (row[0], row[1])
|
||||
except Exception as e:
|
||||
log.error(f"Sequence DB error: {e}")
|
||||
import random
|
||||
return (random.randint(0, 0xFFFF), random.randint(0, 0xFFFF))
|
||||
|
||||
def generate(
|
||||
self,
|
||||
event_type: str,
|
||||
currency: str,
|
||||
current_network: str = 'ETHEREUM',
|
||||
parent_network: str = 'KORECHAIN',
|
||||
treasury: str = 'VECTOR_ZULU',
|
||||
parent_treasury: str = 'KORE_COLLECTIVE',
|
||||
nonce: str = None,
|
||||
type_token: str,
|
||||
domain_token: str,
|
||||
payload: dict,
|
||||
parent: str = GENESIS_ROOT,
|
||||
date: Optional[datetime] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Generate a KoreID for a Vector Zulu event.
|
||||
Generate a KoreID.
|
||||
|
||||
Returns dict with:
|
||||
- koreid: the display ID string
|
||||
- guid: UUIDv4 for internal storage
|
||||
- metadata: all components for traceability
|
||||
Args:
|
||||
type_token: e.g. KoreType.TXN
|
||||
domain_token: e.g. KoreDomain.PAY
|
||||
payload: the event data dict
|
||||
parent: parent KoreID or GENESIS-07
|
||||
date: UTC datetime (defaults to now)
|
||||
|
||||
Returns:
|
||||
dict with koreid, guid, hash, components, payload
|
||||
"""
|
||||
now = datetime.now(timezone.utc)
|
||||
mmyy = now.strftime('%m%y')
|
||||
ddmmyy = now.strftime('%d%m%y')
|
||||
if date is None:
|
||||
date = datetime.now(timezone.utc)
|
||||
|
||||
# Resolve codes
|
||||
ct = ID_TYPES.get(event_type.upper(), 'ff')
|
||||
pt = PARENT_ID_TYPE
|
||||
t = TREASURIES.get(treasury.upper(), '0')
|
||||
pt_val = TREASURIES.get(parent_treasury.upper(), '0')
|
||||
pnnnnn = PARENT_NETWORKS.get(parent_network.upper(), '00000')
|
||||
ccc = CURRENCIES.get(currency.upper(), '000')
|
||||
net8 = CURRENT_NETWORKS.get(current_network.upper(), '00000000')
|
||||
date_str = date.strftime('%Y%m%d')
|
||||
|
||||
# S1: DDMMYY + CT
|
||||
s1 = ddmmyy + ct # 8 chars
|
||||
# Canonical payload for hashing
|
||||
canonical = json.dumps({
|
||||
'payload': payload,
|
||||
'parent': parent,
|
||||
'type': type_token,
|
||||
'domain': domain_token,
|
||||
'date': date_str,
|
||||
}, sort_keys=True, separators=(',', ':'))
|
||||
|
||||
# S2: PT + T(parent) + PNNNNN
|
||||
s2 = pt + pt_val + pnnnnn # 8 chars
|
||||
# First 16 hex chars of SHA256
|
||||
hash_val = hashlib.sha256(canonical.encode()).hexdigest()[:16]
|
||||
|
||||
# S3: CCC + checksum
|
||||
ck = self._xor_checksum(ccc + '0') # XOR of currency bytes
|
||||
s3 = ccc + ck # 4 chars
|
||||
|
||||
# S4 / S5: split current network code
|
||||
s4 = net8[:4] # 4 chars
|
||||
s5 = net8[4:] # 4 chars
|
||||
|
||||
# S6: PPPP + T(current) + CCCC + RRR
|
||||
parent_seq, current_seq = self._get_sequence(mmyy, ct, t, pnnnnn, ccc)
|
||||
pppp = format(parent_seq & 0xFFFF, '04x')
|
||||
cccc = format(current_seq & 0xFFFF, '04x')
|
||||
|
||||
if nonce is None:
|
||||
import random
|
||||
nonce = format(random.randint(0, 0xFFF), '03x')
|
||||
|
||||
s6 = pppp + t + cccc + nonce # 12 chars
|
||||
|
||||
koreid = f"{s1}/{s2}-{s3}-{s4}-{s5}-{s6}"
|
||||
koreid = f"KID-{type_token}-{domain_token}-{date_str}-{hash_val}"
|
||||
guid = str(uuid.uuid4())
|
||||
|
||||
return {
|
||||
'koreid': koreid,
|
||||
'guid': guid,
|
||||
'koreid': koreid,
|
||||
'guid': guid,
|
||||
'hash': hash_val,
|
||||
'parent': parent,
|
||||
'components': {
|
||||
's1': s1, 's2': s2, 's3': s3,
|
||||
's4': s4, 's5': s5, 's6': s6,
|
||||
'type': type_token,
|
||||
'domain': domain_token,
|
||||
'date': date_str,
|
||||
'hash': hash_val,
|
||||
},
|
||||
'metadata': {
|
||||
'event_type': event_type,
|
||||
'currency': currency,
|
||||
'current_network': current_network,
|
||||
'parent_network': parent_network,
|
||||
'treasury': treasury,
|
||||
'generated_at': now.isoformat(),
|
||||
'id_type_code': ct,
|
||||
'parent_seq': parent_seq,
|
||||
'current_seq': current_seq,
|
||||
}
|
||||
'payload': payload,
|
||||
'generated_at': date.isoformat(),
|
||||
}
|
||||
|
||||
def for_mirror_mint(self, uetr: str, amount: float,
|
||||
currency: str, vault: str, parent: str = GENESIS_ROOT) -> dict:
|
||||
return self.generate(
|
||||
KoreType.MINT, KoreDomain.MIRROR,
|
||||
{'uetr': uetr, 'amount': amount, 'currency': currency, 'vault': vault},
|
||||
parent=parent
|
||||
)
|
||||
|
||||
def wrap_ethereum_event(
|
||||
event_type: str,
|
||||
amount: float,
|
||||
currency: str,
|
||||
tx_hash: str,
|
||||
block_number: int,
|
||||
from_address: str,
|
||||
to_address: str,
|
||||
token_symbol: str,
|
||||
koreid_generator: KoreIDGenerator,
|
||||
linked_koreid: str = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Convert an Ethereum event into a KoreID-formatted payload
|
||||
for posting back to KoreChain via KoreOracle.
|
||||
def for_mirror_settle(self, uetr: str, parent_koreid: str) -> dict:
|
||||
return self.generate(
|
||||
KoreType.SETTLE, KoreDomain.SETTLE,
|
||||
{'uetr': uetr},
|
||||
parent=parent_koreid
|
||||
)
|
||||
|
||||
Preserves full traceability per Bryan's spec:
|
||||
- KoreID reference
|
||||
- hash (TX hash)
|
||||
- signature (block + tx proof)
|
||||
- chain linkage
|
||||
"""
|
||||
kid = koreid_generator.generate(
|
||||
event_type=event_type,
|
||||
currency=currency,
|
||||
current_network='ETHEREUM',
|
||||
parent_network='KORECHAIN',
|
||||
)
|
||||
def for_mirror_redeem(self, uetr: str, amount: float, parent_koreid: str) -> dict:
|
||||
return self.generate(
|
||||
KoreType.REDEEM, KoreDomain.PAY,
|
||||
{'uetr': uetr, 'amount': amount},
|
||||
parent=parent_koreid
|
||||
)
|
||||
|
||||
payload = {
|
||||
'koreid': kid['koreid'],
|
||||
'guid': kid['guid'],
|
||||
'value': amount,
|
||||
'proof': {
|
||||
'hash': tx_hash,
|
||||
'block_number': block_number,
|
||||
'chain_reference': kid['koreid'],
|
||||
'network': 'ETHEREUM',
|
||||
'network_code': CURRENT_NETWORKS.get('ETHEREUM'),
|
||||
},
|
||||
'source': {
|
||||
'type': 'ETHEREUM_EVENT',
|
||||
'event_type': event_type,
|
||||
'token': token_symbol,
|
||||
'from_address': from_address,
|
||||
'to_address': to_address,
|
||||
'tx_hash': tx_hash,
|
||||
},
|
||||
'linked_to': linked_koreid,
|
||||
'generated_at': datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
def for_mirror_burn(self, uetr: str, amount: float,
|
||||
venue: str, parent_koreid: str) -> dict:
|
||||
return self.generate(
|
||||
KoreType.BURN, KoreDomain.MIRROR,
|
||||
{'uetr': uetr, 'amount': amount, 'venue': venue},
|
||||
parent=parent_koreid
|
||||
)
|
||||
|
||||
return payload
|
||||
def for_mirror_close(self, uetr: str, parent_koreid: str) -> dict:
|
||||
return self.generate(
|
||||
KoreType.CLOSE, KoreDomain.MIRROR,
|
||||
{'uetr': uetr},
|
||||
parent=parent_koreid
|
||||
)
|
||||
|
||||
def for_oracle_anchor(self, tx_hash: str, balance: float,
|
||||
round_id: int, parent: str = GENESIS_ROOT) -> dict:
|
||||
return self.generate(
|
||||
KoreType.ORACLE, KoreDomain.ORACLE,
|
||||
{'tx_hash': tx_hash, 'balance': balance, 'round_id': round_id},
|
||||
parent=parent
|
||||
)
|
||||
|
||||
def for_ethereum_anchor(self, original_koreid: str, tx_hash: str,
|
||||
block_number: int, event_type: str) -> dict:
|
||||
return self.generate(
|
||||
KoreType.ANCHOR, KoreDomain.ANCHOR,
|
||||
{'tx_hash': tx_hash, 'block_number': block_number, 'event_type': event_type},
|
||||
parent=original_koreid
|
||||
)
|
||||
|
||||
def for_vault_event(self, vault_id: str, amount: float,
|
||||
currency: str, event_type: str) -> dict:
|
||||
type_map = {
|
||||
'DEPOSIT': KoreType.DEPOSIT,
|
||||
'WITHDRAW': KoreType.WITHDRAW,
|
||||
'TRANSFER': KoreType.TRANSFER,
|
||||
}
|
||||
return self.generate(
|
||||
type_map.get(event_type, KoreType.EVENT), KoreDomain.VAULT,
|
||||
{'vault_id': vault_id, 'amount': amount, 'currency': currency}
|
||||
)
|
||||
|
||||
|
||||
# ── KoreID Parser ─────────────────────────────────────────────────────────────
|
||||
|
||||
class KoreIDParser:
|
||||
|
||||
def parse(self, koreid: str) -> Optional[dict]:
|
||||
"""
|
||||
Parse a KoreID string into components.
|
||||
Returns None if invalid format.
|
||||
"""
|
||||
try:
|
||||
if not koreid.startswith('KID-'):
|
||||
return None
|
||||
parts = koreid.split('-')
|
||||
if len(parts) != 5:
|
||||
return None
|
||||
_, type_token, domain_token, date_str, hash_val = parts
|
||||
if len(hash_val) != 16:
|
||||
return None
|
||||
date = datetime.strptime(date_str, '%Y%m%d').replace(tzinfo=timezone.utc)
|
||||
return {
|
||||
'koreid': koreid,
|
||||
'type': type_token,
|
||||
'domain': domain_token,
|
||||
'date': date_str,
|
||||
'date_parsed': date.isoformat(),
|
||||
'hash': hash_val,
|
||||
'structurally_valid': True,
|
||||
}
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def is_valid_format(self, koreid: str) -> bool:
|
||||
return self.parse(koreid) is not None
|
||||
|
||||
def get_type(self, koreid: str) -> Optional[str]:
|
||||
parsed = self.parse(koreid)
|
||||
return parsed['type'] if parsed else None
|
||||
|
||||
def get_domain(self, koreid: str) -> Optional[str]:
|
||||
parsed = self.parse(koreid)
|
||||
return parsed['domain'] if parsed else None
|
||||
|
||||
|
||||
# ── KoreID Validator (calls benjii-prime) ────────────────────────────────────
|
||||
|
||||
class KoreIDValidator:
|
||||
|
||||
BENJII_BASE = 'https://benjii-prime.korenet.cloud'
|
||||
|
||||
def __init__(self, cert_path: str, key_path: str):
|
||||
self.cert = (cert_path, key_path)
|
||||
|
||||
def _post(self, endpoint: str, payload: dict) -> Optional[dict]:
|
||||
try:
|
||||
resp = requests.post(
|
||||
f'{self.BENJII_BASE}{endpoint}',
|
||||
cert=self.cert,
|
||||
verify=False,
|
||||
json=payload,
|
||||
timeout=15,
|
||||
headers={'Content-Type': 'application/json'}
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
return resp.json()
|
||||
log.warning(f"Validator {endpoint} returned {resp.status_code}")
|
||||
return None
|
||||
except Exception as e:
|
||||
log.error(f"Validator error: {e}")
|
||||
return None
|
||||
|
||||
def validate_structure(self, koreid: str) -> Optional[dict]:
|
||||
"""POST /api/v1/koreid/validate — structure check only."""
|
||||
return self._post('/api/v1/koreid/validate', {'kore_id': koreid})
|
||||
|
||||
def verify_integrity(self, koreid: str, type_token: str,
|
||||
domain_token: str, payload: dict) -> Optional[dict]:
|
||||
"""POST /api/v1/koreid/verify — boolean integrity check."""
|
||||
return self._post('/api/v1/koreid/verify', {
|
||||
'kore_id': koreid,
|
||||
'type': type_token,
|
||||
'domain': domain_token,
|
||||
'payload': payload,
|
||||
})
|
||||
|
||||
def verify_lineage(self, koreid: str, type_token: str = None,
|
||||
domain_token: str = None, payload: dict = None,
|
||||
parent: str = GENESIS_ROOT) -> Optional[dict]:
|
||||
"""
|
||||
POST /api/v1/koreid/verify-lineage
|
||||
Full validation including GENESIS-07 anchoring.
|
||||
Phase-4 requirement: treat as GENESIS-07-anchored only when
|
||||
integrity_verified=True AND genesis_anchored=True.
|
||||
"""
|
||||
body = {'kore_id': koreid, 'parent': parent}
|
||||
if type_token: body['type'] = type_token
|
||||
if domain_token: body['domain'] = domain_token
|
||||
if payload: body['payload'] = payload
|
||||
|
||||
result = self._post('/api/v1/koreid/verify-lineage', body)
|
||||
if result:
|
||||
anchored = (
|
||||
result.get('integrity_verified') is True and
|
||||
result.get('genesis_anchored') is True
|
||||
)
|
||||
result['phase4_compliant'] = anchored
|
||||
if not anchored:
|
||||
log.warning(f"KoreID {koreid} failed Phase-4 check: {result.get('reason')}")
|
||||
return result
|
||||
|
||||
def is_genesis_anchored(self, koreid: str, type_token: str,
|
||||
domain_token: str, payload: dict) -> bool:
|
||||
"""
|
||||
Convenience method — returns True only if Phase-4 compliant.
|
||||
Fail-closed: any error returns False.
|
||||
"""
|
||||
try:
|
||||
result = self.verify_lineage(koreid, type_token, domain_token, payload)
|
||||
return result.get('phase4_compliant', False) if result else False
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Test generation
|
||||
gen = KoreIDGenerator()
|
||||
result = gen.generate('MINT', 'GBP', 'ETHEREUM', 'KORECHAIN', 'VECTOR_ZULU')
|
||||
print("Generated KoreID:", result['koreid'])
|
||||
print("GUID:", result['guid'])
|
||||
print("Components:", result['components'])
|
||||
|
||||
# Test event wrapping
|
||||
payload = wrap_ethereum_event(
|
||||
event_type='MINT',
|
||||
amount=1000000.00,
|
||||
currency='GBP',
|
||||
tx_hash='0xabc123def456',
|
||||
block_number=25375588,
|
||||
from_address='0x0000000000000000000000000000000000000000',
|
||||
to_address='0x8B59352944134D6c2e175d92Bb07b25443163A53',
|
||||
token_symbol='SVGBP',
|
||||
koreid_generator=gen,
|
||||
)
|
||||
import json
|
||||
print("\nWrapped payload:")
|
||||
print(json.dumps(payload, indent=2))
|
||||
|
||||
gen = KoreIDGenerator()
|
||||
parser = KoreIDParser()
|
||||
|
||||
# Test generation
|
||||
print("=== KoreID Generation ===")
|
||||
|
||||
mint_kid = gen.for_mirror_mint(
|
||||
uetr='550e8400-e29b-41d4-a716-446655440000',
|
||||
amount=10000000,
|
||||
currency='USD',
|
||||
vault='VZ-USD-VAULT'
|
||||
)
|
||||
print(f"MINT: {mint_kid['koreid']}")
|
||||
|
||||
settle_kid = gen.for_mirror_settle(
|
||||
uetr='550e8400-e29b-41d4-a716-446655440000',
|
||||
parent_koreid=mint_kid['koreid']
|
||||
)
|
||||
print(f"SETTLE: {settle_kid['koreid']}")
|
||||
|
||||
oracle_kid = gen.for_oracle_anchor(
|
||||
tx_hash='0xabc123def456',
|
||||
balance=49918000000,
|
||||
round_id=47
|
||||
)
|
||||
print(f"ORACLE: {oracle_kid['koreid']}")
|
||||
|
||||
# Test parsing
|
||||
print("\n=== KoreID Parsing ===")
|
||||
parsed = parser.parse(mint_kid['koreid'])
|
||||
print(f"Type: {parsed['type']} | Domain: {parsed['domain']} | Date: {parsed['date']}")
|
||||
print(f"Valid: {parsed['structurally_valid']}")
|
||||
|
||||
# Test invalid
|
||||
print(f"Invalid: {parser.is_valid_format('NOT-A-KOREID')}")
|
||||
|
||||
Reference in New Issue
Block a user