create_yubikey_did_plc.py

from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
from ykman.device import list_all_devices
from yubikit.core.smartcard import SmartCardConnection
from yubikit.piv import PivSession, SLOT, KEY_TYPE
import multiformats
import dag_cbor
import hashlib
import base64
import requests
import json

HANDLE = "yubikey.dev.retr0.id"
PDS_SERVER = "pds.dev.retr0.id"
PLC_SERVER = "plc.bsky-sandbox.dev"

"""
(you should probably read the yubikey docs before proceeding - this is a PoC)


Generate a key with:

yubico-piv-tool --action generate --slot 9c --algorithm ECCP256 --hash SHA256

put the result in yk_pubkey.pem
"""

def encode_did_pubkey(pubkey: ec.EllipticCurvePublicKey):
	assert(type(pubkey.curve) is ec.SECP256R1)
	compressed_public_bytes = pubkey.public_bytes(
		serialization.Encoding.X962,
		serialization.PublicFormat.CompressedPoint
	)
	return "did:key:" + multiformats.multibase.encode(
		multiformats.multicodec.wrap("p256-pub", compressed_public_bytes),
		"base58btc"
	)

pubkey = serialization.load_pem_public_key(open("yk_pubkey.pem", "rb").read())

genesis = {
	"type": "plc_operation",
	"rotationKeys": [
		encode_did_pubkey(pubkey),
	],
	"verificationMethods": {
		"atproto": encode_did_pubkey(pubkey), #XXX should really be separate from rotationKeys
	},
	"alsoKnownAs": [
		"at://" + HANDLE
	],
	"services": {
		"atproto_pds": {
			"type": "AtprotoPersonalDataServer",
			"endpoint": "https://" + PDS_SERVER
		}
	},
	"prev": None,
}

genesis_bytes = dag_cbor.encode(genesis)

dev, info = list_all_devices()[0] # XXX: assume only 1 device
with dev.open_connection(SmartCardConnection) as connection:
	piv = PivSession(connection)
	piv.verify_pin(input("Yubikey PIV PIN: "))

	dss_sig = piv.sign(
		slot=SLOT.SIGNATURE,
		key_type=KEY_TYPE.ECCP256, # secp256r1
		hash_algorithm=hashes.SHA256(),
		message=genesis_bytes
	)

r, s = decode_dss_signature(dss_sig)

# apply low-s malleability mitigation
SECP256R1_N = 0xffffffff00000000ffffffffffffffffbce6faada7179e84f3b9cac2fc632551
if s > SECP256R1_N // 2:
	s = SECP256R1_N - s

signature = base64.urlsafe_b64encode(r.to_bytes(32, "big") + s.to_bytes(32, "big")).decode().strip("=")
signed_genesis = genesis | {"sig": signature}
signed_genesis_bytes = dag_cbor.encode(signed_genesis)

plc = "did:plc:" + base64.b32encode(hashlib.sha256(signed_genesis_bytes).digest())[:24].lower().decode()

json_blob = json.dumps(signed_genesis, indent="\t")

with open("yk_did_plc.txt", "w") as logfile:
	print("Created DID:", plc, file=logfile)
	print("", file=logfile)
	print(json_blob, file=logfile)

print(json_blob)
print()
print("Created DID:", plc)

print("Publishing...")

plc_url = "https://" + PLC_SERVER + "/" + plc
r = requests.post(plc_url, json=signed_genesis)
print(r, r.content)
assert(r.ok)

print("Your PLC should now be live at", plc_url)