# v0.2.16 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * from dataclasses import dataclass import json from urllib.parse import urlparse UserError = gl.vm.UserError 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 _normalize_repo_url(url: str, is_patch: bool = False) -> tuple: scheme, hostname, port = _extract_origin(url) parsed = urlparse(url.strip()) raw_path = parsed.path.strip() segments = [s for s in raw_path.split("/") if s] if not is_patch: if len(segments) < 2: raise UserError("Official repository base must include owner and repository name") if segments[-1].lower().endswith(".git"): segments[-1] = segments[-1][:-4] if not segments[-1]: raise UserError("Invalid repository name in URL") norm_path = "/" + "/".join(segments).lower() return scheme, hostname, port, norm_path else: norm_path = "/" + "/".join(segments).lower() return scheme, hostname, port, norm_path def _validate_patch_commit_url(patch_url: str, base_repo_url: str) -> None: t_scheme, t_host, t_port, t_path = _normalize_repo_url(patch_url, is_patch=True) b_scheme, b_host, b_port, b_path = _normalize_repo_url(base_repo_url, is_patch=False) if t_scheme != b_scheme or t_port != b_port: raise UserError("Patch commit URL protocol or port does not match registered repository origin") if t_host != b_host and not t_host.endswith("." + b_host): raise UserError("Patch commit URL host does not match registered repository origin") # Enforce path binding: patch path must strictly start with the registered repository path if not t_path.startswith(b_path + "/"): raise UserError("Patch commit URL must belong to the registered repository path") # Enforce commit endpoint requirement (ideally commit/ or commits/ or -/commit/) rel_path = t_path[len(b_path) + 1:].strip("/") segs = [s for s in rel_path.split("/") if s] commit_hash = "" if len(segs) == 2 and segs[0] in ("commit", "commits"): commit_hash = segs[1] elif len(segs) == 3 and segs[0] == "-" and segs[1] == "commit": commit_hash = segs[2] else: raise UserError("Patch URL must point to a valid commit endpoint in the registered repository") if len(commit_hash) < 4 or not commit_hash.isalnum(): raise UserError("Patch commit hash is invalid") 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", "payout_pct": 0, "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 ("CRITICAL_VALID", "MEDIUM_VALID", "SPAM_INVALID", "ABORT"): return None pct = data.get("payout_pct", 0) if isinstance(pct, float): pct = int(pct) if not isinstance(pct, int) or not (0 <= pct <= 100): return None if verdict == "CRITICAL_VALID": pct = 100 elif verdict == "MEDIUM_VALID" and not (1 <= pct <= 99): pct = 50 elif verdict in ("SPAM_INVALID", "ABORT"): pct = 0 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" pct = 0 reason = f"[low_confidence: {conf}%] " + reason return { "verdict": verdict, "payout_pct": pct, "confidence": conf, "reason": reason[:300], } @allow_storage @dataclass class BugBountyProgram: id: str sponsor: str project_name: str official_repo_base: str total_budget: bigint remaining_balance: bigint status: str # ACTIVE | CLOSED reports_count: bigint @allow_storage @dataclass class VulnerabilityReport: id: str program_id: str whitehat: str poc_report_url: str patch_commit_url: str requested_bounty: bigint status: str # PENDING | AUDITING | SETTLED | REJECTED | ESCALATED verdict: str # CRITICAL_VALID | MEDIUM_VALID | SPAM_INVALID | ABORT payout_pct: bigint confidence: bigint reason: str class Contract(gl.Contract): programs: TreeMap[str, BugBountyProgram] reports: TreeMap[str, VulnerabilityReport] program_counter: bigint treasury_address: str total_locked_funds: bigint platform_arbiter: str def __init__(self, treasury_addr: str): self.program_counter = bigint(0) treasury_addr = treasury_addr.strip() if treasury_addr else "" if not treasury_addr: raise UserError("Treasury address is required") try: Address(treasury_addr) except Exception: raise UserError("Invalid treasury address format") self.treasury_address = treasury_addr.lower() self.total_locked_funds = bigint(0) self.platform_arbiter = _addr_str(gl.message.sender_address) def _treasury(self) -> Address: if not self.treasury_address: raise UserError("Treasury address is not configured") return Address(self.treasury_address) @gl.public.write.payable def create_program(self, project_name: str, official_repo_base: str) -> str: """DeFi project creates bug bounty escrow with locked funds and registered repository.""" budget = gl.message.value if budget <= bigint(0): raise UserError("Bounty pool budget must be greater than 0") project_name = project_name.strip() official_repo_base = official_repo_base.strip() if len(project_name) < 3: raise UserError("Project name too short") # Enforce strict origin extraction and repository path validation on official repository URL _normalize_repo_url(official_repo_base, is_patch=False) self.program_counter += bigint(1) pid = str(self.program_counter) self.programs[pid] = BugBountyProgram( id=pid, sponsor=_addr_str(gl.message.sender_address), project_name=project_name, official_repo_base=official_repo_base, total_budget=budget, remaining_balance=budget, status="ACTIVE", reports_count=bigint(0), ) self.total_locked_funds += budget return pid @gl.public.write def submit_and_audit_vulnerability( self, program_id: str, poc_report_url: str, patch_commit_url: str, requested_bounty: bigint, ) -> str: """Whitehat hacker submits vulnerability PoC and patch commit for AI consensus audit.""" if program_id not in self.programs: raise UserError("Bounty program not found") prog = self.programs[program_id] if prog.status != "ACTIVE": raise UserError("Bounty program is not active") if requested_bounty <= bigint(0): raise UserError("Requested bounty must be greater than 0") if requested_bounty > prog.remaining_balance: raise UserError("Requested bounty exceeds remaining pool balance") poc_report_url = poc_report_url.strip() patch_commit_url = patch_commit_url.strip() # Validate origin & repository path: patch commit must strictly belong to official repo base _validate_patch_commit_url(patch_commit_url, prog.official_repo_base) _extract_origin(poc_report_url) prog.reports_count += bigint(1) rid = str(prog.reports_count) r_key = program_id + "_" + rid whitehat_addr = _addr_str(gl.message.sender_address) self.reports[r_key] = VulnerabilityReport( id=rid, program_id=program_id, whitehat=whitehat_addr, poc_report_url=poc_report_url, patch_commit_url=patch_commit_url, requested_bounty=requested_bounty, status="AUDITING", verdict="", payout_pct=bigint(0), confidence=bigint(0), reason="", ) self.programs[program_id] = prog p_name = str(prog.project_name) u_poc = str(poc_report_url) u_patch = str(patch_commit_url) # Inline nondet block: Pass GenVM static AST linter def leader_fn(): # 1. Scrape PoC vulnerability description try: res_poc = gl.nondet.web.render(u_poc, mode="text") poc_text = res_poc.content if hasattr(res_poc, "content") else str(res_poc) if not poc_text or len(poc_text.strip()) < 30: return {"verdict": "ABORT", "payout_pct": 0, "confidence": 0, "reason": "PoC URL returned empty"} except Exception as e: return {"verdict": "ABORT", "payout_pct": 0, "confidence": 0, "reason": f"PoC fetch failed: {str(e)}"} # 2. Scrape project official patch commit diff try: res_patch = gl.nondet.web.render(u_patch, mode="text") patch_text = res_patch.content if hasattr(res_patch, "content") else str(res_patch) if not patch_text or len(patch_text.strip()) < 30: return {"verdict": "ABORT", "payout_pct": 0, "confidence": 0, "reason": "Patch commit URL returned empty"} except Exception as e: return {"verdict": "ABORT", "payout_pct": 0, "confidence": 0, "reason": f"Patch fetch failed: {str(e)}"} prompt = f""" SYSTEM: You are an elite Smart Contract Security & Bug Bounty Triager for {p_name}. Audit whether the submitted Vulnerability PoC report was valid and addressed by the official patch commit diff. SUBMITTED VULNERABILITY POC: {poc_text[:3500]} OFFICIAL PATCH COMMIT DIFF: {patch_text[:3500]} Rules: - CRITICAL_VALID (payout_pct=100, conf >= 75): Report accurately details a critical exploit (fund drain, reentrancy, access control) solved by the patch commit. - MEDIUM_VALID (payout_pct=50, conf >= 75): Report identifies an authentic medium/low issue or edge-case bug partially mitigated. - SPAM_INVALID (payout_pct=0, conf >= 75): Report is irrelevant AI spam, false positive, or unrelated to the patch commit. - ABORT: Web links require login/captcha, rate-limited, 404, or unreadable. OUTPUT ONLY STRICT JSON: {{ "verdict": "CRITICAL_VALID" | "MEDIUM_VALID" | "SPAM_INVALID" | "ABORT", "payout_pct": 0-100, "confidence": 0-100, "reason": "max 300 chars technical explanation" }} """ 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", "payout_pct": 0, "confidence": 0, "reason": "parse_failed"} if (p1["verdict"] != p2["verdict"]) or (p1["payout_pct"] != p2["payout_pct"]): return {"verdict": "ABORT", "payout_pct": 0, "confidence": 0, "reason": "multi_sample_divergence"} p1["confidence"] = (p1["confidence"] + p2["confidence"]) // 2 return p1 except Exception as e: return {"verdict": "ABORT", "payout_pct": 0, "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 # Semantic agreement on verdict, payout percentage, and unified 75% confidence bucket return ( mine["verdict"] == leader["verdict"] and mine["payout_pct"] == leader["payout_pct"] 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", "payout_pct": 0, "confidence": 0, "reason": "adjudication_failed"} verdict = result["verdict"] payout_pct = result["payout_pct"] confidence = result["confidence"] reason = result["reason"] # Deterministic post-consensus normalization if confidence < 75 and verdict != "ABORT": verdict = "ABORT" payout_pct = 0 rep = self.reports[r_key] rep.verdict = verdict rep.payout_pct = bigint(payout_pct) rep.confidence = bigint(confidence) rep.reason = reason req_bounty = rep.requested_bounty whitehat_wallet = Address(rep.whitehat) if verdict in ("CRITICAL_VALID", "MEDIUM_VALID"): payout = (req_bounty * bigint(payout_pct)) // bigint(100) fee = (payout * bigint(2)) // bigint(100) whitehat_net = payout - fee if fee > bigint(0): gl.get_contract_at(self._treasury()).emit_transfer(value=fee) if whitehat_net > bigint(0): gl.get_contract_at(whitehat_wallet).emit_transfer(value=whitehat_net) rep.status = "SETTLED" # Deduct disbursed funds from program balance and global locked reserves if prog.remaining_balance >= payout: prog.remaining_balance -= payout else: prog.remaining_balance = bigint(0) if self.total_locked_funds >= payout: self.total_locked_funds -= payout else: self.total_locked_funds = bigint(0) self.programs[program_id] = prog elif verdict == "SPAM_INVALID": rep.status = "REJECTED" else: rep.status = "ESCALATED" self.reports[r_key] = rep return r_key @gl.public.write def resolve_escalated_report( self, program_id: str, report_id: str, payout_percentage: int, ) -> None: """Sponsor or platform arbiter manually settles an ESCALATED report.""" r_key = program_id + "_" + report_id if r_key not in self.reports: raise UserError("Report not found") rep = self.reports[r_key] prog = self.programs[program_id] if rep.status != "ESCALATED": raise UserError("Report is not in ESCALATED status") sender = _addr_str(gl.message.sender_address) if sender != prog.sponsor and sender != self.platform_arbiter: raise UserError("Only sponsor or platform arbiter can resolve escalated report") if not (0 <= payout_percentage <= 100): raise UserError("payout_percentage must be between 0 and 100") req_bounty = rep.requested_bounty payout = (req_bounty * bigint(payout_percentage)) // bigint(100) if payout > prog.remaining_balance: payout = prog.remaining_balance fee = (payout * bigint(2)) // bigint(100) whitehat_net = payout - fee if fee > bigint(0): gl.get_contract_at(self._treasury()).emit_transfer(value=fee) if whitehat_net > bigint(0): gl.get_contract_at(Address(rep.whitehat)).emit_transfer(value=whitehat_net) rep.status = "SETTLED" rep.verdict = f"RESOLVED_MANUALLY_{payout_percentage}_PERCENT" rep.reason = f"Manually resolved by {sender}" self.reports[r_key] = rep if prog.remaining_balance >= payout: prog.remaining_balance -= payout else: prog.remaining_balance = bigint(0) if self.total_locked_funds >= payout: self.total_locked_funds -= payout else: self.total_locked_funds = bigint(0) self.programs[program_id] = prog @gl.public.write def close_program_and_refund_remainder(self, program_id: str) -> None: """Sponsor can close program and reclaim unspent bounty funds once all reports are settled.""" if program_id not in self.programs: raise UserError("Program not found") prog = self.programs[program_id] if _addr_str(gl.message.sender_address) != prog.sponsor: raise UserError("Only sponsor can close program") if prog.status != "ACTIVE": raise UserError("Program is not active") # Verify no pending reports count = int(prog.reports_count) for i in range(1, count + 1): r_key = program_id + "_" + str(i) if r_key in self.reports: r = self.reports[r_key] if r.status in ("PENDING", "AUDITING", "ESCALATED"): raise UserError("Cannot close program while reports are pending or escalated") remainder = prog.remaining_balance prog.status = "CLOSED" prog.remaining_balance = bigint(0) self.programs[program_id] = prog if self.total_locked_funds >= remainder: self.total_locked_funds -= remainder else: self.total_locked_funds = bigint(0) if remainder > bigint(0): gl.get_contract_at(Address(prog.sponsor)).emit_transfer(value=remainder) @gl.public.view def get_program(self, program_id: str) -> str: if program_id not in self.programs: raise UserError("Program not found") p = self.programs[program_id] return json.dumps({ "id": p.id, "sponsor": p.sponsor, "project_name": p.project_name, "official_repo_base": p.official_repo_base, "total_budget": str(p.total_budget), "remaining_balance": str(p.remaining_balance), "status": p.status, "reports_count": str(p.reports_count), }) @gl.public.view def get_report(self, program_id: str, report_id: str) -> str: r_key = program_id + "_" + report_id if r_key not in self.reports: raise UserError("Report not found") r = self.reports[r_key] return json.dumps({ "id": r.id, "program_id": r.program_id, "whitehat": r.whitehat, "poc_report_url": r.poc_report_url, "patch_commit_url": r.patch_commit_url, "requested_bounty": str(r.requested_bounty), "status": r.status, "verdict": r.verdict, "payout_pct": str(r.payout_pct), "confidence": str(r.confidence), "reason": r.reason, }) @gl.public.view def get_program_counter(self) -> int: return int(self.program_counter) @gl.public.view def get_treasury(self) -> str: return self.treasury_address