atproto/lexicon: package for working with lexicon schemas, and runtime data validation

atproto/data/parse.go

@@ -111,6 +111,18 @@ func parseMap(obj map[string]any) (any, error) {
 			return nil, fmt.Errorf("$type field must contain a non-empty string")
 		}
 	}
+	// legacy blob type
+	if len(obj) == 2 {
+		if _, ok := obj["mimeType"]; ok {
+			if _, ok := obj["cid"]; ok {
+				b, err := parseLegacyBlob(obj)
+				if err != nil {
+					return nil, err
+				}
+				return *b, nil
+			}
+		}
+	}
 	out := make(map[string]any, len(obj))
 	for k, val := range obj {
 		if len(k) > MAX_OBJECT_KEY_LEN {
@@ -213,6 +225,30 @@ func parseBlob(obj map[string]any) (*Blob, error) {
 	}, nil
 }
 
+func parseLegacyBlob(obj map[string]any) (*Blob, error) {
+	if len(obj) != 2 {
+		return nil, fmt.Errorf("legacy blobs expected to have 2 fields")
+	}
+	var err error
+	mimeType, ok := obj["mimeType"].(string)
+	if !ok {
+		return nil, fmt.Errorf("blob 'mimeType' missing or not a string")
+	}
+	cidStr, ok := obj["cid"]
+	if !ok {
+		return nil, fmt.Errorf("blob 'cid' missing")
+	}
+	c, err := cid.Parse(cidStr)
+	if err != nil {
+		return nil, fmt.Errorf("invalid CID: %w", err)
+	}
+	return &Blob{
+		Size:     -1,
+		MimeType: mimeType,
+		Ref:      CIDLink(c),
+	}, nil
+}
+
 func parseObject(obj map[string]any) (map[string]any, error) {
 	out, err := parseMap(obj)
 	if err != nil {

atproto/identity/doc.go

@@ -1,8 +1,6 @@
 /*
 Package identity provides types and routines for resolving handles and DIDs from the network
 
-The two main abstractions are a Catalog interface for identity service implementations, and an Identity structure which represents core identity information relevant to atproto. The Catalog interface can be nested, somewhat like HTTP middleware, to provide caching, observability, or other bespoke needs in more complex systems.
-
-Much of the implementation of this SDK is based on existing code in indigo:api/extra.go
+The two main abstractions are a Directory interface for identity service implementations, and an Identity struct which represents core identity information relevant to atproto. The Directory interface can be nested, somewhat like HTTP middleware, to provide caching, observability, or other bespoke needs in more complex systems.
 */
 package identity

atproto/lexicon/catalog.go

@@ -0,0 +1,112 @@
+package lexicon
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+// An aggregation of lexicon schemas, and methods for validating generic data against those schemas.
+type Catalog interface {
+	Resolve(ref string) (*Schema, error)
+}
+
+type BaseCatalog struct {
+	schemas map[string]Schema
+}
+
+func NewBaseCatalog() BaseCatalog {
+	return BaseCatalog{
+		schemas: make(map[string]Schema),
+	}
+}
+
+func (c *BaseCatalog) Resolve(ref string) (*Schema, error) {
+	if ref == "" {
+		return nil, fmt.Errorf("tried to resolve empty string name")
+	}
+	// default to #main if name doesn't have a fragment
+	if !strings.Contains(ref, "#") {
+		ref = ref + "#main"
+	}
+	s, ok := c.schemas[ref]
+	if !ok {
+		return nil, fmt.Errorf("schema not found in catalog: %s", ref)
+	}
+	return &s, nil
+}
+
+func (c *BaseCatalog) AddSchemaFile(sf SchemaFile) error {
+	base := sf.ID
+	for frag, def := range sf.Defs {
+		if len(frag) == 0 || strings.Contains(frag, "#") || strings.Contains(frag, ".") {
+			// TODO: more validation here?
+			return fmt.Errorf("schema name invalid: %s", frag)
+		}
+		name := base + "#" + frag
+		if _, ok := c.schemas[name]; ok {
+			return fmt.Errorf("catalog already contained a schema with name: %s", name)
+		}
+		// "A file can have at most one definition with one of the "primary" types. Primary types should always have the name main. It is possible for main to describe a non-primary type."
+		switch s := def.Inner.(type) {
+		case SchemaRecord, SchemaQuery, SchemaProcedure, SchemaSubscription:
+			if frag != "main" {
+				return fmt.Errorf("record, query, procedure, and subscription types must be 'main', not: %s", frag)
+			}
+		case SchemaToken:
+			// add fully-qualified name to token
+			s.fullName = name
+			def.Inner = s
+		}
+		def.SetBase(base)
+		if err := def.CheckSchema(); err != nil {
+			return err
+		}
+		s := Schema{
+			ID:       name,
+			Revision: sf.Revision,
+			Def:      def.Inner,
+		}
+		c.schemas[name] = s
+	}
+	return nil
+}
+
+func (c *BaseCatalog) LoadDirectory(dirPath string) error {
+	return filepath.WalkDir(dirPath, func(p string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if d.IsDir() {
+			return nil
+		}
+		if !strings.HasSuffix(p, ".json") {
+			return nil
+		}
+		// TODO: logging
+		fmt.Println(p)
+		f, err := os.Open(p)
+		if err != nil {
+			return err
+		}
+		defer func() { _ = f.Close() }()
+
+		b, err := io.ReadAll(f)
+		if err != nil {
+			return err
+		}
+
+		var sf SchemaFile
+		if err = json.Unmarshal(b, &sf); err != nil {
+			return err
+		}
+		if err = c.AddSchemaFile(sf); err != nil {
+			return err
+		}
+		return nil
+	})
+}

atproto/lexicon/cmd/lextool/main.go

@@ -0,0 +1,90 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+
+	"github.com/bluesky-social/indigo/atproto/lexicon"
+
+	"github.com/urfave/cli/v2"
+)
+
+func main() {
+	app := cli.App{
+		Name:  "lex-tool",
+		Usage: "informal debugging CLI tool for atproto lexicons",
+	}
+	app.Commands = []*cli.Command{
+		&cli.Command{
+			Name:   "parse-schema",
+			Usage:  "parse an individual lexicon schema file (JSON)",
+			Action: runParseSchema,
+		},
+		&cli.Command{
+			Name:   "load-directory",
+			Usage:  "try recursively loading all the schemas from a directory",
+			Action: runLoadDirectory,
+		},
+		&cli.Command{
+			Name:   "validate-record",
+			Usage:  "fetch from network, validate against catalog",
+			Action: runValidateRecord,
+		},
+		&cli.Command{
+			Name:   "validate-firehose",
+			Usage:  "subscribe to a firehose, validate every known record against catalog",
+			Action: runValidateFirehose,
+		},
+	}
+	h := slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})
+	slog.SetDefault(slog.New(h))
+	app.RunAndExitOnError()
+}
+
+func runParseSchema(cctx *cli.Context) error {
+	p := cctx.Args().First()
+	if p == "" {
+		return fmt.Errorf("need to provide path to a schema file as an argument")
+	}
+
+	f, err := os.Open(p)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = f.Close() }()
+
+	b, err := io.ReadAll(f)
+	if err != nil {
+		return err
+	}
+
+	var sf lexicon.SchemaFile
+	if err := json.Unmarshal(b, &sf); err != nil {
+		return err
+	}
+	out, err := json.MarshalIndent(sf, "", "  ")
+	if err != nil {
+		return err
+	}
+	fmt.Println(string(out))
+	return nil
+}
+
+func runLoadDirectory(cctx *cli.Context) error {
+	p := cctx.Args().First()
+	if p == "" {
+		return fmt.Errorf("need to provide directory path as an argument")
+	}
+
+	c := lexicon.NewBaseCatalog()
+	err := c.LoadDirectory(p)
+	if err != nil {
+		return err
+	}
+
+	fmt.Println("success!")
+	return nil
+}

