# v0.2.16 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } """ Klin — founder-deployed Intelligent Contract. Writes: register(identity) deposit() save_message(text, response, provenance) Stores the pair bound to model provenance. The contract hashes question + answer + provider + model + generation_id itself. Caller-supplied text is not a seal. verify_message(message_id) Nondet: validators fetch current public sources and fact-check claims. Equivalence of two model wordings is NOT enough for verified=true. withdraw() / withdraw_to(destination) Reads: get_name, get_version, get_owner, get_identity, get_balance, get_history, get_history_count, get_message, get_save_fee, get_verify_fee, get_verify_mode """ from genlayer import * import hashlib import json MAX_TEXT = 4000 MAX_RESPONSE = 8000 MAX_HISTORY = 40 MAX_IDENTITY = 200 MAX_PROVENANCE = 600 SAVE_FEE = u256(10**14) VERIFY_FEE = u256(10**15) @gl.evm.contract_interface class _Eoa: class View: pass class Write: pass class Klin(gl.Contract): owner: Address identities: TreeMap[str, str] balances: TreeMap[str, u256] history_json: TreeMap[str, str] next_id: u256 trusted_signer: Address nonces: TreeMap[str, u256] def __init__(self): # NOTE: do NOT instantiate TreeMap() here. `nonces` is a declared storage # field (persistent) like `identities`/`balances`/`history_json`; re-creating # it in __init__ would wipe replay protection on re-execution. GenLayer # auto-initializes declared TreeMap fields to an empty map. self.owner = gl.message.sender_address self.next_id = u256(1) self.trusted_signer = Address("0x0000000000000000000000000000000000000000") def _k(self, addr) -> str: return str(addr) def _load_history(self, addr) -> list: key = self._k(addr) if key in self.history_json: try: data = json.loads(self.history_json[key]) if isinstance(data, list): return data except Exception: return [] return [] def _save_history(self, addr, items: list) -> None: if len(items) > MAX_HISTORY: items = items[-MAX_HISTORY:] self.history_json[self._k(addr)] = json.dumps(items) def _balance_of(self, addr) -> u256: key = self._k(addr) if key in self.balances: return self.balances[key] return u256(0) def _credit(self, addr, amount: u256) -> None: key = self._k(addr) self.balances[key] = self._balance_of(addr) + amount def _debit(self, addr, amount: u256) -> None: key = self._k(addr) bal = self._balance_of(addr) if bal < amount: raise gl.vm.UserError("insufficient contract balance") self.balances[key] = bal - amount def _send_gen(self, to: Address, amount: u256) -> None: _Eoa(to).emit_transfer(value=amount) def _clip(self, value: str, limit: int) -> str: text = value if isinstance(value, str) else str(value or "") if len(text) > limit: return text[:limit] return text def _content_hash(self, question: str, answer: str, meta: dict) -> str: digest = hashlib.sha256() blob = "\n".join( [ question, answer, str(meta.get("provider") or ""), str(meta.get("model") or ""), str(meta.get("generation_id") or ""), str(meta.get("generated_at") or ""), ] ) digest.update(blob.encode("utf-8")) return digest.hexdigest() def _attest_message(self, sender: Address, question: str, answer: str, meta: dict) -> str: """Canonical ASCII payload, identical to the server's buildAttestMessage. Only hex addresses/hashes and the ASCII provider/model strings are used, so Python len() and the client's UTF-8 byte length agree on the prefix. CRITICAL: `separators=(",", ":")` emits COMPACT JSON with no spaces. The server signs this payload via viem's `JSON.stringify`, which never adds whitespace. If we used the Python default separators (", ", ": "), the contract would recover the signer over DIFFERENT bytes than the server signed, so ecrecover returns the wrong address and EVERY legitimate attestation is rejected as "unauthorized provenance signer". """ return json.dumps( { "a": str(sender).lower(), "g": str(meta.get("generation_id") or "")[:80], "m": str(meta.get("model") or "")[:80], "p": str(meta.get("provider") or "")[:40], "q": Keccak256(question.encode("utf-8")).digest().hex()[:64], "r": Keccak256(answer.encode("utf-8")).digest().hex()[:64], "t": int(meta.get("generated_at") or 0), }, separators=(",", ":"), ) def _recover_signer(self, message: str, signature: str) -> Address: """Recover the ECDSA signer of an EIP-191 personal message. `signature` is a 65-byte blob (r || s || v). - `v` may be 27/28 (legacy) or 0/1 (normalized). Both are accepted; we normalize to 0/1 because that is what the EVM recovery primitive expects. NOTE: `gl.ecrecover` is the EVM-compatible recovery primitive exposed by GenLayer's py-genlayer. If the runtime names it differently, change only this one call. """ msg_bytes = message.encode("utf-8") # Canonical EIP-191 personal-message prefix (NO separator after the # length). This MUST match the prefix viem's `account.signMessage` / # `hashMessage` computes, otherwise the contract recovers the wrong # signer and every legitimate attestation is rejected: # keccak256("\x19Ethereum Signed Message:\n" + len(payload) + payload) prefix = ( b"\x19Ethereum Signed Message:\n" + str(len(msg_bytes)).encode("ascii") ) digest = Keccak256(prefix + msg_bytes).digest() try: sig = bytes.fromhex(str(signature).removeprefix("0x")) except Exception: raise gl.vm.UserError("bad signature encoding") if len(sig) != 65: raise gl.vm.UserError("signature must be 65 bytes") r = sig[:32] s = sig[32:64] v = int(sig[64]) # Normalize legacy 27/28 to 0/1 for the recovery primitive. if v in (27, 28): v = v - 27 elif v not in (0, 1): raise gl.vm.UserError("invalid signature recovery byte") try: recovered = gl.ecrecover(digest, v, r, s) except Exception: raise gl.vm.UserError("invalid signature format or recovery failure") if not recovered: raise gl.vm.UserError("invalid signature format or recovery failure") return recovered def _parse_signature(self, raw: str) -> str: blob = (raw or "").strip() if blob == "": return "" if blob[:2] != "0x" or len(blob) != 132: raise gl.vm.UserError("signature must be 0x + 130 hex") return blob def _parse_provenance(self, raw: str) -> dict: blob = (raw or "").strip() if len(blob) < 2 or len(blob) > MAX_PROVENANCE: raise gl.vm.UserError("provenance required") try: data = json.loads(blob) except Exception: raise gl.vm.UserError("provenance must be JSON") if not isinstance(data, dict): raise gl.vm.UserError("provenance must be JSON") provider = self._clip(str(data.get("provider") or "").strip(), 40) model = self._clip(str(data.get("model") or "").strip(), 80) generation_id = self._clip(str(data.get("generation_id") or "").strip(), 80) generated_at = data.get("generated_at") if provider == "" or model == "": raise gl.vm.UserError("provenance must name provider and model") try: generated_at_n = int(generated_at) except Exception: generated_at_n = 0 if generated_at_n < 0: generated_at_n = 0 return { "provider": provider, "model": model, "generation_id": generation_id, "generated_at": generated_at_n, } @gl.public.write def register(self, identity: str) -> None: sender = gl.message.sender_address cleaned = (identity or "").strip() if len(cleaned) < 1 or len(cleaned) > MAX_IDENTITY: raise gl.vm.UserError("identity must be 1-200 characters") self.identities[self._k(sender)] = cleaned @gl.public.write.payable def deposit(self) -> None: sender = gl.message.sender_address incoming = gl.message.value if incoming <= u256(0): raise gl.vm.UserError("deposit value must be positive") self._credit(sender, incoming) @gl.public.write def save_message( self, text: str, response: str, provenance: str, signature: str, nonce: u256, ) -> None: sender = gl.message.sender_address question = self._clip((text or "").strip(), MAX_TEXT) answer = self._clip(response or "", MAX_RESPONSE) if len(question) < 1: raise gl.vm.UserError("text must not be empty") meta = self._parse_provenance(provenance) # Anti-replay: each sender's nonce must advance by exactly one. expected_nonce = self.nonces.get(self._k(sender), u256(0)) if nonce != expected_nonce: raise gl.vm.UserError("invalid nonce") # Authenticated provenance: verify the server attestation before binding. # Dev mode (trusted_signer == 0 address) accepts unsealed entries so the # preview works without a configured signer; production must set it. sig = self._parse_signature(signature) attested = False if self.trusted_signer != Address("0x0000000000000000000000000000000000000000"): if sig == "": raise gl.vm.UserError("attestation signature required") message = self._attest_message(sender, question, answer, meta) recovered = self._recover_signer(message, sig) if str(recovered).lower() != str(self.trusted_signer).lower(): raise gl.vm.UserError("unauthorized provenance signer") attested = True self._debit(sender, SAVE_FEE) self.nonces[self._k(sender)] = expected_nonce + u256(1) message_id = str(int(self.next_id)) items = self._load_history(sender) items.append( { "id": message_id, "text": question, "response": answer, "provider": meta["provider"], "model": meta["model"], "generation_id": meta["generation_id"], "generated_at": meta["generated_at"], "content_hash": self._content_hash(question, answer, meta), "verified": False, "checked": False, "sources": "", "attested": attested, } ) self._save_history(sender, items) self.next_id = self.next_id + u256(1) def _qenc(self, text: str) -> str: out = [] for ch in text[:90]: o = ord(ch) if ch == " ": out.append("+") elif (48 <= o <= 57) or (65 <= o <= 90) or (97 <= o <= 122) or ch in "-._": out.append(ch) else: for b in ch.encode("utf-8"): out.append("%{:02X}".format(b)) return "".join(out) def _web_text(self, url: str) -> str: try: raw = gl.nondet.web.get(url) body = getattr(raw, "body", raw) if isinstance(body, bytes): text = body.decode("utf-8", "replace") else: text = str(body or "") if len(text) > 2800: return text[:2800] return text except Exception: return "" def _collect_sources(self, question: str) -> str: q = self._qenc(question) if not q: return "" ddg = self._web_text( "https://api.duckduckgo.com/?q=" + q + "&format=json&no_html=1&skip_disambig=1" ) wiki = self._web_text( "https://en.wikipedia.org/w/api.php?action=query&list=search&srsearch=" + q + "&srlimit=3&format=json&utf8=1" ) combined = (ddg + "\n" + wiki).strip() if len(combined) > 4500: return combined[:4500] return combined @gl.public.write def verify_message(self, message_id: str) -> None: sender = gl.message.sender_address target = str(message_id or "").strip() if not target: raise gl.vm.UserError("unknown message") items = self._load_history(sender) found = None index = -1 for i, item in enumerate(items): if str(item.get("id")) == target: found = item index = i break if found is None: raise gl.vm.UserError("unknown message") self._debit(sender, VERIFY_FEE) question = str(found.get("text") or "") original = str(found.get("response") or "") meta = { "provider": str(found.get("provider") or ""), "model": str(found.get("model") or ""), "generation_id": str(found.get("generation_id") or ""), # MUST include generated_at: save_message stored the content_hash WITH # it, so recomputing without it would always miss and mark verified=False. "generated_at": int(found.get("generated_at") or 0), } expected_hash = str(found.get("content_hash") or "") actual_hash = self._content_hash(question, original, meta) if (not expected_hash) or expected_hash != actual_hash: items[index]["verified"] = False items[index]["checked"] = True items[index]["sources"] = "hash-mismatch" self._save_history(sender, items) return provider = meta["provider"] or "unknown" model = meta["model"] or "unknown" def check_sources() -> str: evidence = self._collect_sources(question) if not evidence.strip(): return "false" prompt = f"""You are an independent fact-checking validator. Question: {question} Recorded answer (from {model} via {provider}): {original} Current source material fetched now from DuckDuckGo Instant Answer and Wikipedia search: {evidence} Rules: - Extract factual claims in the recorded answer. - Check those claims against the source material. - verified=true only if checkable claims are supported by the sources, or the answer has no checkable factual claim (pure how-to / opinion). - verified=false if claims are unsupported, contradicted, stale, or a checkable claim has no source coverage. - A claim is "supported" only when the source material actually states or entails it; do not infer from unrelated passages. - Do NOT mark verified just because you would phrase the answer similarly. Return ONLY JSON: {{"verified": true}} or {{"verified": false}} """ raw = gl.nondet.exec_prompt(prompt) cleaned = raw.strip() if cleaned.startswith("```"): cleaned = cleaned.strip("`") cleaned = cleaned.replace("json", "", 1).strip() start = cleaned.find("{") end = cleaned.rfind("}") if start != -1 and end > start: cleaned = cleaned[start : end + 1] parsed = json.loads(cleaned) flag = bool(parsed.get("verified")) return "true" if flag else "false" verdict = gl.eq_principle.strict_eq(check_sources) ok = str(verdict).strip().lower() == "true" items[index]["verified"] = ok items[index]["checked"] = True items[index]["sources"] = "ddg+wikipedia" items[index]["evidence"] = self._collect_sources(question)[:4000] self._save_history(sender, items) def _withdraw_to(self, destination: Address) -> None: sender = gl.message.sender_address amount = self._balance_of(sender) if amount <= u256(0): raise gl.vm.UserError("nothing to withdraw") self.balances[self._k(sender)] = u256(0) self._send_gen(destination, amount) @gl.public.write def withdraw(self) -> None: self._withdraw_to(gl.message.sender_address) @gl.public.write def withdraw_to(self, destination: Address) -> None: self._withdraw_to(destination) @gl.public.view def get_name(self) -> str: return "Klin" @gl.public.view def get_version(self) -> str: return "0.5.0" @gl.public.view def get_owner(self) -> str: return str(self.owner) @gl.public.view def get_identity(self, address: Address) -> str: key = self._k(address) if key in self.identities: return self.identities[key] return "" @gl.public.view def get_balance(self, address: Address) -> u256: return self._balance_of(address) @gl.public.view def get_history(self, address: Address) -> str: return json.dumps(self._load_history(address)) @gl.public.view def get_history_count(self, address: Address) -> u256: return u256(len(self._load_history(address))) @gl.public.view def get_message(self, address: Address, message_id: str) -> str: target = str(message_id or "").strip() for item in self._load_history(address): if str(item.get("id")) == target: return json.dumps(item) return "" @gl.public.view def get_save_fee(self) -> u256: return SAVE_FEE @gl.public.view def get_verify_fee(self) -> u256: return VERIFY_FEE @gl.public.view def get_verify_mode(self) -> str: return "sources" @gl.public.write def set_trusted_signer(self, signer: Address) -> None: if gl.message.sender_address != self.owner: raise gl.vm.UserError("only owner can set trusted signer") self.trusted_signer = signer @gl.public.view def get_trusted_signer(self) -> str: return str(self.trusted_signer) @gl.public.view def get_attested(self, address: Address, message_id: str) -> bool: target = str(message_id or "").strip() for item in self._load_history(address): if str(item.get("id")) == target: return bool(item.get("attested") or False) return False