tt/ttstd-infinite-canvas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(media): 资源分发 - 统一 CDN 读取签名与源站降级边界

Browse Source
This commit is contained in:
ddCat
2026-09-22 23:26:28 +08:00
parent 11618e83fb
commit a66799b1f2
6 changed files with 114 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
backend/internal/app/resource.go
View File

@@ -15,6 +15,7 @@ import (
_ "image/png"
"infinite-canvas/backend/internal/kernel"
"io"
"log"
"mime"
"mime/multipart"
"net/http"
@@ -324,7 +325,7 @@ func (s *Service) openResourceRange(userID string, resource *model.Resource, ran
setting.Provider = firstNonEmpty(resource.Provider, setting.Provider)
setting.Endpoint = firstNonEmpty(resource.Endpoint, setting.Endpoint)
setting.Bucket = firstNonEmpty(resource.Bucket, setting.Bucket)
stream, err := getOSSObjectRange(setting, resource.ObjectKey, normalizeSingleByteRange(rangeHeader))
stream, err := getOriginOSSObjectRange(setting, resource.ObjectKey, normalizeSingleByteRange(rangeHeader))
if err != nil {
return nil, err
}
@@ -432,6 +433,9 @@ func writeLocalResourceObject(filePath string, body io.Reader) error {
}
// A cloud write is successful only after the configured origin accepts it.
// storeResourceObject writes to the configured origin and degrades to local storage
// when the external origin is unavailable. The resource binding is rewritten before
// the caller persists the ready state, so later reads follow the actual object location.
func (s *Service) storeResourceObject(resource *model.Resource, fileName string, body io.Reader) (string, error) {
if resource == nil {
return "", errors.New("资源不存在")
@@ -439,11 +443,36 @@ func (s *Service) storeResourceObject(resource *model.Resource, fileName string,
if resource.Provider == "local" {
return "", writeLocalResourceObject(filepath.Join(s.dataDir, "resources", filepath.FromSlash(resource.ObjectKey)), body)
}
setting, err := s.ossSettingForResource(resource.UserID, resource)
if err != nil {
return "", err
setting, settingErr := s.ossSettingForResource(resource.UserID, resource)
var etag string
var putErr error
if settingErr == nil {
etag, putErr = putOSSObject(setting, resource.ObjectKey, resource.MimeType, resource.Size, body)
}
return putOSSObject(setting, resource.ObjectKey, resource.MimeType, resource.Size, body)
if putErr == nil && settingErr == nil {
return etag, nil
}
fallbackErr := putErr
if fallbackErr == nil {
fallbackErr = settingErr
}
if seeker, ok := body.(io.Seeker); ok {
if _, seekErr := seeker.Seek(0, io.SeekStart); seekErr != nil {
return "", errors.Join(fallbackErr, fmt.Errorf("降级本地存储时重置读取位置失败：%w", seekErr))
}
}
localKey := localObjectKey(resource.UserID, resource.Kind, fileName, resource.MimeType, time.Now())
resource.Provider = "local"
resource.ObjectKey = localKey
resource.Endpoint = ""
resource.Bucket = ""
resource.StorageSettingID = ""
resource.ETag = ""
if localErr := writeLocalResourceObject(filepath.Join(s.dataDir, "resources", filepath.FromSlash(localKey)), body); localErr != nil {
return "", errors.Join(fallbackErr, fmt.Errorf("降级本地存储失败：%w", localErr))
}
log.Printf("object storage upload degraded to local storage: resource=%s error=%v", resource.ID, fallbackErr)
return "", nil
}
func (s *Service) retryStoredResource(userID string, resource *model.Resource, kind string, mimeType string, size int64, body io.Reader) (*model.Resource, error) {

1
backend/internal/app/resource_oss_fallback_test.go
View File

@@ -35,6 +35,7 @@ func newResourceFallbackTestService(t *testing.T) (*Service, *gorm.DB) {
// 配置了 OSS（enabled=1、endpoint 不可达）时上传应回退本地存储成功，
// 而不是把上传标记为失败——本地媒体导入不应因外部存储故障整体失败。
func TestStoreResourceDegradesToLocalWhenOSSUnavailable(t *testing.T) {
t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "127.0.0.1")
service, db := newResourceFallbackTestService(t)
seedOSSEnabled(t, db, "user-1", "http://127.0.0.1:1")

19
backend/internal/app/resource_test.go
View File

@@ -731,8 +731,9 @@ func TestNormalizeSingleByteRange(t *testing.T) {
func TestHydrateNewAPIChannel1ResourceUsesSignedOSSURL(t *testing.T) {
svc := newResourceTestService(t)
settingJSON, _ := json.Marshal(ossSettingValue{
Enabled: true, Provider: "aliyun", Endpoint: "https://oss-cn-test.aliyuncs.com", Bucket: "private-bucket",
AccessKeyID: "access-id", AccessKeySecret: "secret-value",
Enabled: true, Provider: "aliyun", Endpoint: "https://oss-cn-test.aliyuncs.com", CDNBaseURL: "https://media.example.com",
Bucket: "private-bucket", AccessKeyID: "access-id", AccessKeySecret: "secret-value",
Delivery: storage.DeliverySettings{CDNAuthMode: "public"},
})
if err := svc.repo.SaveSystemSetting(&model.SystemSetting{Key: ossSettingKey, ValueJSON: string(settingJSON)}); err != nil {
t.Fatal(err)
@@ -749,7 +750,7 @@ func TestHydrateNewAPIChannel1ResourceUsesSignedOSSURL(t *testing.T) {
if err := svc.hydrateProviderMedia("user-1", &media, providerMediaHydrationPolicy{requireURL: true}); err != nil {
t.Fatalf("hydrateProviderMedia() error = %v", err)
}
if !strings.HasPrefix(media.URL, "https://private-bucket.oss-cn-test.aliyuncs.com/") || media.DataURL != "" || !strings.Contains(media.URL, "Signature=") {
if media.URL != "https://media.example.com/users/user-1/image/reference.png" || media.DataURL != "" {
t.Fatalf("media = %#v", media)
}
if err := svc.hydrateProviderMedia("other-user", &providerMedia{StorageKey: "resource:resource-1"}, providerMediaHydrationPolicy{requireURL: true}); err == nil {
@@ -759,10 +760,11 @@ func TestHydrateNewAPIChannel1ResourceUsesSignedOSSURL(t *testing.T) {
func TestHydrateNewAPIChannel1ResourceUsesSignedLocalURL(t *testing.T) {
t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true")
t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "127.0.0.1")
server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
defer server.Close()
svc := newResourceTestService(t)
settingJSON, _ := json.Marshal(ossSettingValue{Provider: "aliyun", PublicBaseURL: server.URL})
settingJSON, _ := json.Marshal(ossSettingValue{Provider: "aliyun", PublicBaseURL: "https://127.0.0.1"})
if err := svc.repo.SaveSystemSetting(&model.SystemSetting{Key: ossSettingKey, ValueJSON: string(settingJSON)}); err != nil {
t.Fatal(err)
}
@@ -774,7 +776,7 @@ func TestHydrateNewAPIChannel1ResourceUsesSignedLocalURL(t *testing.T) {
if err := svc.hydrateProviderMedia("user-1", &media, providerMediaHydrationPolicy{requireURL: true}); err != nil {
t.Fatalf("hydrateProviderMedia() error = %v", err)
}
if !strings.HasPrefix(media.URL, server.URL+"/api/public/resources/resource-local/file/resource-local.png?") || !strings.Contains(media.URL, "signature=") || media.DataURL != "" {
if !strings.HasPrefix(media.URL, "https://127.0.0.1/api/public/resources/resource-local/file?") || !strings.Contains(media.URL, "signature=") || media.DataURL != "" {
t.Fatalf("media = %#v", media)
}
stored, err := svc.repo.Resource("resource-local")
@@ -786,8 +788,9 @@ func TestHydrateNewAPIChannel1ResourceUsesSignedLocalURL(t *testing.T) {
func TestHydratePreferredURLUsesObjectStorageAndFallsBackLocal(t *testing.T) {
svc := newResourceTestService(t)
settingJSON, _ := json.Marshal(ossSettingValue{
Enabled: true, Provider: "aliyun", Endpoint: "https://oss-cn-test.aliyuncs.com", Bucket: "private-bucket",
AccessKeyID: "access-id", AccessKeySecret: "secret-value",
Enabled: true, Provider: "aliyun", Endpoint: "https://oss-cn-test.aliyuncs.com", CDNBaseURL: "https://media.example.com",
Bucket: "private-bucket", AccessKeyID: "access-id", AccessKeySecret: "secret-value",
Delivery: storage.DeliverySettings{CDNAuthMode: "public"},
})
if err := svc.repo.SaveSystemSetting(&model.SystemSetting{Key: ossSettingKey, ValueJSON: string(settingJSON)}); err != nil {
t.Fatal(err)
@@ -804,7 +807,7 @@ func TestHydratePreferredURLUsesObjectStorageAndFallsBackLocal(t *testing.T) {
if err := svc.hydrateProviderMedia("user-1", &media, providerMediaHydrationPolicy{preferURL: true}); err != nil {
t.Fatalf("hydrateProviderMedia(prefer object) error = %v", err)
}
if !strings.HasPrefix(media.URL, "https://private-bucket.oss-cn-test.aliyuncs.com/") || media.DataURL != "" || !strings.Contains(media.URL, "Signature=") {
if media.URL != "https://media.example.com/users/user-1/image/prefer.png" || media.DataURL != "" {
t.Fatalf("object media = %#v", media)
}

2
backend/internal/app/storage_adapters.go
View File

@@ -12,6 +12,8 @@ var (
getOSSObjectRange = storage.GetOSSObjectRange
getAliyunOSSObjectRange = storage.GetAliyunOSSObjectRange
signedOSSObjectURL = storage.SignedOSSObjectURL
signedOriginObjectURL = storage.SignedOriginObjectURL
getOriginOSSObjectRange = storage.GetOriginObjectRange
signedAliyunOSSObjectURL = storage.SignedAliyunOSSObjectURL
putCOSObject = storage.PutCOSObject
putQiniuObject = storage.PutQiniuObject

2
backend/internal/assets/access.go
View File

@@ -129,7 +129,7 @@ func ResolveAccess(resource *model.Resource, setting storage.Settings, options A
} else if setting.Delivery.RequireCDN {
return nil, AccessError(503, "resource_cdn_unconfigured", "CDN 访问鉴权未配置，请检查存储分发设置")
} else if storage.PublicOrigin(setting) {
access.URL, err = storage.SignedOSSObjectURL(setting, resource.ObjectKey, expires)
access.URL, err = storage.SignedOriginObjectURL(setting, resource.ObjectKey, expires)
access.Delivery = DeliveryOrigin
if setting.CDNBaseURL != "" {
access.FallbackReason = "cdn_auth_unconfigured"

66
backend/internal/storage/objects.go
View File

@@ -16,6 +16,7 @@ import (
"infinite-canvas/backend/internal/kernel"
"infinite-canvas/backend/internal/outbound"
"io"
"net"
"net/http"
"net/url"
"strconv"
@@ -70,6 +71,18 @@ type ObjectStream struct {
}
func GetOSSObjectRange(setting Settings, objectKey string, rangeHeader string) (*ObjectStream, error) {
setting = NormalizeSettings(setting)
// This compatibility entry point is also used by older callers that only
// know CDNBaseURL. The application delivery policy uses GetOriginObjectRange
// for an explicit private-proxy fallback, so a missing CDN auth mode never
// turns that fallback into an accidental CDN request.
if setting.CDNBaseURL != "" && (setting.Delivery.CDNAuthMode == "" || CDNEnabled(setting)) {
return getCDNObjectRange(setting, objectKey, rangeHeader)
}
return GetOriginObjectRange(setting, objectKey, rangeHeader)
}
func GetOriginObjectRange(setting Settings, objectKey string, rangeHeader string) (*ObjectStream, error) {
setting = NormalizeSettings(setting)
if setting.Provider == s3Provider {
return GetS3ObjectRange(setting, objectKey, rangeHeader)
@@ -83,6 +96,39 @@ func GetOSSObjectRange(setting Settings, objectKey string, rangeHeader string) (
return GetAliyunOSSObjectRange(setting, objectKey, rangeHeader)
}
func getCDNObjectRange(setting Settings, objectKey string, rangeHeader string) (*ObjectStream, error) {
var (
urlValue string
err error
)
if setting.Delivery.CDNAuthMode == "" {
urlValue, err = OssCDNObjectURL(setting.CDNBaseURL, objectKey)
} else {
urlValue, err = SignCDNURL(setting, objectKey, time.Now().Add(resourceAccessURLTTL))
}
if err != nil {
return nil, err
}
req, err := http.NewRequest(http.MethodGet, urlValue, nil)
if err != nil {
return nil, err
}
if rangeHeader != "" {
req.Header.Set("Range", rangeHeader)
}
outbound.ApplyDefaultOutboundHeaders(req)
resp, err := outbound.OutboundHTTPClient(2 * time.Minute).Do(req)
if err != nil {
return nil, fmt.Errorf("CDN 读取失败：%w", err)
}
if (resp.StatusCode < 200 || resp.StatusCode >= 300) && resp.StatusCode != http.StatusRequestedRangeNotSatisfiable {
defer resp.Body.Close()
detail, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
return nil, fmt.Errorf("CDN 读取失败：%s %s", resp.Status, strings.TrimSpace(string(detail)))
}
return &ObjectStream{Body: resp.Body, StatusCode: resp.StatusCode, ContentLength: resp.ContentLength, ContentRange: resp.Header.Get("Content-Range"), AcceptRanges: kernel.FirstNonEmpty(resp.Header.Get("Accept-Ranges"), "bytes")}, nil
}
func GetAliyunOSSObjectRange(setting Settings, objectKey string, rangeHeader string) (*ObjectStream, error) {
req, err := NewOSSRequest(http.MethodGet, setting, objectKey, "", nil)
if err != nil {
@@ -104,6 +150,20 @@ func GetAliyunOSSObjectRange(setting Settings, objectKey string, rangeHeader str
}
func SignedOSSObjectURL(setting Settings, objectKey string, expiresAt time.Time) (string, error) {
setting = NormalizeSettings(setting)
// Keep the low-level helper compatible with historical callers that passed
// only CDNBaseURL. The application policy remains strict and calls
// SignedOriginObjectURL when CDN authentication is not explicitly enabled.
if setting.CDNBaseURL != "" && (setting.Delivery.CDNAuthMode == "" || CDNEnabled(setting)) {
if setting.Delivery.CDNAuthMode == "" {
return OssCDNObjectURL(setting.CDNBaseURL, objectKey)
}
return SignCDNURL(setting, objectKey, expiresAt)
}
return SignedOriginObjectURL(setting, objectKey, expiresAt)
}
func SignedOriginObjectURL(setting Settings, objectKey string, expiresAt time.Time) (string, error) {
setting = NormalizeSettings(setting)
if setting.Provider == s3Provider {
return SignedS3ObjectURL(setting, objectKey, expiresAt)
@@ -451,7 +511,11 @@ func OssBucketBaseURL(setting Settings) (*url.URL, error) {
if parsed.Host == "" {
return nil, errors.New("OSS Endpoint 格式不正确")
}
if !strings.HasPrefix(parsed.Host, setting.Bucket+".") {
host := parsed.Hostname()
// Local test/dev endpoints and explicit IP endpoints are commonly path-style
// services. Do not turn 127.0.0.1 into test-bucket.127.0.0.1, which both
// breaks the endpoint and makes the upload fallback impossible to exercise.
if host != "localhost" && net.ParseIP(host) == nil && !strings.HasPrefix(strings.ToLower(host), "localhost:") && !strings.HasPrefix(parsed.Host, setting.Bucket+".") {
parsed.Host = setting.Bucket + "." + parsed.Host
}
return parsed, nil