Bluesky migration to your own PDS

migrate.py

import atproto
import atproto_crypto.algs
import atproto_crypto.did
import atproto_crypto.consts
import tqdm
import cryptography.hazmat.primitives.asymmetric.ec as ec
import cryptography.hazmat.primitives.serialization as serialization


'''
This script was mostly copied from 
https://github.com/bluesky-social/pds/blob/main/ACCOUNT_MIGRATION.md
'''

OLD_PDS_URL = 'https://bsky.social'
NEW_PDS_URL = 'https://example.com'
CURRENT_HANDLE = 'to-migrate.bsky.social'
CURRENT_PASSWORD = 'password'
NEW_HANDLE = 'migrated.example.com'
NEW_ACCOUNT_EMAIL = '[email protected]'
NEW_ACCOUNT_PASSWORD = 'password'
NEW_PDS_INVITE_CODE = 'example-com-12345-abcde'

def main():
    old_client = atproto.Client(OLD_PDS_URL)
    new_client = atproto.Client(NEW_PDS_URL)
    profile = old_client.login(CURRENT_HANDLE, CURRENT_PASSWORD)
    print("{} logged in".format(profile.display_name))
    account_did = profile.did

    # Create account
    describe_res = new_client.com.atproto.server.describe_server()
    new_server_did = describe_res.did
    service_jwt_res = old_client.com.atproto.server.get_service_auth({
        "aud": new_server_did,
        "lxm": "com.atproto.server.createAccount"
    })
    service_jwt = service_jwt_res.token
    print("Obtained JWT from old server.")

    new_client.com.atproto.server.create_account({
        "handle": NEW_HANDLE,
        "email": NEW_ACCOUNT_EMAIL,
        "password": NEW_ACCOUNT_PASSWORD,
        "did": account_did,
        "invite_code": NEW_PDS_INVITE_CODE,
    }, headers={
        "authorization": "Bearer " + service_jwt,
        "encoding": "application/json"
    })
    # new_client.login will not work since it will call app.bsky.actor.get_profile but we are deactivated
    new_profile = new_client._get_and_set_session(NEW_HANDLE, NEW_ACCOUNT_PASSWORD)
    print("Account {} created".format(new_profile.did))

    # migrate data
    repo = old_client.com.atproto.sync.get_repo({"did": account_did})
    new_client.com.atproto.repo.import_repo(repo)

    print("Migrating blobs...")
    with tqdm.tqdm(total=0) as progress:
        blob_cursor = None
        while True:
            listed_blobs = old_client.com.atproto.sync.list_blobs({
                "did": account_did,
                "cursor": blob_cursor,
            })
            items = len(listed_blobs.cids)
            if items == 0:
                break
            progress.total += items
            progress.refresh()
            for cid in listed_blobs.cids:
                blob_res = old_client.com.atproto.sync.get_blob({
                    "did": account_did,
                    "cid": cid
                })
                new_client.com.atproto.repo.upload_blob(blob_res)
                progress.update(1)
            blob_cursor = listed_blobs.cursor

    prefs = old_client.app.bsky.actor.get_preferences()
    new_client.app.bsky.actor.put_preferences({
        "preferences": prefs.preferences
    })
    print("Data migrated")

    # Migrate Identity
    recover_key = ec.generate_private_key(ec.SECP256K1())
    private_key = hex(recover_key.private_numbers().private_value)[2:]
    old_client.com.atproto.identity.request_plc_operation_signature()
    get_did_cred = new_client.com.atproto.identity.get_recommended_did_credentials()
    rotation_keys = get_did_cred.rotation_keys
    if not rotation_keys:
        raise Exception("No rotation key provided")

    public_key = recover_key.public_key().public_bytes(
        encoding=serialization.Encoding.X962,
        format=serialization.PublicFormat.CompressedPoint
    )
    req = {
        "token": input("Please enter token from your email: ").strip(),
        "rotation_keys": [
            atproto_crypto.did.format_did_key(atproto_crypto.consts.SECP256K1_JWT_ALG, public_key),
            *rotation_keys
        ],
        "alsoKnownAs": get_did_cred.also_known_as,
        "verificationMethods": get_did_cred.verification_methods,
        "services": get_did_cred.services,
    }
    plc_op = old_client.com.atproto.identity.sign_plc_operation(req)
    print("❗ Your private recovery key is: {}. Please store this in a secure location! ❗".format(private_key))

    new_client.com.atproto.identity.submit_plc_operation({
        "operation": plc_op.operation
    })

    # Finalize Migration
    new_client.com.atproto.server.activate_account()
    old_client.com.atproto.server.deactivate_account({})
    print("Migration success!!!")

if __name__ == '__main__':
    main()

requirements.txt

atproto==0.0.58
tqdm==4.67.1
cryptography==43.0.3