chore(*): 代码同步 - 合并远端 main 最新变更
This commit is contained in:
@@ -2,10 +2,12 @@ package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"path"
|
||||
"strings"
|
||||
@@ -13,6 +15,7 @@ import (
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
|
||||
"github.com/tencentyun/cos-go-sdk-v5"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
@@ -76,27 +79,9 @@ func (s *Service) testOSSSetting(scope string, ownerID string, actorID string, r
|
||||
}()
|
||||
|
||||
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 {
|
||||
if err := verifyOSSConnection(value, testKey); 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 {
|
||||
@@ -111,6 +96,79 @@ func (s *Service) testOSSSetting(scope string, ownerID string, actorID string, r
|
||||
return &OSSConnectionTestResult{OK: true, Message: "连接测试通过", TestedAt: now, TestedDigest: location.TestedDigest}, nil
|
||||
}
|
||||
|
||||
func verifyOSSConnection(value ossSettingValue, testKey string) error {
|
||||
payload := []byte("yingce-storage-test")
|
||||
// 连接测试验证服务端到对象存储的真实读写权限。CDN 是浏览器读取出口,
|
||||
// 可能存在回源鉴权或边缘同步延迟,不能参与刚写入对象的最小读写测试。
|
||||
testValue := value
|
||||
testValue.CDNBaseURL = ""
|
||||
if _, err := putOSSObject(testValue, testKey, "application/octet-stream", int64(len(payload)), bytes.NewReader(payload)); err != nil {
|
||||
return storageConnectionTestError(testValue.Provider, "写入", err)
|
||||
}
|
||||
stream, err := getOSSObjectRange(testValue, 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(testValue, testKey); err != nil {
|
||||
return storageConnectionTestError(testValue.Provider, "删除", err)
|
||||
}
|
||||
if rangeErr != nil {
|
||||
return storageConnectionTestError(testValue.Provider, "Range 读取", rangeErr)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func storageConnectionTestError(provider string, operation string, cause error) error {
|
||||
provider = strings.ToLower(strings.TrimSpace(provider))
|
||||
providerName := "对象存储"
|
||||
switch provider {
|
||||
case aliyunOSSProvider:
|
||||
providerName = "阿里云 OSS"
|
||||
case tencentCOSProvider:
|
||||
providerName = "腾讯云 COS"
|
||||
case qiniuKodoProvider:
|
||||
providerName = "七牛云 Kodo"
|
||||
case s3Provider:
|
||||
providerName = "S3 兼容存储"
|
||||
}
|
||||
|
||||
if errors.Is(cause, context.DeadlineExceeded) {
|
||||
return WrapAppError(http.StatusGatewayTimeout, fmt.Sprintf("%s %s测试超时,请检查 Endpoint 和服务端网络", providerName, operation), cause)
|
||||
}
|
||||
var networkErr net.Error
|
||||
if errors.As(cause, &networkErr) && networkErr.Timeout() {
|
||||
return WrapAppError(http.StatusGatewayTimeout, fmt.Sprintf("%s %s测试超时,请检查 Endpoint 和服务端网络", providerName, operation), cause)
|
||||
}
|
||||
|
||||
if provider == tencentCOSProvider {
|
||||
var responseErr *cos.ErrorResponse
|
||||
if errors.As(cause, &responseErr) && responseErr.Response != nil {
|
||||
switch responseErr.Response.StatusCode {
|
||||
case http.StatusBadRequest, http.StatusUnprocessableEntity:
|
||||
return WrapAppError(http.StatusBadGateway, fmt.Sprintf("腾讯云 COS %s请求被拒绝,请检查 Bucket、Region 和 Endpoint", operation), cause)
|
||||
case http.StatusUnauthorized, http.StatusForbidden:
|
||||
return WrapAppError(http.StatusBadGateway, fmt.Sprintf("腾讯云 COS %s鉴权失败,请检查 SecretId、SecretKey 和 Bucket 权限", operation), cause)
|
||||
case http.StatusNotFound:
|
||||
return WrapAppError(http.StatusBadGateway, fmt.Sprintf("腾讯云 COS %s未找到 Bucket,请检查 Bucket、Region 和 Endpoint", operation), cause)
|
||||
case http.StatusTooManyRequests:
|
||||
return WrapAppError(http.StatusBadGateway, fmt.Sprintf("腾讯云 COS %s请求过于频繁,请稍后重试", operation), cause)
|
||||
}
|
||||
if responseErr.Response.StatusCode >= http.StatusInternalServerError {
|
||||
return WrapAppError(http.StatusBadGateway, fmt.Sprintf("腾讯云 COS %s服务暂时不可用,请稍后重试", operation), cause)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return WrapAppError(http.StatusBadGateway, fmt.Sprintf("%s %s测试失败,请检查 Endpoint、Bucket、访问密钥和存储桶权限", providerName, operation), cause)
|
||||
}
|
||||
|
||||
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)
|
||||
|
||||
111
backend/internal/service/storage_location_test.go
Normal file
111
backend/internal/service/storage_location_test.go
Normal file
@@ -0,0 +1,111 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"hash/crc64"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestTencentCOSConnectionTestUsesStorageEndpointInsteadOfCDN(t *testing.T) {
|
||||
t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true")
|
||||
var storageMethods []string
|
||||
storageServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
storageMethods = append(storageMethods, r.Method)
|
||||
if !strings.HasPrefix(r.URL.Path, "/open-ai-canvas/.yingce-tests/platform/") {
|
||||
t.Errorf("path = %q", r.URL.Path)
|
||||
}
|
||||
if !strings.Contains(r.Header.Get("Authorization"), "q-sign-algorithm=sha1") {
|
||||
t.Errorf("Authorization = %q", r.Header.Get("Authorization"))
|
||||
}
|
||||
switch r.Method {
|
||||
case http.MethodPut:
|
||||
payload, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
if string(payload) != "yingce-storage-test" {
|
||||
t.Errorf("payload = %q", payload)
|
||||
}
|
||||
w.Header().Set("ETag", `"cos-test-etag"`)
|
||||
w.Header().Set("x-cos-hash-crc64ecma", strconv.FormatUint(crc64.Checksum(payload, crc64.MakeTable(crc64.ECMA)), 10))
|
||||
case http.MethodGet:
|
||||
if r.Header.Get("Range") != "bytes=0-3" {
|
||||
t.Errorf("Range = %q", r.Header.Get("Range"))
|
||||
}
|
||||
w.Header().Set("Accept-Ranges", "bytes")
|
||||
w.Header().Set("Content-Range", "bytes 0-3/19")
|
||||
w.WriteHeader(http.StatusPartialContent)
|
||||
_, _ = io.WriteString(w, "ying")
|
||||
case http.MethodDelete:
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
default:
|
||||
t.Errorf("unexpected method %s", r.Method)
|
||||
}
|
||||
}))
|
||||
defer storageServer.Close()
|
||||
|
||||
cdnRequests := 0
|
||||
cdnServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
cdnRequests++
|
||||
http.Error(w, "CDN must not be used by storage connection tests", http.StatusBadGateway)
|
||||
}))
|
||||
defer cdnServer.Close()
|
||||
|
||||
err := verifyOSSConnection(ossSettingValue{
|
||||
Provider: tencentCOSProvider,
|
||||
Region: "ap-guangzhou",
|
||||
Endpoint: storageServer.URL,
|
||||
CDNBaseURL: cdnServer.URL,
|
||||
Bucket: "example-1250000000",
|
||||
AccessKeyID: "secret-id",
|
||||
AccessKeySecret: "secret-key",
|
||||
PathPrefix: defaultOSSPathPrefix,
|
||||
}, defaultOSSPathPrefix+"/.yingce-tests/platform/test-id")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if strings.Join(storageMethods, ",") != "PUT,GET,DELETE" {
|
||||
t.Fatalf("storage methods = %v", storageMethods)
|
||||
}
|
||||
if cdnRequests != 0 {
|
||||
t.Fatalf("CDN requests = %d", cdnRequests)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTencentCOSConnectionTestReturnsActionableAuthError(t *testing.T) {
|
||||
t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true")
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/xml")
|
||||
w.WriteHeader(http.StatusForbidden)
|
||||
_, _ = io.WriteString(w, `<Error><Code>SignatureDoesNotMatch</Code><Message>signature mismatch</Message><RequestId>request-id</RequestId></Error>`)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
err := verifyOSSConnection(ossSettingValue{
|
||||
Provider: tencentCOSProvider,
|
||||
Region: "ap-guangzhou",
|
||||
Endpoint: server.URL,
|
||||
Bucket: "example-1250000000",
|
||||
AccessKeyID: "secret-id",
|
||||
AccessKeySecret: "secret-key",
|
||||
PathPrefix: defaultOSSPathPrefix,
|
||||
}, defaultOSSPathPrefix+"/.yingce-tests/platform/test-id")
|
||||
if err == nil {
|
||||
t.Fatal("verifyOSSConnection() error = nil")
|
||||
}
|
||||
var appErr *AppError
|
||||
if !errors.As(err, &appErr) {
|
||||
t.Fatalf("error type = %T, error = %v", err, err)
|
||||
}
|
||||
if appErr.Status != http.StatusBadGateway || !strings.Contains(appErr.Message, "腾讯云 COS 写入鉴权失败") {
|
||||
t.Fatalf("AppError = %#v", appErr)
|
||||
}
|
||||
if strings.Contains(appErr.Message, "secret-key") || strings.Contains(appErr.Message, "signature mismatch") {
|
||||
t.Fatalf("public error leaked upstream detail: %q", appErr.Message)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user