atproto/lexicon/cmd/lextool/net.go

@@ -0,0 +1,95 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+
+	"github.com/bluesky-social/indigo/atproto/data"
+	"github.com/bluesky-social/indigo/atproto/identity"
+	"github.com/bluesky-social/indigo/atproto/lexicon"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+
+	"github.com/urfave/cli/v2"
+)
+
+func runValidateRecord(cctx *cli.Context) error {
+	ctx := context.Background()
+	args := cctx.Args().Slice()
+	if len(args) != 2 {
+		return fmt.Errorf("expected two args (catalog path and AT-URI)")
+	}
+	p := args[0]
+	if p == "" {
+		return fmt.Errorf("need to provide directory path as an argument")
+	}
+
+	cat := lexicon.NewBaseCatalog()
+	err := cat.LoadDirectory(p)
+	if err != nil {
+		return err
+	}
+
+	aturi, err := syntax.ParseATURI(args[1])
+	if err != nil {
+		return err
+	}
+	if aturi.RecordKey() == "" {
+		return fmt.Errorf("need a full, not partial, AT-URI: %s", aturi)
+	}
+	dir := identity.DefaultDirectory()
+	ident, err := dir.Lookup(ctx, aturi.Authority())
+	if err != nil {
+		return fmt.Errorf("resolving AT-URI authority: %v", err)
+	}
+	pdsURL := ident.PDSEndpoint()
+	if pdsURL == "" {
+		return fmt.Errorf("could not resolve PDS endpoint for AT-URI account: %s", ident.DID.String())
+	}
+
+	slog.Info("fetching record", "did", ident.DID.String(), "collection", aturi.Collection().String(), "rkey", aturi.RecordKey().String())
+	url := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
+		pdsURL, ident.DID, aturi.Collection(), aturi.RecordKey())
+	resp, err := http.Get(url)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("fetch failed")
+	}
+	respBytes, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+
+	body, err := data.UnmarshalJSON(respBytes)
+	record, ok := body["value"].(map[string]any)
+	if !ok {
+		return fmt.Errorf("fetched record was not an object")
+	}
+
+	slog.Info("validating", "did", ident.DID.String(), "collection", aturi.Collection().String(), "rkey", aturi.RecordKey().String())
+	err = lexicon.ValidateRecord(&cat, record, aturi.Collection().String(), lexicon.LenientMode)
+	if err != nil {
+		return err
+	}
+	fmt.Println("success!")
+	return nil
+}
+
+func runValidateFirehose(cctx *cli.Context) error {
+	p := cctx.Args().First()
+	if p == "" {
+		return fmt.Errorf("need to provide directory path as an argument")
+	}
+
+	cat := lexicon.NewBaseCatalog()
+	err := cat.LoadDirectory(p)
+	if err != nil {
+		return err
+	}
+
+	return fmt.Errorf("UNIMPLEMENTED")
+}

atproto/lexicon/docs.go

@@ -0,0 +1,4 @@
+/*
+Package atproto/lexicon provides generic Lexicon schema parsing and run-time validation.
+*/
+package lexicon

atproto/lexicon/extract.go

@@ -0,0 +1,20 @@
+package lexicon
+
+import (
+	"encoding/json"
+)
+
+// Helper type for extracting record type from JSON
+type genericSchemaDef struct {
+	Type string `json:"type"`
+}
+
+// Parses the top-level $type field from generic atproto JSON data
+func ExtractTypeJSON(b []byte) (string, error) {
+	var gsd genericSchemaDef
+	if err := json.Unmarshal(b, &gsd); err != nil {
+		return "", err
+	}
+
+	return gsd.Type, nil
+}