mirrors/at-container-registry
1
0
Fork 0
mirror of https://tangled.org/evan.jarrett.net/at-container-registry synced 2026-08-29 04:06:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

remove user oauth flow. hold now contains captain record indicating owner

Browse Source
This commit is contained in:
Evan Jarrett
2025-10-15 14:47:53 -05:00
parent a271d3d8e3
commit fade86abaa
12 changed files with 1087 additions and 562 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/hold/main.go
+2 -59
View File
@@ -2,18 +2,14 @@ package main
import (
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"strconv"
"strings"
"time"
"atcr.io/pkg/atproto"
"atcr.io/pkg/hold"
"atcr.io/pkg/hold/pds"
indigooauth "github.com/bluesky-social/indigo/atproto/auth/oauth"
// Import storage drivers
_ "github.com/distribution/distribution/v3/registry/storage/driver/filesystem"
@@ -48,8 +44,8 @@ func main() {
log.Fatalf("Failed to initialize embedded PDS: %v", err)
}
// Bootstrap PDS with hold owner as first crew member
if err := holdPDS.Bootstrap(ctx, cfg.Registration.OwnerDID); err != nil {
// Bootstrap PDS with captain record and hold owner as first crew member
if err := holdPDS.Bootstrap(ctx, cfg.Registration.OwnerDID, cfg.Server.Public, cfg.Registration.AllowAllCrew); err != nil {
log.Fatalf("Failed to bootstrap PDS: %v", err)
}
@@ -114,41 +110,6 @@ func main() {
service.HandleMultipartPartUpload(w, r, uploadID, partNumber, did, service.MultipartMgr)
})
// Pre-register OAuth callback route (will be populated by auto-registration)
var oauthCallbackHandler http.HandlerFunc
mux.HandleFunc("/auth/oauth/callback", func(w http.ResponseWriter, r *http.Request) {
if oauthCallbackHandler != nil {
oauthCallbackHandler(w, r)
} else {
http.Error(w, "OAuth callback not initialized", http.StatusServiceUnavailable)
}
})
// OAuth client metadata endpoint for ATProto OAuth
// The hold service serves its metadata at /client-metadata.json
// This is referenced by its client ID URL
mux.HandleFunc("/client-metadata.json", func(w http.ResponseWriter, r *http.Request) {
// Create a temporary config to generate metadata (indigo provides this)
redirectURI := cfg.Server.PublicURL + "/auth/oauth/callback"
clientID := cfg.Server.PublicURL + "/client-metadata.json"
// Define scopes needed for hold registration and crew management
// Omit action parameter to allow all actions (create, update, delete)
scopes := []string{
"atproto",
fmt.Sprintf("repo:%s", atproto.HoldCollection),
fmt.Sprintf("repo:%s", atproto.HoldCrewCollection),
fmt.Sprintf("repo:%s", atproto.SailorProfileCollection),
}
config := indigooauth.NewPublicConfig(clientID, redirectURI, scopes)
metadata := config.ClientMetadata()
// Serve as JSON
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Access-Control-Allow-Origin", "*")
json.NewEncoder(w).Encode(metadata)
})
mux.HandleFunc("/blobs/", func(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet, http.MethodHead:
@@ -183,24 +144,6 @@ func main() {
}
}()
// Give server a moment to start
time.Sleep(100 * time.Millisecond)
// Auto-register if owner DID is set (now that server is running)
if cfg.Registration.OwnerDID != "" {
if err := service.AutoRegister(&oauthCallbackHandler); err != nil {
log.Printf("WARNING: Auto-registration failed: %v", err)
log.Printf("You can register manually later using the /register endpoint")
} else {
log.Printf("Successfully registered hold service in PDS")
}
// Reconcile allow-all crew state
if err := service.ReconcileAllowAllCrew(&oauthCallbackHandler); err != nil {
log.Printf("WARNING: Failed to reconcile allow-all crew state: %v", err)
}
}
// Wait for server error or shutdown
if err := <-serverErr; err != nil {
log.Fatalf("Server failed: %v", err)
docs/EMBEDDED_PDS.md
+267 -18
View File
@@ -146,7 +146,8 @@ Hold Service = Minimal PDS (did:web:hold1.example.com)
│ ├── io.atcr.hold.exportImage (data portability)
│ └── io.atcr.hold.getStats (metadata)
└── Records (hold's own PDS):
├── io.atcr.hold.crew (crew membership)
├── io.atcr.hold.captain (single record: ownership & metadata)
├── io.atcr.hold.crew/* (crew membership & permissions)
└── io.atcr.hold.config (hold configuration)
```
@@ -270,8 +271,20 @@ Storage continues to use distribution's existing S3 layout. The PDS interface is
**This is standard ATProto federation** - services pass OAuth tokens with DPoP proofs between each other. Hold independently validates tokens against the user's PDS, so there's no trust relationship required.
**Crew records stored in hold's PDS:**
**Records stored in hold's PDS:**
```json
// io.atcr.hold.captain (single record - hold metadata)
{
"$type": "io.atcr.hold.captain",
"owner": "did:plc:alice123",
"public": false,
"deployedAt": "2025-10-14T...",
"region": "iad",
"provider": "fly.io"
}
// io.atcr.hold.crew/* (access control records)
{
"$type": "io.atcr.hold.crew",
"member": "did:plc:alice123",
@@ -281,6 +294,10 @@ Storage continues to use distribution's existing S3 layout. The PDS interface is
}
```
**Semantic separation:**
- **Captain record** = Hold ownership and metadata (who owns it, where it's deployed)
- **Crew records** = Access control (who can use it, what permissions they have)
**Security considerations:**
- User's OAuth token is exposed to hold during delegation
- However, hold independently validates it (can't be forged)
@@ -467,14 +484,85 @@ Can we build a reusable "static PDS" library for apps like ATCR, tangled.org, st
### 6. Hold Discovery & Registration
**Current:** Hold registers by creating records in owner's PDS
**New:** Hold is its own identity - how does AppView discover available holds?
**Decision: No registration records needed in owner's PDS.**
Possibilities:
- Holds publish to feeds
- AppView maintains directory
- DIDs are manually configured
- ATProto directory service
Since holds are ATProto actors with did:web identity, they are self-describing:
**Hold's PDS contains everything:**
```
did:web:hold01.atcr.io
├── io.atcr.hold.captain → { owner: "did:plc:alice123", ... }
└── io.atcr.hold.crew/* → Access control records
```
**DID Document with Multiple Services:**
Holds expose multiple service endpoints to distinguish themselves from generic PDSs:
```json
{
"@context": ["https://www.w3.org/ns/did/v1", ...],
"id": "did:web:hold01.atcr.io",
"service": [
{
"id": "#atproto_pds",
"type": "AtprotoPersonalDataServer",
"serviceEndpoint": "https://hold01.atcr.io"
},
{
"id": "#atcr_hold",
"type": "AtcrHoldService",
"serviceEndpoint": "https://hold01.atcr.io"
}
]
}
```
**Service semantics:**
- **`#atproto_pds`** - Standard ATProto PDS operations (crew queries, record sync)
- **`#atcr_hold`** - ATCR-specific operations (blob storage, presigned URLs)
**Discovery patterns:**
1. **Direct deployment** - Owner deploys hold, knows the DID
2. **Sailor profiles** - Users reference holds by DID in their profile
3. **DID resolution** - `did:web:hold01.atcr.io` → `https://hold01.atcr.io/.well-known/did.json`
4. **Service lookup** - Check for `#atcr_hold` service to identify ATCR holds
5. **Crew queries** - AppView queries hold's PDS directly via `#atproto_pds` endpoint
**AppView resolution flow:**
```go
// 1. Get hold DID from sailor profile
holdDID := profile.DefaultHold // "did:web:hold01.atcr.io"
// 2. Resolve DID document
didDoc := resolveDidWeb(holdDID)
// 3. Extract service endpoints
pdsEndpoint := didDoc.GetService("#atproto_pds") // XRPC operations
holdEndpoint := didDoc.GetService("#atcr_hold") // Blob operations
// 4. Query crew list via PDS endpoint
crew := xrpcClient.ListRecords(pdsEndpoint, "io.atcr.hold.crew")
// 5. Check if user has access
hasAccess := crew.Contains(userDID)
```
**No need for reverse lookup** (owner → holds). Users know their holds because they deployed them.
**Benefits:**
- ✅ Single source of truth (hold's PDS)
- ✅ No cross-PDS writes during registration
- ✅ Self-describing ATProto actors
- ✅ Standard DID resolution patterns
- ✅ Clear service semantics (PDS vs ATCR-specific)
- ✅ Discoverable via service type
**OAuth implications:**
- OAuth registration flow no longer needed (hold is self-describing)
- OAuth code kept for backward compatibility with legacy registration records
- Future: Remove OAuth after migration period
### 7. Multi-Tenancy
@@ -603,12 +691,22 @@ We're building the minimal PDS needed for discoverability, not a full social cli
**Key insight:** Other ATProto services will "just work" as long as they can retrieve records from the hold's PDS. We don't need to implement full social features for the hold to participate in the ecosystem.
### Crew Management: Individual Records
### Crew Management: Captain + Individual Records
**Decision: Individual crew record per user (remove wildcard logic)**
**Decision: Captain record (ownership) + Individual crew records (access control)**
```json
// io.atcr.hold.crew/{rkey}
// io.atcr.hold.captain (single record - hold metadata)
{
"$type": "io.atcr.hold.captain",
"owner": "did:plc:alice123",
"public": false,
"deployedAt": "2025-10-14T...",
"region": "iad",
"provider": "fly.io"
}
// io.atcr.hold.crew/{rkey} (access control)
{
"$type": "io.atcr.hold.crew",
"member": "did:plc:alice123",
@@ -617,7 +715,7 @@ We're building the minimal PDS needed for discoverability, not a full social cli
"addedAt": "2025-10-14T..."
}
// io.atcr.hold.config/policy
// io.atcr.hold.config/policy (optional)
{
"$type": "io.atcr.hold.config",
"access": "public", // or "allowlist"
@@ -627,6 +725,11 @@ We're building the minimal PDS needed for discoverability, not a full social cli
}
```
**Semantic separation:**
- **Captain record** = Who owns/deployed the hold (billing, deletion, migration rights)
- **Crew records** = Who can use the hold (access control, permissions)
- **Config record** = Hold-wide policies
**Authorization logic:**
```go
func (p *HoldPDS) CheckAccess(ctx context.Context, userDID string) (bool, error) {
@@ -661,13 +764,159 @@ func (p *HoldPDS) CheckAccess(ctx context.Context, userDID string) (bool, error)
- **Private team hold:** `access: "allowlist"` - explicit crew membership
- **Hybrid:** Public access + explicit admin crew records for elevated permissions
### Phase 2: XRPC Endpoints Implementation ✅ COMPLETED
**Critical Implementation Lessons Learned:**
#### 1. Custom Record Types Require Manual CBOR Decoding
Indigo's `repo.GetRecord()` uses its lexicon decoder which only knows about built-in ATProto types. For custom types, you must use `GetRecordBytes()` and decode manually:
```go
// ❌ WRONG - Fails with "unrecognized lexicon type"
record, err := repo.GetRecord(ctx, path, &CrewRecord{})
// ✅ CORRECT - Manual CBOR decoding
recordCID, recBytes, err := repo.GetRecordBytes(ctx, path)
var crewRecord CrewRecord
err = crewRecord.UnmarshalCBOR(bytes.NewReader(*recBytes))
```
**Why:** Indigo's lexicon system doesn't know about `io.atcr.hold.crew` or other custom types.
#### 2. JSON Struct Tags Must Match CBOR Tags Exactly
For CID verification to work, JSON and CBOR encodings must produce identical bytes:
```go
// ❌ WRONG - JSON uses capital field names (Member, Role)
type CrewRecord struct {
Type string `cborgen:"$type"`
Member string `cborgen:"member"`
Role string `cborgen:"role"`
Permissions []string `cborgen:"permissions"`
AddedAt string `cborgen:"addedAt"`
}
// ✅ CORRECT - JSON tags match CBOR tags
type CrewRecord struct {
Type string `json:"$type" cborgen:"$type"`
Member string `json:"member" cborgen:"member"`
Role string `json:"role" cborgen:"role"`
Permissions []string `json:"permissions" cborgen:"permissions"`
AddedAt string `json:"addedAt" cborgen:"addedAt"`
}
```
**Why:** Verification code CBOR-encodes the JSON record and compares the CID. Mismatched field names produce different bytes and thus different CIDs.
#### 3. MST ForEach Returns Full Paths
The `repo.ForEach()` callback receives full collection paths, not just record keys:
```go
// ❌ WRONG - Prepends collection prefix again
err := repo.ForEach(ctx, "io.atcr.hold.crew", func(k string, v cid.Cid) error {
// k is already "io.atcr.hold.crew/3m37dr2ddit22"
path := fmt.Sprintf("%s/%s", collection, k) // Double path!
return nil
})
// ✅ CORRECT - Extract just the rkey
err := repo.ForEach(ctx, "io.atcr.hold.crew", func(k string, v cid.Cid) error {
// k = "io.atcr.hold.crew/3m37dr2ddit22"
parts := strings.Split(k, "/")
rkey := parts[len(parts)-1] // "3m37dr2ddit22"
return nil
})
```
#### 4. All Record Endpoints Must Return CIDs
Per ATProto spec, `com.atproto.repo.getRecord` and `listRecords` must include the record's CID:
```go
// ✅ CORRECT - Include CID in response
response := map[string]any{
"uri": fmt.Sprintf("at://%s/%s/%s", did, collection, rkey),
"cid": recordCID.String(), // Required!
"value": record,
}
```
**Why:** Clients need the CID to verify record integrity via `com.atproto.sync.getRecord`.
#### 5. sync.getRecord CAR Files Must Include Full MST Path
The `com.atproto.sync.getRecord` endpoint must return a CAR file with ALL blocks needed to verify the record:
```go
// ❌ WRONG - Only includes the record block
blk, _ := repo.Blockstore().Get(ctx, recordCID)
// Write single block to CAR
// ✅ CORRECT - Capture all accessed blocks
loggingBS := util.NewLoggingBstore(session)
tempRepo, _ := repo.OpenRepo(ctx, loggingBS, repoHead)
_, _, _ = tempRepo.GetRecordBytes(ctx, path)
blocks := loggingBS.GetLoggedBlocks() // Commit + MST nodes + record
// Write all blocks to CAR
```
**Components included:**
1. **Commit block** - Repo head with signature, data root, version
2. **MST tree nodes** - Path from root to record (log N depth)
3. **Record block** - The actual record data
**Why:** Clients need the full Merkle path to cryptographically verify the record against the repo head.
#### 6. CAR Root Must Be Repo Head, Not Record CID
The CAR file's root CID must be the repo head (commit), not the record:
```go
// ❌ WRONG - Uses record CID as root
header := &car.CarHeader{
Roots: []cid.Cid{recordCID},
Version: 1,
}
// ✅ CORRECT - Uses repo head as root
repoHead, _ := carstore.GetUserRepoHead(ctx, uid)
header := &car.CarHeader{
Roots: []cid.Cid{repoHead}, // Commit CID
Version: 1,
}
```
**Why:** The CAR represents a slice of the repo from head to record, not just the record itself.
#### 7. Empty Collections Should Return Empty Arrays
Handle empty collections gracefully instead of returning errors:
```go
// ✅ CORRECT - Return empty array for missing collection
err := repo.ForEach(ctx, collection, func(k string, v cid.Cid) error {
// ...
})
if err != nil {
if err.Error() == "mst: not found" {
return []*CrewMemberWithKey{}, nil // Empty collection
}
return nil, err // Real error
}
```
**Why:** ATProto expects empty arrays for non-existent collections, not 404 errors.
### Next Steps
1. **Add indigo dependencies** - carstore, repo, MST
2. **Implement HoldPDS with carstore** - Create pkg/hold/pds
3. **Add crew management** - CRUD operations for crew records
4. **Implement standard PDS endpoints** - describeServer, describeRepo, getRecord, listRecords
5. **Add DID document** - did:web identity generation
1. ~~**Add indigo dependencies**~~ ✅
2. ~~**Implement HoldPDS with carstore**~~ ✅
3. ~~**Add crew management**~~ ✅
4. ~~**Implement standard PDS endpoints**~~ ✅
5. ~~**Add DID document**~~ ✅
6. **Custom XRPC methods** - getUploadUrl, getDownloadUrl (presigned URLs)
7. **Wire up in cmd/hold** - Serve XRPC alongside existing HTTP
8. **Test basic operations** - Add/list crew, policy checks
gen/main.go
+2 -1
View File
@@ -20,9 +20,10 @@ import (
)
func main() {
// Generate map-style encoders for CrewRecord
// Generate map-style encoders for CrewRecord and CaptainRecord
if err := cbg.WriteMapEncodersToFile("pkg/hold/pds/cbor_gen.go", "pds",
pds.CrewRecord{},
pds.CaptainRecord{},
); err != nil {
fmt.Printf("Failed to generate CBOR encoders: %v\n", err)
os.Exit(1)
pkg/hold/pds/auth.go
+194
View File
@@ -0,0 +1,194 @@
package pds
import (
"context"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"github.com/bluesky-social/indigo/atproto/identity"
"github.com/bluesky-social/indigo/atproto/syntax"
)
// ValidatedUser represents a successfully validated user from DPoP + OAuth
type ValidatedUser struct {
DID string
Handle string
PDS string
Authorized bool
}
// ValidateDPoPRequest validates a request with DPoP + OAuth tokens
// This implements the standard ATProto token validation flow:
// 1. Extract Authorization header (DPoP <token>)
// 2. Extract DPoP header (proof JWT)
// 3. Call user's PDS to validate token via com.atproto.server.getSession
// 4. Return validated user DID
func ValidateDPoPRequest(r *http.Request) (*ValidatedUser, error) {
// Extract Authorization header
authHeader := r.Header.Get("Authorization")
if authHeader == "" {
return nil, fmt.Errorf("missing Authorization header")
}
// Check for DPoP authorization scheme
parts := strings.SplitN(authHeader, " ", 2)
if len(parts) != 2 {
return nil, fmt.Errorf("invalid Authorization header format")
}
if parts[0] != "DPoP" {
return nil, fmt.Errorf("expected DPoP authorization scheme, got: %s", parts[0])
}
accessToken := parts[1]
if accessToken == "" {
return nil, fmt.Errorf("missing access token")
}
// Extract DPoP header
dpopProof := r.Header.Get("DPoP")
if dpopProof == "" {
return nil, fmt.Errorf("missing DPoP header")
}
// TODO: We could verify the DPoP proof locally (signature, HTM, HTU, etc.)
// For now, we'll rely on the PDS to validate everything
// The token contains the user's DID in its claims, but we can't trust it without validation
// We need to call the user's PDS to validate the token
// Problem: We don't know which PDS to call yet!
// For now, we'll parse the JWT to extract the DID/PDS hint (unverified)
// Then validate against that PDS
// This is safe because the PDS will verify the token is valid for that DID
did, pds, err := extractDIDFromToken(accessToken)
if err != nil {
return nil, fmt.Errorf("failed to extract DID from token: %w", err)
}
// Validate token with the user's PDS
session, err := validateTokenWithPDS(r.Context(), pds, accessToken, dpopProof)
if err != nil {
return nil, fmt.Errorf("token validation failed: %w", err)
}
// Verify the DID matches
if session.DID != did {
return nil, fmt.Errorf("token DID mismatch: expected %s, got %s", did, session.DID)
}
return &ValidatedUser{
DID: session.DID,
Handle: session.Handle,
PDS: pds,
Authorized: true,
}, nil
}
// extractDIDFromToken extracts the DID and PDS from an unverified JWT token
// This is just for routing purposes - the token will be validated by the PDS
func extractDIDFromToken(token string) (string, string, error) {
// JWT format: header.payload.signature
parts := strings.Split(token, ".")
if len(parts) != 3 {
return "", "", fmt.Errorf("invalid JWT format")
}
// Decode payload (base64url)
payload, err := decodeBase64URL(parts[1])
if err != nil {
return "", "", fmt.Errorf("failed to decode payload: %w", err)
}
// Parse JSON
var claims struct {
Sub string `json:"sub"` // DID
Iss string `json:"iss"` // PDS URL (issuer)
}
if err := json.Unmarshal(payload, &claims); err != nil {
return "", "", fmt.Errorf("failed to parse claims: %w", err)
}
if claims.Sub == "" {
return "", "", fmt.Errorf("missing sub claim (DID)")
}
if claims.Iss == "" {
return "", "", fmt.Errorf("missing iss claim (PDS)")
}
return claims.Sub, claims.Iss, nil
}
// decodeBase64URL decodes base64url (RFC 4648)
func decodeBase64URL(s string) ([]byte, error) {
// Use Go's RawURLEncoding (base64url without padding)
return base64.RawURLEncoding.DecodeString(s)
}
// SessionResponse represents the response from com.atproto.server.getSession
type SessionResponse struct {
DID string `json:"did"`
Handle string `json:"handle"`
}
// validateTokenWithPDS calls the user's PDS to validate the token
func validateTokenWithPDS(ctx context.Context, pdsURL, accessToken, dpopProof string) (*SessionResponse, error) {
// Call com.atproto.server.getSession with DPoP headers
url := fmt.Sprintf("%s/xrpc/com.atproto.server.getSession", strings.TrimSuffix(pdsURL, "/"))
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
// Add DPoP authorization headers
req.Header.Set("Authorization", "DPoP "+accessToken)
req.Header.Set("DPoP", dpopProof)
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to call PDS: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))
}
var session SessionResponse
if err := json.NewDecoder(resp.Body).Decode(&session); err != nil {
return nil, fmt.Errorf("failed to decode session: %w", err)
}
return &session, nil
}
// ResolveDIDToPDS resolves a DID to its PDS endpoint (for reference)
// This is an alternative approach if we don't trust the token's issuer claim
func ResolveDIDToPDS(ctx context.Context, did string) (string, error) {
directory := identity.DefaultDirectory()
didParsed, err := syntax.ParseDID(did)
if err != nil {
return "", fmt.Errorf("invalid DID: %w", err)
}
ident, err := directory.LookupDID(ctx, didParsed)
if err != nil {
return "", fmt.Errorf("failed to resolve DID: %w", err)
}
pdsEndpoint := ident.PDSEndpoint()
if pdsEndpoint == "" {
return "", fmt.Errorf("no PDS endpoint found for DID")
}
return pdsEndpoint, nil
}
pkg/hold/pds/captain.go
+146
View File
@@ -0,0 +1,146 @@
package pds
import (
"bytes"
"context"
"fmt"
"time"
"github.com/bluesky-social/indigo/repo"
"github.com/ipfs/go-cid"
)
const (
// CaptainRkey is the fixed rkey for the captain record (singleton)
CaptainRkey = "self"
)
// CreateCaptainRecord creates the captain record for the hold
func (p *HoldPDS) CreateCaptainRecord(ctx context.Context, ownerDID string, public bool, allowAllCrew bool) (cid.Cid, error) {
captainRecord := &CaptainRecord{
Type: CaptainCollection,
Owner: ownerDID,
Public: public,
AllowAllCrew: allowAllCrew,
DeployedAt: time.Now().Format(time.RFC3339),
}
// Create record in repo with fixed rkey "self"
recordCID, rkey, err := p.repo.CreateRecord(ctx, CaptainCollection, captainRecord)
if err != nil {
return cid.Undef, fmt.Errorf("failed to create captain record: %w", err)
}
// Create signer function from signing key
signer := func(ctx context.Context, did string, data []byte) ([]byte, error) {
return p.signingKey.HashAndSign(data)
}
// Commit the changes to get new root CID
root, rev, err := p.repo.Commit(ctx, signer)
if err != nil {
return cid.Undef, fmt.Errorf("failed to commit captain record: %w", err)
}
// Close the delta session with the new root
_, err = p.session.CloseWithRoot(ctx, root, rev)
if err != nil {
return cid.Undef, fmt.Errorf("failed to persist commit: %w", err)
}
// Create a new session for the next operation
rootStr := root.String()
newSession, err := p.carstore.NewDeltaSession(ctx, p.uid, &rootStr)
if err != nil {
return cid.Undef, fmt.Errorf("failed to create new session: %w", err)
}
// Load repo from the newly committed head
newRepo, err := repo.OpenRepo(ctx, newSession, root)
if err != nil {
return cid.Undef, fmt.Errorf("failed to reload repo after commit: %w", err)
}
// Update the stored session and repo
p.session = newSession
p.repo = newRepo
fmt.Printf("Created captain record with rkey: %s, cid: %s\n", rkey, recordCID)
return recordCID, nil
}
// GetCaptainRecord retrieves the captain record
func (p *HoldPDS) GetCaptainRecord(ctx context.Context) (cid.Cid, *CaptainRecord, error) {
path := fmt.Sprintf("%s/%s", CaptainCollection, CaptainRkey)
// Get the record bytes and decode manually
recordCID, recBytes, err := p.repo.GetRecordBytes(ctx, path)
if err != nil {
return cid.Undef, nil, fmt.Errorf("failed to get captain record: %w", err)
}
// Decode the CBOR bytes into our CaptainRecord type
var captainRecord CaptainRecord
if err := captainRecord.UnmarshalCBOR(bytes.NewReader(*recBytes)); err != nil {
return cid.Undef, nil, fmt.Errorf("failed to decode captain record: %w", err)
}
return recordCID, &captainRecord, nil
}
// UpdateCaptainRecord updates the captain record (e.g., to change public/allowAllCrew settings)
func (p *HoldPDS) UpdateCaptainRecord(ctx context.Context, public bool, allowAllCrew bool) (cid.Cid, error) {
// Get existing record to preserve other fields
_, existing, err := p.GetCaptainRecord(ctx)
if err != nil {
return cid.Undef, fmt.Errorf("failed to get existing captain record: %w", err)
}
// Update the fields
existing.Public = public
existing.AllowAllCrew = allowAllCrew
// Update record in repo
path := fmt.Sprintf("%s/%s", CaptainCollection, CaptainRkey)
recordCID, err := p.repo.UpdateRecord(ctx, path, existing)
if err != nil {
return cid.Undef, fmt.Errorf("failed to update captain record: %w", err)
}
// Create signer function from signing key
signer := func(ctx context.Context, did string, data []byte) ([]byte, error) {
return p.signingKey.HashAndSign(data)
}
// Commit the changes
root, rev, err := p.repo.Commit(ctx, signer)
if err != nil {
return cid.Undef, fmt.Errorf("failed to commit captain record update: %w", err)
}
// Close the delta session with the new root
_, err = p.session.CloseWithRoot(ctx, root, rev)
if err != nil {
return cid.Undef, fmt.Errorf("failed to persist commit: %w", err)
}
// Create a new session for the next operation
rootStr := root.String()
newSession, err := p.carstore.NewDeltaSession(ctx, p.uid, &rootStr)
if err != nil {
return cid.Undef, fmt.Errorf("failed to create new session: %w", err)
}
// Load repo from the newly committed head
newRepo, err := repo.OpenRepo(ctx, newSession, root)
if err != nil {
return cid.Undef, fmt.Errorf("failed to reload repo after commit: %w", err)
}
// Update the stored session and repo
p.session = newSession
p.repo = newRepo
return recordCID, nil
}
pkg/hold/pds/cbor_gen.go
+319
View File
@@ -293,3 +293,322 @@ func (t *CrewRecord) UnmarshalCBOR(r io.Reader) (err error) {
return nil
}
func (t *CaptainRecord) MarshalCBOR(w io.Writer) error {
if t == nil {
_, err := w.Write(cbg.CborNull)
return err
}
cw := cbg.NewCborWriter(w)
fieldCount := 7
if t.Region == "" {
fieldCount--
}
if t.Provider == "" {
fieldCount--
}
if _, err := cw.Write(cbg.CborEncodeMajorType(cbg.MajMap, uint64(fieldCount))); err != nil {
return err
}
// t.Type (string) (string)
if len("$type") > 8192 {
return xerrors.Errorf("Value in field \"$type\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("$type"))); err != nil {
return err
}
if _, err := cw.WriteString(string("$type")); err != nil {
return err
}
if len(t.Type) > 8192 {
return xerrors.Errorf("Value in field t.Type was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Type))); err != nil {
return err
}
if _, err := cw.WriteString(string(t.Type)); err != nil {
return err
}
// t.Owner (string) (string)
if len("owner") > 8192 {
return xerrors.Errorf("Value in field \"owner\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("owner"))); err != nil {
return err
}
if _, err := cw.WriteString(string("owner")); err != nil {
return err
}
if len(t.Owner) > 8192 {
return xerrors.Errorf("Value in field t.Owner was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Owner))); err != nil {
return err
}
if _, err := cw.WriteString(string(t.Owner)); err != nil {
return err
}
// t.Public (bool) (bool)
if len("public") > 8192 {
return xerrors.Errorf("Value in field \"public\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("public"))); err != nil {
return err
}
if _, err := cw.WriteString(string("public")); err != nil {
return err
}
if err := cbg.WriteBool(w, t.Public); err != nil {
return err
}
// t.Region (string) (string)
if t.Region != "" {
if len("region") > 8192 {
return xerrors.Errorf("Value in field \"region\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("region"))); err != nil {
return err
}
if _, err := cw.WriteString(string("region")); err != nil {
return err
}
if len(t.Region) > 8192 {
return xerrors.Errorf("Value in field t.Region was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Region))); err != nil {
return err
}
if _, err := cw.WriteString(string(t.Region)); err != nil {
return err
}
}
// t.Provider (string) (string)
if t.Provider != "" {
if len("provider") > 8192 {
return xerrors.Errorf("Value in field \"provider\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("provider"))); err != nil {
return err
}
if _, err := cw.WriteString(string("provider")); err != nil {
return err
}
if len(t.Provider) > 8192 {
return xerrors.Errorf("Value in field t.Provider was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Provider))); err != nil {
return err
}
if _, err := cw.WriteString(string(t.Provider)); err != nil {
return err
}
}
// t.DeployedAt (string) (string)
if len("deployedAt") > 8192 {
return xerrors.Errorf("Value in field \"deployedAt\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("deployedAt"))); err != nil {
return err
}
if _, err := cw.WriteString(string("deployedAt")); err != nil {
return err
}
if len(t.DeployedAt) > 8192 {
return xerrors.Errorf("Value in field t.DeployedAt was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.DeployedAt))); err != nil {
return err
}
if _, err := cw.WriteString(string(t.DeployedAt)); err != nil {
return err
}
// t.AllowAllCrew (bool) (bool)
if len("allowAllCrew") > 8192 {
return xerrors.Errorf("Value in field \"allowAllCrew\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("allowAllCrew"))); err != nil {
return err
}
if _, err := cw.WriteString(string("allowAllCrew")); err != nil {
return err
}
if err := cbg.WriteBool(w, t.AllowAllCrew); err != nil {
return err
}
return nil
}
func (t *CaptainRecord) UnmarshalCBOR(r io.Reader) (err error) {
*t = CaptainRecord{}
cr := cbg.NewCborReader(r)
maj, extra, err := cr.ReadHeader()
if err != nil {
return err
}
defer func() {
if err == io.EOF {
err = io.ErrUnexpectedEOF
}
}()
if maj != cbg.MajMap {
return fmt.Errorf("cbor input should be of type map")
}
if extra > cbg.MaxLength {
return fmt.Errorf("CaptainRecord: map struct too large (%d)", extra)
}
n := extra
nameBuf := make([]byte, 12)
for i := uint64(0); i < n; i++ {
nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 8192)
if err != nil {
return err
}
if !ok {
// Field doesn't exist on this type, so ignore it
if err := cbg.ScanForLinks(cr, func(cid.Cid) {}); err != nil {
return err
}
continue
}
switch string(nameBuf[:nameLen]) {
// t.Type (string) (string)
case "$type":
{
sval, err := cbg.ReadStringWithMax(cr, 8192)
if err != nil {
return err
}
t.Type = string(sval)
}
// t.Owner (string) (string)
case "owner":
{
sval, err := cbg.ReadStringWithMax(cr, 8192)
if err != nil {
return err
}
t.Owner = string(sval)
}
// t.Public (bool) (bool)
case "public":
maj, extra, err = cr.ReadHeader()
if err != nil {
return err
}
if maj != cbg.MajOther {
return fmt.Errorf("booleans must be major type 7")
}
switch extra {
case 20:
t.Public = false
case 21:
t.Public = true
default:
return fmt.Errorf("booleans are either major type 7, value 20 or 21 (got %d)", extra)
}
// t.Region (string) (string)
case "region":
{
sval, err := cbg.ReadStringWithMax(cr, 8192)
if err != nil {
return err
}
t.Region = string(sval)
}
// t.Provider (string) (string)
case "provider":
{
sval, err := cbg.ReadStringWithMax(cr, 8192)
if err != nil {
return err
}
t.Provider = string(sval)
}
// t.DeployedAt (string) (string)
case "deployedAt":
{
sval, err := cbg.ReadStringWithMax(cr, 8192)
if err != nil {
return err
}
t.DeployedAt = string(sval)
}
// t.AllowAllCrew (bool) (bool)
case "allowAllCrew":
maj, extra, err = cr.ReadHeader()
if err != nil {
return err
}
if maj != cbg.MajOther {
return fmt.Errorf("booleans must be major type 7")
}
switch extra {
case 20:
t.AllowAllCrew = false
case 21:
t.AllowAllCrew = true
default:
return fmt.Errorf("booleans are either major type 7, value 20 or 21 (got %d)", extra)
}
default:
// Field doesn't exist on this type, so ignore it
if err := cbg.ScanForLinks(r, func(cid.Cid) {}); err != nil {
return err
}
}
}
return nil
}
pkg/hold/pds/did.go
+5
View File
@@ -77,6 +77,11 @@ func (p *HoldPDS) GenerateDIDDocument(publicURL string) (*DIDDocument, error) {
Type: "AtprotoPersonalDataServer",
ServiceEndpoint: publicURL,
},
{
ID: "#atcr_hold",
Type: "AtcrHoldService",
ServiceEndpoint: publicURL,
},
},
}
pkg/hold/pds/server.go
+10 -2
View File
@@ -98,8 +98,8 @@ func (p *HoldPDS) SigningKey() *atcrypto.PrivateKeyK256 {
return p.signingKey
}
// Bootstrap initializes the hold with the owner as the first crew member
func (p *HoldPDS) Bootstrap(ctx context.Context, ownerDID string) error {
// Bootstrap initializes the hold with the captain record and owner as first crew member
func (p *HoldPDS) Bootstrap(ctx context.Context, ownerDID string, public bool, allowAllCrew bool) error {
if ownerDID == "" {
return nil
}
@@ -115,6 +115,14 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, ownerDID string) error {
return nil
}
// Create captain record (hold ownership and settings)
_, err = p.CreateCaptainRecord(ctx, ownerDID, public, allowAllCrew)
if err != nil {
return fmt.Errorf("failed to create captain record: %w", err)
}
fmt.Printf("✅ Created captain record (public=%v, allowAllCrew=%v)\n", public, allowAllCrew)
// Add hold owner as first crew member with admin role
_, err = p.AddCrewMember(ctx, ownerDID, "admin", []string{"blob:read", "blob:write", "crew:admin"})
if err != nil {
pkg/hold/pds/types.go
+17 -1
View File
@@ -1,8 +1,23 @@
package pds
//go:generate go run github.com/whyrusleeping/cbor-gen --map-encoding CrewRecord CaptainRecord
// ATProto record types for the hold service
// CaptainRecord represents the hold's ownership and metadata
// Collection: io.atcr.hold.captain (single record per hold)
type CaptainRecord struct {
Type string `json:"$type" cborgen:"$type"`
Owner string `json:"owner" cborgen:"owner"` // DID of hold owner
Public bool `json:"public" cborgen:"public"` // Public read access
AllowAllCrew bool `json:"allowAllCrew" cborgen:"allowAllCrew"` // Allow any authenticated user to register as crew
DeployedAt string `json:"deployedAt" cborgen:"deployedAt"` // RFC3339 timestamp
Region string `json:"region,omitempty" cborgen:"region,omitempty"` // S3 region (optional)
Provider string `json:"provider,omitempty" cborgen:"provider,omitempty"` // Deployment provider (optional)
}
// CrewRecord represents a crew member in the hold
// Collection: io.atcr.hold.crew (one record per member)
type CrewRecord struct {
Type string `json:"$type" cborgen:"$type"`
Member string `json:"member" cborgen:"member"`
@@ -12,5 +27,6 @@ type CrewRecord struct {
}
const (
CrewCollection = "io.atcr.hold.crew"
CaptainCollection = "io.atcr.hold.captain"
CrewCollection = "io.atcr.hold.crew"
)
pkg/hold/pds/xrpc.go
+103
View File
@@ -79,6 +79,9 @@ func (h *XRPCHandler) RegisterHandlers(mux *http.ServeMux) {
// DID document and handle resolution
mux.HandleFunc("/.well-known/did.json", corsMiddleware(h.HandleDIDDocument))
mux.HandleFunc("/.well-known/atproto-did", corsMiddleware(h.HandleAtprotoDID))
// Custom ATCR endpoints
mux.HandleFunc("/xrpc/io.atcr.hold.requestCrew", corsMiddleware(h.HandleRequestCrew))
}
// HandleHealth returns health check information
@@ -480,3 +483,103 @@ func (h *XRPCHandler) HandleAtprotoDID(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain")
fmt.Fprint(w, h.pds.DID())
}
// HandleRequestCrew handles crew membership requests
// This endpoint allows authenticated users to request crew membership
// Authorization is checked against captain record settings
func (h *XRPCHandler) HandleRequestCrew(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
// Validate DPoP + OAuth token from Authorization and DPoP headers
user, err := ValidateDPoPRequest(r)
if err != nil {
http.Error(w, fmt.Sprintf("authentication failed: %v", err), http.StatusUnauthorized)
return
}
// Parse request body (optional parameters)
var req struct {
Role string `json:"role"` // Requested role (default: "member")
Permissions []string `json:"permissions"` // Requested permissions
}
// Body is optional - if empty, just use defaults
if r.Body != nil && r.ContentLength > 0 {
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest)
return
}
}
// Get captain record to check authorization settings
_, captain, err := h.pds.GetCaptainRecord(r.Context())
if err != nil {
http.Error(w, fmt.Sprintf("failed to get captain record: %v", err), http.StatusInternalServerError)
return
}
// Check authorization:
// 1. If allowAllCrew is true, any authenticated user can join
// 2. If user is the owner, they can always join (though they should already be crew)
// 3. Otherwise, deny
isOwner := user.DID == captain.Owner
if !captain.AllowAllCrew && !isOwner {
http.Error(w, "crew registration not allowed (HOLD_ALLOW_ALL_CREW=false)", http.StatusForbidden)
return
}
// Set defaults if not provided
if req.Role == "" {
req.Role = "member"
}
if len(req.Permissions) == 0 {
req.Permissions = []string{"blob:read", "blob:write"}
}
// Check if user is already a crew member
// List all crew members and check if this DID is already present
crew, err := h.pds.ListCrewMembers(r.Context())
if err != nil {
http.Error(w, fmt.Sprintf("failed to list crew members: %v", err), http.StatusInternalServerError)
return
}
for _, member := range crew {
if member.Record.Member == user.DID {
// Already a crew member, return success with existing record
response := map[string]any{
"uri": fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), CrewCollection, member.Rkey),
"cid": member.Cid.String(),
"status": "already_member",
"message": "User is already a crew member",
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(response)
return
}
}
// Create new crew record
recordCID, err := h.pds.AddCrewMember(r.Context(), user.DID, req.Role, req.Permissions)
if err != nil {
http.Error(w, fmt.Sprintf("failed to create crew record: %v", err), http.StatusInternalServerError)
return
}
// Return success response
// Note: rkey is generated by AddCrewMember (TID), we don't have direct access to it
// For now, return just the CID. In production, AddCrewMember should return both CID and rkey
response := map[string]any{
"cid": recordCID.String(),
"status": "created",
"message": "Successfully added to crew",
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusCreated)
json.NewEncoder(w).Encode(response)
}
pkg/hold/registration.go
-481
View File
@@ -1,481 +0,0 @@
package hold
import (
"context"
"encoding/json"
"errors"
"fmt"
"log"
"net/http"
"net/url"
"strings"
"time"
"atcr.io/pkg/atproto"
"atcr.io/pkg/auth/oauth"
"github.com/bluesky-social/indigo/atproto/identity"
"github.com/bluesky-social/indigo/atproto/syntax"
)
// HealthHandler handles health check requests
func (s *HoldService) HealthHandler(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.Write([]byte(`{"status":"ok"}`))
}
// isHoldRegistered checks if a hold with the given public URL is already registered in the PDS
func (s *HoldService) isHoldRegistered(ctx context.Context, did, pdsEndpoint, publicURL string) (bool, error) {
// We need to query the PDS without authentication to check public records
// ATProto records are publicly readable, so we can use an unauthenticated client
client := atproto.NewClient(pdsEndpoint, did, "")
// List all hold records for this DID
records, err := client.ListRecords(ctx, atproto.HoldCollection, 100)
if err != nil {
return false, fmt.Errorf("failed to list hold records: %w", err)
}
// Check if any hold record matches our public URL
for _, record := range records {
var holdRecord atproto.HoldRecord
if err := json.Unmarshal(record.Value, &holdRecord); err != nil {
continue
}
if holdRecord.Endpoint == publicURL {
return true, nil
}
}
return false, nil
}
// AutoRegister registers this hold service in the owner's PDS
// Checks if already registered first, then does OAuth if needed
func (s *HoldService) AutoRegister(callbackHandler *http.HandlerFunc) error {
reg := &s.config.Registration
publicURL := s.config.Server.PublicURL
if publicURL == "" {
return fmt.Errorf("HOLD_PUBLIC_URL not set")
}
if reg.OwnerDID == "" {
return fmt.Errorf("HOLD_OWNER not set - required for registration")
}
ctx := context.Background()
log.Printf("Checking registration status for DID: %s", reg.OwnerDID)
// Resolve DID to PDS endpoint using indigo
directory := identity.DefaultDirectory()
didParsed, err := syntax.ParseDID(reg.OwnerDID)
if err != nil {
return fmt.Errorf("invalid owner DID: %w", err)
}
ident, err := directory.LookupDID(ctx, didParsed)
if err != nil {
return fmt.Errorf("failed to resolve PDS for DID: %w", err)
}
pdsEndpoint := ident.PDSEndpoint()
if pdsEndpoint == "" {
return fmt.Errorf("no PDS endpoint found for DID")
}
log.Printf("PDS endpoint: %s", pdsEndpoint)
// Check if hold is already registered
isRegistered, err := s.isHoldRegistered(ctx, reg.OwnerDID, pdsEndpoint, publicURL)
if err != nil {
log.Printf("Warning: failed to check registration status: %v", err)
log.Printf("Proceeding with OAuth registration...")
} else if isRegistered {
log.Printf("✓ Hold service already registered in PDS")
log.Printf("Public URL: %s", publicURL)
return nil
}
// Not registered, need to do OAuth
log.Printf("Hold not registered, starting OAuth flow...")
// Get handle from DID document (already resolved above)
handle := ident.Handle.String()
if handle == "" || handle == "handle.invalid" {
return fmt.Errorf("no valid handle found for DID")
}
log.Printf("Resolved handle: %s", handle)
log.Printf("Starting OAuth registration for hold service")
log.Printf("Public URL: %s", publicURL)
return s.registerWithOAuth(publicURL, handle, reg.OwnerDID, pdsEndpoint, callbackHandler)
}
// registerWithOAuth performs OAuth flow and registers the hold
func (s *HoldService) registerWithOAuth(publicURL, handle, did, pdsEndpoint string, callbackHandler *http.HandlerFunc) error {
// Run OAuth flow to get authenticated client
client, err := s.runOAuthFlow(callbackHandler, "Hold service registration")
if err != nil {
return err
}
log.Printf("Authorization received!")
log.Printf("OAuth session obtained successfully")
log.Printf("DID: %s", did)
log.Printf("PDS: %s", pdsEndpoint)
return s.registerWithClient(publicURL, did, client)
}
// registerWithClient registers the hold using an authenticated ATProto client
func (s *HoldService) registerWithClient(publicURL, did string, client *atproto.Client) error {
// Derive hold name from URL (hostname)
holdName, err := extractHostname(publicURL)
if err != nil {
return fmt.Errorf("failed to extract hostname from URL: %w", err)
}
log.Printf("Registering hold service: url=%s, name=%s, owner=%s", publicURL, holdName, did)
ctx := context.Background()
// Create HoldRecord
holdRecord := atproto.NewHoldRecord(publicURL, did, s.config.Server.Public)
// Use hostname as record key
holdResult, err := client.PutRecord(ctx, atproto.HoldCollection, holdName, holdRecord)
if err != nil {
return fmt.Errorf("failed to create hold record: %w", err)
}
log.Printf("✓ Created hold record: %s", holdResult.URI)
// Create HoldCrewRecord for the owner
crewRecord := atproto.NewHoldCrewRecord(holdResult.URI, did, "owner")
crewRKey := fmt.Sprintf("%s-%s", holdName, did)
crewResult, err := client.PutRecord(ctx, atproto.HoldCrewCollection, crewRKey, crewRecord)
if err != nil {
return fmt.Errorf("failed to create crew record: %w", err)
}
log.Printf("✓ Created crew record: %s", crewResult.URI)
// Update sailor profile to set this as the default hold
profile, err := atproto.GetProfile(ctx, client)
if err != nil {
log.Printf("Warning: failed to get sailor profile: %v", err)
} else {
if profile == nil {
// Create new profile with this hold as default
profile = atproto.NewSailorProfileRecord(publicURL)
} else {
// Update existing profile with new defaultHold
profile.DefaultHold = publicURL
profile.UpdatedAt = time.Now()
}
err = atproto.UpdateProfile(ctx, client, profile)
if err != nil {
log.Printf("Warning: failed to update sailor profile: %v", err)
} else {
log.Printf("✓ Updated sailor profile defaultHold: %s", publicURL)
}
}
log.Print("\n" + strings.Repeat("=", 80))
log.Printf("REGISTRATION COMPLETE")
log.Print(strings.Repeat("=", 80))
log.Printf("Hold service is now registered and ready to use!")
log.Print(strings.Repeat("=", 80) + "\n")
return nil
}
// extractHostname extracts the hostname from a URL to use as the hold name
func extractHostname(urlStr string) (string, error) {
u, err := url.Parse(urlStr)
if err != nil {
return "", err
}
// Remove port if present
hostname := u.Hostname()
if hostname == "" {
return "", fmt.Errorf("no hostname in URL")
}
return hostname, nil
}
// ReconcileAllowAllCrew reconciles the allow-all crew record state with the environment variable
// Called on every startup to ensure the PDS record matches the desired configuration
func (s *HoldService) ReconcileAllowAllCrew(callbackHandler *http.HandlerFunc) error {
ownerDID := s.config.Registration.OwnerDID
if ownerDID == "" {
// No owner DID configured, skip reconciliation
return nil
}
desiredState := s.config.Registration.AllowAllCrew
log.Printf("Checking allow-all crew state (desired: %v)", desiredState)
// Query PDS for current state
actualState, err := s.hasAllowAllCrewRecord()
if err != nil {
return fmt.Errorf("failed to check allow-all crew record: %w", err)
}
log.Printf("Allow-all crew record exists: %v", actualState)
// States match - nothing to do
if desiredState == actualState {
if desiredState {
log.Printf("✓ Allow-all crew enabled (all authenticated users can push)")
} else {
log.Printf("✓ Allow-all crew disabled (explicit crew membership required)")
}
return nil
}
// State mismatch - need to reconcile
if desiredState && !actualState {
// Need to create wildcard crew record
log.Printf("Creating allow-all crew record (HOLD_ALLOW_ALL_CREW=true)")
return s.createAllowAllCrewRecord(callbackHandler)
}
if !desiredState && actualState {
// Need to delete wildcard crew record
log.Printf("Deleting allow-all crew record (HOLD_ALLOW_ALL_CREW=false)")
return s.deleteAllowAllCrewRecord(callbackHandler)
}
return nil
}
// hasAllowAllCrewRecord checks if the allow-all crew record exists in the PDS for THIS hold
func (s *HoldService) hasAllowAllCrewRecord() (bool, error) {
ownerDID := s.config.Registration.OwnerDID
publicURL := s.config.Server.PublicURL
if ownerDID == "" {
return false, fmt.Errorf("hold owner DID not configured")
}
if publicURL == "" {
return false, fmt.Errorf("hold public URL not configured")
}
ctx := context.Background()
// Resolve owner's PDS endpoint
directory := identity.DefaultDirectory()
ownerDIDParsed, err := syntax.ParseDID(ownerDID)
if err != nil {
return false, fmt.Errorf("invalid owner DID: %w", err)
}
ident, err := directory.LookupDID(ctx, ownerDIDParsed)
if err != nil {
return false, fmt.Errorf("failed to resolve owner PDS: %w", err)
}
pdsEndpoint := ident.PDSEndpoint()
if pdsEndpoint == "" {
return false, fmt.Errorf("no PDS endpoint found for owner")
}
// Build hold-specific rkey
holdName, err := extractHostname(publicURL)
if err != nil {
return false, fmt.Errorf("failed to extract hostname: %w", err)
}
crewRKey := fmt.Sprintf("allow-all-%s", holdName)
// Create unauthenticated client to read public records
client := atproto.NewClient(pdsEndpoint, ownerDID, "")
// Query for hold-specific allow-all record
record, err := client.GetRecord(ctx, atproto.HoldCrewCollection, crewRKey)
if err != nil {
// Record doesn't exist
if errors.Is(err, atproto.ErrRecordNotFound) {
return false, nil
}
return false, fmt.Errorf("failed to get crew record: %w", err)
}
// Verify it's the wildcard record (memberPattern: "*")
var crewRecord atproto.HoldCrewRecord
if err := json.Unmarshal(record.Value, &crewRecord); err != nil {
return false, fmt.Errorf("failed to unmarshal crew record: %w", err)
}
// Check if it's the exact wildcard pattern
if crewRecord.MemberPattern == nil || *crewRecord.MemberPattern != "*" {
return false, nil
}
// Verify it's for this hold (defensive check)
expectedHoldURI := fmt.Sprintf("at://%s/%s/%s", ownerDID, atproto.HoldCollection, holdName)
return crewRecord.Hold == expectedHoldURI, nil
}
// createAllowAllCrewRecord creates a wildcard crew record allowing all authenticated users
func (s *HoldService) createAllowAllCrewRecord(callbackHandler *http.HandlerFunc) error {
ownerDID := s.config.Registration.OwnerDID
publicURL := s.config.Server.PublicURL
// Run OAuth flow to get authenticated client
client, err := s.runOAuthFlow(callbackHandler, "Creating allow-all crew record")
if err != nil {
return err
}
ctx := context.Background()
// Get hold URI
holdName, err := extractHostname(publicURL)
if err != nil {
return fmt.Errorf("failed to extract hostname: %w", err)
}
holdURI := fmt.Sprintf("at://%s/%s/%s", ownerDID, atproto.HoldCollection, holdName)
// Create wildcard crew record
crewRecord := atproto.NewHoldCrewRecordWithPattern(holdURI, "*", "write")
// Use hold-specific rkey to support multiple holds with different allow-all settings
crewRKey := fmt.Sprintf("allow-all-%s", holdName)
_, err = client.PutRecord(ctx, atproto.HoldCrewCollection, crewRKey, crewRecord)
if err != nil {
return fmt.Errorf("failed to create allow-all crew record: %w", err)
}
log.Printf("✓ Created allow-all crew record (allows all authenticated users)")
return nil
}
// deleteAllowAllCrewRecord deletes the wildcard crew record for this hold
func (s *HoldService) deleteAllowAllCrewRecord(callbackHandler *http.HandlerFunc) error {
// Safety check: only delete if it's the exact wildcard pattern for THIS hold
isWildcard, err := s.hasAllowAllCrewRecord()
if err != nil {
return fmt.Errorf("failed to check allow-all crew record: %w", err)
}
if !isWildcard {
log.Printf("Note: 'allow-all' crew record not found for this hold (may exist for other holds)")
return nil
}
// Get hold name for rkey
holdName, err := extractHostname(s.config.Server.PublicURL)
if err != nil {
return fmt.Errorf("failed to extract hostname: %w", err)
}
crewRKey := fmt.Sprintf("allow-all-%s", holdName)
// Run OAuth flow to get authenticated client
client, err := s.runOAuthFlow(callbackHandler, "Deleting allow-all crew record")
if err != nil {
return err
}
ctx := context.Background()
// Delete the hold-specific allow-all record
err = client.DeleteRecord(ctx, atproto.HoldCrewCollection, crewRKey)
if err != nil {
return fmt.Errorf("failed to delete allow-all crew record: %w", err)
}
log.Printf("✓ Deleted allow-all crew record for this hold")
return nil
}
// getHoldRegistrationScopes returns the OAuth scopes needed for hold registration and crew management
func getHoldRegistrationScopes() []string {
return []string{
"atproto",
fmt.Sprintf("repo:%s", atproto.HoldCollection),
fmt.Sprintf("repo:%s", atproto.HoldCrewCollection),
fmt.Sprintf("repo:%s", atproto.SailorProfileCollection),
}
}
// runOAuthFlow performs OAuth flow and returns an authenticated client
// Reusable helper to avoid code duplication across registration and reconciliation
func (s *HoldService) runOAuthFlow(callbackHandler *http.HandlerFunc, purpose string) (*atproto.Client, error) {
ownerDID := s.config.Registration.OwnerDID
publicURL := s.config.Server.PublicURL
ctx := context.Background()
// Resolve owner's PDS endpoint
directory := identity.DefaultDirectory()
ownerDIDParsed, err := syntax.ParseDID(ownerDID)
if err != nil {
return nil, fmt.Errorf("invalid owner DID: %w", err)
}
ident, err := directory.LookupDID(ctx, ownerDIDParsed)
if err != nil {
return nil, fmt.Errorf("failed to resolve owner PDS: %w", err)
}
pdsEndpoint := ident.PDSEndpoint()
if pdsEndpoint == "" {
return nil, fmt.Errorf("no PDS endpoint found for owner")
}
handle := ident.Handle.String()
if handle == "" || handle == "handle.invalid" {
return nil, fmt.Errorf("no valid handle found for DID")
}
// Determine base URL for OAuth
var baseURL string
if s.config.Server.TestMode {
parsedURL, err := url.Parse(publicURL)
if err != nil {
return nil, fmt.Errorf("failed to parse public URL: %w", err)
}
port := parsedURL.Port()
if port == "" {
port = "8080"
}
baseURL = fmt.Sprintf("http://127.0.0.1:%s", port)
} else {
baseURL = publicURL
}
// Run OAuth flow
result, err := oauth.InteractiveFlowWithCallback(
ctx,
baseURL,
handle,
getHoldRegistrationScopes(),
func(handler http.HandlerFunc) error {
*callbackHandler = handler
return nil
},
func(authURL string) error {
log.Print("\n" + strings.Repeat("=", 80))
log.Printf("OAUTH REQUIRED: %s", purpose)
log.Print(strings.Repeat("=", 80))
log.Printf("\nVisit: %s\n", authURL)
log.Printf("Waiting for authorization...")
log.Print(strings.Repeat("=", 80) + "\n")
return nil
},
)
if err != nil {
return nil, fmt.Errorf("OAuth flow failed: %w", err)
}
// Create authenticated client
apiClient := result.Session.APIClient()
return atproto.NewClientWithIndigoClient(pdsEndpoint, ownerDID, apiClient), nil
}
pkg/hold/service.go
+22
View File
@@ -4,6 +4,8 @@ import (
"context"
"fmt"
"log"
"net/http"
"net/url"
"github.com/aws/aws-sdk-go/service/s3"
storagedriver "github.com/distribution/distribution/v3/registry/storage/driver"
@@ -47,3 +49,23 @@ func NewHoldService(cfg *Config) (*HoldService, error) {
func (s *HoldService) GetPresignedURL(ctx context.Context, operation PresignedURLOperation, digest string, did string) (string, error) {
return s.getPresignedURL(ctx, operation, digest, did)
}
// HealthHandler handles health check requests
func (s *HoldService) HealthHandler(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.Write([]byte(`{"status":"ok"}`))
}
// extractHostname extracts the hostname from a URL
func extractHostname(urlStr string) (string, error) {
u, err := url.Parse(urlStr)
if err != nil {
return "", err
}
// Remove port if present
hostname := u.Hostname()
if hostname == "" {
return "", fmt.Errorf("no hostname in URL")
}
return hostname, nil
}