···

       
4
       
 
       
	Context             []string                   `json:"@context"`

     

       
5
       
 
       
	Id                  string                     `json:"id"`

     

       
6
       
 
       
	AlsoKnownAs         []string                   `json:"alsoKnownAs"`

     

       
7
       
-
       
	VerificationMethods []DidDocVerificationMethod `json:"verificationMethod"`

     

       
8
       
 
       
	Service             []DidDocService            `json:"service"`

     

       
9
       
 
       
}

     

       
10
       
 
       


     

···

       
4
       
 
       
	Context             []string                   `json:"@context"`

     

       
5
       
 
       
	Id                  string                     `json:"id"`

     

       
6
       
 
       
	AlsoKnownAs         []string                   `json:"alsoKnownAs"`

     

       
7
       
+
       
	VerificationMethods []DidDocVerificationMethod `json:"verificationMethods"`

     

       
8
       
 
       
	Service             []DidDocService            `json:"service"`

     

       
9
       
 
       
}

     

       
10
       
 
       


     

···

       
25
       
 
       
	return db.cli.Clauses(clauses...).Create(value)

     

       
26
       
 
       
}

     

       
27
       
 
       


     

       
28
       
-
       
func (db *DB) Save(value any, clauses []clause.Expression) *gorm.DB {

     

       
29
       
-
       
	db.mu.Lock()

     

       
30
       
-
       
	defer db.mu.Unlock()

     

       
31
       
-
       
	return db.cli.Clauses(clauses...).Save(value)

     

       
32
       
-
       
}

     

       
33
       
-
       


     

       
34
       
 
       
func (db *DB) Exec(sql string, clauses []clause.Expression, values ...any) *gorm.DB {

     

       
35
       
 
       
	db.mu.Lock()

     

       
36
       
 
       
	defer db.mu.Unlock()

     

···

       
25
       
 
       
	return db.cli.Clauses(clauses...).Create(value)

     

       
26
       
 
       
}

     

       
27
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
28
       
 
       
func (db *DB) Exec(sql string, clauses []clause.Expression, values ...any) *gorm.DB {

     

       
29
       
 
       
	db.mu.Lock()

     

       
30
       
 
       
	defer db.mu.Unlock()

     

···

       
32
       
 
       
	return genericError(e, 400, msg)

     

       
33
       
 
       
}

     

       
34
       
 
       


     

       
35
       
-
       
func UnauthorizedError(e echo.Context, suffix *string) error {

     

       
36
       
-
       
	msg := "Unauthorized"

     

       
37
       
-
       
	if suffix != nil {

     

       
38
       
-
       
		msg += ". " + *suffix

     

       
39
       
-
       
	}

     

       
40
       
-
       
	return genericError(e, 401, msg)

     

       
41
       
-
       
}

     

       
42
       
-
       


     

       
43
       
-
       
func ForbiddenError(e echo.Context, suffix *string) error {

     

       
44
       
-
       
	msg := "Forbidden"

     

       
45
       
-
       
	if suffix != nil {

     

       
46
       
-
       
		msg += ". " + *suffix

     

       
47
       
-
       
	}

     

       
48
       
-
       
	return genericError(e, 403, msg)

     

       
49
       
-
       
}

     

       
50
       
-
       


     

       
51
       
 
       
func InvalidTokenError(e echo.Context) error {

     

       
52
       
 
       
	return InputError(e, to.StringPtr("InvalidToken"))

     

       
53
       
 
       
}

     

···

       
32
       
 
       
	return genericError(e, 400, msg)

     

       
33
       
 
       
}

     

       
34
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
35
       
 
       
func InvalidTokenError(e echo.Context) error {

     

       
36
       
 
       
	return InputError(e, to.StringPtr("InvalidToken"))

     

       
37
       
 
       
}

     

···

       
106
       
 
       
	Did       string `gorm:"index;index:idx_blob_did_cid"`

     

       
107
       
 
       
	Cid       []byte `gorm:"index;index:idx_blob_did_cid"`

     

       
108
       
 
       
	RefCount  int

     

       
109
       
-
       
	Storage   string `gorm:"default:sqlite"`

     

       
110
       
 
       
}

     

       
111
       
 
       


     

       
112
       
 
       
type BlobPart struct {

     

···

       
106
       
 
       
	Did       string `gorm:"index;index:idx_blob_did_cid"`

     

       
107
       
 
       
	Cid       []byte `gorm:"index;index:idx_blob_did_cid"`

     

       
108
       
 
       
	RefCount  int

     

       
109
       
+
       
	Storage   string `gorm:"default:sqlite;check:storage in ('sqlite', 's3')"`

     

       
110
       
 
       
}

     

       
111
       
 
       


     

       
112
       
 
       
