fix(slave): compensate master clock skew on signature expiry (#3242)

Slave-to-master API signatures carry a 60s TTL computed from the slave's
local clock; when the slave clock lagged the master by more than that,
every signed request - including the transfer task's upload-session
creation - arrived already expired and transfers failed with "signature
expired". The remote client now detects CodeSignExpired responses,
learns the master-minus-local offset from the response Date header,
retries the request with a compensated signature base time, and applies
the offset to all subsequent signed requests on that client.

Authored By: TDvorak <info@tdvorak.dev>

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
pull/3582/head
Tomas Dvorak 2 weeks ago
parent 8d226d2d07
commit 93e11533c0

@ -10,6 +10,7 @@ import (
"net/http" "net/http"
"net/url" "net/url"
"strings" "strings"
"sync/atomic"
"time" "time"
"github.com/cloudreve/Cloudreve/v4/application/constants" "github.com/cloudreve/Cloudreve/v4/application/constants"
@ -91,6 +92,37 @@ type remoteClient struct {
httpClient request.Client httpClient request.Client
settings setting.Provider settings setting.Provider
l logging.Logger l logging.Logger
// clockOffset compensates master/slave clock skew, learned from a
// signature-expired response's Date header (upstream #3242).
clockOffset atomic.Int64
}
// request sends a signed slave API request, applying the learned clock offset
// to the signature base time so skewed master/slave clocks do not produce
// expired signatures.
func (c *remoteClient) request(method, target string, body io.Reader, opts ...request.Option) *request.Response {
if off := c.clockOffset.Load(); off != 0 {
opts = append(opts, request.WithSignBaseTime(time.Now().Add(time.Duration(off)*time.Second)))
}
return c.httpClient.Request(method, target, body, opts...)
}
// learnClockOffset stores the master-minus-local clock offset computed from
// the response Date header.
func (c *remoteClient) learnClockOffset(resp *http.Response) {
if resp == nil {
return
}
masterTime, err := http.ParseTime(resp.Header.Get("Date"))
if err != nil {
return
}
// Date has second precision; round up so truncation never under-compensates.
offset := int64(masterTime.Sub(time.Now()).Seconds()) + 1
if offset != 0 {
c.l.Info("Compensating master clock offset of %ds after signature expiry", offset)
c.clockOffset.Store(offset)
}
} }
func (c *remoteClient) Upload(ctx context.Context, file *fs.UploadRequest) error { func (c *remoteClient) Upload(ctx context.Context, file *fs.UploadRequest) error {
@ -133,7 +165,7 @@ func (c *remoteClient) Upload(ctx context.Context, file *fs.UploadRequest) error
} }
func (c *remoteClient) DeleteUploadSession(ctx context.Context, sessionID string) error { func (c *remoteClient) DeleteUploadSession(ctx context.Context, sessionID string) error {
resp, err := c.httpClient.Request( resp, err := c.request(
"DELETE", "DELETE",
"upload/"+sessionID, "upload/"+sessionID,
nil, nil,
@ -161,7 +193,7 @@ func (c *remoteClient) DeleteFiles(ctx context.Context, files ...string) ([]stri
return files, fmt.Errorf("failed to marshal delete request: %w", err) return files, fmt.Errorf("failed to marshal delete request: %w", err)
} }
resp, err := c.httpClient.Request( resp, err := c.request(
"DELETE", "DELETE",
"file", "file",
bytes.NewReader(reqStr), bytes.NewReader(reqStr),
@ -185,7 +217,7 @@ func (c *remoteClient) DeleteFiles(ctx context.Context, files ...string) ([]stri
} }
func (c *remoteClient) MediaMeta(ctx context.Context, src, ext, language string) ([]driver.MediaMeta, error) { func (c *remoteClient) MediaMeta(ctx context.Context, src, ext, language string) ([]driver.MediaMeta, error) {
resp, err := c.httpClient.Request( resp, err := c.request(
http.MethodGet, http.MethodGet,
routes.SlaveMediaMetaRoute(src, ext, language), routes.SlaveMediaMetaRoute(src, ext, language),
nil, nil,
@ -214,14 +246,33 @@ func (c *remoteClient) CreateUploadSession(ctx context.Context, session *fs.Uplo
return err return err
} }
bodyReader := strings.NewReader(string(reqBodyEncoded)) send := func() (*serializer.Response, error) {
resp, err := c.httpClient.Request( raw := c.request(
"PUT", "PUT",
"upload", "upload",
bodyReader, strings.NewReader(string(reqBodyEncoded)),
request.WithContext(ctx), request.WithContext(ctx),
request.WithLogger(c.l), request.WithLogger(c.l),
).CheckHTTPResponse(200).DecodeResponse() )
httpResp := raw.Response
resp, err := raw.CheckHTTPResponse(200).DecodeResponse()
if err == nil && resp.Code == serializer.CodeSignExpired {
// Master clock is ahead of ours beyond the sign TTL; learn the
// offset from the response Date header and retry once (#3242).
c.learnClockOffset(httpResp)
raw = c.request(
"PUT",
"upload",
strings.NewReader(string(reqBodyEncoded)),
request.WithContext(ctx),
request.WithLogger(c.l),
)
resp, err = raw.CheckHTTPResponse(200).DecodeResponse()
}
return resp, err
}
resp, err := send()
if err != nil { if err != nil {
return err return err
} }
@ -234,7 +285,7 @@ func (c *remoteClient) CreateUploadSession(ctx context.Context, session *fs.Uplo
} }
func (c *remoteClient) List(ctx context.Context, path string, recursive bool) ([]fs.PhysicalObject, error) { func (c *remoteClient) List(ctx context.Context, path string, recursive bool) ([]fs.PhysicalObject, error) {
resp, err := c.httpClient.Request( resp, err := c.request(
http.MethodGet, http.MethodGet,
routes.SlaveFileListRoute(path, recursive), routes.SlaveFileListRoute(path, recursive),
nil, nil,
@ -271,7 +322,7 @@ func (c *remoteClient) GetUploadURL(ctx context.Context, expires time.Time, sess
} }
func (c *remoteClient) uploadChunk(ctx context.Context, sessionID string, index int, chunk io.Reader, overwrite bool, size int64) error { func (c *remoteClient) uploadChunk(ctx context.Context, sessionID string, index int, chunk io.Reader, overwrite bool, size int64) error {
resp, err := c.httpClient.Request( resp, err := c.request(
"POST", "POST",
fmt.Sprintf("upload/%s?chunk=%d", sessionID, index), fmt.Sprintf("upload/%s?chunk=%d", sessionID, index),
chunk, chunk,

@ -0,0 +1,76 @@
package remote
import (
"context"
"net/http"
"net/http/httptest"
"strconv"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/cloudreve/Cloudreve/v4/pkg/auth"
"github.com/cloudreve/Cloudreve/v4/pkg/filemanager/fs"
"github.com/cloudreve/Cloudreve/v4/pkg/logging"
"github.com/cloudreve/Cloudreve/v4/pkg/request"
"github.com/stretchr/testify/require"
)
// TestCreateUploadSessionClockSkewRetry reproduces upstream #3242: when the
// slave's clock is behind the master's, the upload-session signature arrives
// already expired. The client must learn the offset from the response Date
// header and retry with a compensated base time.
func TestCreateUploadSessionClockSkewRetry(t *testing.T) {
const skewSeconds = 3600 // master is 1h ahead of slave
masterNow := func() time.Time { return time.Now().Add(skewSeconds * time.Second) }
var calls atomic.Int32
var lastExpires int64
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
calls.Add(1)
authHeader := r.Header.Get("Authorization")
parts := strings.Split(authHeader, ":")
expires, _ := strconv.ParseInt(parts[len(parts)-1], 10, 64)
lastExpires = expires
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Date", masterNow().UTC().Format(http.TimeFormat))
if expires < masterNow().Unix() {
w.Write([]byte(`{"code":40005,"msg":"signature expired"}`))
return
}
w.Write([]byte(`{"code":0}`))
}))
defer srv.Close()
c := &remoteClient{
l: logging.NewConsoleLogger(logging.LevelError),
httpClient: request.NewClient(
nil,
request.WithEndpoint(srv.URL),
request.WithCredential(auth.HMACAuth{SecretKey: []byte("test")}, 60),
),
}
err := c.CreateUploadSession(context.Background(), &fs.UploadSession{}, false)
require.NoError(t, err)
require.Equal(t, int32(2), calls.Load(), "expected one expired attempt plus one compensated retry")
require.InDelta(t, skewSeconds, c.clockOffset.Load(), 1)
require.GreaterOrEqual(t, lastExpires, masterNow().Unix())
}
// TestLearnClockOffsetIgnoresBadHeaders ensures malformed or missing Date
// headers leave the offset untouched.
func TestLearnClockOffsetIgnoresBadHeaders(t *testing.T) {
c := &remoteClient{l: logging.NewConsoleLogger(logging.LevelError)}
c.learnClockOffset(nil)
require.Zero(t, c.clockOffset.Load())
c.learnClockOffset(&http.Response{Header: http.Header{}})
require.Zero(t, c.clockOffset.Load())
c.learnClockOffset(&http.Response{Header: http.Header{"Date": {"garbage"}}})
require.Zero(t, c.clockOffset.Load())
}

@ -20,6 +20,7 @@ type options struct {
header http.Header header http.Header
sign auth.Auth sign auth.Auth
signTTL int64 signTTL int64
signBaseTime time.Time
ctx context.Context ctx context.Context
contentLength int64 contentLength int64
masterMeta bool masterMeta bool
@ -88,6 +89,15 @@ func WithCredential(instance auth.Auth, ttl int64) Option {
}) })
} }
// WithSignBaseTime overrides the "now" used to compute the signature expiry
// timestamp. Slave clients use it to compensate clock skew against the master
// once the offset is learned from a signature-expired response (#3242).
func WithSignBaseTime(t time.Time) Option {
return optionFunc(func(o *options) {
o.signBaseTime = t
})
}
// WithHeader 设置请求Header // WithHeader 设置请求Header
func WithHeader(header http.Header) Option { func WithHeader(header http.Header) Option {
return optionFunc(func(o *options) { return optionFunc(func(o *options) {

@ -176,7 +176,11 @@ func (c *HTTPClient) Request(method, target string, body io.Reader, opts ...Opti
if options.ctx == nil { if options.ctx == nil {
ctx = context.Background() ctx = context.Background()
} }
expire := time.Now().Add(time.Second * time.Duration(options.signTTL)) expireBase := options.signBaseTime
if expireBase.IsZero() {
expireBase = time.Now()
}
expire := expireBase.Add(time.Second * time.Duration(options.signTTL))
switch method { switch method {
case "PUT", "POST", "PATCH": case "PUT", "POST", "PATCH":
auth.SignRequest(ctx, options.sign, req, &expire) auth.SignRequest(ctx, options.sign, req, &expire)

@ -6,9 +6,12 @@ import (
"github.com/cloudreve/Cloudreve/v4/pkg/auth" "github.com/cloudreve/Cloudreve/v4/pkg/auth"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
testMock "github.com/stretchr/testify/mock" testMock "github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
"io" "io"
"io/ioutil" "io/ioutil"
"net/http" "net/http"
"net/http/httptest"
"strconv"
"strings" "strings"
"testing" "testing"
"time" "time"
@ -271,3 +274,43 @@ func TestHTTPClient_TPSLimit(t *testing.T) {
} }
} }
func TestWithSignBaseTime(t *testing.T) {
asserts := assert.New(t)
options := newDefaultOption()
asserts.True(options.signBaseTime.IsZero())
future := time.Now().Add(time.Hour)
WithSignBaseTime(future).apply(options)
asserts.Equal(future, options.signBaseTime)
}
func TestSignBaseTimeShiftsExpiry(t *testing.T) {
asserts := assert.New(t)
var authHeader string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
authHeader = r.Header.Get("Authorization")
w.WriteHeader(200)
_, _ = w.Write([]byte(`{"code":0}`))
}))
defer srv.Close()
client := NewClientDeprecated(WithEndpoint(srv.URL))
base := time.Now().Add(2 * time.Hour)
resp := client.Request(
"POST",
"/upload",
strings.NewReader("{}"),
WithCredential(auth.HMACAuth{SecretKey: []byte("k")}, 60),
WithSignBaseTime(base),
)
asserts.NoError(resp.Err)
// Authorization: Cr <base64sign>:<expires>
parts := strings.Split(authHeader, ":")
require.GreaterOrEqual(t, len(parts), 2)
expires, err := strconv.ParseInt(parts[len(parts)-1], 10, 64)
asserts.NoError(err)
asserts.InDelta(base.Add(60*time.Second).Unix(), expires, 5)
}

Loading…
Cancel
Save