354 lines
12 KiB
Python
354 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Vector Zulu — KoreID Generator & Validator v2.0
|
|
Correct format per VZ-Welcome-Pack FINAL v1.0 (2026-07-08)
|
|
|
|
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
|
|
|
|
GENESIS-07 is the public chain-root token (GENESIS_ROOT).
|
|
Validation is by calling benjii-prime.korenet.cloud, not table lookup.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import uuid
|
|
import logging
|
|
import requests
|
|
import urllib3
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
urllib3.disable_warnings()
|
|
log = logging.getLogger(__name__)
|
|
|
|
# ── KoreID type tokens ────────────────────────────────────────────────────────
|
|
|
|
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
|
|
|
|
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
|
|
|
|
GENESIS_ROOT = 'GENESIS-07'
|
|
|
|
|
|
# ── KoreID Generator ──────────────────────────────────────────────────────────
|
|
|
|
class KoreIDGenerator:
|
|
|
|
def generate(
|
|
self,
|
|
type_token: str,
|
|
domain_token: str,
|
|
payload: dict,
|
|
parent: str = GENESIS_ROOT,
|
|
date: Optional[datetime] = None,
|
|
) -> dict:
|
|
"""
|
|
Generate a KoreID.
|
|
|
|
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
|
|
"""
|
|
if date is None:
|
|
date = datetime.now(timezone.utc)
|
|
|
|
date_str = date.strftime('%Y%m%d')
|
|
|
|
# Canonical payload for hashing
|
|
canonical = json.dumps({
|
|
'payload': payload,
|
|
'parent': parent,
|
|
'type': type_token,
|
|
'domain': domain_token,
|
|
'date': date_str,
|
|
}, sort_keys=True, separators=(',', ':'))
|
|
|
|
# First 16 hex chars of SHA256
|
|
hash_val = hashlib.sha256(canonical.encode()).hexdigest()[:16]
|
|
|
|
koreid = f"KID-{type_token}-{domain_token}-{date_str}-{hash_val}"
|
|
guid = str(uuid.uuid4())
|
|
|
|
return {
|
|
'koreid': koreid,
|
|
'guid': guid,
|
|
'hash': hash_val,
|
|
'parent': parent,
|
|
'components': {
|
|
'type': type_token,
|
|
'domain': domain_token,
|
|
'date': date_str,
|
|
'hash': hash_val,
|
|
},
|
|
'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 for_mirror_settle(self, uetr: str, parent_koreid: str) -> dict:
|
|
return self.generate(
|
|
KoreType.SETTLE, KoreDomain.SETTLE,
|
|
{'uetr': uetr},
|
|
parent=parent_koreid
|
|
)
|
|
|
|
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
|
|
)
|
|
|
|
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
|
|
)
|
|
|
|
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__':
|
|
import json
|
|
|
|
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')}")
|