type BlobPart struct {

     

···

       
10
       
 
       
	"github.com/Azure/go-autorest/autorest/to"

     

       
11
       
 
       
	"github.com/bluesky-social/indigo/api/atproto"

     

       
12
       
 
       
	"github.com/bluesky-social/indigo/atproto/atcrypto"

     

       
0
       
 
       

     

       
13
       
 
       
	"github.com/bluesky-social/indigo/events"

     

       
14
       
 
       
	"github.com/bluesky-social/indigo/repo"

     

       
15
       
 
       
	"github.com/bluesky-social/indigo/util"

     
···

       
38
       
 
       
func (s *Server) handleCreateAccount(e echo.Context) error {

     

       
39
       
 
       
	var request ComAtprotoServerCreateAccountRequest

     

       
40
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
41
       
 
       
	if err := e.Bind(&request); err != nil {

     

       
42
       
 
       
		s.logger.Error("error receiving request", "endpoint", "com.atproto.server.createAccount", "error", err)

     

       
43
       
 
       
		return helpers.ServerError(e, nil)

     
···

       
68
       
 
       
			}

     

       
69
       
 
       
		}

     

       
70
       
 
       
	}

     

       
71
       
-
       
	

     

       
72
       
-
       
	var signupDid string

     

       
73
       
-
       
	if request.Did != nil {

     

       
74
       
-
       
		signupDid = *request.Did;

     

       
75
       
-
       
		

     

       
76
       
-
       
		token := strings.TrimSpace(strings.Replace(e.Request().Header.Get("authorization"), "Bearer ", "", 1))

     

       
77
       
-
       
		if token == "" {

     

       
78
       
-
       
			return helpers.UnauthorizedError(e, to.StringPtr("must authenticate to use an existing did"))

     

       
79
       
-
       
		}

     

       
80
       
-
       
		authDid, err := s.validateServiceAuth(e.Request().Context(), token, "com.atproto.server.createAccount")

     

       
81
       
-
       


     

       
82
       
-
       
		if err != nil {

     

       
83
       
-
       
			s.logger.Warn("error validating authorization token", "endpoint", "com.atproto.server.createAccount", "error", err)

     

       
84
       
-
       
			return helpers.UnauthorizedError(e, to.StringPtr("invalid authorization token"))

     

       
85
       
-
       
		}

     

       
86
       
-
       


     

       
87
       
-
       
		if authDid != signupDid {

     

       
88
       
-
       
			return helpers.ForbiddenError(e, to.StringPtr("auth did did not match signup did"))

     

       
89
       
-
       
		}

     

       
90
       
-
       
	}

     

       
91
       
 
       


     

       
92
       
 
       
	// see if the handle is already taken

     

       
93
       
-
       
	actor, err := s.getActorByHandle(request.Handle)

     

       
94
       
 
       
	if err != nil && err != gorm.ErrRecordNotFound {

     

       
95
       
 
       
		s.logger.Error("error looking up handle in db", "endpoint", "com.atproto.server.createAccount", "error", err)

     

       
96
       
 
       
		return helpers.ServerError(e, nil)

     

       
97
       
 
       
	}

     

       
98
       
-
       
	if err == nil && actor.Did != signupDid {

     

       
99
       
 
       
		return helpers.InputError(e, to.StringPtr("HandleNotAvailable"))

     

       
100
       
 
       
	}

     

       
101
       
 
       


     

       
102
       
-
       
	if did, err := s.passport.ResolveHandle(e.Request().Context(), request.Handle); err == nil && did != signupDid {

     

       
103
       
 
       
		return helpers.InputError(e, to.StringPtr("HandleNotAvailable"))

     

       
104
       
 
       
	}

     

       
105
       
 
       


     
···

       
117
       
 
       
	}

     

       
118
       
 
       


     

       
119
       
 
       
	// see if the email is already taken

     

       
120
       
-
       
	existingRepo, err := s.getRepoByEmail(request.Email)

     

       
121
       
 
       
	if err != nil && err != gorm.ErrRecordNotFound {

     

       
122
       
 
       
		s.logger.Error("error looking up email in db", "endpoint", "com.atproto.server.createAccount", "error", err)

     

       
123
       
 
       
		return helpers.ServerError(e, nil)

     

       
124
       
 
       
	}

     

       
125
       
-
       
	if err == nil && existingRepo.Did != signupDid {

     

       
126
       
 
       
		return helpers.InputError(e, to.StringPtr("EmailNotAvailable"))

     

       
127
       
 
       
	}

     

       
