Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
11 kB · 317 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318package spxrpc
import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "net/http" "net/http/httptest" "net/url" "testing"
"github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/xrpc" "github.com/ipfs/go-cid" "github.com/ipld/go-car" carutil "github.com/ipld/go-car/util" "github.com/labstack/echo/v4" "github.com/multiformats/go-multihash" "github.com/stretchr/testify/require"
"stream.place/streamplace/pkg/atproto" "stream.place/streamplace/pkg/comatproto" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/reposync" "stream.place/streamplace/pkg/statedb")
// newSyncTestNode stands up a real file-backed server repo and an echo router// carrying the generated com.atproto.* routes, so the sync handlers are// exercised through the same stub layer production uses (query parsing,// status codes and all).//// serverHost == broadcasterHost puts the node in single-PDS mode, where// isServerPDS is true no matter what Host header the request carries — which// is what lets an httptest server (Host: 127.0.0.1:port) reach the server repo.func newSyncTestNode(t *testing.T, serverHost, broadcasterHost string) (*config.CLI, *echo.Echo) { t.Helper() cli := &config.CLI{ BroadcasterHost: broadcasterHost, ServerHost: serverHost, DBURL: ":memory:", } cli.DataDir = t.TempDir() mod, err := model.MakeDB(":memory:") require.NoError(t, err) state, err := statedb.MakeDB(context.Background(), cli, nil, mod) require.NoError(t, err)
// The server repo lives in package-level globals; don't let a previous // test in this binary leak into this one. atproto.ServerRepo = nil atproto.ServerCarStore = nil atproto.ServerPubMultibase = ""
handle, err := atproto.MakeServerRepo(context.Background(), cli, state) require.NoError(t, err) t.Cleanup(func() { _ = handle.Close() })
s := &Server{cli: cli} e := echo.New() e.Use(s.ContextPreservingMiddleware()) require.NoError(t, s.RegisterHandlersComatproto(e)) return cli, e}
// commitTestRecord writes one record at collection/rkey. The value is always a// viewerCount because the MST doesn't care what the bytes say — only the key// placement matters for these tests.func commitTestRecord(t *testing.T, cli *config.CLI, collection, rkey string) { t.Helper() updatedAt := "2026-03-21T00:00:00Z" vc := placestream.LiveViewerCount{ LexiconTypeID: constants.PLACE_STREAM_LIVE_VIEWERCOUNT, Count: 7, Server: cli.ServerDID(), Streamer: "did:plc:" + rkey, UpdatedAt: &updatedAt, } require.NoError(t, atproto.CommitServerRepoRecord(context.Background(), cli, collection, rkey, &vc))}
// syncGet issues a GET against e as if it arrived on host.func syncGet(t *testing.T, e *echo.Echo, host, method string, q url.Values) *httptest.ResponseRecorder { t.Helper() req := httptest.NewRequest(http.MethodGet, fmt.Sprintf("http://%s/xrpc/%s?%s", host, method, q.Encode()), nil) rec := httptest.NewRecorder() e.ServeHTTP(rec, req) return rec}
// readRootlessCAR parses a getBlocks response: a CARv1 whose header declares no// roots. go-car's NewCarReader rejects that shape, so this mirrors what// pkg/reposync does with raw car.ReadHeader.func readRootlessCAR(t *testing.T, raw []byte) map[cid.Cid][]byte { t.Helper() br := bufio.NewReader(bytes.NewReader(raw)) hdr, err := car.ReadHeader(br) require.NoError(t, err) require.EqualValues(t, 1, hdr.Version) require.Empty(t, hdr.Roots, "getBlocks must return a rootless CAR") out := map[cid.Cid][]byte{} for { c, data, err := carutil.ReadNode(br) if err == io.EOF { return out } require.NoError(t, err) out[c] = data }}
func TestComAtprotoSyncGetLatestCommitAndGetBlocks(t *testing.T) { const host = "sync1.example.com" cli, e := newSyncTestNode(t, host, "broadcaster.example.com") did := atproto.ServerRepo.RepoDid() require.Equal(t, "did:web:"+host, did)
commitTestRecord(t, cli, constants.PLACE_STREAM_LIVE_VIEWERCOUNT, "aaa") commitTestRecord(t, cli, constants.PLACE_STREAM_MEDIA_ORIGIN, "babczxv1")
// getLatestCommit reports a decodable commit CID plus its rev. rec := syncGet(t, e, host, "com.atproto.sync.getLatestCommit", url.Values{"did": {did}}) require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) var latest comatproto.SyncGetLatestCommit_Output require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &latest)) commitCID, err := cid.Decode(latest.Cid) require.NoError(t, err) require.NotEmpty(t, latest.Rev)
// ...and it agrees with the repo's own view of head. wantCID, wantRev, err := atproto.ServerRepoLatestCommit(context.Background()) require.NoError(t, err) require.Equal(t, wantCID.String(), latest.Cid) require.Equal(t, wantRev, latest.Rev)
// A repo we don't host is a 404, same as getRepo. rec = syncGet(t, e, host, "com.atproto.sync.getLatestCommit", url.Values{"did": {"did:web:somewhere.else"}}) require.Equal(t, http.StatusNotFound, rec.Code) require.Contains(t, rec.Body.String(), "RepoNotFound")
// getBlocks hands back exactly the requested block, and the bytes hash // to the CID getLatestCommit just advertised. rec = syncGet(t, e, host, "com.atproto.sync.getBlocks", url.Values{ "did": {did}, "cids": {latest.Cid}, }) require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) blocks := readRootlessCAR(t, rec.Body.Bytes()) require.Len(t, blocks, 1) require.Contains(t, blocks, commitCID) require.NoError(t, reposync.VerifyBlock(commitCID, blocks[commitCID]))
// Several CIDs in one call, deduped. rec = syncGet(t, e, host, "com.atproto.sync.getBlocks", url.Values{ "did": {did}, "cids": {latest.Cid, latest.Cid}, }) require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) require.Len(t, readRootlessCAR(t, rec.Body.Bytes()), 1)
// A CID we don't have fails the whole request rather than quietly // returning a short bag of blocks. missing, err := cid.Prefix{ Version: 1, Codec: cid.DagCBOR, MhType: multihash.SHA2_256, MhLength: 32, }.Sum([]byte("definitely not in this repo")) require.NoError(t, err) rec = syncGet(t, e, host, "com.atproto.sync.getBlocks", url.Values{ "did": {did}, "cids": {latest.Cid, missing.String()}, }) require.Equal(t, http.StatusNotFound, rec.Code) require.Contains(t, rec.Body.String(), "BlockNotFound")
// Garbage CID -> InvalidRequest, not a 500. rec = syncGet(t, e, host, "com.atproto.sync.getBlocks", url.Values{ "did": {did}, "cids": {"not-a-cid"}, }) require.Equal(t, http.StatusBadRequest, rec.Code) require.Contains(t, rec.Body.String(), "InvalidRequest")
// Over the per-request cap -> InvalidRequest. tooMany := make([]string, maxGetBlocksCIDs+1) for i := range tooMany { tooMany[i] = latest.Cid } rec = syncGet(t, e, host, "com.atproto.sync.getBlocks", url.Values{ "did": {did}, "cids": tooMany, }) require.Equal(t, http.StatusBadRequest, rec.Code) require.Contains(t, rec.Body.String(), "InvalidRequest")
// Wrong repo -> RepoNotFound. rec = syncGet(t, e, host, "com.atproto.sync.getBlocks", url.Values{ "did": {"did:web:somewhere.else"}, "cids": {latest.Cid}, }) require.Equal(t, http.StatusNotFound, rec.Code) require.Contains(t, rec.Body.String(), "RepoNotFound")}
// TestComAtprotoSyncReposyncRoundTrip is the payoff: the phase 1-3 client// (verified head + prefix-bounded MST walk) driven over HTTP against our own// handlers. If this passes, a streamplace node is walkable by any peer running// the same code path — no full-CAR fallback.func TestComAtprotoSyncReposyncRoundTrip(t *testing.T) { ctx := context.Background() const host = "node.example.com" cli, e := newSyncTestNode(t, host, host) did := atproto.ServerRepo.RepoDid()
inRange := [][2]string{ {constants.PLACE_STREAM_LIVE_VIEWERCOUNT, "aaa"}, {constants.PLACE_STREAM_LIVE_VIEWERCOUNT, "bbb"}, {constants.PLACE_STREAM_MEDIA_ORIGIN, "babczxv1"}, {constants.PLACE_STREAM_MEDIA_ORIGIN, "babczxv2"}, } // Keys immediately outside ["place.stream.", "place.stream/") on both // sides, so a walker that ignored the range bound would be caught. outOfRange := [][2]string{ {"app.bsky.feed.post", "3abcxyz"}, {"place.streamx.thing", "1"}, {"xyz.example.thing", "1"}, } for _, rec := range append(append([][2]string{}, inRange...), outOfRange...) { commitTestRecord(t, cli, rec[0], rec[1]) }
want := map[string]string{} for _, r := range inRange { out, err := atproto.ServerRepoGetRecord(ctx, did, r[0], r[1]) require.NoError(t, err) require.NotNil(t, out.Cid) want[r[0]+"/"+r[1]] = *out.Cid }
ts := httptest.NewServer(e) defer ts.Close()
client := &xrpc.Client{Host: ts.URL} fetcher := &reposync.XRPCBlockFetcher{Client: client, DID: did}
// The node signs its commits with its server-repo key; a verifying peer // gets that key from the did:web document. Mock the directory with the // very same multibase the DID doc publishes. dir := identity.NewMockDirectory() dir.Insert(identity.Identity{ DID: syntax.DID(did), Handle: syntax.HandleInvalid, Keys: map[string]identity.VerificationMethod{ "atproto": {Type: "Multikey", PublicKeyMultibase: atproto.ServerPubMultibase}, }, })
head, err := reposync.FetchVerifiedHead(ctx, client, fetcher, &dir, did) require.NoError(t, err) wantCID, wantRev, err := atproto.ServerRepoLatestCommit(ctx) require.NoError(t, err) require.Equal(t, wantCID, head.CID) require.Equal(t, wantRev, head.Rev) require.Equal(t, did, head.Commit.DID)
// Signature verification is load-bearing, not decorative: point the // directory at somebody else's key and the head must be rejected. otherPriv, err := atcrypto.GeneratePrivateKeyK256() require.NoError(t, err) otherPub, err := otherPriv.PublicKey() require.NoError(t, err) wrongDir := identity.NewMockDirectory() wrongDir.Insert(identity.Identity{ DID: syntax.DID(did), Handle: syntax.HandleInvalid, Keys: map[string]identity.VerificationMethod{ "atproto": {Type: "Multikey", PublicKeyMultibase: otherPub.Multibase()}, }, }) _, err = reposync.FetchVerifiedHead(ctx, client, fetcher, &wrongDir, did) require.Error(t, err) require.Contains(t, err.Error(), "signature")
got := map[string]string{} walker := &reposync.Walker{Fetcher: fetcher} require.NoError(t, walker.WalkPrefix(ctx, head.Root, "place.stream.", func(path string, rcid cid.Cid, rec []byte) error { require.NoError(t, reposync.VerifyBlock(rcid, rec)) got[path] = rcid.String() return nil })) require.Equal(t, want, got)
// A walk that starts from a verified head must also see writes that // happen afterwards, once the head is re-fetched. commitTestRecord(t, cli, constants.PLACE_STREAM_MEDIA_ORIGIN, "babczxv3") head2, err := reposync.FetchVerifiedHead(ctx, client, fetcher, &dir, did) require.NoError(t, err) require.NotEqual(t, head.CID, head2.CID) require.Greater(t, head2.Rev, head.Rev)
got2 := map[string]string{} require.NoError(t, walker.WalkPrefix(ctx, head2.Root, "place.stream.", func(path string, rcid cid.Cid, rec []byte) error { got2[path] = rcid.String() return nil })) require.Len(t, got2, len(want)+1) require.Contains(t, got2, constants.PLACE_STREAM_MEDIA_ORIGIN+"/babczxv3")}