this repo has no description
working on some cross pds event streams, finished plc client impl
+240
-40
-2
api/plc.go
-2
api/plc.go
+7
-1
cmd/pds/main.go
+7
-1
cmd/pds/main.go
···
39
39
&cli.BoolFlag{
40
40
Name: "dbtracing",
41
41
},
42
+
&cli.StringFlag{
43
+
Name: "pdshost",
44
+
Usage: "hostname of the pds",
45
+
Value: "localhost:4989",
46
+
},
42
47
}
43
48
44
49
app.Commands = []*cli.Command{
···
112
117
return err
113
118
}
114
119
115
-
srv, err := server.NewServer(db, cs, "server.key", ".pdstest")
120
+
pdshost := cctx.String("pdshost")
121
+
srv, err := server.NewServer(db, cs, "server.key", ".pdstest", pdshost, server.NewFakeDid(db), []byte("jwtsecretplaceholder"))
116
122
if err != nil {
117
123
return err
118
124
}
+17
key/key.go
+17
key/key.go
···
6
6
"crypto/elliptic"
7
7
"crypto/rand"
8
8
"crypto/sha256"
9
+
"crypto/x509"
9
10
"fmt"
10
11
11
12
"github.com/multiformats/go-multibase"
···
68
69
69
70
return "did:key:" + kstr
70
71
}
72
+
73
+
func (k *Key) RawBytes() ([]byte, error) {
74
+
switch k.Type {
75
+
case "ed25519":
76
+
return k.Raw.([]byte), nil
77
+
case "P-256":
78
+
b, err := x509.MarshalECPrivateKey(k.Raw.(*ecdsa.PrivateKey))
79
+
if err != nil {
80
+
return nil, err
81
+
}
82
+
83
+
return b, nil
84
+
default:
85
+
return nil, fmt.Errorf("unsupported key type: %q", k.Type)
86
+
}
87
+
}
+6
-5
server/auth.go
+6
-5
server/auth.go
···
4
4
"context"
5
5
"crypto/rand"
6
6
"encoding/base64"
7
+
"fmt"
7
8
"time"
8
9
9
10
"github.com/lestrrat-go/jwx/jwa"
···
32
33
rand.Read(rval)
33
34
refreshTok.Set("jti", base64.StdEncoding.EncodeToString(rval))
34
35
35
-
accSig, err := jwt.Sign(accessTok, jwa.HS256, s.signingKey)
36
+
accSig, err := jwt.Sign(accessTok, jwa.HS256, s.jwtSigningKey)
36
37
if err != nil {
37
-
return nil, err
38
+
return nil, fmt.Errorf("signing access token: %w", err)
38
39
}
39
40
40
-
refSig, err := jwt.Sign(refreshTok, jwa.HS256, s.signingKey)
41
+
refSig, err := jwt.Sign(refreshTok, jwa.HS256, s.jwtSigningKey)
41
42
if err != nil {
42
-
return nil, err
43
+
return nil, fmt.Errorf("signing refresh token: %w", err)
43
44
}
44
45
45
46
return &xrpc.AuthInfo{
···
56
57
rval := make([]byte, 10)
57
58
rand.Read(rval)
58
59
59
-
accSig, err := jwt.Sign(accessTok, jwa.HS256, s.signingKey)
60
+
accSig, err := jwt.Sign(accessTok, jwa.HS256, s.jwtSigningKey)
60
61
if err != nil {
61
62
return nil, err
62
63
}
+8
-1
server/fakedid.go
+8
-1
server/fakedid.go
···
1
1
package schemagen
2
2
3
3
import (
4
+
"context"
4
5
"crypto/rand"
5
6
"encoding/hex"
6
7
8
+
"github.com/whyrusleeping/go-did"
9
+
"github.com/whyrusleeping/gosky/key"
7
10
"gorm.io/gorm"
8
11
)
9
12
···
22
25
return &FakeDid{db}
23
26
}
24
27
25
-
func (fd *FakeDid) NewForHandle(handle string) (string, error) {
28
+
func (fd *FakeDid) GetDocument(ctx context.Context, did string) (*did.Document, error) {
29
+
panic("nyi")
30
+
}
31
+
32
+
func (fd *FakeDid) CreateDID(ctx context.Context, sigkey *key.Key, recovery string, handle string, service string) (string, error) {
26
33
buf := make([]byte, 8)
27
34
rand.Read(buf)
28
35
d := "did:plc:" + hex.EncodeToString(buf)
+121
-4
server/federation_test.go
+121
-4
server/federation_test.go
···
1
1
package schemagen
2
2
3
3
import (
4
+
"context"
4
5
"crypto/ecdsa"
5
6
"crypto/elliptic"
6
7
"crypto/rand"
···
10
11
"os"
11
12
"path/filepath"
12
13
"testing"
14
+
"time"
13
15
14
16
"github.com/lestrrat-go/jwx/v2/jwk"
17
+
"github.com/whyrusleeping/gosky/api"
18
+
atproto "github.com/whyrusleeping/gosky/api/atproto"
19
+
bsky "github.com/whyrusleeping/gosky/api/bsky"
15
20
"github.com/whyrusleeping/gosky/carstore"
21
+
"github.com/whyrusleeping/gosky/xrpc"
16
22
"gorm.io/driver/sqlite"
17
23
"gorm.io/gorm"
18
24
)
19
25
20
26
func makeKey(t *testing.T, fname string) {
21
-
raw, err := ecdsa.GenerateKey(elliptic.P384(), rand.Reader)
27
+
raw, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
22
28
if err != nil {
23
29
t.Fatal(fmt.Errorf("failed to generate new ECDSA private key: %s", err))
24
30
}
···
42
48
if err := os.WriteFile(fname, buf, 0664); err != nil {
43
49
t.Fatal(err)
44
50
}
45
-
46
51
}
47
52
48
53
type testPDS struct {
49
54
dir string
50
55
server *Server
56
+
plc *api.PLCServer
57
+
58
+
host string
59
+
60
+
shutdown func()
51
61
}
52
62
53
-
func setupPDS(t *testing.T, suffix string) *testPDS {
63
+
func (tp *testPDS) Cleanup() {
64
+
if tp.shutdown != nil {
65
+
tp.shutdown()
66
+
}
67
+
68
+
if tp.dir != "" {
69
+
_ = os.RemoveAll(tp.dir)
70
+
}
71
+
}
72
+
73
+
func setupPDS(t *testing.T, host, suffix string) *testPDS {
54
74
dir, err := ioutil.TempDir("", "fedtest")
55
75
if err != nil {
56
76
t.Fatal(err)
···
79
99
kfile := filepath.Join(dir, "server.key")
80
100
makeKey(t, kfile)
81
101
82
-
srv, err := NewServer(maindb, cs, kfile, suffix)
102
+
plc := &api.PLCServer{
103
+
Host: "http://localhost:2582",
104
+
}
105
+
106
+
srv, err := NewServer(maindb, cs, kfile, suffix, host, plc, []byte(host+suffix))
83
107
if err != nil {
84
108
t.Fatal(err)
85
109
}
···
87
111
return &testPDS{
88
112
dir: dir,
89
113
server: srv,
114
+
host: host,
90
115
}
91
116
}
92
117
118
+
func (tp *testPDS) Run(t *testing.T) {
119
+
// TODO: rig this up so it t.Fatals if the RunAPI call fails immediately
120
+
go func() {
121
+
if err := tp.server.RunAPI(tp.host); err != nil {
122
+
fmt.Println(err)
123
+
}
124
+
}()
125
+
time.Sleep(time.Millisecond * 10)
126
+
127
+
tp.shutdown = func() {
128
+
tp.server.echo.Shutdown(context.TODO())
129
+
}
130
+
}
131
+
132
+
type testUser struct {
133
+
handle string
134
+
pds *testPDS
135
+
did string
136
+
137
+
client *xrpc.Client
138
+
}
139
+
140
+
func (tp *testPDS) PeerWith(t *testing.T, op *testPDS) {
141
+
panic("no")
142
+
}
143
+
144
+
func (tp *testPDS) NewUser(t *testing.T, handle string) *testUser {
145
+
ctx := context.TODO()
146
+
147
+
c := &xrpc.Client{
148
+
Host: "http://" + tp.host,
149
+
}
150
+
151
+
out, err := atproto.AccountCreate(ctx, c, &atproto.AccountCreate_Input{
152
+
Email: handle + "@fake.com",
153
+
Handle: handle,
154
+
Password: "password",
155
+
})
156
+
if err != nil {
157
+
t.Fatal(err)
158
+
}
159
+
160
+
c.Auth = &xrpc.AuthInfo{
161
+
AccessJwt: out.AccessJwt,
162
+
RefreshJwt: out.RefreshJwt,
163
+
Handle: out.Handle,
164
+
Did: out.Did,
165
+
}
166
+
167
+
return &testUser{
168
+
pds: tp,
169
+
handle: out.Handle,
170
+
client: c,
171
+
did: out.Did,
172
+
}
173
+
}
174
+
175
+
func (u *testUser) Post(t *testing.T, body string) string {
176
+
t.Helper()
177
+
178
+
ctx := context.TODO()
179
+
resp, err := atproto.RepoCreateRecord(ctx, u.client, &atproto.RepoCreateRecord_Input{
180
+
Collection: "app.bsky.feed.post",
181
+
Did: u.did,
182
+
Record: &bsky.FeedPost{
183
+
CreatedAt: time.Now().Format(time.RFC3339),
184
+
Text: body,
185
+
},
186
+
})
187
+
188
+
if err != nil {
189
+
t.Fatal(err)
190
+
}
191
+
192
+
return resp.Uri
193
+
}
194
+
93
195
func TestBasicFederation(t *testing.T) {
196
+
p1 := setupPDS(t, "localhost:8812", ".pdsone")
197
+
p2 := setupPDS(t, "localhost:8813", ".pdstwo")
198
+
199
+
defer p1.Cleanup()
200
+
defer p2.Cleanup()
201
+
202
+
p1.Run(t)
203
+
p2.Run(t)
204
+
205
+
bob := p1.NewUser(t, "bob.pdsone")
206
+
laura := p2.NewUser(t, "laura.pdstwo")
207
+
208
+
//p1.PeerWith(p2)
209
+
bob.Post(t, "hello world")
210
+
laura.Post(t, "hello bob")
94
211
95
212
}
+11
-2
server/fedmgr.go
+11
-2
server/fedmgr.go
···
12
12
)
13
13
14
14
type FederationManager struct {
15
-
indexCallback func(context.Context, string, *Event) error
15
+
indexCallback IndexCallback
16
+
}
17
+
18
+
type IndexCallback func(context.Context, string, *Event) error
19
+
20
+
func NewFederationManager(cb IndexCallback) *FederationManager {
21
+
return &FederationManager{
22
+
indexCallback: cb,
23
+
}
16
24
}
17
25
18
26
func (fm *FederationManager) SubscribeToPds(ctx context.Context, host string) error {
···
32
40
time.Sleep(sleepForBackoff(backoff))
33
41
backoff++
34
42
}
35
-
_ = res
43
+
44
+
fmt.Println("event subscription response code: ", res.StatusCode)
36
45
37
46
if err := fm.handleConnection(host, con); err != nil {
38
47
log.Printf("connection to %q failed: %s", host, err)
+2
-2
server/handlers.go
+2
-2
server/handlers.go
···
419
419
return nil, err
420
420
}
421
421
422
-
d, err := s.fakeDid.NewForHandle(input.Handle)
422
+
d, err := s.plc.CreateDID(ctx, s.signingKey, recoveryKey, input.Handle, s.serviceUrl)
423
423
if err != nil {
424
-
return nil, err
424
+
return nil, fmt.Errorf("create did: %w", err)
425
425
}
426
426
427
427
u.DID = d
+10
server/indexer.go
+10
server/indexer.go
···
28
28
db.AutoMigrate(&FollowRecord{})
29
29
db.AutoMigrate(&VoteRecord{})
30
30
db.AutoMigrate(&RepostRecord{})
31
+
db.AutoMigrate(&ExternalFollow{})
31
32
32
33
return &Indexer{
33
34
db: db,
···
109
110
Cid string
110
111
}
111
112
113
+
type ExternalFollow struct {
114
+
gorm.Model
115
+
PDS uint
116
+
User uint
117
+
}
118
+
112
119
func (ix *Indexer) catchup(ctx context.Context, evt *repomgr.RepoEvent) error {
113
120
// TODO: catch up on events that happened since this event (in the event of a crash or downtime)
114
121
return nil
···
136
143
default:
137
144
log.Println("unrecognized repo event type: ", evt.Kind)
138
145
}
146
+
}
139
147
148
+
func (ix *Indexer) handleFedEvent(ctx context.Context, host string, evt *Event) error {
149
+
panic("TODO")
140
150
}
141
151
142
152
func (ix *Indexer) handleRecordCreate(ctx context.Context, evt *repomgr.RepoEvent, local bool) error {
+58
-23
server/server.go
+58
-23
server/server.go
···
3
3
import (
4
4
"context"
5
5
"crypto/ecdsa"
6
-
"crypto/elliptic"
7
6
"encoding/json"
8
7
"fmt"
9
8
"log"
···
16
15
"github.com/ipfs/go-cid"
17
16
"github.com/labstack/echo/v4"
18
17
"github.com/labstack/echo/v4/middleware"
18
+
"github.com/lestrrat-go/jwx/jwa"
19
19
jwk "github.com/lestrrat-go/jwx/jwk"
20
20
jwt "github.com/lestrrat-go/jwx/jwt"
21
+
"github.com/whyrusleeping/go-did"
21
22
comatprototypes "github.com/whyrusleeping/gosky/api/atproto"
22
23
appbskytypes "github.com/whyrusleeping/gosky/api/bsky"
23
24
"github.com/whyrusleeping/gosky/carstore"
25
+
"github.com/whyrusleeping/gosky/key"
24
26
"github.com/whyrusleeping/gosky/lex/util"
25
27
"github.com/whyrusleeping/gosky/repomgr"
26
28
"github.com/whyrusleeping/gosky/xrpc"
···
28
30
)
29
31
30
32
type Server struct {
31
-
db *gorm.DB
32
-
cs *carstore.CarStore
33
-
repoman *repomgr.RepoManager
34
-
feedgen *FeedGenerator
35
-
notifman *NotificationManager
36
-
indexer *Indexer
37
-
events *EventManager
38
-
signingKey []byte
33
+
db *gorm.DB
34
+
cs *carstore.CarStore
35
+
repoman *repomgr.RepoManager
36
+
feedgen *FeedGenerator
37
+
notifman *NotificationManager
38
+
indexer *Indexer
39
+
events *EventManager
40
+
fedmgr *FederationManager
41
+
signingKey *key.Key
42
+
echo *echo.Echo
43
+
jwtSigningKey []byte
39
44
40
45
handleSuffix string
46
+
serviceUrl string
41
47
42
-
fakeDid *FakeDid
48
+
plc PLCClient
43
49
}
44
50
45
51
const UserActorDeclCid = "bafyreid27zk7lbis4zw5fz4podbvbs4fc5ivwji3dmrwa6zggnj4bnd57u"
46
52
const UserActorDeclType = "app.bsky.system.actorUser"
47
53
48
-
func NewServer(db *gorm.DB, cs *carstore.CarStore, kfile string, handleSuffix string) (*Server, error) {
54
+
type PLCClient interface {
55
+
GetDocument(ctx context.Context, didstr string) (*did.Document, error)
56
+
CreateDID(ctx context.Context, sigkey *key.Key, recovery string, handle string, service string) (string, error)
57
+
}
58
+
59
+
func NewServer(db *gorm.DB, cs *carstore.CarStore, kfile string, handleSuffix, serviceUrl string, plc PLCClient, jwtkey []byte) (*Server, error) {
49
60
db.AutoMigrate(&User{})
50
61
51
62
serkey, err := loadKey(kfile)
···
64
75
}
65
76
66
77
s := &Server{
67
-
signingKey: serkey,
68
-
db: db,
69
-
cs: cs,
70
-
notifman: notifman,
71
-
indexer: ix,
72
-
fakeDid: NewFakeDid(db),
73
-
events: evtman,
74
-
repoman: repoman,
75
-
handleSuffix: handleSuffix,
78
+
signingKey: serkey,
79
+
db: db,
80
+
cs: cs,
81
+
notifman: notifman,
82
+
indexer: ix,
83
+
plc: NewFakeDid(db),
84
+
events: evtman,
85
+
repoman: repoman,
86
+
handleSuffix: handleSuffix,
87
+
serviceUrl: serviceUrl,
88
+
jwtSigningKey: jwtkey,
76
89
}
90
+
91
+
s.fedmgr = NewFederationManager(s.handleFedEvent)
77
92
78
93
repoman.SetEventHandler(func(ctx context.Context, evt *repomgr.RepoEvent) {
79
94
ix.HandleRepoEvent(ctx, evt)
···
98
113
99
114
s.feedgen = feedgen
100
115
116
+
go evtman.Run()
117
+
101
118
return s, nil
102
119
}
103
120
121
+
func (s *Server) handleFedEvent(ctx context.Context, host string, evt *Event) error {
122
+
switch evt.Kind {
123
+
case EvtKindCreateRecord:
124
+
case EvtKindUpdateRecord:
125
+
126
+
}
127
+
panic("nyi")
128
+
}
129
+
104
130
func (s *Server) repoEventToFedEvent(ctx context.Context, evt *repomgr.RepoEvent) (*Event, error) {
105
131
out := &Event{
106
132
CarSlice: evt.RepoSlice,
···
143
169
return util.CborDecodeValue(blk.RawData())
144
170
}
145
171
146
-
func loadKey(kfile string) ([]byte, error) {
172
+
func loadKey(kfile string) (*key.Key, error) {
147
173
kb, err := os.ReadFile(kfile)
148
174
if err != nil {
149
175
return nil, err
···
158
184
if err := sk.Raw(&spk); err != nil {
159
185
return nil, err
160
186
}
161
-
return elliptic.Marshal(spk.Curve, spk.X, spk.Y), nil
187
+
curve, ok := sk.Get("crv")
188
+
if !ok {
189
+
return nil, fmt.Errorf("need a curve set")
190
+
}
191
+
192
+
return &key.Key{
193
+
Raw: &spk,
194
+
Type: string(curve.(jwa.EllipticCurveAlgorithm)),
195
+
}, nil
162
196
}
163
197
164
198
func (s *Server) RunAPI(listen string) error {
165
199
e := echo.New()
200
+
s.echo = e
166
201
e.Use(middleware.LoggerWithConfig(middleware.LoggerConfig{
167
202
Format: "method=${method}, uri=${uri}, status=${status} latency=${latency_human}\n",
168
203
}))
···
181
216
}
182
217
},
183
218
//KeyFunc: s.getKey,
184
-
SigningKey: s.signingKey,
219
+
SigningKey: s.jwtSigningKey,
185
220
}
186
221
187
222
e.HTTPErrorHandler = func(err error, ctx echo.Context) {