128
       
 
       


     
···

       
163
       
 
       
		SigningKey:            k.Bytes(),

     

       
164
       
 
       
	}

     

       
165
       
 
       


     

       
166
       
-
       
	if actor == nil {

     

       
167
       
-
       
		actor = &models.Actor{

     

       
168
       
-
       
			Did:    signupDid,

     

       
169
       
-
       
			Handle: request.Handle,

     

       
170
       
-
       
		}

     

       
171
       
 
       


     

       
172
       
-
       
		if err := s.db.Create(&urepo, nil).Error; err != nil {

     

       
173
       
-
       
			s.logger.Error("error inserting new repo", "error", err)

     

       
174
       
-
       
			return helpers.ServerError(e, nil)

     

       
175
       
-
       
		}

     

       
176
       
-
       
	

     

       
177
       
-
       
		if err := s.db.Create(&actor, nil).Error; err != nil {

     

       
178
       
-
       
			s.logger.Error("error inserting new actor", "error", err)

     

       
179
       
-
       
			return helpers.ServerError(e, nil)

     

       
180
       
-
       
		}

     

       
181
       
-
       
	} else {

     

       
182
       
-
       
		if err := s.db.Save(&actor, nil).Error; err != nil {

     

       
183
       
-
       
			s.logger.Error("error inserting new actor", "error", err)

     

       
184
       
-
       
			return helpers.ServerError(e, nil)

     

       
185
       
-
       
		}

     

       
186
       
 
       
	}

     

       
187
       
 
       


     

       
188
       
-
       
	if request.Did == nil || *request.Did == "" {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
189
       
 
       
		bs := s.getBlockstore(signupDid)

     

       
190
       
 
       
		r := repo.NewRepo(context.TODO(), signupDid, bs)

     

       
191
       
 
       


     

···

       
10
       
 
       
	"github.com/Azure/go-autorest/autorest/to"

     

       
11
       
 
       
	"github.com/bluesky-social/indigo/api/atproto"

     

       
12
       
 
       
	"github.com/bluesky-social/indigo/atproto/atcrypto"

     

       
13
       
+
       
	"github.com/bluesky-social/indigo/atproto/syntax"

     

       
14
       
 
       
	"github.com/bluesky-social/indigo/events"

     

       
15
       
 
       
	"github.com/bluesky-social/indigo/repo"

     

       
16
       
 
       
	"github.com/bluesky-social/indigo/util"

     
···

       
39
       
 
       
func (s *Server) handleCreateAccount(e echo.Context) error {

     

       
40
       
 
       
	var request ComAtprotoServerCreateAccountRequest

     

       
41
       
 
       


     

       
42
       
+
       
	var signupDid string

     

       
43
       
+
       
	customDidHeader := e.Request().Header.Get("authorization")

     

       
44
       
+
       
	if customDidHeader != "" {

     

       
45
       
+
       
		pts := strings.Split(customDidHeader, " ")

     

       
46
       
+
       
		if len(pts) != 2 {

     

       
47
       
+
       
			return helpers.InputError(e, to.StringPtr("InvalidDid"))

     

       
48
       
+
       
		}

     

       
49
       
+
       


     

       
50
       
+
       
		_, err := syntax.ParseDID(pts[1])

     

       
51
       
+
       
		if err != nil {

     

       
52
       
+
       
			return helpers.InputError(e, to.StringPtr("InvalidDid"))

     

       
53
       
+
       
		}

     

       
54
       
+
       


     

       
55
       
+
       
		signupDid = pts[1]

     

       
56
       
+
       
	}

     

       
57
       
+
       


     

       
58
       
 
       
	if err := e.Bind(&request); err != nil {

     

       
59
       
 
       
		s.logger.Error("error receiving request", "endpoint", "com.atproto.server.createAccount", "error", err)

     

       
60
       
 
       
		return helpers.ServerError(e, nil)

     
···

       
85
       
 
       
			}

     

       
86
       
 
       
		}

     

       
87
       
 
       
	}

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
88
       
 
       


     

       
89
       
 
       
	// see if the handle is already taken

     

       
90
       
+
       
	_, err := s.getActorByHandle(request.Handle)

     

       
91
       
 
       
	if err != nil && err != gorm.ErrRecordNotFound {

     

       
92
       
 
       
		s.logger.Error("error looking up handle in db", "endpoint", "com.atproto.server.createAccount", "error", err)

     

       
93
       
 
       
		return helpers.ServerError(e, nil)

     

       
94
       
 
       
	}

     

       
95
       
