rclone/backend/mailru/mailru.go
Roman Kredentser 55ad1354b6 link: Add --expire and --unlink flags
This adds expire and unlink fields to the PublicLink interface.

This fixes up the affected backends and removes unlink parameters
where they are present.
2020-06-05 14:51:05 +01:00

2400 lines
63 KiB
Go

package mailru
import (
"bytes"
"context"
"fmt"
gohash "hash"
"io"
"path"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"time"
"encoding/hex"
"encoding/json"
"io/ioutil"
"net/http"
"net/url"
"github.com/rclone/rclone/backend/mailru/api"
"github.com/rclone/rclone/backend/mailru/mrhash"
"github.com/rclone/rclone/fs"
"github.com/rclone/rclone/fs/config"
"github.com/rclone/rclone/fs/config/configmap"
"github.com/rclone/rclone/fs/config/configstruct"
"github.com/rclone/rclone/fs/config/obscure"
"github.com/rclone/rclone/fs/fserrors"
"github.com/rclone/rclone/fs/fshttp"
"github.com/rclone/rclone/fs/hash"
"github.com/rclone/rclone/fs/object"
"github.com/rclone/rclone/fs/operations"
"github.com/rclone/rclone/lib/encoder"
"github.com/rclone/rclone/lib/oauthutil"
"github.com/rclone/rclone/lib/pacer"
"github.com/rclone/rclone/lib/rest"
"github.com/pkg/errors"
"golang.org/x/oauth2"
)
// Global constants
const (
minSleepPacer = 10 * time.Millisecond
maxSleepPacer = 2 * time.Second
decayConstPacer = 2 // bigger for slower decay, exponential
metaExpirySec = 20 * 60 // meta server expiration time
serverExpirySec = 3 * 60 // download server expiration time
shardExpirySec = 30 * 60 // upload server expiration time
maxServerLocks = 4 // maximum number of locks per single download server
maxInt32 = 2147483647 // used as limit in directory list request
speedupMinSize = 512 // speedup is not optimal if data is smaller than average packet
)
// Global errors
var (
ErrorDirAlreadyExists = errors.New("directory already exists")
ErrorDirSourceNotExists = errors.New("directory source does not exist")
ErrorInvalidName = errors.New("invalid characters in object name")
// MrHashType is the hash.Type for Mailru
MrHashType hash.Type
)
// Description of how to authorize
var oauthConfig = &oauth2.Config{
ClientID: api.OAuthClientID,
ClientSecret: "",
Endpoint: oauth2.Endpoint{
AuthURL: api.OAuthURL,
TokenURL: api.OAuthURL,
AuthStyle: oauth2.AuthStyleInParams,
},
}
// Register with Fs
func init() {
MrHashType = hash.RegisterHash("MailruHash", 40, mrhash.New)
fs.Register(&fs.RegInfo{
Name: "mailru",
Description: "Mail.ru Cloud",
NewFs: NewFs,
Options: []fs.Option{{
Name: "user",
Help: "User name (usually email)",
Required: true,
}, {
Name: "pass",
Help: "Password",
Required: true,
IsPassword: true,
}, {
Name: "speedup_enable",
Default: true,
Advanced: false,
Help: `Skip full upload if there is another file with same data hash.
This feature is called "speedup" or "put by hash". It is especially efficient
in case of generally available files like popular books, video or audio clips,
because files are searched by hash in all accounts of all mailru users.
Please note that rclone may need local memory and disk space to calculate
content hash in advance and decide whether full upload is required.
Also, if rclone does not know file size in advance (e.g. in case of
streaming or partial uploads), it will not even try this optimization.`,
Examples: []fs.OptionExample{{
Value: "true",
Help: "Enable",
}, {
Value: "false",
Help: "Disable",
}},
}, {
Name: "speedup_file_patterns",
Default: "*.mkv,*.avi,*.mp4,*.mp3,*.zip,*.gz,*.rar,*.pdf",
Advanced: true,
Help: `Comma separated list of file name patterns eligible for speedup (put by hash).
Patterns are case insensitive and can contain '*' or '?' meta characters.`,
Examples: []fs.OptionExample{{
Value: "",
Help: "Empty list completely disables speedup (put by hash).",
}, {
Value: "*",
Help: "All files will be attempted for speedup.",
}, {
Value: "*.mkv,*.avi,*.mp4,*.mp3",
Help: "Only common audio/video files will be tried for put by hash.",
}, {
Value: "*.zip,*.gz,*.rar,*.pdf",
Help: "Only common archives or PDF books will be tried for speedup.",
}},
}, {
Name: "speedup_max_disk",
Default: fs.SizeSuffix(3 * 1024 * 1024 * 1024),
Advanced: true,
Help: `This option allows you to disable speedup (put by hash) for large files
(because preliminary hashing can exhaust you RAM or disk space)`,
Examples: []fs.OptionExample{{
Value: "0",
Help: "Completely disable speedup (put by hash).",
}, {
Value: "1G",
Help: "Files larger than 1Gb will be uploaded directly.",
}, {
Value: "3G",
Help: "Choose this option if you have less than 3Gb free on local disk.",
}},
}, {
Name: "speedup_max_memory",
Default: fs.SizeSuffix(32 * 1024 * 1024),
Advanced: true,
Help: `Files larger than the size given below will always be hashed on disk.`,
Examples: []fs.OptionExample{{
Value: "0",
Help: "Preliminary hashing will always be done in a temporary disk location.",
}, {
Value: "32M",
Help: "Do not dedicate more than 32Mb RAM for preliminary hashing.",
}, {
Value: "256M",
Help: "You have at most 256Mb RAM free for hash calculations.",
}},
}, {
Name: "check_hash",
Default: true,
Advanced: true,
Help: "What should copy do if file checksum is mismatched or invalid",
Examples: []fs.OptionExample{{
Value: "true",
Help: "Fail with error.",
}, {
Value: "false",
Help: "Ignore and continue.",
}},
}, {
Name: "user_agent",
Default: "",
Advanced: true,
Hide: fs.OptionHideBoth,
Help: `HTTP user agent used internally by client.
Defaults to "rclone/VERSION" or "--user-agent" provided on command line.`,
}, {
Name: "quirks",
Default: "",
Advanced: true,
Hide: fs.OptionHideBoth,
Help: `Comma separated list of internal maintenance flags.
This option must not be used by an ordinary user. It is intended only to
facilitate remote troubleshooting of backend issues. Strict meaning of
flags is not documented and not guaranteed to persist between releases.
Quirks will be removed when the backend grows stable.
Supported quirks: atomicmkdir binlist gzip insecure retry400`,
}, {
Name: config.ConfigEncoding,
Help: config.ConfigEncodingHelp,
Advanced: true,
// Encode invalid UTF-8 bytes as json doesn't handle them properly.
Default: (encoder.Display |
encoder.EncodeWin | // :?"*<>|
encoder.EncodeBackSlash |
encoder.EncodeInvalidUtf8),
}},
})
}
// Options defines the configuration for this backend
type Options struct {
Username string `config:"user"`
Password string `config:"pass"`
UserAgent string `config:"user_agent"`
CheckHash bool `config:"check_hash"`
SpeedupEnable bool `config:"speedup_enable"`
SpeedupPatterns string `config:"speedup_file_patterns"`
SpeedupMaxDisk fs.SizeSuffix `config:"speedup_max_disk"`
SpeedupMaxMem fs.SizeSuffix `config:"speedup_max_memory"`
Quirks string `config:"quirks"`
Enc encoder.MultiEncoder `config:"encoding"`
}
// retryErrorCodes is a slice of error codes that we will retry
var retryErrorCodes = []int{
429, // Too Many Requests.
500, // Internal Server Error
502, // Bad Gateway
503, // Service Unavailable
504, // Gateway Timeout
509, // Bandwidth Limit Exceeded
}
// shouldRetry returns a boolean as to whether this response and err
// deserve to be retried. It returns the err as a convenience.
// Retries password authorization (once) in a special case of access denied.
func shouldRetry(res *http.Response, err error, f *Fs, opts *rest.Opts) (bool, error) {
if res != nil && res.StatusCode == 403 && f.opt.Password != "" && !f.passFailed {
reAuthErr := f.reAuthorize(opts, err)
return reAuthErr == nil, err // return an original error
}
if res != nil && res.StatusCode == 400 && f.quirks.retry400 {
return true, err
}
return fserrors.ShouldRetry(err) || fserrors.ShouldRetryHTTP(res, retryErrorCodes), err
}
// errorHandler parses a non 2xx error response into an error
func errorHandler(res *http.Response) (err error) {
data, err := rest.ReadBody(res)
if err != nil {
return err
}
fileError := &api.FileErrorResponse{}
err = json.NewDecoder(bytes.NewReader(data)).Decode(fileError)
if err == nil {
fileError.Message = fileError.Body.Home.Error
return fileError
}
serverError := &api.ServerErrorResponse{}
err = json.NewDecoder(bytes.NewReader(data)).Decode(serverError)
if err == nil {
return serverError
}
serverError.Message = string(data)
if serverError.Message == "" || strings.HasPrefix(serverError.Message, "{") {
// Replace empty or JSON response with a human readable text.
serverError.Message = res.Status
}
serverError.Status = res.StatusCode
return serverError
}
// Fs represents a remote mail.ru
type Fs struct {
name string
root string // root path
opt Options // parsed options
speedupGlobs []string // list of file name patterns eligible for speedup
speedupAny bool // true if all file names are aligible for speedup
features *fs.Features // optional features
srv *rest.Client // REST API client
cli *http.Client // underlying HTTP client (for authorize)
m configmap.Mapper // config reader (for authorize)
source oauth2.TokenSource // OAuth token refresher
pacer *fs.Pacer // pacer for API calls
metaMu sync.Mutex // lock for meta server switcher
metaURL string // URL of meta server
metaExpiry time.Time // time to refresh meta server
shardMu sync.Mutex // lock for upload shard switcher
shardURL string // URL of upload shard
shardExpiry time.Time // time to refresh upload shard
fileServers serverPool // file server dispatcher
authMu sync.Mutex // mutex for authorize()
passFailed bool // true if authorize() failed after 403
quirks quirks // internal maintenance flags
}
// NewFs constructs an Fs from the path, container:path
func NewFs(name, root string, m configmap.Mapper) (fs.Fs, error) {
// fs.Debugf(nil, ">>> NewFs %q %q", name, root)
ctx := context.Background() // Note: NewFs does not pass context!
// Parse config into Options struct
opt := new(Options)
err := configstruct.Set(m, opt)
if err != nil {
return nil, err
}
if opt.Password != "" {
opt.Password = obscure.MustReveal(opt.Password)
}
// Trailing slash signals us to optimize out one file check
rootIsDir := strings.HasSuffix(root, "/")
// However the f.root string should not have leading or trailing slashes
root = strings.Trim(root, "/")
f := &Fs{
name: name,
root: root,
opt: *opt,
m: m,
}
if err := f.parseSpeedupPatterns(opt.SpeedupPatterns); err != nil {
return nil, err
}
f.quirks.parseQuirks(opt.Quirks)
f.pacer = fs.NewPacer(pacer.NewDefault(pacer.MinSleep(minSleepPacer), pacer.MaxSleep(maxSleepPacer), pacer.DecayConstant(decayConstPacer)))
f.features = (&fs.Features{
CaseInsensitive: true,
CanHaveEmptyDirectories: true,
// Can copy/move across mailru configs (almost, thus true here), but
// only when they share common account (this is checked in Copy/Move).
ServerSideAcrossConfigs: true,
}).Fill(f)
// Override few config settings and create a client
clientConfig := *fs.Config
if opt.UserAgent != "" {
clientConfig.UserAgent = opt.UserAgent
}
clientConfig.NoGzip = !f.quirks.gzip // Send not "Accept-Encoding: gzip" like official client
f.cli = fshttp.NewClient(&clientConfig)
f.srv = rest.NewClient(f.cli)
f.srv.SetRoot(api.APIServerURL)
f.srv.SetHeader("Accept", "*/*") // Send "Accept: */*" with every request like official client
f.srv.SetErrorHandler(errorHandler)
if f.quirks.insecure {
transport := f.cli.Transport.(*fshttp.Transport).Transport
transport.TLSClientConfig.InsecureSkipVerify = true
transport.ProxyConnectHeader = http.Header{"User-Agent": {clientConfig.UserAgent}}
}
if err = f.authorize(ctx, false); err != nil {
return nil, err
}
f.fileServers = serverPool{
pool: make(pendingServerMap),
fs: f,
path: "/d",
expirySec: serverExpirySec,
}
if !rootIsDir {
_, dirSize, err := f.readItemMetaData(ctx, f.root)
rootIsDir = (dirSize >= 0)
// Ignore non-existing item and other errors
if err == nil && !rootIsDir {
root = path.Dir(f.root)
if root == "." {
root = ""
}
f.root = root
// Return fs that points to the parent and signal rclone to do filtering
return f, fs.ErrorIsFile
}
}
return f, nil
}
// Internal maintenance flags (to be removed when the backend matures).
// Primarily intended to facilitate remote support and troubleshooting.
type quirks struct {
gzip bool
insecure bool
binlist bool
atomicmkdir bool
retry400 bool
}
func (q *quirks) parseQuirks(option string) {
for _, flag := range strings.Split(option, ",") {
switch strings.ToLower(strings.TrimSpace(flag)) {
case "gzip":
// This backend mimics the official client which never sends the
// "Accept-Encoding: gzip" header. However, enabling compression
// might be good for performance.
// Use this quirk to investigate the performance impact.
// Remove this quirk if performance does not improve.
q.gzip = true
case "insecure":
// The mailru disk-o protocol is not documented. To compare HTTP
// stream against the official client one can use Telerik Fiddler,
// which introduces a self-signed certificate. This quirk forces
// the Go http layer to accept it.
// Remove this quirk when the backend reaches maturity.
q.insecure = true
case "binlist":
// The official client sometimes uses a so called "bin" protocol,
// implemented in the listBin file system method below. This method
// is generally faster than non-recursive listM1 but results in
// sporadic deserialization failures if total size of tree data
// approaches 8Kb (?). The recursive method is normally disabled.
// This quirk can be used to enable it for further investigation.
// Remove this quirk when the "bin" protocol support is complete.
q.binlist = true
case "atomicmkdir":
// At the moment rclone requires Mkdir to return success if the
// directory already exists. However, such programs as borgbackup
// or restic use mkdir as a locking primitive and depend on its
// atomicity. This quirk is a workaround. It can be removed
// when the above issue is investigated.
q.atomicmkdir = true
case "retry400":
// This quirk will help in troubleshooting a very rare "Error 400"
// issue. It can be removed if the problem does not show up
// for a year or so. See the below issue:
// https://github.com/ivandeex/rclone/issues/14
q.retry400 = true
default:
// Just ignore all unknown flags
}
}
}
// Note: authorize() is not safe for concurrent access as it updates token source
func (f *Fs) authorize(ctx context.Context, force bool) (err error) {
var t *oauth2.Token
if !force {
t, err = oauthutil.GetToken(f.name, f.m)
}
if err != nil || !tokenIsValid(t) {
fs.Infof(f, "Valid token not found, authorizing.")
ctx := oauthutil.Context(f.cli)
t, err = oauthConfig.PasswordCredentialsToken(ctx, f.opt.Username, f.opt.Password)
}
if err == nil && !tokenIsValid(t) {
err = errors.New("Invalid token")
}
if err != nil {
return errors.Wrap(err, "Failed to authorize")
}
if err = oauthutil.PutToken(f.name, f.m, t, false); err != nil {
return err
}
// Mailru API server expects access token not in the request header but
// in the URL query string, so we must use a bare token source rather than
// client provided by oauthutil.
//
// WARNING: direct use of the returned token source triggers a bug in the
// `(*token != *ts.token)` comparison in oauthutil.TokenSource.Token()
// crashing with panic `comparing uncomparable type map[string]interface{}`
// As a workaround, mimic oauth2.NewClient() wrapping token source in
// oauth2.ReuseTokenSource
_, ts, err := oauthutil.NewClientWithBaseClient(f.name, f.m, oauthConfig, f.cli)
if err == nil {
f.source = oauth2.ReuseTokenSource(nil, ts)
}
return err
}
func tokenIsValid(t *oauth2.Token) bool {
return t.Valid() && t.RefreshToken != "" && t.Type() == "Bearer"
}
// reAuthorize is called after getting 403 (access denied) from the server.
// It handles the case when user has changed password since a previous
// rclone invocation and obtains a new access token, if needed.
func (f *Fs) reAuthorize(opts *rest.Opts, origErr error) error {
// lock and recheck the flag to ensure authorize() is attempted only once
f.authMu.Lock()
defer f.authMu.Unlock()
if f.passFailed {
return origErr
}
ctx := context.Background() // Note: reAuthorize is called by ShouldRetry, no context!
fs.Debugf(f, "re-authorize with new password")
if err := f.authorize(ctx, true); err != nil {
f.passFailed = true
return err
}
// obtain new token, if needed
tokenParameter := ""
if opts != nil && opts.Parameters.Get("token") != "" {
tokenParameter = "token"
}
if opts != nil && opts.Parameters.Get("access_token") != "" {
tokenParameter = "access_token"
}
if tokenParameter != "" {
token, err := f.accessToken()
if err != nil {
f.passFailed = true
return err
}
opts.Parameters.Set(tokenParameter, token)
}
return nil
}
// accessToken() returns OAuth token and possibly refreshes it
func (f *Fs) accessToken() (string, error) {
token, err := f.source.Token()
if err != nil {
return "", errors.Wrap(err, "cannot refresh access token")
}
return token.AccessToken, nil
}
// absPath converts root-relative remote to absolute home path
func (f *Fs) absPath(remote string) string {
return path.Join("/", f.root, remote)
}
// relPath converts absolute home path to root-relative remote
// Note that f.root can not have leading and trailing slashes
func (f *Fs) relPath(absPath string) (string, error) {
target := strings.Trim(absPath, "/")
if f.root == "" {
return target, nil
}
if target == f.root {
return "", nil
}
if strings.HasPrefix(target+"/", f.root+"/") {
return target[len(f.root)+1:], nil
}
return "", fmt.Errorf("path %q should be under %q", absPath, f.root)
}
// metaServer ...
func (f *Fs) metaServer(ctx context.Context) (string, error) {
f.metaMu.Lock()
defer f.metaMu.Unlock()
if f.metaURL != "" && time.Now().Before(f.metaExpiry) {
return f.metaURL, nil
}
opts := rest.Opts{
RootURL: api.DispatchServerURL,
Method: "GET",
Path: "/m",
}
var (
res *http.Response
url string
err error
)
err = f.pacer.Call(func() (bool, error) {
res, err = f.srv.Call(ctx, &opts)
if err == nil {
url, err = readBodyWord(res)
}
return fserrors.ShouldRetry(err), err
})
if err != nil {
closeBody(res)
return "", err
}
f.metaURL = url
f.metaExpiry = time.Now().Add(metaExpirySec * time.Second)
fs.Debugf(f, "new meta server: %s", f.metaURL)
return f.metaURL, nil
}
// readBodyWord reads the single line response to completion
// and extracts the first word from the first line.
func readBodyWord(res *http.Response) (word string, err error) {
var body []byte
body, err = rest.ReadBody(res)
if err == nil {
line := strings.Trim(string(body), " \r\n")
word = strings.Split(line, " ")[0]
}
if word == "" {
return "", errors.New("Empty reply from dispatcher")
}
return word, nil
}
// readItemMetaData returns a file/directory info at given full path
// If it can't be found it fails with fs.ErrorObjectNotFound
// For the return value `dirSize` please see Fs.itemToEntry()
func (f *Fs) readItemMetaData(ctx context.Context, path string) (entry fs.DirEntry, dirSize int, err error) {
token, err := f.accessToken()
if err != nil {
return nil, -1, err
}
opts := rest.Opts{
Method: "GET",
Path: "/api/m1/file",
Parameters: url.Values{
"access_token": {token},
"home": {f.opt.Enc.FromStandardPath(path)},
"offset": {"0"},
"limit": {strconv.Itoa(maxInt32)},
},
}
var info api.ItemInfoResponse
err = f.pacer.Call(func() (bool, error) {
res, err := f.srv.CallJSON(ctx, &opts, nil, &info)
return shouldRetry(res, err, f, &opts)
})
if err != nil {
if apiErr, ok := err.(*api.FileErrorResponse); ok {
switch apiErr.Status {
case 404:
err = fs.ErrorObjectNotFound
case 400:
fs.Debugf(f, "object %q status %d (%s)", path, apiErr.Status, apiErr.Message)
err = fs.ErrorObjectNotFound
}
}
return
}
entry, dirSize, err = f.itemToDirEntry(ctx, &info.Body)
return
}
// itemToEntry converts API item to rclone directory entry
// The dirSize return value is:
// <0 - for a file or in case of error
// =0 - for an empty directory
// >0 - for a non-empty directory
func (f *Fs) itemToDirEntry(ctx context.Context, item *api.ListItem) (entry fs.DirEntry, dirSize int, err error) {
remote, err := f.relPath(f.opt.Enc.ToStandardPath(item.Home))
if err != nil {
return nil, -1, err
}
switch item.Kind {
case "folder":
dir := fs.NewDir(remote, time.Unix(item.Mtime, 0)).SetSize(item.Size)
dirSize := item.Count.Files + item.Count.Folders
return dir, dirSize, nil
case "file":
binHash, err := mrhash.DecodeString(item.Hash)
if err != nil {
return nil, -1, err
}
file := &Object{
fs: f,
remote: remote,
hasMetaData: true,
size: item.Size,
mrHash: binHash,
modTime: time.Unix(item.Mtime, 0),
}
return file, -1, nil
default:
return nil, -1, fmt.Errorf("Unknown resource type %q", item.Kind)
}
}
// List the objects and directories in dir into entries.
// The entries can be returned in any order but should be for a complete directory.
// dir should be "" to list the root, and should not have trailing slashes.
// This should return ErrDirNotFound if the directory isn't found.
func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err error) {
// fs.Debugf(f, ">>> List: %q", dir)
if f.quirks.binlist {
entries, err = f.listBin(ctx, f.absPath(dir), 1)
} else {
entries, err = f.listM1(ctx, f.absPath(dir), 0, maxInt32)
}
if err == nil && fs.Config.LogLevel >= fs.LogLevelDebug {
names := []string{}
for _, entry := range entries {
names = append(names, entry.Remote())
}
sort.Strings(names)
// fs.Debugf(f, "List(%q): %v", dir, names)
}
return
}
// list using protocol "m1"
func (f *Fs) listM1(ctx context.Context, dirPath string, offset int, limit int) (entries fs.DirEntries, err error) {
token, err := f.accessToken()
if err != nil {
return nil, err
}
params := url.Values{}
params.Set("access_token", token)
params.Set("offset", strconv.Itoa(offset))
params.Set("limit", strconv.Itoa(limit))
data := url.Values{}
data.Set("home", f.opt.Enc.FromStandardPath(dirPath))
opts := rest.Opts{
Method: "POST",
Path: "/api/m1/folder",
Parameters: params,
Body: strings.NewReader(data.Encode()),
ContentType: api.BinContentType,
}
var (
info api.FolderInfoResponse
res *http.Response
)
err = f.pacer.Call(func() (bool, error) {
res, err = f.srv.CallJSON(ctx, &opts, nil, &info)
return shouldRetry(res, err, f, &opts)
})
if err != nil {
apiErr, ok := err.(*api.FileErrorResponse)
if ok && apiErr.Status == 404 {
return nil, fs.ErrorDirNotFound
}
return nil, err
}
if info.Body.Kind != "folder" {
return nil, fs.ErrorIsFile
}
for _, item := range info.Body.List {
entry, _, err := f.itemToDirEntry(ctx, &item)
if err == nil {
entries = append(entries, entry)
} else {
fs.Debugf(f, "Excluding path %q from list: %v", item.Home, err)
}
}
return entries, nil
}
// list using protocol "bin"
func (f *Fs) listBin(ctx context.Context, dirPath string, depth int) (entries fs.DirEntries, err error) {
options := api.ListOptDefaults
req := api.NewBinWriter()
req.WritePu16(api.OperationFolderList)
req.WriteString(f.opt.Enc.FromStandardPath(dirPath))
req.WritePu32(int64(depth))
req.WritePu32(int64(options))
req.WritePu32(0)
token, err := f.accessToken()
if err != nil {
return nil, err
}
metaURL, err := f.metaServer(ctx)
if err != nil {
return nil, err
}
opts := rest.Opts{
Method: "POST",
RootURL: metaURL,
Parameters: url.Values{
"client_id": {api.OAuthClientID},
"token": {token},
},
ContentType: api.BinContentType,
Body: req.Reader(),
}
var res *http.Response
err = f.pacer.Call(func() (bool, error) {
res, err = f.srv.Call(ctx, &opts)
return shouldRetry(res, err, f, &opts)
})
if err != nil {
closeBody(res)
return nil, err
}
r := api.NewBinReader(res.Body)
defer closeBody(res)
// read status
switch status := r.ReadByteAsInt(); status {
case api.ListResultOK:
// go on...
case api.ListResultNotExists:
return nil, fs.ErrorDirNotFound
default:
return nil, fmt.Errorf("directory list error %d", status)
}
t := &treeState{
f: f,
r: r,
options: options,
rootDir: parentDir(dirPath),
lastDir: "",
level: 0,
}
t.currDir = t.rootDir
// read revision
if err := t.revision.Read(r); err != nil {
return nil, err
}
// read space
if (options & api.ListOptTotalSpace) != 0 {
t.totalSpace = int64(r.ReadULong())
}
if (options & api.ListOptUsedSpace) != 0 {
t.usedSpace = int64(r.ReadULong())
}
t.fingerprint = r.ReadBytesByLength()
// deserialize
for {
entry, err := t.NextRecord()
if err != nil {
break
}
if entry != nil {
entries = append(entries, entry)
}
}
if err != nil && err != fs.ErrorListAborted {
fs.Debugf(f, "listBin failed at offset %d: %v", r.Count(), err)
return nil, err
}
return entries, nil
}
func (t *treeState) NextRecord() (fs.DirEntry, error) {
r := t.r
parseOp := r.ReadByteAsShort()
if r.Error() != nil {
return nil, r.Error()
}
switch parseOp {
case api.ListParseDone:
return nil, fs.ErrorListAborted
case api.ListParsePin:
if t.lastDir == "" {
return nil, errors.New("last folder is null")
}
t.currDir = t.lastDir
t.level++
return nil, nil
case api.ListParsePinUpper:
if t.currDir == t.rootDir {
return nil, nil
}
if t.level <= 0 {
return nil, errors.New("no parent folder")
}
t.currDir = parentDir(t.currDir)
t.level--
return nil, nil
case api.ListParseUnknown15:
skip := int(r.ReadPu32())
for i := 0; i < skip; i++ {
r.ReadPu32()
r.ReadPu32()
}
return nil, nil
case api.ListParseReadItem:
// get item (see below)
default:
return nil, fmt.Errorf("unknown parse operation %d", parseOp)
}
// get item
head := r.ReadIntSpl()
itemType := head & 3
if (head & 4096) != 0 {
t.dunnoNodeID = r.ReadNBytes(api.DunnoNodeIDLength)
}
name := t.f.opt.Enc.FromStandardPath(string(r.ReadBytesByLength()))
t.dunno1 = int(r.ReadULong())
t.dunno2 = 0
t.dunno3 = 0
if r.Error() != nil {
return nil, r.Error()
}
var (
modTime time.Time
size int64
binHash []byte
dirSize int64
isDir = true
)
switch itemType {
case api.ListItemMountPoint:
t.treeID = r.ReadNBytes(api.TreeIDLength)
t.dunno2 = int(r.ReadULong())
t.dunno3 = int(r.ReadULong())
case api.ListItemFolder:
t.dunno2 = int(r.ReadULong())
case api.ListItemSharedFolder:
t.dunno2 = int(r.ReadULong())
t.treeID = r.ReadNBytes(api.TreeIDLength)
case api.ListItemFile:
isDir = false
modTime = r.ReadDate()
size = int64(r.ReadULong())
binHash = r.ReadNBytes(mrhash.Size)
default:
return nil, fmt.Errorf("unknown item type %d", itemType)
}
if isDir {
t.lastDir = path.Join(t.currDir, name)
if (t.options & api.ListOptDelete) != 0 {
t.dunnoDel1 = int(r.ReadPu32())
t.dunnoDel2 = int(r.ReadPu32())
}
if (t.options & api.ListOptFolderSize) != 0 {
dirSize = int64(r.ReadULong())
}
}
if r.Error() != nil {
return nil, r.Error()
}
if fs.Config.LogLevel >= fs.LogLevelDebug {
ctime, _ := modTime.MarshalJSON()
fs.Debugf(t.f, "binDir %d.%d %q %q (%d) %s", t.level, itemType, t.currDir, name, size, ctime)
}
if t.level != 1 {
// TODO: implement recursion and ListR
// Note: recursion is broken because maximum buffer size is 8K
return nil, nil
}
remote, err := t.f.relPath(path.Join(t.currDir, name))
if err != nil {
return nil, err
}
if isDir {
return fs.NewDir(remote, modTime).SetSize(dirSize), nil
}
obj := &Object{
fs: t.f,
remote: remote,
hasMetaData: true,
size: size,
mrHash: binHash,
modTime: modTime,
}
return obj, nil
}
type treeState struct {
f *Fs
r *api.BinReader
options int
rootDir string
currDir string
lastDir string
level int
revision treeRevision
totalSpace int64
usedSpace int64
fingerprint []byte
dunno1 int
dunno2 int
dunno3 int
dunnoDel1 int
dunnoDel2 int
dunnoNodeID []byte
treeID []byte
}
type treeRevision struct {
ver int16
treeID []byte
treeIDNew []byte
bgn uint64
bgnNew uint64
}
func (rev *treeRevision) Read(data *api.BinReader) error {
rev.ver = data.ReadByteAsShort()
switch rev.ver {
case 0:
// Revision()
case 1, 2:
rev.treeID = data.ReadNBytes(api.TreeIDLength)
rev.bgn = data.ReadULong()
case 3, 4:
rev.treeID = data.ReadNBytes(api.TreeIDLength)
rev.bgn = data.ReadULong()
rev.treeIDNew = data.ReadNBytes(api.TreeIDLength)
rev.bgnNew = data.ReadULong()
case 5:
rev.treeID = data.ReadNBytes(api.TreeIDLength)
rev.bgn = data.ReadULong()
rev.treeIDNew = data.ReadNBytes(api.TreeIDLength)
default:
return fmt.Errorf("unknown directory revision %d", rev.ver)
}
return data.Error()
}
// CreateDir makes a directory (parent must exist)
func (f *Fs) CreateDir(ctx context.Context, path string) error {
// fs.Debugf(f, ">>> CreateDir %q", path)
req := api.NewBinWriter()
req.WritePu16(api.OperationCreateFolder)
req.WritePu16(0) // revision
req.WriteString(f.opt.Enc.FromStandardPath(path))
req.WritePu32(0)
token, err := f.accessToken()
if err != nil {
return err
}
metaURL, err := f.metaServer(ctx)
if err != nil {
return err
}
opts := rest.Opts{
Method: "POST",
RootURL: metaURL,
Parameters: url.Values{
"client_id": {api.OAuthClientID},
"token": {token},
},
ContentType: api.BinContentType,
Body: req.Reader(),
}
var res *http.Response
err = f.pacer.Call(func() (bool, error) {
res, err = f.srv.Call(ctx, &opts)
return shouldRetry(res, err, f, &opts)
})
if err != nil {
closeBody(res)
return err
}
reply := api.NewBinReader(res.Body)
defer closeBody(res)
switch status := reply.ReadByteAsInt(); status {
case api.MkdirResultOK:
return nil
case api.MkdirResultAlreadyExists, api.MkdirResultExistsDifferentCase:
return ErrorDirAlreadyExists
case api.MkdirResultSourceNotExists:
return ErrorDirSourceNotExists
case api.MkdirResultInvalidName:
return ErrorInvalidName
default:
return fmt.Errorf("mkdir error %d", status)
}
}
// Mkdir creates the container (and its parents) if it doesn't exist.
// Normally it ignores the ErrorDirAlreadyExist, as required by rclone tests.
// Nevertheless, such programs as borgbackup or restic use mkdir as a locking
// primitive and depend on its atomicity, i.e. mkdir should fail if directory
// already exists. As a workaround, users can add string "atomicmkdir" in the
// hidden `quirks` parameter or in the `--mailru-quirks` command-line option.
func (f *Fs) Mkdir(ctx context.Context, dir string) error {
// fs.Debugf(f, ">>> Mkdir %q", dir)
err := f.mkDirs(ctx, f.absPath(dir))
if err == ErrorDirAlreadyExists && !f.quirks.atomicmkdir {
return nil
}
return err
}
// mkDirs creates container and its parents by absolute path,
// fails with ErrorDirAlreadyExists if it already exists.
func (f *Fs) mkDirs(ctx context.Context, path string) error {
if path == "/" || path == "" {
return nil
}
switch err := f.CreateDir(ctx, path); err {
case nil:
return nil
case ErrorDirSourceNotExists:
fs.Debugf(f, "mkDirs by part %q", path)
// fall thru...
default:
return err
}
parts := strings.Split(strings.Trim(path, "/"), "/")
path = ""
for _, part := range parts {
if part == "" {
continue
}
path += "/" + part
switch err := f.CreateDir(ctx, path); err {
case nil, ErrorDirAlreadyExists:
continue
default:
return err
}
}
return nil
}
func parentDir(absPath string) string {
parent := path.Dir(strings.TrimRight(absPath, "/"))
if parent == "." {
parent = ""
}
return parent
}
// mkParentDirs creates parent containers by absolute path,
// ignores the ErrorDirAlreadyExists
func (f *Fs) mkParentDirs(ctx context.Context, path string) error {
err := f.mkDirs(ctx, parentDir(path))
if err == ErrorDirAlreadyExists {
return nil
}
return err
}
// Rmdir deletes a directory.
// Returns an error if it isn't empty.
func (f *Fs) Rmdir(ctx context.Context, dir string) error {
// fs.Debugf(f, ">>> Rmdir %q", dir)
return f.purgeWithCheck(ctx, dir, true, "rmdir")
}
// Purge deletes all the files and the root directory
// Optional interface: Only implement this if you have a way of deleting
// all the files quicker than just running Remove() on the result of List()
func (f *Fs) Purge(ctx context.Context) error {
// fs.Debugf(f, ">>> Purge")
return f.purgeWithCheck(ctx, "", false, "purge")
}
// purgeWithCheck() removes the root directory.
// Refuses if `check` is set and directory has anything in.
func (f *Fs) purgeWithCheck(ctx context.Context, dir string, check bool, opName string) error {
path := f.absPath(dir)
if path == "/" || path == "" {
// Mailru will not allow to purge root space returning status 400
return fs.ErrorNotDeletingDirs
}
_, dirSize, err := f.readItemMetaData(ctx, path)
if err != nil {
return errors.Wrapf(err, "%s failed", opName)
}
if check && dirSize > 0 {
return fs.ErrorDirectoryNotEmpty
}
return f.delete(ctx, path, false)
}
func (f *Fs) delete(ctx context.Context, path string, hardDelete bool) error {
token, err := f.accessToken()
if err != nil {
return err
}
data := url.Values{"home": {f.opt.Enc.FromStandardPath(path)}}
opts := rest.Opts{
Method: "POST",
Path: "/api/m1/file/remove",
Parameters: url.Values{
"access_token": {token},
},
Body: strings.NewReader(data.Encode()),
ContentType: api.BinContentType,
}
var response api.GenericResponse
err = f.pacer.Call(func() (bool, error) {
res, err := f.srv.CallJSON(ctx, &opts, nil, &response)
return shouldRetry(res, err, f, &opts)
})
switch {
case err != nil:
return err
case response.Status == 200:
return nil
default:
return fmt.Errorf("delete failed with code %d", response.Status)
}
}
// Copy src to this remote using server side copy operations.
// This is stored with the remote path given.
// It returns the destination Object and a possible error.
// Will only be called if src.Fs().Name() == f.Name()
// If it isn't possible then return fs.ErrorCantCopy
func (f *Fs) Copy(ctx context.Context, src fs.Object, remote string) (fs.Object, error) {
// fs.Debugf(f, ">>> Copy %q %q", src.Remote(), remote)
srcObj, ok := src.(*Object)
if !ok {
fs.Debugf(src, "Can't copy - not same remote type")
return nil, fs.ErrorCantCopy
}
if srcObj.fs.opt.Username != f.opt.Username {
// Can copy across mailru configs only if they share common account
fs.Debugf(src, "Can't copy - not same account")
return nil, fs.ErrorCantCopy
}
srcPath := srcObj.absPath()
dstPath := f.absPath(remote)
overwrite := false
// fs.Debugf(f, "copy %q -> %q\n", srcPath, dstPath)
err := f.mkParentDirs(ctx, dstPath)
if err != nil {
return nil, err
}
data := url.Values{}
data.Set("home", f.opt.Enc.FromStandardPath(srcPath))
data.Set("folder", f.opt.Enc.FromStandardPath(parentDir(dstPath)))
data.Set("email", f.opt.Username)
data.Set("x-email", f.opt.Username)
if overwrite {
data.Set("conflict", "rewrite")
} else {
data.Set("conflict", "rename")
}
token, err := f.accessToken()
if err != nil {
return nil, err
}
opts := rest.Opts{
Method: "POST",
Path: "/api/m1/file/copy",
Parameters: url.Values{
"access_token": {token},
},
Body: strings.NewReader(data.Encode()),
ContentType: api.BinContentType,
}
var response api.GenericBodyResponse
err = f.pacer.Call(func() (bool, error) {
res, err := f.srv.CallJSON(ctx, &opts, nil, &response)
return shouldRetry(res, err, f, &opts)
})
if err != nil {
return nil, errors.Wrap(err, "couldn't copy file")
}
if response.Status != 200 {
return nil, fmt.Errorf("copy failed with code %d", response.Status)
}
tmpPath := f.opt.Enc.ToStandardPath(response.Body)
if tmpPath != dstPath {
// fs.Debugf(f, "rename temporary file %q -> %q\n", tmpPath, dstPath)
err = f.moveItemBin(ctx, tmpPath, dstPath, "rename temporary file")
if err != nil {
_ = f.delete(ctx, tmpPath, false) // ignore error
return nil, err
}
}
// fix modification time at destination
dstObj := &Object{
fs: f,
remote: remote,
}
err = dstObj.readMetaData(ctx, true)
if err == nil && dstObj.modTime != srcObj.modTime {
dstObj.modTime = srcObj.modTime
err = dstObj.addFileMetaData(ctx, true)
}
if err != nil {
dstObj = nil
}
return dstObj, err
}
// Move src to this remote using server side move operations.
// This is stored with the remote path given.
// It returns the destination Object and a possible error.
// Will only be called if src.Fs().Name() == f.Name()
// If it isn't possible then return fs.ErrorCantMove
func (f *Fs) Move(ctx context.Context, src fs.Object, remote string) (fs.Object, error) {
// fs.Debugf(f, ">>> Move %q %q", src.Remote(), remote)
srcObj, ok := src.(*Object)
if !ok {
fs.Debugf(src, "Can't move - not same remote type")
return nil, fs.ErrorCantMove
}
if srcObj.fs.opt.Username != f.opt.Username {
// Can move across mailru configs only if they share common account
fs.Debugf(src, "Can't move - not same account")
return nil, fs.ErrorCantMove
}
srcPath := srcObj.absPath()
dstPath := f.absPath(remote)
err := f.mkParentDirs(ctx, dstPath)
if err != nil {
return nil, err
}
err = f.moveItemBin(ctx, srcPath, dstPath, "move file")
if err != nil {
return nil, err
}
return f.NewObject(ctx, remote)
}
// move/rename an object using BIN protocol
func (f *Fs) moveItemBin(ctx context.Context, srcPath, dstPath, opName string) error {
token, err := f.accessToken()
if err != nil {
return err
}
metaURL, err := f.metaServer(ctx)
if err != nil {
return err
}
req := api.NewBinWriter()
req.WritePu16(api.OperationRename)
req.WritePu32(0) // old revision
req.WriteString(f.opt.Enc.FromStandardPath(srcPath))
req.WritePu32(0) // new revision
req.WriteString(f.opt.Enc.FromStandardPath(dstPath))
req.WritePu32(0) // dunno
opts := rest.Opts{
Method: "POST",
RootURL: metaURL,
Parameters: url.Values{
"client_id": {api.OAuthClientID},
"token": {token},
},
ContentType: api.BinContentType,
Body: req.Reader(),
}
var res *http.Response
err = f.pacer.Call(func() (bool, error) {
res, err = f.srv.Call(ctx, &opts)
return shouldRetry(res, err, f, &opts)
})
if err != nil {
closeBody(res)
return err
}
reply := api.NewBinReader(res.Body)
defer closeBody(res)
switch status := reply.ReadByteAsInt(); status {
case api.MoveResultOK:
return nil
default:
return fmt.Errorf("%s failed with error %d", opName, status)
}
}
// DirMove moves src, srcRemote to this remote at dstRemote
// using server side move operations.
// Will only be called if src.Fs().Name() == f.Name()
// If it isn't possible then return fs.ErrorCantDirMove
// If destination exists then return fs.ErrorDirExists
func (f *Fs) DirMove(ctx context.Context, src fs.Fs, srcRemote, dstRemote string) error {
// fs.Debugf(f, ">>> DirMove %q %q", srcRemote, dstRemote)
srcFs, ok := src.(*Fs)
if !ok {
fs.Debugf(srcFs, "Can't move directory - not same remote type")
return fs.ErrorCantDirMove
}
if srcFs.opt.Username != f.opt.Username {
// Can move across mailru configs only if they share common account
fs.Debugf(src, "Can't move - not same account")
return fs.ErrorCantDirMove
}
srcPath := srcFs.absPath(srcRemote)
dstPath := f.absPath(dstRemote)
// fs.Debugf(srcFs, "DirMove [%s]%q --> [%s]%q\n", srcRemote, srcPath, dstRemote, dstPath)
// Refuse to move to or from the root
if len(srcPath) <= len(srcFs.root) || len(dstPath) <= len(f.root) {
fs.Debugf(src, "DirMove error: Can't move root")
return errors.New("can't move root directory")
}
err := f.mkParentDirs(ctx, dstPath)
if err != nil {
return err
}
_, _, err = f.readItemMetaData(ctx, dstPath)
switch err {
case fs.ErrorObjectNotFound:
// OK!
case nil:
return fs.ErrorDirExists
default:
return err
}
return f.moveItemBin(ctx, srcPath, dstPath, "directory move")
}
// PublicLink generates a public link to the remote path (usually readable by anyone)
func (f *Fs) PublicLink(ctx context.Context, remote string, expire fs.Duration, unlink bool) (link string, err error) {
// fs.Debugf(f, ">>> PublicLink %q", remote)
token, err := f.accessToken()
if err != nil {
return "", err
}
data := url.Values{}
data.Set("home", f.opt.Enc.FromStandardPath(f.absPath(remote)))
data.Set("email", f.opt.Username)
data.Set("x-email", f.opt.Username)
opts := rest.Opts{
Method: "POST",
Path: "/api/m1/file/publish",
Parameters: url.Values{
"access_token": {token},
},
Body: strings.NewReader(data.Encode()),
ContentType: api.BinContentType,
}
var response api.GenericBodyResponse
err = f.pacer.Call(func() (bool, error) {
res, err := f.srv.CallJSON(ctx, &opts, nil, &response)
return shouldRetry(res, err, f, &opts)
})
if err == nil && response.Body != "" {
return api.PublicLinkURL + response.Body, nil
}
if err == nil {
return "", errors.New("server returned empty link")
}
if apiErr, ok := err.(*api.FileErrorResponse); ok && apiErr.Status == 404 {
return "", fs.ErrorObjectNotFound
}
return "", err
}
// CleanUp permanently deletes all trashed files/folders
func (f *Fs) CleanUp(ctx context.Context) error {
// fs.Debugf(f, ">>> CleanUp")
token, err := f.accessToken()
if err != nil {
return err
}
data := url.Values{
"email": {f.opt.Username},
"x-email": {f.opt.Username},
}
opts := rest.Opts{
Method: "POST",
Path: "/api/m1/trashbin/empty",
Parameters: url.Values{
"access_token": {token},
},
Body: strings.NewReader(data.Encode()),
ContentType: api.BinContentType,
}
var response api.CleanupResponse
err = f.pacer.Call(func() (bool, error) {
res, err := f.srv.CallJSON(ctx, &opts, nil, &response)
return shouldRetry(res, err, f, &opts)
})
if err != nil {
return err
}
switch response.StatusStr {
case "200":
return nil
default:
return fmt.Errorf("cleanup failed (%s)", response.StatusStr)
}
}
// About gets quota information
func (f *Fs) About(ctx context.Context) (*fs.Usage, error) {
// fs.Debugf(f, ">>> About")
token, err := f.accessToken()
if err != nil {
return nil, err
}
opts := rest.Opts{
Method: "GET",
Path: "/api/m1/user",
Parameters: url.Values{
"access_token": {token},
},
}
var info api.UserInfoResponse
err = f.pacer.Call(func() (bool, error) {
res, err := f.srv.CallJSON(ctx, &opts, nil, &info)
return shouldRetry(res, err, f, &opts)
})
if err != nil {
return nil, err
}
total := info.Body.Cloud.Space.BytesTotal
used := int64(info.Body.Cloud.Space.BytesUsed)
usage := &fs.Usage{
Total: fs.NewUsageValue(total),
Used: fs.NewUsageValue(used),
Free: fs.NewUsageValue(total - used),
}
return usage, nil
}
// Put the object
// Copy the reader in to the new object which is returned
// The new object may have been created if an error is returned
func (f *Fs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) {
o := &Object{
fs: f,
remote: src.Remote(),
size: src.Size(),
modTime: src.ModTime(ctx),
}
// fs.Debugf(f, ">>> Put: %q %d '%v'", o.remote, o.size, o.modTime)
return o, o.Update(ctx, in, src, options...)
}
// Update an existing object
// Copy the reader into the object updating modTime and size
// The new object may have been created if an error is returned
func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) error {
wrapIn := in
size := src.Size()
if size < 0 {
return errors.New("mailru does not support streaming uploads")
}
err := o.fs.mkParentDirs(ctx, o.absPath())
if err != nil {
return err
}
var (
fileBuf []byte
fileHash []byte
newHash []byte
trySpeedup bool
)
// Don't disturb the source if file fits in hash.
// Skip an extra speedup request if file fits in hash.
if size > mrhash.Size {
// Request hash from source.
if srcHash, err := src.Hash(ctx, MrHashType); err == nil && srcHash != "" {
fileHash, _ = mrhash.DecodeString(srcHash)
}
// Try speedup if it's globally enabled and source hash is available.
trySpeedup = o.fs.opt.SpeedupEnable
if trySpeedup && fileHash != nil {
if o.putByHash(ctx, fileHash, src, "source") {
return nil
}
trySpeedup = false // speedup failed, force upload
}
}
// Need to calculate hash, check whether file is still eligible for speedup
if trySpeedup {
trySpeedup = o.fs.eligibleForSpeedup(o.Remote(), size, options...)
}
// Attempt to put by calculating hash in memory
if trySpeedup && size <= int64(o.fs.opt.SpeedupMaxMem) {
//fs.Debugf(o, "attempt to put by hash from memory")
fileBuf, err = ioutil.ReadAll(in)
if err != nil {
return err
}
fileHash = mrhash.Sum(fileBuf)
if o.putByHash(ctx, fileHash, src, "memory") {
return nil
}
wrapIn = bytes.NewReader(fileBuf)
trySpeedup = false // speedup failed, force upload
}
// Attempt to put by hash using a spool file
if trySpeedup {
tmpFs, err := fs.TemporaryLocalFs()
if err != nil {
fs.Infof(tmpFs, "Failed to create spool FS: %v", err)
} else {
defer func() {
if err := operations.Purge(ctx, tmpFs, ""); err != nil {
fs.Infof(tmpFs, "Failed to cleanup spool FS: %v", err)
}
}()
spoolFile, mrHash, err := makeTempFile(ctx, tmpFs, wrapIn, src)
if err != nil {
return errors.Wrap(err, "Failed to create spool file")
}
if o.putByHash(ctx, mrHash, src, "spool") {
// If put by hash is successful, ignore transitive error
return nil
}
if wrapIn, err = spoolFile.Open(ctx); err != nil {
return err
}
fileHash = mrHash
}
}
// Upload object data
if size <= mrhash.Size {
// Optimize upload: skip extra request if data fits in the hash buffer.
if fileBuf == nil {
fileBuf, err = ioutil.ReadAll(wrapIn)
}
if fileHash == nil && err == nil {
fileHash = mrhash.Sum(fileBuf)
}
newHash = fileHash
} else {
var hasher gohash.Hash
if fileHash == nil {
// Calculate hash in transit
hasher = mrhash.New()
wrapIn = io.TeeReader(wrapIn, hasher)
}
newHash, err = o.upload(ctx, wrapIn, size, options...)
if fileHash == nil && err == nil {
fileHash = hasher.Sum(nil)
}
}
if err != nil {
return err
}
if bytes.Compare(fileHash, newHash) != 0 {
if o.fs.opt.CheckHash {
return mrhash.ErrorInvalidHash
}
fs.Infof(o, "hash mismatch on upload: expected %x received %x", fileHash, newHash)
}
o.mrHash = newHash
o.size = size
o.modTime = src.ModTime(ctx)
return o.addFileMetaData(ctx, true)
}
// eligibleForSpeedup checks whether file is eligible for speedup method (put by hash)
func (f *Fs) eligibleForSpeedup(remote string, size int64, options ...fs.OpenOption) bool {
if !f.opt.SpeedupEnable {
return false
}
if size <= mrhash.Size || size < speedupMinSize || size >= int64(f.opt.SpeedupMaxDisk) {
return false
}
_, _, partial := getTransferRange(size, options...)
if partial {
return false
}
if f.speedupAny {
return true
}
if f.speedupGlobs == nil {
return false
}
nameLower := strings.ToLower(strings.TrimSpace(path.Base(remote)))
for _, pattern := range f.speedupGlobs {
if matches, _ := filepath.Match(pattern, nameLower); matches {
return true
}
}
return false
}
// parseSpeedupPatterns converts pattern string into list of unique glob patterns
func (f *Fs) parseSpeedupPatterns(patternString string) (err error) {
f.speedupGlobs = nil
f.speedupAny = false
uniqueValidPatterns := make(map[string]interface{})
for _, pattern := range strings.Split(patternString, ",") {
pattern = strings.ToLower(strings.TrimSpace(pattern))
if pattern == "" {
continue
}
if pattern == "*" {
f.speedupAny = true
}
if _, err := filepath.Match(pattern, ""); err != nil {
return fmt.Errorf("invalid file name pattern %q", pattern)
}
uniqueValidPatterns[pattern] = nil
}
for pattern := range uniqueValidPatterns {
f.speedupGlobs = append(f.speedupGlobs, pattern)
}
return nil
}
func (o *Object) putByHash(ctx context.Context, mrHash []byte, info fs.ObjectInfo, method string) bool {
oNew := new(Object)
*oNew = *o
oNew.mrHash = mrHash
oNew.size = info.Size()
oNew.modTime = info.ModTime(ctx)
if err := oNew.addFileMetaData(ctx, true); err != nil {
fs.Debugf(o, "Cannot put by hash from %s, performing upload", method)
return false
}
*o = *oNew
fs.Debugf(o, "File has been put by hash from %s", method)
return true
}
func makeTempFile(ctx context.Context, tmpFs fs.Fs, wrapIn io.Reader, src fs.ObjectInfo) (spoolFile fs.Object, mrHash []byte, err error) {
// Local temporary file system must support SHA1
hashType := hash.SHA1
// Calculate Mailru and spool verification hashes in transit
hashSet := hash.NewHashSet(MrHashType, hashType)
hasher, err := hash.NewMultiHasherTypes(hashSet)
if err != nil {
return nil, nil, err
}
wrapIn = io.TeeReader(wrapIn, hasher)
// Copy stream into spool file
tmpInfo := object.NewStaticObjectInfo(src.Remote(), src.ModTime(ctx), src.Size(), false, nil, nil)
hashOption := &fs.HashesOption{Hashes: hashSet}
if spoolFile, err = tmpFs.Put(ctx, wrapIn, tmpInfo, hashOption); err != nil {
return nil, nil, err
}
// Validate spool file
sums := hasher.Sums()
checkSum := sums[hashType]
fileSum, err := spoolFile.Hash(ctx, hashType)
if spoolFile.Size() != src.Size() || err != nil || checkSum == "" || fileSum != checkSum {
return nil, nil, mrhash.ErrorInvalidHash
}
mrHash, err = mrhash.DecodeString(sums[MrHashType])
return
}
func (o *Object) upload(ctx context.Context, in io.Reader, size int64, options ...fs.OpenOption) ([]byte, error) {
token, err := o.fs.accessToken()
if err != nil {
return nil, err
}
shardURL, err := o.fs.uploadShard(ctx)
if err != nil {
return nil, err
}
opts := rest.Opts{
Method: "PUT",
RootURL: shardURL,
Body: in,
Options: options,
ContentLength: &size,
Parameters: url.Values{
"client_id": {api.OAuthClientID},
"token": {token},
},
ExtraHeaders: map[string]string{
"Accept": "*/*",
},
}
var (
res *http.Response
strHash string
)
err = o.fs.pacer.Call(func() (bool, error) {
res, err = o.fs.srv.Call(ctx, &opts)
if err == nil {
strHash, err = readBodyWord(res)
}
return fserrors.ShouldRetry(err), err
})
if err != nil {
closeBody(res)
return nil, err
}
switch res.StatusCode {
case 200, 201:
return mrhash.DecodeString(strHash)
default:
return nil, fmt.Errorf("upload failed with code %s (%d)", res.Status, res.StatusCode)
}
}
func (f *Fs) uploadShard(ctx context.Context) (string, error) {
f.shardMu.Lock()
defer f.shardMu.Unlock()
if f.shardURL != "" && time.Now().Before(f.shardExpiry) {
return f.shardURL, nil
}
token, err := f.accessToken()
if err != nil {
return "", err
}
opts := rest.Opts{
Method: "GET",
Path: "/api/m1/dispatcher",
Parameters: url.Values{
"client_id": {api.OAuthClientID},
"access_token": {token},
},
}
var info api.ShardInfoResponse
err = f.pacer.Call(func() (bool, error) {
res, err := f.srv.CallJSON(ctx, &opts, nil, &info)
return shouldRetry(res, err, f, &opts)
})
if err != nil {
return "", err
}
f.shardURL = info.Body.Upload[0].URL
f.shardExpiry = time.Now().Add(shardExpirySec * time.Second)
fs.Debugf(f, "new upload shard: %s", f.shardURL)
return f.shardURL, nil
}
// Object describes a mailru object
type Object struct {
fs *Fs // what this object is part of
remote string // The remote path
hasMetaData bool // whether info below has been set
size int64 // Bytes in the object
modTime time.Time // Modified time of the object
mrHash []byte // Mail.ru flavored SHA1 hash of the object
}
// NewObject finds an Object at the remote.
// If object can't be found it fails with fs.ErrorObjectNotFound
func (f *Fs) NewObject(ctx context.Context, remote string) (fs.Object, error) {
// fs.Debugf(f, ">>> NewObject %q", remote)
o := &Object{
fs: f,
remote: remote,
}
err := o.readMetaData(ctx, true)
if err != nil {
return nil, err
}
return o, nil
}
// absPath converts root-relative remote to absolute home path
func (o *Object) absPath() string {
return o.fs.absPath(o.remote)
}
// Object.readMetaData reads and fills a file info
// If object can't be found it fails with fs.ErrorObjectNotFound
func (o *Object) readMetaData(ctx context.Context, force bool) error {
if o.hasMetaData && !force {
return nil
}
entry, dirSize, err := o.fs.readItemMetaData(ctx, o.absPath())
if err != nil {
return err
}
newObj, ok := entry.(*Object)
if !ok || dirSize >= 0 {
return fs.ErrorNotAFile
}
if newObj.remote != o.remote {
return fmt.Errorf("File %q path has changed to %q", o.remote, newObj.remote)
}
o.hasMetaData = true
o.size = newObj.size
o.modTime = newObj.modTime
o.mrHash = newObj.mrHash
return nil
}
// Fs returns the parent Fs
func (o *Object) Fs() fs.Info {
return o.fs
}
// Return a string version
func (o *Object) String() string {
if o == nil {
return "<nil>"
}
//return fmt.Sprintf("[%s]%q", o.fs.root, o.remote)
return o.remote
}
// Remote returns the remote path
func (o *Object) Remote() string {
return o.remote
}
// ModTime returns the modification time of the object
// It attempts to read the objects mtime and if that isn't present the
// LastModified returned in the http headers
func (o *Object) ModTime(ctx context.Context) time.Time {
err := o.readMetaData(ctx, false)
if err != nil {
fs.Errorf(o, "%v", err)
}
return o.modTime
}
// Size returns the size of an object in bytes
func (o *Object) Size() int64 {
ctx := context.Background() // Note: Object.Size does not pass context!
err := o.readMetaData(ctx, false)
if err != nil {
fs.Errorf(o, "%v", err)
}
return o.size
}
// Hash returns the MD5 or SHA1 sum of an object
// returning a lowercase hex string
func (o *Object) Hash(ctx context.Context, t hash.Type) (string, error) {
if t == MrHashType {
return hex.EncodeToString(o.mrHash), nil
}
return "", hash.ErrUnsupported
}
// Storable returns whether this object is storable
func (o *Object) Storable() bool {
return true
}
// SetModTime sets the modification time of the local fs object
//
// Commits the datastore
func (o *Object) SetModTime(ctx context.Context, modTime time.Time) error {
// fs.Debugf(o, ">>> SetModTime [%v]", modTime)
o.modTime = modTime
return o.addFileMetaData(ctx, true)
}
func (o *Object) addFileMetaData(ctx context.Context, overwrite bool) error {
if len(o.mrHash) != mrhash.Size {
return mrhash.ErrorInvalidHash
}
token, err := o.fs.accessToken()
if err != nil {
return err
}
metaURL, err := o.fs.metaServer(ctx)
if err != nil {
return err
}
req := api.NewBinWriter()
req.WritePu16(api.OperationAddFile)
req.WritePu16(0) // revision
req.WriteString(o.fs.opt.Enc.FromStandardPath(o.absPath()))
req.WritePu64(o.size)
req.WritePu64(o.modTime.Unix())
req.WritePu32(0)
req.Write(o.mrHash)
if overwrite {
// overwrite
req.WritePu32(1)
} else {
// don't add if not changed, add with rename if changed
req.WritePu32(55)
req.Write(o.mrHash)
req.WritePu64(o.size)
}
opts := rest.Opts{
Method: "POST",
RootURL: metaURL,
Parameters: url.Values{
"client_id": {api.OAuthClientID},
"token": {token},
},
ContentType: api.BinContentType,
Body: req.Reader(),
}
var res *http.Response
err = o.fs.pacer.Call(func() (bool, error) {
res, err = o.fs.srv.Call(ctx, &opts)
return shouldRetry(res, err, o.fs, &opts)
})
if err != nil {
closeBody(res)
return err
}
reply := api.NewBinReader(res.Body)
defer closeBody(res)
switch status := reply.ReadByteAsInt(); status {
case api.AddResultOK, api.AddResultNotModified, api.AddResultDunno04, api.AddResultDunno09:
return nil
case api.AddResultInvalidName:
return ErrorInvalidName
default:
return fmt.Errorf("add file error %d", status)
}
}
// Remove an object
func (o *Object) Remove(ctx context.Context) error {
// fs.Debugf(o, ">>> Remove")
return o.fs.delete(ctx, o.absPath(), false)
}
// getTransferRange detects partial transfers and calculates start/end offsets into file
func getTransferRange(size int64, options ...fs.OpenOption) (start int64, end int64, partial bool) {
var offset, limit int64 = 0, -1
for _, option := range options {
switch opt := option.(type) {
case *fs.SeekOption:
offset = opt.Offset
case *fs.RangeOption:
offset, limit = opt.Decode(size)
default:
if option.Mandatory() {
fs.Errorf(nil, "Unsupported mandatory option: %v", option)
}
}
}
if limit < 0 {
limit = size - offset
}
end = offset + limit
if end > size {
end = size
}
partial = !(offset == 0 && end == size)
return offset, end, partial
}
// Open an object for read and download its content
func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.ReadCloser, err error) {
// fs.Debugf(o, ">>> Open")
token, err := o.fs.accessToken()
if err != nil {
return nil, err
}
start, end, partial := getTransferRange(o.size, options...)
// TODO: set custom timeouts
opts := rest.Opts{
Method: "GET",
Options: options,
Path: url.PathEscape(strings.TrimLeft(o.fs.opt.Enc.FromStandardPath(o.absPath()), "/")),
Parameters: url.Values{
"client_id": {api.OAuthClientID},
"token": {token},
},
ExtraHeaders: map[string]string{
"Accept": "*/*",
"Range": fmt.Sprintf("bytes=%d-%d", start, end-1),
},
}
var res *http.Response
server := ""
err = o.fs.pacer.Call(func() (bool, error) {
server, err = o.fs.fileServers.Dispatch(ctx, server)
if err != nil {
return false, err
}
opts.RootURL = server
res, err = o.fs.srv.Call(ctx, &opts)
return shouldRetry(res, err, o.fs, &opts)
})
if err != nil {
if res != nil && res.Body != nil {
closeBody(res)
}
return nil, err
}
var hasher gohash.Hash
if !partial {
// Cannot check hash of partial download
hasher = mrhash.New()
}
wrapStream := &endHandler{
ctx: ctx,
stream: res.Body,
hasher: hasher,
o: o,
server: server,
}
return wrapStream, nil
}
type endHandler struct {
ctx context.Context
stream io.ReadCloser
hasher gohash.Hash
o *Object
server string
done bool
}
func (e *endHandler) Read(p []byte) (n int, err error) {
n, err = e.stream.Read(p)
if e.hasher != nil {
// hasher will not return an error, just panic
_, _ = e.hasher.Write(p[:n])
}
if err != nil { // io.Error or EOF
err = e.handle(err)
}
return
}
func (e *endHandler) Close() error {
_ = e.handle(nil) // ignore returned error
return e.stream.Close()
}
func (e *endHandler) handle(err error) error {
if e.done {
return err
}
e.done = true
o := e.o
o.fs.fileServers.Free(e.server)
if err != io.EOF || e.hasher == nil {
return err
}
newHash := e.hasher.Sum(nil)
if bytes.Compare(o.mrHash, newHash) == 0 {
return io.EOF
}
if o.fs.opt.CheckHash {
return mrhash.ErrorInvalidHash
}
fs.Infof(o, "hash mismatch on download: expected %x received %x", o.mrHash, newHash)
return io.EOF
}
// serverPool backs server dispacher
type serverPool struct {
pool pendingServerMap
mu sync.Mutex
path string
expirySec time.Duration
fs *Fs
}
type pendingServerMap map[string]*pendingServer
type pendingServer struct {
locks int
expiry time.Time
}
// Dispatch dispatches next download server.
// It prefers switching and tries to avoid current server
// in use by caller because it may be overloaded or slow.
func (p *serverPool) Dispatch(ctx context.Context, current string) (string, error) {
now := time.Now()
url := p.getServer(current, now)
if url != "" {
return url, nil
}
// Server not found - ask Mailru dispatcher.
opts := rest.Opts{
Method: "GET",
RootURL: api.DispatchServerURL,
Path: p.path,
}
var (
res *http.Response
err error
)
err = p.fs.pacer.Call(func() (bool, error) {
res, err = p.fs.srv.Call(ctx, &opts)
if err != nil {
return fserrors.ShouldRetry(err), err
}
url, err = readBodyWord(res)
return fserrors.ShouldRetry(err), err
})
if err != nil || url == "" {
closeBody(res)
return "", errors.Wrap(err, "Failed to request file server")
}
p.addServer(url, now)
return url, nil
}
func (p *serverPool) Free(url string) {
if url == "" {
return
}
p.mu.Lock()
defer p.mu.Unlock()
srv := p.pool[url]
if srv == nil {
return
}
if srv.locks <= 0 {
// Getting here indicates possible race
fs.Infof(p.fs, "Purge file server: locks -, url %s", url)
delete(p.pool, url)
return
}
srv.locks--
if srv.locks == 0 && time.Now().After(srv.expiry) {
delete(p.pool, url)
fs.Debugf(p.fs, "Free file server: locks 0, url %s", url)
return
}
fs.Debugf(p.fs, "Unlock file server: locks %d, url %s", srv.locks, url)
}
// Find an underlocked server
func (p *serverPool) getServer(current string, now time.Time) string {
p.mu.Lock()
defer p.mu.Unlock()
for url, srv := range p.pool {
if url == "" || srv.locks < 0 {
continue // Purged server slot
}
if url == current {
continue // Current server - prefer another
}
if srv.locks >= maxServerLocks {
continue // Overlocked server
}
if now.After(srv.expiry) {
continue // Expired server
}
srv.locks++
fs.Debugf(p.fs, "Lock file server: locks %d, url %s", srv.locks, url)
return url
}
return ""
}
func (p *serverPool) addServer(url string, now time.Time) {
p.mu.Lock()
defer p.mu.Unlock()
expiry := now.Add(p.expirySec * time.Second)
expiryStr := []byte("-")
if fs.Config.LogLevel >= fs.LogLevelInfo {
expiryStr, _ = expiry.MarshalJSON()
}
// Attach to a server proposed by dispatcher
srv := p.pool[url]
if srv != nil {
srv.locks++
srv.expiry = expiry
fs.Debugf(p.fs, "Reuse file server: locks %d, url %s, expiry %s", srv.locks, url, expiryStr)
return
}
// Add new server
p.pool[url] = &pendingServer{locks: 1, expiry: expiry}
fs.Debugf(p.fs, "Switch file server: locks 1, url %s, expiry %s", url, expiryStr)
}
// Name of the remote (as passed into NewFs)
func (f *Fs) Name() string {
return f.name
}
// Root of the remote (as passed into NewFs)
func (f *Fs) Root() string {
return f.root
}
// String converts this Fs to a string
func (f *Fs) String() string {
return fmt.Sprintf("[%s]", f.root)
}
// Precision return the precision of this Fs
func (f *Fs) Precision() time.Duration {
return time.Second
}
// Hashes returns the supported hash sets
func (f *Fs) Hashes() hash.Set {
return hash.Set(MrHashType)
}
// Features returns the optional features of this Fs
func (f *Fs) Features() *fs.Features {
return f.features
}
// close response body ignoring errors
func closeBody(res *http.Response) {
if res != nil {
_ = res.Body.Close()
}
}
// Check the interfaces are satisfied
var (
_ fs.Fs = (*Fs)(nil)
_ fs.Purger = (*Fs)(nil)
_ fs.Copier = (*Fs)(nil)
_ fs.Mover = (*Fs)(nil)
_ fs.DirMover = (*Fs)(nil)
_ fs.PublicLinker = (*Fs)(nil)
_ fs.CleanUpper = (*Fs)(nil)
_ fs.Abouter = (*Fs)(nil)
_ fs.Object = (*Object)(nil)
)