// Package oss implements media interface by storing media objects in Alibaba Cloud OSS. package oss import ( "encoding/base32" "encoding/json" "errors" "hash/fnv" "io" "net/http" "net/url" "path" "strconv" "strings" "sync/atomic" "time" aliyunoss "github.com/aliyun/aliyun-oss-go-sdk/oss" "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/media" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" ) const ( defaultServeURL = "/v0/file/s/" defaultCacheControl = "no-cache, must-revalidate" handlerName = "oss" // Presign GET URLs for this number of seconds. defaultPresignDuration = 120 ) type ossConfig struct { AccessKeyID string `json:"access_key_id"` AccessKeySecret string `json:"access_key_secret"` Endpoint string `json:"endpoint"` BucketName string `json:"bucket"` RegionID string `json:"region_id"` CorsOrigins []string `json:"cors_origins"` ServeURL string `json:"serve_url"` PresignTTL int `json:"presign_ttl"` CacheControl string `json:"cache_control"` VerifyBucket bool `json:"verify_bucket"` UseSysConfig bool `json:"use_sys_config"` } type ossHandler struct { client *aliyunoss.Client bucket *aliyunoss.Bucket conf ossConfig corsOrigins []media.AllowedOrigin } type readerCounter struct { count int64 reader io.Reader } func (rc *readerCounter) Read(buf []byte) (int, error) { n, err := rc.reader.Read(buf) atomic.AddInt64(&rc.count, int64(n)) return n, err } // Init initializes the media handler. func (oh *ossHandler) Init(jsconf string) error { var err error if err = json.Unmarshal([]byte(jsconf), &oh.conf); err != nil { return errors.New("failed to parse config: " + err.Error()) } if err = oh.loadSysConfig(); err != nil { return err } if oh.conf.AccessKeyID == "" { return errors.New("missing Access Key ID") } if oh.conf.AccessKeySecret == "" { return errors.New("missing Access Key Secret") } if oh.conf.Endpoint == "" { return errors.New("missing Endpoint") } if oh.conf.BucketName == "" { return errors.New("missing Bucket") } if oh.conf.PresignTTL <= 0 { oh.conf.PresignTTL = defaultPresignDuration } if oh.conf.CacheControl == "" { oh.conf.CacheControl = defaultCacheControl } if oh.conf.ServeURL == "" { oh.conf.ServeURL = defaultServeURL } oh.corsOrigins, err = media.ParseCORSAllow(oh.conf.CorsOrigins) if err != nil { return errors.New("failed to parse CORS allowed origins: " + err.Error()) } oh.client, err = aliyunoss.New(oh.conf.Endpoint, oh.conf.AccessKeyID, oh.conf.AccessKeySecret) if err != nil { return err } if oh.conf.VerifyBucket { exists, err := oh.client.IsBucketExist(oh.conf.BucketName) if err != nil { return err } if !exists { return errors.New("bucket does not exist") } } oh.bucket, err = oh.client.Bucket(oh.conf.BucketName) return err } func (oh *ossHandler) loadSysConfig() error { if !oh.conf.UseSysConfig { return nil } values, err := store.ThirdPartyConfig.Get("im_third_party", "oss_enabled", "oss_provider", "oss_endpoint", "oss_bucket", "oss_region_id", "oss_access_key_id", "oss_access_key_secret", "oss_presign_ttl", "oss_serve_url", "oss_cache_control", "oss_cors_origins", "oss_verify_bucket", ) if err != nil { return errors.New("failed to load OSS config from sys config: " + err.Error()) } if !truthy(values["oss_enabled"]) { return errors.New("OSS is disabled in sys config") } if provider := strings.TrimSpace(values["oss_provider"]); provider != "" && provider != handlerName { return errors.New("unsupported storage provider in sys config: " + provider) } applyString(&oh.conf.Endpoint, values["oss_endpoint"]) applyString(&oh.conf.BucketName, values["oss_bucket"]) applyString(&oh.conf.RegionID, values["oss_region_id"]) applyString(&oh.conf.AccessKeyID, values["oss_access_key_id"]) applyString(&oh.conf.AccessKeySecret, values["oss_access_key_secret"]) applyString(&oh.conf.ServeURL, values["oss_serve_url"]) applyString(&oh.conf.CacheControl, values["oss_cache_control"]) if origins := parseStringList(values["oss_cors_origins"]); len(origins) > 0 { oh.conf.CorsOrigins = origins } if verify := strings.TrimSpace(values["oss_verify_bucket"]); verify != "" { oh.conf.VerifyBucket = truthy(verify) } if ttl := strings.TrimSpace(values["oss_presign_ttl"]); ttl != "" { parsed, err := strconv.Atoi(ttl) if err != nil { return errors.New("invalid OSS presign TTL in sys config: " + err.Error()) } oh.conf.PresignTTL = parsed } return nil } func parseStringList(value string) []string { value = strings.TrimSpace(value) if value == "" { return nil } var items []string if strings.HasPrefix(value, "[") { if err := json.Unmarshal([]byte(value), &items); err == nil { return compactStrings(items) } } return compactStrings(strings.Split(value, ",")) } func compactStrings(values []string) []string { items := make([]string, 0, len(values)) for _, value := range values { if value = strings.TrimSpace(value); value != "" { items = append(items, value) } } return items } func applyString(target *string, value string) { if value = strings.TrimSpace(value); value != "" { *target = value } } func truthy(value string) bool { switch strings.ToLower(strings.TrimSpace(value)) { case "1", "true", "yes", "on": return true default: return false } } // Headers adds CORS headers and redirects GET and HEAD requests to OSS. 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 { return headers, status, nil } fid := oh.GetIdFromUrl(url.String()) if fid.IsZero() { return nil, 0, types.ErrNotFound } fdef, err := oh.getFileRecord(tenantID, fid) if err != nil { return nil, 0, err } if fdef.ETag != "" && strings.Trim(headers.Get("If-None-Match"), "\"") == fdef.ETag { return http.Header{ "ETag": {`"` + fdef.ETag + `"`}, "Cache-Control": {oh.conf.CacheControl}, }, 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 != "" { return http.Header{ "Location": {redirURL}, "ETag": {`"` + fdef.ETag + `"`}, "Content-Type": {"application/json; charset=utf-8"}, "Cache-Control": {oh.conf.CacheControl}, }, http.StatusPermanentRedirect, nil } return nil, 0, nil } // Upload processes request for a file upload. func (oh *ossHandler) Upload(fdef *types.FileDef, file io.Reader) (string, int64, error) { key, err := tenantObjectKey(fdef.TenantID, fdef.Uid().String32(), fdef.CreatedAt) if err != nil { return "", 0, err } if err = store.Files.StartUpload(fdef.TenantID, fdef); err != nil { logs.Warn.Println("failed to create file record", fdef.Id, err) return "", 0, err } rc := readerCounter{reader: file} options := []aliyunoss.Option{aliyunoss.CacheControl(oh.conf.CacheControl)} if fdef.MimeType != "" { options = append(options, aliyunoss.ContentType(fdef.MimeType)) } if err = oh.bucket.PutObject(key, &rc, options...); err != nil { return "", 0, err } fname := media.FileURLName(fdef) fdef.Location = key fdef.ETag = etagFromLocation(key) meta, err := oh.bucket.GetObjectDetailedMeta(key) if err != nil { return "", 0, err } if etag := strings.Trim(meta.Get("ETag"), "\""); etag != "" { fdef.ETag = etag } return oh.conf.ServeURL + fname, rc.count, nil } func (oh *ossHandler) UploadPublic(fdef *media.PublicFileDef, file io.Reader) (string, int64, error) { key, err := publicObjectKey(fdef.Purpose, fdef.ID, fdef.CreatedAt) if err != nil { return "", 0, err } rc := readerCounter{reader: file} options := []aliyunoss.Option{aliyunoss.CacheControl(oh.conf.CacheControl)} if fdef.MimeType != "" { options = append(options, aliyunoss.ContentType(fdef.MimeType)) } if err = oh.bucket.PutObject(key, &rc, options...); err != nil { return "", 0, err } fdef.Location = key fdef.ETag = etagFromLocation(key) meta, err := oh.bucket.GetObjectDetailedMeta(key) if err != nil { return "", 0, err } if etag := strings.Trim(meta.Get("ETag"), "\""); etag != "" { fdef.ETag = etag } return key, rc.count, nil } // Download is unsupported because OSS objects are served through signed redirects. func (oh *ossHandler) Download(tenantID types.TenantID, url string) (*types.FileDef, media.ReadSeekCloser, error) { return nil, nil, types.ErrUnsupported } // Delete deletes files from OSS by provided object keys. func (oh *ossHandler) Delete(locations []string) error { for i := 0; i < len(locations); i += 1000 { end := i + 1000 if end > len(locations) { end = len(locations) } if _, err := oh.bucket.DeleteObjects(locations[i:end], aliyunoss.DeleteObjectsQuiet(true)); err != nil { return err } } return nil } // GetIdFromUrl converts an attachment URL to a file UID. func (oh *ossHandler) GetIdFromUrl(url string) types.Uid { return media.GetIdFromUrl(url, oh.conf.ServeURL) } func (oh *ossHandler) getFileRecord(tenantID types.TenantID, fid types.Uid) (*types.FileDef, error) { fd, err := store.Files.Get(tenantID, fid.String()) if err != nil { return nil, err } if fd == nil { return nil, types.ErrNotFound } return fd, nil } func (oh *ossHandler) resolveObjectKey(location string) (string, error) { location = strings.TrimSpace(location) if location == "" { return "", types.ErrNotFound } if _, err := oh.bucket.GetObjectDetailedMeta(location); err == nil { return location, nil } else if !isOSSObjectNotFound(err) { return "", err } if legacy := strings.TrimPrefix(location, "uploads/"); legacy != location { if _, err := oh.bucket.GetObjectDetailedMeta(legacy); err == nil { return legacy, nil } else if !isOSSObjectNotFound(err) { return "", err } } return "", types.ErrNotFound } func isOSSObjectNotFound(err error) bool { if err == nil { return false } if serviceErr, ok := err.(aliyunoss.ServiceError); ok { return serviceErr.StatusCode == http.StatusNotFound || serviceErr.Code == "NoSuchKey" } return false } func tenantObjectKey(tenantID types.TenantID, fileID string, createdAt time.Time) (string, error) { if !tenantID.IsValid() || fileID == "" { return "", types.ErrMalformed } if createdAt.IsZero() { createdAt = time.Now().UTC() } return path.Join("tenants", strconv.FormatInt(int64(tenantID), 10), createdAt.Format("2006"), createdAt.Format("01"), fileID), nil } func publicObjectKey(purpose, fileID string, createdAt time.Time) (string, error) { purpose = cleanPublicPurpose(purpose) if purpose == "" || fileID == "" { return "", types.ErrMalformed } if createdAt.IsZero() { createdAt = time.Now().UTC() } return path.Join("public", purpose, createdAt.Format("2006"), createdAt.Format("01"), strings.Trim(fileID, "/")), nil } func cleanPublicPurpose(purpose string) string { purpose = strings.Trim(strings.ToLower(strings.TrimSpace(purpose)), "/") return strings.Map(func(r rune) rune { if r >= 'a' && r <= 'z' || r >= '0' && r <= '9' || r == '_' || r == '-' { return r } return -1 }, purpose) } func etagFromLocation(location string) string { hasher := fnv.New128() hasher.Write([]byte(location)) return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding). EncodeToString(hasher.Sum(make([]byte, 0, hasher.Size())))) } func init() { store.RegisterMediaHandler(handlerName, &ossHandler{}) }