+
       
	if err == nil {

     

       
96
       
 
       
		return helpers.InputError(e, to.StringPtr("HandleNotAvailable"))

     

       
97
       
 
       
	}

     

       
98
       
 
       


     

       
99
       
+
       
	if did, err := s.passport.ResolveHandle(e.Request().Context(), request.Handle); err == nil && did != "" {

     

       
100
       
 
       
		return helpers.InputError(e, to.StringPtr("HandleNotAvailable"))

     

       
101
       
 
       
	}

     

       
102
       
 
       


     
···

       
114
       
 
       
	}

     

       
115
       
 
       


     

       
116
       
 
       
	// see if the email is already taken

     

       
117
       
+
       
	_, err = s.getRepoByEmail(request.Email)

     

       
118
       
 
       
	if err != nil && err != gorm.ErrRecordNotFound {

     

       
119
       
 
       
		s.logger.Error("error looking up email in db", "endpoint", "com.atproto.server.createAccount", "error", err)

     

       
120
       
 
       
		return helpers.ServerError(e, nil)

     

       
121
       
 
       
	}

     

       
122
       
+
       
	if err == nil {

     

       
123
       
 
       
		return helpers.InputError(e, to.StringPtr("EmailNotAvailable"))

     

       
124
       
 
       
	}

     

       
125
       
 
       


     
···

       
160
       
 
       
		SigningKey:            k.Bytes(),

     

       
161
       
 
       
	}

     

       
162
       
 
       


     

       
163
       
+
       
	actor := models.Actor{

     

       
164
       
+
       
		Did:    signupDid,

     

       
165
       
+
       
		Handle: request.Handle,

     

       
166
       
+
       
	}

     

       
0
       
 
       

     

       
167
       
 
       


     

       
168
       
+
       
	if err := s.db.Create(&urepo, nil).Error; err != nil {

     

       
169
       
+
       
		s.logger.Error("error inserting new repo", "error", err)

     

       
170
       
+
       
		return helpers.ServerError(e, nil)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
171
       
 
       
	}

     

       
172
       
 
       


     

       
173
       
+
       
	if err := s.db.Create(&actor, nil).Error; err != nil {

     

       
174
       
+
       
		s.logger.Error("error inserting new actor", "error", err)

     

       
175
       
+
       
		return helpers.ServerError(e, nil)

     

       
176
       
+
       
	}

     

       
177
       
+
       


     

       
178
       
