this repo has no description
add HostChecker abstraction for PDS+account lookups
+176
-24
+1
-11
cmd/relayered/relay/account.go
+1
-11
cmd/relayered/relay/account.go
···
8
8
"strings"
9
9
"time"
10
10
11
-
comatproto "github.com/bluesky-social/indigo/api/atproto"
12
11
"github.com/bluesky-social/indigo/atproto/syntax"
13
12
"github.com/bluesky-social/indigo/cmd/relayered/relay/models"
14
-
"github.com/bluesky-social/indigo/xrpc"
15
13
16
14
"github.com/ipfs/go-cid"
17
15
"gorm.io/gorm"
···
152
150
// TODO: what do we actually want to track about the source we immediately got this message from vs the canonical Host?
153
151
r.Logger.Warn("pds discovered in new user flow", "pds", durl.String(), "did", did)
154
152
155
-
// Do a trivial API request against the Host to verify that it exists
156
-
pclient := &xrpc.Client{Host: durl.String()}
157
-
if r.Config.ApplyHostClientSettings != nil {
158
-
r.Config.ApplyHostClientSettings(pclient)
159
-
}
160
-
cfg, err := comatproto.ServerDescribeServer(ctx, pclient)
153
+
err = r.HostChecker.CheckHost(ctx, durl.String())
161
154
if err != nil {
162
155
// TODO: failing this shouldn't halt our indexing
163
156
return nil, fmt.Errorf("failed to check unrecognized pds: %w", err)
164
157
}
165
-
166
-
// since handles can be anything, checking against this list doesn't matter...
167
-
_ = cfg
168
158
169
159
// could check other things, a valid response is good enough for now
170
160
canonicalHost.Hostname = durl.Host
+105
cmd/relayered/relay/host_checker.go
+105
cmd/relayered/relay/host_checker.go
···
1
+
package relay
2
+
3
+
import (
4
+
"context"
5
+
"errors"
6
+
"fmt"
7
+
"net/http"
8
+
9
+
comatproto "github.com/bluesky-social/indigo/api/atproto"
10
+
"github.com/bluesky-social/indigo/atproto/identity"
11
+
"github.com/bluesky-social/indigo/xrpc"
12
+
)
13
+
14
+
var ErrNotPDS = errors.New("server is not a PDS")
15
+
16
+
// Simple interface for doing host and account status checks.
17
+
//
18
+
// The main reason this is an interface is to make testing/mocking easy.
19
+
type HostChecker interface {
20
+
// host should be a URL, including scheme, hostname (and optional port), but no path segment
21
+
CheckHost(ctx context.Context, host string) error
22
+
FetchAccountStatus(ctx context.Context, ident *identity.Identity) (string, error)
23
+
}
24
+
25
+
type HostClient struct {
26
+
Client *http.Client
27
+
UserAgent string
28
+
}
29
+
30
+
func NewHostClient(userAgent string) *HostClient {
31
+
if userAgent == "" {
32
+
userAgent = "indigo-relay"
33
+
}
34
+
return &HostClient{
35
+
Client: http.DefaultClient,
36
+
UserAgent: userAgent,
37
+
}
38
+
}
39
+
40
+
func (hc *HostClient) CheckHost(ctx context.Context, host string) error {
41
+
xrpcc := xrpc.Client{
42
+
Client: hc.Client,
43
+
UserAgent: &hc.UserAgent,
44
+
Host: host,
45
+
}
46
+
47
+
_, err := comatproto.ServerDescribeServer(ctx, &xrpcc)
48
+
if err != nil {
49
+
return fmt.Errorf("%w: %w", ErrNotPDS, err)
50
+
}
51
+
return nil
52
+
}
53
+
54
+
func (hc *HostClient) FetchAccountStatus(ctx context.Context, ident *identity.Identity) (string, error) {
55
+
pdsEndpoint := ident.PDSEndpoint()
56
+
if pdsEndpoint == "" {
57
+
return "", fmt.Errorf("account does not declare a PDS: %s", ident.DID)
58
+
}
59
+
60
+
xrpcc := xrpc.Client{
61
+
Client: hc.Client,
62
+
UserAgent: &hc.UserAgent,
63
+
Host: pdsEndpoint,
64
+
}
65
+
66
+
info, err := comatproto.SyncGetRepoStatus(ctx, &xrpcc, ident.DID.String())
67
+
if err != nil {
68
+
return "", err
69
+
}
70
+
if info.Active == true {
71
+
return "active", nil
72
+
} else if info.Status != nil {
73
+
return *info.Status, nil
74
+
} else {
75
+
return "inactive", nil
76
+
}
77
+
}
78
+
79
+
type MockHostChecker struct {
80
+
Hosts map[string]bool
81
+
Accounts map[string]string
82
+
}
83
+
84
+
func NewMockHostChecker() *MockHostChecker {
85
+
return &MockHostChecker{
86
+
Hosts: make(map[string]bool),
87
+
Accounts: make(map[string]string),
88
+
}
89
+
}
90
+
91
+
func (hc *MockHostChecker) CheckHost(ctx context.Context, host string) error {
92
+
_, ok := hc.Hosts[host]
93
+
if !ok {
94
+
return ErrNotPDS
95
+
}
96
+
return nil
97
+
}
98
+
99
+
func (hc *MockHostChecker) FetchAccountStatus(ctx context.Context, ident *identity.Identity) (string, error) {
100
+
status, ok := hc.Accounts[ident.DID.String()]
101
+
if !ok {
102
+
return "", ErrAccountNotFound
103
+
}
104
+
return status, nil
105
+
}
+53
cmd/relayered/relay/host_checker_test.go
+53
cmd/relayered/relay/host_checker_test.go
···
1
+
package relay
2
+
3
+
import (
4
+
"testing"
5
+
6
+
"github.com/bluesky-social/indigo/atproto/identity"
7
+
"github.com/bluesky-social/indigo/atproto/syntax"
8
+
9
+
"github.com/stretchr/testify/assert"
10
+
)
11
+
12
+
func TestMockHostChecker(t *testing.T) {
13
+
assert := assert.New(t)
14
+
ctx := t.Context()
15
+
var err error
16
+
17
+
hc := NewMockHostChecker()
18
+
hc.Hosts["https://pds.example.com"] = true
19
+
hc.Accounts["did:web:active.example.com"] = "active"
20
+
21
+
assert.NoError(hc.CheckHost(ctx, "https://pds.example.com"))
22
+
assert.Error(hc.CheckHost(ctx, ""))
23
+
assert.Error(hc.CheckHost(ctx, "https://dummy.example.com"))
24
+
25
+
s1, err := hc.FetchAccountStatus(ctx, &identity.Identity{DID: syntax.DID("did:web:active.example.com")})
26
+
assert.NoError(err)
27
+
assert.Equal("active", s1)
28
+
29
+
_, err = hc.FetchAccountStatus(ctx, &identity.Identity{DID: syntax.DID("did:web:nope.example.com")})
30
+
assert.Error(err)
31
+
}
32
+
33
+
func TestLiveHostChecker(t *testing.T) {
34
+
assert := assert.New(t)
35
+
ctx := t.Context()
36
+
var err error
37
+
38
+
dir := identity.DefaultDirectory()
39
+
hc := NewHostClient("indigo-tests")
40
+
41
+
assert.NoError(hc.CheckHost(ctx, "https://morel.us-east.host.bsky.network"))
42
+
assert.Error(hc.CheckHost(ctx, "https://dummy.example.com"))
43
+
44
+
ident, err := dir.LookupHandle(ctx, syntax.Handle("atproto.com"))
45
+
46
+
s1, err := hc.FetchAccountStatus(ctx, ident)
47
+
assert.NoError(err)
48
+
assert.Equal("active", s1)
49
+
50
+
ident.DID = syntax.DID("did:web:dummy.example.com")
51
+
_, err = hc.FetchAccountStatus(ctx, ident)
52
+
assert.Error(err)
53
+
}
+17
-13
cmd/relayered/relay/relay.go
+17
-13
cmd/relayered/relay/relay.go
···
17
17
var tracer = otel.Tracer("relay")
18
18
19
19
type Relay struct {
20
-
db *gorm.DB
21
-
dir identity.Directory
22
-
Logger *slog.Logger
23
-
Slurper *Slurper
24
-
Events *eventmgr.EventManager
25
-
Validator *Validator
26
-
Config RelayConfig
20
+
db *gorm.DB
21
+
dir identity.Directory
22
+
Logger *slog.Logger
23
+
Slurper *Slurper
24
+
Events *eventmgr.EventManager
25
+
Validator *Validator
26
+
HostChecker HostChecker
27
+
Config RelayConfig
27
28
28
29
// extUserLk serializes a section of syncHostAccount()
29
30
// TODO: at some point we will want to lock specific DIDs, this lock as is
···
65
66
66
67
uc, _ := lru.New[string, *models.Account](2_000_000)
67
68
69
+
hc := NewHostClient("relayered") // TODO: pass-through a user-agent from config?
70
+
68
71
r := &Relay{
69
-
db: db,
70
-
dir: dir,
71
-
Logger: slog.Default().With("system", "relay"),
72
-
Events: evtman,
73
-
Validator: vldtr,
74
-
Config: *config,
72
+
db: db,
73
+
dir: dir,
74
+
Logger: slog.Default().With("system", "relay"),
75
+
Events: evtman,
76
+
Validator: vldtr,
77
+
HostChecker: hc,
78
+
Config: *config,
75
79
76
80
consumersLk: sync.RWMutex{},
77
81
consumers: make(map[uint64]*SocketConsumer),