# 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_contribution: 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_contribution: 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[:2]).lower() return scheme, hostname, port, norm_path else: norm_path = "/" + "/".join(segments).lower() return scheme, hostname, port, norm_path def _validate_contribution_url(contribution_url: str, base_repo_url: str) -> None: t_scheme, t_host, t_port, t_path = _normalize_repo_url(contribution_url, is_contribution=True) b_scheme, b_host, b_port, b_path = _normalize_repo_url(base_repo_url, is_contribution=False) if t_scheme != b_scheme or t_port != b_port: raise UserError("Contribution URL protocol or port does not match registered repository origin") if t_host != b_host and not t_host.endswith("." + b_host): raise UserError("Contribution URL host does not match registered repository origin") # Enforce path binding: contribution path must strictly start with the registered repository path (/owner/repo) if not t_path.startswith(b_path + "/"): raise UserError("Contribution URL must belong to the registered repository path") # Enforce pull request, commit, or release endpoint requirement rel_path = t_path[len(b_path) + 1:].strip("/") segs = [s for s in rel_path.split("/") if s] if len(segs) < 2: raise UserError("Contribution URL must point to a specific pull request, commit, or release endpoint") endpoint_type = segs[0] if endpoint_type not in ("pull", "pulls", "commit", "commits", "releases"): raise UserError("URL must be a valid pull request (/pull/ID), commit (/commit/HASH), or release (/releases/TAG) endpoint") 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 ("CRITICAL_MAINTENANCE", "ROUTINE_UPDATE", "SPAM_INFLATION", "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 PackageRegistry: package_id: str maintainer: str package_name: str repository_base: str ecosystem: str total_attestations: bigint @allow_storage @dataclass class ContributionAttestation: attestation_id: str package_id: str contributor: str release_or_pr_url: str declared_summary: str status: str # PENDING | CRITICAL | ROUTINE | REJECTED_SPAM | ESCALATED verdict: str # CRITICAL_MAINTENANCE | ROUTINE_UPDATE | SPAM_INFLATION | ABORT confidence: bigint audit_rationale: str class Contract(gl.Contract): packages: TreeMap[str, PackageRegistry] attestations: TreeMap[str, ContributionAttestation] package_counter: bigint total_attestations_completed: bigint governance_arbiter: str def __init__(self): self.package_counter = bigint(0) self.total_attestations_completed = bigint(0) self.governance_arbiter = _addr_str(gl.message.sender_address) @gl.public.write def register_package( self, package_name: str, repository_base: str, ecosystem: str, ) -> str: """Register a critical open-source package or dependency.""" package_name = package_name.strip() repository_base = repository_base.strip() ecosystem = ecosystem.strip().upper() if len(package_name) < 2: raise UserError("Package name too short") if len(ecosystem) < 2: raise UserError("Ecosystem name too short") _normalize_repo_url(repository_base, is_contribution=False) self.package_counter += bigint(1) pid = str(self.package_counter) self.packages[pid] = PackageRegistry( package_id=pid, maintainer=_addr_str(gl.message.sender_address), package_name=package_name, repository_base=repository_base, ecosystem=ecosystem, total_attestations=bigint(0), ) return pid @gl.public.write def attest_contribution( self, package_id: str, contributor_addr: str, release_or_pr_url: str, declared_summary: str, ) -> str: """Audit and attest an open-source release or PR contribution against spam/inflation.""" if package_id not in self.packages: raise UserError("Package not registered") pkg = self.packages[package_id] try: contrib_addr = Address(contributor_addr.strip()) except Exception: raise UserError("Invalid contributor address format") release_or_pr_url = release_or_pr_url.strip() declared_summary = declared_summary.strip() if len(declared_summary) < 15: raise UserError("Declared summary too short (min 15 chars)") # Enforce path binding: prevent same-host different-repository attacks _validate_contribution_url(release_or_pr_url, pkg.repository_base) pkg.total_attestations += bigint(1) self.total_attestations_completed += bigint(1) aid = package_id + "_" + str(pkg.total_attestations) self.attestations[aid] = ContributionAttestation( attestation_id=aid, package_id=package_id, contributor=contrib_addr.as_hex.lower(), release_or_pr_url=release_or_pr_url, declared_summary=declared_summary, status="PENDING", verdict="", confidence=bigint(0), audit_rationale="", ) self.packages[package_id] = pkg p_name = str(pkg.package_name) u_contrib = str(release_or_pr_url) summary = str(declared_summary) def leader_fn(): try: res = gl.nondet.web.render(u_contrib, mode="text") doc_text = res.content if hasattr(res, "content") else str(res) if not doc_text or len(doc_text.strip()) < 30: return {"verdict": "ABORT", "confidence": 0, "reason": "Contribution page returned empty text"} if any(err in doc_text[:400].lower() for err in ["404 not found", "error 404"]): return {"verdict": "ABORT", "confidence": 0, "reason": "Contribution page 404/unreachable"} except Exception as e: return {"verdict": "ABORT", "confidence": 0, "reason": f"Web fetch error: {str(e)}"} prompt = f""" SYSTEM: You are the Autonomous Open-Source Contribution & Dependency Integrity Verifier. Audit the target pull request or release for genuine development impact, bug fixes, or malicious spam inflation. PACKAGE NAME: {p_name} CONTRIBUTOR CLAIMED IMPACT: {summary} FETCHED REPOSITORY COMMIT / PR CONTENT: {doc_text[:4000]} Rules: - CRITICAL_MAINTENANCE (conf >= 75): Substantial security patch, architectural refactoring, critical bug resolution, or high-impact dependency maintenance. - ROUTINE_UPDATE (conf >= 75): Authentic standard contribution: minor bugfix, documentation enhancements, standard dependency bump, or non-critical improvement. - SPAM_INFLATION (conf >= 75): Artificial churn, trivial typo-farming, auto-generated bot commits, or zero substantive technical value intended to game public funding. - ABORT: Content is rate-limited, captcha-blocked, login-protected, or unreadable. OUTPUT ONLY STRICT JSON: {{ "verdict": "CRITICAL_MAINTENANCE" | "ROUTINE_UPDATE" | "SPAM_INFLATION" | "ABORT", "confidence": 0-100, "reason": "max 300 chars technical impact 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" att = self.attestations[aid] att.verdict = verdict att.confidence = bigint(confidence) att.audit_rationale = reason if verdict == "CRITICAL_MAINTENANCE": att.status = "CRITICAL" elif verdict == "ROUTINE_UPDATE": att.status = "ROUTINE" elif verdict == "SPAM_INFLATION": att.status = "REJECTED_SPAM" else: att.status = "ESCALATED" self.attestations[aid] = att return aid @gl.public.write def resolve_escalated_attestation( self, attestation_id: str, manual_status: str, override_reason: str, ) -> None: """Governance arbiter manually resolves an ESCALATED attestation caused by network rate limits.""" if attestation_id not in self.attestations: raise UserError("Attestation record not found") att = self.attestations[attestation_id] if att.status != "ESCALATED": raise UserError("Attestation is not in ESCALATED state") sender = _addr_str(gl.message.sender_address) if sender != self.governance_arbiter: raise UserError("Only authorized governance arbiter can resolve escalated attestations") s_upper = manual_status.strip().upper() if s_upper not in ("CRITICAL", "ROUTINE", "REJECTED_SPAM"): raise UserError("Invalid manual status assignment") att.status = s_upper att.verdict = f"RESOLVED_MANUALLY_{s_upper}" att.audit_rationale = f"Arbiter override ({sender}): {override_reason[:200]}" self.attestations[attestation_id] = att @gl.public.view def is_contribution_verified(self, attestation_id: str) -> bool: """Lightweight oracle query for Gitcoin, RetroPGF, or DAO grant distribution contracts.""" if attestation_id not in self.attestations: return False return self.attestations[attestation_id].status in ("CRITICAL", "ROUTINE") @gl.public.view def get_package(self, package_id: str) -> str: if package_id not in self.packages: raise UserError("Package not found") p = self.packages[package_id] return json.dumps({ "package_id": p.package_id, "maintainer": p.maintainer, "package_name": p.package_name, "repository_base": p.repository_base, "ecosystem": p.ecosystem, "total_attestations": str(p.total_attestations), }) @gl.public.view def get_attestation(self, attestation_id: str) -> str: if attestation_id not in self.attestations: raise UserError("Attestation not found") a = self.attestations[attestation_id] return json.dumps({ "attestation_id": a.attestation_id, "package_id": a.package_id, "contributor": a.contributor, "release_or_pr_url": a.release_or_pr_url, "declared_summary": a.declared_summary, "status": a.status, "verdict": a.verdict, "confidence": str(a.confidence), "audit_rationale": a.audit_rationale, }) @gl.public.view def get_metrics(self) -> str: return json.dumps({ "total_registered_packages": str(self.package_counter), "total_attestations_completed": str(self.total_attestations_completed), })