+
       
	if customDidHeader == "" {

     

       
179
       
 
       
		bs := s.getBlockstore(signupDid)

     

       
180
       
 
       
		r := repo.NewRepo(context.TODO(), signupDid, bs)

     

       
181
       
 
       


     

···

       
64
       
 
       
		for _, p := range parts {

     

       
65
       
 
       
			buf.Write(p.Data)

     

       
66
       
 
       
		}

     

       
67
       
-
       
	} else if blob.Storage == "s3" {

     

       
68
       
-
       
		if  !(s.s3Config != nil && s.s3Config.BlobstoreEnabled) {

     

       
69
       
-
       
			s.logger.Error("s3 storage disabled")

     

       
70
       
-
       
			return helpers.ServerError(e, nil)

     

       
71
       
-
       
		}

     

       
72
       
-
       


     

       
73
       
 
       
		config := &aws.Config{

     

       
74
       
 
       
			Region:      aws.String(s.s3Config.Region),

     

       
75
       
 
       
			Credentials: credentials.NewStaticCredentials(s.s3Config.AccessKey, s.s3Config.SecretKey, ""),

     

···

       
64
       
 
       
		for _, p := range parts {

     

       
65
       
 
       
			buf.Write(p.Data)

     

       
66
       
 
       
		}

     

       
67
       
+
       
	} else if blob.Storage == "s3" && s.s3Config != nil && s.s3Config.BlobstoreEnabled {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
68
       
 
       
		config := &aws.Config{

     

       
69
       
 
       
			Region:      aws.String(s.s3Config.Region),

     

       
70
       
 
       
			Credentials: credentials.NewStaticCredentials(s.s3Config.AccessKey, s.s3Config.SecretKey, ""),

     

···

       
1
       
 
       
package server

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
4
       
-
       
	"context"

     

       
5
       
-
       
	"time"

     

       
6
       
 
       


     

       
7
       
 
       
	"github.com/bluesky-social/indigo/events"

     

       
8
       
 
       
	"github.com/bluesky-social/indigo/lex/util"

     
···

       
11
       
 
       
)

     

       
12
       
 
       


     

       
13
       
 
       
func (s *Server) handleSyncSubscribeRepos(e echo.Context) error {

     

       
14
       
-
       
	ctx := e.Request().Context()

     

       
15
       
-
       
	logger := s.logger.With("component", "subscribe-repos-websocket")

     

       
16
       
-
       


     

       
17
       
 
       
	conn, err := websocket.Upgrade(e.Response().Writer, e.Request(), e.Response().Header(), 1<<10, 1<<10)

     

       
18
       
 
       
	if err != nil {

     

       
19
       
-
       
		logger.Error("unable to establish websocket with relay", "err", err)

     

       
20
       
 
       
		return err

     

       
21
       
 
       
	}

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
22
       
 
       


     

       
23
       
 
       
	ident := e.RealIP() + "-" + e.Request().UserAgent()

     

       
24
       
-
       
	logger = logger.With("ident", ident)

     

       
25
       
-
       
	logger.Info("new connection established")

     

       
26
       
 
       


     

       
27
       
 
       
	evts, cancel, err := s.evtman.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool {

     

       
28
       
 
       
		return true

     
···

       
36
       
 
       
	for evt := range evts {

     

       
37
       
 
       
		wc, err := conn.NextWriter(websocket.BinaryMessage)

     

       
38
       
 
       
		if err != nil {

     

       
39
       
-
       
			logger.Error("error writing message to relay", "err", err)

     

       
40
       
-
       
			break

     

       
41
       
-
       
		}

     

       
42
       
-
       


     

       
43
       
-
       
		if ctx.Err() != nil {

     

       
44
       
-
       
			logger.Error("context error", "err", err)

     

       
45
       
-
       
			break

     

       
46
       
 
       
		}

     

       
47
       
 
       


     

       
48
       
 
       
		var obj util.CBOR

     

       
0
       
 
       

     

       
49
       
 
       
		switch {

     

       
50
       
 
       
		case evt.Error != nil:

     

       
51
       
 
       
			header.Op = events.EvtKindErrorFrame

     
···

       
63
       
 
       
			header.MsgType = "#info"

     

       
64
       
 
       
			obj = evt.RepoInfo

     

       
65
       
 
       
		default:

     

       
66
       
-
       
			logger.Warn("unrecognized event kind")

     

       
67
       
-
       
			return nil

     

       
68
       
 
       
		}

     

       
69
       
 
       


     

       
70
       
 
       
		if err := header.MarshalCBOR(wc); err != nil {

     

       
71
       
-
       
			logger.Error("failed to write header to relay", "err", err)

     

       
72
       
-
       
			break

     

       
73
       
 
       
		}

     

       
74
       
 
       


     

       
75
       
 
       
		if err := obj.MarshalCBOR(wc); err != nil {

     

       
76
       
-
       
			logger.Error("failed to write event to relay", "err", err)

     

       
77
       
-
       
			break

     

       
78
       
 
       
		}

     

       
79
       
 
       


     

       
80
       
 
       
		if err := wc.Close(); err != nil {

     

       
81
       
-
       
			logger.Error("failed to flush-close our event write", "err", err)

     

       
82
       
-
       
			break

     

       
83
       
 
       
		}

     

       
84
       
-
       
	}

     

       
85
       
-
       


     

       
86
       
-
       
	// we should tell the relay to request a new crawl at this point if we got disconnected

     

       
87
       
-
       
	// use a new context since the old one might be cancelled at this point

     

       
88
       
-
       
	ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second)

     

       
89
       
-
       
	defer cancel()

     

       
90
       
-
       
	if err := s.requestCrawl(ctx); err != nil {

     

       
91
       
-
       
		logger.Error("error requesting crawls", "err", err)

     

       
92
       
 
       
	}

     

       
93
       
 
       


     

       
94
       
 
       
	return nil

     

···

       
1
       
 
       
package server

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
4
       
+
       
	"fmt"

     

       
0
       
 
       

     

       
5
       
 
       


     

       
6
       
 
       
	"github.com/bluesky-social/indigo/events"

     

       
7
       
 
       
	"github.com/bluesky-social/indigo/lex/util"

     
···

       
10
       
 
       
)

     

       
11
       
 
       


     

       
12
       
 
       
func (s *Server) handleSyncSubscribeRepos(e echo.Context) error {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
13
       
 
       
	conn, err := websocket.Upgrade(e.Response().Writer, e.Request(), e.Response().Header(), 1<<10, 1<<10)

     

       
14
       
 
       
	if err != nil {

     

       
0
       
 
       

     

       
15
       
 
       
		return err

     

       
16
       
 
       
	}

     

       
17
       
+
       


     

       
18
       
+
       
	s.logger.Info("new connection", "ua", e.Request().UserAgent())

     

       
19
       
+
       


     

       
20
       
+
       
	ctx := e.Request().Context()

     

       
21
       
 
       


     

       
22
       
 
       
	ident := e.RealIP() + "-" + e.Request().UserAgent()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
23
       
 
       


     

       
24
       
 
       
	evts, cancel, err := s.evtman.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool {

     

       
25
       
 
       
		return true

     
···

       
33
       
 
       
	for evt := range evts {

     

       
34
       
 
       
		wc, err := conn.NextWriter(websocket.BinaryMessage)

     

       
35
       
 
       
		if err != nil {

     

       
36
       
+
       
			return err

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
37
       
 
       
		}

     

       
38
       
 
       


     

       
39
       
 
       
		var obj util.CBOR

     

       
40
       
+
       


     

       
41
       
 
       
		switch {

     

       
42
       
 
       
		case evt.Error != nil:

     

       
43
       
 
       
			header.Op = events.EvtKindErrorFrame

     
···

       
55
       
 
       
			header.MsgType = "#info"

     

       
56
       
 
       
			obj = evt.RepoInfo

     

       
57
       
 
       
		default:

     

       
58
       
+
       
			return fmt.Errorf("unrecognized event kind")

     

       
0
       
 
       

     

       
59
       
 
       
		}

     

       
60
       
 
       


     

       
61
       
 
       
		if err := header.MarshalCBOR(wc); err != nil {

     

       
62
       
+
       
			return fmt.Errorf("failed to write header: %w", err)

     

       
0
       
 
       

     

       
63
       
 
       
		}

     

       
64
       
 
       


     

       
65
       
 
       
		if err := obj.MarshalCBOR(wc); err != nil {

     

       
66
       
+
       
			return fmt.Errorf("failed to write event: %w", err)

     

       
0
       
 
       

     

       
67
       
 
       
		}

     

       
68
       
 
       


     

       
69
       
 
       
		if err := wc.Close(); err != nil {

     

       
70
       
+
       
			return fmt.Errorf("failed to flush-close our event write: %w", err)

     

       
0
       
 
       

     

       
71
       
 
       
		}

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
72
       
 
       
	}

     

       
73
       
 
       


     

       
74
       
 
       
	return nil

     

···

       
78
       
 
       
	passport      *identity.Passport

     

       
79
       
 
       
	fallbackProxy string

     

       
80
       
 
       


     

       
81
       
-
       
	lastRequestCrawl time.Time

     

       
82
       
-
       
	requestCrawlMu   sync.Mutex

     

       
83
       
-
       


     

       
84
       
 
       
	dbName   string

     

       
85
       
 
       
	s3Config *S3Config

     

       
86
       
 
       
}

     
···

       
423
       
 
       
	// public

     

       
424
       
 
       
	s.echo.GET("/xrpc/com.atproto.identity.resolveHandle", s.handleResolveHandle)

     

       
425
       
 
       
	s.echo.POST("/xrpc/com.atproto.server.createAccount", s.handleCreateAccount)

     

       
0
       
 
       

     

       
426
       
 
       
	s.echo.POST("/xrpc/com.atproto.server.createSession", s.handleCreateSession)

     

       
427
       
 
       
	s.echo.GET("/xrpc/com.atproto.server.describeServer", s.handleDescribeServer)

     

       
428
       
 
       


     
···

       
521
       
 
       


     

       
522
       
 
       
	go s.backupRoutine()

     

       
523
       
 
       


     

       
524
       
-
       
	go func() {

     

       
525
       
-
       
		if err := s.requestCrawl(ctx); err != nil {

     

       
526
       
-
       
			s.logger.Error("error requesting crawls", "err", err)

     

       
527
       
-
       
		}

     

       
528
       
-
       
	}()

     

       
529
       
-
       


     

       
530
       
-
       
	<-ctx.Done()

     

       
531
       
-
       


     

       
532
       
-
       
	fmt.Println("shut down")

     

       
533
       
-
       


     

       
534
       
-
       
	return nil

     

       
535
       
-
       
}

     

       
536
       
-
       


     

       
537
       
-
       
func (s *Server) requestCrawl(ctx context.Context) error {

     

       
538
       
-
       
	logger := s.logger.With("component", "request-crawl")

     

       
539
       
-
       
	s.requestCrawlMu.Lock()

     

       
540
       
-
       
	defer s.requestCrawlMu.Unlock()

     

       
541
       
-
       


     

       
542
       
-
       
	logger.Info("requesting crawl with configured relays")

     

       
543
       
-
       


     

       
544
       
-
       
	if time.Now().Sub(s.lastRequestCrawl) <= 1*time.Minute {

     

       
545
       
-
       
		return fmt.Errorf("a crawl request has already been made within the last minute")

     

       
546
       
-
       
	}

     

       
547
       
-
       


     

       
548
       
 
       
	for _, relay := range s.config.Relays {

     

       
549
       
-
       
		logger := logger.With("relay", relay)

     

       
550
       
-
       
		logger.Info("requesting crawl from relay")

     

       
551
       
 
       
		cli := xrpc.Client{Host: relay}

     

       
552
       
-
       
		if err := atproto.SyncRequestCrawl(ctx, &cli, &atproto.SyncRequestCrawl_Input{

     

       
553
       
 
       
			Hostname: s.config.Hostname,

     

       
554
       
-
       
		}); err != nil {

     

       
555
       
-
       
			logger.Error("error requesting crawl", "err", err)

     

       
556
       
-
       
		} else {

     

       
557
       
-
       
			logger.Info("crawl requested successfully")

     

       
558
       
-
       
		}

     

       
559
       
 
       
	}

     

       
560
       
 
       


     

       
561
       
-
       
	s.lastRequestCrawl = time.Now()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
562
       
 
       


     

       
563
       
 
       
	return nil

     

       
564
       
 
       
}

     

···

       
78
       
 
       
	passport      *identity.Passport

     

       
79
       
 
       
	fallbackProxy string

     

       
80
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
81
       
 
       
	dbName   string

     

       
82
       
 
       
	s3Config *S3Config

     

       
83
       
 
       
}

     
···

       
420
       
 
       
	// public

     

       
421
       
 
       
	s.echo.GET("/xrpc/com.atproto.identity.resolveHandle", s.handleResolveHandle)

     

       
422
       
 
       
	s.echo.POST("/xrpc/com.atproto.server.createAccount", s.handleCreateAccount)

     

       
423
       
+
       
	s.echo.POST("/xrpc/com.atproto.server.createAccount", s.handleCreateAccount)

     

       
424
       
 
       
	s.echo.POST("/xrpc/com.atproto.server.createSession", s.handleCreateSession)

     

       
425
       
 
       
	s.echo.GET("/xrpc/com.atproto.server.describeServer", s.handleDescribeServer)

     

       
426
       
 
       


     
···

       
519
       
 
       


     

       
520
       
 
       
	go s.backupRoutine()

     

       
521
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
522
       
 
       
	for _, relay := range s.config.Relays {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
523
       
 
       
		cli := xrpc.Client{Host: relay}

     

       
524
       
+
       
		atproto.SyncRequestCrawl(ctx, &cli, &atproto.SyncRequestCrawl_Input{

     

       
525
       
 
       
			Hostname: s.config.Hostname,

     

       
526
       
+
       
		})

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
527
       
 
       
	}

     

       
528
       
 
       


     

       
529
       
+
       
	<-ctx.Done()

     

       
530
       
+
       


     

       
531
       
+
       
	fmt.Println("shut down")

     

       
532
       
 
       


     

       
533
       
 
       
	return nil

     

       
534
       
 
       
}

     

