# { "Depends": "py-genlayer:test" } import json from dataclasses import dataclass from genlayer import * ERROR_EXPECTED = "[EXPECTED]" ERROR_LLM = "[LLM_ERROR]" @allow_storage @dataclass class ForensicsReport: report_id: u256 submitter: Address wallet_address: str chain: str risk_score: u256 risk_level: str findings_json: str summary: str patterns_json: str def _risk_level(score: int) -> str: if score >= 80: return "CRITICAL" if score >= 60: return "HIGH" if score >= 40: return "MEDIUM" if score >= 20: return "LOW" return "CLEAN" def _clamp_score(raw) -> int: try: value = int(round(float(str(raw).strip()))) except (ValueError, TypeError): return 0 if value < 0: return 0 if value > 100: return 100 return value def _investigate_prompt(address: str, chain: str) -> str: return f"""You are WalletDetective, an AI on-chain forensics investigator. Analyze this wallet address for suspicious activity. Wallet Address: {address} Chain: {chain} Perform a comprehensive forensics analysis: 1. Identify wallet type (EOA, multisig, contract, exchange) 2. Analyze transaction patterns (frequency, amounts, counterparties) 3. Check for known associations (mixers, sanctioned addresses, exploits) 4. Detect unusual patterns (wash trading, layering, structuring) 5. Assess fund sources and destinations 6. Check for smart contract interactions (DeFi, bridges, DEXs) Return ONLY valid JSON: {{ "risk_score": <0-100>, "wallet_type": "EOA|MULTISIG|CONTRACT|EXCHANGE", "findings": [ {{ "category": "MIXER|SANCTION|EXPLOIT|WASH_TRADE|LAYERING|SUSPICIOUS|NORMAL", "severity": "CRITICAL|HIGH|MEDIUM|LOW|INFO", "description": "", "confidence": <0-100> }} ], "patterns": [ {{"pattern": "", "description": "", "frequency": ""}} ], "summary": "<3-4 sentence forensics summary>", "associated_protocols": ["", ""], "estimated_volume": "", "first_seen": "" }} """ class WalletDetective(gl.Contract): owner: Address report_count: u256 reports: TreeMap[u256, ForensicsReport] report_ids: DynArray[u256] def __init__(self): self.owner = gl.message.sender_address self.report_count = u256(0) def _investigate(self, address: str, chain: str) -> dict: prompt = _investigate_prompt(address, chain) def leader_fn(): result = gl.nondet.exec_prompt(prompt, response_format="json") if not isinstance(result, dict): raise gl.vm.UserError(f"{ERROR_LLM} Invalid response") result["risk_score"] = _clamp_score(result.get("risk_score", 0)) result["risk_level"] = _risk_level(result["risk_score"]) return result def validator_fn(leaders_res): if not isinstance(leaders_res, gl.vm.Return): return False v = leader_fn() leader_score = _clamp_score(leaders_res.calldata.get("risk_score", 0)) validator_score = _clamp_score(v.get("risk_score", 0)) if abs(leader_score - validator_score) > 20: return False return _risk_level(leader_score) == _risk_level(validator_score) return gl.vm.run_nondet_unsafe(leader_fn, validator_fn) @gl.public.write def investigate_wallet(self, address: str, chain: str) -> u256: cleaned = address.strip() if len(cleaned) < 10: raise gl.vm.UserError(f"{ERROR_EXPECTED} Invalid wallet address") analysis = self._investigate(cleaned, chain.strip()) report_id = self.report_count findings_json = json.dumps(analysis.get("findings", []), sort_keys=True) patterns_json = json.dumps(analysis.get("patterns", []), sort_keys=True) self.reports[report_id] = ForensicsReport( report_id=report_id, submitter=gl.message.sender_address, wallet_address=cleaned, chain=chain.strip(), risk_score=int(analysis.get("risk_score", 0)), risk_level=str(analysis.get("risk_level", "UNKNOWN")), findings_json=findings_json, summary=str(analysis.get("summary", ""))[:1000], patterns_json=patterns_json, ) self.report_ids.append(report_id) self.report_count = report_id + 1 return report_id @gl.public.view def get_report(self, id: u256) -> dict: if id not in self.reports: raise gl.vm.UserError(f"{ERROR_EXPECTED} Report not found") r = self.reports[id] return { "report_id": r.report_id, "submitter": r.submitter.as_hex, "wallet_address": r.wallet_address, "chain": r.chain, "risk_score": r.risk_score, "risk_level": r.risk_level, "findings": json.loads(r.findings_json), "summary": r.summary, "patterns": json.loads(r.patterns_json), } @gl.public.view def get_all_reports(self) -> list: results = [] count = int(self.report_count) i = count - 1 while i >= 0 and len(results) < 50: r = self.reports[i] results.append({ "report_id": r.report_id, "submitter": r.submitter.as_hex, "wallet_address": r.wallet_address, "chain": r.chain, "risk_score": r.risk_score, "risk_level": r.risk_level, }) i -= 1 return results