this repo has no description
goat: verify sync v1.1 firehose (#988)
+260
-12
+25
atproto/label/label.go
+25
atproto/label/label.go
···
25
25
Sig data.Bytes `json:"sig,omitempty" cborgen:"sig,omitempty"`
26
26
}
27
27
28
+
// converts to map[string]any for printing as JSON
29
+
func (l *Label) Data() map[string]any {
30
+
d := map[string]any{
31
+
"cid": l.CID,
32
+
"cts": l.CreatedAt,
33
+
"src": l.SourceDID,
34
+
"uri": l.URI,
35
+
"val": l.Val,
36
+
"ver": l.Version,
37
+
}
38
+
if l.CID != nil {
39
+
d["cid"] = l.CID
40
+
}
41
+
if l.ExpiresAt != nil {
42
+
d["exp"] = l.ExpiresAt
43
+
}
44
+
if l.Negated != nil {
45
+
d["neg"] = l.Negated
46
+
}
47
+
if l.Sig != nil {
48
+
d["sig"] = data.Bytes(l.Sig)
49
+
}
50
+
return d
51
+
}
52
+
28
53
// does basic checks on syntax and structure
29
54
func (l *Label) VerifySyntax() error {
30
55
if l.Version != ATPROTO_LABEL_VERSION {
+18
atproto/repo/commit.go
+18
atproto/repo/commit.go
···
5
5
"fmt"
6
6
7
7
"github.com/bluesky-social/indigo/atproto/crypto"
8
+
"github.com/bluesky-social/indigo/atproto/data"
8
9
"github.com/bluesky-social/indigo/atproto/syntax"
9
10
10
11
"github.com/ipfs/go-cid"
···
37
38
return fmt.Errorf("invalid commit data: %w", err)
38
39
}
39
40
return nil
41
+
}
42
+
43
+
// returns a representation of the commit object as atproto data (eg, for JSON serialization)
44
+
func (c *Commit) AsData() map[string]any {
45
+
d := map[string]any{
46
+
"did": c.DID,
47
+
"version": c.Version,
48
+
"prev": (*data.CIDLink)(c.Prev),
49
+
"data": data.CIDLink(c.Data),
50
+
}
51
+
if c.Sig != nil {
52
+
d["sig"] = data.Bytes(c.Sig)
53
+
}
54
+
if c.Rev != "" {
55
+
d["rev"] = c.Rev
56
+
}
57
+
return d
40
58
}
41
59
42
60
// Encodes the commit object as DAG-CBOR, without the signature field. Used for signing or validating signatures.
-1
atproto/repo/sync.go
-1
atproto/repo/sync.go
+185
-11
cmd/goat/firehose.go
+185
-11
cmd/goat/firehose.go
···
11
11
"net/url"
12
12
"os"
13
13
"strings"
14
+
"time"
14
15
15
16
comatproto "github.com/bluesky-social/indigo/api/atproto"
16
17
"github.com/bluesky-social/indigo/atproto/data"
18
+
"github.com/bluesky-social/indigo/atproto/identity"
17
19
"github.com/bluesky-social/indigo/atproto/repo"
18
20
"github.com/bluesky-social/indigo/atproto/syntax"
19
21
"github.com/bluesky-social/indigo/events"
···
49
51
Usage: "only print account and identity events",
50
52
},
51
53
&cli.BoolFlag{
54
+
Name: "quiet",
55
+
Aliases: []string{"q"},
56
+
Usage: "don't actually print events to stdout (eg, errors only)",
57
+
},
58
+
&cli.BoolFlag{
59
+
Name: "verify-basic",
60
+
Usage: "parse events and do basic syntax and structure checks",
61
+
},
62
+
&cli.BoolFlag{
63
+
Name: "verify-sig",
64
+
Usage: "verify account signatures on commits",
65
+
},
66
+
&cli.BoolFlag{
67
+
Name: "verify-mst",
68
+
Usage: "run inductive verification of ops and MST structure",
69
+
},
70
+
&cli.BoolFlag{
52
71
Name: "ops",
53
72
Aliases: []string{"records"},
54
73
Usage: "instead of printing entire events, print individual record ops",
···
58
77
}
59
78
60
79
type GoatFirehoseConsumer struct {
61
-
// for pretty-printing events to stdout
62
-
EventLogger *slog.Logger
63
80
OpsMode bool
64
81
AccountsOnly bool
82
+
Quiet bool
83
+
VerifyBasic bool
84
+
VerifySig bool
85
+
VerifyMST bool
65
86
// filter to specified collections
66
87
CollectionFilter []string
88
+
// for signature verification
89
+
Dir identity.Directory
67
90
}
68
91
69
92
func runFirehose(cctx *cli.Context) error {
70
93
ctx := context.Background()
71
94
72
-
slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, nil)))
95
+
slog.SetDefault(configLogger(cctx, os.Stderr))
96
+
97
+
// main thing is skipping handle verification
98
+
bdir := identity.BaseDirectory{
99
+
SkipHandleVerification: true,
100
+
TryAuthoritativeDNS: false,
101
+
SkipDNSDomainSuffixes: []string{".bsky.social"},
102
+
UserAgent: "goat/" + versioninfo.Short(),
103
+
}
104
+
cdir := identity.NewCacheDirectory(&bdir, 1_000_000, time.Hour*24, time.Minute*2, time.Minute*5)
73
105
74
106
gfc := GoatFirehoseConsumer{
75
-
EventLogger: slog.New(slog.NewJSONHandler(os.Stdout, nil)),
76
107
OpsMode: cctx.Bool("ops"),
77
108
AccountsOnly: cctx.Bool("account-events"),
78
109
CollectionFilter: cctx.StringSlice("collection"),
110
+
Quiet: cctx.Bool("quiet"),
111
+
VerifyBasic: cctx.Bool("verify-basic"),
112
+
VerifySig: cctx.Bool("verify-sig"),
113
+
VerifyMST: cctx.Bool("verify-mst"),
114
+
Dir: &cdir,
79
115
}
80
116
81
117
var relayHost string
···
104
140
u.RawQuery = fmt.Sprintf("cursor=%d", cctx.Int("cursor"))
105
141
}
106
142
urlString := u.String()
107
-
slog.Debug("GET", "url", urlString)
108
143
con, _, err := dialer.Dial(urlString, http.Header{
109
144
"User-Agent": []string{fmt.Sprintf("goat/%s", versioninfo.Short())},
110
145
})
···
114
149
115
150
rsc := &events.RepoStreamCallbacks{
116
151
RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error {
117
-
slog.Debug("commit event", "did", evt.Repo, "seq", evt.Seq)
152
+
//slog.Debug("commit event", "did", evt.Repo, "seq", evt.Seq)
118
153
if !gfc.AccountsOnly && !gfc.OpsMode {
119
154
return gfc.handleCommitEvent(ctx, evt)
120
155
} else if !gfc.AccountsOnly && gfc.OpsMode {
···
123
158
return nil
124
159
},
125
160
RepoSync: func(evt *comatproto.SyncSubscribeRepos_Sync) error {
126
-
slog.Debug("sync event", "did", evt.Did, "seq", evt.Seq)
161
+
//slog.Debug("sync event", "did", evt.Did, "seq", evt.Seq)
127
162
if !gfc.AccountsOnly && !gfc.OpsMode {
128
163
return gfc.handleSyncEvent(ctx, evt)
129
164
}
130
165
return nil
131
166
},
132
167
RepoIdentity: func(evt *comatproto.SyncSubscribeRepos_Identity) error {
133
-
slog.Debug("identity event", "did", evt.Did, "seq", evt.Seq)
168
+
//slog.Debug("identity event", "did", evt.Did, "seq", evt.Seq)
134
169
if !gfc.OpsMode {
135
170
return gfc.handleIdentityEvent(ctx, evt)
136
171
}
137
172
return nil
138
173
},
139
174
RepoAccount: func(evt *comatproto.SyncSubscribeRepos_Account) error {
140
-
slog.Debug("account event", "did", evt.Did, "seq", evt.Seq)
175
+
//slog.Debug("account event", "did", evt.Did, "seq", evt.Seq)
141
176
if !gfc.OpsMode {
142
177
return gfc.handleAccountEvent(ctx, evt)
143
178
}
144
179
return nil
145
180
},
181
+
RepoHandle: func(evt *comatproto.SyncSubscribeRepos_Handle) error {
182
+
if gfc.VerifyBasic {
183
+
slog.Info("deprecated event type", "eventType", "handle", "did", evt.Did, "seq", evt.Seq)
184
+
}
185
+
return nil
186
+
},
187
+
RepoMigrate: func(evt *comatproto.SyncSubscribeRepos_Migrate) error {
188
+
if gfc.VerifyBasic {
189
+
slog.Info("deprecated event type", "eventType", "migrate", "did", evt.Did, "seq", evt.Seq)
190
+
}
191
+
return nil
192
+
},
193
+
RepoTombstone: func(evt *comatproto.SyncSubscribeRepos_Tombstone) error {
194
+
if gfc.VerifyBasic {
195
+
slog.Info("deprecated event type", "eventType", "handle", "did", evt.Did, "seq", evt.Seq)
196
+
}
197
+
return nil
198
+
},
146
199
}
147
200
148
201
scheduler := parallel.NewScheduler(
···
156
209
}
157
210
158
211
func (gfc *GoatFirehoseConsumer) handleIdentityEvent(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Identity) error {
212
+
if gfc.VerifySig {
213
+
did, err := syntax.ParseDID(evt.Did)
214
+
if err != nil {
215
+
return err
216
+
}
217
+
gfc.Dir.Purge(ctx, did.AtIdentifier())
218
+
}
219
+
if gfc.VerifyBasic {
220
+
if _, err := syntax.ParseDID(evt.Did); err != nil {
221
+
slog.Warn("invalid DID", "eventType", "identity", "did", evt.Did, "seq", evt.Seq)
222
+
}
223
+
}
224
+
if gfc.Quiet {
225
+
return nil
226
+
}
159
227
out := make(map[string]interface{})
160
228
out["type"] = "identity"
161
229
out["payload"] = evt
···
168
236
}
169
237
170
238
func (gfc *GoatFirehoseConsumer) handleAccountEvent(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Account) error {
239
+
if gfc.VerifyBasic {
240
+
if _, err := syntax.ParseDID(evt.Did); err != nil {
241
+
slog.Warn("invalid DID", "eventType", "account", "did", evt.Did, "seq", evt.Seq)
242
+
}
243
+
}
244
+
if gfc.Quiet {
245
+
return nil
246
+
}
171
247
out := make(map[string]interface{})
172
248
out["type"] = "account"
173
249
out["payload"] = evt
···
180
256
}
181
257
182
258
func (gfc *GoatFirehoseConsumer) handleSyncEvent(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Sync) error {
259
+
commit, err := repo.LoadCARCommit(ctx, bytes.NewReader(evt.Blocks))
260
+
if err != nil {
261
+
return err
262
+
}
263
+
if gfc.VerifyBasic {
264
+
if err := commit.VerifyStructure(); err != nil {
265
+
slog.Warn("bad commit object", "eventType", "sync", "did", evt.Did, "seq", evt.Seq, "err", err)
266
+
}
267
+
if _, err := syntax.ParseDID(evt.Did); err != nil {
268
+
slog.Warn("invalid DID", "eventType", "account", "did", evt.Did, "seq", evt.Seq)
269
+
}
270
+
}
271
+
if gfc.Quiet {
272
+
return nil
273
+
}
274
+
evt.Blocks = nil
183
275
out := make(map[string]interface{})
184
276
out["type"] = "sync"
277
+
out["commit"] = commit.AsData()
185
278
out["payload"] = evt
186
279
b, err := json.Marshal(out)
187
280
if err != nil {
···
194
287
// this is the simple version, when not in "records" mode: print the event as JSON, but don't include blocks
195
288
func (gfc *GoatFirehoseConsumer) handleCommitEvent(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error {
196
289
290
+
if gfc.VerifyBasic || gfc.VerifySig || gfc.VerifyMST {
291
+
292
+
logger := slog.With("eventType", "commit", "did", evt.Repo, "seq", evt.Seq, "rev", evt.Rev)
293
+
294
+
did, err := syntax.ParseDID(evt.Repo)
295
+
if err != nil {
296
+
return err
297
+
}
298
+
299
+
commit, err := repo.LoadCARCommit(ctx, bytes.NewReader(evt.Blocks))
300
+
if err != nil {
301
+
return err
302
+
}
303
+
304
+
if gfc.VerifySig {
305
+
ident, err := gfc.Dir.LookupDID(ctx, did)
306
+
if err != nil {
307
+
return err
308
+
}
309
+
pubkey, err := ident.PublicKey()
310
+
if err != nil {
311
+
return err
312
+
}
313
+
logger = logger.With("pds", ident.PDSEndpoint())
314
+
if err := commit.VerifySignature(pubkey); err != nil {
315
+
logger.Warn("commit signature validation failed", "err", err)
316
+
}
317
+
}
318
+
319
+
if len(evt.Blocks) == 0 {
320
+
logger.Warn("commit message missing blocks")
321
+
}
322
+
323
+
if gfc.VerifyBasic {
324
+
// the commit itself
325
+
if err := commit.VerifyStructure(); err != nil {
326
+
logger.Warn("bad commit object", "err", err)
327
+
}
328
+
// the event fields
329
+
rev, err := syntax.ParseTID(evt.Rev)
330
+
if err != nil {
331
+
logger.Warn("bad TID syntax in commit rev", "err", err)
332
+
}
333
+
if rev.String() != commit.Rev {
334
+
logger.Warn("event rev != commit rev", "commitRev", commit.Rev)
335
+
}
336
+
if did.String() != commit.DID {
337
+
logger.Warn("event DID != commit DID", "commitDID", commit.DID)
338
+
}
339
+
_, err = syntax.ParseDatetime(evt.Time)
340
+
if err != nil {
341
+
logger.Warn("bad datetime syntax in commit time", "time", evt.Time, "err", err)
342
+
}
343
+
if evt.TooBig {
344
+
logger.Warn("deprecated tooBig commit flag set")
345
+
}
346
+
if evt.Rebase {
347
+
logger.Warn("deprecated rebase commit flag set")
348
+
}
349
+
}
350
+
351
+
if gfc.VerifyMST {
352
+
if evt.PrevData == nil {
353
+
logger.Warn("prevData is nil, skipping MST check")
354
+
} else {
355
+
// TODO: break out this function in to smaller chunks
356
+
if _, err := repo.VerifyCommitMessage(ctx, evt); err != nil {
357
+
logger.Warn("failed to invert commit MST", "err", err)
358
+
}
359
+
}
360
+
}
361
+
}
362
+
363
+
if gfc.Quiet {
364
+
return nil
365
+
}
366
+
197
367
// apply collections filter
198
368
if len(gfc.CollectionFilter) > 0 {
199
369
keep := false
···
302
472
if err != nil {
303
473
return err
304
474
}
305
-
fmt.Println(string(b))
475
+
if !gfc.Quiet {
476
+
fmt.Println(string(b))
477
+
}
306
478
case "delete":
307
479
out["action"] = "delete"
308
480
b, err := json.Marshal(out)
309
481
if err != nil {
310
482
return err
311
483
}
312
-
fmt.Println(string(b))
484
+
if !gfc.Quiet {
485
+
fmt.Println(string(b))
486
+
}
313
487
default:
314
488
logger.Error("unexpected record op kind")
315
489
}
+7
cmd/goat/main.go
+7
cmd/goat/main.go
···
23
23
Name: "goat",
24
24
Usage: "Go AT protocol CLI tool",
25
25
Version: versioninfo.Short(),
26
+
Flags: []cli.Flag{
27
+
&cli.StringFlag{
28
+
Name: "log-level",
29
+
Usage: "log verbosity level (eg: warn, info, debug)",
30
+
EnvVars: []string{"GOAT_LOG_LEVEL", "GO_LOG_LEVEL", "LOG_LEVEL"},
31
+
},
32
+
},
26
33
}
27
34
app.Commands = []*cli.Command{
28
35
cmdRecordGet,
+25
cmd/goat/util.go
+25
cmd/goat/util.go
···
3
3
import (
4
4
"context"
5
5
"io"
6
+
"log/slog"
6
7
"os"
8
+
"strings"
7
9
8
10
"github.com/bluesky-social/indigo/atproto/identity"
9
11
"github.com/bluesky-social/indigo/atproto/syntax"
12
+
13
+
"github.com/urfave/cli/v2"
10
14
)
11
15
12
16
func resolveIdent(ctx context.Context, arg string) (*identity.Identity, error) {
···
42
46
}
43
47
return file, nil
44
48
}
49
+
50
+
func configLogger(cctx *cli.Context, writer io.Writer) *slog.Logger {
51
+
var level slog.Level
52
+
switch strings.ToLower(cctx.String("log-level")) {
53
+
case "error":
54
+
level = slog.LevelError
55
+
case "warn":
56
+
level = slog.LevelWarn
57
+
case "info":
58
+
level = slog.LevelInfo
59
+
case "debug":
60
+
level = slog.LevelDebug
61
+
default:
62
+
level = slog.LevelInfo
63
+
}
64
+
logger := slog.New(slog.NewJSONHandler(writer, &slog.HandlerOptions{
65
+
Level: level,
66
+
}))
67
+
slog.SetDefault(logger)
68
+
return logger
69
+
}