···

       
1
       
-
       
package server

     

       
2
       
-
       


     

       
3
       
-
       
import (

     

       
4
       
-
       
	"context"

     

       
5
       
-
       
	"fmt"

     

       
6
       
-
       
	"strings"

     

       
7
       
-
       


     

       
8
       
-
       
	"github.com/bluesky-social/indigo/atproto/atcrypto"

     

       
9
       
-
       
	"github.com/bluesky-social/indigo/atproto/identity"

     

       
10
       
-
       
	atproto_identity "github.com/bluesky-social/indigo/atproto/identity"

     

       
11
       
-
       
	"github.com/bluesky-social/indigo/atproto/syntax"

     

       
12
       
-
       
	"github.com/golang-jwt/jwt/v4"

     

       
13
       
-
       
)

     

       
14
       
-
       


     

       
15
       
-
       
type ES256KSigningMethod struct {

     

       
16
       
-
       
	alg string

     

       
17
       
-
       
}

     

       
18
       
-
       


     

       
19
       
-
       
func (m *ES256KSigningMethod) Alg() string {

     

       
20
       
-
       
	return m.alg

     

       
21
       
-
       
}

     

       
22
       
-
       


     

       
23
       
-
       
func (m *ES256KSigningMethod) Verify(signingString string, signature string, key interface{}) error {

     

       
24
       
-
       
	signatureBytes, err := jwt.DecodeSegment(signature)

     

       
25
       
-
       
	if err != nil {

     

       
26
       
-
       
		return err

     

       
27
       
-
       
	}

     

       
28
       
-
       
	return key.(atcrypto.PublicKey).HashAndVerifyLenient([]byte(signingString), signatureBytes)

     

       
29
       
-
       
}

     

       
30
       
