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