From f0de364063ef9169ef57dd343fab34c74c066dd7 Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 22 Jun 2026 23:49:24 +0800 Subject: [PATCH] fix: harden desktop release OSS uploads --- .../src/lib/desktop-client-releases.ts | 97 +++++++++++++------ .../ops/app/desktop_client_release.go | 48 ++++++++- .../ops/app/desktop_client_release_test.go | 55 ++++++++++- .../internal/shared/objectstorage/aliyun.go | 34 ++++--- .../internal/shared/objectstorage/client.go | 4 + server/internal/shared/objectstorage/minio.go | 34 ++++--- .../shared/objectstorage/reloadable.go | 9 ++ 7 files changed, 218 insertions(+), 63 deletions(-) diff --git a/apps/ops-web/src/lib/desktop-client-releases.ts b/apps/ops-web/src/lib/desktop-client-releases.ts index c32aae6..318743f 100644 --- a/apps/ops-web/src/lib/desktop-client-releases.ts +++ b/apps/ops-web/src/lib/desktop-client-releases.ts @@ -1,7 +1,7 @@ import axios from 'axios' -import { createMD5, createSHA256 } from 'hash-wasm' +import { createSHA256 } from 'hash-wasm' -import { http } from '@/lib/http' +import { OpsApiError, http } from '@/lib/http' export type DesktopReleasePlatform = 'darwin' | 'win32' | 'linux' export type DesktopReleaseArch = 'universal' | 'x64' | 'arm64' | 'ia32' @@ -126,7 +126,6 @@ interface DesktopClientReleaseDirectUploadSession { interface DesktopClientReleaseFileHashes { sha256: string - contentMD5: string } export const platformOptions = [ @@ -175,7 +174,20 @@ export const desktopClientReleaseApi = { file: File, target: DesktopClientReleaseUploadTarget, ) { - return uploadOSSDirect(file, target) + try { + return await uploadOSSDirect(file, target) + } catch (error) { + if (!shouldFallbackToChunkedUpload(error)) { + throw error + } + target.onProgress?.({ + loaded: 0, + total: file.size, + percent: 0, + stage: 'uploading', + }) + return uploadOSSInChunks(file, target) + } }, async uploadOSSLegacy( file: File, @@ -246,7 +258,6 @@ async function uploadOSSDirect( file_name: file.name, file_size_bytes: file.size, sha256: hashes.sha256, - content_md5: hashes.contentMD5, }, ) @@ -254,24 +265,42 @@ async function uploadOSSDirect( if (!session.upload_url) { throw new Error('missing_direct_upload_url') } - await axios.put(session.upload_url, file, { - headers: session.headers, - timeout: DESKTOP_CLIENT_RELEASE_UPLOAD_TIMEOUT_MS, - onUploadProgress(event) { - const total = event.total ?? file.size - target.onProgress?.({ - loaded: event.loaded, - total, - percent: total ? Math.min(99, Math.round((event.loaded / total) * 100)) : undefined, - stage: 'uploading', - }) - }, - }) + try { + await axios.put(session.upload_url, file, { + headers: directUploadHeaders(session.headers), + timeout: DESKTOP_CLIENT_RELEASE_UPLOAD_TIMEOUT_MS, + onUploadProgress(event) { + const total = event.total ?? file.size + target.onProgress?.({ + loaded: event.loaded, + total, + percent: total ? Math.min(99, Math.round((event.loaded / total) * 100)) : undefined, + stage: 'uploading', + }) + }, + }) + } catch (uploadError) { + try { + return await completeDirectUpload(session, target) + } catch (completeError) { + if (isDirectUploadNotFoundError(completeError)) { + throw uploadError + } + throw completeError + } + } } + return completeDirectUpload(session, target) +} + +async function completeDirectUpload( + session: DesktopClientReleaseDirectUploadSession, + target: DesktopClientReleaseUploadTarget, +): Promise { target.onProgress?.({ - loaded: file.size, - total: file.size, + loaded: session.file_size_bytes, + total: session.file_size_bytes, percent: 100, stage: 'verifying', }) @@ -298,34 +327,40 @@ async function uploadOSSDirect( return result } +function isDirectUploadNotFoundError(error: unknown): boolean { + return error instanceof OpsApiError && error.code === 40078 +} + +function directUploadHeaders(headers: Record | undefined): Record { + if (!headers) return {} + return Object.fromEntries( + Object.entries(headers).filter(([key]) => key.toLowerCase() !== 'content-md5'), + ) +} + +function shouldFallbackToChunkedUpload(error: unknown): boolean { + if (axios.isAxiosError(error)) return true + return error instanceof Error && error.message === 'missing_direct_upload_url' +} + async function hashFile( file: File, onProgress?: (loaded: number) => void, ): Promise { - const [sha256Hasher, md5Hasher] = await Promise.all([createSHA256(), createMD5()]) + const sha256Hasher = await createSHA256() let loaded = 0 while (loaded < file.size) { const end = Math.min(loaded + DESKTOP_CLIENT_RELEASE_HASH_CHUNK_SIZE_BYTES, file.size) const chunk = new Uint8Array(await file.slice(loaded, end).arrayBuffer()) sha256Hasher.update(chunk) - md5Hasher.update(chunk) loaded = end onProgress?.(loaded) } return { sha256: sha256Hasher.digest('hex') as string, - contentMD5: uint8ArrayToBase64(md5Hasher.digest('binary') as Uint8Array), } } -function uint8ArrayToBase64(value: Uint8Array): string { - let binary = '' - for (let index = 0; index < value.length; index += 1) { - binary += String.fromCharCode(value[index]) - } - return window.btoa(binary) -} - async function uploadOSSInChunks( file: File, target: DesktopClientReleaseUploadTarget, diff --git a/server/internal/ops/app/desktop_client_release.go b/server/internal/ops/app/desktop_client_release.go index daf6031..7dc0aa2 100644 --- a/server/internal/ops/app/desktop_client_release.go +++ b/server/internal/ops/app/desktop_client_release.go @@ -602,8 +602,7 @@ func (s *DesktopClientReleaseService) InitiateDirectUpload( return nil, err } } - contentMD5, err := normalizeDesktopReleaseContentMD5(in.ContentMD5) - if err != nil { + if _, err := normalizeDesktopReleaseContentMD5(in.ContentMD5); err != nil { return nil, err } @@ -635,7 +634,7 @@ func (s *DesktopClientReleaseService) InitiateDirectUpload( ctx, objectKey, detectDesktopReleaseContentType(fileName, nil), - contentMD5, + "", desktopClientReleaseDirectUploadTTL, ) if err != nil { @@ -724,6 +723,21 @@ func (s *DesktopClientReleaseService) CompleteDirectUpload( return nil, response.ErrBadRequest(40078, "direct_upload_not_found", "OSS 直传文件尚未完成") } } + actualSHA, actualSize, err := hashDesktopClientReleaseStorageObject(ctx, s.storage, objectKey) + if err != nil { + if errors.Is(err, objectstorage.ErrObjectNotFound) { + return nil, response.ErrBadRequest(40078, "direct_upload_not_found", "OSS 直传文件尚未完成") + } + appErr := response.ErrInternal(50092, "desktop_client_release_upload_failed", err.Error()) + appErr.Cause = err + return nil, appErr + } + if actualSize != in.FileSizeBytes { + return nil, response.ErrBadRequest(40079, "direct_upload_size_mismatch", "OSS 直传文件大小不匹配，请重新上传") + } + if actualSHA != sha256Hex { + return nil, response.ErrBadRequest(40070, "invalid_sha256", "OSS 直传文件 SHA256 不匹配，请重新上传") + } return &DesktopClientReleaseUploadResult{ OSSObjectKey: objectKey, @@ -733,6 +747,34 @@ func (s *DesktopClientReleaseService) CompleteDirectUpload( }, nil } +func hashDesktopClientReleaseStorageObject( + ctx context.Context, + storage objectstorage.Client, + objectKey string, +) (string, int64, error) { + if readerClient, ok := storage.(objectstorage.ObjectReaderClient); ok { + reader, err := readerClient.GetReader(ctx, objectKey) + if err != nil { + return "", 0, err + } + defer reader.Close() + + hasher := sha256.New() + size, err := io.Copy(hasher, reader) + if err != nil { + return "", 0, err + } + return hex.EncodeToString(hasher.Sum(nil)), size, nil + } + + content, err := storage.GetBytes(ctx, objectKey) + if err != nil { + return "", 0, err + } + actualSHA := sha256.Sum256(content) + return hex.EncodeToString(actualSHA[:]), int64(len(content)), nil +} + func MaxDesktopClientReleaseMultipartBytes() int64 { return MaxDesktopClientReleaseUploadBytes + desktopClientReleaseMultipartOverheadBytes } diff --git a/server/internal/ops/app/desktop_client_release_test.go b/server/internal/ops/app/desktop_client_release_test.go index c355355..e236f45 100644 --- a/server/internal/ops/app/desktop_client_release_test.go +++ b/server/internal/ops/app/desktop_client_release_test.go @@ -6,6 +6,7 @@ import ( "crypto/sha256" "encoding/hex" "errors" + "fmt" "io" "net/http" "os" @@ -343,12 +344,15 @@ func TestDesktopClientReleaseDirectUploadUsesPresignedPutURL(t *testing.T) { require.Equal(t, "desktop-client/releases/blobs/"+wantSHA+".dmg", initResult.OSSObjectKey) require.Equal(t, "https://oss.example.test/upload/"+initResult.OSSObjectKey, initResult.UploadURL) require.Equal(t, "application/x-apple-diskimage", initResult.Headers["Content-Type"]) - require.Equal(t, "pt74NOtqjIwBJtPUYvJ8AA==", initResult.Headers["Content-Md5"]) + require.NotContains(t, initResult.Headers, "Content-Md5") + require.NotContains(t, initResult.Headers, "Content-MD5") require.Len(t, storage.presignedPutCalls, 1) require.Equal(t, initResult.OSSObjectKey, storage.presignedPutCalls[0]) + require.Equal(t, []string{""}, storage.presignedPutMD5s) storage.existsByKey[initResult.OSSObjectKey] = true storage.objectSizes[initResult.OSSObjectKey] = int64(len(content)) + storage.objects[initResult.OSSObjectKey] = content completeResult, err := svc.CompleteDirectUpload(context.Background(), DesktopClientReleaseDirectUploadCompleteInput{ Platform: "darwin", Arch: "arm64", @@ -392,6 +396,34 @@ func TestDesktopClientReleaseDirectUploadRejectsSizeMismatch(t *testing.T) { require.Contains(t, err.Error(), "direct_upload_size_mismatch") } +func TestDesktopClientReleaseDirectUploadRejectsSHA256Mismatch(t *testing.T) { + t.Parallel() + + sum := sha256.Sum256([]byte("release-package")) + sha256Hex := hex.EncodeToString(sum[:]) + objectKey := buildDesktopClientReleaseObjectKey(sha256Hex, "client.zip") + storage := &desktopReleaseDownloadURLStorage{ + existsByKey: map[string]bool{objectKey: true}, + objects: map[string][]byte{objectKey: []byte("different-package")}, + objectSizes: map[string]int64{objectKey: int64(len("different-package"))}, + } + svc := NewDesktopClientReleaseService(nil, storage, nil) + + _, err := svc.CompleteDirectUpload(context.Background(), DesktopClientReleaseDirectUploadCompleteInput{ + Platform: "darwin", + Arch: "arm64", + Channel: "stable", + Version: "1.2.3", + Kind: DesktopClientAssetKindUpdater, + OSSObjectKey: objectKey, + FileName: "client.zip", + FileSizeBytes: int64(len("different-package")), + SHA256: sha256Hex, + }) + require.Error(t, err) + require.Contains(t, err.Error(), "invalid_sha256") +} + type desktopReleaseDownloadURLStorage struct { publicURLCalls []string downloadURLCalls []string @@ -399,6 +431,7 @@ type desktopReleaseDownloadURLStorage struct { objects map[string][]byte objectSizes map[string]int64 presignedPutCalls []string + presignedPutMD5s []string } func (s *desktopReleaseDownloadURLStorage) Validate() error { @@ -429,8 +462,23 @@ func (s *desktopReleaseDownloadURLStorage) PutReader(_ context.Context, objectKe return nil } -func (s *desktopReleaseDownloadURLStorage) GetBytes(context.Context, string) ([]byte, error) { - return nil, errors.New("not implemented") +func (s *desktopReleaseDownloadURLStorage) GetBytes(_ context.Context, objectKey string) ([]byte, error) { + if s.objects == nil { + return nil, fmt.Errorf("%w: %s", objectstorage.ErrObjectNotFound, objectKey) + } + content, ok := s.objects[objectKey] + if !ok { + return nil, fmt.Errorf("%w: %s", objectstorage.ErrObjectNotFound, objectKey) + } + return append([]byte(nil), content...), nil +} + +func (s *desktopReleaseDownloadURLStorage) GetReader(_ context.Context, objectKey string) (io.ReadCloser, error) { + content, err := s.GetBytes(context.Background(), objectKey) + if err != nil { + return nil, err + } + return io.NopCloser(bytes.NewReader(content)), nil } func (s *desktopReleaseDownloadURLStorage) Exists(_ context.Context, objectKey string) (bool, error) { @@ -448,6 +496,7 @@ func (s *desktopReleaseDownloadURLStorage) PresignedPutURL( ttl time.Duration, ) (*objectstorage.PresignedPutResult, error) { s.presignedPutCalls = append(s.presignedPutCalls, objectKey) + s.presignedPutMD5s = append(s.presignedPutMD5s, contentMD5Base64) headers := http.Header{} headers.Set("Content-Type", contentType) if contentMD5Base64 != "" { diff --git a/server/internal/shared/objectstorage/aliyun.go b/server/internal/shared/objectstorage/aliyun.go index 6456d17..25223ac 100644 --- a/server/internal/shared/objectstorage/aliyun.go +++ b/server/internal/shared/objectstorage/aliyun.go @@ -217,6 +217,26 @@ func aliyunStatusCode(err error) int { } func (c *aliyunClient) GetBytes(ctx context.Context, objectKey string) ([]byte, error) { + reader, err := c.GetReader(ctx, objectKey) + if err != nil { + return nil, err + } + defer reader.Close() + + data, err := io.ReadAll(reader) + if err != nil { + c.logError(ctx, "aliyun read object failed", + zap.String("bucket", c.bucket), + zap.String("object_key", objectKey), + zap.Error(err), + ) + return nil, fmt.Errorf("read aliyun object: %w", err) + } + + return data, nil +} + +func (c *aliyunClient) GetReader(ctx context.Context, objectKey string) (io.ReadCloser, error) { if err := c.Validate(); err != nil { return nil, err } @@ -243,19 +263,7 @@ func (c *aliyunClient) GetBytes(ctx context.Context, objectKey string) ([]byte, } return nil, fmt.Errorf("get aliyun object: %w", err) } - defer reader.Close() - - data, err := io.ReadAll(reader) - if err != nil { - c.logError(ctx, "aliyun read object failed", - zap.String("bucket", c.bucket), - zap.String("object_key", objectKey), - zap.Error(err), - ) - return nil, fmt.Errorf("read aliyun object: %w", err) - } - - return data, nil + return reader, nil } func isAliyunObjectNotFound(err error) bool { diff --git a/server/internal/shared/objectstorage/client.go b/server/internal/shared/objectstorage/client.go index eb48b8d..d9397ae 100644 --- a/server/internal/shared/objectstorage/client.go +++ b/server/internal/shared/objectstorage/client.go @@ -33,6 +33,10 @@ type ReaderClient interface { PutReader(ctx context.Context, objectKey string, reader io.Reader, size int64, contentType string) error } +type ObjectReaderClient interface { + GetReader(ctx context.Context, objectKey string) (io.ReadCloser, error) +} + type PresignedPutResult struct { URL string Headers http.Header diff --git a/server/internal/shared/objectstorage/minio.go b/server/internal/shared/objectstorage/minio.go index 2e5a5be..8a9f3d4 100644 --- a/server/internal/shared/objectstorage/minio.go +++ b/server/internal/shared/objectstorage/minio.go @@ -161,20 +161,9 @@ func (c *minioClient) PutReader( } func (c *minioClient) GetBytes(ctx context.Context, objectKey string) ([]byte, error) { - if err := c.Validate(); err != nil { - return nil, err - } - reader, err := c.client.GetObject(ctx, c.bucket, objectKey, minio.GetObjectOptions{}) + reader, err := c.GetReader(ctx, objectKey) if err != nil { - c.logError(ctx, "minio get object failed", - zap.String("bucket", c.bucket), - zap.String("object_key", objectKey), - zap.Error(err), - ) - if isMinIOObjectNotFound(err) { - return nil, fmt.Errorf("%w: %s", ErrObjectNotFound, objectKey) - } - return nil, fmt.Errorf("get minio object: %w", err) + return nil, err } defer reader.Close() @@ -193,6 +182,25 @@ func (c *minioClient) GetBytes(ctx context.Context, objectKey string) ([]byte, e return data, nil } +func (c *minioClient) GetReader(ctx context.Context, objectKey string) (io.ReadCloser, error) { + if err := c.Validate(); err != nil { + return nil, err + } + reader, err := c.client.GetObject(ctx, c.bucket, objectKey, minio.GetObjectOptions{}) + if err != nil { + c.logError(ctx, "minio get object failed", + zap.String("bucket", c.bucket), + zap.String("object_key", objectKey), + zap.Error(err), + ) + if isMinIOObjectNotFound(err) { + return nil, fmt.Errorf("%w: %s", ErrObjectNotFound, objectKey) + } + return nil, fmt.Errorf("get minio object: %w", err) + } + return reader, nil +} + func isMinIOObjectNotFound(err error) bool { responseErr := minio.ToErrorResponse(err) return responseErr.StatusCode == 404 || diff --git a/server/internal/shared/objectstorage/reloadable.go b/server/internal/shared/objectstorage/reloadable.go index 3c1143b..59c640b 100644 --- a/server/internal/shared/objectstorage/reloadable.go +++ b/server/internal/shared/objectstorage/reloadable.go @@ -49,6 +49,15 @@ func (c *ReloadableClient) GetBytes(ctx context.Context, objectKey string) ([]by return c.snapshot().GetBytes(ctx, objectKey) } +func (c *ReloadableClient) GetReader(ctx context.Context, objectKey string) (io.ReadCloser, error) { + client := c.snapshot() + readerClient, ok := client.(ObjectReaderClient) + if !ok { + return nil, ErrNotConfigured + } + return readerClient.GetReader(ctx, objectKey) +} + func (c *ReloadableClient) Exists(ctx context.Context, objectKey string) (bool, error) { return c.snapshot().Exists(ctx, objectKey) }