From 256cc883c9dc073072ae77a42ee76c066cd57f03 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sat, 11 Oct 2025 23:13:20 -0500 Subject: [PATCH] consolidate presigned url endpoint. fix manifest pull blob from pds --- cmd/hold/main.go | 4 +- pkg/atproto/client.go | 21 ++++++ pkg/hold/handlers.go | 112 ++++++++++++++++---------------- pkg/hold/types.go | 7 +- pkg/storage/proxy_blob_store.go | 101 +++++++--------------------- 5 files changed, 103 insertions(+), 142 deletions(-) diff --git a/cmd/hold/main.go b/cmd/hold/main.go index 3501606..e329ee4 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -35,9 +35,7 @@ func main() { mux := http.NewServeMux() mux.HandleFunc("/health", service.HealthHandler) mux.HandleFunc("/register", service.HandleRegister) - mux.HandleFunc("/get-presigned-url", service.HandlePresignedURL(hold.OperationGet)) - mux.HandleFunc("/head-presigned-url", service.HandlePresignedURL(hold.OperationHead)) - mux.HandleFunc("/put-presigned-url", service.HandlePresignedURL(hold.OperationPut)) + mux.HandleFunc("/presigned-url", service.HandlePresignedURL) mux.HandleFunc("/move", service.HandleMove) // Multipart upload endpoints diff --git a/pkg/atproto/client.go b/pkg/atproto/client.go index 2b1e832..5f2c3a1 100644 --- a/pkg/atproto/client.go +++ b/pkg/atproto/client.go @@ -3,6 +3,7 @@ package atproto import ( "bytes" "context" + "encoding/base64" "encoding/json" "fmt" "io" @@ -343,6 +344,26 @@ func (c *Client) GetBlob(ctx context.Context, cid string) ([]byte, error) { return nil, fmt.Errorf("failed to read blob data: %w", err) } + // Check if PDS returned JSON-wrapped blob (Bluesky implementation) + // PDS may wrap blobs as JSON-encoded base64 strings + // Detection: Check if content starts with a quote (indicating JSON string) + if len(data) > 0 && data[0] == '"' { + // Blob is JSON-encoded - decode it + var base64Str string + if err := json.Unmarshal(data, &base64Str); err != nil { + return nil, fmt.Errorf("failed to unmarshal JSON-wrapped blob: %w", err) + } + + // Base64-decode the blob content + decoded, err := base64.StdEncoding.DecodeString(base64Str) + if err != nil { + return nil, fmt.Errorf("failed to base64-decode blob: %w", err) + } + + return decoded, nil + } + + // Raw blob response (expected ATProto behavior) return data, nil } diff --git a/pkg/hold/handlers.go b/pkg/hold/handlers.go index 56d0efd..5b628e3 100644 --- a/pkg/hold/handlers.go +++ b/pkg/hold/handlers.go @@ -12,64 +12,62 @@ import ( "atcr.io/pkg/atproto" ) -// HandlePresignedURL returns an HTTP handler for presigned URL requests (GET, HEAD, or PUT) -// This consolidates the three separate handlers into a single parameterized implementation -func (s *HoldService) HandlePresignedURL(operation PresignedURLOperation) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) - return - } - - var req PresignedURLRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest) - return - } - - // Validate DID authorization based on operation type - var authorized bool - switch operation { - case OperationGet, OperationHead: - authorized = s.isAuthorizedRead(req.DID) - case OperationPut: - authorized = s.isAuthorizedWrite(req.DID) - default: - http.Error(w, "unsupported operation", http.StatusBadRequest) - return - } - - if !authorized { - log.Printf("[HandlePresignedURL:%s] Authorization FAILED", operation) - if req.DID == "" { - http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized) - } else { - http.Error(w, "forbidden: access denied", http.StatusForbidden) - } - return - } - - // Generate presigned URL (15 minute expiry) - ctx := context.Background() - expiry := time.Now().Add(15 * time.Minute) - - url, err := s.getPresignedURL(ctx, operation, req.Digest, req.DID) - if err != nil { - log.Printf("[HandlePresignedURL:%s] getPresignedURL failed: %v", operation, err) - http.Error(w, fmt.Sprintf("failed to generate URL: %v", err), http.StatusInternalServerError) - return - } - - log.Printf("[HandlePresignedURL:%s] Returning URL to client", operation) - - resp := PresignedURLResponse{ - URL: url, - ExpiresAt: expiry, - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(resp) +// HandlePresignedURL handles presigned URL requests (GET, HEAD, or PUT) +// Operation type is specified in the request body +func (s *HoldService) HandlePresignedURL(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return } + + var req PresignedURLRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, fmt.Sprintf("invalid request: %v", err), http.StatusBadRequest) + return + } + + // Validate DID authorization based on operation type + var authorized bool + switch req.Operation { + case OperationGet, OperationHead: + authorized = s.isAuthorizedRead(req.DID) + case OperationPut: + authorized = s.isAuthorizedWrite(req.DID) + default: + http.Error(w, "unsupported operation", http.StatusBadRequest) + return + } + + if !authorized { + log.Printf("[HandlePresignedURL:%s] Authorization FAILED", req.Operation) + if req.DID == "" { + http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized) + } else { + http.Error(w, "forbidden: access denied", http.StatusForbidden) + } + return + } + + // Generate presigned URL (15 minute expiry) + ctx := context.Background() + expiry := time.Now().Add(15 * time.Minute) + + url, err := s.getPresignedURL(ctx, req.Operation, req.Digest, req.DID) + if err != nil { + log.Printf("[HandlePresignedURL:%s] getPresignedURL failed: %v", req.Operation, err) + http.Error(w, fmt.Sprintf("failed to generate URL: %v", err), http.StatusInternalServerError) + return + } + + log.Printf("[HandlePresignedURL:%s] Returning URL to client", req.Operation) + + resp := PresignedURLResponse{ + URL: url, + ExpiresAt: expiry, + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) } // HandleProxyGet proxies a blob download through the service diff --git a/pkg/hold/types.go b/pkg/hold/types.go index 69c875d..44d491e 100644 --- a/pkg/hold/types.go +++ b/pkg/hold/types.go @@ -15,9 +15,10 @@ const ( // PresignedURLRequest represents a request for a presigned URL (GET, HEAD, or PUT) type PresignedURLRequest struct { - DID string `json:"did"` - Digest string `json:"digest"` - Size int64 `json:"size,omitempty"` // Only required for PUT operations + Operation PresignedURLOperation `json:"operation"` + DID string `json:"did"` + Digest string `json:"digest"` + Size int64 `json:"size,omitempty"` // Only required for PUT operations } // PresignedURLResponse contains the presigned URL diff --git a/pkg/storage/proxy_blob_store.go b/pkg/storage/proxy_blob_store.go index 21324ff..beb390b 100644 --- a/pkg/storage/proxy_blob_store.go +++ b/pkg/storage/proxy_blob_store.go @@ -270,11 +270,17 @@ func (p *ProxyBlobStore) Resume(ctx context.Context, id string) (distribution.Bl return writer, nil } -// getDownloadURL requests a presigned download URL from the storage service -func (p *ProxyBlobStore) getDownloadURL(ctx context.Context, dgst digest.Digest) (string, error) { +// getPresignedURL requests a presigned URL from the storage service for any operation +func (p *ProxyBlobStore) getPresignedURL(ctx context.Context, operation, dgst string, size int64) (string, error) { reqBody := map[string]any{ - "did": p.did, - "digest": dgst.String(), + "operation": operation, + "did": p.did, + "digest": dgst, + } + + // Only include size for PUT operations + if size > 0 { + reqBody["size"] = size } body, err := json.Marshal(reqBody) @@ -282,7 +288,7 @@ func (p *ProxyBlobStore) getDownloadURL(ctx context.Context, dgst digest.Digest) return "", err } - url := fmt.Sprintf("%s/get-presigned-url", p.storageEndpoint) + url := fmt.Sprintf("%s/presigned-url", p.storageEndpoint) req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) if err != nil { return "", err @@ -296,7 +302,7 @@ func (p *ProxyBlobStore) getDownloadURL(ctx context.Context, dgst digest.Digest) defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return "", fmt.Errorf("failed to get download URL: status %d", resp.StatusCode) + return "", fmt.Errorf("failed to get presigned URL: status %d", resp.StatusCode) } var result struct { @@ -309,87 +315,24 @@ func (p *ProxyBlobStore) getDownloadURL(ctx context.Context, dgst digest.Digest) return result.URL, nil } +// getDownloadURL requests a presigned download URL from the storage service +func (p *ProxyBlobStore) getDownloadURL(ctx context.Context, dgst digest.Digest) (string, error) { + return p.getPresignedURL(ctx, "GET", dgst.String(), 0) +} + // getHeadURL requests a presigned HEAD URL from the storage service func (p *ProxyBlobStore) getHeadURL(ctx context.Context, dgst digest.Digest) (string, error) { - reqBody := map[string]any{ - "did": p.did, - "digest": dgst.String(), - } - - body, err := json.Marshal(reqBody) - if err != nil { - return "", err - } - - url := fmt.Sprintf("%s/head-presigned-url", p.storageEndpoint) - req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) - if err != nil { - return "", err - } - req.Header.Set("Content-Type", "application/json") - - resp, err := p.httpClient.Do(req) - if err != nil { - return "", err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return "", fmt.Errorf("failed to get HEAD URL: status %d", resp.StatusCode) - } - - var result struct { - URL string `json:"url"` - } - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return "", err - } - - return result.URL, nil + return p.getPresignedURL(ctx, "HEAD", dgst.String(), 0) } // 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(), - "size": size, + url, err := p.getPresignedURL(ctx, "PUT", dgst.String(), size) + if err == nil { + fmt.Printf("DEBUG [proxy_blob_store/getUploadURL]: Got presigned URL=%s\n", url) } - - body, err := json.Marshal(reqBody) - if err != nil { - return "", err - } - - 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 - } - req.Header.Set("Content-Type", "application/json") - - resp, err := p.httpClient.Do(req) - if err != nil { - return "", err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return "", fmt.Errorf("failed to get upload URL: status %d", resp.StatusCode) - } - - var result struct { - URL string `json:"url"` - } - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return "", err - } - - fmt.Printf("DEBUG [proxy_blob_store/getUploadURL]: Got presigned URL=%s\n", result.URL) - return result.URL, nil + return url, err } // startMultipartUpload initiates a multipart upload via hold service