extro.rook.host
This repository has no description
9.5 kB
283 lines
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3
4"""
5Minimal DID:PLC genesis op generator + registrar.
6
7Requirements (Python 3.9+):
8 pip install cryptography dag-cbor base58 requests
9
10Example:
11 # Minimal (k256, no AKA/services):
12 python plc_genesis_register.py
13
14 # With handle and PDS (repeatable; does not upload keys):
15 python plc_genesis_register.py --aka at://alice.example --pds https://pds.example.com
16
17 # Use p256 instead of k256:
18 python plc_genesis_register.py --curve p256 --aka at://bob.example --pds https://pds.example.com
19"""
20
21import argparse
22import base64
23import dataclasses
24import hashlib
25import json
26import pathlib
27import sys
28import typing as t
29
30import base58
31import dag_cbor
32import requests
33from cryptography.hazmat.backends import default_backend
34from cryptography.hazmat.primitives import hashes, serialization
35from cryptography.hazmat.primitives.asymmetric import ec, utils
36
37# --- Constants from atproto crypto spec (multicodec varints for compressed pubkeys) ---
38# p256-pub code 0x1200 -> varint bytes [0x80, 0x24]
39# secp256k1-pub code 0xE7 -> varint bytes [0xE7, 0x01]
40# Ref: https://atproto.com/specs/cryptography
41MC_P256_PUB = bytes([0x80, 0x24])
42MC_K256_PUB = bytes([0xE7, 0x01])
43
44# Curve orders (for low-S canonicalization)
45# Ref: standards / well-known constants
46N_K256 = int("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141", 16)
47N_P256 = int("FFFFFFFF00000000FFFFFFFFFFFFFFFFBCE6FAADA7179E84F3B9CAC2FC632551", 16)
48
49
50def ensure_dir(path: pathlib.Path) -> None:
51 path.mkdir(parents=True, exist_ok=True)
52
53
54def b64url_nopad(data: bytes) -> str:
55 return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
56
57
58def base32_lower_nopad(data: bytes) -> str:
59 # Python emits uppercase with padding; PLC uses lowercase, no padding
60 return base64.b32encode(data).decode("ascii").lower().rstrip("=")
61
62
63def compress_pubkey_bytes(pubkey) -> bytes:
64 # cryptography supports X9.62 compressed point for p256 and k256
65 return pubkey.public_bytes(
66 encoding=serialization.Encoding.X962,
67 format=serialization.PublicFormat.CompressedPoint,
68 )
69
70
71def did_key_for_pub(curve: str, pubkey_bytes_compressed: bytes) -> str:
72 if curve == "k256":
73 pref = MC_K256_PUB
74 elif curve == "p256":
75 pref = MC_P256_PUB
76 else:
77 raise ValueError("curve must be 'k256' or 'p256'")
78 mb = b"z" + base58.b58encode(pref + pubkey_bytes_compressed)
79 return "did:key:" + mb.decode("ascii")
80
81
82def low_s_r_s(curve: str, r: int, s: int) -> t.Tuple[int, int]:
83 if curve == "k256":
84 n = N_K256
85 else:
86 n = N_P256
87 # Enforce low-S (s <= n/2)
88 if s > n // 2:
89 s = n - s
90 return r, s
91
92
93def sign_low_s_raw(curve: str, private_key, message: bytes) -> bytes:
94 # cryptography signs ECDSA over the given hash algorithm
95 der_sig = private_key.sign(message, ec.ECDSA(hashes.SHA256()))
96 r, s = utils.decode_dss_signature(der_sig)
97 r, s = low_s_r_s(curve, r, s)
98 # raw 64-byte r||s big-endian
99 rb = r.to_bytes(32, "big")
100 sb = s.to_bytes(32, "big")
101 return rb + sb
102
103
104@dataclasses.dataclass
105class KeyBundle:
106 curve: str # 'k256' or 'p256'
107 priv_pem_path: pathlib.Path
108 pub_pem_path: pathlib.Path
109 did_key: str
110 signing_key_bytes: bytes # compressed point (33 bytes)
111
112
113def generate_rotation_key(outdir: pathlib.Path, curve: str, verbose: bool = False) -> KeyBundle:
114 ensure_dir(outdir)
115 if curve == "k256":
116 priv = ec.generate_private_key(ec.SECP256K1(), backend=default_backend())
117 elif curve == "p256":
118 priv = ec.generate_private_key(ec.SECP256R1(), backend=default_backend())
119 else:
120 raise ValueError("curve must be 'k256' or 'p256'")
121
122 pub = priv.public_key()
123
124 # Save keys (unencrypted PEM for simplicity; protect these in real deployments)
125 priv_pem = priv.private_bytes(
126 encoding=serialization.Encoding.PEM,
127 format=serialization.PrivateFormat.PKCS8,
128 encryption_algorithm=serialization.NoEncryption(),
129 )
130 pub_pem = pub.public_bytes(
131 encoding=serialization.Encoding.PEM,
132 format=serialization.PublicFormat.SubjectPublicKeyInfo,
133 )
134
135 priv_path = outdir / f"rotation_{curve}_private.pem"
136 pub_path = outdir / f"rotation_{curve}_public.pem"
137 priv_path.write_bytes(priv_pem)
138 pub_path.write_bytes(pub_pem)
139
140 # did:key for rotation key is allowed;
141 # PLC only requires rotationKeys to be did:key of k256 or p256
142 compressed = compress_pubkey_bytes(pub)
143 didk = did_key_for_pub(curve, compressed)
144
145 if verbose:
146 print(f"[verbose] Generated {curve.upper()} keypair")
147 print(f"[verbose] Compressed pubkey: {compressed.hex()}")
148
149 return KeyBundle(
150 curve=curve,
151 priv_pem_path=priv_path,
152 pub_pem_path=pub_path,
153 did_key=didk,
154 signing_key_bytes=compressed,
155 )
156
157
158def build_unsigned_op(
159 rotation_did_keys: t.List[str], aka_list: t.List[str], pds_endpoint: t.Optional[str]
160) -> dict:
161 # Minimal regular op shape; empty maps/lists are fine
162 verification_methods = {}
163 services = {}
164 if pds_endpoint:
165 services["atproto_pds"] = {
166 "type": "AtprotoPersonalDataServer",
167 "endpoint": pds_endpoint,
168 }
169 return {
170 "type": "plc_operation",
171 "rotationKeys": rotation_did_keys,
172 "verificationMethods": verification_methods,
173 "alsoKnownAs": aka_list,
174 "services": services,
175 "prev": None, # must be present and null for genesis
176 # 'sig' inserted after signing
177 }
178
179
180def derive_plc_did(signed_op: dict) -> str:
181 cbor = dag_cbor.encode(signed_op)
182 digest = hashlib.sha256(cbor).digest()
183 suffix = base32_lower_nopad(digest)[:24]
184 return "did:plc:" + suffix
185
186
187def main():
188 ap = argparse.ArgumentParser(description="Generate & register a DID:PLC genesis operation.")
189 ap.add_argument(
190 "--out", default="plc_keys", help="Output directory for keys (default: plc_keys)"
191 )
192 ap.add_argument("--curve", choices=["k256", "p256"], default="k256", help="Rotation key curve")
193 ap.add_argument(
194 "--aka",
195 action="append",
196 default=[],
197 help="alsoKnownAs entry (e.g., at://alice.example). May repeat.",
198 )
199 ap.add_argument("--pds", default=None, help="PDS endpoint URL (e.g., https://pds.example.com)")
200 ap.add_argument("--dry-run", action="store_true", help="Build & print but do not POST to PLC")
201 ap.add_argument("-v", "--verbose", action="store_true", help="Show verbose output")
202 args = ap.parse_args()
203
204 outdir = pathlib.Path(args.out)
205 kb = generate_rotation_key(outdir, args.curve, args.verbose)
206
207 unsigned = build_unsigned_op([kb.did_key], args.aka, args.pds)
208
209 if args.verbose:
210 print("[verbose] Unsigned operation:")
211 print(json.dumps(unsigned, indent=2))
212
213 # Sign DAG-CBOR bytes of unsigned op (without 'sig')
214 unsigned_cbor = dag_cbor.encode(unsigned)
215 if args.verbose:
216 print(f"[verbose] Encoded CBOR size: {len(unsigned_cbor)} bytes")
217
218 # cryptography ECDSA expects the message; spec requires ECDSA-SHA256;
219 # library will hash internally. To sign the SHA256 digest explicitly,
220 # replace message with hashlib.sha256(unsigned_cbor).digest()
221 # and switch to ec.Prehashed(hashes.SHA256()).
222 # Low-S enforcement + raw r||s per atproto crypto guidance.
223 priv = serialization.load_pem_private_key(kb.priv_pem_path.read_bytes(), password=None)
224 raw_sig = sign_low_s_raw(args.curve, priv, unsigned_cbor)
225 sig_b64u = b64url_nopad(raw_sig)
226
227 if args.verbose:
228 print(f"[verbose] Signature (base64url): {sig_b64u}")
229
230 signed = dict(unsigned)
231 signed["sig"] = sig_b64u
232
233 did = derive_plc_did(signed)
234
235 if args.verbose:
236 signed_cbor = dag_cbor.encode(signed)
237 digest = hashlib.sha256(signed_cbor).digest()
238 print(f"[verbose] SHA256 of signed op: {digest.hex()}")
239
240 # Save artifacts
241 (outdir / "genesis_unsigned.dag-cbor").write_bytes(unsigned_cbor)
242 (outdir / "genesis_signed.json").write_text(
243 json.dumps(signed, separators=(",", ":"), ensure_ascii=False) + "\n"
244 )
245 (outdir / "did.txt").write_text(did + "\n")
246 (outdir / "rotation_did_key.txt").write_text(kb.did_key + "\n")
247
248 if args.verbose:
249 print(f"[verbose] Wrote genesis_unsigned.dag-cbor ({len(unsigned_cbor)} bytes)")
250 print("[verbose] Wrote genesis_signed.json")
251 print("[verbose] Wrote did.txt")
252 print("[verbose] Wrote rotation_did_key.txt")
253
254 print(f"Rotation key (did:key): {kb.did_key}")
255 print(f"DID (derived): {did}")
256 print(f"Wrote keys & artifacts to: {outdir.resolve()}")
257
258 if args.dry_run:
259 print("Dry run selected; not POSTing to PLC.")
260 return
261
262 # Submit to PLC directory (HTTP POST JSON to .../{did})
263 url = f"https://plc.directory/{did}"
264 if args.verbose:
265 print(f"[verbose] POSTing to {url}")
266 print("[verbose] Request body:")
267 print(json.dumps(signed, indent=2))
268
269 try:
270 resp = requests.post(url, json=signed, timeout=10)
271 print(f"POST {url} -> {resp.status_code}")
272 print(resp.text[:5000])
273 if 200 <= resp.status_code < 300:
274 print("Registration appears successful.")
275 else:
276 print("Registration failed; see response above.")
277 except Exception as e:
278 print(f"Error POSTing to PLC: {e}", file=sys.stderr)
279 sys.exit(2)
280
281
282if __name__ == "__main__":
283 main()