diff --git a/backend/internal/application/application.go b/backend/internal/application/application.go index dc269a9..ff59d71 100644 --- a/backend/internal/application/application.go +++ b/backend/internal/application/application.go @@ -306,7 +306,7 @@ func New(options Options) (*App, error) { } webhookBridge := orchestration.NewWebhookBridge(webhook.NewDeliverer(webhookSender, getenv("ZHINIAN_WEBHOOK_SECRET"), nil)) outputs := orchestration.NewAssetOutputRegistrar(assetService, orchestration.ResolveProviderOutputURLs) - providerProcessor := jobs.ProviderProcessor{Providers: providerResolver, Store: jobStore} + providerProcessor := jobs.ProviderProcessor{Providers: providerResolver, Store: jobStore, AssetURLs: assetService, AssetURLTTL: assets.MaximumSignedURLTTL} settlementProcessor := orchestration.NewSettlementProcessor(providerProcessor, ledger, settlementState, nil) processor := orchestration.NewOutputRegisteringProcessor(settlementProcessor, outputs, jobState) artifacts := orchestration.NewAssetArtifacts(assetService) diff --git a/backend/internal/application/runtime.go b/backend/internal/application/runtime.go index 188d3b6..c0dc854 100644 --- a/backend/internal/application/runtime.go +++ b/backend/internal/application/runtime.go @@ -4,6 +4,7 @@ import ( "context" "crypto/rand" "encoding/hex" + "errors" "fmt" "io" "net/http" @@ -480,6 +481,14 @@ func (store prefixedBlobStore) Delete(ctx context.Context, key string) error { return store.store.Delete(ctx, path.Join(store.prefix, key)) } +func (store prefixedBlobStore) SignReadURL(key string, ttl time.Duration) (string, error) { + signer, ok := store.store.(assets.BlobURLSigner) + if !ok { + return "", errors.New("blob store does not support signed URLs") + } + return signer.SignReadURL(path.Join(store.prefix, key), ttl) +} + func configuredOSSBlobStore(getenv postgres.Getenv) (assets.BlobStore, bool, error) { endpoint, bucket := strings.TrimSpace(getenv("ALI_OSS_ENDPOINT")), strings.TrimSpace(getenv("ALI_OSS_BUCKET")) accessKeyID, secret := strings.TrimSpace(getenv("ALI_OSS_ACCESS_KEY_ID")), strings.TrimSpace(getenv("ALI_OSS_ACCESS_KEY_SECRET")) @@ -491,7 +500,7 @@ func configuredOSSBlobStore(getenv postgres.Getenv) (assets.BlobStore, bool, err if err != nil { return nil, false, err } - store, err := assets.NewOSS(assets.OSSConfig{Endpoint: endpoint, Bucket: bucket, PublicBaseURL: publicURL, PublicRead: true}, client) + store, err := assets.NewOSS(assets.OSSConfig{Endpoint: endpoint, Bucket: bucket, PublicBaseURL: publicURL, PublicRead: false}, client) if err != nil { return nil, false, err } diff --git a/backend/internal/application/runtime_test.go b/backend/internal/application/runtime_test.go index 158f326..aabe4c5 100644 --- a/backend/internal/application/runtime_test.go +++ b/backend/internal/application/runtime_test.go @@ -12,6 +12,7 @@ import ( "reflect" "strings" "testing" + "time" "git.nianxx.cn/wangxuming/NianAIGC/backend/internal/assets" "git.nianxx.cn/wangxuming/NianAIGC/backend/internal/billing" @@ -320,9 +321,42 @@ func TestPrefixedBlobStoreKeepsApplicationStoragePathStable(t *testing.T) { if !reflect.DeepEqual([]string{inner.readKey, inner.deleteKey}, []string{"tenant-prefix/uploads/day/file.png", "tenant-prefix/uploads/day/file.png"}) { t.Fatalf("read/delete=%q/%q", inner.readKey, inner.deleteKey) } + if _, err := store.SignReadURL(stored.Key, time.Hour); err != nil || inner.signedKey != "tenant-prefix/uploads/day/file.png" || inner.signedTTL != time.Hour { + t.Fatalf("signed URL delegation = %q / %s, err=%v", inner.signedKey, inner.signedTTL, err) + } } -type recordingBlobStore struct{ putKey, readKey, deleteKey string } +func TestConfiguredOSSBlobStoreDoesNotRequestPublicObjectACL(t *testing.T) { + var requests []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + requests = append(requests, request.Method+" "+request.URL.RequestURI()) + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + values := map[string]string{ + "ALI_OSS_ENDPOINT": server.URL + "/private-bucket", + "ALI_OSS_BUCKET": "private-bucket", + "ALI_OSS_ACCESS_KEY_ID": "test-access-key", + "ALI_OSS_ACCESS_KEY_SECRET": "test-access-secret", + "ALI_OSS_PUBLIC_BASE_URL": server.URL + "/private-bucket", + } + store, configured, err := configuredOSSBlobStore(func(name string) string { return values[name] }) + if err != nil || !configured { + t.Fatalf("configured store = %T, %v, configured=%v", store, err, configured) + } + if _, err := store.Put(context.Background(), "uploads/day/private.png", bytes.NewReader([]byte("png")), 3, "image/png"); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(requests, []string{"PUT /private-bucket/zhinian/uploads/day/private.png"}) { + t.Fatalf("OSS requests = %#v, want one private PutObject request", requests) + } +} + +type recordingBlobStore struct { + putKey, readKey, deleteKey, signedKey string + signedTTL time.Duration +} type applicationRuntimeSettingsRepository struct { values map[string]string @@ -360,3 +394,7 @@ func (s *recordingBlobStore) Delete(_ context.Context, key string) error { s.deleteKey = key return nil } +func (s *recordingBlobStore) SignReadURL(key string, ttl time.Duration) (string, error) { + s.signedKey, s.signedTTL = key, ttl + return "https://signed.example/" + key, nil +} diff --git a/backend/internal/assets/assets.go b/backend/internal/assets/assets.go index 646d99a..76aed66 100644 --- a/backend/internal/assets/assets.go +++ b/backend/internal/assets/assets.go @@ -126,6 +126,9 @@ type BlobStore interface { Read(context.Context, string) (Blob, error) Delete(context.Context, string) error } +type BlobURLSigner interface { + SignReadURL(string, time.Duration) (string, error) +} type RemoteFetcher interface { Fetch(context.Context, string) (Blob, error) } @@ -364,6 +367,52 @@ func (s *Service) Download(ctx context.Context, scope Scope, id string) (Blob, e return s.remote.Fetch(ctx, a.URL) } +// SignedDownloadURL returns a short-lived direct URL only when the configured +// blob store supports it. Callers can otherwise fall back to Download without +// exposing storage credentials or assuming that every backend is OSS. +func (s *Service) SignedDownloadURL(ctx context.Context, scope Scope, id string, ttl time.Duration) (string, bool, error) { + a, err := s.Get(ctx, scope, id) + if err != nil { + return "", false, err + } + return s.signedDownloadURL(a, ttl) +} + +func (s *Service) signedDownloadURL(a Asset, ttl time.Duration) (string, bool, error) { + if a.StoragePath == "" || s.blobs == nil { + return "", false, nil + } + signer, ok := s.blobs.(BlobURLSigner) + if !ok { + return "", false, nil + } + signed, err := signer.SignReadURL(a.StoragePath, ttl) + if err != nil { + return "", false, err + } + return signed, true, nil +} + +// ResolveProviderAssetURL implements the jobs package's narrow resolver seam +// without importing jobs. The stable source URL is used only to replace the +// matching value in a transient provider request; the signed URL is never +// persisted in the asset catalog or job payload. +func (s *Service) ResolveProviderAssetURL(ctx context.Context, ownerID, id string, ttl time.Duration) (string, string, error) { + scope := PlatformScope(ownerID) + a, err := s.Get(ctx, scope, id) + if err != nil { + return "", "", err + } + signed, ok, err := s.signedDownloadURL(a, ttl) + if err != nil { + return "", "", err + } + if ok { + return a.URL, signed, nil + } + return a.URL, a.URL, nil +} + // DownloadPath serves a stored object only after its metadata has been found // in the requesting owner's catalog. Catalogs that support HTTP file serving // implement the optional storage-path lookup without widening other callers. diff --git a/backend/internal/assets/oss.go b/backend/internal/assets/oss.go index cb50f08..d6fdbcd 100644 --- a/backend/internal/assets/oss.go +++ b/backend/internal/assets/oss.go @@ -9,6 +9,7 @@ import ( "net/url" "path" "strings" + "time" ) const OSSACLPublicRead = "public-read" @@ -37,6 +38,10 @@ type OSSClient interface { Delete(context.Context, OSSObjectRequest) error } +type OSSURLSigner interface { + SignGetURL(OSSObjectRequest, time.Duration) (string, error) +} + type OSSError struct { Status int Code string @@ -116,6 +121,17 @@ func (s *OSS) Delete(ctx context.Context, key string) error { return mapOSSError(err) } +func (s *OSS) SignReadURL(key string, ttl time.Duration) (string, error) { + if err := validOSSKey(key); err != nil { + return "", err + } + signer, ok := s.client.(OSSURLSigner) + if !ok { + return "", errors.New("OSS client does not support signed URLs") + } + return signer.SignGetURL(OSSObjectRequest{Endpoint: s.config.PublicBaseURL, Bucket: s.config.Bucket, Key: key}, ttl) +} + type ossOperationDiagnostic struct { Operation string Status int @@ -182,3 +198,4 @@ func escapeOSSKey(key string) string { } var _ BlobStore = (*OSS)(nil) +var _ BlobURLSigner = (*OSS)(nil) diff --git a/backend/internal/assets/oss_http.go b/backend/internal/assets/oss_http.go index fe2fa69..049d7e5 100644 --- a/backend/internal/assets/oss_http.go +++ b/backend/internal/assets/oss_http.go @@ -16,6 +16,7 @@ import ( "net/url" "path" "sort" + "strconv" "strings" "syscall" "time" @@ -23,6 +24,11 @@ import ( const maxOSSErrorBody = 64 << 10 +const ( + DefaultSignedURLTTL = time.Hour + MaximumSignedURLTTL = 9 * time.Hour +) + // OSSHTTPClient implements OSSClient using Aliyun OSS's HTTP authorization // protocol. The supplied HTTP client owns timeout and transport policy. type OSSHTTPClient struct { @@ -112,6 +118,31 @@ func (c *OSSHTTPClient) Delete(ctx context.Context, request OSSObjectRequest) er return consumeOSSResponse(response) } +// SignGetURL creates an OSS V1 query-signed GET URL. The expiration is bounded +// to OSS's documented maximum and the operation is local: no network request is +// performed and no secret is embedded in the resulting URL. +func (c *OSSHTTPClient) SignGetURL(request OSSObjectRequest, ttl time.Duration) (string, error) { + if ttl <= 0 || ttl > MaximumSignedURLTTL { + return "", errors.New("OSS signed URL lifetime must be between zero and nine hours") + } + requestURL, err := ossObjectURL(request.Endpoint, request.Bucket, request.Key, "") + if err != nil { + return "", err + } + expires := c.now().UTC().Add(ttl).Unix() + canonicalResource := "/" + request.Bucket + "/" + request.Key + stringToSign := strings.Join([]string{http.MethodGet, "", "", strconv.FormatInt(expires, 10), canonicalResource}, "\n") + mac := hmac.New(sha1.New, []byte(c.accessKeySecret)) + _, _ = mac.Write([]byte(stringToSign)) + signature := base64.StdEncoding.EncodeToString(mac.Sum(nil)) + query := requestURL.Query() + query.Set("OSSAccessKeyId", c.accessKeyID) + query.Set("Expires", strconv.FormatInt(expires, 10)) + query.Set("Signature", signature) + requestURL.RawQuery = query.Encode() + return requestURL.String(), nil +} + func (c *OSSHTTPClient) newRequest(ctx context.Context, method, endpoint, bucket, key, subresource string, body io.Reader) (*http.Request, error) { requestURL, err := ossObjectURL(endpoint, bucket, key, subresource) if err != nil { @@ -307,3 +338,4 @@ func readOSSError(response *http.Response) error { } var _ OSSClient = (*OSSHTTPClient)(nil) +var _ OSSURLSigner = (*OSSHTTPClient)(nil) diff --git a/backend/internal/assets/oss_http_test.go b/backend/internal/assets/oss_http_test.go index d521f0f..d17b1f3 100644 --- a/backend/internal/assets/oss_http_test.go +++ b/backend/internal/assets/oss_http_test.go @@ -48,6 +48,36 @@ func TestOSSHTTPClientSignsPutAndBuildsVirtualHostURL(t *testing.T) { } } +func TestOSSHTTPClientPresignsPrivateGetURL(t *testing.T) { + client := newTestOSSHTTPClient(t, func(*http.Request) *http.Response { + t.Fatal("presigning must not perform a network request") + return nil + }) + + signed, err := client.SignGetURL(OSSObjectRequest{ + Endpoint: "https://oss-cn-test.aliyuncs.com", + Bucket: "bucket-a", + Key: "photos/cat.png", + }, time.Hour) + if err != nil { + t.Fatal(err) + } + parsed, err := url.Parse(signed) + if err != nil { + t.Fatal(err) + } + if parsed.Scheme != "https" || parsed.Host != "bucket-a.oss-cn-test.aliyuncs.com" || parsed.EscapedPath() != "/photos/cat.png" { + t.Fatalf("signed URL target = %s", parsed) + } + query := parsed.Query() + if query.Get("OSSAccessKeyId") != "access-key" || query.Get("Expires") != "1786582800" || query.Get("Signature") != "uA2xP4TwgB6YB17MaALkharfEmM=" { + t.Fatalf("signed URL query = %#v", query) + } + if _, err := client.SignGetURL(OSSObjectRequest{Endpoint: "https://oss-cn-test.aliyuncs.com", Bucket: "bucket-a", Key: "photos/cat.png"}, 10*time.Hour); err == nil { + t.Fatal("expected excessive signed URL lifetime to fail") + } +} + func TestOSSHTTPClientSignsACLGetDeleteAndStreamsGetMetadata(t *testing.T) { var calls int client := newTestOSSHTTPClient(t, func(r *http.Request) *http.Response { diff --git a/backend/internal/assets/oss_test.go b/backend/internal/assets/oss_test.go index d10898a..e971c96 100644 --- a/backend/internal/assets/oss_test.go +++ b/backend/internal/assets/oss_test.go @@ -9,6 +9,7 @@ import ( "log" "strings" "testing" + "time" ) type ossClientStub struct { @@ -19,6 +20,9 @@ type ossClientStub struct { body []byte contentType string getErr, errorToReturn error + signedRequest OSSObjectRequest + signedTTL time.Duration + signedURL string } func (s *ossClientStub) Put(_ context.Context, r OSSPutRequest) error { @@ -41,6 +45,10 @@ func (s *ossClientStub) Delete(_ context.Context, r OSSObjectRequest) error { s.deletes = append(s.deletes, r) return s.errorToReturn } +func (s *ossClientStub) SignGetURL(request OSSObjectRequest, ttl time.Duration) (string, error) { + s.signedRequest, s.signedTTL = request, ttl + return s.signedURL, s.errorToReturn +} func TestOSSPutUsesConfiguredAddressContentTypeAndPublicACL(t *testing.T) { client := &ossClientStub{} @@ -85,6 +93,23 @@ func TestOSSReadDeleteAndMissingMapping(t *testing.T) { t.Fatalf("delete missing=%v", err) } } + +func TestOSSSignsExternalReadsAgainstPublicBucketBaseURL(t *testing.T) { + client := &ossClientStub{signedURL: "https://bucket-a.oss-cn-test.aliyuncs.com/uploads/a.png?Signature=signed"} + store, err := NewOSS(OSSConfig{ + Endpoint: "https://oss-cn-test-internal.aliyuncs.com", Bucket: "bucket-a", PublicBaseURL: "https://bucket-a.oss-cn-test.aliyuncs.com", + }, client) + if err != nil { + t.Fatal(err) + } + got, err := store.SignReadURL("uploads/a.png", time.Hour) + if err != nil || got != client.signedURL { + t.Fatalf("signed URL = %q, err=%v", got, err) + } + if client.signedRequest.Endpoint != "https://bucket-a.oss-cn-test.aliyuncs.com" || client.signedRequest.Bucket != "bucket-a" || client.signedRequest.Key != "uploads/a.png" || client.signedTTL != time.Hour { + t.Fatalf("signed request = %#v, ttl=%s", client.signedRequest, client.signedTTL) + } +} func TestOSSRejectsUnsafeKeysAndIncompleteConfiguration(t *testing.T) { if _, err := NewOSS(OSSConfig{Endpoint: "e", Bucket: "b"}, &ossClientStub{}); err == nil { t.Fatal("expected config error") diff --git a/backend/internal/assets/service_test.go b/backend/internal/assets/service_test.go index 33522c2..e5151c4 100644 --- a/backend/internal/assets/service_test.go +++ b/backend/internal/assets/service_test.go @@ -6,6 +6,7 @@ import ( "errors" "io" "reflect" + "strings" "testing" "time" ) @@ -132,6 +133,25 @@ func TestDownloadUsesBlobBeforeRemoteFetcher(t *testing.T) { } } +func TestProviderAssetURLSignsStoredObjectWithoutPersistingSignature(t *testing.T) { + cat := &memoryCatalog{assets: []Asset{{ + ID: "asset-private", OwnerID: "owner-a", URL: "https://private-bucket.test/uploads/a.png", StoragePath: "uploads/a.png", + }}} + blobs := &signedMemoryBlobs{memoryBlobs: &memoryBlobs{}, signedURL: "https://private-bucket.test/uploads/a.png?Signature=temporary"} + svc := NewService(cat, blobs, nil, time.Now, nil) + + sourceURL, accessURL, err := svc.ResolveProviderAssetURL(context.Background(), "owner-a", "asset-private", time.Hour) + if err != nil { + t.Fatal(err) + } + if sourceURL != cat.assets[0].URL || accessURL != blobs.signedURL || blobs.signedKey != "uploads/a.png" || blobs.signedTTL != time.Hour { + t.Fatalf("resolved URLs = %q / %q, signer = %q / %s", sourceURL, accessURL, blobs.signedKey, blobs.signedTTL) + } + if strings.Contains(cat.assets[0].URL, "Signature=") { + t.Fatalf("temporary signature was persisted: %q", cat.assets[0].URL) + } +} + type memoryCatalog struct { assets []Asset createErr error @@ -205,6 +225,18 @@ type memoryBlobs struct { deleteErr error } +type signedMemoryBlobs struct { + *memoryBlobs + signedURL string + signedKey string + signedTTL time.Duration +} + +func (m *signedMemoryBlobs) SignReadURL(key string, ttl time.Duration) (string, error) { + m.signedKey, m.signedTTL = key, ttl + return m.signedURL, nil +} + func (m *memoryBlobs) Put(_ context.Context, key string, body io.Reader, _ int64, contentType string) (StoredObject, error) { m.putKey = key m.putBody, _ = io.ReadAll(body) diff --git a/backend/internal/httpapi/assets.go b/backend/internal/httpapi/assets.go index de93aa7..1c29db1 100644 --- a/backend/internal/httpapi/assets.go +++ b/backend/internal/httpapi/assets.go @@ -271,6 +271,20 @@ func (h *assetsHandler) download(w http.ResponseWriter, r *http.Request, public } return } + inline := r.URL.Query().Get("inline") == "1" + if inline { + signedURL, signed, signErr := h.service.SignedDownloadURL(r.Context(), scope, id, assets.DefaultSignedURLTTL) + if signErr != nil { + writeAssetError(w, signErr, public, "") + return + } + if signed { + w.Header().Set("Cache-Control", "private, no-store") + w.Header().Set("Location", signedURL) + w.WriteHeader(http.StatusTemporaryRedirect) + return + } + } blob, err := h.service.Download(r.Context(), scope, id) if err != nil { if public { @@ -281,7 +295,11 @@ func (h *assetsHandler) download(w http.ResponseWriter, r *http.Request, public return } defer blob.Body.Close() - writeBlob(w, blob, "private, no-store", contentDisposition(a.Name)) + disposition := contentDisposition(a.Name) + if inline { + disposition = inlineContentDisposition(a.Name) + } + writeBlob(w, blob, "private, no-store", disposition) } func (h *assetsHandler) serveStored(w http.ResponseWriter, r *http.Request, key string) { @@ -383,6 +401,12 @@ func writeBlob(w http.ResponseWriter, blob assets.Blob, cache, disposition strin _, _ = io.Copy(w, blob.Body) } func contentDisposition(name string) string { + return namedContentDisposition("attachment", name) +} +func inlineContentDisposition(name string) string { + return namedContentDisposition("inline", name) +} +func namedContentDisposition(kind, name string) string { clean := strings.TrimSpace(strings.NewReplacer("\r", "_", "\n", "_", "/", "_", "\\", "_").Replace(name)) if clean == "" { clean = "download" @@ -395,7 +419,7 @@ func contentDisposition(name string) string { ascii.WriteByte('_') } } - return `attachment; filename="` + ascii.String() + `"; filename*=UTF-8''` + url.PathEscape(clean) + return kind + `; filename="` + ascii.String() + `"; filename*=UTF-8''` + url.PathEscape(clean) } func requestOrigin(r *http.Request) string { scheme := "http" diff --git a/backend/internal/httpapi/assets_test.go b/backend/internal/httpapi/assets_test.go index 425c360..386634d 100644 --- a/backend/internal/httpapi/assets_test.go +++ b/backend/internal/httpapi/assets_test.go @@ -81,6 +81,13 @@ func (c *assetCatalog) DeleteOwner(_ context.Context, owner, id string) (assets. type assetBlobs struct{ values map[string][]byte } +type signedAssetBlobs struct { + *assetBlobs + url string +} + +func (b *signedAssetBlobs) SignReadURL(string, time.Duration) (string, error) { return b.url, nil } + func (b *assetBlobs) Put(_ context.Context, key string, r io.Reader, _ int64, _ string) (assets.StoredObject, error) { p, _ := io.ReadAll(r) b.values[key] = p @@ -164,6 +171,24 @@ func TestAssetsMultipartDownloadServingAndMethods(t *testing.T) { } } +func TestAssetInlineReadRedirectsToShortLivedSignedURL(t *testing.T) { + cat := &assetCatalog{values: []assets.Asset{{ + ID: "asset-private", OwnerID: "demo-merchant", Name: "private.png", URL: "https://private.test/private.png", StoragePath: "uploads/private.png", + }}} + blobs := &signedAssetBlobs{ + assetBlobs: &assetBlobs{values: map[string][]byte{"uploads/private.png": []byte("png")}}, + url: "https://private.test/private.png?OSSAccessKeyId=test&Expires=1&Signature=signed", + } + svc := assets.NewService(cat, blobs, nil, time.Now, nil) + platform, _ := httpapi.NewPlatformAuthorizer(httpapi.AuthState{}, nil) + h, _ := httpapi.NewAssetsHandler(svc, platform, publicapi.NewAuthenticator(publicapi.Config{APIKeys: "a:k"}), httpapi.AssetsConfig{}) + + inline := request(t, h, http.MethodGet, "/api/assets/asset-private/download?inline=1", nil, nil) + if inline.Code != http.StatusTemporaryRedirect || inline.Header().Get("Location") != blobs.url || inline.Header().Get("Cache-Control") != "private, no-store" || inline.Header().Get("Content-Disposition") != "" { + t.Fatalf("inline response = %d %#v", inline.Code, inline.Header()) + } +} + func TestAssetsLimitsAndInfrastructureErrorsDoNotLeak(t *testing.T) { cat := &assetCatalog{} svc := assets.NewService(cat, &assetBlobs{values: map[string][]byte{}}, nil, time.Now, nil) diff --git a/backend/internal/jobs/provider.go b/backend/internal/jobs/provider.go index c5a34de..bf64386 100644 --- a/backend/internal/jobs/provider.go +++ b/backend/internal/jobs/provider.go @@ -20,13 +20,19 @@ type ProviderResolver interface { Resolve(context.Context, string) (providers.Adapter, error) } +type ProviderAssetURLResolver interface { + ResolveProviderAssetURL(context.Context, string, string, time.Duration) (sourceURL string, accessURL string, err error) +} + func (registry ProviderRegistry) Resolve(_ context.Context, name string) (providers.Adapter, error) { return registry[name], nil } type ProviderProcessor struct { - Providers ProviderResolver - Store Store + Providers ProviderResolver + Store Store + AssetURLs ProviderAssetURLResolver + AssetURLTTL time.Duration } func (p ProviderProcessor) Advance(ctx context.Context, job Job) (Job, error) { @@ -49,6 +55,12 @@ func (p ProviderProcessor) Advance(ctx context.Context, job Job) (Job, error) { if err := json.Unmarshal(job.RequestPayload, &request); err != nil { return Job{}, errors.New("invalid provider request payload") } + if job.ProviderTaskID == "" && job.ProviderDispatchStartedAt == nil { + request, err = p.refreshAssetURLs(ctx, job, request) + if err != nil { + return Job{}, fmt.Errorf("prepare provider asset URLs: %w", err) + } + } var result providers.Result expectedStatus := job.Status if job.ProviderTaskID == "" { @@ -128,6 +140,47 @@ func (p ProviderProcessor) Advance(ctx context.Context, job Job) (Job, error) { return job, nil } +func (p ProviderProcessor) refreshAssetURLs(ctx context.Context, job Job, request providers.Request) (providers.Request, error) { + if p.AssetURLs == nil || len(job.InputAssetIDs) == 0 { + return request, nil + } + ttl := p.AssetURLTTL + if ttl <= 0 { + ttl = time.Hour + } + replacements := make(map[string]string, len(job.InputAssetIDs)) + seen := make(map[string]struct{}, len(job.InputAssetIDs)) + for _, assetID := range job.InputAssetIDs { + assetID = strings.TrimSpace(assetID) + if assetID == "" { + continue + } + if _, ok := seen[assetID]; ok { + continue + } + seen[assetID] = struct{}{} + sourceURL, accessURL, err := p.AssetURLs.ResolveProviderAssetURL(ctx, job.OwnerID, assetID, ttl) + if err != nil { + return providers.Request{}, err + } + if strings.TrimSpace(sourceURL) == "" || strings.TrimSpace(accessURL) == "" { + return providers.Request{}, errors.New("provider asset URL is unavailable") + } + replacements[sourceURL] = accessURL + } + for index, inputURL := range request.InputURLs { + if accessURL, ok := replacements[inputURL]; ok { + request.InputURLs[index] = accessURL + } + } + for index := range request.Materials { + if accessURL, ok := replacements[request.Materials[index].URL]; ok { + request.Materials[index].URL = accessURL + } + } + return request, nil +} + type ProviderJobBuilder struct { ImageProvider, VideoProvider string ImageModel, VideoModel string diff --git a/backend/internal/jobs/provider_test.go b/backend/internal/jobs/provider_test.go index 11ffa18..821bda3 100644 --- a/backend/internal/jobs/provider_test.go +++ b/backend/internal/jobs/provider_test.go @@ -203,6 +203,35 @@ func TestProviderBuilderAndProcessor(t *testing.T) { } } +func TestProviderProcessorRefreshesPrivateAssetURLImmediatelyBeforeSubmit(t *testing.T) { + store := newMemoryJobStore() + job := Job{ + ID: "job-private", OwnerID: "owner-a", Provider: "fixture", ReqKey: "model-a", Capability: "image.generate", Status: StatusQueued, + InputAssetIDs: []string{"asset-private"}, InputURLs: []string{"https://private.test/uploads/a.png"}, + RequestPayload: json.RawMessage(`{"capability":"image.generate","model":"model-a","prompt":"draw","inputUrls":["https://private.test/uploads/a.png"],"materials":[{"url":"https://private.test/uploads/a.png","type":"image"}]}`), + } + store.jobs[job.ID] = job + adapter := &countingProvider{result: providers.Result{TaskID: "provider-task", Status: providers.StatusQueued}} + resolver := &recordingProviderAssetURLResolver{ + sourceURL: "https://private.test/uploads/a.png", + accessURL: "https://private.test/uploads/a.png?OSSAccessKeyId=test&Signature=temporary", + } + processor := ProviderProcessor{Providers: ProviderRegistry{"fixture": adapter}, Store: store, AssetURLs: resolver} + + if _, err := processor.Advance(context.Background(), job); err != nil { + t.Fatal(err) + } + if resolver.owner != "owner-a" || resolver.assetID != "asset-private" || resolver.ttl != time.Hour { + t.Fatalf("resolver call = owner %q asset %q ttl %s", resolver.owner, resolver.assetID, resolver.ttl) + } + if !reflect.DeepEqual(adapter.request.InputURLs, []string{resolver.accessURL}) || len(adapter.request.Materials) != 1 || adapter.request.Materials[0].URL != resolver.accessURL { + t.Fatalf("submitted request = %#v", adapter.request) + } + if strings.Contains(string(job.RequestPayload), "Signature=temporary") || strings.Contains(string(store.jobs[job.ID].RequestPayload), "Signature=temporary") { + t.Fatal("temporary signature must not be persisted in the job request") + } +} + func TestProviderProcessorNeverResubmitsAfterPersistedDispatchIntent(t *testing.T) { store := newMemoryJobStore() started := time.Date(2026, 8, 13, 8, 0, 0, 0, time.UTC) @@ -290,6 +319,18 @@ func TestProviderJobBuilderClampsPriorityToPublicContract(t *testing.T) { type countingProvider struct { submits int result providers.Result + request providers.Request +} + +type recordingProviderAssetURLResolver struct { + sourceURL, accessURL string + owner, assetID string + ttl time.Duration +} + +func (resolver *recordingProviderAssetURLResolver) ResolveProviderAssetURL(_ context.Context, owner, assetID string, ttl time.Duration) (string, string, error) { + resolver.owner, resolver.assetID, resolver.ttl = owner, assetID, ttl + return resolver.sourceURL, resolver.accessURL, nil } type modelQueryProvider struct { @@ -308,8 +349,9 @@ func (provider *modelQueryProvider) QueryModel(_ context.Context, _ string, mode return provider.result, nil } -func (p *countingProvider) Submit(context.Context, providers.Request) (providers.Result, error) { +func (p *countingProvider) Submit(_ context.Context, request providers.Request) (providers.Result, error) { p.submits++ + p.request = request return p.result, nil } func (p *countingProvider) Query(context.Context, string) (providers.Result, error) { diff --git a/backend/internal/settings/service.go b/backend/internal/settings/service.go index 30245d3..6daa75b 100644 --- a/backend/internal/settings/service.go +++ b/backend/internal/settings/service.go @@ -585,7 +585,7 @@ func definitions() []Group { {ID: "evolink", Title: "EvoLink 图片 API", Description: "GPT Image 2 图片生成。", Fields: []Field{{Key: "EVOLINK_API_KEY", Label: "EvoLink API Key", Secret: true, Type: "password"}, {Key: "EVOLINK_BASE_URL", Label: "Base URL", DefaultValue: "https://api.evolink.ai"}, {Key: "EVOLINK_IMAGE_MODEL", Label: "图片模型", DefaultValue: "gpt-image-2"}, {Key: "EVOLINK_IMAGE_QUALITY", Label: "质量", DefaultValue: "medium"}}}, {ID: "seedance", Title: "Seedance 视频 API", Description: "火山方舟 API Key。", Fields: []Field{{Key: "SEEDANCE_API_KEY", Label: "方舟 API Key", Secret: true, Type: "password"}}}, {ID: "bailian", Title: "阿里云百炼 API", Description: "万相图片与视频。", Fields: []Field{{Key: "BAILIAN_API_KEY", Label: "百炼 API Key", Secret: true, Type: "password"}, {Key: "BAILIAN_BASE_URL", Label: "Base URL", DefaultValue: "https://llm-126wneubbdo6dbr5.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"}, {Key: "BAILIAN_IMAGE_MODEL", Label: "图片模型", DefaultValue: "wan2.7-image-pro"}, {Key: "BAILIAN_VIDEO_MODEL", Label: "视频模型", DefaultValue: "wan2.7-i2v-2026-04-25"}}}, - {ID: "oss", Title: "OSS 资产存储", Description: "共享资产存储。", Fields: []Field{{Key: "ALI_OSS_ENDPOINT", Label: "Endpoint"}, {Key: "ALI_OSS_BUCKET", Label: "Bucket"}, {Key: "ALI_OSS_ACCESS_KEY_ID", Label: "Access Key ID", Secret: true, Type: "password"}, {Key: "ALI_OSS_ACCESS_KEY_SECRET", Label: "Access Key Secret", Secret: true, Type: "password"}, {Key: "ALI_OSS_PUBLIC_BASE_URL", Label: "公开访问 Base URL"}}}, + {ID: "oss", Title: "OSS 资产存储", Description: "私有桶通过后端签名访问。", Fields: []Field{{Key: "ALI_OSS_ENDPOINT", Label: "Endpoint"}, {Key: "ALI_OSS_BUCKET", Label: "Bucket"}, {Key: "ALI_OSS_ACCESS_KEY_ID", Label: "Access Key ID", Secret: true, Type: "password"}, {Key: "ALI_OSS_ACCESS_KEY_SECRET", Label: "Access Key Secret", Secret: true, Type: "password"}, {Key: "ALI_OSS_PUBLIC_BASE_URL", Label: "Bucket Base URL（无需公共读）"}}}, } } diff --git a/components/asset-manager.tsx b/components/asset-manager.tsx index 23bcfb8..887e888 100644 --- a/components/asset-manager.tsx +++ b/components/asset-manager.tsx @@ -5,6 +5,7 @@ import { Download, Eye, ImageIcon, Info, Loader2, Music, RefreshCw, Trash2, X } import { clampPage, pageItems, Pagination } from "@/components/pagination"; import { modalEnter, modalExit, pulseFeedback, revealChildren, runScopedMotion } from "@/lib/ui/motion"; import { formatBillingAmount } from "@/lib/billing"; +import { assetPreviewUrl } from "@/lib/client/asset-urls"; import type { Asset, GenerationJob } from "@/lib/types"; type AssetView = "assets" | "tasks"; @@ -386,10 +387,10 @@ function renderAssetPreview(asset: Asset) { ); } if (asset.kind === "video" || isVideo(asset)) { - return