chadtmiller.com/pds.js

Switch repository

Loading…

// After an inbound migration the DID document changes (new signing key, new PDS
// endpoint). The PDS must announce an #identity firehose event so relays and
// AppViews re-resolve the DID — otherwise they keep verifying the account's
// service tokens against the *old* signing key and reject them with
// "could not verify jwt signature". These tests pin the two places that must
// emit it: activateAccount (the last migration step) and submitPlcOperation
// (where the DID doc actually changes).

import { afterEach, describe, expect, it, vi } from 'vitest';
import { PersonalDataServer } from '../packages/core/src/pds.js';
import { cborDecode } from '../packages/core/src/repo.js';

const HOSTNAME = 'pds.example.com';
const DID = 'did:plc:migratedaccount00000000000';
const HANDLE = 'alice.example.com';
const FULL_ACCESS = { did: DID, scope: 'atproto' };

/** A minimal initialized actor store that records emitted firehose events. */
function createStorage() {
  const state = { status: 'deactivated' };
  let seq = 0;
  /** @type {any[]} */
  const events = [];
  return {
    _events: events,
    _state: state,
    async getDid() {
      return DID;
    },
    async getHandle() {
      return HANDLE;
    },
    async getKemKey() {
      return null;
    },
    async getAccountStatus() {
      return state.status;
    },
    async setAccountStatus(/** @type {string} */ s) {
      state.status = s;
    },
    async getLatestSeq() {
      return seq;
    },
    async putEvent(
      /** @type {number} */ s,
      /** @type {string} */ _did,
      /** @type {string} */ _cid,
      /** @type {Uint8Array} */ evt,
    ) {
      seq = s;
      events.push(cborDecode(evt));
    },
  };
}

/**
 * @param {ReturnType<typeof createStorage>} storage
 * @param {string} [relayUrl]
 */
function createServer(storage, relayUrl) {
  return new PersonalDataServer({
    ...(relayUrl ? { relayUrl } : {}),
    actorStorage: /** @type {any} */ (storage),
    sharedStorage: /** @type {any} */ ({ putActor: async () => {} }),
    blobs: /** @type {any} */ ({}),
    jwtSecret: 'test-secret',
    hostname: HOSTNAME,
    password: 'test-password',
    plcUrl: 'https://plc.test',
    // No relayUrl: keeps notifyRelay a no-op so no network is touched.
  });
}

describe('migration announces identity so tokens verify against the new key', () => {
  afterEach(() => {
    vi.unstubAllGlobals();
  });

  it('activateAccount emits both #account (active) and #identity', async () => {
    const storage = createStorage();
    const pds = createServer(storage);

    const activate =
      pds.routes['/xrpc/com.atproto.server.activateAccount'].handler;
    const res = await activate.call(
      pds,
      new Request(
        `https://${HOSTNAME}/xrpc/com.atproto.server.activateAccount`,
        {
          method: 'POST',
          body: '{}',
        },
      ),
      new URL(`https://${HOSTNAME}/`),
      /** @type {any} */ (FULL_ACCESS),
    );
    expect(res.status).toBe(200);
    expect(storage._state.status).toBe('active');

    const types = storage._events.map((e) => e.type);
    expect(types).toContain('account');
    expect(types).toContain('identity');
    // Exactly one of each — setAccountActive is the sole #account emitter, so a
    // duplicate here would mean the handler double-announced.
    expect(types.filter((t) => t === 'account')).toHaveLength(1);
    expect(types.filter((t) => t === 'identity')).toHaveLength(1);

    const account = storage._events.find((e) => e.type === 'account');
    expect(account.active).toBe(true);

    const identity = storage._events.find((e) => e.type === 'identity');
    expect(identity.did).toBe(DID);
    expect(identity.handle).toBe(HANDLE);
  });

  it('activateAccount without full access is rejected and emits nothing', async () => {
    const storage = createStorage();
    const pds = createServer(storage);

    const activate =
      pds.routes['/xrpc/com.atproto.server.activateAccount'].handler;
    const res = await activate.call(
      pds,
      new Request(
        `https://${HOSTNAME}/xrpc/com.atproto.server.activateAccount`,
        {
          method: 'POST',
          body: '{}',
        },
      ),
      new URL(`https://${HOSTNAME}/`),
      /** @type {any} */ ({ did: DID, scope: 'repo:app.bsky.feed.post' }),
    );
    expect(res.status).toBe(403);
    expect(storage._events).toHaveLength(0);
  });

  it('submitPlcOperation emits #identity once the PLC directory accepts it', async () => {
    const storage = createStorage();
    const pds = createServer(storage);

    // Stub the PLC directory POST to succeed.
    const calls = /** @type {string[]} */ ([]);
    vi.stubGlobal(
      'fetch',
      vi.fn(async (/** @type {string} */ url) => {
        calls.push(url);
        return new Response('', { status: 200 });
      }),
    );

    const submit =
      pds.routes['/xrpc/com.atproto.identity.submitPlcOperation'].handler;
    const res = await submit.call(
      pds,
      new Request(
        `https://${HOSTNAME}/xrpc/com.atproto.identity.submitPlcOperation`,
        {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({ operation: { type: 'plc_operation' } }),
        },
      ),
      new URL(`https://${HOSTNAME}/`),
      /** @type {any} */ (FULL_ACCESS),
    );
    expect(res.status).toBe(200);
    // It actually submitted to the PLC directory for this DID.
    expect(calls.some((u) => u.includes(encodeURIComponent(DID)))).toBe(true);

    const identity = storage._events.filter((e) => e.type === 'identity');
    expect(identity).toHaveLength(1);
    expect(identity[0].did).toBe(DID);
  });

  it('submitPlcOperation emits nothing when the PLC directory rejects it', async () => {
    const storage = createStorage();
    const pds = createServer(storage);
    vi.stubGlobal(
      'fetch',
      vi.fn(async () => new Response('nope', { status: 400 })),
    );

    const submit =
      pds.routes['/xrpc/com.atproto.identity.submitPlcOperation'].handler;
    const res = await submit.call(
      pds,
      new Request(
        `https://${HOSTNAME}/xrpc/com.atproto.identity.submitPlcOperation`,
        {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({ operation: { type: 'plc_operation' } }),
        },
      ),
      new URL(`https://${HOSTNAME}/`),
      /** @type {any} */ (FULL_ACCESS),
    );
    expect(res.status).toBe(400);
    expect(storage._events).toHaveLength(0);
  });
});

