Merge branch 'main' into backfill

atproto/heap/cid.go

···
+package heap
+
+import (
+	"github.com/ipfs/go-cid"
+	"github.com/multiformats/go-multihash"
+)
+
+func computeCID(b []byte) (*cid.Cid, error) {
+	// TODO: not sure why this would ever fail; could we ignore or panic?
+	// TODO: is there a more performant way to call SHA256, then wrap?
+	builder := cid.NewPrefixV1(cid.Raw, multihash.SHA2_256)
+	c, err := builder.Sum(b)
+	if err != nil {
+		return nil, err
+	}
+	return &c, err
+}

atproto/heap/examples_test.go

···
+package heap
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/bluesky-social/indigo/atproto/repo"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+)
+
+func ExampleNetClient_GetRepoCAR() {
+
+	ctx := context.Background()
+	nc := NewNetClient()
+	did := syntax.DID("did:plc:ewvi7nxzyoun6zhxrhs64oiz")
+
+	stream, err := nc.GetRepoCAR(ctx, did)
+	if err != nil {
+		panic("failed to download CAR: " + err.Error())
+	}
+	defer stream.Close()
+
+	// NOTE: could also use LoadCommitFromCAR
+	commit, _, err := repo.LoadRepoFromCAR(ctx, stream)
+	if err != nil {
+		panic("failed to parse CAR: " + err.Error())
+	}
+
+	ident, _ := nc.Dir.LookupDID(ctx, did)
+	pub, _ := ident.PublicKey()
+
+	if err := commit.VerifySignature(pub); err != nil {
+		panic("failed to verify commit signature: " + err.Error())
+	}
+
+	fmt.Println(commit.DID)
+	// did:plc:ewvi7nxzyoun6zhxrhs64oiz
+}
+
+func ExampleNetClient_GetBlob() {
+
+	ctx := context.Background()
+	nc := NewNetClient()
+	did := syntax.DID("did:plc:ewvi7nxzyoun6zhxrhs64oiz")
+	cid := syntax.CID("bafkreieya7iitpu4okjtm7iexiwikj7t63ttlthad32ojsvjqhqbc3iwmi")
+
+	buf := bytes.Buffer{}
+	if err := nc.GetBlob(ctx, did, cid, &buf); err != nil {
+		panic("failed to download blob: " + err.Error())
+	}
+
+	fmt.Println(buf.Len())
+	// 518394
+}
+
+func ExampleNetClient_GetAccountStatus() {
+
+	ctx := context.Background()
+	nc := NewNetClient()
+	did := syntax.DID("did:plc:ewvi7nxzyoun6zhxrhs64oiz")
+
+	active, status, err := nc.GetAccountStatus(ctx, did)
+	if err != nil {
+		panic("failed to check account status: " + err.Error())
+	}
+
+	fmt.Printf("active=%t status=%s\n", active, status)
+	// active=true status=
+}
+
+func ExampleNetClient_GetRecordUnverified() {
+
+	ctx := context.Background()
+	nc := NewNetClient()
+	did := syntax.DID("did:plc:ewvi7nxzyoun6zhxrhs64oiz")
+	collection := syntax.NSID("app.bsky.actor.profile")
+	rkey := syntax.RecordKey("self")
+
+	raw, _, err := nc.GetRecordUnverified(ctx, did, collection, rkey)
+	if err != nil {
+		panic("failed to fetch record: " + err.Error())
+	}
+	var record map[string]any
+	_ = json.Unmarshal(*raw, &record)
+
+	fmt.Println(record["displayName"])
+	// AT Protocol Developers
+}
+
+func ExampleNetClient_GetRecord() {
+
+	ctx := context.Background()
+	nc := NewNetClient()
+	did := syntax.DID("did:plc:ewvi7nxzyoun6zhxrhs64oiz")
+	collection := syntax.NSID("app.bsky.actor.profile")
+	rkey := syntax.RecordKey("self")
+
+	var record map[string]any
+	_, err := nc.GetRecord(ctx, did, collection, rkey, &record)
+	if err != nil {
+		panic("failed to fetch record: " + err.Error())
+	}
+
+	fmt.Println(record["displayName"])
+	// Output: AT Protocol Developers
+}

