# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * from dataclasses import dataclass import json # Compatibility guard: Ensure gl.UserError is available across GenVM and gltest direct environments if not hasattr(gl, "UserError"): try: gl.UserError = gl.vm.UserError except Exception: class UserError(Exception): pass gl.UserError = UserError def _addr_str(addr: Address) -> str: """Safely format an Address instance into a lowercase hex string.""" try: return addr.as_hex.lower() except Exception: return str(addr).lower() def _get_sender() -> Address: """Safely obtain transaction sender across GenVM runtime versions.""" try: return gl.message.sender except Exception: try: return gl.message.sender_address except Exception: raise gl.UserError("Cannot resolve sender address.") def _safe_transfer(recipient: Address, amount: bigint) -> None: """Safely disburse native GEN to an address using official GenLayer SDK pattern.""" if amount <= bigint(0): return gl.get_contract_at(recipient).emit_transfer(value=u256(int(amount))) DEFAULT_TRADEMARK_DOMAINS = ( "wipo.int", "www.wipo.int", "branddb.wipo.int", "uspto.gov", "www.uspto.gov", "tsdr.uspto.gov", "euipo.europa.eu", "www.euipo.europa.eu", "tmdn.org", "www.tmdn.org", "mock-trademark.genlayer.com", ) def _parse_url_host(raw_url: str) -> str: """Extract and validate normalized hostname from URL.""" clean = raw_url.strip() if "?" in clean: clean = clean.split("?")[0] if "#" in clean: clean = clean.split("#")[0] clean = clean.strip() if clean.startswith("https://"): rest = clean[8:] elif clean.startswith("http://"): rest = clean[7:] else: raise gl.UserError("URL must begin with http:// or https://") host = rest.split("/", 1)[0].strip().lower() if ":" in host: host = host.split(":")[0].strip() if not host: raise gl.UserError("Invalid URL: missing host") return host @allow_storage @dataclass class DomainDispute: dispute_id: str claimant: Address accused_holder: Address target_domain: str # e.g., "apple-pay.eth", "nike-web3.eth" trademark_name: str # Registered trademark text registration_number: str # Official trademark registration ID trademark_registry_url: str # Authoritative registry verification URL dispute_bond: bigint # Staked challenge bond status: str # "FILED", "ADJUDICATED", "CANCELLED" verdict: str # "PENDING", "BAD_FAITH_SQUATTING", "LEGITIMATE_FAIR_USE", "INSUFFICIENT_EVIDENCE" reason: str # Consensus rationale deadline: bigint # Resolution deadline created_at: bigint resolved_at: bigint class Contract(gl.Contract): """ TrademarkDomainDisputeX: Autonomous Web3 Domain Squatting & Trademark Defense Arbiter Track: Onchain Justice / Autonomous Protocols """ owner: Address dispute_count: bigint disputes: TreeMap[str, DomainDispute] revoked_domains: TreeMap[str, bool] custom_allowed_registries: TreeMap[str, bool] min_dispute_bond: bigint def __init__(self): # GenVM automatically initializes TreeMap storage fields to empty. self.owner = _get_sender() self.dispute_count = bigint(0) self.min_dispute_bond = bigint(1000) def _get_current_timestamp(self) -> bigint: """Derive trusted deterministic execution timestamp from GenLayer transaction context.""" try: from datetime import datetime dt_raw = getattr(gl.message, "datetime", None) if dt_raw is None and hasattr(gl, "message_raw") and isinstance(gl.message_raw, dict): dt_raw = gl.message_raw.get("datetime") if dt_raw: dt_str = str(dt_raw).strip().replace("Z", "+00:00") dt = datetime.fromisoformat(dt_str) ts = int(dt.timestamp()) if ts > 0: return bigint(ts) except Exception: pass try: if hasattr(gl, "block") and hasattr(gl.block, "timestamp"): ts = int(gl.block.timestamp) if ts > 0: return bigint(ts) except Exception: pass return bigint(0) def _parse_llm_json(self, text: str) -> dict: """Safely parse LLM responses, stripping markdown wrappers if present.""" 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": "INSUFFICIENT_EVIDENCE", "confidence": 0, "reason": f"Failed to parse LLM JSON: {str(e)[:100]}" } @gl.public.write def add_allowed_registry(self, domain: str) -> None: """Owner can add trusted intellectual property registries.""" if _addr_str(_get_sender()) != _addr_str(self.owner): raise gl.UserError("Only owner can add allowed registry domains.") clean = domain.strip().lower() if len(clean) < 3: raise gl.UserError("Invalid registry domain name.") self.custom_allowed_registries[clean] = True @gl.public.view def is_registry_allowed(self, domain: str) -> bool: """Check if registry host is authorized.""" clean = domain.strip().lower() if clean in DEFAULT_TRADEMARK_DOMAINS: return True try: return bool(self.custom_allowed_registries[clean]) except Exception: return False @gl.public.write.payable def file_dispute( self, accused_holder: Address, target_domain: str, trademark_name: str, registration_number: str, trademark_registry_url: str, challenge_duration_sec: int ) -> str: """ Trademark owner stakes a bond to file a cybersquatting dispute against a domain. """ bond = bigint(gl.message.value) if bond < self.min_dispute_bond: raise gl.UserError(f"Dispute bond must be at least {self.min_dispute_bond} wei.") clean_domain = target_domain.strip().lower() clean_tm = trademark_name.strip() clean_reg_num = registration_number.strip() clean_url = trademark_registry_url.strip() if len(clean_domain) < 3: raise gl.UserError("target_domain must be at least 3 characters.") if len(clean_tm) < 2: raise gl.UserError("trademark_name must be at least 2 characters.") if len(clean_reg_num) < 3: raise gl.UserError("registration_number is invalid.") if _addr_str(_get_sender()) == _addr_str(accused_holder): raise gl.UserError("Claimant and accused holder cannot be identical.") # Check if domain was already revoked is_revoked = False try: is_revoked = bool(self.revoked_domains[clean_domain]) except Exception: is_revoked = False if is_revoked: raise gl.UserError("This domain has already been formally revoked on-chain.") if not (clean_url.startswith("http://") or clean_url.startswith("https://")): raise gl.UserError("trademark_registry_url must start with http:// or https://") # Canonical Registry Validation host = _parse_url_host(clean_url) is_allowed = host in DEFAULT_TRADEMARK_DOMAINS if not is_allowed: try: is_allowed = bool(self.custom_allowed_registries[host]) except Exception: is_allowed = False if not is_allowed: raise gl.UserError(f"Registry host '{host}' is not in the allowed trademark whitelist.") # Canonical Registration ID Binding if clean_reg_num.lower() not in clean_url.lower(): raise gl.UserError(f"Registry URL must contain the registration number '{clean_reg_num}'.") now_ts = self._get_current_timestamp() duration = bigint(challenge_duration_sec if challenge_duration_sec >= 86400 else 86400) deadline = now_ts + duration if now_ts > bigint(0) else bigint(86400 * 7) self.dispute_count += bigint(1) did = str(self.dispute_count) self.disputes[did] = DomainDispute( dispute_id=did, claimant=_get_sender(), accused_holder=accused_holder, target_domain=clean_domain, trademark_name=clean_tm, registration_number=clean_reg_num, trademark_registry_url=clean_url, dispute_bond=bond, status="FILED", verdict="PENDING", reason="Dispute filed. Awaiting autonomous AI adjudication.", deadline=deadline, created_at=self.dispute_count, resolved_at=bigint(0) ) return did @gl.public.write def adjudicate_dispute(self, dispute_id: str) -> None: """ Validators render the trademark database page, cross-examine the target domain, and reach 100% discrete consensus on cybersquatting infringement. """ if dispute_id not in self.disputes: raise gl.UserError("Dispute not found.") dispute = self.disputes[dispute_id] if dispute.status != "FILED": raise gl.UserError("Dispute is already resolved or cancelled.") domain_local = str(dispute.target_domain) tm_local = str(dispute.trademark_name) reg_num_local = str(dispute.registration_number) url_local = str(dispute.trademark_registry_url) def leader_fn(): web_content = "" try: res = gl.nondet.web.render(url_local, mode="text") if hasattr(res, "content"): web_content = res.content elif isinstance(res, dict) and "body" in res: web_content = res["body"] else: web_content = str(res) except Exception: web_content = "" lower_web = web_content[:500].lower() if web_content else "" if len(web_content.strip()) < 15 or "404 not found" in lower_web or "record not found" in lower_web: return { "verdict": "INSUFFICIENT_EVIDENCE", "confidence": 100, "reason": "Trademark registry lookup returned 404, offline, or record not found." } snippet = web_content[:4000] prompt = f"""You are an Autonomous Trademark & Domain Dispute Arbiter (UDRP Specialist) on GenLayer. Evaluate whether the registration of target domain '{domain_local}' constitutes bad-faith cybersquatting against registered trademark '{tm_local}' (Reg No: {reg_num_local}). TRADEMARK REGISTRY SOURCE CONTENT: \"\"\" {snippet} \"\"\" TARGET DOMAIN UNDER SCRUTINY: {domain_local} DISCRETE CLASSIFICATION RULES: - "BAD_FAITH_SQUATTING": The official registry page confirms an active trademark for '{tm_local}', and the domain name is confusingly similar, capitalizing on brand confusion or held for extortion without legitimate rights. - "LEGITIMATE_FAIR_USE": The domain uses generic, descriptive, or unrelated words, or the accused holder demonstrates a bona fide offering of goods/services without deceptive intent. - "INSUFFICIENT_EVIDENCE": The registry entry is expired, cancelled, abandoned, or ambiguous. OUTPUT FORMAT: Respond ONLY with a VALID JSON object (no markdown, no backticks): {{ "verdict": "BAD_FAITH_SQUATTING" | "LEGITIMATE_FAIR_USE" | "INSUFFICIENT_EVIDENCE", "confidence": , "reason": "" }}""" try: raw_res = gl.nondet.exec_prompt(prompt, response_format="json") parsed = None if isinstance(raw_res, dict): parsed = raw_res elif hasattr(raw_res, "content") and isinstance(raw_res.content, dict): parsed = raw_res.content else: text = raw_res.content if hasattr(raw_res, "content") else str(raw_res) cleaned = str(text).strip() if cleaned.startswith("```json"): cleaned = cleaned[7:] elif cleaned.startswith("```"): cleaned = cleaned[3:] if cleaned.endswith("```"): cleaned = cleaned[:-3] parsed = json.loads(cleaned.strip()) verdict_candidate = str(parsed.get("verdict", "INSUFFICIENT_EVIDENCE")).strip().upper() valid_verdicts = ("BAD_FAITH_SQUATTING", "LEGITIMATE_FAIR_USE", "INSUFFICIENT_EVIDENCE") if verdict_candidate not in valid_verdicts: verdict_candidate = "INSUFFICIENT_EVIDENCE" try: conf = int(parsed.get("confidence", 0)) conf = max(0, min(100, conf)) except Exception: conf = 50 reason_str = str(parsed.get("reason", "Evaluated by autonomous trademark jury."))[:220] return { "verdict": verdict_candidate, "confidence": conf, "reason": reason_str } except Exception as e: return { "verdict": "INSUFFICIENT_EVIDENCE", "confidence": 0, "reason": f"Evaluation error: {str(e)[:100]}" } def validator_fn(leader_res) -> bool: if not isinstance(leader_res, gl.vm.Return): return False leader = leader_res.calldata if not isinstance(leader, dict) or "verdict" not in leader: return False valid_verdicts = ("BAD_FAITH_SQUATTING", "LEGITIMATE_FAIR_USE", "INSUFFICIENT_EVIDENCE") l_verdict = str(leader.get("verdict", "")).strip().upper() if l_verdict not in valid_verdicts: return False mine = leader_fn() m_verdict = str(mine.get("verdict", "")).strip().upper() # DISCRETE EQUIVALENCE: 100% agreement on exact discrete legal verdict return l_verdict == m_verdict adjudication_res = gl.vm.run_nondet(leader_fn, validator_fn) if isinstance(adjudication_res, dict): final_res = adjudication_res else: final_res = self._parse_llm_json(str(adjudication_res)) verdict = str(final_res.get("verdict", "INSUFFICIENT_EVIDENCE")).strip().upper() valid_verdicts = ("BAD_FAITH_SQUATTING", "LEGITIMATE_FAIR_USE", "INSUFFICIENT_EVIDENCE") if verdict not in valid_verdicts: verdict = "INSUFFICIENT_EVIDENCE" reason = str(final_res.get("reason", "Consensus concluded.")) dispute.status = "ADJUDICATED" dispute.verdict = verdict dispute.reason = reason dispute.resolved_at = self.dispute_count bond_val = dispute.dispute_bond if verdict == "BAD_FAITH_SQUATTING": # Blacklist/Revoke domain on-chain self.revoked_domains[dispute.target_domain.lower()] = True self.disputes[dispute_id] = dispute # Refund bond back to claimant if bond_val > bigint(0): _safe_transfer(dispute.claimant, bond_val) elif verdict == "LEGITIMATE_FAIR_USE": # Domain holder was legitimate; claimant bond is forfeited self.disputes[dispute_id] = dispute else: # INSUFFICIENT_EVIDENCE: refund bond to claimant self.disputes[dispute_id] = dispute if bond_val > bigint(0): _safe_transfer(dispute.claimant, bond_val) @gl.public.write def cancel_expired_dispute(self, dispute_id: str) -> None: """ Safe Recovery Path: If dispute passes deadline without adjudication, claimant can recover their bond. """ if dispute_id not in self.disputes: raise gl.UserError("Dispute not found.") dispute = self.disputes[dispute_id] if _addr_str(_get_sender()) != _addr_str(dispute.claimant): raise gl.UserError("Only claimant can cancel expired dispute.") if dispute.status != "FILED": raise gl.UserError("Dispute is already resolved or cancelled.") now_ts = self._get_current_timestamp() if now_ts > bigint(0) and now_ts <= dispute.deadline: raise gl.UserError("Dispute deadline has not passed yet.") bond_val = dispute.dispute_bond dispute.status = "CANCELLED" dispute.verdict = "INSUFFICIENT_EVIDENCE" dispute.reason = "Dispute lapsed past deadline and was cancelled by claimant." dispute.resolved_at = self.dispute_count self.disputes[dispute_id] = dispute if bond_val > bigint(0): _safe_transfer(dispute.claimant, bond_val) @gl.public.view def is_domain_revoked(self, domain: str) -> bool: """Check if a domain has been officially marked as bad-faith cybersquatted on-chain.""" clean = domain.strip().lower() try: return bool(self.revoked_domains[clean]) except Exception: return False @gl.public.view def get_dispute(self, dispute_id: str) -> str: """Retrieve details of a domain dispute as a JSON string.""" if dispute_id not in self.disputes: raise gl.UserError("Dispute not found.") d = self.disputes[dispute_id] return json.dumps({ "dispute_id": d.dispute_id, "claimant": _addr_str(d.claimant), "accused_holder": _addr_str(d.accused_holder), "target_domain": d.target_domain, "trademark_name": d.trademark_name, "registration_number": d.registration_number, "trademark_registry_url": d.trademark_registry_url, "dispute_bond": str(d.dispute_bond), "status": d.status, "verdict": d.verdict, "reason": d.reason, "deadline": str(d.deadline), "created_at": str(d.created_at), "resolved_at": str(d.resolved_at) }) @gl.public.view def get_current_time(self) -> int: """Retrieve current contract execution timestamp.""" return int(str(self._get_current_timestamp())) @gl.public.view def get_dispute_count(self) -> int: return int(self.dispute_count) @gl.public.view def get_owner(self) -> str: return _addr_str(self.owner)