Low-latency AT Protocol interaction indexer.
firehose: verify commit sigs against repo signing keys
Aly Raffauf
(Jul 4, 2026, 11:25 AM EDT)
c859f5b8
37dfb20b
+49
-1
+1
README.md
+1
README.md
+1
cmd/asterism/main.go
+1
cmd/asterism/main.go
+45
-1
internal/firehose/commit.go
+45
-1
internal/firehose/commit.go
···
5
5
"context"
6
6
"fmt"
7
7
"strings"
8
+
"time"
8
9
9
10
"github.com/alyraffauf/asterism/internal/index"
10
11
"github.com/bluesky-social/indigo/api/atproto"
12
+
"github.com/bluesky-social/indigo/atproto/syntax"
11
13
indigorepo "github.com/bluesky-social/indigo/repo"
12
14
"github.com/ipfs/go-cid"
13
15
)
14
16
15
17
func (c *Consumer) HandleCommit(ctx context.Context, event *atproto.SyncSubscribeRepos_Commit) error {
16
18
// fmt.Println("repo:", event.Repo, "commit:", event.Rev)
17
-
//
19
+
18
20
if err := c.Store.SaveCursor(ctx, event.Seq); err != nil {
19
21
fmt.Println("could not save cursor:", err)
20
22
}
···
33
35
return err
34
36
}
35
37
38
+
if !c.hasWantedOps(event.Ops) {
39
+
return nil
40
+
}
41
+
42
+
if err := c.verifyCommit(ctx, repo); err != nil {
43
+
fmt.Println("commit verification failed:", event.Repo, err)
44
+
return nil
45
+
}
46
+
36
47
for _, operation := range event.Ops {
37
48
if err := c.handleOperation(ctx, event, repo, operation); err != nil {
38
49
fmt.Println("could not handle operation:", err)
···
73
84
74
85
return index.Record(ctx, c.Store, event.Repo, collection, recordKey, recordCid.String(), event.Rev, *recordBytes)
75
86
}
87
+
88
+
func (c *Consumer) hasWantedOps(ops []*atproto.SyncSubscribeRepos_RepoOp) bool {
89
+
for _, op := range ops {
90
+
collection, _, ok := strings.Cut(op.Path, "/")
91
+
if ok && c.wants(collection) {
92
+
return true
93
+
}
94
+
}
95
+
return false
96
+
}
97
+
98
+
func (c *Consumer) verifyCommit(ctx context.Context, repo *indigorepo.Repo) error {
99
+
sc := repo.SignedCommit()
100
+
101
+
resolveCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
102
+
identity, err := c.Directory.LookupDID(resolveCtx, syntax.DID(sc.Did))
103
+
cancel()
104
+
if err != nil {
105
+
return fmt.Errorf("resolve did: %w", err)
106
+
}
107
+
108
+
pubKey, err := identity.PublicKey()
109
+
if err != nil {
110
+
return fmt.Errorf("get public key: %w", err)
111
+
}
112
+
113
+
unsignedBytes, err := sc.Unsigned().BytesForSigning()
114
+
if err != nil {
115
+
return fmt.Errorf("marshal unsigned commit: %w", err)
116
+
}
117
+
118
+
return pubKey.HashAndVerify(unsignedBytes, sc.Sig)
119
+
}
+2
internal/firehose/consumer.go
+2
internal/firehose/consumer.go
···
3
3
import (
4
4
"github.com/alyraffauf/asterism/internal/backfill"
5
5
"github.com/alyraffauf/asterism/internal/store"
6
+
"github.com/bluesky-social/indigo/atproto/identity"
6
7
)
7
8
8
9
type Consumer struct {
9
10
WantedCollections map[string]struct{}
10
11
Store *store.Store
12
+
Directory identity.Directory
11
13
Backfill *backfill.Backfill
12
14
}
13
15