diff --git a/koreid.py b/koreid.py index 49e25b6..f893447 100644 --- a/koreid.py +++ b/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: /---- -Example: 061925a1/b231f4e5-3487-0a1b-03e8-001540009abc +KoreID format: KID---- + 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')}")