#!/usr/bin/env python3 """zhiva_verify.py — независимая проверка цепи Живы (НОВЫЙ МИР). Написан отдельно от ноды: другой язык, другой код. Проверяет каждую строку цепи: 1) подпись владельца ключа над канон-строкой события (Ed25519); 2) решения модератора: подписи одобряющих над "Approve|<канон>" и наличие подписи модератора; 3) сцепку: хвост = sha256("<прошлый хвост>|<хеш события>"), поле prev совпадает с хвостом. Запуск: python3 zhiva_verify.py # скачать с nw.com.ru и проверить python3 zhiva_verify.py zhiva.chain # проверить свой файл python3 zhiva_verify.py --replica ~/zhiva-replica # реплика: хранить копию и следить, что прошлое не переписано Нужно: pip install cryptography """ import hashlib, json, os, sys, time, urllib.request URL = "https://nw.com.ru/chain/zhiva.chain" MODERATOR = os.environ.get("ZHIVA_MODERATOR", "c1bf6d7b103d70093dd6d77060659a359761af58d7c238f3e8dcfdcf77224089") RESPONSIBLE = set() # ключи ответственных — как в ноде RESPONSIBLE_REQUIRED = 0 def sha(s): return hashlib.sha256(s.encode()).hexdigest() def canon(e): t = e.get("type"); g = lambda k: e.get(k, "") if t == "IdentityRegister": return f"IdentityRegister|{g('name')}|{g('pubkey')}|{g('birth_hash')}|{g('ts')}" if t == "Endorse": return f"Endorse|{g('sponsor_pubkey')}|{g('candidate_hash')}|{g('ts')}" if t == "NodeJoin": return f"NodeJoin|{g('pubkey')}|{g('node')}|{g('ts')}" if t == "NodeCreate": return f"NodeCreate|{g('pubkey')}|{g('node')}|{g('about_hash')}|{g('ts')}" if t == "Invite": return f"Invite|{g('pubkey')}|{g('host')}|{g('ts')}" if t == "Transfer": return f"Transfer|{g('from')}|{g('to')}|{g('amount')}|{g('note')}|{g('ts')}" if t == "NodeProfile": return f"NodeProfile|{g('pubkey')}|{g('node')}|{g('profile_hash')}|{g('ts')}" if t == "TaskDone": return f"TaskDone|{g('pubkey')}|{g('task_id')}|{g('photo_hash')}|{g('ts')}" if t == "GameReward": return f"GameReward|{g('pubkey')}|{g('task_id')}|{g('amount')}|{g('verdict')}|{g('ts')}" if t == "BirthAmend": return f"BirthAmend|{g('pubkey')}|{g('new_birth_hash')}|{g('reason_hash')}|{g('req_ts')}" if t == "Merit": return f"Merit|{g('pubkey')}|{g('amount')}|{g('reason')}|{g('refid')}" if t == "NodeRename": return f"NodeRename|{g('pubkey')}|{g('node')}|{g('new_name')}|{g('ts')}" if t == "AssetMint": return f"AssetMint|{g('pubkey')}|{g('asset_id')}|{g('kind')}|{g('title')}|{sha(g('about'))}|{g('ts')}" if t == "MarketList": return f"MarketList|{g('pubkey')}|{g('asset_id')}|{g('price')}|{g('ts')}" if t == "DealOpen": return f"DealOpen|{g('pubkey')}|{g('asset_id')}|{g('price')}|{g('ts')}" if t == "DealStep": return f"DealStep|{g('pubkey')}|{g('deal_id')}|{g('step')}|{g('ts')}" if t == "Dispute": return f"Dispute|{g('pubkey')}|{g('deal_id')}|{g('reason_hash')}|{g('ts')}" if t == "DisputeResolve": return f"DisputeResolve|{g('deal_id')}|{g('winner')}|{g('note_hash')}" raise ValueError("неизвестный тип события: %s" % t) DECISIONS = ("BirthAmend", "Merit", "DisputeResolve") def ev_hash(e): c = canon(e) if e.get("type") in DECISIONS: a = ",".join("%s:%s" % (x["pubkey"], x["sig"]) for x in e.get("approvals", [])) return sha("%s|%s|%s" % (c, e.get("ts"), a)) return sha(c) def verify_sig(pk, sig, msg): from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey try: Ed25519PublicKey.from_public_bytes(bytes.fromhex(pk)).verify(bytes.fromhex(sig), msg.encode()); return True except Exception: return False def check_event(e): t = e.get("type"); c = canon(e) if t == "GameReward": return e.get("sig") == "oracle", "награда без печати оракула" if t in DECISIONS: seen = [] for a in e.get("approvals", []): if a["pubkey"] in seen: return False, "одобряющий повторяется" if a["pubkey"] != MODERATOR and a["pubkey"] not in RESPONSIBLE: return False, "одобряет посторонний ключ" if not verify_sig(a["pubkey"], a["sig"], "Approve|" + c): return False, "подпись одобрения не сходится" seen.append(a["pubkey"]) if MODERATOR not in seen: return False, "нет подписи модератора" if len([k for k in seen if k in RESPONSIBLE]) < RESPONSIBLE_REQUIRED: return False, "мало ответственных" if t != "BirthAmend": return True, "" key = e.get("from") if t == "Transfer" else e.get("sponsor_pubkey") if t == "Endorse" else e.get("pubkey") return verify_sig(key, e.get("sig", ""), c), "подпись владельца не сходится" def verify_lines(lines): tail = ""; ok = 0; bad = []; tails = [] for i, line in enumerate(lines, 1): if not line.strip(): continue try: d = json.loads(line) except Exception: bad.append((i, "не JSON")); continue e = d.get("ev", d); prev = d.get("prev") if "ev" in d else None try: good, why = check_event(e); h = ev_hash(e) except Exception as x: bad.append((i, str(x))); continue if prev is not None and prev != tail: good, why = False, "prev не совпадает с хвостом — цепь разорвана" if good: ok += 1 else: bad.append((i, "%s · %s" % (e.get("type"), why))) tail = sha("%s|%s" % (tail, h)); tails.append(tail) return ok, bad, tail, tails def main(): args = sys.argv[1:] replica = None if args and args[0] == "--replica": replica = os.path.expanduser(args[1]); args = args[2:] if args: raw = open(args[0], encoding="utf-8").read() else: raw = urllib.request.urlopen(URL, timeout=30).read().decode("utf-8") lines = [l for l in raw.splitlines() if l.strip()] ok, bad, tail, tails = verify_lines(lines) print("событий: %d · целых: %d · хвост: %s" % (len(lines), ok, tail)) for i, why in bad[:20]: print(" ✗ строка %d: %s" % (i, why)) status = "ЦЕПЬ ЦЕЛА" if not bad else "ЕСТЬ НАРУШЕНИЯ" if replica: os.makedirs(replica, exist_ok=True); mine = os.path.join(replica, "zhiva.chain") if os.path.exists(mine): old = [l for l in open(mine, encoding="utf-8").read().splitlines() if l.strip()] if lines[:len(old)] != old: status = "ПРОШЛОЕ ПЕРЕПИСАНО" cut = next((i for i, (a, b) in enumerate(zip(old, lines)) if a != b), min(len(old), len(lines))) open(os.path.join(replica, "ALERT-%s.txt" % time.strftime("%Y%m%d-%H%M")), "w").write( "Строка %d отличается от сохранённой копии. Старая копия сохранена рядом.\n" % (cut + 1)) os.replace(mine, mine + ".before-" + time.strftime("%Y%m%d-%H%M")) if not bad and status != "ПРОШЛОЕ ПЕРЕПИСАНО": open(mine + ".tmp", "w", encoding="utf-8").write("\n".join(lines) + "\n"); os.replace(mine + ".tmp", mine) with open(os.path.join(replica, "log.txt"), "a") as fh: fh.write("%s · %s · %d событий · %s\n" % (time.strftime("%Y-%m-%d %H:%M"), status, len(lines), tail)) print(status) sys.exit(0 if status == "ЦЕПЬ ЦЕЛА" else 1) if __name__ == "__main__": main()