diff --git a/server/media/oss/oss.go b/server/media/oss/oss.go index c1511a9..56425b1 100644 --- a/server/media/oss/oss.go +++ b/server/media/oss/oss.go @@ -58,12 +58,80 @@ type readerCounter struct { reader io.Reader } +type ossReadSeekCloser struct { + open func(offset int64) (io.ReadCloser, error) + size int64 + pos int64 + body io.ReadCloser + closed bool +} + func (rc *readerCounter) Read(buf []byte) (int, error) { n, err := rc.reader.Read(buf) atomic.AddInt64(&rc.count, int64(n)) return n, err } +func (r *ossReadSeekCloser) Read(buf []byte) (int, error) { + if r.closed { + return 0, errors.New("read from closed OSS object") + } + if r.pos >= r.size { + return 0, io.EOF + } + if r.body == nil { + var err error + r.body, err = r.open(r.pos) + if err != nil { + return 0, err + } + } + n, err := r.body.Read(buf) + r.pos += int64(n) + if err != nil { + _ = r.body.Close() + r.body = nil + } + return n, err +} + +func (r *ossReadSeekCloser) Seek(offset int64, whence int) (int64, error) { + var next int64 + switch whence { + case io.SeekStart: + next = offset + case io.SeekCurrent: + next = r.pos + offset + case io.SeekEnd: + next = r.size + offset + default: + return 0, errors.New("invalid seek origin") + } + if next < 0 { + return 0, errors.New("negative seek position") + } + if next == r.pos { + return next, nil + } + if r.body != nil { + _ = r.body.Close() + r.body = nil + } + r.pos = next + return next, nil +} + +func (r *ossReadSeekCloser) Close() error { + if r.closed { + return nil + } + r.closed = true + if r.body != nil { + return r.body.Close() + } + return nil +} + // Init initializes the media handler. func (oh *ossHandler) Init(jsconf string) error { var err error @@ -207,7 +275,9 @@ func truthy(value string) bool { } } -// Headers adds CORS headers and redirects GET and HEAD requests to OSS. +// Headers adds CORS and cache headers. Downloads stay on the IM origin: forwarding the authenticated +// browser request through a signed OSS redirect also forwards custom headers, which forces an OSS +// preflight and fails unless the bucket exposes those private headers through CORS. func (oh *ossHandler) Headers(tenantID types.TenantID, method string, url *url.URL, headers http.Header, serve bool) (http.Header, int, error) { headers, status := media.CORSHandler(method, headers, oh.corsOrigins, serve) if status != 0 || method == http.MethodPost || method == http.MethodPut { @@ -229,48 +299,11 @@ func (oh *ossHandler) Headers(tenantID types.TenantID, method string, url *url.U headers.Set("Cache-Control", oh.conf.CacheControl) return headers, http.StatusNotModified, nil } - objectKey, err := oh.resolveObjectKey(fdef.Location) - if err != nil { - return nil, 0, err - } - - var redirURL string - switch method { - case http.MethodGet: - options := []aliyunoss.Option{ - aliyunoss.ResponseCacheControl(oh.conf.CacheControl), - } - if isAttachment, _ := strconv.ParseBool(url.Query().Get("asatt")); isAttachment { - options = append(options, aliyunoss.ResponseContentDisposition("attachment")) - } - redirURL, err = oh.bucket.SignURL(objectKey, aliyunoss.HTTPGet, int64(oh.conf.PresignTTL), options...) - case http.MethodHead: - redirURL, err = oh.bucket.SignURL(objectKey, aliyunoss.HTTPHead, int64(oh.conf.PresignTTL)) - } - if err != nil { - return nil, 0, err - } - - if redirURL != "" { - // Keep the CORS and Vary headers produced above. Replacing the map here - // makes preflight succeed but causes the actual browser GET to fail with - // MissingAllowOriginHeader before it can follow the signed OSS redirect. - headers.Set("Location", normalizeSignedRedirectURL(redirURL)) + if fdef.ETag != "" { headers.Set("ETag", `"`+fdef.ETag+`"`) - headers.Set("Content-Type", "application/json; charset=utf-8") - headers.Set("Cache-Control", oh.conf.CacheControl) - return headers, http.StatusFound, nil } - return nil, 0, nil -} - -func normalizeSignedRedirectURL(rawURL string) string { - parsed, err := url.Parse(strings.TrimSpace(rawURL)) - if err != nil || parsed.Scheme != "http" || parsed.Host == "" { - return rawURL - } - parsed.Scheme = "https" - return parsed.String() + headers.Set("Cache-Control", oh.conf.CacheControl) + return headers, 0, nil } // Upload processes request for a file upload. @@ -333,9 +366,41 @@ func (oh *ossHandler) UploadPublic(fdef *media.PublicFileDef, file io.Reader) (s return key, rc.count, nil } -// Download is unsupported because OSS objects are served through signed redirects. +// Download streams a private OSS object through the authenticated IM endpoint. The seekable reader +// reopens the object with a byte range after every seek, allowing http.ServeContent to handle HEAD +// and Range requests without buffering an entire image, video, or document in server memory. func (oh *ossHandler) Download(tenantID types.TenantID, url string) (*types.FileDef, media.ReadSeekCloser, error) { - return nil, nil, types.ErrUnsupported + fid := oh.GetIdFromUrl(url) + if fid.IsZero() { + return nil, nil, types.ErrNotFound + } + fd, err := oh.getFileRecord(tenantID, fid) + if err != nil { + return nil, nil, err + } + objectKey, err := oh.resolveObjectKey(fd.Location) + if err != nil { + return nil, nil, err + } + if fd.Size <= 0 { + meta, err := oh.bucket.GetObjectDetailedMeta(objectKey) + if err != nil { + return nil, nil, err + } + fd.Size, err = strconv.ParseInt(meta.Get("Content-Length"), 10, 64) + if err != nil || fd.Size < 0 { + return nil, nil, types.ErrMalformed + } + } + + reader := &ossReadSeekCloser{size: fd.Size} + reader.open = func(offset int64) (io.ReadCloser, error) { + if offset <= 0 { + return oh.bucket.GetObject(objectKey) + } + return oh.bucket.GetObject(objectKey, aliyunoss.Range(offset, fd.Size-1)) + } + return fd, reader, nil } // Delete deletes files from OSS by provided object keys. diff --git a/server/media/oss/oss_tenant_test.go b/server/media/oss/oss_tenant_test.go index fb77a67..9a5799b 100644 --- a/server/media/oss/oss_tenant_test.go +++ b/server/media/oss/oss_tenant_test.go @@ -1,6 +1,8 @@ package oss import ( + "bytes" + "io" "net/http" "testing" "time" @@ -58,14 +60,31 @@ func TestIsOSSObjectNotFound(t *testing.T) { } } -func TestNormalizeSignedRedirectURL(t *testing.T) { - rawURL := "http://im-oss-t1.oss-cn-shenzhen.aliyuncs.com/tenants/1/avatar.jpg?Expires=1&OSSAccessKeyId=ak&Signature=sig" - want := "https://im-oss-t1.oss-cn-shenzhen.aliyuncs.com/tenants/1/avatar.jpg?Expires=1&OSSAccessKeyId=ak&Signature=sig" - if got := normalizeSignedRedirectURL(rawURL); got != want { - t.Fatalf("normalizeSignedRedirectURL()=%q, want %q", got, want) +func TestOSSReadSeekCloser(t *testing.T) { + data := []byte("0123456789") + reader := &ossReadSeekCloser{ + size: int64(len(data)), + open: func(offset int64) (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(data[offset:])), nil + }, } - if got := normalizeSignedRedirectURL("/v0/file/s/avatar.jpg"); got != "/v0/file/s/avatar.jpg" { - t.Fatalf("relative URL must be unchanged, got %q", got) + defer reader.Close() + + buf := make([]byte, 3) + if n, err := reader.Read(buf); err != nil || n != 3 || string(buf) != "012" { + t.Fatalf("initial read = %q, %d, %v", buf, n, err) + } + if pos, err := reader.Seek(-2, io.SeekEnd); err != nil || pos != 8 { + t.Fatalf("seek end = %d, %v", pos, err) + } + if n, err := reader.Read(buf); err != nil && err != io.EOF || n != 2 || string(buf[:n]) != "89" { + t.Fatalf("ranged read = %q, %d, %v", buf[:n], n, err) + } + if pos, err := reader.Seek(4, io.SeekStart); err != nil || pos != 4 { + t.Fatalf("seek start = %d, %v", pos, err) + } + if n, err := reader.Read(buf[:2]); err != nil || n != 2 || string(buf[:2]) != "45" { + t.Fatalf("second ranged read = %q, %d, %v", buf[:2], n, err) } }