gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(rpc2): add intercepor of auth

Browse Source 
. #22537366

Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher 2024-08-26 17:27:21 +08:00
parent fb6f396b3e
commit 12dd967d35
20 changed files with 422 additions and 308 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
blobstore/cli/blobnode/disk.go
View File

@ -34,7 +34,7 @@ import (
"github.com/cubefs/cubefs/blobstore/cli/config"
"github.com/cubefs/cubefs/blobstore/common/kvstore"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth"
auth_proto "github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
)
func addCmdDisk(cmd *grumble.Command) {
@ -301,7 +301,7 @@ func newCmClient(c *grumble.Context) *clustermgr.Client {
secret := config.ClusterMgrSecret()
cfg := &clustermgr.Config{}
cfg.LbConfig.Hosts = cmHosts
cfg.LbConfig.Config.Tc.Auth = auth.Config{EnableAuth: secret != "", Secret: secret}
cfg.LbConfig.Config.Tc.Auth = auth_proto.Config{EnableAuth: secret != "", Secret: secret}
return clustermgr.New(cfg)
}

4
blobstore/cli/config/cluster.go
View File

@ -24,7 +24,7 @@ import (
"github.com/cubefs/cubefs/blobstore/cli/common/fmt"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth"
auth_proto "github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
)
// NewConsulClient returns client of consul with address.
@ -85,7 +85,7 @@ func NewCluster(clusterID string, hosts []string, secret string) *cmapi.Client {
Hosts: hosts,
Config: rpc.Config{
Tc: rpc.TransportConfig{
Auth: auth.Config{
Auth: auth_proto.Config{
EnableAuth: secret != "",
Secret: secret,
},

4
blobstore/cli/scheduler/scheduler.go
View File

@ -23,7 +23,7 @@ import (
"github.com/cubefs/cubefs/blobstore/cli/common/fmt"
"github.com/cubefs/cubefs/blobstore/cli/config"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth"
auth_proto "github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
"github.com/cubefs/cubefs/blobstore/scheduler/client"
)
@ -51,7 +51,7 @@ func cmConfig(clusterID proto.ClusterID) *clustermgr.Config {
secret := config.ClusterMgrSecret()
cfg := &clustermgr.Config{}
cfg.LbConfig.Hosts = addrs
cfg.LbConfig.Config.Tc.Auth = auth.Config{EnableAuth: secret != "", Secret: secret}
cfg.LbConfig.Config.Tc.Auth = auth_proto.Config{EnableAuth: secret != "", Secret: secret}
return cfg
}

9
blobstore/cmd/cmd.go
View File

@ -32,6 +32,7 @@ import (
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/rpc/auditlog"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth"
auth_proto "github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
"github.com/cubefs/cubefs/blobstore/util/graceful"
"github.com/cubefs/cubefs/blobstore/util/log"
@ -57,8 +58,8 @@ type Config struct {
BindAddr string `json:"bind_addr"`
ShutdownTimeoutS int `json:"shutdown_timeout_s"`
AuditLog auditlog.Config `json:"auditlog"`
Auth auth.Config `json:"auth"`
AuditLog auditlog.Config `json:"auditlog"`
Auth auth_proto.Config `json:"auth"`
}
type Module struct {
@ -208,14 +209,14 @@ func Main(args []string) {
// 4. the fourth is Auth handler if config,
// 5. others self define handlers by modules.
func reorderMiddleWareHandlers(r *rpc.Router, lh, profileHandler rpc.ProgressHandler,
authCfg auth.Config, handlers []rpc.ProgressHandler,
authCfg auth_proto.Config, handlers []rpc.ProgressHandler,
) (mux http.Handler) {
hs := []rpc.ProgressHandler{lh}
if profileHandler != nil {
hs = append(hs, profileHandler)
}
if authCfg.EnableAuth && authCfg.Secret != "" {
hs = append(hs, auth.NewAuthHandler(&authCfg))
hs = append(hs, auth.New(&authCfg))
}
hs = append(hs, handlers...)
return rpc.MiddlewareHandlerWith(r, hs...)

8
blobstore/common/profile/profile.go
View File

@ -173,15 +173,14 @@ func randomLocalServer() {
}
addr := fmt.Sprintf("127.0.0.1:%d", ln.Addr().(*net.TCPAddr).Port)
genMetricExporter("http://" + addr)
genDumpScript("http://" + addr)
server := &http.Server{
Addr: addr,
Handler: rpc.MiddlewareHandlerWith(rpc.New(), &profileHandler{}),
}
go func() {
server.Serve(ln)
}()
go func() { server.Serve(ln) }()
}
func NewProfileHandler(addr string) rpc.ProgressHandler {
@ -209,7 +208,6 @@ func NewProfileHandler(addr string) rpc.ProgressHandler {
}
serverOnce.Do(randomLocalServer)
return &profileHandler{}
}

4
blobstore/common/rpc/auditlog/decoder.go
View File

@ -23,7 +23,7 @@ import (
"net/http"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth"
auth_proto "github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
)
const maxSeekableBodyLength = 1 << 10
@ -54,7 +54,7 @@ var DefaultRequestHeaderKeys = []string{
"X-Upload-Encoding",
"X-Src",
auth.TokenHeaderKey,
auth_proto.TokenHeaderKey,
}
type DecodedReq struct {

103
blobstore/common/rpc/auth/auth.go
View File

@ -1,103 +0,0 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package auth
import (
"bytes"
"crypto/md5"
"encoding/base64"
"encoding/binary"
"errors"
"net/http"
)
const (
// md5 need 16 byte
TokenKeyLenth = 16
// #nosec G101
TokenHeaderKey = "BLOB-STORE-AUTH-TOKEN"
)
var errMismatchToken = errors.New("mismatch token")
type Config struct {
EnableAuth bool `json:"enable_auth"`
Secret string `json:"secret"`
}
// simply: use timestamp as a token calculate param
type authInfo struct {
timestamp int64
token []byte
// other auth content
others []byte
}
func encodeAuthInfo(info *authInfo) (ret string, err error) {
w := bytes.NewBuffer([]byte{})
if err = binary.Write(w, binary.LittleEndian, &info.timestamp); err != nil {
return
}
if err = binary.Write(w, binary.LittleEndian, &info.token); err != nil {
return
}
return base64.URLEncoding.EncodeToString(w.Bytes()), nil
}
func decodeAuthInfo(encodeStr string) (info *authInfo, err error) {
info = new(authInfo)
b, err := base64.URLEncoding.DecodeString(encodeStr)
if err != nil {
return
}
info.token = make([]byte, TokenKeyLenth)
r := bytes.NewBuffer(b)
if err = binary.Read(r, binary.LittleEndian, &info.timestamp); err != nil {
return
}
if err = binary.Read(r, binary.LittleEndian, &info.token); err != nil {
return
}
return
}
// calculate auth token with params and secret
func calculate(info *authInfo, secret []byte) (err error) {
hash := md5.New()
b := make([]byte, 8)
binary.LittleEndian.PutUint64(b, uint64(info.timestamp))
hash.Write(info.others)
hash.Write(b)
hash.Write(secret)
info.token = hash.Sum(nil)
return
}
// verify auth token with params and secret
func verify(info *authInfo, secret []byte) (err error) {
checkAuthInfo := &authInfo{timestamp: info.timestamp, others: info.others}
calculate(checkAuthInfo, secret)
if !bytes.Equal(checkAuthInfo.token, info.token) {
return errMismatchToken
}
return
}
func genEncodeStr(req *http.Request) []byte {
calStr := req.URL.Path + req.URL.RawQuery
return []byte(calStr)
}

93
blobstore/common/rpc/auth/auth_test.go
View File

@ -1,93 +0,0 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package auth
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strconv"
"testing"
"github.com/stretchr/testify/require"
)
var testServer *httptest.Server
var (
testSecret = "testSecret"
testName = "test"
)
type ret struct {
Name string `json:"name"`
}
func init() {
authHandler := NewAuthHandler(&Config{EnableAuth: true, Secret: testSecret})
http.HandleFunc("/get/name", func(w http.ResponseWriter, r *http.Request) {
authHandler.Handler(w, r, func(w http.ResponseWriter, r *http.Request) {
data, _ := json.Marshal(&ret{Name: testName})
w.Write(data)
})
})
testServer = httptest.NewServer(http.DefaultServeMux)
}
func TestAuth(t *testing.T) {
// invalid secret
tc := NewAuthTransport(&http.Transport{}, &Config{EnableAuth: true, Secret: "wrongSecret"})
client := http.Client{
Transport: tc,
}
req, err := http.NewRequest("POST", testServer.URL+"/get/name?id="+strconv.Itoa(101), nil)
require.NoError(t, err)
response, err := client.Do(req)
require.NoError(t, err)
defer response.Body.Close()
require.Equal(t, http.StatusForbidden, response.StatusCode)
// valid secret
tc = NewAuthTransport(&http.Transport{}, &Config{EnableAuth: true, Secret: testSecret})
client = http.Client{
Transport: tc,
}
result := &ret{}
req, err = http.NewRequest("POST", testServer.URL+"/get/name?id="+strconv.Itoa(101), nil)
require.NoError(t, err)
response, err = client.Do(req)
require.NoError(t, err)
defer response.Body.Close()
require.Equal(t, http.StatusOK, response.StatusCode)
err = json.NewDecoder(response.Body).Decode(result)
require.NoError(t, err)
require.Equal(t, testName, result.Name)
// test empty token
c := http.Client{}
req, err = http.NewRequest("POST", testServer.URL+"/get/name?id="+strconv.Itoa(101), nil)
require.NoError(t, err)
response, err = c.Do(req)
require.NoError(t, err)
require.Equal(t, http.StatusForbidden, response.StatusCode)
response.Body.Close()
// test token failed
req.Header.Set(TokenHeaderKey, "#$@%DF#$@#$")
response, err = c.Do(req)
require.NoError(t, err)
require.Equal(t, http.StatusForbidden, response.StatusCode)
response.Body.Close()
}

57
blobstore/common/rpc/auth/authhandler.go
View File

@ -1,57 +0,0 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package auth
import (
"net/http"
)
type AuthHandler struct {
Secret []byte
}
func NewAuthHandler(cfg *Config) *AuthHandler {
if cfg.EnableAuth {
if cfg.Secret == "" {
panic("auth secret can not be nil")
}
return &AuthHandler{
Secret: []byte(cfg.Secret),
}
}
return nil
}
func (self *AuthHandler) Handler(w http.ResponseWriter, req *http.Request, f func(http.ResponseWriter, *http.Request)) {
token := req.Header.Get(TokenHeaderKey)
if token == "" {
w.WriteHeader(http.StatusForbidden)
return
}
info, err := decodeAuthInfo(token)
if err != nil {
w.WriteHeader(http.StatusForbidden)
return
}
info.others = genEncodeStr(req)
err = verify(info, self.Secret)
if err != nil && err == errMismatchToken {
w.WriteHeader(http.StatusForbidden)
return
}
f(w, req)
}

56
blobstore/common/rpc/auth/handler.go Normal file
View File

@ -0,0 +1,56 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package auth
import (
"net/http"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc2"
)
type handler struct {
Secret []byte
}
func New(cfg *proto.Config) interface {
rpc2.Interceptor
rpc.ProgressHandler
} {
if !cfg.EnableAuth || cfg.Secret == "" {
panic("auth secret can not be empty")
}
return &handler{Secret: []byte(cfg.Secret)}
}
func (h *handler) Handler(w http.ResponseWriter, req *http.Request, f func(http.ResponseWriter, *http.Request)) {
if err := proto.Decode(req.Header.Get(proto.TokenHeaderKey), proto.ParamFromRequest(req), h.Secret); err != nil {
w.WriteHeader(http.StatusForbidden)
return
}
f(w, req)
}
func (h *handler) Handle(w rpc2.ResponseWriter, req *rpc2.Request, f rpc2.Handle) error {
if err := proto.Decode(req.Header.Get(proto.TokenHeaderKey), []byte(req.RemotePath), h.Secret); err != nil {
return &rpc2.Error{
Status: http.StatusForbidden,
Reason: "Auth",
Detail: err.Error(),
}
}
return f(w, req)
}

83
blobstore/common/rpc/auth/handler_test.go Normal file
View File

@ -0,0 +1,83 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package auth
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc2"
)
var (
testServer *httptest.Server
testSecret = "testSecret"
authHandler interface {
rpc2.Interceptor
rpc.ProgressHandler
}
)
func init() {
authHandler = New(&proto.Config{EnableAuth: true, Secret: testSecret})
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
authHandler.Handler(w, r, func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(200)
})
})
testServer = httptest.NewServer(http.DefaultServeMux)
}
func TestHandlerRpc(t *testing.T) {
require.Panics(t, func() { New(&proto.Config{EnableAuth: false, Secret: testSecret}) })
require.Panics(t, func() { New(&proto.Config{EnableAuth: true, Secret: ""}) })
client := http.Client{}
req, _ := http.NewRequest("POST", testServer.URL+"/", nil)
token := proto.Encode(time.Now().Unix(), proto.ParamFromRequest(req), []byte("wrongSecret"))
req.Header.Set(proto.TokenHeaderKey, token)
response, _ := client.Do(req)
require.Equal(t, http.StatusForbidden, response.StatusCode)
response.Body.Close()
req, _ = http.NewRequest("POST", testServer.URL+"/", nil)
token = proto.Encode(time.Now().Unix(), proto.ParamFromRequest(req), []byte(testSecret))
req.Header.Set(proto.TokenHeaderKey, token)
response, _ = client.Do(req)
require.Equal(t, http.StatusOK, response.StatusCode)
response.Body.Close()
testServer.Close()
}
func TestHandlerRpc2(t *testing.T) {
req, _ := rpc2.NewRequest(context.Background(), "/", "/path", nil, nil)
token := proto.Encode(time.Now().Unix(), []byte(req.RemotePath), []byte("wrongSecret"))
req.Header.Set(proto.TokenHeaderKey, token)
err := authHandler.Handle(nil, req, func(rpc2.ResponseWriter, *rpc2.Request) error { return nil })
require.Equal(t, http.StatusForbidden, rpc.DetectStatusCode(err))
token = proto.Encode(time.Now().Unix(), []byte(req.RemotePath), []byte(testSecret))
req.Header.Set(proto.TokenHeaderKey, token)
err = authHandler.Handle(nil, req, func(rpc2.ResponseWriter, *rpc2.Request) error { return nil })
require.Equal(t, http.StatusOK, rpc.DetectStatusCode(err))
}

93
blobstore/common/rpc/auth/proto/auth.go Normal file
View File

@ -0,0 +1,93 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package proto
import (
"bytes"
"crypto/md5"
"encoding/base64"
"encoding/binary"
"errors"
"net/http"
)
const (
tokenLenth = 8 + md5.Size
// #nosec G101
TokenHeaderKey = "BLOB-STORE-AUTH-TOKEN"
)
var errMismatchToken = errors.New("mismatch token")
type Config struct {
EnableAuth bool `json:"enable_auth"`
Secret string `json:"secret"`
}
// token simply: use timestamp as a token calculate param
type token struct {
param []byte
timestamp [8]byte
md5sum [md5.Size]byte
}
func Encode(timestamp int64, param, secret []byte) string {
var t token
t.param = param
binary.LittleEndian.PutUint64(t.timestamp[:], uint64(timestamp))
t.calculate(secret)
return t.encode()
}
func Decode(tokenStr string, param, secret []byte) error {
b, err := base64.URLEncoding.DecodeString(tokenStr)
if err != nil {
return err
}
if len(b) != tokenLenth {
return errMismatchToken
}
t := new(token)
t.param = param
copy(t.timestamp[:], b[:8])
t.calculate(secret)
if !bytes.Equal(t.md5sum[:], b[8:]) {
return errMismatchToken
}
return nil
}
func (t *token) encode() string {
w := bytes.NewBuffer(make([]byte, 0, tokenLenth))
w.Write(t.timestamp[:])
w.Write(t.md5sum[:])
return base64.URLEncoding.EncodeToString(w.Bytes())
}
// calculate auth token with params and secret
func (t *token) calculate(secret []byte) {
hasher := md5.New()
hasher.Write(t.param)
hasher.Write(t.timestamp[:])
hasher.Write(secret)
copy(t.md5sum[:], hasher.Sum(nil))
}
func ParamFromRequest(req *http.Request) []byte {
return []byte(req.URL.Path + req.URL.RawQuery)
}

41
blobstore/common/rpc/auth/proto/auth_test.go Normal file
View File

@ -0,0 +1,41 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package proto
import (
"encoding/base64"
"net/http"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func TestAuth(t *testing.T) {
secret := []byte("testSecret")
timestamp := time.Now().Unix()
req, err := http.NewRequest("POST", "http://127.0.0.1:80/get/name", nil)
require.NoError(t, err)
param := ParamFromRequest(req)
encoded := Encode(timestamp, param, secret)
require.NoError(t, Decode(encoded, param, secret))
require.Error(t, Decode(encoded, nil, secret))
require.Error(t, Decode(encoded, param, nil))
require.Error(t, Decode(encoded[:6], param, secret))
require.Error(t, Decode(base64.URLEncoding.EncodeToString(make([]byte, tokenLenth-1)), param, secret))
}

39
blobstore/common/rpc/auth/authtransport.go → blobstore/common/rpc/auth/transport/transport.go
View File

@ -17,42 +17,31 @@ package auth
import (
"net/http"
"time"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
)
type AuthTransport struct {
type transport struct {
Secret []byte
Tr http.RoundTripper
}
func NewAuthTransport(tr http.RoundTripper, cfg *Config) http.RoundTripper {
func New(tr http.RoundTripper, cfg *proto.Config) http.RoundTripper {
if cfg.EnableAuth {
if cfg.Secret == "" {
panic("auth secret can not be nil")
}
return &AuthTransport{
Secret: []byte(cfg.Secret),
Tr: tr,
if cfg.Secret != "" {
return &transport{
Secret: []byte(cfg.Secret),
Tr: tr,
}
}
}
return nil
return tr
}
// a simple auth token
func (self *AuthTransport) RoundTrip(req *http.Request) (resp *http.Response, err error) {
func (t *transport) RoundTrip(req *http.Request) (resp *http.Response, err error) {
now := time.Now().Unix()
info := &authInfo{timestamp: now, others: genEncodeStr(req)}
err = calculate(info, self.Secret)
if err != nil {
return self.Tr.RoundTrip(req)
}
token, err := encodeAuthInfo(info)
if err != nil {
return self.Tr.RoundTrip(req)
}
req.Header.Set(TokenHeaderKey, token)
return self.Tr.RoundTrip(req)
param := proto.ParamFromRequest(req)
req.Header.Set(proto.TokenHeaderKey, proto.Encode(now, param, t.Secret))
return t.Tr.RoundTrip(req)
}

81
blobstore/common/rpc/auth/transport/transport_test.go Normal file
View File

@ -0,0 +1,81 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package auth
import (
"net/http"
"net/http/httptest"
"testing"
"github.com/stretchr/testify/require"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
)
var (
testServer *httptest.Server
testSecret = "testSecret"
)
func init() {
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
token := r.Header.Get(proto.TokenHeaderKey)
if token == "" {
w.WriteHeader(http.StatusForbidden)
return
}
if err := proto.Decode(token, proto.ParamFromRequest(r), []byte(testSecret)); err != nil {
w.WriteHeader(http.StatusForbidden)
return
}
w.WriteHeader(200)
})
testServer = httptest.NewServer(http.DefaultServeMux)
}
func TestTransport(t *testing.T) {
// invalid secret
tc := New(&http.Transport{}, &proto.Config{EnableAuth: true, Secret: "wrongSecret"})
client := http.Client{Transport: tc}
req, _ := http.NewRequest("POST", testServer.URL+"/", nil)
response, _ := client.Do(req)
require.Equal(t, http.StatusForbidden, response.StatusCode)
response.Body.Close()
// valid secret
tc = New(&http.Transport{}, &proto.Config{EnableAuth: true, Secret: testSecret})
client = http.Client{Transport: tc}
req, _ = http.NewRequest("POST", testServer.URL+"/", nil)
response, _ = client.Do(req)
require.Equal(t, http.StatusOK, response.StatusCode)
response.Body.Close()
// test empty token
tc = New(&http.Transport{}, &proto.Config{EnableAuth: true, Secret: ""})
client = http.Client{Transport: tc}
req, _ = http.NewRequest("POST", testServer.URL+"/", nil)
response, _ = client.Do(req)
require.Equal(t, http.StatusForbidden, response.StatusCode)
response.Body.Close()
// test token failed
req, _ = http.NewRequest("POST", testServer.URL+"/", nil)
req.Header.Set(proto.TokenHeaderKey, "#$@%DF#$@#$")
response, _ = client.Do(req)
require.Equal(t, http.StatusForbidden, response.StatusCode)
response.Body.Close()
testServer.Close()
}

16
blobstore/common/rpc/client_config.go
View File

@ -19,7 +19,8 @@ import (
"net/http"
"time"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth"
auth_proto "github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
auth_transport "github.com/cubefs/cubefs/blobstore/common/rpc/auth/transport"
)
// TransportConfig http transport config
@ -41,14 +42,14 @@ type TransportConfig struct {
DisableCompression bool `json:"disable_compression"`
// auth config
Auth auth.Config `json:"auth"`
Auth auth_proto.Config `json:"auth"`
}
// Default returns default transport if none setting.
// Disable Auth config.
func (tc TransportConfig) Default() TransportConfig {
noAuth := tc
noAuth.Auth = auth.Config{}
noAuth.Auth = auth_proto.Config{}
none := TransportConfig{}
if noAuth == none {
return TransportConfig{
@ -81,12 +82,5 @@ func NewTransport(cfg *TransportConfig) http.RoundTripper {
Timeout: time.Duration(cfg.DialTimeoutMs) * time.Millisecond,
KeepAlive: 30 * time.Second,
}).DialContext
if cfg.Auth.EnableAuth {
authTr := auth.NewAuthTransport(tr, &cfg.Auth)
if authTr != nil {
return authTr
}
}
return tr
return auth_transport.New(tr, &cfg.Auth)
}

4
blobstore/common/rpc/client_test.go
View File

@ -26,7 +26,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/cubefs/cubefs/blobstore/common/crc32block"
"github.com/cubefs/cubefs/blobstore/common/rpc/auth"
auth_proto "github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
)
var (
@ -123,7 +123,7 @@ var simpleCfg = &Config{
MaxIdleConnsPerHost: 10,
IdleConnTimeoutMs: 60000,
DisableCompression: true,
Auth: auth.Config{
Auth: auth_proto.Config{
EnableAuth: true,
Secret: "test",
},

7
blobstore/common/rpc2/client.go
View File

@ -22,6 +22,7 @@ import (
"time"
"github.com/cubefs/cubefs/blobstore/common/rpc"
auth_proto "github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
"github.com/cubefs/cubefs/blobstore/util/defaulter"
"github.com/cubefs/cubefs/blobstore/util/retry"
)
@ -39,6 +40,8 @@ type Client struct {
RequestTimeout time.Duration
ResponseTimeout time.Duration
Auth auth_proto.Config
Selector rpc.Selector // lb client
LbConfig struct {
Hosts []string
@ -71,6 +74,10 @@ func (c *Client) Do(req *Request, ret Unmarshaler) (resp *Response, err error) {
return nil, ErrConnNoAddress
}
if c.Auth.EnableAuth && c.Auth.Secret != "" {
req.Header.Set(auth_proto.TokenHeaderKey, auth_proto.Encode(time.Now().Unix(),
[]byte(req.RemotePath), []byte(c.Auth.Secret)))
}
for _, opt := range req.opts {
opt(req)
}

23
blobstore/common/rpc2/client_test.go
View File

@ -21,6 +21,7 @@ import (
"strings"
"testing"
auth_proto "github.com/cubefs/cubefs/blobstore/common/rpc/auth/proto"
"github.com/stretchr/testify/require"
)
@ -87,3 +88,25 @@ func TestClientLbSelector(t *testing.T) {
}
cli.Close()
}
func handleClientAuth(w ResponseWriter, req *Request) error {
secret := req.Header.Get("token-secret")
return auth_proto.Decode(req.Header.Get(auth_proto.TokenHeaderKey),
[]byte(req.RemotePath), []byte(secret))
}
func TestClientAuth(t *testing.T) {
var handler Router
handler.Register("/", handleClientAuth)
server, cli, shutdown := newServer("tcp", &handler)
defer shutdown()
cli.Auth.EnableAuth = true
cli.Auth.Secret = "secret"
req, _ := NewRequest(testCtx, server.Name, "/", nil, nil)
require.Error(t, cli.DoWith(req, nil))
req, _ = NewRequest(testCtx, server.Name, "/", nil, nil)
req.Header.Set("token-secret", cli.Auth.Secret)
require.NoError(t, cli.DoWith(req, nil))
}

1
blobstore/common/rpc2/server.go
View File

@ -296,6 +296,7 @@ func (s *Server) handleStream(stream *transport.Stream) {
}
}
resp.WriteOK(nil)
if err = resp.Flush(); err != nil {
return err
}