diff --git a/.env.example b/.env.example index 8796bd9..3f3c8f5 100644 --- a/.env.example +++ b/.env.example @@ -16,7 +16,7 @@ HOLD_PUBLIC_URL=http://127.0.0.1:8080 # Storage driver type (s3, filesystem) # Default: s3 -STORAGE_DRIVER=s3 +STORAGE_DRIVER=filesystem # For S3/Storj/Minio: AWS_ACCESS_KEY_ID=your_access_key @@ -50,7 +50,7 @@ HOLD_PUBLIC=false # Your ATProto DID (REQUIRED for registration) # Get your DID: https://bsky.social/xrpc/com.atproto.identity.resolveHandle?handle=yourhandle.bsky.social # -# On first run with HOLD_CREW_OWNER set: +# On first run with HOLD_OWNER set: # 1. Hold service will print an OAuth URL to the logs # 2. Visit the URL in your browser to authorize # 3. Hold service creates hold + crew records in your PDS @@ -60,4 +60,4 @@ HOLD_PUBLIC=false # - Hold service checks if already registered # - Skips OAuth if records exist # -HOLD_CREW_OWNER=did:plc:your-did-here +HOLD_OWNER=did:plc:your-did-here diff --git a/CLAUDE.md b/CLAUDE.md index 49bb921..1b9b537 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -40,7 +40,7 @@ export ATPROTO_ACCESS_TOKEN=your-token export HOLD_PUBLIC_URL=http://127.0.0.1:8080 export STORAGE_DRIVER=filesystem export STORAGE_ROOT_DIR=/tmp/atcr-hold -export HOLD_CREW_OWNER=did:plc:your-did-here +export HOLD_OWNER=did:plc:your-did-here ./atcr-hold # Check logs for OAuth URL, visit in browser to complete registration ``` @@ -299,7 +299,7 @@ Key insight: "Private" gates anonymous access, not authenticated access. This re - `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY` - S3 credentials - `S3_BUCKET`, `S3_ENDPOINT` - S3 configuration - `HOLD_PUBLIC` - Allow public reads (default: false) -- `HOLD_CREW_OWNER` - DID for auto-registration (optional) +- `HOLD_OWNER` - DID for auto-registration (optional) **Deployment:** Can run on Fly.io, Railway, Docker, Kubernetes, etc. @@ -386,7 +386,7 @@ This ensures: - Storage driver config via env vars: `STORAGE_DRIVER`, `AWS_*`, `S3_*` - Authorization: Based on PDS records (`hold.public`, crew records) - Server settings: `HOLD_SERVER_ADDR`, `HOLD_PUBLIC_URL`, `HOLD_PUBLIC` -- Auto-registration: `HOLD_CREW_OWNER` (optional) +- Auto-registration: `HOLD_OWNER` (optional) **Credential Helper**: - Token storage: `~/.atcr/oauth-token.json` @@ -439,7 +439,7 @@ When writing tests: **Adding BYOS support for a user**: 1. User sets environment variables (storage credentials, public URL) -2. User runs hold service with `HOLD_CREW_OWNER` set - auto-registration via OAuth +2. User runs hold service with `HOLD_OWNER` set - auto-registration via OAuth 3. Hold service creates `io.atcr.hold` + `io.atcr.hold.crew` records in PDS 4. AppView automatically queries PDS and routes blobs to user's storage 5. No AppView changes needed - fully decentralized diff --git a/SPEC.md b/SPEC.md index cdff788..9884f3c 100644 --- a/SPEC.md +++ b/SPEC.md @@ -431,7 +431,7 @@ ATProto OAuth Implementation Plan Unified Model - Every hold service requires HOLD_CREW_OWNER: + Every hold service requires HOLD_OWNER: - Owner's PDS has the io.atcr.hold record - Owner's PDS has all io.atcr.hold.crew records - Authorization is always governed by PDS records diff --git a/cmd/hold/main.go b/cmd/hold/main.go index a7ff4df..55e2563 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -34,7 +34,7 @@ type Config struct { // RegistrationConfig defines auto-registration settings type RegistrationConfig struct { - // OwnerDID is the owner's ATProto DID (from env: HOLD_CREW_OWNER) + // OwnerDID is the owner's ATProto DID (from env: HOLD_OWNER) // If set, auto-registration is enabled OwnerDID string `yaml:"owner_did"` } @@ -55,6 +55,9 @@ type ServerConfig struct { // Public controls whether this hold allows public blob reads without auth (from env: HOLD_PUBLIC) Public bool `yaml:"public"` + // TestMode uses localhost for OAuth redirects while storing real URL in hold record (from env: TEST_MODE) + TestMode bool `yaml:"test_mode"` + // ReadTimeout for HTTP requests ReadTimeout time.Duration `yaml:"read_timeout"` @@ -64,8 +67,12 @@ type ServerConfig struct { // HoldService provides presigned URLs for blob storage in a hold type HoldService struct { - driver storagedriver.StorageDriver - config *Config + driver storagedriver.StorageDriver + config *Config + oauthCodeCh chan string + oauthErrCh chan error + oauthState string + codeVerifier string } // NewHoldService creates a new hold service @@ -78,8 +85,10 @@ func NewHoldService(cfg *Config) (*HoldService, error) { } return &HoldService{ - driver: driver, - config: cfg, + driver: driver, + config: cfg, + oauthCodeCh: make(chan string, 1), + oauthErrCh: make(chan error, 1), }, nil } @@ -183,7 +192,7 @@ func (s *HoldService) HandlePutPresignedURL(w http.ResponseWriter, r *http.Reque ctx := context.Background() expiry := time.Now().Add(15 * time.Minute) - url, err := s.getUploadURL(ctx, req.Digest, req.Size) + url, err := s.getUploadURL(ctx, req.Digest, req.Size, req.DID) if err != nil { http.Error(w, fmt.Sprintf("failed to generate URL: %v", err), http.StatusInternalServerError) return @@ -200,7 +209,7 @@ func (s *HoldService) HandlePutPresignedURL(w http.ResponseWriter, r *http.Reque // HandleProxyGet proxies a blob download through the service func (s *HoldService) HandleProxyGet(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { + if r.Method != http.MethodGet && r.Method != http.MethodHead { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } @@ -228,10 +237,23 @@ func (s *HoldService) HandleProxyGet(w http.ResponseWriter, r *http.Request) { return } - // Read blob from storage ctx := r.Context() - path := fmt.Sprintf("/docker/registry/v2/blobs/%s/data", digest) + path := blobPath(digest) + // For HEAD requests, just check if blob exists + if r.Method == http.MethodHead { + stat, err := s.driver.Stat(ctx, path) + if err != nil { + http.Error(w, "blob not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Length", fmt.Sprintf("%d", stat.Size())) + w.WriteHeader(http.StatusOK) + return + } + + // For GET requests, read and return the blob content, err := s.driver.GetContent(ctx, path) if err != nil { http.Error(w, "blob not found", http.StatusNotFound) @@ -244,6 +266,8 @@ func (s *HoldService) HandleProxyGet(w http.ResponseWriter, r *http.Request) { // HandleProxyPut proxies a blob upload through the service func (s *HoldService) HandleProxyPut(w http.ResponseWriter, r *http.Request) { + log.Printf("HandleProxyPut: method=%s, path=%s, query=%s", r.Method, r.URL.Path, r.URL.RawQuery) + if r.Method != http.MethodPut { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return @@ -260,11 +284,17 @@ func (s *HoldService) HandleProxyPut(w http.ResponseWriter, r *http.Request) { did = r.Header.Get("X-ATCR-DID") } + log.Printf("HandleProxyPut: digest=%s, did=%s", digest, did) + // Authorize WRITE access - if !s.isAuthorizedWrite(did) { + authorized := s.isAuthorizedWrite(did) + log.Printf("HandleProxyPut: authorization check: did=%s, authorized=%v", did, authorized) + if !authorized { if did == "" { + log.Printf("HandleProxyPut: rejecting - no DID provided") http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized) } else { + log.Printf("HandleProxyPut: rejecting - DID not authorized for write") http.Error(w, "forbidden: write access denied", http.StatusForbidden) } return @@ -272,19 +302,23 @@ func (s *HoldService) HandleProxyPut(w http.ResponseWriter, r *http.Request) { // Write blob to storage ctx := r.Context() - path := fmt.Sprintf("/docker/registry/v2/blobs/%s/data", digest) + path := blobPath(digest) content, err := io.ReadAll(r.Body) if err != nil { + log.Printf("HandleProxyPut: failed to read body: %v", err) http.Error(w, "failed to read body", http.StatusBadRequest) return } + log.Printf("HandleProxyPut: writing blob to path=%s, size=%d bytes", path, len(content)) if err := s.driver.PutContent(ctx, path, content); err != nil { + log.Printf("HandleProxyPut: failed to store blob: %v", err) http.Error(w, "failed to store blob", http.StatusInternalServerError) return } + log.Printf("HandleProxyPut: successfully stored blob digest=%s, size=%d", digest, len(content)) w.WriteHeader(http.StatusCreated) } @@ -400,7 +434,7 @@ func (s *HoldService) isCrewMember(did string) (bool, error) { // getDownloadURL generates a download URL for a blob func (s *HoldService) getDownloadURL(ctx context.Context, digest string) (string, error) { // Check if blob exists - path := fmt.Sprintf("/docker/registry/v2/blobs/%s/data", digest) + path := blobPath(digest) _, err := s.driver.Stat(ctx, path) if err != nil { return "", fmt.Errorf("blob not found: %w", err) @@ -408,14 +442,15 @@ func (s *HoldService) getDownloadURL(ctx context.Context, digest string) (string // For drivers that support presigned URLs (S3), use those // For now, return a proxy URL through this service - return fmt.Sprintf("http://%s/blobs/%s", s.config.Server.Addr, digest), nil + return fmt.Sprintf("%s/blobs/%s", s.config.Server.PublicURL, digest), nil } // getUploadURL generates an upload URL for a blob -func (s *HoldService) getUploadURL(ctx context.Context, digest string, size int64) (string, error) { +// Note: This is called from HandlePutPresignedURL which has the DID in the request +func (s *HoldService) getUploadURL(ctx context.Context, digest string, size int64, did string) (string, error) { // For drivers that support presigned URLs (S3), use those - // For now, return a proxy URL through this service - return fmt.Sprintf("http://%s/blobs/%s", s.config.Server.Addr, digest), nil + // For now, return a proxy URL through this service with DID for authorization + return fmt.Sprintf("%s/blobs/%s?did=%s", s.config.Server.PublicURL, digest, did), nil } // RegisterRequest represents a request to register this hold in a user's PDS @@ -515,6 +550,34 @@ func (s *HoldService) HealthHandler(w http.ResponseWriter, r *http.Request) { }) } +// HandleOAuthCallback handles OAuth callback from authorization server +func (s *HoldService) HandleOAuthCallback(w http.ResponseWriter, r *http.Request) { + code := r.URL.Query().Get("code") + receivedState := r.URL.Query().Get("state") + + if receivedState != s.oauthState { + s.oauthErrCh <- fmt.Errorf("invalid state parameter") + http.Error(w, "Invalid state", http.StatusBadRequest) + return + } + + if code == "" { + s.oauthErrCh <- fmt.Errorf("no authorization code received") + http.Error(w, "No code", http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "text/html") + fmt.Fprintf(w, `