-
       


     

       
31
       
-
       
func (m *ES256KSigningMethod) Sign(signingString string, key interface{}) (string, error) {

     

       
32
       
-
       
	return "", fmt.Errorf("unimplemented")

     

       
33
       
-
       
}

     

       
34
       
-
       


     

       
35
       
-
       
func init() {

     

       
36
       
-
       
	ES256K := ES256KSigningMethod{alg: "ES256K"}

     

       
37
       
-
       
	jwt.RegisterSigningMethod(ES256K.Alg(), func() jwt.SigningMethod {

     

       
38
       
-
       
		return &ES256K

     

       
39
       
-
       
	})

     

       
40
       
-
       
}

     

       
41
       
-
       


     

       
42
       
-
       
func (s *Server) validateServiceAuth(ctx context.Context, rawToken string, nsid string) (string, error) {

     

       
43
       
-
       
	token := strings.TrimSpace(rawToken)

     

       
44
       
-
       


     

       
45
       
-
       
	parsedToken, err := jwt.ParseWithClaims(token, jwt.MapClaims{}, func(token *jwt.Token) (interface{}, error) {

     

       
46
       
-
       
		did := syntax.DID(token.Claims.(jwt.MapClaims)["iss"].(string))

     

       
47
       
-
       
		didDoc, err := s.passport.FetchDoc(ctx, did.String());

     

       
48
       
-
       
		if err != nil {

     

       
49
       
-
       
			return nil, fmt.Errorf("unable to resolve did %s: %s", did, err)

     

       
50
       
-
       
		}

     

       
51
       
-
       


     

       
52
       
-
       
		verificationMethods := make([]atproto_identity.DocVerificationMethod, len(didDoc.VerificationMethods))

     

       
53
       
-
       
		for i, verificationMethod := range didDoc.VerificationMethods {

     

       
54
       
-
       
			verificationMethods[i] = atproto_identity.DocVerificationMethod{

     

       
55
       
-
       
				ID: verificationMethod.Id,

     

       
56
       
-
       
				Type: verificationMethod.Type,

     

       
57
       
-
       
				PublicKeyMultibase: verificationMethod.PublicKeyMultibase,

     

       
58
       
-
       
				Controller: verificationMethod.Controller,

     

       
59
       
-
       
			}

     

       
60
       
-
       
		}

     

       
61
       
-
       
		services := make([]atproto_identity.DocService, len(didDoc.Service))

     

       
62
       
-
       
		for i, service := range didDoc.Service {

     

       
63
       
-
       
			services[i] = atproto_identity.DocService{

     

       
64
       
-
       
				ID: service.Id,

     

       
65
       
-
       
				Type: service.Type,

     

       
66
       
-
       
				ServiceEndpoint: service.ServiceEndpoint,

     

       
67
       
-
       
			}

     

       
68
       
-
       
		}

     

       
69
       
-
       
		parsedIdentity := atproto_identity.ParseIdentity(&identity.DIDDocument{

     

       
70
       
-
       
			DID: did,

     

       
71
       
-
       
			AlsoKnownAs: didDoc.AlsoKnownAs,

     

       
72
       
-
       
			VerificationMethod: verificationMethods,

     

       
73
       
-
       
			Service: services,

     

       
74
       
-
       
		})

     

       
75
       
-
       


     

       
76
       
-
       
		key, err := parsedIdentity.PublicKey()

     

       
77
       
-
       
		if err != nil {

     

       
78
       
-
       
			return nil, fmt.Errorf("signing key not found for did %s: %s", did, err)

     

       
79
       
-
       
		}

     

       
80
       
-
       
		return key, nil

     

       
81
       
-
       
	})

     

       
82
       
-
       
	if err != nil {

     

       
83
       
-
       
		return "", fmt.Errorf("invalid token: %s", err)

     

       
84
       
-
       
	}

     

       
85
       
-
       


     

       
86
       
-
       
	claims := parsedToken.Claims.(jwt.MapClaims)

     

       
87
       
-
       
	if claims["lxm"] != nsid {

     

       
88
       
-
       
		return "", fmt.Errorf("bad jwt lexicon method (\"lxm\"). must match: %s", nsid)

     

       
89
       
-
       
	}

     

       
90
       
-
       
	return claims["iss"].(string), nil

     

       
91
       
-
       
}
     

···

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0