// Peer — a self-custody, single-tenant atproto node (peer/bridge split, Model B).
//
// The source of truth: it HOLDS the user's own signing key (derived from their seed /
// hyperident identity, never uploaded anywhere) and authors + SIGNS its own atproto repo
// locally over the atproto-store substrate. atproto-bridge later PROJECTS this peer to
// bsky (fetch repo over Hyperswarm -> transcode CAR; aggregate its firehose) — but the
// bridge never holds the key and never signs. Custody = key possession.
//
// This is the inversion of the old custodial Pds: there the server minted + held the key;
// here the peer is the sole key holder. See docs/DECISION-2026-07-03-peer-bridge-split.md
// (in atproto-bridge) — the no-key invariant.
import { readFileSync, writeFileSync, existsSync, mkdirSync } from 'bare-fs'
import { join } from 'bare-path'
import b4a from 'b4a'
import c from 'compact-encoding'
import ProtomuxRPC from 'protomux-rpc'
import Protomux from 'protomux'
import { randomBytes } from 'hypercore-crypto'
import { hashPassword, verifyPassword, signJwt, verifyJwt } from 'atproto-identity/lib/auth.js'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { RepoStore } from 'atproto-store/lib/repo-store.js'
import { RepoManager } from 'atproto-store/lib/repo-manager.js'
import { getRepo } from 'atproto-store/lib/read-view.js'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { createAtprotoIdentity, createAtprotoIdentityFromKey, deriveAtprotoKey, selfCustodyIdentity } from 'atproto-identity'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
const VAULT_FILE = 'peer-identity.json' // the peer's self-custody identity vault (device key + genesis)
const RPC_PROTOCOL = 'hyper-proto-peer/rpc/v1' // peer-signs-over-RPC channel, muxed over the swarm conn
const ANNOUNCE_PROTOCOL = 'atproto/did-announce/v1' // discovery push — MUST match serve-membrane + atproto-bridge PeerRouter
const ACCESS_TTL_SEC = 2 * 60 * 60 // 2h — standard atproto access-token lifetime
const REFRESH_TTL_SEC = 90 * 24 * 60 * 60 // 90d — standard atproto refresh-token lifetime
export class Peer {
constructor ({ corestore, did, handle, privateKey, didKey, signedOp, genesisOp = null, rotationOp = null, rootDidKey = null, rotationKeys = null, jwtSecret = null, appPasswordHash = null, clock, now }) {
this.corestore = corestore
this.did = did
this.handle = handle
this.privateKey = privateKey // SELF-HELD device signing key — never leaves the peer
this.didKey = didKey // the device signing key's did:key (= the DID doc's #atproto vm after rotation)
this.signedOp = signedOp // legacy/compat alias = the genesis op (single-key adopt/fromMnemonic paths)
this.genesisOp = genesisOp // deterministic, root-vm did:plc genesis (publish to PLC first)
this.rotationOp = rotationOp // post-genesis device-attestation rotation (publish second; vm → device)
this.rootDidKey = rootDidKey // the hyperident ROOT rotation key (recovery); null for single-key peers
this.rotationKeys = rotationKeys // rotation authority set baked into the genesis
this._jwtSecret = jwtSecret || null // peer-held HMAC secret for its OWN session tokens
this._appPasswordHash = appPasswordHash || null // scrypt hash of the app-password (bsky-app login)
this._clock = clock // micros clock for RepoManager rev (injectable for tests)
this._nowFn = now || (() => new Date().toISOString()) // ISO time stamped into firehose frames
this.store = null
this.mgr = null
this.seq = null // the peer's OWN signed firehose (FirehoseSequencer over a Hypercore)
}
// Self-custody onboarding: derive the identity from the user's OWN seed. The signing key
// is held here, on the peer. (A hyperident-anchored variant will derive the same key from
// the shared vault + attach the device/keet attestation — same custody, richer chain.)
static fromMnemonic ({ storage, mnemonic, handle, pdsUrl, clock, now } = {}) {
const id = createAtprotoIdentity(mnemonic, { handle, pdsUrl })
return Peer.open({ storage, handle, ...id, clock, now })
}
// Bring-your-own raw key (e.g. a key already held in the hyperident vault).
static fromPrivateKey ({ storage, privateKey, handle, pdsUrl, clock, now } = {}) {
const id = createAtprotoIdentityFromKey(privateKey, { handle, pdsUrl })
return Peer.open({ storage, handle, ...id, clock, now })
}
// ---- Init UX: hyperident-anchored, device-attested onboarding (the fresh-install flow) ----
//
// The atproto-correct self-custody model (atproto-base/DESIGN.md §4a). A DETERMINISTIC,
// root-derived genesis pins the DID (did:plc = f(rootMnemonic+handle+pdsUrl) — reproducible and
// device-INDEPENDENT), and a POST-genesis rotation attests a per-device SIGNING key. The DID is
// NOT a function of the device key, so it reproduces across runs and can span multiple devices
// (the flaw in the old buildAttestedIdentity path, which baked the device key into the genesis).
// The hyperident ROOT is the rotation/recovery authority: it signs the genesis + the rotation,
// then is discarded — never stored on the peer. If the device is lost, the root rotates to a new
// device and the DID/handle/repo persist.
//
// Persists the device key + BOTH ops (genesis, rotation) to a vault (`peer-identity.json`) so the
// peer RELOADS the same identity across restarts. The root seed never touches the vault. The
// returned peer holds `genesisOp` + `rotationOp` ready to submit to the PLC directory in order.
static async init ({ storage, rootMnemonic, deviceMnemonic, deviceKey, handle, pdsUrl, appPassword, clock, now } = {}) {
if (typeof storage !== 'string') throw new Error('Peer.init: storage must be a directory path')
if (!rootMnemonic) throw new Error('Peer.init: rootMnemonic (the hyperident recovery seed) is required')
mkdirSync(storage, { recursive: true })
const vaultPath = join(storage, VAULT_FILE)
if (existsSync(vaultPath)) throw new Error(`Peer.init: identity already exists at ${vaultPath} — use Peer.load`)
// device: use a provided device key/mnemonic, else selfCustodyIdentity mints a fresh one.
const dk = deviceKey || (deviceMnemonic ? deriveAtprotoKey(deviceMnemonic).privateKey : undefined)
const id = selfCustodyIdentity({ rootMnemonic, handle, pdsUrl, deviceKey: dk })
const vault = {
version: 1,
did: id.did,
handle,
pdsUrl,
devicePrivateKey: b4a.toString(id.deviceKey.privateKey, 'hex'),
deviceDidKey: id.deviceKey.didKey,
rootDidKey: id.rootDidKey,
rotationKeys: id.rotationKeys,
genesisOp: id.genesisOp, // deterministic root-vm genesis (submit to PLC first)
rotationOp: id.rotationOp, // device-attestation rotation, vm → device (submit second)
signedOp: id.genesisOp, // compat alias (peer.signedOp = the genesis)
// session auth (the peer is its own auth authority): a random HMAC secret for its session
// tokens, and — if set — an app-password hash so the bsky app can log in (the app can't
// sign with the user's key). Neither is the account key; the key stays the device key.
jwtSecret: b4a.toString(randomBytes(32), 'hex'),
appPasswordHash: appPassword ? hashPassword(appPassword) : null
}
writeFileSync(vaultPath, JSON.stringify(vault, null, 2))
return Peer._openFromVault(storage, vault, { clock, now })
}
// Set (or rotate) the app-password on an already-onboarded peer.
static setAppPassword ({ storage, appPassword } = {}) {
const vaultPath = join(storage, VAULT_FILE)
if (!existsSync(vaultPath)) throw new Error(`Peer.setAppPassword: no vault at ${vaultPath}`)
const vault = JSON.parse(readFileSync(vaultPath, 'utf8'))
vault.appPasswordHash = hashPassword(appPassword)
if (!vault.jwtSecret) vault.jwtSecret = b4a.toString(randomBytes(32), 'hex')
writeFileSync(vaultPath, JSON.stringify(vault, null, 2))
}
// Adopt an EXISTING did:plc (MIGRATION): open with a bring-your-own DID + signing key (no
// mint), persist a vault so the peer reloads it durably. Used to migrate a custodial account
// to self-custody WITHOUT changing the DID — the serviceEndpoint (the bridge) is unchanged,
// so no PLC op. signedOp is null (the DID is already published).
static async adopt ({ storage, did, handle, privateKey, didKey, pdsUrl, rotationKeys = null, rootDidKey = null, appPassword, clock, now } = {}) {
if (typeof storage !== 'string') throw new Error('Peer.adopt: storage must be a directory path')
if (!did || !privateKey || !didKey) throw new Error('Peer.adopt: did, privateKey, didKey required')
mkdirSync(storage, { recursive: true })
const vaultPath = join(storage, VAULT_FILE)
if (existsSync(vaultPath)) throw new Error(`Peer.adopt: identity already exists at ${vaultPath} — use Peer.load`)
const devicePrivateKey = typeof privateKey === 'string' ? privateKey : b4a.toString(b4a.from(privateKey), 'hex')
const vault = {
version: 1,
did,
handle,
pdsUrl,
devicePrivateKey,
deviceDidKey: didKey,
rootDidKey,
rotationKeys,
signedOp: null, // DID already published; not needed to serve
jwtSecret: b4a.toString(randomBytes(32), 'hex'),
appPasswordHash: appPassword ? hashPassword(appPassword) : null
}
writeFileSync(vaultPath, JSON.stringify(vault, null, 2))
return Peer._openFromVault(storage, vault, { clock, now })
}
// Import records (e.g. from a migrated repo CAR) — re-authors each under THIS peer's key.
// Record bytes are canonical DAG-CBOR so record CIDs are preserved (bsky sees the same posts);
// only the commit is fresh (a normal repo update at a higher rev).
async importRecords (records) {
const out = []
for (const r of records) out.push(await this.write({ collection: r.collection, rkey: r.rkey, record: r.value || r.record }))
return out
}
// Reload an already-onboarded peer from its vault (the normal startup path).
static async load ({ storage, clock, now } = {}) {
if (typeof storage !== 'string') throw new Error('Peer.load: storage must be a directory path')
const vaultPath = join(storage, VAULT_FILE)
if (!existsSync(vaultPath)) throw new Error(`Peer.load: no identity vault at ${vaultPath} — run Peer.init first`)
const vault = JSON.parse(readFileSync(vaultPath, 'utf8'))
return Peer._openFromVault(storage, vault, { clock, now })
}
static _openFromVault (storage, vault, { clock, now } = {}) {
return Peer.open({
storage,
did: vault.did,
handle: vault.handle,
privateKey: Uint8Array.from(b4a.from(vault.devicePrivateKey, 'hex')),
didKey: vault.deviceDidKey,
signedOp: vault.signedOp,
genesisOp: vault.genesisOp || vault.signedOp || null, // old vaults: the genesis was `signedOp`
rotationOp: vault.rotationOp || null,
rootDidKey: vault.rootDidKey,
rotationKeys: vault.rotationKeys,
jwtSecret: vault.jwtSecret,
appPasswordHash: vault.appPasswordHash,
clock,
now
})
}
static async open ({ storage, did, handle, privateKey, didKey, signedOp, genesisOp = null, rotationOp = null, rootDidKey, rotationKeys, jwtSecret, appPasswordHash, clock, now } = {}) {
// GOTCHA: Corestore CLEARS the directory it is given on init. So when `storage` is a path,
// nest the Corestore under <storage>/store/ and keep the identity vault in <storage>/
// (the parent) where the wipe can't reach it. A Corestore instance is used as-is.
const corestore = openCorestore(typeof storage === 'string' ? join(storage, 'store') : storage)
const peer = new Peer({ corestore, did, handle, privateKey, didKey, signedOp, genesisOp, rotationOp, rootDidKey, rotationKeys, jwtSecret, appPasswordHash, clock, now })
peer.store = await new RepoStore(corestore, { name: `atproto-repo:${did}` }).ready()
peer.mgr = await new RepoManager(peer.store, { did, privateKey, clock }).init()
peer.seq = await new FirehoseSequencer(corestore.get({ name: 'firehose' })).ready()
return peer
}
get rev () { return this.mgr.rev }
get commitCid () { return this.mgr.commitCid }
// Author a record with the peer's OWN key, then sequence the signed #commit into the peer's
// own firehose. The firehose frame CARRIES the signed commit block — so a bridge/replica
// reads the signature from the replicated firehose (keyless) instead of re-signing. The peer
// never hands over the key.
async _authorAndSequence (op) {
const write = await this.mgr.applyWrite(op)
const { seq } = await this.seq.sequenceCommit(write, { time: this._nowFn() })
return { ...write, seq }
}
write ({ collection, rkey, record }) {
return this._authorAndSequence({ action: 'create', collection, rkey, record })
}
update ({ collection, rkey, record }) {
return this._authorAndSequence({ action: 'update', collection, rkey, record })
}
delete ({ collection, rkey }) {
return this._authorAndSequence({ action: 'delete', collection, rkey })
}
// The signed repo as an atproto CAR (what a bridge fetches over the swarm and serves to bsky).
getRepo () {
return getRepo(this.store, { did: this.did, rev: this.mgr.rev, prev: null, privateKey: this.privateKey })
}
// The repo Hypercore's key — a bridge/replica opens it read-only (`new RepoStore(store,
// { key })`) after replicating this peer's Corestore, to read the byte-exact signed repo.
get repoKey () { return this.store.core.key }
get repoKeyHex () { return b4a.toString(this.store.core.key, 'hex') }
// The firehose Hypercore's key — a bridge/replica opens it read-only after replicating the
// Corestore, reads the latest #commit frame, and extracts the SIGNED commit (keyless). This
// is how the bridge gets a signature it cannot re-derive without the key.
get firehoseKey () { return this.seq.core.key }
get firehoseKeyHex () { return b4a.toString(this.seq.core.key, 'hex') }
// ---- sessions: the peer is its OWN auth authority (bsky-app login via app-password) ----
get canLogin () { return !!(this._jwtSecret && this._appPasswordHash) }
_nowSec () { const t = this._nowFn ? Date.parse(this._nowFn()) : Date.now(); return Math.floor(t / 1000) }
_mintSession () {
const nowSec = this._nowSec()
const base = { sub: this.did }
return {
did: this.did,
handle: this.handle,
accessJwt: signJwt({ ...base, scope: 'com.atproto.access', exp: nowSec + ACCESS_TTL_SEC }, this._jwtSecret),
refreshJwt: signJwt({ ...base, scope: 'com.atproto.refresh', exp: nowSec + REFRESH_TTL_SEC }, this._jwtSecret)
}
}
// Verify the app-password and mint session tokens with the peer-held jwtSecret (NOT the
// account key). The bsky app logs in here (relayed by the bridge); the bridge never verifies
// credentials nor mints tokens — the peer is the sole auth authority.
createSession ({ identifier, password } = {}) {
if (!this.canLogin) throw new Error('createSession: sessions not configured on this peer')
if (identifier && identifier !== this.handle && identifier !== this.did) throw new Error('createSession: unknown identifier')
if (!password || !verifyPassword(password, this._appPasswordHash)) throw new Error('createSession: invalid credentials')
return this._mintSession()
}
refreshSession (refreshJwt) {
if (!this.canLogin) throw new Error('refreshSession: sessions not configured on this peer')
const payload = verifyJwt(refreshJwt, this._jwtSecret, { nowSec: this._nowSec() }) // throws if invalid/expired
if (payload.scope !== 'com.atproto.refresh') throw new Error('refreshSession: not a refresh token')
return this._mintSession()
}
// Serve this peer's repo (its whole Corestore) for replication over Hyperswarm — the P2P
// backhaul the bridge pulls from. Rendezvous matches the fleet convention (topic = the
// swarm's own pubkey = serverKey), so atproto-bridge's fetch side (FleetNode /
// connectReplica) finds it unchanged. The peer replicates its cores; it NEVER sends its
// signing key — a replica gets the signed repo and verifies it locally.
serve (swarm, { canWrite = null } = {}) {
this._serverKey = b4a.toString(swarm.keyPair.publicKey, 'hex')
const onconn = (conn) => {
this.corestore.replicate(conn)
const rk = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
console.log(JSON.stringify({ msg: 'peer inbound connection', from: rk, did: this.did }))
conn.on('close', () => console.log(JSON.stringify({ msg: 'peer connection closed', from: rk })))
this._attachRpc(conn, { canWrite }) // apply-write / sessions / blobs (conditionally)
// ANNOUNCE-PUSH (atproto/did-announce/v1): push our core keys + lengths on connect + on each new
// commit, so a bridge/consumer learns them with NO round-trip and NO serverKey directionality.
// The SOLE discovery path (get-coords removed); matches serve-membrane + atproto-bridge PeerRouter.
try {
const achan = Protomux.from(conn).createChannel({ protocol: ANNOUNCE_PROTOCOL })
if (achan) {
const amsg = achan.addMessage({ encoding: c.json, onmessage: () => {} })
achan.open()
const push = () => { try { amsg.send({ did: this.did, handle: this.handle, repoKey: this.repoKeyHex, firehoseKey: this.firehoseKeyHex, repoLength: this.store.core.length, firehoseLength: this.seq.core.length }) } catch {} }
push()
const onapp = () => push()
this.seq.core.on('append', onapp)
conn.on('close', () => { try { this.seq.core.off('append', onapp) } catch {} })
}
} catch {}
}
swarm.on('connection', onconn)
// Join BOTH the serverKey topic (fleet-replicate compat) AND the DID-derived discovery
// topic — the latter makes the peer findable from the DID alone (Option 2), no DID-doc entry.
// client:true is LOAD-BEARING, not just for discovery: both the peer AND the bridge are
// typically firewalled (NAT'd self-custody box ↔ NAT'd bridge host). A hyperswarm member that
// joins client:false never sends outbound holepunch packets, so its NAT mapping never opens and
// the peer punching from the other side is dropped. Two firewalled peers must BOTH be client:true
// to punch simultaneously — proven: symmetric connects in ~2s, asymmetric never opens (cAtt>0,
// cOpen:0). Mirrors hyperident-gateway's rock-solid {server:true,client:true} mesh.
const keyTopic = swarm.keyPair.publicKey
const didTopic = discoveryTopic(this.did)
const dKey = swarm.join(keyTopic, { server: true, client: true })
const dDid = swarm.join(didTopic, { server: true, client: true })
return {
serverKey: this._serverKey,
discoveryTopic: b4a.toString(didTopic, 'hex'),
ready: () => Promise.all([dKey.flushed(), dDid.flushed()]),
close: () => swarm.off('connection', onconn)
}
}
// Peer-signs-over-RPC (Model B): expose the keyed operations over the SAME muxed swarm
// connection (protomux-rpc shares Protomux.from(conn) with the corestore replication). A
// bridge — the DID's serviceEndpoint for a NAT'd peer — relays the user's request here and
// the PEER applies+signs / verifies+mints; the bridge never holds the key or the jwtSecret.
// apply-write -> peer signs a repo write; gated by canWrite(remotePubKeyHex).
// create-session -> peer verifies the app-password + mints tokens (gated by the password).
// refresh-session -> peer re-mints from a valid refresh token.
_attachRpc (conn, { canWrite } = {}) {
const rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json })
// Discovery is ANNOUNCE-PUSH (see serve()): the bridge learns our repo/firehose keys from the
// pushed announce, replicates, and verifies our repo signs against the DID key. This RPC channel
// is only the keyed relay (apply-write / sessions / blobs) — no get-coords.
if (canWrite) {
rpc.respond('apply-write', async (op) => {
const remote = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
if (!remote || !canWrite(remote)) throw new Error(`unauthorized: caller may not write to ${this.did}`)
const w = await this._authorAndSequence(op)
return { did: this.did, commitCid: w.commitCid.toString(), rev: w.rev, seq: w.seq }
})
}
if (this.canLogin) {
rpc.respond('create-session', async (params) => this.createSession(params))
rpc.respond('refresh-session', async ({ refreshJwt } = {}) => this.refreshSession(refreshJwt))
}
return rpc
}
async close () {
if (this.seq) await this.seq.close()
if (this.store) await this.store.close()
// close the Corestore too — it holds the RocksDB fd-lock; without this a reopen
// (Peer.load) on the same dir fails with "File descriptor could not be locked".
if (this.corestore) await this.corestore.close()
}
}