# v0.2.16 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * from genlayer.gl.vm import UserError from dataclasses import dataclass import json from urllib.parse import urlparse from datetime import datetime def _addr_str(addr: Address) -> str: try: return addr.as_hex.lower() except Exception: return str(addr).lower() def _extract_origin(url: str) -> tuple: u = url.strip() if not (u.startswith("http://") or u.startswith("https://")): raise UserError("URL must start with http:// or https://") try: parsed = urlparse(u) except Exception: raise UserError("Invalid URL format") scheme = parsed.scheme.lower() if scheme not in ("http", "https"): raise UserError("Only http and https protocols are supported") if parsed.username is not None or parsed.password is not None: raise UserError("URL credentials are not allowed") hostname = parsed.hostname if not hostname: raise UserError("URL missing valid hostname") hostname = hostname.lower().strip() if not hostname or ".." in hostname or hostname.startswith(".") or hostname.endswith("."): raise UserError("Ambiguous or invalid hostname") port = parsed.port if port is None: port = 80 if scheme == "http" else 443 return scheme, hostname, port def _is_origin_valid(target_url: str, base_url: str) -> bool: t_scheme, t_host, t_port = _extract_origin(target_url) b_scheme, b_host, b_port = _extract_origin(base_url) if t_scheme != b_scheme or t_port != b_port: return False if t_host == b_host: return True if t_host.endswith("." + b_host): return True return False def _parse_llm_json(text) -> dict: if isinstance(text, dict): return text if hasattr(text, "content"): text = text.content try: cleaned = str(text).strip() if cleaned.startswith("```json"): cleaned = cleaned[7:] elif cleaned.startswith("```"): cleaned = cleaned[3:] if cleaned.endswith("```"): cleaned = cleaned[:-3] return json.loads(cleaned.strip()) except Exception as e: return {"verdict": "ABORT", "confidence": 0, "reason": f"Parse error: {str(e)}"} def _safe_parse(raw) -> dict: data = _parse_llm_json(raw) if not isinstance(data, dict): return None verdict = str(data.get("verdict", "")).strip().upper() if verdict not in ("CLEARED_COMPLIANT", "SANCTIONED_BLOCKED", "SUSPICIOUS_WATCHLIST", "ABORT"): return None conf = data.get("confidence", 0) if isinstance(conf, float): conf = int(conf) if not isinstance(conf, int) or not (0 <= conf <= 100): return None reason = str(data.get("reason", "")) # Unified 75% confidence threshold across entire pipeline if conf < 75 and verdict != "ABORT": verdict = "ABORT" reason = f"[low_confidence: {conf}%] " + reason return { "verdict": verdict, "confidence": conf, "reason": reason[:300], } @allow_storage @dataclass class ComplianceRecord: target_identifier: str entity_jurisdiction: str compliance_status: str audit_confidence: bigint legal_justification: str last_audited_epoch: bigint class Contract(gl.Contract): compliance_cache: TreeMap[str, ComplianceRecord] total_screened_entities: bigint official_sanction_registry_base: str compliance_officer: str clearance_validity_duration: bigint def __init__(self, compliance_officer: str = ""): self.total_screened_entities = bigint(0) self.clearance_validity_duration = bigint(2592000) officer_input = compliance_officer.strip() if officer_input.startswith("http://") or officer_input.startswith("https://"): _extract_origin(officer_input) self.official_sanction_registry_base = officer_input self.compliance_officer = _addr_str(gl.message.sender_address) elif officer_input: try: Address(officer_input) self.compliance_officer = officer_input.lower() except Exception: self.compliance_officer = _addr_str(gl.message.sender_address) self.official_sanction_registry_base = "https://sanctionssearch.ofac.treas.gov" else: self.compliance_officer = _addr_str(gl.message.sender_address) self.official_sanction_registry_base = "https://sanctionssearch.ofac.treas.gov" def _get_current_timestamp(self) -> bigint: """Derive trusted timestamp from GenLayer transaction execution context (gl.message_raw). SECURITY: Raises UserError if trusted datetime is missing or malformed - never defaults to 0, as that would allow an attacker to bypass all time-based guards.""" try: dt_str = str(gl.message_raw.get("datetime", "")) if not dt_str: raise UserError("Trusted timestamp unavailable: gl.message_raw missing 'datetime' field") dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00")) ts = bigint(int(dt.timestamp())) if ts <= bigint(0): raise UserError("Trusted timestamp resolved to zero or negative - refusing to proceed") return ts except UserError: raise except Exception as e: raise UserError(f"Trusted timestamp parse failure: {str(e)}") def _construct_or_validate_evidence_url(self, target_id_clean: str, live_url: str) -> str: """Bind registry evidence to audited entity. Ensures search URL strictly binds to target identifier and official registry origin.""" target_url = live_url.strip() if not target_url: return f"{self.official_sanction_registry_base}/default.aspx?name={target_id_clean}" if not _is_origin_valid(target_url, self.official_sanction_registry_base): raise UserError("Sanction search URL does not match registered official sanction registry origin") if target_id_clean not in target_url.lower(): raise UserError("Sanction search URL does not bind to target identifier") return target_url @gl.public.write def request_compliance_audit( self, target_identifier: str, entity_jurisdiction: str, live_sanction_search_url: str = "", ) -> str: """Autonomous AML/Sanction screening audit with evidence binding and cache protection.""" target_id_clean = target_identifier.strip().lower() jurisdiction_clean = entity_jurisdiction.strip().upper() if len(target_id_clean) < 3: raise UserError("Target identifier too short (minimum 3 characters)") if len(jurisdiction_clean) < 2: raise UserError("Jurisdiction code too short") current_ts = self._get_current_timestamp() # Cache Protection & Anti-Tampering: # 1. Strictly prevent overwriting SANCTIONED_BLOCKED entities without officer intervention if target_id_clean in self.compliance_cache: existing_rec = self.compliance_cache[target_id_clean] if existing_rec.compliance_status == "SANCTIONED_BLOCKED": raise UserError("Entity is sanctioned and blocked; cannot re-audit without compliance officer resolution") # 2. Prevent arbitrary callers from overwriting valid, fresh CLEARED_COMPLIANT records if existing_rec.compliance_status == "CLEARED_COMPLIANT": if (current_ts - existing_rec.last_audited_epoch) <= self.clearance_validity_duration: raise UserError("Entity already holds valid fresh clearance; re-audit disallowed until expiration") # Evidence Binding: construct deterministic search URL or validate target binding in caller URL search_url = self._construct_or_validate_evidence_url(target_id_clean, live_sanction_search_url) target_str = str(target_id_clean) jur_str = str(jurisdiction_clean) u_search = str(search_url) def leader_fn(): try: res = gl.nondet.web.render(u_search, mode="text") page_text = res.content if hasattr(res, "content") else str(res) if not page_text or len(page_text.strip()) < 30: return {"verdict": "ABORT", "confidence": 0, "reason": "Sanction registry page returned empty text"} if any(err in page_text[:400].lower() for err in ["404 not found", "error 404", "service unavailable"]): return {"verdict": "ABORT", "confidence": 0, "reason": "Sanction registry page unreachable/404"} except Exception as e: return {"verdict": "ABORT", "confidence": 0, "reason": f"Web fetch error: {str(e)}"} prompt = f""" SYSTEM: You are the Autonomous International Sanctions & Financial Crime Compliance Oracle. Audit whether the target entity appears on international sanction watchlists, SDN lists, or export control embargoes. TARGET ENTITY IDENTIFIER: {target_str} DECLARED JURISDICTION: {jur_str} LIVE OFFICIAL SANCTIONS REGISTRY EVIDENCE: {page_text[:4000]} EVIDENCE BINDING VALIDATION: 1. Verify that the search evidence above strictly and explicitly corresponds to TARGET ENTITY IDENTIFIER '{target_str}'. 2. If the page reflects a search for a different entity, or does not contain query evidence for '{target_str}', you MUST output verdict 'ABORT' with reason 'evidence_target_mismatch'. 3. Do NOT clear or sanction any entity if the evidence does not explicitly bind to '{target_str}'. Classification Rules: - SANCTIONED_BLOCKED (conf >= 75): Exact match or verified entity link on active sanction lists (terrorism, cyber attacks, state sanctions). - SUSPICIOUS_WATCHLIST (conf >= 75): Fuzzy name match, secondary watchlists, PEP (Politically Exposed Person), or unconfirmed nexus requiring enhanced due diligence. - CLEARED_COMPLIANT (conf >= 75): Search confirms zero matches, clean record, or explicit non-sanctioned certification on official registry for '{target_str}'. - ABORT: Evidence target mismatch, portal requires login, captcha, rate-limited, or query failed. OUTPUT ONLY STRICT JSON: {{ "verdict": "CLEARED_COMPLIANT" | "SANCTIONED_BLOCKED" | "SUSPICIOUS_WATCHLIST" | "ABORT", "confidence": 0-100, "reason": "max 300 chars technical regulatory justification" }} """ try: raw1 = gl.nondet.exec_prompt(prompt, response_format="json") raw2 = gl.nondet.exec_prompt(prompt, response_format="json") p1 = _safe_parse(raw1) p2 = _safe_parse(raw2) if p1 is None or p2 is None: return {"verdict": "ABORT", "confidence": 0, "reason": "parse_failed"} if p1["verdict"] != p2["verdict"]: return {"verdict": "ABORT", "confidence": 0, "reason": "multi_sample_divergence"} p1["confidence"] = (p1["confidence"] + p2["confidence"]) // 2 return p1 except Exception as e: return {"verdict": "ABORT", "confidence": 0, "reason": f"LLM error: {str(e)}"} def validator_fn(leader_res) -> bool: if not isinstance(leader_res, gl.vm.Return): return False leader_data = leader_res.calldata if hasattr(leader_res, "calldata") else leader_res leader = _safe_parse(leader_data) if leader is None: return False mine = _safe_parse(leader_fn()) if mine is None: return False return ( mine["verdict"] == leader["verdict"] and (mine["confidence"] >= 75) == (leader["confidence"] >= 75) ) result_raw = gl.vm.run_nondet(leader_fn, validator_fn) result = _safe_parse(result_raw) if result is None: result = {"verdict": "ABORT", "confidence": 0, "reason": "adjudication_failed"} verdict = result["verdict"] confidence = result["confidence"] reason = result["reason"] # Deterministic post-consensus normalization if confidence < 75 and verdict != "ABORT": verdict = "ABORT" if target_id_clean not in self.compliance_cache: self.total_screened_entities += bigint(1) record = ComplianceRecord( target_identifier=target_id_clean, entity_jurisdiction=jurisdiction_clean, compliance_status="PENDING", audit_confidence=bigint(0), legal_justification="", last_audited_epoch=bigint(0), ) else: record = self.compliance_cache[target_id_clean] record.entity_jurisdiction = jurisdiction_clean record.audit_confidence = bigint(confidence) record.legal_justification = reason record.last_audited_epoch = current_ts if verdict in ("CLEARED_COMPLIANT", "SANCTIONED_BLOCKED", "SUSPICIOUS_WATCHLIST"): record.compliance_status = verdict else: record.compliance_status = "ESCALATED" self.compliance_cache[target_id_clean] = record return record.compliance_status @gl.public.write def resolve_escalated_screening( self, target_identifier: str, manual_status: str, override_reason: str, ) -> None: """Compliance officer resolves an ESCALATED screening caused by network or registry blocks.""" target_id_clean = target_identifier.strip().lower() if target_id_clean not in self.compliance_cache: raise UserError("Entity record not found") record = self.compliance_cache[target_id_clean] if record.compliance_status != "ESCALATED": raise UserError("Entity record is not in ESCALATED state") sender = _addr_str(gl.message.sender_address) if sender != self.compliance_officer: raise UserError("Only authorized compliance officer can resolve escalated screenings") manual_status_upper = manual_status.strip().upper() if manual_status_upper not in ("CLEARED_COMPLIANT", "SANCTIONED_BLOCKED", "SUSPICIOUS_WATCHLIST"): raise UserError("Invalid compliance status assignment") record.compliance_status = manual_status_upper record.legal_justification = f"Officer manual override ({sender}): {override_reason[:200]}" record.last_audited_epoch = self._get_current_timestamp() self.compliance_cache[target_id_clean] = record @gl.public.write def officer_override_sanction( self, target_identifier: str, new_status: str, appeal_legal_rationale: str, ) -> None: """Authorized compliance officer appeal / override for SANCTIONED_BLOCKED or contested entities.""" target_id_clean = target_identifier.strip().lower() if target_id_clean not in self.compliance_cache: raise UserError("Entity record not found") sender = _addr_str(gl.message.sender_address) if sender != self.compliance_officer: raise UserError("Only authorized compliance officer can override sanction determinations") new_status_upper = new_status.strip().upper() if new_status_upper not in ("CLEARED_COMPLIANT", "SANCTIONED_BLOCKED", "SUSPICIOUS_WATCHLIST", "ESCALATED"): raise UserError("Invalid target compliance status") record = self.compliance_cache[target_id_clean] record.compliance_status = new_status_upper record.legal_justification = f"Officer appeal/override ({sender}): {appeal_legal_rationale[:200]}" record.last_audited_epoch = self._get_current_timestamp() self.compliance_cache[target_id_clean] = record @gl.public.view def is_entity_cleared(self, target_identifier: str) -> bool: """Lightweight boolean gatekeeper interface for other DeFi/Bridge contracts to query. Returns True only if entity has CLEARED_COMPLIANT status AND clearance is fresh.""" target_id_clean = target_identifier.strip().lower() if target_id_clean not in self.compliance_cache: return False rec = self.compliance_cache[target_id_clean] if rec.compliance_status != "CLEARED_COMPLIANT": return False current_ts = self._get_current_timestamp() if current_ts < rec.last_audited_epoch: return False if (current_ts - rec.last_audited_epoch) > self.clearance_validity_duration: return False return True @gl.public.view def is_clearance_fresh(self, target_identifier: str) -> bool: """Check if target entity holds an unexpired compliance record.""" target_id_clean = target_identifier.strip().lower() if target_id_clean not in self.compliance_cache: return False rec = self.compliance_cache[target_id_clean] if rec.last_audited_epoch <= bigint(0): return False current_ts = self._get_current_timestamp() if current_ts < rec.last_audited_epoch: return False return (current_ts - rec.last_audited_epoch) <= self.clearance_validity_duration @gl.public.view def get_clearance_age(self, target_identifier: str) -> int: """Returns age of compliance audit in seconds.""" target_id_clean = target_identifier.strip().lower() if target_id_clean not in self.compliance_cache: raise UserError("Entity record not found") rec = self.compliance_cache[target_id_clean] if rec.last_audited_epoch <= bigint(0): raise UserError("Entity has not completed a valid audit epoch") current_ts = self._get_current_timestamp() if current_ts < rec.last_audited_epoch: return 0 return int(current_ts - rec.last_audited_epoch) @gl.public.view def get_clearance_validity_duration(self) -> int: return int(self.clearance_validity_duration) @gl.public.write def update_clearance_validity_duration(self, new_duration_seconds: int) -> None: sender = _addr_str(gl.message.sender_address) if sender != self.compliance_officer: raise UserError("Only authorized compliance officer can update validity duration") if new_duration_seconds < 3600: raise UserError("Clearance validity duration must be at least 1 hour (3600 seconds)") if new_duration_seconds > 31536000: raise UserError("Clearance validity duration cannot exceed 365 days (31536000 seconds)") self.clearance_validity_duration = bigint(new_duration_seconds) @gl.public.view def get_compliance_record(self, target_identifier: str) -> str: """Full audit detail query interface for auditing and compliance verification.""" target_id_clean = target_identifier.strip().lower() if target_id_clean not in self.compliance_cache: raise UserError("Entity record not found") rec = self.compliance_cache[target_id_clean] return json.dumps({ "target_identifier": rec.target_identifier, "entity_jurisdiction": rec.entity_jurisdiction, "compliance_status": rec.compliance_status, "audit_confidence": str(rec.audit_confidence), "legal_justification": rec.legal_justification, "last_audited_epoch": str(rec.last_audited_epoch), "clearance_validity_duration": str(self.clearance_validity_duration), }) @gl.public.view def get_total_screened_entities(self) -> int: return int(self.total_screened_entities) @gl.public.view def get_registry_base(self) -> str: return self.official_sanction_registry_base @gl.public.view def get_compliance_officer(self) -> str: return self.compliance_officer @gl.public.write def update_registry_base(self, new_registry_base: str) -> None: sender = _addr_str(gl.message.sender_address) if sender != self.compliance_officer: raise UserError("Only authorized compliance officer can update registry base") new_base = new_registry_base.strip() _extract_origin(new_base) self.official_sanction_registry_base = new_base @gl.public.write def update_compliance_officer(self, new_officer: str) -> None: sender = _addr_str(gl.message.sender_address) if sender != self.compliance_officer: raise UserError("Only authorized compliance officer can transfer authority") officer_clean = new_officer.strip() try: Address(officer_clean) except Exception: raise UserError("Invalid officer address format") self.compliance_officer = officer_clean.lower()