支持 S3 兼容对象存储 (#336)

* feat(storage): 支持 S3 对象存储

* fix(storage): 补齐 S3 对象文件后缀
This commit is contained in:
Rou authored and GitHub committed 2026-08-27 22:16:08 +08:00
1 parent d42e392ed7
commit 21b69d8bc7
25 files changed
+1374 -88

No files matched your search

+1
View File
@@ -43,6 +43,7 @@ func Models() []any {
&model.UserPluginState{},
&model.ArkPrivateAssetBinding{},
&model.UserOSSSetting{},
&model.StorageLocation{},
&model.UserDailyUploadUsage{},
&model.Skill{},
&model.UserSkillState{},
+23
View File
@@ -477,6 +477,7 @@ func RegisterAdminRoutes(r *gin.RouterGroup, svc *service.Service) {
failService(c, err)
return
}
c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 64<<10)
var req service.OSSSettingRequest
if err := c.ShouldBindJSON(&req); err != nil {
fail(c, http.StatusBadRequest, err)
@@ -489,6 +490,28 @@ func RegisterAdminRoutes(r *gin.RouterGroup, svc *service.Service) {
}
ok(c, gin.H{"setting": setting})
})
r.POST("/admin/settings/oss/test", func(c *gin.Context) {
user, err := currentUser(c, svc)
if err != nil {
failService(c, err)
return
}
if !enforceRateLimit(c, "admin-storage-test:"+user.ID, 6, time.Minute) {
return
}
c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 64<<10)
var req service.OSSSettingRequest
if err := c.ShouldBindJSON(&req); err != nil {
fail(c, http.StatusBadRequest, err)
return
}
result, err := svc.TestAdminOSSSetting(user, req)
if err != nil {
failService(c, err)
return
}
ok(c, result)
})
r.GET("/admin/settings/ark-private-assets", func(c *gin.Context) {
user, err := currentUser(c, svc)
if err != nil {
@@ -0,0 +1,32 @@
package handler
import (
"testing"
"infinite-canvas/backend/internal/service"
"github.com/gin-gonic/gin"
)
func TestStorageConnectionTestRoutesAreRegistered(t *testing.T) {
gin.SetMode(gin.TestMode)
router := gin.New()
group := router.Group("/api")
RegisterAdminRoutes(group, &service.Service{})
RegisterUserDataRoutes(group, &service.Service{})
wanted := map[string]bool{
"POST /api/admin/settings/oss/test": false,
"POST /api/settings/oss/test": false,
}
for _, route := range router.Routes() {
key := route.Method + " " + route.Path
if _, exists := wanted[key]; exists {
wanted[key] = true
}
}
for route, found := range wanted {
if !found {
t.Errorf("route %s is not registered", route)
}
}
}
+22
View File
@@ -92,6 +92,28 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) {
}
ok(c, gin.H{"setting": setting})
})
r.POST("/settings/oss/test", func(c *gin.Context) {
user, err := currentUser(c, svc)
if err != nil {
failService(c, err)
return
}
if !enforceRateLimit(c, "user-storage-test:"+user.ID, 6, time.Minute) {
return
}
c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 64<<10)
var req service.OSSSettingRequest
if err := c.ShouldBindJSON(&req); err != nil {
fail(c, http.StatusBadRequest, err)
return
}
result, err := svc.TestUserOSSSetting(user, req)
if err != nil {
failService(c, err)
return
}
ok(c, result)
})
r.GET("/resources", func(c *gin.Context) {
user, err := currentUser(c, svc)
if err != nil {
+16
View File
@@ -50,6 +50,22 @@ type UserOSSSetting struct {
UpdatedAt time.Time `json:"updatedAt"`
}
// StorageLocation is an immutable object-storage address. Credentials may be
// rotated in place, while resources keep referring to the same location ID.
type StorageLocation struct {
ID string `json:"id" gorm:"primaryKey;size:36"`
Scope string `json:"scope" gorm:"size:16;index:idx_storage_locations_scope_owner_active,priority:1;uniqueIndex:idx_storage_locations_identity,priority:1"`
OwnerID string `json:"ownerId" gorm:"size:36;index:idx_storage_locations_scope_owner_active,priority:2;uniqueIndex:idx_storage_locations_identity,priority:2"`
Provider string `json:"provider" gorm:"size:24;index;uniqueIndex:idx_storage_locations_identity,priority:3"`
LocationDigest string `json:"-" gorm:"size:64;uniqueIndex:idx_storage_locations_identity,priority:4"`
ValueJSON string `json:"-" gorm:"type:text"`
TestedDigest string `json:"-" gorm:"size:64"`
TestedAt *time.Time `json:"testedAt,omitempty"`
Active bool `json:"active" gorm:"index:idx_storage_locations_scope_owner_active,priority:3"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
type UserDailyUploadUsage struct {
ID string `json:"id" gorm:"primaryKey;size:64"`
UserID string `json:"userId" gorm:"size:36;index;uniqueIndex:idx_user_daily_upload_day,priority:1"`
+55
View File
@@ -840,6 +840,61 @@ func (r *Repository) CreateUserOSSSetting(setting *model.UserOSSSetting) error {
return r.db.Create(setting).Error
}
func (r *Repository) StorageLocation(id string) (*model.StorageLocation, error) {
var location model.StorageLocation
if err := r.db.First(&location, "id = ?", id).Error; err != nil {
return nil, err
}
return &location, nil
}
func (r *Repository) StorageLocationByDigest(scope string, ownerID string, provider string, digest string) (*model.StorageLocation, error) {
var location model.StorageLocation
if err := r.db.First(&location, "scope = ? AND owner_id = ? AND provider = ? AND location_digest = ?", scope, ownerID, provider, digest).Error; err != nil {
return nil, err
}
return &location, nil
}
func (r *Repository) StorageLocationHistoryCount(scope string, ownerID string) (int64, error) {
var count int64
err := r.db.Model(&model.StorageLocation{}).Where("scope = ? AND owner_id = ?", scope, ownerID).Count(&count).Error
return count, err
}
func (r *Repository) StorageLocationResourceCount(id string) (int64, error) {
var count int64
err := r.db.Model(&model.Resource{}).Where("storage_setting_id = ?", id).Count(&count).Error
return count, err
}
func (r *Repository) CreateStorageLocation(location *model.StorageLocation) error {
return r.db.Create(location).Error
}
func (r *Repository) SaveStorageLocation(location *model.StorageLocation) error {
return r.db.Save(location).Error
}
func (r *Repository) ActivateStorageLocation(scope string, ownerID string, id string, active bool) error {
return r.db.Transaction(func(tx *gorm.DB) error {
if err := tx.Model(&model.StorageLocation{}).Where("scope = ? AND owner_id = ? AND active = ?", scope, ownerID, true).Update("active", false).Error; err != nil {
return err
}
if !active {
return nil
}
result := tx.Model(&model.StorageLocation{}).Where("id = ? AND scope = ? AND owner_id = ?", id, scope, ownerID).Update("active", true)
if result.Error != nil {
return result.Error
}
if result.RowsAffected != 1 {
return gorm.ErrRecordNotFound
}
return nil
})
}
func (r *Repository) ReserveDailyUpload(userID string, day string, size int64, limit int64) error {
usage := model.UserDailyUploadUsage{ID: userID + ":" + day, UserID: userID, Day: day}
return r.db.Transaction(func(tx *gorm.DB) error {
+74 -15
View File
@@ -103,6 +103,9 @@ func (s *Service) directResourceURL(resource *model.Resource, expiresAt time.Tim
setting.Provider = firstNonEmpty(resource.Provider, setting.Provider)
setting.Endpoint = firstNonEmpty(resource.Endpoint, setting.Endpoint)
setting.Bucket = firstNonEmpty(resource.Bucket, setting.Bucket)
if setting.Provider == s3Provider && !publicHTTPSStorageEndpoint(setting.Endpoint) {
return s.signedHTTPSPublicResourceURL(resource.ID, expiresAt)
}
return signedOSSObjectURL(setting, resource.ObjectKey, expiresAt)
}
@@ -130,6 +133,11 @@ func (s *Service) prepareResourceDelivery(userID string, resource *model.Resourc
if err != nil {
return nil, err
}
// S3 兼容 Endpoint 可能是私网服务;浏览器默认始终使用同源代理。
// 只有明确的服务端上游需求才签发可公开访问的短时地址。
if setting.Provider == s3Provider && !options.ForceDirect {
return &ResourceDelivery{Resource: resource}, nil
}
if setting.Provider == qiniuKodoProvider && setting.CDNBaseURL != "" {
// 七牛私有空间即使配置了绑定域名,也不能匿名访问;必须使用
// Kodo 私有下载签名,否则浏览器会收到 NotSupportAnonymous。
@@ -147,6 +155,13 @@ func (s *Service) prepareResourceDelivery(userID string, resource *model.Resourc
return &ResourceDelivery{Resource: resource, RedirectURL: redirectURL}, nil
}
if options.ForceDirect {
if setting.Provider == s3Provider && !publicHTTPSStorageEndpoint(setting.Endpoint) {
redirectURL, err := s.signedHTTPSPublicResourceURL(resource.ID, time.Now().Add(directResourceURLTTL))
if err != nil {
return nil, err
}
return &ResourceDelivery{Resource: resource, RedirectURL: redirectURL}, nil
}
redirectURL, err := signedOSSObjectURL(setting, resource.ObjectKey, time.Now().Add(directResourceURLTTL))
if err != nil {
return nil, err
@@ -175,6 +190,17 @@ func (s *Service) signedPublicResourceURL(resourceID string, expiresAt time.Time
return baseURL.String(), nil
}
func (s *Service) signedHTTPSPublicResourceURL(resourceID string, expiresAt time.Time) (string, error) {
baseURL, err := s.publicResourceBaseURL()
if err != nil {
return "", err
}
if baseURL.Scheme != "https" {
return "", BadAuthRequest("私网或 HTTP S3 用于上游资源时,服务器公开访问地址必须使用 HTTPS")
}
return s.signedPublicResourceURL(resourceID, expiresAt)
}
func (s *Service) verifyPublicResourceSignature(resourceID string, expires string, signature string) error {
if strings.TrimSpace(signature) == "" || !decimalDigits(expires) {
return Forbidden("匿名下载链接无效")
@@ -323,7 +349,7 @@ func (s *Service) OpenPublicResourceRange(id string, expires string, signature s
if err != nil {
return nil, Forbidden("匿名下载链接无效")
}
if resource.Provider != "local" {
if resource.Provider != "local" && resource.Provider != s3Provider {
return nil, Forbidden("匿名下载链接无效")
}
if err := s.verifyPublicResourceSignature(resource.ID, expires, signature); err != nil {
@@ -368,11 +394,11 @@ func (s *Service) storeResource(userID string, kind string, fileName string, mim
return nil, err
}
provider := "local"
objectKey := localObjectKey(userID, kind, fileName, now)
objectKey := localObjectKey(userID, kind, fileName, mimeType, now)
resource := model.Resource{ID: newID(), UserID: userID, Kind: kind, Status: model.ResourceStatusPending, Provider: provider, ObjectKey: objectKey, MimeType: mimeType, Size: size, Width: width, Height: height, DurationMs: durationMs, CreatedAt: now, UpdatedAt: now}
if useOSS {
provider = setting.Provider
objectKey = ossObjectKey(setting, userID, kind, fileName, now)
objectKey = ossObjectKey(setting, userID, kind, fileName, mimeType, now)
resource.Provider = provider
resource.Endpoint = setting.Endpoint
resource.Bucket = setting.Bucket
@@ -431,8 +457,8 @@ func (s *Service) storeResource(userID string, kind string, fileName string, mim
return &resource, nil
}
func localObjectKey(userID string, kind string, fileName string, now time.Time) string {
ext := strings.ToLower(filepath.Ext(fileName))
func localObjectKey(userID string, kind string, fileName string, mimeType string, now time.Time) string {
ext := resourceFileExtension(fileName, mimeType, kind)
return path.Join("users", safeObjectSegment(userID), kind, now.Format("2006/01/02"), newID()+ext)
}
@@ -676,6 +702,26 @@ func extensionFromMimeType(mimeType string) string {
return "bin"
}
func resourceFileExtension(fileName string, mimeType string, kind string) string {
if ext := strings.ToLower(filepath.Ext(strings.TrimSpace(fileName))); ext != "" && ext != "." {
return ext
}
cleanMimeType := strings.TrimSpace(strings.Split(mimeType, ";")[0])
if extensions, err := mime.ExtensionsByType(cleanMimeType); err == nil && len(extensions) > 0 {
return strings.ToLower(extensions[0])
}
switch kind {
case "image":
return ".png"
case "video":
return ".mp4"
case "audio":
return ".mp3"
default:
return ".bin"
}
}
func imageDimensions(data []byte) (int, int) {
config, _, err := image.DecodeConfig(bytes.NewReader(data))
if err != nil {
@@ -697,26 +743,30 @@ func (s *Service) activeResourceOSSSetting(userID string) (ossSettingValue, stri
if err != nil {
return ossSettingValue{}, "", false, err
}
if userSetting != nil && value.Enabled {
value, err = validateActiveOSSSetting(value, "用户 OSS 尚未启用", "你的 OSS 配置不完整")
return value, userSetting.ID, true, err
}
_, systemValue, err := s.readOSSSetting()
if err != nil {
return ossSettingValue{}, "", false, err
}
userAllowed := value.Provider != s3Provider || systemValue.AllowUserS3
if userSetting != nil && value.Enabled && userAllowed {
value, err = validateActiveOSSSetting(value, "用户 OSS 尚未启用", "你的 OSS 配置不完整")
return value, firstNonEmpty(value.StorageLocationID, userSetting.ID), true, err
}
if !systemValue.Enabled {
return ossSettingValue{}, "", false, nil
}
systemValue, err = validateActiveOSSSetting(systemValue, "管理员尚未启用 OSS", "平台 OSS 配置不完整,请联系管理员")
return systemValue, "", true, err
return systemValue, systemValue.StorageLocationID, true, err
}
func (s *Service) ossSettingForResource(userID string, resource *model.Resource) (ossSettingValue, error) {
var setting ossSettingValue
var err error
if resource.StorageSettingID != "" {
_, setting, err = s.readUserOSSSettingByID(userID, resource.StorageSettingID)
_, setting, err = s.storageLocationValue(resource.StorageSettingID)
if errors.Is(err, gorm.ErrRecordNotFound) {
_, setting, err = s.readUserOSSSettingByID(userID, resource.StorageSettingID)
}
if err == nil {
_, current, currentErr := s.readUserOSSSetting(userID)
if currentErr != nil {
@@ -805,8 +855,8 @@ func validateActiveOSSSetting(setting ossSettingValue, disabledMessage string, i
if !setting.Enabled {
return ossSettingValue{}, BadAuthRequest(disabledMessage)
}
if setting.Provider != aliyunOSSProvider && setting.Provider != tencentCOSProvider && setting.Provider != qiniuKodoProvider {
return ossSettingValue{}, BadAuthRequest("仅支持阿里云 OSS、腾讯云 COS 和七牛云 Kodo")
if setting.Provider != aliyunOSSProvider && setting.Provider != tencentCOSProvider && setting.Provider != qiniuKodoProvider && setting.Provider != s3Provider {
return ossSettingValue{}, BadAuthRequest("仅支持阿里云 OSS、腾讯云 COS、七牛云 Kodo 和通用 S3")
}
if setting.Bucket == "" || setting.Endpoint == "" || setting.AccessKeyID == "" || setting.AccessKeySecret == "" {
return ossSettingValue{}, BadAuthRequest(incompleteMessage)
@@ -832,8 +882,8 @@ func normalizeResourceKind(kind string, mimeType string) string {
return "file"
}
func ossObjectKey(setting ossSettingValue, userID string, kind string, fileName string, now time.Time) string {
ext := strings.ToLower(filepath.Ext(fileName))
func ossObjectKey(setting ossSettingValue, userID string, kind string, fileName string, mimeType string, now time.Time) string {
ext := resourceFileExtension(fileName, mimeType, kind)
name := newID()
parts := []string{setting.PathPrefix, "users", safeObjectSegment(userID), kind, now.Format("2006/01/02"), name + ext}
return strings.Trim(strings.Join(nonEmptySegments(parts), "/"), "/")
@@ -847,6 +897,9 @@ func putOSSObject(setting ossSettingValue, objectKey string, mimeType string, si
if setting.Provider == qiniuKodoProvider {
return putQiniuObject(setting, objectKey, mimeType, size, body)
}
if setting.Provider == s3Provider {
return putS3Object(setting, objectKey, mimeType, size, body)
}
return putAliyunOSSObject(setting, objectKey, mimeType, size, body)
}
@@ -884,6 +937,9 @@ type ossObjectStream struct {
func getOSSObjectRange(setting ossSettingValue, objectKey string, rangeHeader string) (*ossObjectStream, error) {
setting = normalizeOSSSetting(setting)
if setting.Provider == s3Provider {
return getS3ObjectRange(setting, objectKey, rangeHeader)
}
if setting.CDNBaseURL != "" {
return getOSSObjectRangeViaCDN(setting, objectKey, rangeHeader)
}
@@ -939,6 +995,9 @@ func decimalDigits(value string) bool {
func signedOSSObjectURL(setting ossSettingValue, objectKey string, expiresAt time.Time) (string, error) {
setting = normalizeOSSSetting(setting)
if setting.Provider == s3Provider {
return signedS3ObjectURL(setting, objectKey, expiresAt)
}
if setting.Provider == qiniuKodoProvider {
return signedQiniuObjectURL(setting, objectKey, expiresAt)
}
@@ -310,6 +310,12 @@ func (s *Service) deleteStoredResourceObject(userID string, resource *model.Reso
return fmt.Errorf("无法读取七牛云 Kodo 配置:%w", err)
}
return deleteQiniuObject(setting, resource.ObjectKey)
case s3Provider:
setting, err := s.ossSettingForResource(userID, resource)
if err != nil {
return fmt.Errorf("无法读取 S3 配置:%w", err)
}
return deleteS3Object(setting, resource.ObjectKey)
default:
return fmt.Errorf("资源 %s 使用了不支持的存储类型 %q", resource.ID, resource.Provider)
}
+1 -1
View File
@@ -825,7 +825,7 @@ func newResourceTestService(t *testing.T) *Service {
if err != nil {
t.Fatal(err)
}
if err := db.AutoMigrate(&model.SystemSetting{}, &model.UserOSSSetting{}, &model.UserDailyUploadUsage{}, &model.Resource{}, &model.SessionFile{}); err != nil {
if err := db.AutoMigrate(&model.SystemSetting{}, &model.UserOSSSetting{}, &model.StorageLocation{}, &model.UserDailyUploadUsage{}, &model.Resource{}, &model.SessionFile{}); err != nil {
t.Fatal(err)
}
return &Service{repo: repository.New(db), dataDir: t.TempDir()}
+3 -1
View File
@@ -25,6 +25,8 @@ type Service struct {
emailCodeMu sync.Mutex
redeemBatchMu sync.Mutex
storageMu sync.Mutex
storageTestMu sync.Mutex
activeStorageTests map[string]bool
characterTaskMu sync.Mutex
activeCancels map[string]context.CancelFunc
pendingStorage map[string]int64
@@ -102,7 +104,7 @@ func New(repo *repository.Repository, dataDir string) *Service {
func NewWithRuntimeCapabilities(repo *repository.Repository, dataDir string, capabilities RuntimeCapabilities) *Service {
coordinator, err := newRuntimeCoordinator(repo.Dialect())
pluginRuntime, pluginRuntimeErr := newPluginRuntime(dataDir)
service := &Service{repo: repo, dataDir: dataDir, runtimeCapabilities: capabilities, activeCancels: make(map[string]context.CancelFunc), coordinator: coordinator, runtimeErr: err, pluginRuntime: pluginRuntime, pluginRuntimeErr: pluginRuntimeErr, workerID: newID(), routeCatalogTTL: 30 * time.Second, routeCatalogMaxStale: 5 * time.Minute, routeHealthBlocked: make(map[string]time.Time)}
service := &Service{repo: repo, dataDir: dataDir, runtimeCapabilities: capabilities, activeStorageTests: make(map[string]bool), activeCancels: make(map[string]context.CancelFunc), coordinator: coordinator, runtimeErr: err, pluginRuntime: pluginRuntime, pluginRuntimeErr: pluginRuntimeErr, workerID: newID(), routeCatalogTTL: 30 * time.Second, routeCatalogMaxStale: 5 * time.Minute, routeHealthBlocked: make(map[string]time.Time)}
service.taskBillingCoordinator = newTaskBillingCoordinator(service.repo)
service.taskTerminalCoordinator = newTaskTerminalCoordinator(service)
service.taskRouteExecutor = newTaskRouteExecutor(service)
+197 -37
View File
@@ -4,13 +4,16 @@ import (
"crypto/aes"
"crypto/cipher"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"path/filepath"
"strconv"
"strings"
"time"
@@ -21,11 +24,13 @@ import (
const ossSettingKey = "oss"
const encryptedSettingPrefix = "enc:v1:"
const defaultOSSPathPrefix = "open-ai-canvas"
const (
aliyunOSSProvider = "aliyun"
tencentCOSProvider = "tencent"
qiniuKodoProvider = "qiniu"
s3Provider = "s3"
)
type OSSSettingRequest struct {
@@ -39,35 +44,53 @@ type OSSSettingRequest struct {
AccessKeySecret string `json:"accessKeySecret"`
PublicBaseURL string `json:"publicBaseUrl"`
PathPrefix string `json:"pathPrefix"`
S3Preset string `json:"s3Preset"`
PathStyle bool `json:"pathStyle"`
SessionToken string `json:"sessionToken"`
AllowUserS3 bool `json:"allowUserS3"`
}
type PublicOSSSetting struct {
Enabled bool `json:"enabled"`
Provider string `json:"provider"`
Region string `json:"region"`
Endpoint string `json:"endpoint"`
CDNBaseURL string `json:"cdnBaseUrl"`
Bucket string `json:"bucket"`
AccessKeyID string `json:"accessKeyId"`
HasAccessKeySecret bool `json:"hasAccessKeySecret"`
PublicBaseURL string `json:"publicBaseUrl"`
PathPrefix string `json:"pathPrefix"`
UpdatedBy string `json:"updatedBy"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
Enabled bool `json:"enabled"`
Provider string `json:"provider"`
Region string `json:"region"`
Endpoint string `json:"endpoint"`
CDNBaseURL string `json:"cdnBaseUrl"`
Bucket string `json:"bucket"`
AccessKeyID string `json:"accessKeyId"`
HasAccessKeySecret bool `json:"hasAccessKeySecret"`
PublicBaseURL string `json:"publicBaseUrl"`
PathPrefix string `json:"pathPrefix"`
S3Preset string `json:"s3Preset"`
PathStyle bool `json:"pathStyle"`
HasSessionToken bool `json:"hasSessionToken"`
StorageLocationID string `json:"storageLocationId"`
TestedAt *time.Time `json:"testedAt,omitempty"`
TestedDigest string `json:"testedDigest,omitempty"`
HistoryCount int64 `json:"historyCount"`
ReferencedResourceCount int64 `json:"referencedResourceCount"`
AllowUserS3 bool `json:"allowUserS3"`
UpdatedBy string `json:"updatedBy"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
type ossSettingValue struct {
Enabled bool `json:"enabled"`
Provider string `json:"provider"`
Region string `json:"region"`
Endpoint string `json:"endpoint"`
CDNBaseURL string `json:"cdnBaseUrl"`
Bucket string `json:"bucket"`
AccessKeyID string `json:"accessKeyId"`
AccessKeySecret string `json:"accessKeySecret"`
PublicBaseURL string `json:"publicBaseUrl"`
PathPrefix string `json:"pathPrefix"`
Enabled bool `json:"enabled"`
Provider string `json:"provider"`
Region string `json:"region"`
Endpoint string `json:"endpoint"`
CDNBaseURL string `json:"cdnBaseUrl"`
Bucket string `json:"bucket"`
AccessKeyID string `json:"accessKeyId"`
AccessKeySecret string `json:"accessKeySecret"`
PublicBaseURL string `json:"publicBaseUrl"`
PathPrefix string `json:"pathPrefix"`
S3Preset string `json:"s3Preset"`
PathStyle bool `json:"pathStyle"`
SessionToken string `json:"sessionToken"`
StorageLocationID string `json:"storageLocationId"`
AllowUserS3 bool `json:"allowUserS3"`
// 平台切换云厂商后仍需读取历史资源,因此仅归档非当前厂商的访问密钥。
ArchivedCredentials map[string]ossProviderCredentials `json:"archivedCredentials,omitempty"`
}
@@ -85,7 +108,10 @@ func (s *Service) AdminOSSSetting(actor *model.User) (*PublicOSSSetting, error)
if err != nil {
return nil, err
}
public := publicOSSSetting(setting, value)
public, err := s.publicOSSSetting(setting, value, "platform", "")
if err != nil {
return nil, err
}
return &public, nil
}
@@ -110,6 +136,15 @@ func (s *Service) UpdateOSSSetting(actor *model.User, req OSSSettingRequest) (*P
}
}
next = archiveOSSProviderCredentials(next, currentValue)
if next.Provider == s3Provider && next.Enabled {
location, err := s.requireTestedS3Location("platform", "", next)
if err != nil {
return nil, err
}
next.StorageLocationID = location.ID
} else if next.Provider == s3Provider && storageLocationDigest(next) == storageLocationDigest(currentValue) {
next.StorageLocationID = currentValue.StorageLocationID
}
stored, err := s.encryptOSSSettingSecrets(next)
if err != nil {
return nil, err
@@ -129,7 +164,19 @@ func (s *Service) UpdateOSSSetting(actor *model.User, req OSSSettingRequest) (*P
if err := s.repo.SaveSystemSetting(&setting); err != nil {
return nil, err
}
public := publicOSSSetting(&setting, next)
if next.Provider == s3Provider && next.StorageLocationID != "" {
if err := s.repo.ActivateStorageLocation("platform", "", next.StorageLocationID, next.Enabled); err != nil {
return nil, err
}
} else if currentValue.Provider == s3Provider && currentValue.StorageLocationID != "" {
if err := s.repo.ActivateStorageLocation("platform", "", "", false); err != nil {
return nil, err
}
}
public, err := s.publicOSSSetting(&setting, next, "platform", "")
if err != nil {
return nil, err
}
return &public, nil
}
@@ -141,7 +188,14 @@ func (s *Service) UserOSSSetting(actor *model.User) (*PublicOSSSetting, error) {
if err != nil {
return nil, err
}
public := publicUserOSSSetting(setting, value)
_, platform, err := s.readOSSSetting()
if err != nil {
return nil, err
}
public, err := s.publicUserOSSSetting(setting, value, actor.ID, platform.AllowUserS3)
if err != nil {
return nil, err
}
return &public, nil
}
@@ -157,6 +211,22 @@ func (s *Service) UpdateUserOSSSetting(actor *model.User, req OSSSettingRequest)
if err != nil {
return nil, err
}
_, platform, err := s.readOSSSetting()
if err != nil {
return nil, err
}
if next.Enabled && next.Provider == s3Provider && !platform.AllowUserS3 {
return nil, Forbidden("平台管理员尚未允许个人 S3 兼容存储")
}
if next.Provider == s3Provider && next.Enabled {
location, err := s.requireTestedS3Location("user", actor.ID, next)
if err != nil {
return nil, err
}
next.StorageLocationID = location.ID
} else if next.Provider == s3Provider && storageLocationDigest(next) == storageLocationDigest(currentValue) {
next.StorageLocationID = currentValue.StorageLocationID
}
stored, err := s.encryptOSSSettingSecrets(next)
if err != nil {
return nil, err
@@ -170,7 +240,19 @@ func (s *Service) UpdateUserOSSSetting(actor *model.User, req OSSSettingRequest)
if err := s.repo.CreateUserOSSSetting(&setting); err != nil {
return nil, err
}
public := publicUserOSSSetting(&setting, next)
if next.Provider == s3Provider && next.StorageLocationID != "" {
if err := s.repo.ActivateStorageLocation("user", actor.ID, next.StorageLocationID, next.Enabled); err != nil {
return nil, err
}
} else if currentValue.Provider == s3Provider && currentValue.StorageLocationID != "" {
if err := s.repo.ActivateStorageLocation("user", actor.ID, "", false); err != nil {
return nil, err
}
}
public, err := s.publicUserOSSSetting(&setting, next, actor.ID, platform.AllowUserS3)
if err != nil {
return nil, err
}
return &public, nil
}
@@ -250,6 +332,10 @@ func (s *Service) encryptOSSSettingSecrets(value ossSettingValue) (ossSettingVal
if err != nil {
return ossSettingValue{}, err
}
value.SessionToken, err = s.encryptSettingSecret(value.SessionToken)
if err != nil {
return ossSettingValue{}, err
}
value.ArchivedCredentials = cloneOSSProviderCredentials(value.ArchivedCredentials)
for provider, credentials := range value.ArchivedCredentials {
credentials.AccessKeySecret, err = s.encryptSettingSecret(credentials.AccessKeySecret)
@@ -268,6 +354,13 @@ func (s *Service) decryptOSSSettingSecrets(value *ossSettingValue) (bool, error)
return false, err
}
value.AccessKeySecret = secret
if value.SessionToken != "" && !strings.HasPrefix(value.SessionToken, encryptedSettingPrefix) {
needsMigration = true
}
value.SessionToken, err = s.decryptSettingSecret(value.SessionToken)
if err != nil {
return false, err
}
value.ArchivedCredentials = cloneOSSProviderCredentials(value.ArchivedCredentials)
for provider, credentials := range value.ArchivedCredentials {
if credentials.AccessKeySecret != "" && !strings.HasPrefix(credentials.AccessKeySecret, encryptedSettingPrefix) {
@@ -466,15 +559,29 @@ func ossSettingFromRequest(req OSSSettingRequest, current ossSettingValue) (ossS
AccessKeySecret: strings.TrimSpace(req.AccessKeySecret),
PublicBaseURL: strings.TrimRight(strings.TrimSpace(req.PublicBaseURL), "/"),
PathPrefix: strings.Trim(strings.TrimSpace(req.PathPrefix), "/"),
S3Preset: strings.TrimSpace(req.S3Preset),
PathStyle: req.PathStyle,
SessionToken: strings.TrimSpace(req.SessionToken),
AllowUserS3: req.AllowUserS3,
})
if next.Provider != aliyunOSSProvider && next.Provider != tencentCOSProvider && next.Provider != qiniuKodoProvider {
return next, BadAuthRequest("仅支持阿里云 OSS、腾讯云 COS 和七牛云 Kodo")
if next.Provider != aliyunOSSProvider && next.Provider != tencentCOSProvider && next.Provider != qiniuKodoProvider && next.Provider != s3Provider {
return next, BadAuthRequest("仅支持阿里云 OSS、腾讯云 COS、七牛云 Kodo 和通用 S3")
}
if next.Provider == s3Provider {
switch next.S3Preset {
case "aws", "r2", "b2", "rustfs", "custom":
default:
return next, BadAuthRequest("S3 预设无效")
}
}
current = normalizeOSSSetting(current)
// 不同云厂商的密钥不能复用;只有继续使用同一厂商时,留空才表示保留原密钥。
// 同一云厂商修改存储位置时,留空仍表示保留现有密钥;切换厂商时不能复用。
if next.AccessKeySecret == "" && next.Provider == current.Provider {
next.AccessKeySecret = current.AccessKeySecret
}
if next.SessionToken == "" && next.Provider == current.Provider {
next.SessionToken = current.SessionToken
}
if next.Enabled {
if next.Bucket == "" {
return next, BadAuthRequest("请填写对象存储 Bucket")
@@ -486,10 +593,22 @@ func ossSettingFromRequest(req OSSSettingRequest, current ossSettingValue) (ossS
if next.Provider == qiniuKodoProvider {
return next, BadAuthRequest("请填写七牛云 Kodo 上传 Endpoint")
}
if next.Provider == s3Provider {
return next, BadAuthRequest("请填写 S3 Endpoint 服务根 URL")
}
return next, BadAuthRequest("请填写阿里云 OSS Endpoint")
}
if _, err := ValidateOutboundURL(next.Endpoint); err != nil {
return next, err
if next.Provider == s3Provider && next.Region == "" {
return next, BadAuthRequest("请填写 S3 Region")
}
var endpointErr error
if next.Provider == s3Provider {
_, endpointErr = validateStorageEndpoint(next.Endpoint)
} else {
_, endpointErr = ValidateOutboundURL(next.Endpoint)
}
if endpointErr != nil {
return next, endpointErr
}
if next.CDNBaseURL != "" {
if _, err := ossCDNBaseURL(next.CDNBaseURL); err != nil {
@@ -551,15 +670,24 @@ func normalizeOSSSetting(value ossSettingValue) ossSettingValue {
value.AccessKeySecret = strings.TrimSpace(value.AccessKeySecret)
value.PublicBaseURL = strings.TrimRight(strings.TrimSpace(value.PublicBaseURL), "/")
value.PathPrefix = strings.Trim(strings.TrimSpace(value.PathPrefix), "/")
if value.PathPrefix == "" {
value.PathPrefix = defaultOSSPathPrefix
}
value.S3Preset = strings.ToLower(strings.TrimSpace(value.S3Preset))
if value.S3Preset == "" {
value.S3Preset = "custom"
}
value.SessionToken = strings.TrimSpace(value.SessionToken)
value.StorageLocationID = strings.TrimSpace(value.StorageLocationID)
value.ArchivedCredentials = cloneOSSProviderCredentials(value.ArchivedCredentials)
return value
}
func defaultOSSSetting() ossSettingValue {
return ossSettingValue{Provider: aliyunOSSProvider}
return ossSettingValue{Provider: aliyunOSSProvider, PathPrefix: defaultOSSPathPrefix}
}
func publicOSSSetting(setting *model.SystemSetting, value ossSettingValue) PublicOSSSetting {
func (s *Service) publicOSSSetting(setting *model.SystemSetting, value ossSettingValue, scope string, ownerID string) (PublicOSSSetting, error) {
result := PublicOSSSetting{
Enabled: value.Enabled,
Provider: value.Provider,
@@ -571,16 +699,24 @@ func publicOSSSetting(setting *model.SystemSetting, value ossSettingValue) Publi
HasAccessKeySecret: strings.TrimSpace(value.AccessKeySecret) != "",
PublicBaseURL: value.PublicBaseURL,
PathPrefix: value.PathPrefix,
S3Preset: value.S3Preset,
PathStyle: value.PathStyle,
HasSessionToken: value.SessionToken != "",
StorageLocationID: value.StorageLocationID,
AllowUserS3: value.AllowUserS3,
}
if setting != nil {
result.UpdatedBy = setting.UpdatedBy
result.CreatedAt = setting.CreatedAt
result.UpdatedAt = setting.UpdatedAt
}
return result
if err := s.populateStorageLocationPublic(&result, scope, ownerID, value.StorageLocationID); err != nil {
return PublicOSSSetting{}, err
}
return result, nil
}
func publicUserOSSSetting(setting *model.UserOSSSetting, value ossSettingValue) PublicOSSSetting {
func (s *Service) publicUserOSSSetting(setting *model.UserOSSSetting, value ossSettingValue, ownerID string, allowUserS3 bool) (PublicOSSSetting, error) {
result := PublicOSSSetting{
Enabled: value.Enabled,
Provider: value.Provider,
@@ -592,11 +728,35 @@ func publicUserOSSSetting(setting *model.UserOSSSetting, value ossSettingValue)
HasAccessKeySecret: strings.TrimSpace(value.AccessKeySecret) != "",
PublicBaseURL: value.PublicBaseURL,
PathPrefix: value.PathPrefix,
S3Preset: value.S3Preset,
PathStyle: value.PathStyle,
HasSessionToken: value.SessionToken != "",
StorageLocationID: value.StorageLocationID,
AllowUserS3: allowUserS3,
}
if value.Provider == s3Provider && !allowUserS3 {
result.Enabled = false
}
if setting != nil {
result.UpdatedBy = setting.UserID
result.CreatedAt = setting.CreatedAt
result.UpdatedAt = setting.UpdatedAt
}
return result
if err := s.populateStorageLocationPublic(&result, "user", ownerID, value.StorageLocationID); err != nil {
return PublicOSSSetting{}, err
}
return result, nil
}
func storageLocationDigest(value ossSettingValue) string {
value = normalizeOSSSetting(value)
payload := strings.Join([]string{value.Provider, value.Endpoint, value.Bucket, value.Region, strconv.FormatBool(value.PathStyle), value.PathPrefix}, "\x00")
sum := sha256.Sum256([]byte(payload))
return hex.EncodeToString(sum[:])
}
func storageTestDigest(value ossSettingValue) string {
payload := strings.Join([]string{storageLocationDigest(value), value.AccessKeyID, value.AccessKeySecret, value.SessionToken}, "\x00")
sum := sha256.Sum256([]byte(payload))
return hex.EncodeToString(sum[:])
}
@@ -0,0 +1,201 @@
package service
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"path"
"strings"
"time"
"infinite-canvas/backend/internal/model"
"gorm.io/gorm"
)
type OSSConnectionTestResult struct {
OK bool `json:"ok"`
Message string `json:"message,omitempty"`
TestedAt time.Time `json:"testedAt"`
TestedDigest string `json:"testedDigest"`
}
func (s *Service) TestAdminOSSSetting(actor *model.User, req OSSSettingRequest) (*OSSConnectionTestResult, error) {
if err := s.RequireAdmin(actor); err != nil {
return nil, err
}
_, current, err := s.readOSSSetting()
if err != nil {
return nil, err
}
return s.testOSSSetting("platform", "", actor.ID, req, current)
}
func (s *Service) TestUserOSSSetting(actor *model.User, req OSSSettingRequest) (*OSSConnectionTestResult, error) {
if actor == nil {
return nil, Unauthorized("请先登录")
}
_, platform, err := s.readOSSSetting()
if err != nil {
return nil, err
}
if strings.EqualFold(strings.TrimSpace(req.Provider), s3Provider) && !platform.AllowUserS3 {
return nil, Forbidden("平台管理员尚未允许个人 S3 兼容存储")
}
_, current, err := s.readUserOSSSetting(actor.ID)
if err != nil {
return nil, err
}
return s.testOSSSetting("user", actor.ID, actor.ID, req, current)
}
func (s *Service) testOSSSetting(scope string, ownerID string, actorID string, req OSSSettingRequest, current ossSettingValue) (*OSSConnectionTestResult, error) {
req.Enabled = true
value, err := ossSettingFromRequest(req, current)
if err != nil {
return nil, err
}
key := scope + ":" + actorID
s.storageTestMu.Lock()
if s.activeStorageTests == nil {
s.activeStorageTests = make(map[string]bool)
}
if s.activeStorageTests[key] {
s.storageTestMu.Unlock()
return nil, NewAppError(http.StatusConflict, "对象存储连接测试正在进行")
}
s.activeStorageTests[key] = true
s.storageTestMu.Unlock()
defer func() {
s.storageTestMu.Lock()
delete(s.activeStorageTests, key)
s.storageTestMu.Unlock()
}()
testKey := path.Join(value.PathPrefix, ".yingce-tests", scope, newID())
payload := []byte("yingce-storage-test")
if _, err := putOSSObject(value, testKey, "application/octet-stream", int64(len(payload)), bytes.NewReader(payload)); err != nil {
return nil, err
}
stream, err := getOSSObjectRange(value, testKey, "bytes=0-3")
var rangeErr error
if err != nil {
rangeErr = err
} else {
data, readErr := io.ReadAll(io.LimitReader(stream.body, 5))
closeErr := stream.body.Close()
if readErr != nil || closeErr != nil || string(data) != "ying" {
rangeErr = errors.New("对象存储 Range 读取测试失败")
}
}
if err := deleteOSSObject(value, testKey); err != nil {
return nil, fmt.Errorf("对象存储删除测试失败:%w", err)
}
if rangeErr != nil {
return nil, rangeErr
}
location, err := s.upsertStorageLocation(scope, ownerID, value)
if err != nil {
return nil, err
}
now := time.Now()
location.TestedDigest = storageTestDigest(value)
location.TestedAt = &now
if err := s.repo.SaveStorageLocation(location); err != nil {
return nil, err
}
return &OSSConnectionTestResult{OK: true, Message: "连接测试通过", TestedAt: now, TestedDigest: location.TestedDigest}, nil
}
func (s *Service) upsertStorageLocation(scope string, ownerID string, value ossSettingValue) (*model.StorageLocation, error) {
digest := storageLocationDigest(value)
location, err := s.repo.StorageLocationByDigest(scope, ownerID, value.Provider, digest)
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
stored, err := s.encryptOSSSettingSecrets(value)
if err != nil {
return nil, err
}
stored.Enabled = true
stored.StorageLocationID = ""
encoded, err := json.Marshal(stored)
if err != nil {
return nil, err
}
if location == nil {
location = &model.StorageLocation{ID: newID(), Scope: scope, OwnerID: ownerID, Provider: value.Provider, LocationDigest: digest, ValueJSON: string(encoded)}
if err := s.repo.CreateStorageLocation(location); err != nil {
return nil, err
}
return location, nil
}
location.ValueJSON = string(encoded)
return location, nil
}
func (s *Service) storageLocationValue(id string) (*model.StorageLocation, ossSettingValue, error) {
location, err := s.repo.StorageLocation(id)
if err != nil {
return nil, ossSettingValue{}, err
}
var value ossSettingValue
if err := json.Unmarshal([]byte(location.ValueJSON), &value); err != nil {
return nil, ossSettingValue{}, errors.New("对象存储位置配置格式无效")
}
if _, err := s.decryptOSSSettingSecrets(&value); err != nil {
return nil, ossSettingValue{}, err
}
value.StorageLocationID = location.ID
return location, normalizeOSSSetting(value), nil
}
func (s *Service) requireTestedS3Location(scope string, ownerID string, value ossSettingValue) (*model.StorageLocation, error) {
location, err := s.repo.StorageLocationByDigest(scope, ownerID, s3Provider, storageLocationDigest(value))
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, BadAuthRequest("S3 关键配置尚未通过连接测试")
}
return nil, err
}
if location.TestedAt == nil || location.TestedDigest != storageTestDigest(value) {
return nil, BadAuthRequest("S3 关键配置或凭据已变化,请重新连接测试")
}
return location, nil
}
func (s *Service) populateStorageLocationPublic(result *PublicOSSSetting, scope string, ownerID string, id string) error {
count, err := s.repo.StorageLocationHistoryCount(scope, ownerID)
if err != nil {
return err
}
result.HistoryCount = count
if id == "" {
return nil
}
location, err := s.repo.StorageLocation(id)
if err != nil {
return err
}
result.TestedAt = location.TestedAt
result.TestedDigest = location.TestedDigest
result.ReferencedResourceCount, err = s.repo.StorageLocationResourceCount(id)
return err
}
func deleteOSSObject(setting ossSettingValue, objectKey string) error {
switch normalizeOSSSetting(setting).Provider {
case s3Provider:
return deleteS3Object(setting, objectKey)
case tencentCOSProvider:
return deleteTencentCOSObject(setting, objectKey)
case qiniuKodoProvider:
return deleteQiniuObject(setting, objectKey)
default:
return deleteAliyunOSSObject(setting, objectKey)
}
}
+165
View File
@@ -0,0 +1,165 @@
package service
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/aws/aws-sdk-go/aws/session"
awss3 "github.com/aws/aws-sdk-go/service/s3"
)
func validateStorageEndpoint(raw string) (*url.URL, error) {
parsed, err := ValidateOutboundURL(raw)
if err != nil {
return nil, err
}
if parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" || strings.Trim(parsed.Path, "/") != "" {
return nil, BadAuthRequest("对象存储 Endpoint 必须是服务根 URL,不能包含认证信息、路径、查询参数或片段")
}
if parsed.Scheme == "http" && !allowedPrivateUpstreamHost(parsed.Hostname()) {
return nil, BadAuthRequest("对象存储 HTTP Endpoint 仅允许访问 CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS 精确放行的主机")
}
return parsed, nil
}
func standardAWSS3Endpoint(endpoint string) bool {
parsed, err := url.Parse(endpoint)
if err != nil {
return false
}
host := strings.ToLower(parsed.Hostname())
return host == "s3.amazonaws.com" || strings.HasPrefix(host, "s3.") && (strings.HasSuffix(host, ".amazonaws.com") || strings.HasSuffix(host, ".amazonaws.com.cn")) || strings.HasPrefix(host, "s3-") && strings.HasSuffix(host, ".amazonaws.com")
}
func publicHTTPSStorageEndpoint(endpoint string) bool {
parsed, err := url.Parse(strings.TrimSpace(endpoint))
if err != nil || parsed.Scheme != "https" || parsed.Hostname() == "" {
return false
}
return validateOutboundHost(parsed.Hostname()) == nil && !allowedPrivateUpstreamHost(parsed.Hostname())
}
func newS3Client(setting ossSettingValue, timeout time.Duration) (*awss3.S3, error) {
setting = normalizeOSSSetting(setting)
endpoint, err := validateStorageEndpoint(setting.Endpoint)
if err != nil {
return nil, err
}
if setting.Region == "" || setting.Bucket == "" || setting.AccessKeyID == "" || setting.AccessKeySecret == "" {
return nil, errors.New("S3 Region、Bucket 或访问密钥不完整")
}
httpClient := OutboundHTTPClient(timeout)
config := aws.NewConfig().
WithRegion(setting.Region).
WithEndpoint(endpoint.String()).
WithCredentials(credentials.NewStaticCredentials(setting.AccessKeyID, setting.AccessKeySecret, setting.SessionToken)).
WithHTTPClient(httpClient).
WithS3ForcePathStyle(setting.PathStyle || !standardAWSS3Endpoint(endpoint.String())).
WithDisableSSL(endpoint.Scheme == "http")
sess, err := session.NewSession(config)
if err != nil {
return nil, err
}
return awss3.New(sess), nil
}
func putS3Object(setting ossSettingValue, objectKey string, mimeType string, size int64, body io.Reader) (string, error) {
client, err := newS3Client(setting, 2*time.Minute)
if err != nil {
return "", err
}
var seekable io.ReadSeeker
if reader, ok := body.(io.ReadSeeker); ok {
seekable = reader
} else {
limit := size + 1
if limit <= 0 {
limit = 64 << 20
}
data, readErr := io.ReadAll(io.LimitReader(body, limit))
if readErr != nil {
return "", readErr
}
if size >= 0 && int64(len(data)) != size {
return "", errors.New("S3 上传内容长度与资源记录不一致")
}
seekable = bytes.NewReader(data)
}
input := &awss3.PutObjectInput{Bucket: aws.String(setting.Bucket), Key: aws.String(strings.TrimLeft(objectKey, "/")), Body: seekable}
if mimeType != "" {
input.ContentType = aws.String(mimeType)
}
if size >= 0 {
input.ContentLength = aws.Int64(size)
}
output, err := client.PutObjectWithContext(context.Background(), input)
if err != nil {
return "", fmt.Errorf("S3 上传失败:%w", err)
}
return strings.Trim(aws.StringValue(output.ETag), `"`), nil
}
func getS3ObjectRange(setting ossSettingValue, objectKey string, rangeHeader string) (*ossObjectStream, error) {
client, err := newS3Client(setting, 2*time.Minute)
if err != nil {
return nil, err
}
input := &awss3.GetObjectInput{Bucket: aws.String(setting.Bucket), Key: aws.String(strings.TrimLeft(objectKey, "/"))}
if rangeHeader != "" {
input.Range = aws.String(rangeHeader)
}
output, err := client.GetObjectWithContext(context.Background(), input)
if err != nil {
if requestFailure, ok := err.(awserr.RequestFailure); ok && requestFailure.StatusCode() == http.StatusRequestedRangeNotSatisfiable {
return &ossObjectStream{body: io.NopCloser(bytes.NewReader(nil)), statusCode: http.StatusRequestedRangeNotSatisfiable, acceptRanges: "bytes"}, nil
}
return nil, fmt.Errorf("S3 读取失败:%w", err)
}
status := http.StatusOK
if rangeHeader != "" && aws.StringValue(output.ContentRange) != "" {
status = http.StatusPartialContent
}
return &ossObjectStream{body: output.Body, statusCode: status, contentLength: aws.Int64Value(output.ContentLength), contentRange: aws.StringValue(output.ContentRange), acceptRanges: firstNonEmpty(aws.StringValue(output.AcceptRanges), "bytes")}, nil
}
func signedS3ObjectURL(setting ossSettingValue, objectKey string, expiresAt time.Time) (string, error) {
client, err := newS3Client(setting, 2*time.Minute)
if err != nil {
return "", err
}
duration := time.Until(expiresAt)
if duration <= 0 {
return "", errors.New("S3 签名有效期必须晚于当前时间")
}
req, _ := client.GetObjectRequest(&awss3.GetObjectInput{Bucket: aws.String(setting.Bucket), Key: aws.String(strings.TrimLeft(objectKey, "/"))})
value, err := req.Presign(duration)
if err != nil {
return "", fmt.Errorf("S3 下载地址签名失败:%w", err)
}
return value, nil
}
func deleteS3Object(setting ossSettingValue, objectKey string) error {
client, err := newS3Client(setting, 2*time.Minute)
if err != nil {
return err
}
_, err = client.DeleteObjectWithContext(context.Background(), &awss3.DeleteObjectInput{Bucket: aws.String(setting.Bucket), Key: aws.String(strings.TrimLeft(objectKey, "/"))})
if requestFailure, ok := err.(awserr.RequestFailure); ok && requestFailure.StatusCode() == http.StatusNotFound {
return nil
}
if err != nil {
return fmt.Errorf("删除 S3 对象失败:%w", err)
}
return nil
}
+178
View File
@@ -0,0 +1,178 @@
package service
import (
"bytes"
"io"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
"time"
)
func TestS3ObjectOperationsUsePathStyleSessionTokenAndNoManagedHeaders(t *testing.T) {
t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "127.0.0.1")
var methods []string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
methods = append(methods, r.Method)
if r.URL.Path != "/bucket/prefix/object.txt" {
t.Errorf("path = %q", r.URL.Path)
}
if r.Header.Get("X-Amz-Security-Token") != "session-token" {
t.Errorf("X-Amz-Security-Token = %q", r.Header.Get("X-Amz-Security-Token"))
}
if r.Header.Get("X-Amz-Acl") != "" || r.Header.Get("X-Amz-Server-Side-Encryption") != "" {
t.Errorf("unexpected ACL/SSE headers: %#v", r.Header)
}
switch r.Method {
case http.MethodPut:
data, _ := io.ReadAll(r.Body)
if string(data) != "payload" {
t.Errorf("body = %q", data)
}
w.Header().Set("ETag", `"etag-value"`)
case http.MethodGet:
if r.Header.Get("Range") != "bytes=0-3" {
t.Errorf("Range = %q", r.Header.Get("Range"))
}
w.Header().Set("Content-Range", "bytes 0-3/7")
w.Header().Set("Accept-Ranges", "bytes")
w.WriteHeader(http.StatusPartialContent)
_, _ = io.WriteString(w, "payl")
case http.MethodDelete:
w.Header().Set("Content-Type", "application/xml")
w.WriteHeader(http.StatusNotFound)
_, _ = io.WriteString(w, `<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>`)
}
}))
defer server.Close()
setting := ossSettingValue{Provider: s3Provider, Region: "us-east-1", Endpoint: server.URL, Bucket: "bucket", AccessKeyID: "access-id", AccessKeySecret: "secret-value", SessionToken: "session-token"}
etag, err := putS3Object(setting, "prefix/object.txt", "text/plain", 7, bytes.NewReader([]byte("payload")))
if err != nil || etag != "etag-value" {
t.Fatalf("putS3Object() = %q, %v", etag, err)
}
stream, err := getS3ObjectRange(setting, "prefix/object.txt", "bytes=0-3")
if err != nil {
t.Fatal(err)
}
data, _ := io.ReadAll(stream.body)
_ = stream.body.Close()
if stream.statusCode != http.StatusPartialContent || stream.contentRange != "bytes 0-3/7" || string(data) != "payl" {
t.Fatalf("stream = %#v, body = %q", stream, data)
}
if err := deleteS3Object(setting, "prefix/object.txt"); err != nil {
t.Fatalf("deleteS3Object(404) error = %v", err)
}
if strings.Join(methods, ",") != "PUT,GET,DELETE" {
t.Fatalf("methods = %v", methods)
}
}
func TestSignedS3ObjectURLUsesSDKPresignAndSessionToken(t *testing.T) {
t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "127.0.0.1")
server := httptest.NewServer(http.NotFoundHandler())
defer server.Close()
value, err := signedS3ObjectURL(ossSettingValue{
Provider: s3Provider, Region: "us-east-1", Endpoint: server.URL, Bucket: "bucket",
AccessKeyID: "access-id", AccessKeySecret: "secret-value", SessionToken: "session-token",
}, "folder/object.png", time.Now().Add(time.Minute))
if err != nil {
t.Fatal(err)
}
parsed, err := url.Parse(value)
if err != nil {
t.Fatal(err)
}
if parsed.Path != "/bucket/folder/object.png" || parsed.Query().Get("X-Amz-Security-Token") != "session-token" || parsed.Query().Get("X-Amz-Signature") == "" {
t.Fatalf("presigned URL = %q", value)
}
if strings.Contains(value, "secret-value") {
t.Fatal("presigned URL leaked secret key")
}
}
func TestS3EndpointAndDigestsEnforceStorageContract(t *testing.T) {
t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "storage.internal")
if _, err := validateStorageEndpoint("https://example.com/path"); err == nil {
t.Fatal("validateStorageEndpoint() accepted a path")
}
if _, err := validateStorageEndpoint("http://example.com"); err == nil {
t.Fatal("validateStorageEndpoint() accepted public HTTP")
}
base := ossSettingValue{Provider: s3Provider, Region: "us-east-1", Endpoint: "https://s3.example.com", Bucket: "bucket", PathPrefix: "assets", AccessKeyID: "id", AccessKeySecret: "secret"}
rotated := base
rotated.AccessKeySecret = "rotated"
moved := base
moved.PathPrefix = "other"
if storageLocationDigest(base) != storageLocationDigest(rotated) {
t.Fatal("credential rotation changed the location digest")
}
if storageTestDigest(base) == storageTestDigest(rotated) {
t.Fatal("credential rotation did not invalidate the tested digest")
}
if storageLocationDigest(base) == storageLocationDigest(moved) {
t.Fatal("location change did not create a different digest")
}
}
func TestOSSSettingKeepsS3SecretsWhenLocationChanges(t *testing.T) {
current := ossSettingValue{
Provider: s3Provider,
Region: "us-east-1",
Endpoint: "https://s3.example.com",
Bucket: "old-bucket",
AccessKeyID: "access-id",
AccessKeySecret: "secret-value",
SessionToken: "session-token",
PathPrefix: defaultOSSPathPrefix,
}
next, err := ossSettingFromRequest(OSSSettingRequest{
Enabled: true,
Provider: s3Provider,
S3Preset: "custom",
Region: "us-east-1",
Endpoint: "https://s3.example.com",
Bucket: "new-bucket",
AccessKeyID: "access-id",
PathPrefix: "changed-prefix",
}, current)
if err != nil {
t.Fatal(err)
}
if next.AccessKeySecret != current.AccessKeySecret || next.SessionToken != current.SessionToken {
t.Fatalf("secrets were not retained: %#v", next)
}
}
func TestDefaultOSSPathPrefix(t *testing.T) {
value := normalizeOSSSetting(ossSettingValue{})
if value.PathPrefix != defaultOSSPathPrefix || defaultOSSSetting().PathPrefix != defaultOSSPathPrefix {
t.Fatalf("default path prefix = %q, default setting = %q", value.PathPrefix, defaultOSSSetting().PathPrefix)
}
}
func TestResourceObjectKeyPreservesOrInfersExtension(t *testing.T) {
setting := ossSettingValue{PathPrefix: "assets"}
now := time.Date(2026, time.August, 27, 20, 31, 0, 0, time.UTC)
for _, test := range []struct {
name string
fileName string
mimeType string
kind string
ext string
}{
{name: "original filename", fileName: "photo.JPEG", mimeType: "image/jpeg", kind: "image", ext: ".jpeg"},
{name: "mime type", mimeType: "image/png", kind: "image", ext: ".png"},
{name: "kind fallback", kind: "video", ext: ".mp4"},
} {
t.Run(test.name, func(t *testing.T) {
objectKey := ossObjectKey(setting, "user-1", test.kind, test.fileName, test.mimeType, now)
if !strings.HasSuffix(objectKey, test.ext) {
t.Fatalf("object key = %q, want suffix %q", objectKey, test.ext)
}
})
}
}