A container registry that uses the AT Protocol for manifest storage and S3 for blob storage.
xrpc cleanup
+271
-37
+1
CLAUDE.md
+1
CLAUDE.md
+43
pkg/hold/pds/auth.go
+43
pkg/hold/pds/auth.go
···
192
192
193
193
return pdsEndpoint, nil
194
194
}
195
+
196
+
// ValidateOwnerOrCrewAdmin validates that the request has valid DPoP + OAuth tokens
197
+
// and that the authenticated user is either the hold owner or a crew member with crew:admin permission
198
+
func ValidateOwnerOrCrewAdmin(r *http.Request, pds *HoldPDS) (*ValidatedUser, error) {
199
+
// Validate DPoP + OAuth token
200
+
user, err := ValidateDPoPRequest(r)
201
+
if err != nil {
202
+
return nil, fmt.Errorf("authentication failed: %w", err)
203
+
}
204
+
205
+
// Get captain record to check owner
206
+
_, captain, err := pds.GetCaptainRecord(r.Context())
207
+
if err != nil {
208
+
return nil, fmt.Errorf("failed to get captain record: %w", err)
209
+
}
210
+
211
+
// Check if user is the owner
212
+
if user.DID == captain.Owner {
213
+
return user, nil
214
+
}
215
+
216
+
// Check if user is crew with admin permission
217
+
crew, err := pds.ListCrewMembers(r.Context())
218
+
if err != nil {
219
+
return nil, fmt.Errorf("failed to check crew membership: %w", err)
220
+
}
221
+
222
+
for _, member := range crew {
223
+
if member.Record.Member == user.DID {
224
+
// Check if this crew member has crew:admin permission
225
+
for _, perm := range member.Record.Permissions {
226
+
if perm == "crew:admin" {
227
+
return user, nil
228
+
}
229
+
}
230
+
// User is crew but doesn't have admin permission
231
+
return nil, fmt.Errorf("crew member lacks required 'crew:admin' permission")
232
+
}
233
+
}
234
+
235
+
// User is neither owner nor authorized crew
236
+
return nil, fmt.Errorf("user is not authorized (must be hold owner or crew admin)")
237
+
}
+20
-5
pkg/hold/pds/crew.go
+20
-5
pkg/hold/pds/crew.go
···
84
84
85
85
// Iterate over all crew records
86
86
err = r.ForEach(ctx, atproto.CrewCollection, func(k string, v cid.Cid) error {
87
-
// Extract rkey from full path (k is like "io.atcr.hold.crew/3m37dr2ddit22")
87
+
// Extract collection and rkey from full path (k is like "io.atcr.hold.crew/3m37dr2ddit22")
88
88
parts := strings.Split(k, "/")
89
+
if len(parts) < 2 {
90
+
return nil // Skip invalid keys
91
+
}
92
+
93
+
// Extract actual collection and rkey
94
+
actualCollection := strings.Join(parts[:len(parts)-1], "/")
89
95
rkey := parts[len(parts)-1]
96
+
97
+
// MST keys are sorted, so once we hit a different collection, stop walking
98
+
if actualCollection != atproto.CrewCollection {
99
+
return repo.ErrDoneIterating
100
+
}
90
101
91
102
// Get the record directly from the repo we already have open
92
103
// (calling GetCrewMember would open a new session unnecessarily)
···
110
121
})
111
122
112
123
if err != nil {
113
-
// If the collection doesn't exist yet (empty repo or no records created),
114
-
// return empty list instead of error
115
-
if err.Error() == "mst: not found" || strings.Contains(err.Error(), "not found") {
124
+
// ErrDoneIterating is expected when we stop walking early
125
+
if err == repo.ErrDoneIterating {
126
+
// Successfully stopped at collection boundary
127
+
} else if err.Error() == "mst: not found" || strings.Contains(err.Error(), "not found") {
128
+
// If the collection doesn't exist yet (empty repo or no records created),
129
+
// return empty list instead of error
116
130
return []*CrewMemberWithKey{}, nil
131
+
} else {
132
+
return nil, fmt.Errorf("failed to list crew members: %w", err)
117
133
}
118
-
return nil, fmt.Errorf("failed to list crew members: %w", err)
119
134
}
120
135
121
136
return crew, nil
+50
pkg/hold/pds/server.go
+50
pkg/hold/pds/server.go
···
5
5
"fmt"
6
6
"os"
7
7
"path/filepath"
8
+
"strings"
8
9
9
10
"atcr.io/pkg/atproto"
10
11
"github.com/bluesky-social/indigo/atproto/atcrypto"
11
12
"github.com/bluesky-social/indigo/carstore"
12
13
lexutil "github.com/bluesky-social/indigo/lex/util"
13
14
"github.com/bluesky-social/indigo/models"
15
+
"github.com/bluesky-social/indigo/repo"
16
+
"github.com/ipfs/go-cid"
14
17
)
15
18
16
19
// init registers our custom ATProto types with indigo's lexutil type registry
···
144
147
145
148
fmt.Printf("✅ Added %s as hold admin\n", ownerDID)
146
149
return nil
150
+
}
151
+
152
+
// ListCollections returns all collections present in the hold's repository
153
+
func (p *HoldPDS) ListCollections(ctx context.Context) ([]string, error) {
154
+
session, err := p.carstore.ReadOnlySession(p.uid)
155
+
if err != nil {
156
+
return nil, fmt.Errorf("failed to create read-only session: %w", err)
157
+
}
158
+
159
+
head, err := p.carstore.GetUserRepoHead(ctx, p.uid)
160
+
if err != nil {
161
+
return nil, fmt.Errorf("failed to get repo head: %w", err)
162
+
}
163
+
164
+
if !head.Defined() {
165
+
// Empty repo, no collections
166
+
return []string{}, nil
167
+
}
168
+
169
+
r, err := repo.OpenRepo(ctx, session, head)
170
+
if err != nil {
171
+
return nil, fmt.Errorf("failed to open repo: %w", err)
172
+
}
173
+
174
+
collections := make(map[string]bool)
175
+
176
+
// Walk all records in the repo to discover collections
177
+
err = r.ForEach(ctx, "", func(k string, v cid.Cid) error {
178
+
// k is like "io.atcr.hold.captain/self" or "io.atcr.hold.crew/3m3by7msdln22"
179
+
parts := strings.Split(k, "/")
180
+
if len(parts) >= 1 {
181
+
collections[parts[0]] = true
182
+
}
183
+
return nil
184
+
})
185
+
186
+
if err != nil {
187
+
return nil, fmt.Errorf("failed to enumerate collections: %w", err)
188
+
}
189
+
190
+
// Convert map to sorted slice
191
+
result := make([]string, 0, len(collections))
192
+
for collection := range collections {
193
+
result = append(result, collection)
194
+
}
195
+
196
+
return result, nil
147
197
}
148
198
149
199
// Close closes the carstore
+157
-32
pkg/hold/pds/xrpc.go
+157
-32
pkg/hold/pds/xrpc.go
···
8
8
"strings"
9
9
10
10
"atcr.io/pkg/atproto"
11
+
lexutil "github.com/bluesky-social/indigo/lex/util"
12
+
"github.com/bluesky-social/indigo/repo"
11
13
"github.com/ipfs/go-cid"
12
14
"github.com/ipld/go-car"
13
15
carutil "github.com/ipld/go-car/util"
···
79
81
mux.HandleFunc("/.well-known/did.json", corsMiddleware(h.HandleDIDDocument))
80
82
mux.HandleFunc("/.well-known/atproto-did", corsMiddleware(h.HandleAtprotoDID))
81
83
84
+
// Write endpoints
85
+
mux.HandleFunc("/xrpc/com.atproto.repo.deleteRecord", corsMiddleware(h.HandleDeleteRecord))
86
+
82
87
// Custom ATCR endpoints
83
88
mux.HandleFunc("/xrpc/io.atcr.hold.requestCrew", corsMiddleware(h.HandleRequestCrew))
84
89
}
···
131
136
}
132
137
133
138
// Get repo parameter
134
-
repo := r.URL.Query().Get("repo")
135
-
if repo == "" || repo != h.pds.DID() {
139
+
repoDID := r.URL.Query().Get("repo")
140
+
if repoDID == "" || repoDID != h.pds.DID() {
136
141
http.Error(w, "invalid repo", http.StatusBadRequest)
137
142
return
138
143
}
···
144
149
return
145
150
}
146
151
147
-
// TODO: Get actual repo head from carstore
152
+
// Get actual collections from repo
153
+
collections, err := h.pds.ListCollections(r.Context())
154
+
if err != nil {
155
+
http.Error(w, fmt.Sprintf("failed to list collections: %v", err), http.StatusInternalServerError)
156
+
return
157
+
}
158
+
148
159
// Note: For did:web, the handle IS the DID (not just hostname)
149
160
response := map[string]any{
150
161
"did": h.pds.DID(),
151
162
"handle": h.pds.DID(),
152
163
"didDoc": didDoc,
153
-
"collections": []string{atproto.CrewCollection},
164
+
"collections": collections,
154
165
"handleIsCorrect": true,
155
166
}
156
167
···
165
176
return
166
177
}
167
178
168
-
repo := r.URL.Query().Get("repo")
179
+
repoDID := r.URL.Query().Get("repo")
169
180
collection := r.URL.Query().Get("collection")
170
181
rkey := r.URL.Query().Get("rkey")
171
182
172
-
if repo == "" || collection == "" || rkey == "" {
183
+
if repoDID == "" || collection == "" || rkey == "" {
173
184
http.Error(w, "missing required parameters", http.StatusBadRequest)
174
185
return
175
186
}
176
187
177
-
if repo != h.pds.DID() {
188
+
if repoDID != h.pds.DID() {
178
189
http.Error(w, "invalid repo", http.StatusBadRequest)
179
190
return
180
191
}
181
192
182
-
// Only support crew collection for now
183
-
if collection != atproto.CrewCollection {
184
-
http.Error(w, "collection not found", http.StatusNotFound)
185
-
return
186
-
}
187
-
188
-
recordCID, crewRecord, err := h.pds.GetCrewMember(r.Context(), rkey)
193
+
// Use generic repomgr.GetRecord - works for any collection
194
+
// lexutil type registry automatically unmarshals to correct type
195
+
recordCID, recordValue, err := h.pds.repomgr.GetRecord(
196
+
r.Context(),
197
+
h.pds.uid,
198
+
collection,
199
+
rkey,
200
+
cid.Undef,
201
+
)
189
202
if err != nil {
190
-
http.Error(w, fmt.Sprintf("failed to get record: %v", err), http.StatusNotFound)
203
+
if strings.Contains(err.Error(), "not found") {
204
+
http.Error(w, "record not found", http.StatusNotFound)
205
+
} else {
206
+
http.Error(w, fmt.Sprintf("failed to get record: %v", err), http.StatusInternalServerError)
207
+
}
191
208
return
192
209
}
193
210
194
211
response := map[string]any{
195
212
"uri": fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), collection, rkey),
196
213
"cid": recordCID.String(),
197
-
"value": crewRecord,
214
+
"value": recordValue,
198
215
}
199
216
200
217
w.Header().Set("Content-Type", "application/json")
···
208
225
return
209
226
}
210
227
211
-
repo := r.URL.Query().Get("repo")
228
+
repoDID := r.URL.Query().Get("repo")
212
229
collection := r.URL.Query().Get("collection")
213
230
214
-
if repo == "" || collection == "" {
231
+
if repoDID == "" || collection == "" {
215
232
http.Error(w, "missing required parameters", http.StatusBadRequest)
216
233
return
217
234
}
218
235
219
-
if repo != h.pds.DID() {
236
+
if repoDID != h.pds.DID() {
220
237
http.Error(w, "invalid repo", http.StatusBadRequest)
221
238
return
222
239
}
223
240
224
-
// Only support crew collection for now
225
-
if collection != atproto.CrewCollection {
226
-
http.Error(w, "collection not found", http.StatusNotFound)
241
+
// Generic implementation using repo.ForEach
242
+
session, err := h.pds.carstore.ReadOnlySession(h.pds.uid)
243
+
if err != nil {
244
+
http.Error(w, fmt.Sprintf("failed to create session: %v", err), http.StatusInternalServerError)
245
+
return
246
+
}
247
+
248
+
head, err := h.pds.carstore.GetUserRepoHead(r.Context(), h.pds.uid)
249
+
if err != nil {
250
+
http.Error(w, fmt.Sprintf("failed to get repo head: %v", err), http.StatusInternalServerError)
251
+
return
252
+
}
253
+
254
+
if !head.Defined() {
255
+
// Empty repo, return empty list
256
+
response := map[string]any{"records": []any{}}
257
+
w.Header().Set("Content-Type", "application/json")
258
+
json.NewEncoder(w).Encode(response)
227
259
return
228
260
}
229
261
230
-
crew, err := h.pds.ListCrewMembers(r.Context())
262
+
repoHandle, err := repo.OpenRepo(r.Context(), session, head)
231
263
if err != nil {
232
-
http.Error(w, fmt.Sprintf("failed to list records: %v", err), http.StatusInternalServerError)
264
+
http.Error(w, fmt.Sprintf("failed to open repo: %v", err), http.StatusInternalServerError)
233
265
return
234
266
}
235
267
236
-
records := make([]map[string]any, len(crew))
237
-
for i, member := range crew {
238
-
records[i] = map[string]any{
239
-
"uri": fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), collection, member.Rkey),
240
-
"cid": member.Cid.String(),
241
-
"value": member.Record,
268
+
var records []map[string]any
269
+
270
+
// Iterate over all records in the collection
271
+
err = repoHandle.ForEach(r.Context(), collection, func(k string, v cid.Cid) error {
272
+
// k is like "io.atcr.hold.captain/self" or "io.atcr.hold.crew/3m3by7msdln22"
273
+
parts := strings.Split(k, "/")
274
+
if len(parts) < 2 {
275
+
return nil // Skip invalid keys
276
+
}
277
+
278
+
// Extract actual collection and rkey from the key path
279
+
actualCollection := strings.Join(parts[:len(parts)-1], "/")
280
+
rkey := parts[len(parts)-1]
281
+
282
+
// Filter: only include records that match the requested collection
283
+
// MST keys are sorted lexicographically, so once we hit a different
284
+
// collection prefix, all remaining keys will also be outside our range
285
+
if actualCollection != collection {
286
+
return repo.ErrDoneIterating // Stop walking the tree
287
+
}
288
+
289
+
// Get the record bytes
290
+
recordCID, recBytes, err := repoHandle.GetRecordBytes(r.Context(), k)
291
+
if err != nil {
292
+
return fmt.Errorf("failed to get record: %w", err)
293
+
}
294
+
295
+
// Decode using lexutil (type registry handles unmarshaling)
296
+
recordValue, err := lexutil.CborDecodeValue(*recBytes)
297
+
if err != nil {
298
+
return fmt.Errorf("failed to decode record: %w", err)
299
+
}
300
+
301
+
records = append(records, map[string]any{
302
+
"uri": fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), actualCollection, rkey),
303
+
"cid": recordCID.String(),
304
+
"value": recordValue,
305
+
})
306
+
return nil
307
+
})
308
+
309
+
if err != nil {
310
+
// ErrDoneIterating is expected when we stop walking early (reached collection boundary)
311
+
if err == repo.ErrDoneIterating {
312
+
// Successfully stopped at collection boundary, continue with collected records
313
+
} else if strings.Contains(err.Error(), "not found") {
314
+
// If the collection doesn't exist yet, return empty list
315
+
records = []map[string]any{}
316
+
} else {
317
+
http.Error(w, fmt.Sprintf("failed to list records: %v", err), http.StatusInternalServerError)
318
+
return
242
319
}
243
320
}
244
321
···
250
327
json.NewEncoder(w).Encode(response)
251
328
}
252
329
330
+
// HandleDeleteRecord deletes a record from the repository
331
+
func (h *XRPCHandler) HandleDeleteRecord(w http.ResponseWriter, r *http.Request) {
332
+
if r.Method != http.MethodPost {
333
+
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
334
+
return
335
+
}
336
+
337
+
repoDID := r.URL.Query().Get("repo")
338
+
collection := r.URL.Query().Get("collection")
339
+
rkey := r.URL.Query().Get("rkey")
340
+
341
+
if repoDID == "" || collection == "" || rkey == "" {
342
+
http.Error(w, "missing required parameters", http.StatusBadRequest)
343
+
return
344
+
}
345
+
346
+
if repoDID != h.pds.DID() {
347
+
http.Error(w, "invalid repo", http.StatusBadRequest)
348
+
return
349
+
}
350
+
351
+
// Validate DPoP + OAuth and check authorization
352
+
_, err := ValidateOwnerOrCrewAdmin(r, h.pds)
353
+
if err != nil {
354
+
http.Error(w, fmt.Sprintf("unauthorized: %v", err), http.StatusForbidden)
355
+
return
356
+
}
357
+
358
+
// Delete the record using repomgr
359
+
err = h.pds.repomgr.DeleteRecord(r.Context(), h.pds.uid, collection, rkey)
360
+
if err != nil {
361
+
if strings.Contains(err.Error(), "not found") {
362
+
http.Error(w, "record not found", http.StatusNotFound)
363
+
} else {
364
+
http.Error(w, fmt.Sprintf("failed to delete record: %v", err), http.StatusInternalServerError)
365
+
}
366
+
return
367
+
}
368
+
369
+
// Return success response
370
+
response := map[string]any{
371
+
"success": true,
372
+
}
373
+
374
+
w.Header().Set("Content-Type", "application/json")
375
+
json.NewEncoder(w).Encode(response)
376
+
}
377
+
253
378
// HandleSyncGetRecord returns a single record as a CAR file for sync
254
379
func (h *XRPCHandler) HandleSyncGetRecord(w http.ResponseWriter, r *http.Request) {
255
380
if r.Method != http.MethodGet {
···
271
396
return
272
397
}
273
398
274
-
// Only support crew collection for now
275
-
if collection != atproto.CrewCollection {
399
+
// Support both captain and crew collections
400
+
if collection != atproto.CaptainCollection && collection != atproto.CrewCollection {
276
401
http.Error(w, "collection not found", http.StatusNotFound)
277
402
return
278
403
}