// The announcement at activation reaches nobody when no relay has subscribed
// yet, and no consumer retries it. maybeAnnounceIdentity repeats it each hour
// until the relay reports that it reads this server.
describe('maybeAnnounceIdentity', () => {
  const RELAY = 'https://relay.test';
  const STATUS = `${RELAY}/xrpc/com.atproto.sync.getHostStatus?hostname=${HOSTNAME}`;

  afterEach(() => {
    vi.unstubAllGlobals();
  });

  /** @param {any} status - What the relay answers, or null for a 404 */
  function stubRelay(status) {
    /** @type {string[]} */
    const calls = [];
    vi.stubGlobal(
      'fetch',
      vi.fn(async (/** @type {string} */ url) => {
        calls.push(String(url));
        if (String(url) === STATUS) {
          return status
            ? new Response(JSON.stringify(status))
            : new Response(JSON.stringify({ error: 'HostNotFound' }), {
                status: 404,
              });
        }
        return new Response('{}');
      }),
    );
    return calls;
  }

  it('announces while the relay has never read this server', async () => {
    const storage = createStorage();
    storage._state.status = 'active';
    const pds = createServer(storage, RELAY);

    const calls = stubRelay({ seq: -1, accountCount: 0, status: 'active' });
    expect(await pds.maybeAnnounceIdentity()).toBe(true);

    const identity = storage._events.filter((e) => e.type === 'identity');
    expect(identity).toHaveLength(1);
    expect(identity[0].handle).toBe(HANDLE);
    // It also asks for the crawl, which is what gets a relay subscribed at all.
    expect(calls.some((u) => u.includes('com.atproto.sync.requestCrawl'))).toBe(
      true,
    );
  });

  it('announces when the relay does not know the host', async () => {
    const storage = createStorage();
    storage._state.status = 'active';
    const pds = createServer(storage, RELAY);

    stubRelay(null);
    expect(await pds.maybeAnnounceIdentity()).toBe(true);
    expect(storage._events.filter((e) => e.type === 'identity')).toHaveLength(
      1,
    );
  });

  it('stops once the relay reads this server', async () => {
    const storage = createStorage();
    storage._state.status = 'active';
    const pds = createServer(storage, RELAY);

    stubRelay({ seq: 42, accountCount: 1, status: 'active' });
    expect(await pds.maybeAnnounceIdentity()).toBe(false);
    expect(storage._events).toHaveLength(0);
  });

  it('says nothing for a deactivated account', async () => {
    const storage = createStorage();
    storage._state.status = 'deactivated';
    const pds = createServer(storage, RELAY);

    stubRelay({ seq: -1, accountCount: 0 });
    expect(await pds.maybeAnnounceIdentity()).toBe(false);
    expect(storage._events).toHaveLength(0);
  });

  it('says nothing when no relay is configured', async () => {
    const storage = createStorage();
    storage._state.status = 'active';
    const pds = createServer(storage);
    const fetchSpy = vi.fn();
    vi.stubGlobal('fetch', fetchSpy);

    expect(await pds.maybeAnnounceIdentity()).toBe(false);
    expect(fetchSpy).not.toHaveBeenCalled();
  });

  it('waits for the next tick when the relay is unreachable', async () => {
    const storage = createStorage();
    storage._state.status = 'active';
    const pds = createServer(storage, RELAY);
    vi.stubGlobal(
      'fetch',
      vi.fn(async () => {
        throw new Error('network down');
      }),
    );

    expect(await pds.maybeAnnounceIdentity()).toBe(false);
    expect(storage._events).toHaveLength(0);
  });
});