this repo has no description
progress on config and XXX removal
+175
-185
+8
-33
cmd/relay/handlers.go
+8
-33
cmd/relay/handlers.go
···
27
27
return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("hostname field empty or invalid: %s", body.Hostname)})
28
28
}
29
29
30
-
if noSSL && !s.relay.Config.SSL {
31
-
return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: "this relay requires SSL"})
30
+
if noSSL && !s.config.AllowInsecureHosts && !admin {
31
+
return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: "this relay requires host SSL"})
32
32
}
33
33
34
34
// TODO: could ensure that query and path are empty
35
-
36
-
// XXX: config if new PDS instances are allowed at all
37
35
38
36
if strings.HasPrefix(hostname, "localhost:") {
39
-
// XXX: config if localhost connections allowed
37
+
if !admin {
38
+
return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: "can not configure localhost via public endpoint"})
39
+
} else {
40
+
// allowed
41
+
}
40
42
} else {
41
43
banned, err := s.relay.DomainIsBanned(ctx, hostname)
42
44
if err != nil {
···
56
58
return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "HostNotFound", Message: fmt.Sprintf("host server unreachable: %s", err)})
57
59
}
58
60
59
-
/* XXX: forwarding requestCrawl should be handled by rainbow, not relay itself
60
-
if len(s.config.NextCrawlers) != 0 {
61
-
blob, err := json.Marshal(body)
62
-
if err != nil {
63
-
s.logger.Warn("could not forward requestCrawl, json err", "err", err)
64
-
} else {
65
-
go func(bodyBlob []byte) {
66
-
for _, rpu := range s.config.NextCrawlers {
67
-
pu := rpu.JoinPath("/xrpc/com.atproto.sync.requestCrawl")
68
-
response, err := s.crawlForwardClient.Post(pu.String(), "application/json", bytes.NewReader(bodyBlob))
69
-
if response != nil && response.Body != nil {
70
-
response.Body.Close()
71
-
}
72
-
if err != nil || response == nil {
73
-
s.logger.Warn("requestCrawl forward failed", "host", rpu, "err", err)
74
-
} else if response.StatusCode != http.StatusOK {
75
-
s.logger.Warn("requestCrawl forward failed", "host", rpu, "status", response.Status)
76
-
} else {
77
-
s.logger.Info("requestCrawl forward successful", "host", rpu)
78
-
}
79
-
}
80
-
}(blob)
81
-
}
82
-
}
83
-
*/
84
-
85
61
return s.relay.SubscribeToHost(hostname, noSSL, false)
86
62
}
87
63
···
147
123
func (s *Service) handleComAtprotoSyncListRepos(c echo.Context, cursor int64, limit int) (*comatproto.SyncListRepos_Output, error) {
148
124
ctx := c.Request().Context()
149
125
150
-
// XXX: document that ListAccounts is ordered by UID (ascending)
151
126
accounts, err := s.relay.ListAccounts(ctx, cursor, limit)
152
127
if err != nil {
153
128
s.logger.Error("failed to query accounts", "err", err)
···
245
220
repo, err := s.relay.GetAccountRepo(ctx, acc.UID)
246
221
if err != nil {
247
222
if errors.Is(err, relay.ErrAccountRepoNotFound) {
248
-
// XXX: return partial result? some special error? desynchronized?
223
+
return nil, c.JSON(http.StatusNotFound, xrpc.XRPCError{ErrStr: "RepoNotSynchronized", Message: "do not know current repo state for account"})
249
224
}
250
225
return nil, err
251
226
}
+21
-37
cmd/relay/handlers_admin.go
+21
-37
cmd/relay/handlers_admin.go
···
34
34
if err != nil {
35
35
return &echo.HTTPError{Code: http.StatusBadRequest, Message: err.Error()}
36
36
}
37
-
s.relay.Config.DisableNewHosts = !enabled
37
+
s.config.DisableRequestCrawl = !enabled
38
38
return c.JSON(http.StatusOK, map[string]any{
39
39
"success": "true",
40
40
})
···
42
42
43
43
func (s *Service) handleAdminGetSubsEnabled(c echo.Context) error {
44
44
return c.JSON(http.StatusOK, map[string]bool{
45
-
"enabled": s.relay.Config.DisableNewHosts,
45
+
"enabled": s.config.DisableRequestCrawl,
46
46
})
47
47
}
48
48
···
59
59
}
60
60
61
61
s.relay.Slurper.NewHostPerDayLimiter.SetLimit(limit)
62
+
63
+
// TODO: forward to SiblingRelayHosts
62
64
return c.JSON(http.StatusOK, map[string]any{
63
65
"success": "true",
64
66
})
···
95
97
Message: err.Error(),
96
98
}
97
99
}
100
+
101
+
// TODO: forward to SiblingRelayHosts
98
102
return c.JSON(http.StatusOK, map[string]any{
99
103
"success": "true",
100
104
})
···
120
124
Message: err.Error(),
121
125
}
122
126
}
127
+
128
+
// TODO: forward to SiblingRelayHosts
123
129
return c.JSON(http.StatusOK, map[string]any{
124
130
"success": "true",
125
131
})
···
240
246
}
241
247
hostInfos[i].EventsSeenSinceStartup = uint64(m.Counter.GetValue())
242
248
243
-
/* XXX: compute these from account limit
249
+
/* TODO: compute these from account limit
244
250
hostInfos[i].PerSecondEventRate = rateLimit{
245
251
Max: p.RateLimit,
246
252
WindowSeconds: 1,
···
317
323
// kill any active connection (there may not be one, so ignore error)
318
324
_ = s.relay.Slurper.KillUpstreamConnection(host.Hostname, false)
319
325
326
+
// TODO: forward to SiblingRelayHosts
320
327
return c.JSON(http.StatusOK, map[string]any{
321
328
"success": "true",
322
329
})
···
345
352
}
346
353
}
347
354
355
+
// TODO: forward to SiblingRelayHosts
348
356
return c.JSON(http.StatusOK, map[string]any{
349
357
"success": "true",
350
358
})
···
390
398
return err
391
399
}
392
400
401
+
// TODO: forward to SiblingRelayHosts
393
402
return c.JSON(http.StatusOK, map[string]any{
394
403
"success": "true",
395
404
})
···
408
417
return err
409
418
}
410
419
420
+
// TODO: forward to SiblingRelayHosts
411
421
return c.JSON(http.StatusOK, map[string]any{
412
422
"success": "true",
413
423
})
···
425
435
return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("invalid body: %s", err))
426
436
}
427
437
428
-
var pds models.Host
429
-
if err := s.db.Where("host = ?", body.Host).First(&pds).Error; err != nil {
438
+
var host models.Host
439
+
if err := s.db.Where("host = ?", body.Host).First(&host).Error; err != nil {
430
440
return err
431
441
}
432
442
433
443
// Update the rate limits in the DB
434
-
pds.RateLimit = float64(body.PerSecond)
435
-
pds.HourlyEventLimit = body.PerHour
436
-
pds.DailyEventLimit = body.PerDay
437
-
pds.RepoLimit = body.RepoLimit
444
+
host.RateLimit = float64(body.PerSecond)
445
+
host.HourlyEventLimit = body.PerHour
446
+
host.DailyEventLimit = body.PerDay
447
+
host.RepoLimit = body.RepoLimit
438
448
439
-
if err := s.db.Save(&pds).Error; err != nil {
449
+
if err := s.db.Save(&host).Error; err != nil {
440
450
return echo.NewHTTPError(http.StatusInternalServerError, fmt.Errorf("failed to save rate limit changes: %w", err))
441
451
}
442
452
443
453
// Update the rate limit in the limiter
444
-
limits := s.relay.Slurper.GetOrCreateLimiters(pds.ID, body.PerSecond, body.PerHour, body.PerDay)
454
+
limits := s.relay.Slurper.GetOrCreateLimiters(host.ID, body.PerSecond, body.PerHour, body.PerDay)
445
455
limits.PerSecond.SetLimit(body.PerSecond)
446
456
limits.PerHour.SetLimit(body.PerHour)
447
457
limits.PerDay.SetLimit(body.PerDay)
···
451
461
})
452
462
}
453
463
*/
454
-
455
-
/* XXX: DREPRECATED
456
-
func (s *Service) handleAdminAddTrustedDomain(c echo.Context) error {
457
-
domain := c.QueryParam("domain")
458
-
if domain == "" {
459
-
return fmt.Errorf("must specify domain in query parameter")
460
-
}
461
-
462
-
// Check if the domain is already trusted
463
-
trustedDomains := s.relay.Slurper.GetTrustedDomains()
464
-
if slices.Contains(trustedDomains, domain) {
465
-
return &echo.HTTPError{
466
-
Code: http.StatusBadRequest,
467
-
Message: "domain is already trusted",
468
-
}
469
-
}
470
-
471
-
if err := s.relay.Slurper.AddTrustedDomain(domain); err != nil {
472
-
return err
473
-
}
474
-
475
-
return c.JSON(http.StatusOK, map[string]any{
476
-
"success": true,
477
-
})
478
-
}
479
-
*/
+19
-8
cmd/relay/main.go
+19
-8
cmd/relay/main.go
···
124
124
Usage: "don't process public (un-authenticated) com.atproto.sync.requestCrawl",
125
125
EnvVars: []string{"RELAY_DISABLE_REQUEST_CRAWL"},
126
126
},
127
-
// XXX: should this be handled by rainbow instead of relays?
127
+
&cli.BoolFlag{
128
+
Name: "allow-insecure-hosts",
129
+
Usage: "enables subscription to non-SSL hosts via requestCrawl",
130
+
EnvVars: []string{"RELAY_ALLOW_INSECURE_HOSTS"},
131
+
},
132
+
&cli.BoolFlag{
133
+
Name: "lenient-sync-validation",
134
+
Usage: "when messages fail atproto 'Sync 1.1' validation, just log, don't drop",
135
+
EnvVars: []string{"RELAY_LENIENT_SYNC_VALIDATION"},
136
+
},
128
137
&cli.StringSliceFlag{
129
-
Name: "forward-crawl-requests",
130
-
Usage: "servers (eg https://example.com) to forward requestCrawl on to; multiple allowed",
131
-
EnvVars: []string{"RELAY_FORWARD_CRAWL_REQUESTS", "RELAY_NEXT_CRAWLER"},
138
+
Name: "sibling-relays",
139
+
Usage: "servers (eg https://example.com) to forward admin state changes to; multiple allowed",
140
+
EnvVars: []string{"RELAY_SIBLING_RELAYS"},
132
141
},
133
142
&cli.StringSliceFlag{
134
143
Name: "trusted-domains",
···
228
237
relayConfig.QueueDepthPerHost = cctx.Int("host-queue-depth")
229
238
relayConfig.DefaultRepoLimit = cctx.Int64("default-account-limit")
230
239
relayConfig.TrustedDomains = cctx.StringSlice("trusted-domains")
240
+
relayConfig.LenientSyncValidation = cctx.Bool("lenient-sync-validation")
231
241
232
242
svcConfig := DefaultServiceConfig()
243
+
svcConfig.AllowInsecureHosts = cctx.Bool("allow-insecure-hosts")
233
244
svcConfig.DisableRequestCrawl = cctx.Bool("disable-request-crawl")
234
-
svcConfig.ForwardCrawlRequestHosts = cctx.StringSlice("forward-crawl-requests")
235
-
if len(svcConfig.ForwardCrawlRequestHosts) > 0 {
236
-
logger.Info("crawl request forwarding enabled", "servers", svcConfig.ForwardCrawlRequestHosts)
245
+
svcConfig.SiblingRelayHosts = cctx.StringSlice("sibling-relays")
246
+
if len(svcConfig.SiblingRelayHosts) > 0 {
247
+
logger.Info("sibling relay hosts configured for admin state forwarding", "servers", svcConfig.SiblingRelayHosts)
237
248
}
238
249
if cctx.IsSet("admin-password") {
239
250
svcConfig.AdminPassword = cctx.String("admin-password")
···
251
262
if err != nil {
252
263
return err
253
264
}
254
-
svc, err := NewService(db, r, svcConfig)
265
+
svc, err := NewService(r, svcConfig)
255
266
if err != nil {
256
267
return err
257
268
}
+13
-3
cmd/relay/relay/account.go
+13
-3
cmd/relay/relay/account.go
···
4
4
"context"
5
5
"errors"
6
6
"fmt"
7
+
"strings"
7
8
8
9
"github.com/bluesky-social/indigo/atproto/syntax"
9
10
"github.com/bluesky-social/indigo/cmd/relay/relay/models"
···
193
194
return nil
194
195
}
195
196
197
+
// Returns the of active accounts (based on local and upstream status). The sort order is by UID, ascending.
196
198
func (r *Relay) ListAccounts(ctx context.Context, cursor int64, limit int) ([]*models.Account, error) {
197
199
198
-
// XXX: what status filter should be in place here? not deleted in addition to not takendown?
199
200
accounts := []*models.Account{}
200
-
if err := r.db.Model(&models.Account{}).Where("uid > ? AND status IS NOT 'takendown' AND (upstream_status IS NULL OR upstream_status = 'active')", cursor).Order("uid").Limit(limit).Find(&accounts).Error; err != nil {
201
+
if err := r.db.Model(&models.Account{}).Where("uid > ? AND status = 'active' AND upstream_status = 'active'", cursor).Order("uid").Limit(limit).Find(&accounts).Error; err != nil {
201
202
return nil, err
202
203
}
203
204
return accounts, nil
···
216
217
return r.db.Exec("INSERT INTO account_repo (uid, rev, commit_cid, commit_data) VALUES (?, ?, ?, ?) ON CONFLICT (uid) DO UPDATE SET rev = EXCLUDED.rev, commit_cid = EXCLUDED.commit_cid, commit_data = EXCLUDED.commit_data", uid, rev, commitCID, commitDataCID).Error
217
218
}
218
219
219
-
// this function with exact name and args implements the `diskpersist.UidSource` interface
220
+
// This implements the `diskpersist.UidSource` interface
220
221
func (r *Relay) DidToUid(ctx context.Context, did string) (uint64, error) {
221
222
// NOTE: not re-parsing DID here (this function is called "loopback" from persister)
222
223
xu, err := r.GetAccount(ctx, syntax.DID(did))
···
228
229
}
229
230
return xu.UID, nil
230
231
}
232
+
233
+
// In the general case, DIDs are case-sensitive. But PLC and did:web should not be, and should normalize to lower-case.
234
+
func NormalizeDID(orig syntax.DID) syntax.DID {
235
+
lower := strings.ToLower(string(orig))
236
+
if strings.HasPrefix(lower, "did:plc:") || strings.HasPrefix(lower, "did:web:") {
237
+
return syntax.DID(lower)
238
+
}
239
+
return orig
240
+
}
+28
cmd/relay/relay/account_test.go
+28
cmd/relay/relay/account_test.go
···
1
+
package relay
2
+
3
+
import (
4
+
"testing"
5
+
6
+
"github.com/stretchr/testify/assert"
7
+
)
8
+
9
+
type DIDFixture struct {
10
+
Val string
11
+
Norm string
12
+
}
13
+
14
+
func TestNormalizeDID(t *testing.T) {
15
+
assert := assert.New(t)
16
+
17
+
fixtures := []HostnameFixture{
18
+
DIDFixture{Val: "did:web:example.com", Norm: "did:web:example.com"},
19
+
DIDFixture{Val: "did:web:example.com", Norm: "did:web:example.com"},
20
+
DIDFixture{Val: "did:web:EXAMPLE.com", Norm: "did:web:example.com"},
21
+
DIDFixture{Val: "did:plc:ABC123", Norm: "did:plc:abc123"},
22
+
DIDFixture{Val: "did:other:ABC", Norm: "did:other:ABC"},
23
+
}
24
+
25
+
for _, f := range fixtures {
26
+
assert.Equal(f.Norm, NormalizeDID(f.Val))
27
+
}
28
+
}
+16
-28
cmd/relay/relay/crawl.go
+16
-28
cmd/relay/relay/crawl.go
···
31
31
}
32
32
}
33
33
34
-
return !r.Config.DisableNewHosts
34
+
return true
35
35
}
36
36
37
37
func (r *Relay) SubscribeToHost(hostname string, noSSL, adminForce bool) error {
38
38
39
-
// if we already have an active subscription going, exit early
39
+
// if we already have an active subscription, exit early
40
40
if r.Slurper.CheckIfSubscribed(hostname) {
41
41
return nil
42
42
}
43
43
44
-
// XXX: new PDS daily rate-limit
45
-
44
+
// fetch host info from database. this query will not error if host does not yet exist
46
45
newHost := false
47
46
var host models.Host
48
47
if err := r.db.Find(&host, "hostname = ?", hostname).Error; err != nil {
···
50
49
}
51
50
52
51
if host.ID == 0 {
52
+
newHost = true
53
53
if !adminForce && !r.canSlurpHost(hostname) {
54
+
// TODO: is this the correct error code?
54
55
return ErrNewSubsDisabled
55
56
}
56
-
// New PDS!
57
-
npds := models.Host{
57
+
58
+
// XXX: new host daily rate-limit
59
+
60
+
host = models.Host{
58
61
Hostname: hostname,
59
62
NoSSL: noSSL,
60
63
Status: models.HostStatusActive,
61
64
AccountLimit: r.Config.DefaultRepoLimit,
62
65
}
63
-
/* XXX
64
-
if rateOverrides != nil {
65
-
npds.RateLimit = float64(rateOverrides.PerSecond)
66
-
npds.HourlyEventLimit = rateOverrides.PerHour
67
-
npds.DailyEventLimit = rateOverrides.PerDay
68
-
npds.RepoLimit = rateOverrides.RepoLimit
69
-
}
70
-
*/
71
-
if err := r.db.Create(&npds).Error; err != nil {
66
+
67
+
if err := r.db.Create(&newHost).Error; err != nil {
72
68
return err
73
69
}
74
70
75
-
newHost = true
76
-
host = npds
71
+
r.Logger.Info("adding new host subscription", "hostname", hostname, "noSSL", noSSL, "adminForce", adminForce)
77
72
} else if host.Status == models.HostStatusBanned {
78
73
return fmt.Errorf("cannot subscribe to banned pds")
79
74
}
80
75
81
-
/* XXX
82
-
if !host.Registered && reg {
83
-
host.Registered = true
84
-
if err := s.db.Model(models.Host{}).Where("id = ?", host.ID).Update("registered", true).Error; err != nil {
85
-
return err
86
-
}
87
-
}
88
-
*/
89
-
90
76
return r.Slurper.Subscribe(&host, newHost)
91
77
}
92
78
···
99
85
}
100
86
101
87
for _, host := range all {
102
-
// copy host
88
+
logger := r.Logger.With("hostID", host.ID, "hostname", host.Hostname)
89
+
logger.Info("re-subscribing to active host")
90
+
// make a copy of host
103
91
host := host
104
92
err := r.Slurper.Subscribe(&host, false)
105
93
if err != nil {
106
-
r.Logger.Warn("failed to re-subscribe to host", "hostID", host.ID, "hostname", host.Hostname, "err", err)
94
+
logger.Warn("failed to re-subscribe to host", "err", err)
107
95
}
108
96
}
109
97
return nil
+2
-1
cmd/relay/relay/domain_ban.go
+2
-1
cmd/relay/relay/domain_ban.go
···
19
19
func (r *Relay) DomainIsBanned(ctx context.Context, hostname string) (bool, error) {
20
20
21
21
if strings.HasPrefix(hostname, "localhost:") {
22
-
// XXX: check localhost config separately
22
+
// this method never allows localhost; need to use admin-mode for that
23
+
return true, nil
23
24
}
24
25
25
26
// otherwise we shouldn't have a port/colon
+1
cmd/relay/relay/errors.go
+1
cmd/relay/relay/errors.go
···
8
8
ErrHostNotFound = errors.New("unknown host or PDS")
9
9
ErrAccountNotFound = errors.New("unknown account")
10
10
ErrAccountRepoNotFound = errors.New("repository state not available")
11
+
ErrNotPDS = errors.New("server is not a PDS")
11
12
12
13
// TODO: these might need better names
13
14
ErrTimeoutShutdown = errors.New("timed out waiting for new events")
+8
-3
cmd/relay/relay/host.go
+8
-3
cmd/relay/relay/host.go
···
82
82
83
83
// parses, normalizes, and validates a raw URL (HTTP or WebSocket) in to a hostname for subscriptions
84
84
//
85
-
// Hostnames much be DNS names, not IP addresses
85
+
// Hostnames must be DNS names, not IP addresses.
86
86
func ParseHostname(raw string) (hostname string, noSSL bool, err error) {
87
87
88
88
// handle case of bare hostname
89
89
if !strings.Contains(raw, "://") {
90
-
raw = "https://" + raw
90
+
if strings.HasPrefix(raw, "localhost:") {
91
+
raw = "http://" + raw
92
+
} else {
93
+
raw = "https://" + raw
94
+
}
91
95
}
92
96
93
97
u, err := url.Parse(raw)
···
100
104
default:
101
105
return "", false, fmt.Errorf("unsupported URL scheme: %s", u.Scheme)
102
106
}
107
+
103
108
// 'localhost' (exact string) is allowed *with* a required port number; SSL is optional
104
109
if u.Hostname() == "localhost" {
105
-
if u.Port() == "" {
110
+
if u.Port() == "" || !strings.HasPrefix(u.Host, "localhost:") {
106
111
return "", false, fmt.Errorf("port number is required for localhost")
107
112
}
108
113
return u.Host, noSSL, nil
-3
cmd/relay/relay/host_checker.go
-3
cmd/relay/relay/host_checker.go
···
2
2
3
3
import (
4
4
"context"
5
-
"errors"
6
5
"fmt"
7
6
"net/http"
8
7
···
10
9
"github.com/bluesky-social/indigo/atproto/identity"
11
10
"github.com/bluesky-social/indigo/xrpc"
12
11
)
13
-
14
-
var ErrNotPDS = errors.New("server is not a PDS")
15
12
16
13
// Simple interface for doing host and account status checks.
17
14
//
+5
-1
cmd/relay/relay/host_test.go
+5
-1
cmd/relay/relay/host_test.go
···
23
23
HostnameFixture{Val: "ws://pds.example.com", Hostname: "pds.example.com", NoSSL: true},
24
24
HostnameFixture{Val: "pds.example.com", Hostname: "pds.example.com", NoSSL: false},
25
25
HostnameFixture{Val: "morel.us-east.host.bsky.network", Hostname: "morel.us-east.host.bsky.network", NoSSL: false},
26
+
HostnameFixture{Val: "https://service.local", Hostname: "service.local", NoSSL: false}, // TODO: SSRF
27
+
HostnameFixture{Val: "localhost:8080", Hostname: "localhost:8080", NoSSL: true},
28
+
HostnameFixture{Val: "https://localhost:8080", Hostname: "localhost:8080", NoSSL: false},
29
+
HostnameFixture{Val: "https://localhost", Error: true},
30
+
HostnameFixture{Val: "localhost", Error: true},
26
31
HostnameFixture{Val: "https://8.8.8.8", Error: true},
27
32
HostnameFixture{Val: "https://internal", Error: true},
28
33
HostnameFixture{Val: "at://pds.example.com", Error: true},
29
34
HostnameFixture{Val: "ftp://pds.example.com", Error: true},
30
-
HostnameFixture{Val: "https://service.local", Hostname: "service.local", NoSSL: false}, // TODO: SSRF
31
35
}
32
36
33
37
for _, f := range fixtures {
+14
-4
cmd/relay/relay/ingest.go
+14
-4
cmd/relay/relay/ingest.go
···
57
57
}
58
58
}
59
59
60
-
// handles the shared part of event processing: that the account existing, is associated with this host, etc
60
+
// Implements the shared part of event processing: that the account existing, is associated with this host, etc.
61
+
//
62
+
// If there is no error, the returned account is always non-nil, but the identity may be nil (if there was a resolution error).
61
63
func (r *Relay) preProcessEvent(ctx context.Context, didStr string, hostname string, hostID uint64, logger *slog.Logger) (*models.Account, *identity.Identity, error) {
62
64
63
65
did, err := syntax.ParseDID(didStr)
64
66
if err != nil {
65
67
return nil, nil, fmt.Errorf("invalid DID in message: %w", err)
66
68
}
67
-
// XXX: did = did.Normalize()
69
+
// TODO: add a test case for non-normalized DID
70
+
did = NormalizeDID(did)
68
71
69
72
acc, err := r.GetAccount(ctx, did)
70
73
if err != nil {
···
95
98
96
99
ident, err := r.Dir.LookupDID(ctx, did)
97
100
if err != nil {
98
-
// XXX: handle more granularly (eg, true NotFound vs other errors); and add tests
99
-
logger.Warn("failed to load identity")
101
+
logger.Warn("failed to load identity", "did", did, "err", err)
100
102
}
101
103
return acc, ident, nil
102
104
}
···
113
115
if !acc.IsActive() {
114
116
logger.Info("dropping commit message for non-active account", "status", acc.Status, "upstreamStatus", acc.UpstreamStatus)
115
117
return nil
118
+
}
119
+
120
+
if ident == nil {
121
+
// XXX: what to do if identity resolution fails
116
122
}
117
123
118
124
prevRepo, err := r.GetAccountRepo(ctx, acc.UID)
···
159
165
if !acc.IsActive() {
160
166
logger.Info("dropping commit message for non-active account", "status", acc.Status, "upstreamStatus", acc.UpstreamStatus)
161
167
return nil
168
+
}
169
+
170
+
if ident == nil {
171
+
// XXX: what to do if identity resolution fails
162
172
}
163
173
164
174
newRepo, err := r.VerifyRepoSync(ctx, evt, ident, hostname)
-7
cmd/relay/relay/metrics.go
-7
cmd/relay/relay/metrics.go
···
31
31
Help: "The total number of sync events received",
32
32
}, []string{"pds"})
33
33
34
-
/* XXX
35
-
var repoCommitsResultCounter = promauto.NewCounterVec(prometheus.CounterOpts{
36
-
Name: "repo_commits_result_counter",
37
-
Help: "The results of commit events received",
38
-
}, []string{"pds", "status"})
39
-
*/
40
-
41
34
var eventsSentCounter = promauto.NewCounterVec(prometheus.CounterOpts{
42
35
Name: "events_sent_counter",
43
36
Help: "The total number of events sent to consumers",
+8
-8
cmd/relay/relay/models/models.go
+8
-8
cmd/relay/relay/models/models.go
···
6
6
"gorm.io/gorm"
7
7
)
8
8
9
-
// TODO: revisit this
10
9
type DomainBan struct {
11
10
gorm.Model
12
11
Domain string `gorm:"unique"`
···
15
14
type HostStatus string
16
15
17
16
const (
18
-
HostStatusActive = HostStatus("active")
19
-
HostStatusIdle = HostStatus("idle")
20
-
HostStatusOffline = HostStatus("offline")
21
-
HostStatusBanned = HostStatus("banned")
17
+
HostStatusActive = HostStatus("active")
18
+
HostStatusIdle = HostStatus("idle")
19
+
HostStatusOffline = HostStatus("offline")
20
+
HostStatusThrottled = HostStatus("throttled")
21
+
HostStatusBanned = HostStatus("banned")
22
22
)
23
23
24
24
type Host struct {
···
26
26
CreatedAt time.Time
27
27
UpdatedAt time.Time
28
28
29
-
// hostname, without URL scheme. might include a port number if localhost, otherwise should not
29
+
// hostname, without URL scheme. if localhost, must include a port number; otherwise must not include port
30
30
Hostname string `gorm:"column:hostname;uniqueIndex;not null"`
31
31
32
32
// indicates ws:// not wss://
···
37
37
38
38
// TODO: ThrottleUntil time.Time
39
39
40
-
// indicates this is a highly trusted PDS (different limits apply)
40
+
// indicates this is a highly trusted host (PDS) and different limits apply
41
41
Trusted bool `gorm:"column:trusted;default:false"`
42
42
43
43
// enum of account status
···
64
64
AccountStatusSuspended = AccountStatus("suspended")
65
65
AccountStatusTakendown = AccountStatus("takendown")
66
66
AccountStatusThrottled = AccountStatus("throttled")
67
-
AccountStatusHostThrottled = AccountStatus("host-throttled") // TODO: actually implement this
67
+
AccountStatusHostThrottled = AccountStatus("host-throttled") // TODO: not yet implemented
68
68
69
69
// generic "not active, but not known" status
70
70
AccountStatusInactive = AccountStatus("inactive")
+7
-10
cmd/relay/relay/relay.go
+7
-10
cmd/relay/relay/relay.go
···
39
39
}
40
40
41
41
type RelayConfig struct {
42
-
SSL bool
43
42
DefaultRepoLimit int64
44
43
ConcurrencyPerHost int
45
44
QueueDepthPerHost int
46
-
SkipAccountHostCheck bool // XXX: only used for testing
47
-
LenientSyncValidation bool // XXX: wire through config
45
+
LenientSyncValidation bool
46
+
TrustedDomains []string
48
47
49
-
// if true, ignore "requestCrawl"
50
-
DisableNewHosts bool
51
-
TrustedDomains []string
48
+
// If true, skip validation that messages for a given account (DID) are coming from the expected upstream host (PDS). Currently only used in tests; might be used for intermediate relays in the future.
49
+
SkipAccountHostCheck bool
52
50
}
53
51
54
52
func DefaultRelayConfig() *RelayConfig {
55
-
// NOTE: many of these defaults are CLI arg defaults
53
+
// NOTE: many of these defaults are clobbered by CLI arguments
56
54
return &RelayConfig{
57
-
SSL: true,
58
55
DefaultRepoLimit: 100,
59
56
ConcurrencyPerHost: 40,
60
57
QueueDepthPerHost: 1000,
···
89
86
return nil, err
90
87
}
91
88
92
-
// XXX: need to pass-through more relay configs
93
89
slurpConfig := DefaultSlurperConfig()
94
-
slurpConfig.SSL = config.SSL
95
90
slurpConfig.DefaultRepoLimit = config.DefaultRepoLimit
96
91
slurpConfig.ConcurrencyPerHost = config.ConcurrencyPerHost
97
92
slurpConfig.QueueDepthPerHost = config.QueueDepthPerHost
93
+
98
94
// register callbacks to persist cursors and host state in database
99
95
slurpConfig.PersistCursorCallback = r.PersistHostCursors
100
96
slurpConfig.PersistHostStatusCallback = r.UpdateHostStatus
97
+
101
98
s, err := NewSlurper(r.processRepoEvent, slurpConfig, r.Logger)
102
99
if err != nil {
103
100
return nil, err
-2
cmd/relay/relay/slurper.go
-2
cmd/relay/relay/slurper.go
···
54
54
}
55
55
56
56
type SlurperConfig struct {
57
-
SSL bool
58
57
DefaultPerSecondLimit int64
59
58
DefaultPerHourLimit int64
60
59
DefaultPerDayLimit int64
···
70
69
func DefaultSlurperConfig() *SlurperConfig {
71
70
// NOTE: many of these defaults are overruled by DefaultRelayConfig, or even process CLI arg defaults
72
71
return &SlurperConfig{
73
-
SSL: false,
74
72
NewHostPerDayLimit: 50,
75
73
DefaultPerSecondLimit: 50,
76
74
DefaultPerHourLimit: 2500,
+7
-14
cmd/relay/service.go
+7
-14
cmd/relay/service.go
···
15
15
"github.com/labstack/echo/v4"
16
16
"github.com/labstack/echo/v4/middleware"
17
17
"github.com/prometheus/client_golang/prometheus/promhttp"
18
-
"gorm.io/gorm"
19
18
)
20
19
21
20
type Service struct {
22
-
db *gorm.DB // XXX
23
21
logger *slog.Logger
24
22
relay *relay.Relay
25
23
config ServiceConfig
···
28
26
}
29
27
30
28
type ServiceConfig struct {
31
-
// list of hosts which get forwarded com.atproto.sync.requestCrawl (HTTP POST)
32
-
ForwardCrawlRequestHosts []string
29
+
// list of hosts which get forwarded admin state changes (takedowns, etc)
30
+
SiblingRelayHosts []string
33
31
34
32
// verified against Basic admin auth
35
33
AdminPassword string
···
39
37
40
38
// if true, don't process public (unauthenticated) requestCrawl
41
39
DisableRequestCrawl bool
40
+
41
+
// if true, allows non-SSL hosts to be added via public requestCrawl
42
+
AllowInsecureHosts bool
42
43
}
43
44
44
45
func DefaultServiceConfig() *ServiceConfig {
···
47
48
}
48
49
}
49
50
50
-
func NewService(db *gorm.DB, r *relay.Relay, config *ServiceConfig) (*Service, error) {
51
+
func NewService(r *relay.Relay, config *ServiceConfig) (*Service, error) {
51
52
52
53
if config == nil {
53
54
config = DefaultServiceConfig()
54
55
}
55
56
56
57
svc := &Service{
57
-
db: db,
58
58
logger: slog.Default().With("system", "relay"),
59
59
relay: r,
60
60
config: *config,
···
90
90
AllowOrigins: []string{"*"},
91
91
AllowHeaders: []string{echo.HeaderOrigin, echo.HeaderContentType, echo.HeaderAccept, echo.HeaderAuthorization},
92
92
}))
93
-
94
-
if !svc.relay.Config.SSL {
95
-
e.Use(middleware.LoggerWithConfig(middleware.LoggerConfig{
96
-
Format: "method=${method}, uri=${uri}, status=${status} latency=${latency_human}\n",
97
-
}))
98
-
} else {
99
-
e.Use(middleware.LoggerWithConfig(middleware.DefaultLoggerConfig))
100
-
}
93
+
e.Use(middleware.LoggerWithConfig(middleware.DefaultLoggerConfig))
101
94
102
95
// React uses a virtual router, so we need to serve the index.html for all
103
96
// routes that aren't otherwise handled or in the /assets directory.
+15
-15
cmd/relay/stream/consumer.go
+15
-15
cmd/relay/stream/consumer.go
···
115
115
// HandleRepoStream
116
116
// con is source of events
117
117
// sched gets AddWork for each event
118
-
// log may be nil for default logger
119
-
func HandleRepoStream(ctx context.Context, con *websocket.Conn, sched Scheduler, log *slog.Logger) error {
120
-
if log == nil {
121
-
log = slog.Default().With("system", "events")
118
+
// logger may be nil for default logger
119
+
func HandleRepoStream(ctx context.Context, con *websocket.Conn, sched Scheduler, logger *slog.Logger) error {
120
+
if logger == nil {
121
+
logger = slog.Default().With("system", "events")
122
122
}
123
123
ctx, cancel := context.WithCancel(ctx)
124
124
defer cancel()
···
136
136
select {
137
137
case <-t.C:
138
138
if err := con.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second*10)); err != nil {
139
-
log.Warn("failed to ping", "err", err)
139
+
logger.Warn("failed to ping", "err", err)
140
140
failcount++
141
141
if failcount >= 4 {
142
-
log.Error("too many ping fails", "count", failcount)
142
+
logger.Error("too many ping fails", "count", failcount)
143
143
con.Close()
144
144
return
145
145
}
···
165
165
166
166
con.SetPongHandler(func(_ string) error {
167
167
if err := con.SetReadDeadline(time.Now().Add(time.Minute)); err != nil {
168
-
log.Error("failed to set read deadline", "err", err)
168
+
logger.Error("failed to set read deadline", "err", err)
169
169
}
170
170
171
171
return nil
···
214
214
}
215
215
216
216
if evt.Seq < lastSeq {
217
-
log.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
217
+
logger.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
218
218
}
219
219
220
220
lastSeq = evt.Seq
···
231
231
}
232
232
233
233
if evt.Seq < lastSeq {
234
-
log.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
234
+
logger.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
235
235
}
236
236
237
237
lastSeq = evt.Seq
···
249
249
}
250
250
251
251
if evt.Seq < lastSeq {
252
-
log.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
252
+
logger.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
253
253
}
254
254
lastSeq = evt.Seq
255
255
···
265
265
}
266
266
267
267
if evt.Seq < lastSeq {
268
-
log.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
268
+
logger.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
269
269
}
270
270
lastSeq = evt.Seq
271
271
···
281
281
}
282
282
283
283
if evt.Seq < lastSeq {
284
-
log.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
284
+
logger.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
285
285
}
286
286
lastSeq = evt.Seq
287
287
···
310
310
}
311
311
312
312
if evt.Seq < lastSeq {
313
-
log.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
313
+
logger.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
314
314
}
315
315
lastSeq = evt.Seq
316
316
···
327
327
}
328
328
329
329
if evt.Seq < lastSeq {
330
-
log.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
330
+
logger.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
331
331
}
332
332
lastSeq = evt.Seq
333
333
···
343
343
}
344
344
345
345
if evt.Seq < lastSeq {
346
-
log.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
346
+
logger.Error("Got events out of order from stream", "seq", evt.Seq, "prev", lastSeq)
347
347
}
348
348
349
349
lastSeq = evt.Seq
-4
cmd/relay/stream/events.go
-4
cmd/relay/stream/events.go
···
5
5
"errors"
6
6
"fmt"
7
7
"io"
8
-
"log/slog"
9
8
10
9
comatproto "github.com/bluesky-social/indigo/api/atproto"
11
10
lexutil "github.com/bluesky-social/indigo/lex/util"
12
11
13
12
cbg "github.com/whyrusleeping/cbor-gen"
14
13
)
15
-
16
-
// XXX
17
-
var log = slog.Default().With("system", "events")
18
14
19
15
const (
20
16
EvtKindErrorFrame = -1
+1
-1
cmd/relay/testing/consumer.go
+1
-1
cmd/relay/testing/consumer.go
+1
-2
cmd/relay/testing/runner.go
+1
-2
cmd/relay/testing/runner.go
···
33
33
func MustSimpleRelay(dir identity.Directory, tmpd string, lenient bool) *SimpleRelay {
34
34
35
35
relayConfig := relay.DefaultRelayConfig()
36
-
relayConfig.SSL = false
37
36
relayConfig.SkipAccountHostCheck = true
38
37
relayConfig.LenientSyncValidation = lenient
39
38
···
165
164
return fmt.Errorf("events didn't match")
166
165
}
167
166
} else {
168
-
// XXX: verify nothing returned? especially if last message in set
167
+
// TODO: verify nothing returned? especially if last message in set
169
168
}
170
169
}
171
170
return nil
+1
-1
cmd/relay/testing/sync_test.go
+1
-1
cmd/relay/testing/sync_test.go