# v0.2.19 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * from dataclasses import dataclass import json import datetime class UserError(Exception): pass @allow_storage @dataclass class Bounty: id: str creator: str title: str target_repo_url: str vulnerability_description: str expected_fix_criteria: str reward_amount: bigint status: str # "OPEN", "RESOLVED", "CANCELLED" winner: str ai_verdict_reason: str patch_pr_url: str created_at: bigint submission_count: bigint commit_hash: str # Bound immutable git commit SHA last_submitter: str # Bound address of the hunter who submitted the patch payout_status: str # "UNPAID", "PAID", "CLAIMABLE", "REFUNDED" class Contract(gl.Contract): bounties: TreeMap[str, Bounty] bounty_ids: DynArray[str] owner: str def __init__(self): # GenVM automatically allocates memory for TreeMap & DynArray self.owner = str(gl.message.sender_address).lower() def _now(self) -> bigint: if not hasattr(gl, "message_raw") or not isinstance(gl.message_raw, dict): raise UserError("Trusted runtime execution timestamp context missing") dt_raw = gl.message_raw.get("datetime", None) if not dt_raw: raise UserError("Trusted timestamp 'datetime' missing from transaction context") try: s = str(dt_raw) if s.endswith("Z"): s = s[:-1] + "+00:00" ts = int(datetime.datetime.fromisoformat(s).timestamp()) if ts <= 0: raise UserError("Invalid non-positive timestamp resolved") return bigint(ts) except Exception as e: raise UserError(f"Failed to parse runtime timestamp: {str(e)}") def _parse_llm_json(self, response) -> dict: """ STRICT OUTPUT PARSER (Steward Gen. Dave & Pavel Kolosov Enforced): - Accepts ONLY valid JSON object with explicit boolean is_valid and non-empty string reason. - Strictly rejects prose, markdown text, pseudo-JSON, truncated JSON, missing fields, nulls, and wrong-type values. - Absolutely NO fallback that approves text lacking 'false'. - Fail-closed on any malformed or unexpected structure. """ if isinstance(response, dict): raw_dict = response else: try: text = str(response).strip() if text.startswith("```json"): text = text[7:] elif text.startswith("```"): text = text[3:] if text.endswith("```"): text = text[:-3] text = text.strip() try: raw_dict = json.loads(text) except Exception: norm = text.replace("'", '"').replace("True", "true").replace("False", "false") raw_dict = json.loads(norm) except Exception as e: return { "is_valid": False, "reason": f"FAIL-CLOSED: Response is not valid JSON ({str(e)})." } if not isinstance(raw_dict, dict): return { "is_valid": False, "reason": "FAIL-CLOSED: Output root must be a JSON object." } if "is_valid" not in raw_dict or "reason" not in raw_dict: return { "is_valid": False, "reason": "FAIL-CLOSED: Missing required fields 'is_valid' or 'reason'." } val = raw_dict["is_valid"] reason = raw_dict["reason"] if type(val) is not bool: return { "is_valid": False, "reason": "FAIL-CLOSED: Field 'is_valid' must be an explicit boolean (true or false), not string/number/null." } if not isinstance(reason, str) or len(reason.strip()) == 0: return { "is_valid": False, "reason": "FAIL-CLOSED: Field 'reason' must be a non-empty string." } return { "is_valid": val, "reason": reason.strip() } @gl.public.write.payable def create_bounty( self, bounty_id: str, title: str, target_repo_url: str, vulnerability_description: str, expected_fix_criteria: str, ) -> None: """ CREATOR PROTECTION & ESCROW LOCK: - Requires positive native GEN escrow lock. - Records immutable block timestamp for time-lock protection. - Supports pull-claim escrow policy if criteria specifies [CLAIMABLE] or [PULL]. """ amount = gl.message.value if amount <= bigint(0): raise UserError("Escrow reward amount must be greater than 0") if bounty_id in self.bounties: raise UserError("Bounty ID already exists") current_time = self._now() self.bounty_ids.append(bounty_id) self.bounties[bounty_id] = Bounty( id=bounty_id, creator=str(gl.message.sender_address).lower(), title=title, target_repo_url=target_repo_url, vulnerability_description=vulnerability_description, expected_fix_criteria=expected_fix_criteria, reward_amount=amount, status="OPEN", winner="", ai_verdict_reason="Awaiting Submissions", patch_pr_url="", created_at=current_time, submission_count=bigint(0), commit_hash="", last_submitter="", payout_status="UNPAID", ) @gl.public.write.payable def top_up_bounty(self, bounty_id: str) -> None: """ CREATOR ENHANCEMENT: Allows creator or community sponsors to increase bounty reward to attract top researchers. """ if bounty_id not in self.bounties: raise UserError("Bounty not found") bounty = self.bounties[bounty_id] if bounty.status != "OPEN": raise UserError("Cannot add reward to non-OPEN bounty") added_amount = gl.message.value if added_amount <= bigint(0): raise UserError("Additional reward must be greater than 0") bounty.reward_amount += added_amount self.bounties[bounty_id] = bounty @gl.public.write def submit_and_evaluate_patch( self, bounty_id: str, commit_hash: str, pr_url: str, ) -> None: """ GROUNDED VALIDATOR CONSENSUS WITH SAFE RECOVERABLE SETTLEMENT: - Fetches authentic git commit diff directly from GitHub via gl.nondet.web.get. - Binds an immutable git commit hash to the on-chain bounty state. - Evaluates fix criteria AND verifies patch introduces no regressions. - Fails closed if diff cannot be fetched or output is malformed. - Verifies pre-flight liquid contract balance before attempting outgoing transfer. - If liquid balance is insufficient or pull-policy is set, safely marks CLAIMABLE for pull claim. """ if bounty_id not in self.bounties: raise UserError("Bounty not found") bounty = self.bounties[bounty_id] if bounty.status != "OPEN": raise UserError("Bounty is not OPEN for submissions") clean_commit = commit_hash.strip().lower() if len(clean_commit) < 7: raise UserError("Invalid commit_hash: must be a valid git commit SHA (minimum 7 characters).") clean_pr = pr_url.strip() if not clean_pr.startswith("http://") and not clean_pr.startswith("https://"): raise UserError("Invalid pr_url: must be a valid HTTP/HTTPS URL.") hunter = str(gl.message.sender_address).lower() title_str = str(bounty.title) repo_url = str(bounty.target_repo_url).strip().rstrip("/") vuln_desc = str(bounty.vulnerability_description) criteria = str(bounty.expected_fix_criteria) is_self_submission = (hunter == bounty.creator.lower()) commit_diff_url = f"{repo_url}/commit/{clean_commit}.diff" pr_diff_url = f"{clean_pr.rstrip('/')}.diff" if "/pull/" in clean_pr else commit_diff_url def leader_fn(): diff_text = "" fetch_error = "" for target_url in [commit_diff_url, pr_diff_url]: try: res = gl.nondet.web.get(target_url) body_bytes = res.body if hasattr(res, "body") else (res if isinstance(res, (bytes, bytearray)) else str(res).encode("utf-8")) text = body_bytes.decode("utf-8", errors="replace") if text and " 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_data = self._parse_llm_json(leader_data) if not isinstance(leader_data, dict) or type(leader_data.get("is_valid")) is not bool: return False mine_data = leader_fn() if not isinstance(mine_data, dict) or type(mine_data.get("is_valid")) is not bool: return False return leader_data["is_valid"] == mine_data["is_valid"] result = gl.vm.run_nondet(leader_fn, validator_fn) parsed_result = self._parse_llm_json(result) bounty.submission_count += bigint(1) sub_count = int(bounty.submission_count) bounty.last_submitter = hunter bounty.commit_hash = clean_commit bounty.patch_pr_url = clean_pr is_valid = parsed_result.get("is_valid") reason = str(parsed_result.get("reason", "")).strip() if is_valid is not True or len(reason) < 5: # Rejection path: Fail-closed, keep escrow safe in contract bounty.status = "OPEN" bounty.payout_status = "UNPAID" bounty.ai_verdict_reason = f"[Submission #{sub_count}] REJECTED (Commit: {clean_commit[:7]}): {reason}" self.bounties[bounty_id] = bounty return # APPROVAL & SAFE RECOVERABLE SETTLEMENT PATH: bounty.status = "RESOLVED" bounty.winner = hunter # 1. Pre-flight Liquid Balance Check has_sufficient_balance = False try: has_sufficient_balance = (self.balance >= bounty.reward_amount) except Exception: has_sufficient_balance = False # 2. Configurable Pull-Escrow Policy Check is_pull_policy = ( "[CLAIMABLE]" in criteria.upper() or "[PULL]" in criteria.upper() or "[RECOVERABLE]" in criteria.upper() or "[CLAIMABLE]" in vuln_desc.upper() ) if has_sufficient_balance and not is_pull_policy: # Verified push transfer: execute emit_transfer to winner gl.get_contract_at(Address(hunter)).emit_transfer(value=u256(bounty.reward_amount)) bounty.payout_status = "PAID" bounty.ai_verdict_reason = f"[Submission #{sub_count}] PASSED & PAID (Commit: {clean_commit[:7]}): {reason}" else: # Automatic transfer withheld or pull-policy enforced: retain escrow as CLAIMABLE bounty.payout_status = "CLAIMABLE" hold_cause = "escrow pull-policy enforced" if is_pull_policy else "insufficient contract liquid balance" bounty.ai_verdict_reason = f"[Submission #{sub_count}] PASSED (Commit: {clean_commit[:7]}): {reason}. Automatic transfer withheld ({hold_cause}). Escrow held as CLAIMABLE for claim_bounty_payout." self.bounties[bounty_id] = bounty @gl.public.write def appeal_rejection( self, bounty_id: str, hunter_justification: str, ) -> None: """ HUNTER PROTECTION - APPEALS TRIBUNAL: Allows a Hunter whose patch was rejected to request re-adjudication with technical justification. Re-runs consensus validator audit with Hunter's defense considered against the authentic git diff. """ if bounty_id not in self.bounties: raise UserError("Bounty not found") bounty = self.bounties[bounty_id] if bounty.status != "OPEN": raise UserError("Only OPEN bounties with prior rejections can be appealed.") clean_commit = getattr(bounty, "commit_hash", "") if not clean_commit or bounty.submission_count == bigint(0): raise UserError("No prior submission found on-chain to appeal.") caller = str(gl.message.sender_address).lower() if bounty.last_submitter and caller != bounty.last_submitter.lower(): raise UserError("Only the security hunter who submitted this patch can file an appeal.") hunter = bounty.last_submitter or caller title_str = str(bounty.title) repo_url = str(bounty.target_repo_url).strip().rstrip("/") vuln_desc = str(bounty.vulnerability_description) criteria = str(bounty.expected_fix_criteria) clean_pr = bounty.patch_pr_url justification_str = hunter_justification.strip()[:1500] commit_diff_url = f"{repo_url}/commit/{clean_commit}.diff" pr_diff_url = f"{clean_pr.rstrip('/')}.diff" if "/pull/" in clean_pr else commit_diff_url def leader_appeal_fn(): diff_text = "" for target_url in [commit_diff_url, pr_diff_url]: try: res = gl.nondet.web.get(target_url) body_bytes = res.body if hasattr(res, "body") else (res if isinstance(res, (bytes, bytearray)) else str(res).encode("utf-8")) text = body_bytes.decode("utf-8", errors="replace") if text and " 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_data = self._parse_llm_json(leader_data) if not isinstance(leader_data, dict) or type(leader_data.get("is_valid")) is not bool: return False mine_data = leader_appeal_fn() if not isinstance(mine_data, dict) or type(mine_data.get("is_valid")) is not bool: return False return leader_data["is_valid"] == mine_data["is_valid"] result = gl.vm.run_nondet(leader_appeal_fn, validator_appeal_fn) parsed_result = self._parse_llm_json(result) is_valid = parsed_result.get("is_valid") reason = str(parsed_result.get("reason", "")).strip() if is_valid is True and len(reason) > 5: bounty.status = "RESOLVED" bounty.winner = hunter has_sufficient_balance = False try: has_sufficient_balance = (self.balance >= bounty.reward_amount) except Exception: has_sufficient_balance = False is_pull_policy = ( "[CLAIMABLE]" in criteria.upper() or "[PULL]" in criteria.upper() or "[RECOVERABLE]" in criteria.upper() ) if has_sufficient_balance and not is_pull_policy: gl.get_contract_at(Address(hunter)).emit_transfer(value=u256(bounty.reward_amount)) bounty.payout_status = "PAID" bounty.ai_verdict_reason = f"[APPEAL UPHELD & PAID]: {reason}" else: bounty.payout_status = "CLAIMABLE" bounty.ai_verdict_reason = f"[APPEAL UPHELD]: {reason}. Escrow held as CLAIMABLE for claim_bounty_payout." self.bounties[bounty_id] = bounty else: bounty.status = "OPEN" bounty.payout_status = "UNPAID" bounty.ai_verdict_reason = f"[APPEAL DENIED]: {reason or 'Appeal denied by consensus tribunal'}" self.bounties[bounty_id] = bounty @gl.public.write def cancel_bounty(self, bounty_id: str) -> None: """ HUNTER PROTECTION - ANTI-RUGPULL FREEZE: 1. Only Creator can cancel. 2. Time-lock: Must wait at least 300s (5 minutes) after creation. 3. Strict Anti-Frontrunning Freeze: If any submission records exist (submission_count > 0), cancellation is completely locked to protect Hunters from Creator snatching code! """ if bounty_id not in self.bounties: raise UserError("Bounty not found") bounty = self.bounties[bounty_id] if str(gl.message.sender_address).lower() != bounty.creator.lower(): raise UserError("Only the Creator can cancel this bounty.") if bounty.status != "OPEN": raise UserError("Bounty is not OPEN for cancellation.") current_time = self._now() if current_time < bounty.created_at + bigint(300): raise UserError("Creator Time-Lock Active: Must wait at least 5 minutes after creation to cancel.") if bounty.submission_count > bigint(0): raise UserError("Anti-Rugpull Lock: This bounty has active submission records. Creator cannot cancel or withdraw escrow while security hunters have submitted patches.") bounty.status = "CANCELLED" has_sufficient_balance = False try: has_sufficient_balance = (self.balance >= bounty.reward_amount) except Exception: has_sufficient_balance = False if has_sufficient_balance: gl.get_contract_at(Address(bounty.creator)).emit_transfer(value=u256(bounty.reward_amount)) bounty.payout_status = "REFUNDED" bounty.ai_verdict_reason = "Cancelled by creator after timelock expired with no active submissions. Escrow refunded." else: bounty.payout_status = "CLAIMABLE" bounty.ai_verdict_reason = "Cancelled by creator. Automatic refund held (insufficient balance). Escrow held as CLAIMABLE for claim_bounty_payout." self.bounties[bounty_id] = bounty @gl.public.write def claim_bounty_payout(self, bounty_id: str) -> None: """ SAFE RECOVERABLE SETTLEMENT CLAIM METHOD (PULL-OVER-PUSH): Allows the verified winner (hunter) or creator (for cancelled bounty) to pull pending escrow if automatic push transfer was withheld or held. Guarantees that escrow funds are never locked or unrecoverable. """ if bounty_id not in self.bounties: raise UserError("Bounty not found") bounty = self.bounties[bounty_id] if bounty.payout_status == "PAID": raise UserError("Bounty payout has already been successfully transferred (PAID).") if bounty.payout_status == "REFUNDED": raise UserError("Bounty escrow has already been refunded (REFUNDED).") caller = str(gl.message.sender_address).lower() if bounty.status == "RESOLVED": if caller != bounty.winner.lower(): raise UserError("Only the verified winner can claim this resolved bounty payout.") if self.balance < bounty.reward_amount: raise UserError("Contract liquid balance insufficient for payout. Sponsor must top up.") gl.get_contract_at(Address(bounty.winner)).emit_transfer(value=u256(bounty.reward_amount)) bounty.payout_status = "PAID" bounty.ai_verdict_reason += " [Escrow payout claimed successfully via claim_bounty_payout]" self.bounties[bounty_id] = bounty elif bounty.status == "CANCELLED": if caller != bounty.creator.lower(): raise UserError("Only the creator can claim a cancelled bounty refund.") if self.balance < bounty.reward_amount: raise UserError("Contract liquid balance insufficient for refund.") gl.get_contract_at(Address(bounty.creator)).emit_transfer(value=u256(bounty.reward_amount)) bounty.payout_status = "REFUNDED" bounty.ai_verdict_reason += " [Escrow refund claimed successfully via claim_bounty_payout]" self.bounties[bounty_id] = bounty else: raise UserError("Bounty is not in a claimable state (must be RESOLVED or CANCELLED with unclaimed escrow).") @gl.public.view def get_bounty(self, bounty_id: str) -> str: """View returns JSON string for easiest compatibility with genlayer-js / Studio""" if bounty_id not in self.bounties: return "{}" b = self.bounties[bounty_id] return json.dumps({ "id": b.id, "creator": b.creator, "title": b.title, "target_repo_url": b.target_repo_url, "vulnerability_description": b.vulnerability_description, "expected_fix_criteria": b.expected_fix_criteria, "reward_amount": str(b.reward_amount), "status": b.status, "winner": b.winner, "ai_verdict_reason": b.ai_verdict_reason, "patch_pr_url": b.patch_pr_url, "commit_hash": getattr(b, "commit_hash", ""), "last_submitter": getattr(b, "last_submitter", ""), "created_at": str(b.created_at), "submission_count": str(b.submission_count), "payout_status": getattr(b, "payout_status", "UNPAID"), }) @gl.public.view def get_bounty_count(self) -> int: return len(self.bounty_ids) @gl.public.view def get_all_bounties(self) -> str: all_list = [] for i in range(len(self.bounty_ids)): bid = self.bounty_ids[i] if bid in self.bounties: b = self.bounties[bid] all_list.append({ "id": b.id, "creator": b.creator, "title": b.title, "target_repo_url": b.target_repo_url, "vulnerability_description": b.vulnerability_description, "expected_fix_criteria": b.expected_fix_criteria, "reward_amount": str(b.reward_amount), "status": b.status, "winner": b.winner, "ai_verdict_reason": b.ai_verdict_reason, "patch_pr_url": b.patch_pr_url, "commit_hash": getattr(b, "commit_hash", ""), "last_submitter": getattr(b, "last_submitter", ""), "created_at": str(b.created_at), "submission_count": str(b.submission_count), "payout_status": getattr(b, "payout_status", "UNPAID"), }) return json.dumps(all_list) @gl.public.view def get_contract_balance(self) -> str: """Returns the contract's liquid native GEN balance held in custody.""" try: return str(self.balance) except Exception: return "0"