Authorization Successful!

You can close this window and return to the terminal.

`) + + // Send code to registration flow + select { + case s.oauthCodeCh <- code: + default: + // Channel already has a value or nobody is listening + } +} + func main() { // Load configuration from environment variables cfg, err := loadConfigFromEnv() @@ -528,24 +591,22 @@ func main() { log.Fatalf("Failed to create hold service: %v", err) } - // Auto-register if owner DID is set - if cfg.Registration.OwnerDID != "" { - if err := service.AutoRegister(); 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") - } - } - // Setup HTTP routes mux := http.NewServeMux() mux.HandleFunc("/health", service.HealthHandler) mux.HandleFunc("/register", service.HandleRegister) mux.HandleFunc("/get-presigned-url", service.HandleGetPresignedURL) mux.HandleFunc("/put-presigned-url", service.HandlePutPresignedURL) + mux.HandleFunc("/oauth/callback", service.HandleOAuthCallback) // OAuth callback on same port + + // OAuth client metadata endpoint for ATProto OAuth + clientID := cfg.Server.PublicURL + "/client-metadata.json" + clientMetadata := oauth.NewClientMetadata(clientID, []string{cfg.Server.PublicURL + "/oauth/callback"}) + clientMetadata.ClientName = "ATCR Hold Service" + clientMetadata.ApplicationType = "web" // Changed from "native" since this is a web service + mux.HandleFunc("/client-metadata.json", oauth.ServeMetadata(clientMetadata)) mux.HandleFunc("/blobs/", func(w http.ResponseWriter, r *http.Request) { - if r.Method == http.MethodGet { + if r.Method == http.MethodGet || r.Method == http.MethodHead { service.HandleProxyGet(w, r) } else if r.Method == http.MethodPut { service.HandleProxyPut(w, r) @@ -562,8 +623,30 @@ func main() { WriteTimeout: cfg.Server.WriteTimeout, } - log.Printf("Starting hold service on %s", cfg.Server.Addr) - if err := server.ListenAndServe(); err != nil { + // Start server in goroutine so we can do auto-registration after it's running + serverErr := make(chan error, 1) + go func() { + log.Printf("Starting hold service on %s", cfg.Server.Addr) + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + serverErr <- err + } + }() + + // 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(); 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") + } + } + + // Wait for server error or shutdown + if err := <-serverErr; err != nil { log.Fatalf("Server failed: %v", err) } } @@ -581,11 +664,12 @@ func loadConfigFromEnv() (*Config, error) { return nil, fmt.Errorf("HOLD_PUBLIC_URL is required") } cfg.Server.Public = os.Getenv("HOLD_PUBLIC") == "true" - cfg.Server.ReadTimeout = 30 * time.Second - cfg.Server.WriteTimeout = 30 * time.Second + cfg.Server.TestMode = os.Getenv("TEST_MODE") == "true" + cfg.Server.ReadTimeout = 5 * time.Minute // Increased for large blob uploads + cfg.Server.WriteTimeout = 5 * time.Minute // Increased for large blob uploads // Registration configuration (optional) - cfg.Registration.OwnerDID = os.Getenv("HOLD_CREW_OWNER") + cfg.Registration.OwnerDID = os.Getenv("HOLD_OWNER") // Storage configuration - build from env vars based on storage type storageType := getEnvOrDefault("STORAGE_DRIVER", "s3") @@ -647,6 +731,29 @@ func getEnvOrDefault(key, defaultValue string) string { return defaultValue } +// blobPath converts a digest (e.g., "sha256:abc123...") to a storage path +// Distribution stores blobs as: /docker/registry/v2/blobs/{algorithm}/{xx}/{hash}/data +// where xx is the first 2 characters of the hash for directory sharding +// NOTE: Path must start with / for filesystem driver +func blobPath(digest string) string { + // Split digest into algorithm and hash + parts := strings.SplitN(digest, ":", 2) + if len(parts) != 2 { + // Fallback for malformed digest + return fmt.Sprintf("/docker/registry/v2/blobs/%s/data", digest) + } + + algorithm := parts[0] + hash := parts[1] + + // Use first 2 characters for sharding + if len(hash) < 2 { + return fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/data", algorithm, hash) + } + + return fmt.Sprintf("/docker/registry/v2/blobs/%s/%s/%s/data", algorithm, hash[:2], hash) +} + // 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 @@ -685,7 +792,7 @@ func (s *HoldService) AutoRegister() error { } if reg.OwnerDID == "" { - return fmt.Errorf("HOLD_CREW_OWNER not set - required for registration") + return fmt.Errorf("HOLD_OWNER not set - required for registration") } ctx := context.Background() @@ -730,12 +837,43 @@ func (s *HoldService) AutoRegister() error { // registerWithOAuth performs OAuth flow and registers the hold func (s *HoldService) registerWithOAuth(publicURL, handle, did, pdsEndpoint string) error { - // Use 127.0.0.1 for localhost callback (works better than "localhost") - callbackAddr := "127.0.0.1:8888" - redirectURI := fmt.Sprintf("http://%s/callback", callbackAddr) + // Extract port from publicURL for test mode + var redirectURI string + var clientID string + + // Define the scopes we need for hold registration + // Need create and update permissions for hold and crew collections + scopes := fmt.Sprintf("atproto repo:%s?action=create repo:%s?action=update repo:%s?action=create repo:%s?action=update", + atproto.HoldCollection, atproto.HoldCollection, + atproto.HoldCrewCollection, atproto.HoldCrewCollection) + + if s.config.Server.TestMode { + // Test mode: Use localhost for OAuth (browser accessible) but store real URL in hold record + // Extract port from publicURL (e.g., "http://172.28.0.3:8080" -> ":8080") + parsedURL, err := url.Parse(publicURL) + if err != nil { + return fmt.Errorf("failed to parse public URL: %w", err) + } + port := parsedURL.Port() + if port == "" { + port = "8080" // default + } + redirectURI = fmt.Sprintf("http://127.0.0.1:%s/oauth/callback", port) + clientID = fmt.Sprintf("http://localhost?redirect_uri=%s&scope=%s", + url.QueryEscape(redirectURI), url.QueryEscape(scopes)) + } else if strings.Contains(publicURL, "127.0.0.1") || strings.Contains(publicURL, "localhost") { + // Localhost development mode per ATProto OAuth spec + redirectURI = publicURL + "/oauth/callback" + clientID = fmt.Sprintf("http://localhost?redirect_uri=%s&scope=%s", + url.QueryEscape(redirectURI), url.QueryEscape(scopes)) + } else { + // Production mode - use client metadata URL + redirectURI = publicURL + "/oauth/callback" + clientID = publicURL + "/client-metadata.json" + } // Create OAuth client - oauthClient, err := oauth.NewClient("http://hold-service", redirectURI) + oauthClient, err := oauth.NewClient(clientID, redirectURI) if err != nil { return fmt.Errorf("failed to create OAuth client: %w", err) } @@ -746,12 +884,22 @@ func (s *HoldService) registerWithOAuth(publicURL, handle, did, pdsEndpoint stri return fmt.Errorf("failed to initialize OAuth: %w", err) } + // Set the scopes we need for hold registration (create and update) + oauthClient.SetScopes([]string{ + "atproto", + fmt.Sprintf("repo:%s?action=create", atproto.HoldCollection), + fmt.Sprintf("repo:%s?action=update", atproto.HoldCollection), + fmt.Sprintf("repo:%s?action=create", atproto.HoldCrewCollection), + fmt.Sprintf("repo:%s?action=update", atproto.HoldCrewCollection), + }) + // Generate authorization URL - state := "hold-registration" - authURL, codeVerifier, err := oauthClient.AuthorizeURL(state) + s.oauthState = "hold-registration" + authURL, codeVerifier, err := oauthClient.AuthorizeURL(s.oauthState) if err != nil { return fmt.Errorf("failed to generate auth URL: %w", err) } + s.codeVerifier = codeVerifier // Print the OAuth URL for user to visit log.Print("\n" + strings.Repeat("=", 80)) @@ -762,61 +910,21 @@ func (s *HoldService) registerWithOAuth(publicURL, handle, did, pdsEndpoint stri log.Printf("Waiting for authorization...") log.Print(strings.Repeat("=", 80) + "\n") - // Start temporary HTTP server for callback - codeChan := make(chan string, 1) - errChan := make(chan error, 1) - - mux := http.NewServeMux() - mux.HandleFunc("/callback", func(w http.ResponseWriter, r *http.Request) { - code := r.URL.Query().Get("code") - receivedState := r.URL.Query().Get("state") - - if receivedState != state { - errChan <- fmt.Errorf("invalid state parameter") - http.Error(w, "Invalid state", http.StatusBadRequest) - return - } - - if code == "" { - errChan <- fmt.Errorf("no authorization code received") - http.Error(w, "No code", http.StatusBadRequest) - return - } - - w.Header().Set("Content-Type", "text/html") - fmt.Fprintf(w, `

