1205 lines
32 KiB
Go
1205 lines
32 KiB
Go
package services
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"encoding/xml"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"path"
|
|
"path/filepath"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/hirochachacha/go-smb2"
|
|
"github.com/minio/minio-go/v7"
|
|
"github.com/minio/minio-go/v7/pkg/credentials"
|
|
"github.com/pkg/sftp"
|
|
"go.etcd.io/bbolt"
|
|
"golang.org/x/crypto/ssh"
|
|
)
|
|
|
|
var storageBackendsBucket = []byte("storage_backends")
|
|
|
|
const (
|
|
StorageBackendLocal = "local"
|
|
StorageBackendS3 = "s3"
|
|
StorageBackendSFTP = "sftp"
|
|
StorageBackendSMB = "smb"
|
|
StorageBackendWebDAV = "webdav"
|
|
|
|
StorageProviderS3 = "s3"
|
|
StorageProviderContabo = "contabo"
|
|
StorageProviderSFTP = "sftp"
|
|
StorageProviderSMB = "smb"
|
|
StorageProviderWebDAV = "webdav"
|
|
)
|
|
|
|
type StorageObject struct {
|
|
Key string
|
|
Size int64
|
|
ContentType string
|
|
ModTime time.Time
|
|
Body io.ReadCloser
|
|
}
|
|
|
|
type StorageBackend interface {
|
|
ID() string
|
|
Type() string
|
|
Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) error
|
|
Get(ctx context.Context, key string) (StorageObject, error)
|
|
Delete(ctx context.Context, key string) error
|
|
DeletePrefix(ctx context.Context, prefix string) error
|
|
Usage(ctx context.Context) (int64, error)
|
|
Test(ctx context.Context) error
|
|
}
|
|
|
|
type StorageBackendConfig struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
Type string `json:"type"`
|
|
Provider string `json:"provider,omitempty"`
|
|
Enabled bool `json:"enabled"`
|
|
LocalPath string `json:"localPath,omitempty"`
|
|
Endpoint string `json:"endpoint,omitempty"`
|
|
Region string `json:"region,omitempty"`
|
|
Bucket string `json:"bucket,omitempty"`
|
|
AccessKey string `json:"accessKey,omitempty"`
|
|
SecretKey string `json:"secretKey,omitempty"`
|
|
UseSSL bool `json:"useSsl,omitempty"`
|
|
PathStyle bool `json:"pathStyle,omitempty"`
|
|
Host string `json:"host,omitempty"`
|
|
Port int `json:"port,omitempty"`
|
|
Username string `json:"username,omitempty"`
|
|
Password string `json:"password,omitempty"`
|
|
PrivateKey string `json:"privateKey,omitempty"`
|
|
HostKey string `json:"hostKey,omitempty"`
|
|
RemotePath string `json:"remotePath,omitempty"`
|
|
Share string `json:"share,omitempty"`
|
|
Domain string `json:"domain,omitempty"`
|
|
CreatedAt time.Time `json:"createdAt"`
|
|
UpdatedAt time.Time `json:"updatedAt"`
|
|
LastTestedAt time.Time `json:"lastTestedAt,omitempty"`
|
|
LastTestError string `json:"lastTestError,omitempty"`
|
|
LastTestSuccess bool `json:"lastTestSuccess,omitempty"`
|
|
}
|
|
|
|
type StorageBackendView struct {
|
|
Config StorageBackendConfig
|
|
UsageBytes int64
|
|
UsageLabel string
|
|
InUse bool
|
|
}
|
|
|
|
type StorageService struct {
|
|
db *bbolt.DB
|
|
localFilesDir string
|
|
}
|
|
|
|
func NewStorageService(db *bbolt.DB, dataDir string) (*StorageService, error) {
|
|
filesDir := filepath.Join(dataDir, "files")
|
|
if err := os.MkdirAll(filesDir, 0o755); err != nil {
|
|
return nil, err
|
|
}
|
|
service := &StorageService{db: db, localFilesDir: filesDir}
|
|
err := db.Update(func(tx *bbolt.Tx) error {
|
|
_, err := tx.CreateBucketIfNotExists(storageBackendsBucket)
|
|
return err
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return service, nil
|
|
}
|
|
|
|
func (s *StorageService) LocalFilesDir() string {
|
|
return s.localFilesDir
|
|
}
|
|
|
|
func (s *StorageService) Backend(id string) (StorageBackend, error) {
|
|
cfg, err := s.BackendConfig(id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !cfg.Enabled {
|
|
return nil, fmt.Errorf("storage backend is disabled")
|
|
}
|
|
return s.backendFromConfig(cfg)
|
|
}
|
|
|
|
func (s *StorageService) BackendConfig(id string) (StorageBackendConfig, error) {
|
|
id = strings.TrimSpace(id)
|
|
if id == "" || id == StorageBackendLocal {
|
|
return s.localConfig(), nil
|
|
}
|
|
var cfg StorageBackendConfig
|
|
err := s.db.View(func(tx *bbolt.Tx) error {
|
|
data := tx.Bucket(storageBackendsBucket).Get([]byte(id))
|
|
if data == nil {
|
|
return os.ErrNotExist
|
|
}
|
|
return json.Unmarshal(data, &cfg)
|
|
})
|
|
if err != nil {
|
|
return StorageBackendConfig{}, err
|
|
}
|
|
return cfg, nil
|
|
}
|
|
|
|
func (s *StorageService) ListBackendConfigs() ([]StorageBackendConfig, error) {
|
|
configs := []StorageBackendConfig{s.localConfig()}
|
|
err := s.db.View(func(tx *bbolt.Tx) error {
|
|
return tx.Bucket(storageBackendsBucket).ForEach(func(_, value []byte) error {
|
|
var cfg StorageBackendConfig
|
|
if err := json.Unmarshal(value, &cfg); err != nil {
|
|
return err
|
|
}
|
|
configs = append(configs, cfg)
|
|
return nil
|
|
})
|
|
})
|
|
sort.Slice(configs, func(i, j int) bool {
|
|
if configs[i].ID == StorageBackendLocal {
|
|
return true
|
|
}
|
|
if configs[j].ID == StorageBackendLocal {
|
|
return false
|
|
}
|
|
return strings.ToLower(configs[i].Name) < strings.ToLower(configs[j].Name)
|
|
})
|
|
return configs, err
|
|
}
|
|
|
|
func (s *StorageService) CreateS3Backend(input StorageBackendConfig) (StorageBackendConfig, error) {
|
|
input.ID = randomID(10)
|
|
input.Provider = normalizeStorageProvider(input.Provider)
|
|
switch input.Provider {
|
|
case StorageProviderSFTP:
|
|
input.Type = StorageBackendSFTP
|
|
case StorageProviderSMB:
|
|
input.Type = StorageBackendSMB
|
|
case StorageProviderWebDAV:
|
|
input.Type = StorageBackendWebDAV
|
|
default:
|
|
input.Type = StorageBackendS3
|
|
}
|
|
if err := normalizeStorageBackendConfig(&input, true); err != nil {
|
|
return StorageBackendConfig{}, err
|
|
}
|
|
now := time.Now().UTC()
|
|
input.Enabled = true
|
|
input.CreatedAt = now
|
|
input.UpdatedAt = now
|
|
if err := s.SaveBackendConfig(input); err != nil {
|
|
return StorageBackendConfig{}, err
|
|
}
|
|
return input, nil
|
|
}
|
|
|
|
func (s *StorageService) UpdateS3Backend(id string, input StorageBackendConfig) (StorageBackendConfig, error) {
|
|
current, err := s.BackendConfig(id)
|
|
if err != nil {
|
|
return StorageBackendConfig{}, err
|
|
}
|
|
if current.ID == StorageBackendLocal {
|
|
return StorageBackendConfig{}, fmt.Errorf("local storage cannot be edited")
|
|
}
|
|
input.ID = current.ID
|
|
input.Type = current.Type
|
|
input.Provider = normalizeStorageProvider(input.Provider)
|
|
switch input.Provider {
|
|
case StorageProviderSFTP:
|
|
input.Type = StorageBackendSFTP
|
|
case StorageProviderSMB:
|
|
input.Type = StorageBackendSMB
|
|
case StorageProviderWebDAV:
|
|
input.Type = StorageBackendWebDAV
|
|
default:
|
|
input.Type = StorageBackendS3
|
|
}
|
|
if strings.TrimSpace(input.SecretKey) == "" {
|
|
input.SecretKey = current.SecretKey
|
|
}
|
|
if strings.TrimSpace(input.Password) == "" {
|
|
input.Password = current.Password
|
|
}
|
|
if strings.TrimSpace(input.PrivateKey) == "" {
|
|
input.PrivateKey = current.PrivateKey
|
|
}
|
|
if strings.TrimSpace(input.HostKey) == "" {
|
|
input.HostKey = current.HostKey
|
|
}
|
|
input.Enabled = current.Enabled
|
|
input.CreatedAt = current.CreatedAt
|
|
input.LastTestedAt = current.LastTestedAt
|
|
input.LastTestError = current.LastTestError
|
|
input.LastTestSuccess = current.LastTestSuccess
|
|
if err := normalizeStorageBackendConfig(&input, false); err != nil {
|
|
return StorageBackendConfig{}, err
|
|
}
|
|
if err := s.SaveBackendConfig(input); err != nil {
|
|
return StorageBackendConfig{}, err
|
|
}
|
|
return input, nil
|
|
}
|
|
|
|
func normalizeStorageBackendConfig(input *StorageBackendConfig, creating bool) error {
|
|
input.Name = strings.TrimSpace(input.Name)
|
|
input.Provider = normalizeStorageProvider(input.Provider)
|
|
if input.Provider == StorageProviderSFTP {
|
|
input.Type = StorageBackendSFTP
|
|
input.Host = strings.TrimSpace(input.Host)
|
|
input.Username = strings.TrimSpace(input.Username)
|
|
input.Password = strings.TrimSpace(input.Password)
|
|
input.PrivateKey = strings.TrimSpace(input.PrivateKey)
|
|
input.HostKey = strings.TrimSpace(input.HostKey)
|
|
input.RemotePath = cleanRemoteRoot(input.RemotePath)
|
|
if input.Port <= 0 {
|
|
input.Port = 22
|
|
}
|
|
if input.Name == "" {
|
|
input.Name = input.Host
|
|
}
|
|
if input.Name == "" || input.Host == "" || input.Username == "" || (input.Password == "" && input.PrivateKey == "") {
|
|
return fmt.Errorf("name, host, username, and password or private key are required")
|
|
}
|
|
return nil
|
|
}
|
|
if input.Provider == StorageProviderSMB {
|
|
input.Type = StorageBackendSMB
|
|
input.Host = strings.TrimSpace(input.Host)
|
|
input.Username = strings.TrimSpace(input.Username)
|
|
input.Password = strings.TrimSpace(input.Password)
|
|
input.Share = strings.Trim(strings.TrimSpace(input.Share), `/\`)
|
|
input.Domain = strings.TrimSpace(input.Domain)
|
|
input.RemotePath = cleanRemoteRoot(input.RemotePath)
|
|
if input.Port <= 0 {
|
|
input.Port = 445
|
|
}
|
|
if input.Name == "" {
|
|
input.Name = input.Share
|
|
}
|
|
if input.Name == "" || input.Host == "" || input.Share == "" || input.Username == "" || input.Password == "" {
|
|
return fmt.Errorf("name, host, share, username, and password are required")
|
|
}
|
|
return nil
|
|
}
|
|
if input.Provider == StorageProviderWebDAV {
|
|
input.Type = StorageBackendWebDAV
|
|
input.Endpoint = strings.TrimSpace(input.Endpoint)
|
|
input.Username = strings.TrimSpace(input.Username)
|
|
input.Password = strings.TrimSpace(input.Password)
|
|
input.RemotePath = cleanRemoteRoot(input.RemotePath)
|
|
if input.Name == "" {
|
|
input.Name = input.Endpoint
|
|
}
|
|
if input.Name == "" || input.Endpoint == "" {
|
|
return fmt.Errorf("name and WebDAV URL are required")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
input.Type = StorageBackendS3
|
|
if input.Provider == StorageProviderContabo {
|
|
input.UseSSL = true
|
|
input.PathStyle = true
|
|
}
|
|
input.Name = strings.TrimSpace(input.Name)
|
|
input.Endpoint = strings.TrimSpace(input.Endpoint)
|
|
input.Region = strings.TrimSpace(input.Region)
|
|
input.Bucket = strings.TrimSpace(input.Bucket)
|
|
input.AccessKey = strings.TrimSpace(input.AccessKey)
|
|
input.SecretKey = strings.TrimSpace(input.SecretKey)
|
|
if input.Name == "" {
|
|
input.Name = input.Bucket
|
|
}
|
|
if input.Name == "" || input.Endpoint == "" || input.Bucket == "" || input.AccessKey == "" || input.SecretKey == "" {
|
|
return fmt.Errorf("name, endpoint, bucket, access key, and secret key are required")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *StorageService) SaveBackendConfig(cfg StorageBackendConfig) error {
|
|
if cfg.ID == "" || cfg.ID == StorageBackendLocal {
|
|
return fmt.Errorf("invalid storage backend id")
|
|
}
|
|
cfg.UpdatedAt = time.Now().UTC()
|
|
data, err := json.Marshal(cfg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.db.Update(func(tx *bbolt.Tx) error {
|
|
return tx.Bucket(storageBackendsBucket).Put([]byte(cfg.ID), data)
|
|
})
|
|
}
|
|
|
|
func (s *StorageService) DisableBackend(id string, inUse bool) error {
|
|
if id == "" || id == StorageBackendLocal {
|
|
return fmt.Errorf("local storage cannot be disabled")
|
|
}
|
|
if inUse {
|
|
return fmt.Errorf("storage backend is in use")
|
|
}
|
|
cfg, err := s.BackendConfig(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
cfg.Enabled = false
|
|
return s.SaveBackendConfig(cfg)
|
|
}
|
|
|
|
func (s *StorageService) DeleteBackend(id string, inUse bool) error {
|
|
if id == "" || id == StorageBackendLocal {
|
|
return fmt.Errorf("local storage cannot be deleted")
|
|
}
|
|
if inUse {
|
|
return fmt.Errorf("storage backend is in use")
|
|
}
|
|
return s.db.Update(func(tx *bbolt.Tx) error {
|
|
return tx.Bucket(storageBackendsBucket).Delete([]byte(id))
|
|
})
|
|
}
|
|
|
|
func (s *StorageService) TestBackend(id string) (StorageBackendConfig, error) {
|
|
cfg, err := s.BackendConfig(id)
|
|
if err != nil {
|
|
return StorageBackendConfig{}, err
|
|
}
|
|
backend, err := s.backendFromConfig(cfg)
|
|
if err != nil {
|
|
return StorageBackendConfig{}, err
|
|
}
|
|
err = backend.Test(context.Background())
|
|
cfg.LastTestedAt = time.Now().UTC()
|
|
cfg.LastTestError = ""
|
|
cfg.LastTestSuccess = err == nil
|
|
if err != nil {
|
|
cfg.LastTestError = err.Error()
|
|
}
|
|
if cfg.ID != StorageBackendLocal {
|
|
_ = s.SaveBackendConfig(cfg)
|
|
}
|
|
return cfg, err
|
|
}
|
|
|
|
func (s *StorageService) backendFromConfig(cfg StorageBackendConfig) (StorageBackend, error) {
|
|
switch cfg.Type {
|
|
case StorageBackendLocal:
|
|
return localStorageBackend{id: cfg.ID, root: cfg.LocalPath}, nil
|
|
case StorageBackendS3:
|
|
return newS3StorageBackend(cfg)
|
|
case StorageBackendSFTP:
|
|
return sftpStorageBackend{cfg: cfg}, nil
|
|
case StorageBackendSMB:
|
|
return smbStorageBackend{cfg: cfg}, nil
|
|
case StorageBackendWebDAV:
|
|
return webDAVStorageBackend{cfg: cfg, client: http.DefaultClient}, nil
|
|
default:
|
|
return nil, fmt.Errorf("unsupported storage backend type %q", cfg.Type)
|
|
}
|
|
}
|
|
|
|
func (s *StorageService) localConfig() StorageBackendConfig {
|
|
now := time.Now().UTC()
|
|
return StorageBackendConfig{
|
|
ID: StorageBackendLocal,
|
|
Name: "Local files",
|
|
Type: StorageBackendLocal,
|
|
Provider: StorageBackendLocal,
|
|
Enabled: true,
|
|
LocalPath: s.localFilesDir,
|
|
CreatedAt: now,
|
|
UpdatedAt: now,
|
|
}
|
|
}
|
|
|
|
type localStorageBackend struct {
|
|
id string
|
|
root string
|
|
}
|
|
|
|
func (b localStorageBackend) ID() string { return b.id }
|
|
func (b localStorageBackend) Type() string { return StorageBackendLocal }
|
|
|
|
func (b localStorageBackend) Put(_ context.Context, key string, body io.Reader, _ int64, _ string) error {
|
|
path, err := b.path(key)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
|
return err
|
|
}
|
|
target, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer target.Close()
|
|
_, err = io.Copy(target, body)
|
|
return err
|
|
}
|
|
|
|
func (b localStorageBackend) Get(_ context.Context, key string) (StorageObject, error) {
|
|
path, err := b.path(key)
|
|
if err != nil {
|
|
return StorageObject{}, err
|
|
}
|
|
source, err := os.Open(path)
|
|
if err != nil {
|
|
return StorageObject{}, err
|
|
}
|
|
stat, err := source.Stat()
|
|
if err != nil {
|
|
source.Close()
|
|
return StorageObject{}, err
|
|
}
|
|
return StorageObject{Key: key, Size: stat.Size(), ModTime: stat.ModTime(), Body: source}, nil
|
|
}
|
|
|
|
func (b localStorageBackend) Delete(_ context.Context, key string) error {
|
|
path, err := b.path(key)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b localStorageBackend) DeletePrefix(_ context.Context, prefix string) error {
|
|
path, err := b.path(prefix)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := os.RemoveAll(path); err != nil && !os.IsNotExist(err) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b localStorageBackend) Usage(_ context.Context) (int64, error) {
|
|
var total int64
|
|
err := filepath.WalkDir(b.root, func(path string, entry os.DirEntry, err error) error {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if entry.IsDir() {
|
|
return nil
|
|
}
|
|
info, err := entry.Info()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
total += info.Size()
|
|
return nil
|
|
})
|
|
if os.IsNotExist(err) {
|
|
return 0, nil
|
|
}
|
|
return total, err
|
|
}
|
|
|
|
func (b localStorageBackend) Test(ctx context.Context) error {
|
|
key := ".warpbox-storage-test-" + randomID(6)
|
|
if err := b.Put(ctx, key, strings.NewReader("ok"), 2, "text/plain"); err != nil {
|
|
return err
|
|
}
|
|
return b.Delete(ctx, key)
|
|
}
|
|
|
|
func (b localStorageBackend) path(key string) (string, error) {
|
|
key = filepath.Clean(strings.TrimPrefix(key, "/"))
|
|
if key == "." || strings.HasPrefix(key, "..") || filepath.IsAbs(key) {
|
|
return "", fmt.Errorf("invalid storage key")
|
|
}
|
|
path := filepath.Join(b.root, key)
|
|
root, err := filepath.Abs(b.root)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
abs, err := filepath.Abs(path)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if abs != root && !strings.HasPrefix(abs, root+string(os.PathSeparator)) {
|
|
return "", fmt.Errorf("invalid storage key")
|
|
}
|
|
return abs, nil
|
|
}
|
|
|
|
type s3StorageBackend struct {
|
|
cfg StorageBackendConfig
|
|
client *minio.Client
|
|
}
|
|
|
|
func newS3StorageBackend(cfg StorageBackendConfig) (*s3StorageBackend, error) {
|
|
endpoint := normalizeS3Endpoint(cfg.Endpoint)
|
|
client, err := minio.New(endpoint, &minio.Options{
|
|
Creds: credentials.NewStaticV4(cfg.AccessKey, cfg.SecretKey, ""),
|
|
Secure: cfg.UseSSL,
|
|
Region: cfg.Region,
|
|
BucketLookup: s3BucketLookup(cfg.PathStyle),
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &s3StorageBackend{cfg: cfg, client: client}, nil
|
|
}
|
|
|
|
func (b *s3StorageBackend) ID() string { return b.cfg.ID }
|
|
func (b *s3StorageBackend) Type() string { return StorageBackendS3 }
|
|
|
|
func (b *s3StorageBackend) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) error {
|
|
opts := minio.PutObjectOptions{ContentType: contentType}
|
|
_, err := b.client.PutObject(ctx, b.cfg.Bucket, cleanObjectKey(key), body, size, opts)
|
|
return err
|
|
}
|
|
|
|
func (b *s3StorageBackend) Get(ctx context.Context, key string) (StorageObject, error) {
|
|
object, err := b.client.GetObject(ctx, b.cfg.Bucket, cleanObjectKey(key), minio.GetObjectOptions{})
|
|
if err != nil {
|
|
return StorageObject{}, err
|
|
}
|
|
info, err := object.Stat()
|
|
if err != nil {
|
|
object.Close()
|
|
return StorageObject{}, err
|
|
}
|
|
return StorageObject{Key: key, Size: info.Size, ContentType: info.ContentType, ModTime: info.LastModified, Body: object}, nil
|
|
}
|
|
|
|
func (b *s3StorageBackend) Delete(ctx context.Context, key string) error {
|
|
return b.client.RemoveObject(ctx, b.cfg.Bucket, cleanObjectKey(key), minio.RemoveObjectOptions{})
|
|
}
|
|
|
|
func (b *s3StorageBackend) DeletePrefix(ctx context.Context, prefix string) error {
|
|
prefix = strings.TrimSuffix(cleanObjectKey(prefix), "/") + "/"
|
|
objects := b.client.ListObjects(ctx, b.cfg.Bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: true})
|
|
for object := range objects {
|
|
if object.Err != nil {
|
|
return object.Err
|
|
}
|
|
if err := b.Delete(ctx, object.Key); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b *s3StorageBackend) Usage(ctx context.Context) (int64, error) {
|
|
var total int64
|
|
for object := range b.client.ListObjects(ctx, b.cfg.Bucket, minio.ListObjectsOptions{Recursive: true}) {
|
|
if object.Err != nil {
|
|
return 0, object.Err
|
|
}
|
|
total += object.Size
|
|
}
|
|
return total, nil
|
|
}
|
|
|
|
func (b *s3StorageBackend) Test(ctx context.Context) error {
|
|
exists, err := b.client.BucketExists(ctx, b.cfg.Bucket)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !exists {
|
|
return fmt.Errorf("bucket %q does not exist", b.cfg.Bucket)
|
|
}
|
|
key := ".warpbox-storage-test-" + randomID(6)
|
|
if err := b.Put(ctx, key, bytes.NewReader([]byte("ok")), 2, "text/plain"); err != nil {
|
|
return err
|
|
}
|
|
return b.Delete(ctx, key)
|
|
}
|
|
|
|
type sftpStorageBackend struct {
|
|
cfg StorageBackendConfig
|
|
}
|
|
|
|
func (b sftpStorageBackend) ID() string { return b.cfg.ID }
|
|
func (b sftpStorageBackend) Type() string { return StorageBackendSFTP }
|
|
|
|
func (b sftpStorageBackend) Put(ctx context.Context, key string, body io.Reader, _ int64, _ string) error {
|
|
client, closer, err := b.client()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer closer()
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
remotePath := b.remotePath(key)
|
|
if err := client.MkdirAll(path.Dir(remotePath)); err != nil {
|
|
return err
|
|
}
|
|
target, err := client.OpenFile(remotePath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer target.Close()
|
|
_, err = io.Copy(target, body)
|
|
return err
|
|
}
|
|
|
|
func (b sftpStorageBackend) Get(ctx context.Context, key string) (StorageObject, error) {
|
|
client, closer, err := b.client()
|
|
if err != nil {
|
|
return StorageObject{}, err
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
closer()
|
|
return StorageObject{}, err
|
|
}
|
|
remotePath := b.remotePath(key)
|
|
source, err := client.Open(remotePath)
|
|
if err != nil {
|
|
closer()
|
|
return StorageObject{}, err
|
|
}
|
|
stat, err := source.Stat()
|
|
if err != nil {
|
|
source.Close()
|
|
closer()
|
|
return StorageObject{}, err
|
|
}
|
|
return StorageObject{Key: key, Size: stat.Size(), ModTime: stat.ModTime(), Body: closeWith(source, closer)}, nil
|
|
}
|
|
|
|
func (b sftpStorageBackend) Delete(ctx context.Context, key string) error {
|
|
client, closer, err := b.client()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer closer()
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
if err := client.Remove(b.remotePath(key)); err != nil && !os.IsNotExist(err) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b sftpStorageBackend) DeletePrefix(ctx context.Context, prefix string) error {
|
|
client, closer, err := b.client()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer closer()
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
remotePath := b.remotePath(prefix)
|
|
if err := client.RemoveDirectory(remotePath); err == nil || os.IsNotExist(err) {
|
|
return nil
|
|
}
|
|
walker := client.Walk(remotePath)
|
|
paths := make([]string, 0)
|
|
for walker.Step() {
|
|
if walker.Err() != nil {
|
|
return walker.Err()
|
|
}
|
|
paths = append(paths, walker.Path())
|
|
}
|
|
sort.Slice(paths, func(i, j int) bool { return len(paths[i]) > len(paths[j]) })
|
|
for _, item := range paths {
|
|
if err := client.Remove(item); err != nil {
|
|
_ = client.RemoveDirectory(item)
|
|
}
|
|
}
|
|
_ = client.RemoveDirectory(remotePath)
|
|
return nil
|
|
}
|
|
|
|
func (b sftpStorageBackend) Usage(ctx context.Context) (int64, error) {
|
|
client, closer, err := b.client()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer closer()
|
|
if err := ctx.Err(); err != nil {
|
|
return 0, err
|
|
}
|
|
var total int64
|
|
walker := client.Walk(cleanRemoteRoot(b.cfg.RemotePath))
|
|
for walker.Step() {
|
|
if walker.Err() != nil {
|
|
return 0, walker.Err()
|
|
}
|
|
info := walker.Stat()
|
|
if info != nil && !info.IsDir() {
|
|
total += info.Size()
|
|
}
|
|
}
|
|
return total, nil
|
|
}
|
|
|
|
func (b sftpStorageBackend) Test(ctx context.Context) error {
|
|
key := ".warpbox-storage-test-" + randomID(6)
|
|
if err := b.Put(ctx, key, strings.NewReader("ok"), 2, "text/plain"); err != nil {
|
|
return err
|
|
}
|
|
return b.Delete(ctx, key)
|
|
}
|
|
|
|
func (b sftpStorageBackend) client() (*sftp.Client, func(), error) {
|
|
auth := make([]ssh.AuthMethod, 0, 2)
|
|
if b.cfg.PrivateKey != "" {
|
|
signer, err := ssh.ParsePrivateKey([]byte(b.cfg.PrivateKey))
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
auth = append(auth, ssh.PublicKeys(signer))
|
|
}
|
|
if b.cfg.Password != "" {
|
|
auth = append(auth, ssh.Password(b.cfg.Password))
|
|
}
|
|
if len(auth) == 0 {
|
|
return nil, nil, fmt.Errorf("sftp password or private key is required")
|
|
}
|
|
hostKeyCallback, err := b.hostKeyCallback()
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
sshClient, err := ssh.Dial("tcp", b.cfg.Host+":"+strconv.Itoa(b.cfg.Port), &ssh.ClientConfig{
|
|
User: b.cfg.Username,
|
|
Auth: auth,
|
|
HostKeyCallback: hostKeyCallback,
|
|
Timeout: 15 * time.Second,
|
|
})
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
client, err := sftp.NewClient(sshClient)
|
|
if err != nil {
|
|
sshClient.Close()
|
|
return nil, nil, err
|
|
}
|
|
return client, func() {
|
|
client.Close()
|
|
sshClient.Close()
|
|
}, nil
|
|
}
|
|
|
|
func (b sftpStorageBackend) hostKeyCallback() (ssh.HostKeyCallback, error) {
|
|
if strings.TrimSpace(b.cfg.HostKey) == "" {
|
|
return ssh.InsecureIgnoreHostKey(), nil
|
|
}
|
|
key, _, _, _, err := ssh.ParseAuthorizedKey([]byte(strings.TrimSpace(b.cfg.HostKey)))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid sftp host public key: %w", err)
|
|
}
|
|
return ssh.FixedHostKey(key), nil
|
|
}
|
|
|
|
func (b sftpStorageBackend) remotePath(key string) string {
|
|
return path.Join(cleanRemoteRoot(b.cfg.RemotePath), cleanObjectKey(key))
|
|
}
|
|
|
|
type joinedReadCloser struct {
|
|
io.ReadCloser
|
|
close func()
|
|
}
|
|
|
|
func closeWith(source io.ReadCloser, close func()) io.ReadCloser {
|
|
return joinedReadCloser{ReadCloser: source, close: close}
|
|
}
|
|
|
|
func (c joinedReadCloser) Close() error {
|
|
err := c.ReadCloser.Close()
|
|
c.close()
|
|
return err
|
|
}
|
|
|
|
type smbStorageBackend struct {
|
|
cfg StorageBackendConfig
|
|
}
|
|
|
|
func (b smbStorageBackend) ID() string { return b.cfg.ID }
|
|
func (b smbStorageBackend) Type() string { return StorageBackendSMB }
|
|
|
|
func (b smbStorageBackend) Put(ctx context.Context, key string, body io.Reader, _ int64, _ string) error {
|
|
share, closer, err := b.share()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer closer()
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
remotePath := b.remotePath(key)
|
|
if err := share.MkdirAll(path.Dir(remotePath), 0o755); err != nil {
|
|
return err
|
|
}
|
|
target, err := share.OpenFile(remotePath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer target.Close()
|
|
_, err = io.Copy(target, body)
|
|
return err
|
|
}
|
|
|
|
func (b smbStorageBackend) Get(ctx context.Context, key string) (StorageObject, error) {
|
|
share, closer, err := b.share()
|
|
if err != nil {
|
|
return StorageObject{}, err
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
closer()
|
|
return StorageObject{}, err
|
|
}
|
|
source, err := share.Open(b.remotePath(key))
|
|
if err != nil {
|
|
closer()
|
|
return StorageObject{}, err
|
|
}
|
|
stat, err := source.Stat()
|
|
if err != nil {
|
|
source.Close()
|
|
closer()
|
|
return StorageObject{}, err
|
|
}
|
|
return StorageObject{Key: key, Size: stat.Size(), ModTime: stat.ModTime(), Body: closeWith(source, closer)}, nil
|
|
}
|
|
|
|
func (b smbStorageBackend) Delete(ctx context.Context, key string) error {
|
|
share, closer, err := b.share()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer closer()
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
if err := share.Remove(b.remotePath(key)); err != nil && !os.IsNotExist(err) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b smbStorageBackend) DeletePrefix(ctx context.Context, prefix string) error {
|
|
share, closer, err := b.share()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer closer()
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
err = share.RemoveAll(b.remotePath(prefix))
|
|
if err != nil && !os.IsNotExist(err) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b smbStorageBackend) Usage(ctx context.Context) (int64, error) {
|
|
share, closer, err := b.share()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer closer()
|
|
if err := ctx.Err(); err != nil {
|
|
return 0, err
|
|
}
|
|
return smbUsage(share, cleanRemoteRoot(b.cfg.RemotePath))
|
|
}
|
|
|
|
func (b smbStorageBackend) Test(ctx context.Context) error {
|
|
key := ".warpbox-storage-test-" + randomID(6)
|
|
if err := b.Put(ctx, key, strings.NewReader("ok"), 2, "text/plain"); err != nil {
|
|
return err
|
|
}
|
|
return b.Delete(ctx, key)
|
|
}
|
|
|
|
func (b smbStorageBackend) share() (*smb2.Share, func(), error) {
|
|
conn, err := net.DialTimeout("tcp", b.cfg.Host+":"+strconv.Itoa(b.cfg.Port), 15*time.Second)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
dialer := &smb2.Dialer{
|
|
Initiator: &smb2.NTLMInitiator{
|
|
User: b.cfg.Username,
|
|
Password: b.cfg.Password,
|
|
Domain: b.cfg.Domain,
|
|
},
|
|
}
|
|
session, err := dialer.Dial(conn)
|
|
if err != nil {
|
|
conn.Close()
|
|
return nil, nil, err
|
|
}
|
|
share, err := session.Mount(b.cfg.Share)
|
|
if err != nil {
|
|
session.Logoff()
|
|
conn.Close()
|
|
return nil, nil, err
|
|
}
|
|
return share, func() {
|
|
share.Umount()
|
|
session.Logoff()
|
|
conn.Close()
|
|
}, nil
|
|
}
|
|
|
|
func (b smbStorageBackend) remotePath(key string) string {
|
|
return strings.TrimPrefix(path.Join(cleanRemoteRoot(b.cfg.RemotePath), cleanObjectKey(key)), "/")
|
|
}
|
|
|
|
func smbUsage(share *smb2.Share, root string) (int64, error) {
|
|
root = strings.TrimPrefix(root, "/")
|
|
entries, err := share.ReadDir(root)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return 0, nil
|
|
}
|
|
return 0, err
|
|
}
|
|
var total int64
|
|
for _, entry := range entries {
|
|
item := path.Join(root, entry.Name())
|
|
if entry.IsDir() {
|
|
size, err := smbUsage(share, item)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
total += size
|
|
continue
|
|
}
|
|
total += entry.Size()
|
|
}
|
|
return total, nil
|
|
}
|
|
|
|
type webDAVStorageBackend struct {
|
|
cfg StorageBackendConfig
|
|
client *http.Client
|
|
}
|
|
|
|
func (b webDAVStorageBackend) ID() string { return b.cfg.ID }
|
|
func (b webDAVStorageBackend) Type() string { return StorageBackendWebDAV }
|
|
|
|
func (b webDAVStorageBackend) Put(ctx context.Context, key string, body io.Reader, _ int64, contentType string) error {
|
|
if err := b.mkcolParents(ctx, key); err != nil {
|
|
return err
|
|
}
|
|
request, err := b.request(ctx, http.MethodPut, key, body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if contentType != "" {
|
|
request.Header.Set("Content-Type", contentType)
|
|
}
|
|
response, err := b.client.Do(request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
|
return fmt.Errorf("webdav put failed: %s", response.Status)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b webDAVStorageBackend) Get(ctx context.Context, key string) (StorageObject, error) {
|
|
request, err := b.request(ctx, http.MethodGet, key, nil)
|
|
if err != nil {
|
|
return StorageObject{}, err
|
|
}
|
|
response, err := b.client.Do(request)
|
|
if err != nil {
|
|
return StorageObject{}, err
|
|
}
|
|
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
|
response.Body.Close()
|
|
return StorageObject{}, fmt.Errorf("webdav get failed: %s", response.Status)
|
|
}
|
|
modTime, _ := time.Parse(http.TimeFormat, response.Header.Get("Last-Modified"))
|
|
return StorageObject{Key: key, Size: response.ContentLength, ContentType: response.Header.Get("Content-Type"), ModTime: modTime, Body: response.Body}, nil
|
|
}
|
|
|
|
func (b webDAVStorageBackend) Delete(ctx context.Context, key string) error {
|
|
return b.deletePath(ctx, key)
|
|
}
|
|
|
|
func (b webDAVStorageBackend) DeletePrefix(ctx context.Context, prefix string) error {
|
|
return b.deletePath(ctx, strings.TrimSuffix(prefix, "/")+"/")
|
|
}
|
|
|
|
func (b webDAVStorageBackend) Usage(ctx context.Context) (int64, error) {
|
|
request, err := b.request(ctx, "PROPFIND", "", nil)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
request.Header.Set("Depth", "infinity")
|
|
request.Header.Set("Content-Type", "application/xml")
|
|
response, err := b.client.Do(request)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
|
return 0, fmt.Errorf("webdav usage failed: %s", response.Status)
|
|
}
|
|
var multi webDAVMultiStatus
|
|
if err := xml.NewDecoder(response.Body).Decode(&multi); err != nil {
|
|
return 0, err
|
|
}
|
|
var total int64
|
|
for _, item := range multi.Responses {
|
|
if item.PropStat.Prop.ResourceType.Collection != nil {
|
|
continue
|
|
}
|
|
total += item.PropStat.Prop.ContentLength
|
|
}
|
|
return total, nil
|
|
}
|
|
|
|
func (b webDAVStorageBackend) Test(ctx context.Context) error {
|
|
key := ".warpbox-storage-test-" + randomID(6)
|
|
if err := b.Put(ctx, key, strings.NewReader("ok"), 2, "text/plain"); err != nil {
|
|
return err
|
|
}
|
|
return b.Delete(ctx, key)
|
|
}
|
|
|
|
func (b webDAVStorageBackend) deletePath(ctx context.Context, key string) error {
|
|
request, err := b.request(ctx, http.MethodDelete, key, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
response, err := b.client.Do(request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode == http.StatusNotFound {
|
|
return nil
|
|
}
|
|
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
|
return fmt.Errorf("webdav delete failed: %s", response.Status)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b webDAVStorageBackend) mkcolParents(ctx context.Context, key string) error {
|
|
dir := path.Dir(cleanObjectKey(key))
|
|
if dir == "." || dir == "/" {
|
|
return nil
|
|
}
|
|
parts := strings.Split(strings.Trim(dir, "/"), "/")
|
|
current := ""
|
|
for _, part := range parts {
|
|
current = path.Join(current, part)
|
|
request, err := b.request(ctx, "MKCOL", strings.TrimSuffix(current, "/")+"/", nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
response, err := b.client.Do(request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
response.Body.Close()
|
|
if response.StatusCode != http.StatusCreated && response.StatusCode != http.StatusMethodNotAllowed && response.StatusCode != http.StatusConflict {
|
|
return fmt.Errorf("webdav mkcol failed: %s", response.Status)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b webDAVStorageBackend) request(ctx context.Context, method, key string, body io.Reader) (*http.Request, error) {
|
|
endpoint := strings.TrimRight(b.cfg.Endpoint, "/")
|
|
if endpoint == "" {
|
|
return nil, fmt.Errorf("webdav url is required")
|
|
}
|
|
remote := path.Join(cleanRemoteRoot(b.cfg.RemotePath), cleanObjectKey(key))
|
|
if strings.HasSuffix(key, "/") && !strings.HasSuffix(remote, "/") {
|
|
remote += "/"
|
|
}
|
|
target := endpoint + "/" + strings.TrimLeft(remote, "/")
|
|
request, err := http.NewRequestWithContext(ctx, method, target, body)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if b.cfg.Username != "" || b.cfg.Password != "" {
|
|
request.SetBasicAuth(b.cfg.Username, b.cfg.Password)
|
|
}
|
|
return request, nil
|
|
}
|
|
|
|
type webDAVMultiStatus struct {
|
|
Responses []webDAVResponse `xml:"response"`
|
|
}
|
|
|
|
type webDAVResponse struct {
|
|
PropStat webDAVPropStat `xml:"propstat"`
|
|
}
|
|
|
|
type webDAVPropStat struct {
|
|
Prop webDAVProp `xml:"prop"`
|
|
}
|
|
|
|
type webDAVProp struct {
|
|
ContentLength int64 `xml:"getcontentlength"`
|
|
ResourceType webDAVResourceType `xml:"resourcetype"`
|
|
}
|
|
|
|
type webDAVResourceType struct {
|
|
Collection *struct{} `xml:"collection"`
|
|
}
|
|
|
|
func s3BucketLookup(pathStyle bool) minio.BucketLookupType {
|
|
if pathStyle {
|
|
return minio.BucketLookupPath
|
|
}
|
|
return minio.BucketLookupAuto
|
|
}
|
|
|
|
func normalizeS3Endpoint(endpoint string) string {
|
|
endpoint = strings.TrimSpace(endpoint)
|
|
if parsed, err := url.Parse(endpoint); err == nil && parsed.Host != "" {
|
|
return parsed.Host
|
|
}
|
|
return strings.TrimPrefix(strings.TrimPrefix(endpoint, "https://"), "http://")
|
|
}
|
|
|
|
func normalizeStorageProvider(provider string) string {
|
|
switch strings.TrimSpace(provider) {
|
|
case StorageProviderContabo:
|
|
return StorageProviderContabo
|
|
case StorageProviderSFTP:
|
|
return StorageProviderSFTP
|
|
case StorageProviderSMB:
|
|
return StorageProviderSMB
|
|
case StorageProviderWebDAV:
|
|
return StorageProviderWebDAV
|
|
default:
|
|
return StorageProviderS3
|
|
}
|
|
}
|
|
|
|
func cleanObjectKey(key string) string {
|
|
return strings.TrimPrefix(filepath.ToSlash(filepath.Clean(strings.TrimPrefix(key, "/"))), "./")
|
|
}
|
|
|
|
func cleanRemoteRoot(value string) string {
|
|
value = strings.TrimSpace(value)
|
|
if value == "" {
|
|
return "."
|
|
}
|
|
cleaned := path.Clean(strings.ReplaceAll(value, "\\", "/"))
|
|
if cleaned == "/" {
|
|
return "/"
|
|
}
|
|
return strings.TrimSuffix(cleaned, "/")
|
|
}
|