atproto/heap/netclient.go

···
+package heap
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+
+	"github.com/bluesky-social/indigo/atproto/identity"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+)
+
+type NetClient struct {
+	Client *http.Client
+	// NOTE: maybe should use a "resolver" which doesn't do handle resolution? or leave that to calling code to configure
+	Dir       identity.Directory
+	UserAgent string
+}
+
+func NewNetClient() *NetClient {
+	return &NetClient{
+		// TODO: maybe custom client: SSRF, retries, timeout
+		Client:    http.DefaultClient,
+		Dir:       identity.DefaultDirectory(),
+		UserAgent: "cobalt-netclient",
+	}
+}
+
+// Fetches repo export (CAR file). Calling code is responsible for closing the returned [io.ReadCloser] on success (often an HTTP response body). Does not verify signatures or CAR format or structure in any way.
+func (nc *NetClient) GetRepoCAR(ctx context.Context, did syntax.DID) (io.ReadCloser, error) {
+	ident, err := nc.Dir.LookupDID(ctx, did)
+	if err != nil {
+		return nil, err
+	}
+	host := ident.PDSEndpoint()
+	if host == "" {
+		return nil, fmt.Errorf("account has no PDS host registered: %s", did.String())
+	}
+	// TODO: validate host
+	// TODO: DID escaping (?)
+	u := fmt.Sprintf("%s/xrpc/com.atproto.sync.getRepo?did=%s", host, did)
+
+	slog.Debug("downloading repo CAR", "did", did, "url", u)
+	req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
+	if err != nil {
+		return nil, err
+	}
+	if nc.UserAgent != "" {
+		req.Header.Set("User-Agent", nc.UserAgent)
+	}
+	req.Header.Set("Accept", "application/vnd.ipld.car")
+
+	resp, err := nc.Client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("fetching repo CAR file (%s): %w", did, err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		resp.Body.Close()
+		return nil, fmt.Errorf("HTTP error fetching repo CAR file (%s): %d", did, resp.StatusCode)
+	}
+
+	return resp.Body, nil
+}
+
+// Resolves and fetches blob from the network. Calling code must close the returned [io.ReadCloser] (eg, HTTP response body). Does not verify CID.
+func (nc *NetClient) GetBlobReader(ctx context.Context, did syntax.DID, cid syntax.CID) (io.ReadCloser, error) {
+	ident, err := nc.Dir.LookupDID(ctx, did)
+	if err != nil {
+		return nil, err
+	}
+	host := ident.PDSEndpoint()
+	if host == "" {
+		return nil, fmt.Errorf("account has no PDS host registered: %s", did.String())
+	}
+	// TODO: validate host
+	// TODO: DID escaping (?)
+	u := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s", host, did, cid)
+
+	slog.Debug("downloading blob", "did", did, "cid", cid, "url", u)
+	req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
+	if err != nil {
+		return nil, err
+	}
+	if nc.UserAgent != "" {
+		req.Header.Set("User-Agent", nc.UserAgent)
+	}
+	req.Header.Set("Accept", "*/*")
+
+	resp, err := nc.Client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("fetching blob (%s, %s): %w", did, cid, err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		resp.Body.Close()
+		return nil, fmt.Errorf("HTTP error fetching blob (%s, %s): %d", did, cid, resp.StatusCode)
+	}
+
+	return resp.Body, nil
+}
+
+var ErrMismatchedBlobCID = errors.New("mismatched blob CID")
+
+// Fetches blob, writes in to provided buffer, and verified CID hash.
+func (nc *NetClient) GetBlob(ctx context.Context, did syntax.DID, cid syntax.CID, buf *bytes.Buffer) error {
+	stream, err := nc.GetBlobReader(ctx, did, cid)
+	if err != nil {
+		return err
+	}
+	defer stream.Close()
+
+	if _, err := io.Copy(buf, stream); err != nil {
+		return err
+	}
+
+	c, err := computeCID(buf.Bytes())
+	if err != nil {
+		return err
+	}
+
+	if c.String() != cid.String() {
+		return ErrMismatchedBlobCID
+	}
+	return nil
+}
+
+type repoStatusResp struct {
+	Active bool   `json:"active"`
+	DID    string `json:"did"`
+	Status string `json:"status,omitempty"`
+}
+
+// Fetches account status. Returns a boolean indicating active state, and a string describing any non-active status.
+func (nc *NetClient) GetAccountStatus(ctx context.Context, did syntax.DID) (active bool, status string, err error) {
+	ident, err := nc.Dir.LookupDID(ctx, did)
+	if err != nil {
+		return false, "", err
+	}
+	host := ident.PDSEndpoint()
+	if host == "" {
+		return false, "", fmt.Errorf("account has no PDS host registered: %s", did.String())
+	}
+	// TODO: validate host
+	// TODO: DID escaping (?)
+	u := fmt.Sprintf("%s/xrpc/com.atproto.sync.getRepoStatus?did=%s", host, did)
+
+	slog.Debug("fetching account status", "did", did, "url", u)
+	req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
+	if err != nil {
+		return false, "", err
+	}
+	if nc.UserAgent != "" {
+		req.Header.Set("User-Agent", nc.UserAgent)
+	}
+	req.Header.Set("Accept", "application/json")
+
+	resp, err := nc.Client.Do(req)
+	if err != nil {
+		return false, "", fmt.Errorf("fetching account status (%s): %w", did, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return false, "", fmt.Errorf("HTTP error fetching account status (%s): %d", did, resp.StatusCode)
+	}
+
+	var rsr repoStatusResp
+	if err := json.NewDecoder(resp.Body).Decode(&rsr); err != nil {
+		return false, "", fmt.Errorf("failed decoding account status response: %w", err)
+	}
+
+	return rsr.Active, rsr.Status, nil
+}

atproto/heap/record.go

···
+package heap
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net/http"
+
+	"github.com/bluesky-social/indigo/atproto/data"
+	"github.com/bluesky-social/indigo/atproto/repo"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+)
+
+type repoRecordResp struct {
+	URI   string          `json:"uri"`
+	CID   syntax.CID      `json:"cid"`
+	Value json.RawMessage `json:"value"`
+}
+
+// Fetches record JSON using com.atproto.repo.getRecord, and returns record as [json.RawMessage] and the CID (as string).
+func (nc *NetClient) GetRecordUnverified(ctx context.Context, did syntax.DID, collection syntax.NSID, rkey syntax.RecordKey) (*json.RawMessage, syntax.CID, error) {
+	ident, err := nc.Dir.LookupDID(ctx, did)
+	if err != nil {
+		return nil, "", err
+	}
+	host := ident.PDSEndpoint()
+	if host == "" {
+		return nil, "", fmt.Errorf("account has no PDS host registered: %s", did.String())
+	}
+	// TODO: validate host
+	// TODO: DID escaping (?)
+	u := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s", host, did, collection, rkey)
+
+	slog.Debug("fetching record JSON", "did", did, "url", u)
+	req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
+	if err != nil {
+		return nil, "", err
+	}
+	if nc.UserAgent != "" {
+		req.Header.Set("User-Agent", nc.UserAgent)
+	}
+	req.Header.Set("Accept", "application/json")
+
+	resp, err := nc.Client.Do(req)
+	if err != nil {
+		return nil, "", fmt.Errorf("fetching record JSON (%s): %w", did, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, "", fmt.Errorf("HTTP error fetching record JSON (%s): %d", did, resp.StatusCode)
+	}
+
+	var rrr repoRecordResp
+	if err := json.NewDecoder(resp.Body).Decode(&rrr); err != nil {
+		return nil, "", fmt.Errorf("failed decoding account status response: %w", err)
+	}
+
+	return &rrr.Value, rrr.CID, nil
+}
+
+// Fetches a record "proof" using com.atproto.sync.getRecord. Verifies signature and merkel chain. Copies record content in out 'out' parameter.
+//
+// If out is nil, record data is not returned. If it is [bytes.Buffer], the record CBOR is copied in. Otherwise, the record is transformed to JSON and Unmarshalled in to provided output, which could be a pointer to a struct, [json.RawMessage], `map[string]any`, etc.
+//
+// TODO: this might not be fully validating MST tree and record CID hashes or encoding yet
+func (nc *NetClient) GetRecord(ctx context.Context, did syntax.DID, collection syntax.NSID, rkey syntax.RecordKey, out any) (syntax.CID, error) {
+	// TODO: "GetRecordProof" variant, which just returns CAR as io.ReadCloser?
+	ident, err := nc.Dir.LookupDID(ctx, did)
+	if err != nil {
+		return "", err
+	}
+	pub, err := ident.PublicKey()
+	if err != nil {
+		return "", err
+	}
+	host := ident.PDSEndpoint()
+	if host == "" {
+		return "", fmt.Errorf("account has no PDS host registered: %s", did.String())
+	}
+	// TODO: validate host
+	// TODO: DID escaping (?)
+	u := fmt.Sprintf("%s/xrpc/com.atproto.sync.getRecord?did=%s&collection=%s&rkey=%s", host, did, collection, rkey)
+
+	slog.Debug("fetching record proof", "did", did, "url", u)
+	req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
+	if err != nil {
+		return "", err
+	}
+	if nc.UserAgent != "" {
+		req.Header.Set("User-Agent", nc.UserAgent)
+	}
+	req.Header.Set("Accept", "application/vnd.ipld.car")
+
+	resp, err := nc.Client.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("fetching record proof (%s): %w", did, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("HTTP error fetching record proof (%s): %d", did, resp.StatusCode)
+	}
+
+	// TODO: re-confirm if loading tree re-checks all CIDs; or if we need to re-compute the tree data CID
+	commit, rp, err := repo.LoadRepoFromCAR(ctx, resp.Body)
+	if err != nil {
+		return "", fmt.Errorf("failed to parse record proof CAR (%s): %w", did, err)
+	}
+
+	// NOTE: LoadRepoFromCAR calls commit.VerifyStructure() internally
+
+	if err := commit.VerifySignature(pub); err != nil {
+		return "", fmt.Errorf("failed to verify record proof signature (%s): %w", did, err)
+	}
+
+	rbytes, rcid, err := rp.GetRecordBytes(ctx, collection, rkey)
+	if err != nil {
+		return "", fmt.Errorf("failed to read record from proof CAR (%s): %w", did, err)
+	}
+	cidStr := syntax.CID(rcid.String())
+
+	// TODO: `GetRecordBytes` does not currently verify record CID, but unpacking CAR file should have done that? but need to confirm CAR implementation does this
+
+	// check that record CBOR is valid, even if we don't return it
+	rdata, err := data.UnmarshalCBOR(rbytes)
+	if err != nil {
+		return "", fmt.Errorf("failed to parse record CBOR (%s): %w", did, err)
+	}
+
+	switch out := out.(type) {
+	case nil:
+		// if output isn't captured, bail out early
+		return cidStr, nil
+	case *bytes.Buffer:
+		// simply copy data over
+		out.Reset()
+		_, err := out.Write(rbytes)
+		if err != nil {
+			return "", err
+		}
+		return cidStr, nil
+	default:
+		// attempt to unmarshal from json
+		jsonBytes, err := json.Marshal(rdata)
+		if err != nil {
+			return "", err
+		}
+		if err := json.Unmarshal(jsonBytes, out); err != nil {
+			return "", fmt.Errorf("failed unmarhsaling record (%s): %w", did, err)
+		}
+		return cidStr, nil
+	}
+}