Authorization Successful!

You can close this window and return to the terminal.

`) - codeChan <- code - }) - - server := &http.Server{ - Addr: callbackAddr, - Handler: mux, - } - - go func() { - if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { - errChan <- err - } - }() - - // Wait for callback or error + // Wait for callback or error (callback happens on main server) var code string select { - case code = <-codeChan: - // Got the code, shutdown callback server - server.Shutdown(context.Background()) - case err := <-errChan: - server.Shutdown(context.Background()) + case code = <-s.oauthCodeCh: + // Got the code from callback + case err := <-s.oauthErrCh: return err case <-time.After(5 * time.Minute): - server.Shutdown(context.Background()) return fmt.Errorf("OAuth timeout - no response after 5 minutes") } log.Printf("Authorization received, exchanging code for token...") // Exchange code for token - token, err := oauthClient.Exchange(ctx, code, codeVerifier) + token, err := oauthClient.Exchange(ctx, code, s.codeVerifier) if err != nil { return fmt.Errorf("failed to exchange code: %w", err) } @@ -825,12 +933,19 @@ func (s *HoldService) registerWithOAuth(publicURL, handle, did, pdsEndpoint stri log.Printf("DID: %s", did) log.Printf("PDS: %s", pdsEndpoint) - // Now register with the token - return s.registerWithToken(publicURL, did, pdsEndpoint, token.AccessToken) + // Now register with the token using DPoP + // Create ATProto client with DPoP transport from OAuth client + dpopKey := oauthClient.DPoPKey() + dpopTransport := oauth.NewDPoPTransport(http.DefaultTransport, dpopKey) + // Set the access token in the transport for "ath" claim computation + dpopTransport.SetAccessToken(token.AccessToken) + client := atproto.NewClientWithDPoP(pdsEndpoint, did, token.AccessToken, dpopKey, dpopTransport) + + return s.registerWithClient(publicURL, did, client) } -// registerWithToken registers the hold using an access token -func (s *HoldService) registerWithToken(publicURL, did, pdsEndpoint, accessToken string) error { +// 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 { @@ -841,9 +956,6 @@ func (s *HoldService) registerWithToken(publicURL, did, pdsEndpoint, accessToken ctx := context.Background() - // Create ATProto client with owner's credentials - client := atproto.NewClient(pdsEndpoint, did, accessToken) - // Create HoldRecord holdRecord := atproto.NewHoldRecord(publicURL, did, s.config.Server.Public) @@ -866,6 +978,28 @@ func (s *HoldService) registerWithToken(publicURL, did, pdsEndpoint, accessToken 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)) diff --git a/cmd/profile-update/main.go b/cmd/profile-update/main.go new file mode 100644 index 0000000..40c55ef --- /dev/null +++ b/cmd/profile-update/main.go @@ -0,0 +1,122 @@ +package main + +import ( + "context" + "encoding/base64" + "encoding/json" + "flag" + "fmt" + "log" + "os" + "path/filepath" + "strings" + + atprotoAuth "atcr.io/pkg/auth/atproto" + "atcr.io/pkg/atproto" +) + +// DockerConfig represents ~/.docker/config.json +type DockerConfig struct { + Auths map[string]AuthEntry `json:"auths"` +} + +type AuthEntry struct { + Auth string `json:"auth"` // base64(username:password) +} + +func main() { + var defaultHold string + var registryURL string + + flag.StringVar(&defaultHold, "default-hold", "", "Default hold endpoint URL (e.g., http://172.28.0.3:8080)") + flag.StringVar(®istryURL, "registry", "127.0.0.1:5000", "Registry URL to read auth from Docker config") + flag.Parse() + + // Read Docker config + home, err := os.UserHomeDir() + if err != nil { + log.Fatalf("Failed to get home directory: %v", err) + } + dockerConfigPath := filepath.Join(home, ".docker", "config.json") + + configData, err := os.ReadFile(dockerConfigPath) + if err != nil { + log.Fatalf("Failed to read Docker config: %v\n\nMake sure you've logged in with: docker login %s", err, registryURL) + } + + var dockerConfig DockerConfig + if err := json.Unmarshal(configData, &dockerConfig); err != nil { + log.Fatalf("Failed to parse Docker config: %v", err) + } + + // Get auth for registry + authEntry, ok := dockerConfig.Auths[registryURL] + if !ok { + log.Fatalf("No auth found for registry %s in Docker config", registryURL) + } + + // Decode base64 auth (format: "username:password") + authBytes, err := base64.StdEncoding.DecodeString(authEntry.Auth) + if err != nil { + log.Fatalf("Failed to decode auth: %v", err) + } + + parts := strings.SplitN(string(authBytes), ":", 2) + if len(parts) != 2 { + log.Fatalf("Invalid auth format") + } + + handle := parts[0] + password := parts[1] // This should be an app password + + fmt.Printf("Handle: %s\n", handle) + + // Create session validator and get access token + validator := atprotoAuth.NewSessionValidator() + ctx := context.Background() + + did, pdsEndpoint, accessToken, err := validator.CreateSessionAndGetToken(ctx, handle, password) + if err != nil { + log.Fatalf("Failed to authenticate: %v", err) + } + + fmt.Printf("DID: %s\n", did) + fmt.Printf("PDS: %s\n\n", pdsEndpoint) + + // Create client with the access token from createSession + client := atproto.NewClient(pdsEndpoint, did, accessToken) + + // Get current profile + profile, err := atproto.GetProfile(ctx, client) + if err != nil { + log.Fatalf("Failed to get current profile: %v", err) + } + + if profile == nil { + if defaultHold == "" { + fmt.Println("No existing profile found.") + fmt.Println("\nTo create profile with default hold, use: -default-hold ") + return + } + fmt.Println("No existing profile found. Creating new profile...") + profile = atproto.NewSailorProfileRecord(defaultHold) + } else { + fmt.Printf("Current defaultHold: %s\n", profile.DefaultHold) + if defaultHold == "" { + // Just show current profile + fmt.Println("\nTo update, use: -default-hold ") + return + } + profile.DefaultHold = defaultHold + } + + // Update profile + if defaultHold != "" { + err = atproto.UpdateProfile(ctx, client, profile) + if err != nil { + log.Fatalf("Failed to update profile: %v", err) + } + + fmt.Printf("\nāœ“ Updated defaultHold to: %s\n", defaultHold) + } +} diff --git a/docker-compose.yml b/docker-compose.yml index cb203b6..51c38af 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -11,12 +11,22 @@ services: # Only auth keys (could be moved to secrets in production) - atcr-auth:/var/lib/atcr/auth restart: unless-stopped + networks: + atcr-network: + ipv4_address: 172.28.0.2 # The registry should be stateless - all storage is external: # - Manifests/Tags -> ATProto PDS # - Blobs/Layers -> Hold service # Future: Add read_only: true for production deployments hold: + environment: + HOLD_PUBLIC_URL: http://172.28.0.3:8080 + HOLD_OWNER: did:plc:pddp4xt5lgnv2qsegbzzs4xg + HOLD_PUBLIC: false + STORAGE_DRIVER: filesystem + STORAGE_ROOT_DIR: /var/lib/atcr/hold + TEST_MODE: true build: context: . dockerfile: Dockerfile.hold @@ -27,8 +37,17 @@ services: volumes: - atcr-hold:/var/lib/atcr/hold restart: unless-stopped + networks: + atcr-network: + ipv4_address: 172.28.0.3 + +networks: + atcr-network: + driver: bridge + ipam: + config: + - subnet: 172.28.0.0/24 volumes: - atcr-blobs: atcr-hold: atcr-auth: diff --git a/docs/BYOS.md b/docs/BYOS.md index a755446..4d34595 100644 --- a/docs/BYOS.md +++ b/docs/BYOS.md @@ -174,9 +174,9 @@ The hold service must be registered in a PDS to be discoverable by the AppView. **Standard registration workflow:** -1. Set `HOLD_CREW_OWNER` to your DID: +1. Set `HOLD_OWNER` to your DID: ```bash - export HOLD_CREW_OWNER=did:plc:your-did-here + export HOLD_OWNER=did:plc:your-did-here ``` 2. Start the hold service: @@ -249,7 +249,7 @@ fly deploy # Set secrets fly secrets set AWS_ACCESS_KEY_ID=... fly secrets set AWS_SECRET_ACCESS_KEY=... -fly secrets set HOLD_CREW_OWNER=did:plc:your-did-here +fly secrets set HOLD_OWNER=did:plc:your-did-here # Check logs for OAuth URL on first run fly logs @@ -404,7 +404,7 @@ Alice wants to use her own Storj account: 1. **Set environment variables**: ```bash export HOLD_PUBLIC_URL=https://alice-storage.fly.dev - export HOLD_CREW_OWNER=did:plc:alice123 + export HOLD_OWNER=did:plc:alice123 export STORAGE_DRIVER=s3 export AWS_ACCESS_KEY_ID=your_storj_access_key export AWS_SECRET_ACCESS_KEY=your_storj_secret_key @@ -423,7 +423,7 @@ A company wants shared storage for their team: 1. **Deploy hold service** with S3 credentials and auto-registration: ```bash export HOLD_PUBLIC_URL=https://company-hold.fly.dev - export HOLD_CREW_OWNER=did:plc:admin + export HOLD_OWNER=did:plc:admin export HOLD_PUBLIC=false export STORAGE_DRIVER=s3 export AWS_ACCESS_KEY_ID=... diff --git a/pkg/atproto/client.go b/pkg/atproto/client.go index df5a4d5..b12c805 100644 --- a/pkg/atproto/client.go +++ b/pkg/atproto/client.go @@ -3,6 +3,7 @@ package atproto import ( "bytes" "context" + "crypto/ecdsa" "encoding/json" "fmt" "io" @@ -15,18 +16,42 @@ type Client struct { did string accessToken string httpClient *http.Client + useDPoP bool // true if using DPoP-bound tokens (OAuth) } -// NewClient creates a new ATProto client +// NewClient creates a new ATProto client for Basic Auth tokens func NewClient(pdsEndpoint, did, accessToken string) *Client { return &Client{ pdsEndpoint: pdsEndpoint, did: did, accessToken: accessToken, httpClient: &http.Client{}, + useDPoP: false, // Basic Auth uses Bearer tokens } } +// NewClientWithDPoP creates a new ATProto client with DPoP support +// This is required for OAuth tokens +func NewClientWithDPoP(pdsEndpoint, did, accessToken string, dpopKey *ecdsa.PrivateKey, transport http.RoundTripper) *Client { + return &Client{ + pdsEndpoint: pdsEndpoint, + did: did, + accessToken: accessToken, + httpClient: &http.Client{ + Transport: transport, + }, + useDPoP: true, // OAuth uses DPoP tokens + } +} + +// authHeader returns the appropriate Authorization header value +func (c *Client) authHeader() string { + if c.useDPoP { + return "DPoP " + c.accessToken + } + return "Bearer " + c.accessToken +} + // Record represents a generic ATProto record type Record struct { URI string `json:"uri"` @@ -57,7 +82,7 @@ func (c *Client) PutRecord(ctx context.Context, collection, rkey string, record return nil, err } - req.Header.Set("Authorization", "Bearer "+c.accessToken) + req.Header.Set("Authorization", c.authHeader()) req.Header.Set("Content-Type", "application/json") resp, err := c.httpClient.Do(req) @@ -89,7 +114,7 @@ func (c *Client) GetRecord(ctx context.Context, collection, rkey string) (*Recor return nil, err } - req.Header.Set("Authorization", "Bearer "+c.accessToken) + req.Header.Set("Authorization", c.authHeader()) resp, err := c.httpClient.Do(req) if err != nil { @@ -133,7 +158,7 @@ func (c *Client) DeleteRecord(ctx context.Context, collection, rkey string) erro return err } - req.Header.Set("Authorization", "Bearer "+c.accessToken) + req.Header.Set("Authorization", c.authHeader()) req.Header.Set("Content-Type", "application/json") resp, err := c.httpClient.Do(req) @@ -160,7 +185,7 @@ func (c *Client) ListRecords(ctx context.Context, collection string, limit int) return nil, err } - req.Header.Set("Authorization", "Bearer "+c.accessToken) + req.Header.Set("Authorization", c.authHeader()) resp, err := c.httpClient.Do(req) if err != nil { diff --git a/pkg/auth/oauth/client.go b/pkg/auth/oauth/client.go index cf0105d..204980e 100644 --- a/pkg/auth/oauth/client.go +++ b/pkg/auth/oauth/client.go @@ -58,12 +58,14 @@ func (c *Client) InitializeForHandle(ctx context.Context, handle string) error { c.metadata = metadata - // Configure OAuth2 client + // Configure OAuth2 client with default scope + // Can be overridden with SetScopes() before calling AuthorizeURL() c.config = &oauth2.Config{ ClientID: c.clientID, Endpoint: oauth2.Endpoint{ - AuthURL: metadata.AuthorizationEndpoint, - TokenURL: metadata.TokenEndpoint, + AuthURL: metadata.AuthorizationEndpoint, + TokenURL: metadata.TokenEndpoint, + PushedAuthURL: metadata.PushedAuthorizationRequestEndpoint, }, RedirectURL: c.redirectURI, Scopes: []string{"atproto"}, @@ -72,6 +74,13 @@ func (c *Client) InitializeForHandle(ctx context.Context, handle string) error { return nil } +// SetScopes sets custom OAuth scopes (must be called after InitializeForHandle) +func (c *Client) SetScopes(scopes []string) { + if c.config != nil { + c.config.Scopes = scopes + } +} + // AuthorizeURL generates the authorization URL with PKCE func (c *Client) AuthorizeURL(state string) (authURL string, codeVerifier string, err error) { if c.config == nil { diff --git a/pkg/auth/oauth/discovery.go b/pkg/auth/oauth/discovery.go index 2841b8f..67de48a 100644 --- a/pkg/auth/oauth/discovery.go +++ b/pkg/auth/oauth/discovery.go @@ -7,6 +7,13 @@ import ( "net/http" ) +// ProtectedResourceMetadata represents the OAuth protected resource metadata +// as defined in ATProto OAuth spec +type ProtectedResourceMetadata struct { + Resource string `json:"resource"` + AuthorizationServers []string `json:"authorization_servers"` +} + // AuthServerMetadata represents the OAuth authorization server metadata // as defined in RFC 8414 type AuthServerMetadata struct { @@ -25,26 +32,72 @@ type AuthServerMetadata struct { AuthorizationResponseIssParameterSupported bool `json:"authorization_response_iss_parameter_supported,omitempty"` } -// DiscoverAuthServer discovers the OAuth authorization server metadata -// from the PDS endpoint using the well-known discovery endpoint -func DiscoverAuthServer(ctx context.Context, pdsEndpoint string) (*AuthServerMetadata, error) { - // Construct the well-known URL per RFC 8414 - discoveryURL := fmt.Sprintf("%s/.well-known/oauth-authorization-server", pdsEndpoint) +// DiscoverProtectedResource discovers the protected resource metadata +// from the PDS endpoint to find the authorization servers +func DiscoverProtectedResource(ctx context.Context, pdsEndpoint string) (*ProtectedResourceMetadata, error) { + // Construct the well-known URL per ATProto OAuth spec + discoveryURL := fmt.Sprintf("%s/.well-known/oauth-protected-resource", pdsEndpoint) req, err := http.NewRequestWithContext(ctx, "GET", discoveryURL, nil) if err != nil { - return nil, fmt.Errorf("failed to create discovery request: %w", err) + return nil, fmt.Errorf("failed to create protected resource discovery request: %w", err) } client := &http.Client{} resp, err := client.Do(req) if err != nil { - return nil, fmt.Errorf("failed to fetch authorization server metadata: %w", err) + return nil, fmt.Errorf("failed to fetch protected resource metadata: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("authorization server discovery failed with status %d", resp.StatusCode) + return nil, fmt.Errorf("protected resource discovery failed with status %d", resp.StatusCode) + } + + var metadata ProtectedResourceMetadata + if err := json.NewDecoder(resp.Body).Decode(&metadata); err != nil { + return nil, fmt.Errorf("failed to decode protected resource metadata: %w", err) + } + + // Validate required fields + if len(metadata.AuthorizationServers) == 0 { + return nil, fmt.Errorf("protected resource metadata missing authorization_servers") + } + + return &metadata, nil +} + +// DiscoverAuthServer discovers the OAuth authorization server metadata +// using the ATProto two-step discovery process: +// 1. Fetch protected resource metadata from PDS to get authorization server URL +// 2. Fetch authorization server metadata from that URL +func DiscoverAuthServer(ctx context.Context, pdsEndpoint string) (*AuthServerMetadata, error) { + // Step 1: Discover the authorization server URL from the protected resource + protectedResource, err := DiscoverProtectedResource(ctx, pdsEndpoint) + if err != nil { + return nil, fmt.Errorf("step 1 failed - discover protected resource from PDS %s: %w", pdsEndpoint, err) + } + + // Use the first authorization server (ATProto spec allows multiple, but typically one) + authServerURL := protectedResource.AuthorizationServers[0] + + // Step 2: Fetch authorization server metadata + discoveryURL := fmt.Sprintf("%s/.well-known/oauth-authorization-server", authServerURL) + + req, err := http.NewRequestWithContext(ctx, "GET", discoveryURL, nil) + if err != nil { + return nil, fmt.Errorf("step 2 failed - create request: %w", err) + } + + client := &http.Client{} + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("step 2 failed - fetch auth server metadata from %s: %w", discoveryURL, err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("step 2 failed - auth server discovery at %s returned status %d", discoveryURL, resp.StatusCode) } var metadata AuthServerMetadata diff --git a/pkg/auth/oauth/storage.go b/pkg/auth/oauth/storage.go index c50262b..eddac32 100644 --- a/pkg/auth/oauth/storage.go +++ b/pkg/auth/oauth/storage.go @@ -95,3 +95,18 @@ func (s *TokenStore) IsExpired() bool { // Add a 60 second buffer to refresh before actual expiry return time.Now().After(s.ExpiresAt.Add(-60 * time.Second)) } + +// ParseDPoPKey parses a PEM-encoded ECDSA private key +func ParseDPoPKey(pemData string) (*ecdsa.PrivateKey, error) { + block, _ := pem.Decode([]byte(pemData)) + if block == nil { + return nil, fmt.Errorf("failed to decode PEM block") + } + + key, err := x509.ParseECPrivateKey(block.Bytes) + if err != nil { + return nil, fmt.Errorf("failed to parse EC private key: %w", err) + } + + return key, nil +} diff --git a/pkg/auth/oauth/transport.go b/pkg/auth/oauth/transport.go index c2d0877..ea1dd8e 100644 --- a/pkg/auth/oauth/transport.go +++ b/pkg/auth/oauth/transport.go @@ -2,6 +2,8 @@ package oauth import ( "crypto/ecdsa" + "crypto/sha256" + "encoding/base64" "fmt" "net/http" "sync" @@ -14,10 +16,11 @@ import ( // DPoPTransport is an HTTP RoundTripper that adds DPoP headers to requests type DPoPTransport struct { - base http.RoundTripper - dpopKey *ecdsa.PrivateKey - nonce string - mu sync.RWMutex // Protects nonce + base http.RoundTripper + dpopKey *ecdsa.PrivateKey + accessToken string // For computing "ath" claim + nonce string + mu sync.RWMutex // Protects nonce } // NewDPoPTransport creates a new DPoP transport with the given private key @@ -80,9 +83,10 @@ func (t *DPoPTransport) RoundTrip(req *http.Request) (*http.Response, error) { // addDPoPHeader generates and adds a DPoP proof header to the request func (t *DPoPTransport) addDPoPHeader(req *http.Request) error { - // Read current nonce + // Read current nonce and access token t.mu.RLock() nonce := t.nonce + accessToken := t.accessToken t.mu.RUnlock() // Create DPoP proof claims @@ -100,6 +104,16 @@ func (t *DPoPTransport) addDPoPHeader(req *http.Request) error { claims.Nonce = nonce } + // Add "ath" (access token hash) if we have an access token + // This is required when using DPoP with an access token + if accessToken != "" { + // Compute SHA-256 hash of the access token + hash := sha256.Sum256([]byte(accessToken)) + // Base64url encode the hash (without padding) + ath := base64.RawURLEncoding.EncodeToString(hash[:]) + claims.AccessTokenHash = ath + } + // Generate DPoP proof // go-dpop automatically adds the JWK to the header proofString, err := dpop.Create(jwt.SigningMethodES256, claims, t.dpopKey) @@ -126,3 +140,10 @@ func (t *DPoPTransport) GetNonce() string { defer t.mu.RUnlock() return t.nonce } + +// SetAccessToken sets the access token for computing "ath" claim +func (t *DPoPTransport) SetAccessToken(token string) { + t.mu.Lock() + defer t.mu.Unlock() + t.accessToken = token +} diff --git a/pkg/storage/proxy_blob_store.go b/pkg/storage/proxy_blob_store.go index ce933ad..cd8d1b9 100644 --- a/pkg/storage/proxy_blob_store.go +++ b/pkg/storage/proxy_blob_store.go @@ -30,10 +30,13 @@ type ProxyBlobStore struct { // NewProxyBlobStore creates a new proxy blob store func NewProxyBlobStore(storageEndpoint, did string) *ProxyBlobStore { + fmt.Printf("DEBUG [proxy_blob_store]: NewProxyBlobStore created with endpoint=%s, did=%s\n", storageEndpoint, did) return &ProxyBlobStore{ storageEndpoint: storageEndpoint, - httpClient: &http.Client{}, - did: did, + httpClient: &http.Client{ + Timeout: 5 * time.Minute, // Timeout for presigned URL requests and uploads + }, + did: did, } } @@ -236,6 +239,8 @@ func (p *ProxyBlobStore) getDownloadURL(ctx context.Context, dgst digest.Digest) // getUploadURL requests a presigned upload URL from the storage service func (p *ProxyBlobStore) getUploadURL(ctx context.Context, dgst digest.Digest, size int64) (string, error) { + fmt.Printf("DEBUG [proxy_blob_store/getUploadURL]: storageEndpoint=%s, digest=%s\n", p.storageEndpoint, dgst) + reqBody := map[string]any{ "did": p.did, "digest": dgst.String(), @@ -248,6 +253,7 @@ func (p *ProxyBlobStore) getUploadURL(ctx context.Context, dgst digest.Digest, s } url := fmt.Sprintf("%s/put-presigned-url", p.storageEndpoint) + fmt.Printf("DEBUG [proxy_blob_store/getUploadURL]: Calling %s\n", url) req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) if err != nil { return "", err @@ -271,6 +277,7 @@ func (p *ProxyBlobStore) getUploadURL(ctx context.Context, dgst digest.Digest, s return "", err } + fmt.Printf("DEBUG [proxy_blob_store/getUploadURL]: Got presigned URL=%s\n", result.URL) return result.URL, nil } @@ -311,7 +318,9 @@ func (w *ProxyBlobWriter) ReadFrom(r io.Reader) (int64, error) { if w.closed { return 0, fmt.Errorf("writer closed") } - return w.buffer.ReadFrom(r) + n, err := w.buffer.ReadFrom(r) + w.size += n + return n, err } // Size returns the current size