From b4123b57483bf27aad4a88373f341b0567c990d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=96=D0=BE=D1=80=D0=B0=20=D0=97=D0=BC=D0=B5=D0=B9=D0=BA?= =?UTF-8?q?=D0=B8=D0=BD?= <48821354+Katze-942@users.noreply.github.com> Date: Thu, 29 Oct 2026 08:44:41 +0400 Subject: [PATCH] XDRIVE transport: Add "s3" storage backend (#7231) https://github.com/XTLS/Xray-core/pull/7231#issuecomment-6488077212 --- infra/conf/transport_method.go | 25 +- infra/conf/transport_test.go | 2 +- transport/internet/xdrive/config.pb.go | 107 ++++- transport/internet/xdrive/config.proto | 8 + transport/internet/xdrive/s3.go | 518 +++++++++++++++++++++++++ transport/internet/xdrive/storage.go | 14 + 6 files changed, 661 insertions(+), 13 deletions(-) create mode 100644 transport/internet/xdrive/s3.go diff --git a/infra/conf/transport_method.go b/infra/conf/transport_method.go index f10a2cf40..3f1379526 100644 --- a/infra/conf/transport_method.go +++ b/infra/conf/transport_method.go @@ -997,12 +997,20 @@ type XDriveConfig struct { EagerWindowMs uint32 `json:"eagerWindowMs"` HoleTimeoutMs uint32 `json:"holeTimeoutMs"` Template json.RawMessage `json:"template"` + S3 *XDriveS3Config `json:"s3"` Xmux XmuxConfig `json:"xmux"` // set by StreamConfig when "alpn" is ["h3"] h3 bool } +type XDriveS3Config struct { + Endpoint string `json:"endpoint"` + Region string `json:"region"` + Bucket string `json:"bucket"` + PathStyle *bool `json:"pathStyle"` +} + // Build implements Buildable. func (c *XDriveConfig) Build() (proto.Message, error) { switch c.Service { @@ -1019,8 +1027,12 @@ func (c *XDriveConfig) Build() (proto.Message, error) { if len(c.Template) == 0 { return nil, errors.New(`service "template" needs a "template" object`) } + case "s3": + if c.S3 == nil { + return nil, errors.New(`service "s3" needs an "s3" object`) + } default: - return nil, errors.New(`unknown "service": "`, c.Service, `", use "google-drive", "template" or "local"`) + return nil, errors.New(`unknown "service": "`, c.Service, `", use "google-drive", "template", "s3" or "local"`) } if c.Xmux.MaxConnections.To > 0 && c.Xmux.MaxConcurrency.To > 0 { return nil, errors.New("maxConnections cannot be specified together with maxConcurrency") @@ -1059,5 +1071,16 @@ func (c *XDriveConfig) Build() (proto.Message, error) { HKeepAlivePeriod: c.Xmux.HKeepAlivePeriod, }, } + if c.Service == "s3" { + config.S3 = &xdrive.S3Config{ + Endpoint: c.S3.Endpoint, + Region: c.S3.Region, + Bucket: c.S3.Bucket, + PathStyle: c.S3.PathStyle == nil || *c.S3.PathStyle, + } + if err := xdrive.ValidateS3Config(config); err != nil { + return nil, err + } + } return config, nil } diff --git a/infra/conf/transport_test.go b/infra/conf/transport_test.go index dfb66c426..44ce2afd1 100644 --- a/infra/conf/transport_test.go +++ b/infra/conf/transport_test.go @@ -324,7 +324,7 @@ func TestXDriveRejectsUnknownService(t *testing.T) { if err := json.Unmarshal([]byte(`{"remoteFolder": "/tmp/xdrive", "service": "Dropbox"}`), config); err != nil { t.Fatalf("Unmarshal: %v", err) } - if _, err := config.Build(); err == nil || !strings.Contains(err.Error(), `use "google-drive", "template" or "local"`) { + if _, err := config.Build(); err == nil || !strings.Contains(err.Error(), `use "google-drive", "template", "s3" or "local"`) { t.Fatalf("expected an unknown service error, got %v", err) } } diff --git a/transport/internet/xdrive/config.pb.go b/transport/internet/xdrive/config.pb.go index a350f0149..82705340c 100644 --- a/transport/internet/xdrive/config.pb.go +++ b/transport/internet/xdrive/config.pb.go @@ -37,6 +37,7 @@ type Config struct { HoleTimeoutMs uint32 `protobuf:"varint,11,opt,name=hole_timeout_ms,json=holeTimeoutMs,proto3" json:"hole_timeout_ms,omitempty"` Template string `protobuf:"bytes,12,opt,name=template,proto3" json:"template,omitempty"` Xmux *splithttp.XmuxConfig `protobuf:"bytes,13,opt,name=xmux,proto3" json:"xmux,omitempty"` + S3 *S3Config `protobuf:"bytes,14,opt,name=s3,proto3" json:"s3,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -162,11 +163,86 @@ func (x *Config) GetXmux() *splithttp.XmuxConfig { return nil } +func (x *Config) GetS3() *S3Config { + if x != nil { + return x.S3 + } + return nil +} + +type S3Config struct { + state protoimpl.MessageState `protogen:"open.v1"` + Endpoint string `protobuf:"bytes,1,opt,name=endpoint,proto3" json:"endpoint,omitempty"` + Region string `protobuf:"bytes,2,opt,name=region,proto3" json:"region,omitempty"` + Bucket string `protobuf:"bytes,3,opt,name=bucket,proto3" json:"bucket,omitempty"` + PathStyle bool `protobuf:"varint,4,opt,name=path_style,json=pathStyle,proto3" json:"path_style,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *S3Config) Reset() { + *x = S3Config{} + mi := &file_transport_internet_xdrive_config_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *S3Config) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*S3Config) ProtoMessage() {} + +func (x *S3Config) ProtoReflect() protoreflect.Message { + mi := &file_transport_internet_xdrive_config_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use S3Config.ProtoReflect.Descriptor instead. +func (*S3Config) Descriptor() ([]byte, []int) { + return file_transport_internet_xdrive_config_proto_rawDescGZIP(), []int{1} +} + +func (x *S3Config) GetEndpoint() string { + if x != nil { + return x.Endpoint + } + return "" +} + +func (x *S3Config) GetRegion() string { + if x != nil { + return x.Region + } + return "" +} + +func (x *S3Config) GetBucket() string { + if x != nil { + return x.Bucket + } + return "" +} + +func (x *S3Config) GetPathStyle() bool { + if x != nil { + return x.PathStyle + } + return false +} + var File_transport_internet_xdrive_config_proto protoreflect.FileDescriptor const file_transport_internet_xdrive_config_proto_rawDesc = "" + "\n" + - "&transport/internet/xdrive/config.proto\x12\x1exray.transport.internet.xdrive\x1a)transport/internet/splithttp/config.proto\"\x8e\x04\n" + + "&transport/internet/xdrive/config.proto\x12\x1exray.transport.internet.xdrive\x1a)transport/internet/splithttp/config.proto\"\xc8\x04\n" + "\x06Config\x12#\n" + "\rremote_folder\x18\x01 \x01(\tR\fremoteFolder\x12\x18\n" + "\aservice\x18\x02 \x01(\tR\aservice\x12\x18\n" + @@ -181,7 +257,14 @@ const file_transport_internet_xdrive_config_proto_rawDesc = "" + " \x01(\rR\reagerWindowMs\x12&\n" + "\x0fhole_timeout_ms\x18\v \x01(\rR\rholeTimeoutMs\x12\x1a\n" + "\btemplate\x18\f \x01(\tR\btemplate\x12A\n" + - "\x04xmux\x18\r \x01(\v2-.xray.transport.internet.splithttp.XmuxConfigR\x04xmuxB5Z3github.com/xtls/xray-core/transport/internet/xdriveb\x06proto3" + "\x04xmux\x18\r \x01(\v2-.xray.transport.internet.splithttp.XmuxConfigR\x04xmux\x128\n" + + "\x02s3\x18\x0e \x01(\v2(.xray.transport.internet.xdrive.S3ConfigR\x02s3\"u\n" + + "\bS3Config\x12\x1a\n" + + "\bendpoint\x18\x01 \x01(\tR\bendpoint\x12\x16\n" + + "\x06region\x18\x02 \x01(\tR\x06region\x12\x16\n" + + "\x06bucket\x18\x03 \x01(\tR\x06bucket\x12\x1d\n" + + "\n" + + "path_style\x18\x04 \x01(\bR\tpathStyleB5Z3github.com/xtls/xray-core/transport/internet/xdriveb\x06proto3" var ( file_transport_internet_xdrive_config_proto_rawDescOnce sync.Once @@ -195,18 +278,20 @@ func file_transport_internet_xdrive_config_proto_rawDescGZIP() []byte { return file_transport_internet_xdrive_config_proto_rawDescData } -var file_transport_internet_xdrive_config_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_transport_internet_xdrive_config_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_transport_internet_xdrive_config_proto_goTypes = []any{ (*Config)(nil), // 0: xray.transport.internet.xdrive.Config - (*splithttp.XmuxConfig)(nil), // 1: xray.transport.internet.splithttp.XmuxConfig + (*S3Config)(nil), // 1: xray.transport.internet.xdrive.S3Config + (*splithttp.XmuxConfig)(nil), // 2: xray.transport.internet.splithttp.XmuxConfig } var file_transport_internet_xdrive_config_proto_depIdxs = []int32{ - 1, // 0: xray.transport.internet.xdrive.Config.xmux:type_name -> xray.transport.internet.splithttp.XmuxConfig - 1, // [1:1] is the sub-list for method output_type - 1, // [1:1] is the sub-list for method input_type - 1, // [1:1] is the sub-list for extension type_name - 1, // [1:1] is the sub-list for extension extendee - 0, // [0:1] is the sub-list for field type_name + 2, // 0: xray.transport.internet.xdrive.Config.xmux:type_name -> xray.transport.internet.splithttp.XmuxConfig + 1, // 1: xray.transport.internet.xdrive.Config.s3:type_name -> xray.transport.internet.xdrive.S3Config + 2, // [2:2] is the sub-list for method output_type + 2, // [2:2] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name } func init() { file_transport_internet_xdrive_config_proto_init() } @@ -220,7 +305,7 @@ func file_transport_internet_xdrive_config_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_transport_internet_xdrive_config_proto_rawDesc), len(file_transport_internet_xdrive_config_proto_rawDesc)), NumEnums: 0, - NumMessages: 1, + NumMessages: 2, NumExtensions: 0, NumServices: 0, }, diff --git a/transport/internet/xdrive/config.proto b/transport/internet/xdrive/config.proto index f10687a9e..b894f6bbd 100644 --- a/transport/internet/xdrive/config.proto +++ b/transport/internet/xdrive/config.proto @@ -19,4 +19,12 @@ message Config { uint32 hole_timeout_ms = 11; string template = 12; xray.transport.internet.splithttp.XmuxConfig xmux = 13; + S3Config s3 = 14; +} + +message S3Config { + string endpoint = 1; + string region = 2; + string bucket = 3; + bool path_style = 4; } diff --git a/transport/internet/xdrive/s3.go b/transport/internet/xdrive/s3.go new file mode 100644 index 000000000..75017dc19 --- /dev/null +++ b/transport/internet/xdrive/s3.go @@ -0,0 +1,518 @@ +package xdrive + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/xml" + stderrors "errors" + "io" + "net" + "net/http" + "net/url" + "strconv" + "strings" + "sync" + "time" + + "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/transport/internet" + "golang.org/x/net/http/httpguts" +) + +type s3Storage struct { + *s3Shared + client *http.Client + inflight chan struct{} + release func() +} + +type s3Shared struct { + conns *connPool + base url.URL + prefix string + region string + accessKey, secretKey string + token string + backoff time.Duration +} + +func s3Region(region string) string { + if region == "" { + return "us-east-1" + } + return region +} + +func ValidateS3Config(c *Config) error { + if c == nil || c.S3 == nil { + return errors.New(`service "s3" needs an "s3" object`) + } + u, err := url.Parse(c.S3.Endpoint) + if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Hostname() == "" || + u.User != nil || u.RawQuery != "" || u.ForceQuery || u.Fragment != "" || u.Opaque != "" { + return errors.New("S3 endpoint must be an http(s) URL without credentials, query or fragment") + } + if !httpguts.ValidHeaderFieldValue(c.S3.Region) || strings.ContainsAny(c.S3.Region, "/, \t") { + return errors.New("invalid S3 signing region") + } + bucket := c.S3.Bucket + if !validS3Name(bucket) || strings.Contains(bucket, "/") { + return errors.New("S3 needs a bucket name without path separators") + } + if !c.S3.PathStyle { + if net.ParseIP(u.Hostname()) != nil { + return errors.New("S3 virtual-hosted addressing requires a DNS endpoint") + } + // In this mode the bucket is part of the HTTP authority, not its path. + if strings.Trim(bucket, "abcdefghijklmnopqrstuvwxyz0123456789-.") != "" || strings.Contains(bucket, "..") || + strings.HasPrefix(bucket, ".") || strings.HasSuffix(bucket, ".") { + return errors.New("S3 virtual-hosted addressing requires a DNS-compatible bucket name") + } + } + if !validS3Name(strings.TrimSuffix(c.RemoteFolder, "/")) { + return errors.New("S3 remoteFolder must be a nonempty relative prefix without empty, '.' or '..' components") + } + if len(c.Secrets) < 2 || len(c.Secrets) > 3 || c.Secrets[0] == "" || c.Secrets[1] == "" { + return errors.New("S3 needs AccessKeyID, SecretAccessKey and optionally SessionToken in secrets") + } + for _, secret := range c.Secrets { + if !httpguts.ValidHeaderFieldValue(secret) { + return errors.New("invalid S3 secret") + } + } + if strings.ContainsAny(c.Secrets[0], "/ ,\t") { + return errors.New("invalid S3 AccessKeyID") + } + return nil +} + +func validS3Name(name string) bool { + for _, ch := range name { + if ch < 0x20 || ch == 0x7f || ch == '\\' { + return false + } + } + for _, part := range strings.Split(name, "/") { + if part == "" || part == "." || part == ".." { + return false + } + } + return true +} + +func newS3Storage(streamSettings *internet.MemoryStreamConfig, c *Config) (*s3Storage, error) { + if err := ValidateS3Config(c); err != nil { + return nil, err + } + base, _ := url.Parse(c.S3.Endpoint) + // Encode the endpoint prefix for SigV4 without turning escaped slashes into separators. + segments := strings.Split(base.EscapedPath(), "/") + for i, segment := range segments { + decoded, _ := url.PathUnescape(segment) + segments[i] = strings.ReplaceAll(s3EscapePath(decoded), "/", "%2F") + } + base.RawPath = strings.Join(segments, "/") + if !strings.HasSuffix(base.RawPath, "/") { + base.Path += "/" + base.RawPath += "/" + } + if c.S3.PathStyle { + base.Path += c.S3.Bucket + "/" + base.RawPath += s3EscapePath(c.S3.Bucket) + "/" + } else { + host := c.S3.Bucket + "." + base.Hostname() + if base.Port() != "" { + host = net.JoinHostPort(host, base.Port()) + } + base.Host = host + } + if streamSettings == nil { + streamSettings = &internet.MemoryStreamConfig{} + } + h1Hosts := new(sync.Map) + newClient := func(ctx context.Context) *http.Client { + client := newServiceClient(ctx, h1Hosts, streamSettings, driveTimeout, driveMaxInflight) + // A redirect changes the signed host/path. Do not follow it with the old + // signature or send the session token to a different endpoint. + client.CheckRedirect = s3NoRedirect + return client + } + s := &s3Storage{ + s3Shared: &s3Shared{ + conns: newConnPool(c.Xmux, newClient), + base: *base, + prefix: strings.TrimSuffix(c.RemoteFolder, "/") + "/", + region: s3Region(c.S3.Region), + accessKey: c.Secrets[0], + secretKey: c.Secrets[1], + backoff: 200 * time.Millisecond, + }, + client: newClient(context.Background()), + inflight: make(chan struct{}, driveMaxInflight), + } + if len(c.Secrets) == 3 { + s.token = c.Secrets[2] + } + return s, nil +} + +func s3NoRedirect(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse } + +func (s *s3Storage) session() Storage { + transport := &sessionTransport{pool: s.conns} + return &s3Storage{ + s3Shared: s.s3Shared, + client: &http.Client{Transport: transport, CheckRedirect: s3NoRedirect}, + inflight: make(chan struct{}, cap(s.inflight)), + release: transport.Close, + } +} + +func (s *s3Storage) Close() error { + if s.release != nil { + s.release() + } + return nil +} + +// SigV4 requires %20 for spaces, uppercase escapes, and unescaped key separators. +func s3EscapePath(value string) string { + const digits = "0123456789ABCDEF" + var out strings.Builder + for i := 0; i < len(value); i++ { + ch := value[i] + if ch >= 'a' && ch <= 'z' || ch >= 'A' && ch <= 'Z' || ch >= '0' && ch <= '9' || + ch == '-' || ch == '_' || ch == '.' || ch == '~' || ch == '/' { + out.WriteByte(ch) + } else { + out.WriteByte('%') + out.WriteByte(digits[ch>>4]) + out.WriteByte(digits[ch&15]) + } + } + return out.String() +} + +func s3Hash(body []byte) string { + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]) +} + +func s3HMAC(key []byte, value string) []byte { + h := hmac.New(sha256.New, key) + h.Write([]byte(value)) + return h.Sum(nil) +} + +func (s *s3Storage) sign(req *http.Request, payloadHash string, now time.Time) { + stamp, date := now.UTC().Format("20060102T150405Z"), now.UTC().Format("20060102") + req.Header.Set("X-Amz-Date", stamp) + req.Header.Set("X-Amz-Content-Sha256", payloadHash) + headers := "host:" + req.URL.Host + "\nx-amz-content-sha256:" + payloadHash + "\nx-amz-date:" + stamp + "\n" + signed := "host;x-amz-content-sha256;x-amz-date" + if contentType := req.Header.Get("Content-Type"); contentType != "" { + headers = "content-type:" + contentType + "\n" + headers + signed = "content-type;" + signed + } + if s.token != "" { + req.Header.Set("X-Amz-Security-Token", s.token) + headers += "x-amz-security-token:" + strings.Join(strings.Fields(s.token), " ") + "\n" + signed += ";x-amz-security-token" + } + // Sign the escaped path exactly as sent. + canonical := req.Method + "\n" + req.URL.EscapedPath() + "\n" + req.URL.RawQuery + "\n" + + headers + "\n" + signed + "\n" + payloadHash + scope := date + "/" + s.region + "/s3/aws4_request" + toSign := "AWS4-HMAC-SHA256\n" + stamp + "\n" + scope + "\n" + s3Hash([]byte(canonical)) + key := s3HMAC([]byte("AWS4"+s.secretKey), date) + key = s3HMAC(key, s.region) + key = s3HMAC(key, "s3") + key = s3HMAC(key, "aws4_request") + req.Header.Set("Authorization", "AWS4-HMAC-SHA256 Credential="+s.accessKey+"/"+scope+ + ", SignedHeaders="+signed+", Signature="+hex.EncodeToString(s3HMAC(key, toSign))) +} + +func (s *s3Storage) target(key string, query url.Values) string { + u := s.base + u.Path += key + u.RawPath += s3EscapePath(key) + // Parameters have fixed ASCII names and one value each; SigV4 uses %20 for spaces. + u.RawQuery = strings.ReplaceAll(query.Encode(), "+", "%20") + return u.String() +} + +func (s *s3Storage) objectKey(name string) (string, error) { + if !validS3Name(name) { + return "", errors.New("invalid relative S3 object name") + } + key := s.prefix + name + if len(key) > 1024 { + return "", errors.New("S3 object key exceeds 1024 bytes") + } + return key, nil +} + +// Retries use fresh signatures and hold a concurrency slot until the response body is read. +func (s *s3Storage) do(ctx context.Context, method, key string, query url.Values, body []byte) (int, []byte, error) { + payloadHash := s3Hash(body) + backoff := s.backoff + var lastErr error + var retryAfter time.Duration + for attempt := 0; attempt < driveMaxAttempts; attempt++ { + if attempt > 0 { + delay := max(jitter(backoff), retryAfter) + timer := time.NewTimer(min(delay, driveMaxBackoff)) + select { + case <-ctx.Done(): + timer.Stop() + return 0, nil, ctx.Err() + case <-timer.C: + } + backoff = min(backoff*2, driveMaxBackoff) + } + select { + case s.inflight <- struct{}{}: + case <-ctx.Done(): + return 0, nil, ctx.Err() + } + status, payload, after, err := s.attempt(ctx, method, key, query, body, payloadHash) + <-s.inflight + if ctx.Err() != nil { + return 0, nil, ctx.Err() + } + if err != nil && (!retryS3Error(err) || status >= 300 && !retryableStatus(status)) { + return 0, nil, err + } + if err == nil && !retryableStatus(status) { + return status, payload, nil + } + lastErr, retryAfter = err, after + if lastErr == nil { + lastErr = s3ResponseError(method, status, payload) + } + } + return 0, nil, lastErr +} + +func retryS3Error(err error) bool { + var dns *net.DNSError + if stderrors.As(err, &dns) { + return dns.IsTimeout || dns.IsTemporary + } + var op *net.OpError + var network net.Error + return stderrors.Is(err, io.EOF) || stderrors.Is(err, io.ErrUnexpectedEOF) || stderrors.Is(err, errStalled) || + stderrors.As(err, &op) || stderrors.As(err, &network) && network.Timeout() +} + +func (s *s3Storage) attempt(ctx context.Context, method, key string, query url.Values, body []byte, hash string) (int, []byte, time.Duration, error) { + req, err := http.NewRequestWithContext(ctx, method, s.target(key, query), bytes.NewReader(body)) + if err != nil { + return 0, nil, 0, errors.New("failed to build S3 request") + } + if method == http.MethodPut { + req.Header.Set("Content-Type", "application/octet-stream") + } + s.sign(req, hash, time.Now()) + resp, err := s.client.Do(req) + if err != nil { + if urlErr, ok := err.(*url.Error); ok { + err = urlErr.Err + } + return 0, nil, 0, errors.New("S3 request failed").Base(err) + } + defer resp.Body.Close() + limit := int64(maxSegmentBytes) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + limit = 64 << 10 + } + payload, err := io.ReadAll(io.LimitReader(resp.Body, limit+1)) + if err != nil { + return resp.StatusCode, nil, 0, errors.New("failed to read S3 response (HTTP ", resp.StatusCode, ")").Base(err) + } + if int64(len(payload)) > limit { + return 0, nil, 0, errors.New("S3 response exceeds size limit") + } + var after time.Duration + if raw := resp.Header.Get("Retry-After"); raw != "" { + if seconds, err := strconv.ParseInt(raw, 10, 32); err == nil && seconds > 0 { + after = min(time.Duration(seconds)*time.Second, driveMaxBackoff) + } else if at, err := http.ParseTime(raw); err == nil { + after = min(max(time.Until(at), 0), driveMaxBackoff) + } + } + return resp.StatusCode, payload, after, nil +} + +func s3Missing(status int, payload []byte) bool { + if status != http.StatusNotFound { + return false + } + code := s3ErrorCode(payload) + return code == "NoSuchKey" || code == "NotFound" +} + +func s3ErrorCode(payload []byte) string { + var response struct { + XMLName xml.Name `xml:"Error"` + Code string + } + if xml.Unmarshal(payload, &response) != nil { + return "" + } + return response.Code +} + +func s3ResponseError(method string, status int, payload []byte) error { + if code := s3ErrorCode(payload); code != "" { + return errors.New("S3 ", method, " answered ", status, " (", strconv.Quote(code), ")") + } + return errors.New("S3 ", method, " answered ", status) +} + +func (s *s3Storage) Put(ctx context.Context, name string, data []byte) error { + key, err := s.objectKey(name) + if err != nil { + return err + } + status, payload, err := s.do(ctx, http.MethodPut, key, nil, data) + if err != nil { + return err + } + if status != http.StatusOK { + return s3ResponseError("PUT", status, payload) + } + return nil +} + +func (s *s3Storage) Get(ctx context.Context, name string) ([]byte, error) { + key, err := s.objectKey(name) + if err != nil { + return nil, err + } + status, payload, err := s.do(ctx, http.MethodGet, key, nil, nil) + if err != nil { + return nil, err + } + if s3Missing(status, payload) { + return nil, errNotFound + } + if status != http.StatusOK { + return nil, s3ResponseError("GET", status, payload) + } + return payload, nil +} + +type s3ListPage struct { + XMLName xml.Name `xml:"ListBucketResult"` + IsTruncated bool + NextContinuationToken string + Contents []struct{ Key string } + CommonPrefixes []struct{ Prefix string } +} + +func (s *s3Storage) list(ctx context.Context, prefix, delimiter string, visit func(*s3ListPage) error) error { + query := url.Values{"list-type": {"2"}, "prefix": {prefix}, "max-keys": {"1000"}} + if delimiter != "" { + query.Set("delimiter", delimiter) + } + seen := make(map[string]bool) + for { + status, payload, err := s.do(ctx, http.MethodGet, "", query, nil) + if err != nil { + return err + } + if status != http.StatusOK { + return s3ResponseError("LIST", status, payload) + } + var page s3ListPage + if err := xml.Unmarshal(payload, &page); err != nil { + return errors.New("invalid S3 list response") + } + // Validate all keys before deleting any objects. + for _, item := range page.Contents { + if !strings.HasPrefix(item.Key, prefix) { + return errors.New("S3 returned an object outside the requested prefix") + } + } + for _, item := range page.CommonPrefixes { + if !strings.HasPrefix(item.Prefix, prefix) { + return errors.New("S3 returned a directory outside the requested prefix") + } + } + if page.IsTruncated { + if page.NextContinuationToken == "" || seen[page.NextContinuationToken] { + return errors.New("S3 list pagination did not advance") + } + seen[page.NextContinuationToken] = true + } + if err := visit(&page); err != nil { + return err + } + if !page.IsTruncated { + return nil + } + query.Set("continuation-token", page.NextContinuationToken) + } +} + +func (s *s3Storage) List(ctx context.Context, prefix string) ([]Entry, error) { + key, err := s.objectKey(prefix) + if err != nil { + return nil, err + } + key += "/" + seen := make(map[string]bool) + var entries []Entry + add := func(full string) { + rest := strings.TrimPrefix(full, key) + name, _, _ := strings.Cut(rest, "/") + if name != "" && validS3Name(name) && !seen[name] { + seen[name] = true + entries = append(entries, Entry{Name: name}) + } + } + err = s.list(ctx, key, "/", func(page *s3ListPage) error { + for _, item := range page.Contents { + add(item.Key) + } + for _, item := range page.CommonPrefixes { + add(item.Prefix) + } + return nil + }) + return entries, err +} + +func (s *s3Storage) deleteObject(ctx context.Context, key string) error { + status, payload, err := s.do(ctx, http.MethodDelete, key, nil, nil) + if err != nil { + return err + } + if status != http.StatusNoContent && status != http.StatusOK && !s3Missing(status, payload) { + return s3ResponseError("DELETE", status, payload) + } + return nil +} + +// The trailing "/" prevents deleting streams/abcd when removing streams/abc. +func (s *s3Storage) Delete(ctx context.Context, name string) error { + key, err := s.objectKey(name) + if err != nil { + return err + } + if err := s.deleteObject(ctx, key); err != nil { + return err + } + return s.list(ctx, key+"/", "", func(page *s3ListPage) error { + for _, item := range page.Contents { + if err := s.deleteObject(ctx, item.Key); err != nil { + return err + } + } + return nil + }) +} diff --git a/transport/internet/xdrive/storage.go b/transport/internet/xdrive/storage.go index dd49a13f9..3f79af591 100644 --- a/transport/internet/xdrive/storage.go +++ b/transport/internet/xdrive/storage.go @@ -43,6 +43,10 @@ func newStorage(streamSettings *internet.MemoryStreamConfig) (Storage, error) { return sharedStorage(streamSettings, config, func() (Storage, error) { return newTemplateStorage(streamSettings, config) }) + case "s3": + return sharedStorage(streamSettings, config, func() (Storage, error) { + return newS3Storage(streamSettings, config) + }) default: return nil, errors.New("unsupported service: ", config.Service) } @@ -57,6 +61,8 @@ func sessionStorage(storage Storage) Storage { return s.session() case *templateStorage: return s.session() + case *s3Storage: + return s.session() } return storage } @@ -68,6 +74,14 @@ var ( func shareKey(streamSettings *internet.MemoryStreamConfig, config *Config) string { parts := []string{config.Service, config.RemoteFolder} + if config.S3 != nil { + parts = append(parts, config.S3.Endpoint, s3Region(config.S3.Region), config.S3.Bucket) + if config.S3.PathStyle { + parts = append(parts, "path-style") + } else { + parts = append(parts, "virtual-hosted-style") + } + } parts = append(parts, config.Secrets...) if streamSettings != nil { parts = append(parts, streamSettings.SecurityType)