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

Feature: support ip addr acl check

Browse Source 
Signed-off-by: leonrayang <chl696@sina.com>
This commit is contained in:
leonrayang 2022-12-18 01:21:34 +08:00 committed by leonrayang
parent 53a8d282c6
commit b525c104a5
17 changed files with 482 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

161
cli/cmd/acl.go Normal file
View File

@ -0,0 +1,161 @@
package cmd
import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/master"
"github.com/cubefs/cubefs/util"
"github.com/spf13/cobra"
)
const (
cmdAclUse = "acl [COMMAND]"
cmdAclShort = "Manage cluster volumes acl black list"
cmdAclAddShort = "add volume acl ip"
cmdAclDelShort = "del volume acl ip"
cmdAclListShort = "list volume acl ip list"
cmdAclCheckShort = "check volume acl ip"
//acl op
CliAclAdd = "aclAdd"
cliAclListShort = "aclList"
CliAclDel = "aclDel"
CliAclCheck = "aclCheck"
)
func newAclCmd(client *master.MasterClient) *cobra.Command {
var cmd = &cobra.Command{
Use: cmdAclUse,
Short: cmdAclShort,
Args: cobra.MinimumNArgs(0),
Aliases: []string{"acl"},
}
cmd.AddCommand(
newAclAddCmd(client),
newAclDelCmd(client),
newAclListCmd(client),
newAclCheckCmd(client),
)
return cmd
}
func newAclAddCmd(client *master.MasterClient) *cobra.Command {
var optKeyword string
var cmd = &cobra.Command{
Use: CliAclAdd,
Short: cmdAclAddShort,
Aliases: []string{"add"},
Run: func(cmd *cobra.Command, args []string) {
if len(args) <= 1 {
stdout("example:cfs-cli acl aclAdd volName 192.168.0.1\n")
return
}
var err error
defer func() {
if err != nil {
errout("Error: %v", err)
}
}()
var aclInfo *proto.AclRsp
if aclInfo, err = client.UserAPI().AclOperation(args[0], args[1], util.AclAddIP); err != nil || !aclInfo.OK {
return
}
stdout("success!\n")
},
}
cmd.Flags().StringVar(&optKeyword, "keyword", "", "Specify keyword of volume name to filter")
return cmd
}
func newAclListCmd(client *master.MasterClient) *cobra.Command {
var optKeyword string
var cmd = &cobra.Command{
Use: cliAclListShort,
Short: cmdAclListShort,
Aliases: []string{"list"},
Run: func(cmd *cobra.Command, args []string) {
var volumeName = args[0]
var err error
if len(args) == 0 {
stdout("need volume name\n")
return
}
defer func() {
if err != nil {
errout("Error: %v", err)
}
}()
var aclInfo *proto.AclRsp
if aclInfo, err = client.UserAPI().AclOperation(volumeName, "", util.AclListIP); err != nil || !aclInfo.OK {
stdout("AclOperation return \n")
return
}
stdout("%v\n", volumeAclTableHeader)
for _, info := range aclInfo.List {
stdout("%v\n", formatAclInfoTableRow(info))
}
},
}
cmd.Flags().StringVar(&optKeyword, "keyword", "", "Specify keyword of volume name to filter")
return cmd
}
func newAclDelCmd(client *master.MasterClient) *cobra.Command {
var optKeyword string
var cmd = &cobra.Command{
Use: CliAclDel,
Short: cmdAclDelShort,
Aliases: []string{"del"},
Run: func(cmd *cobra.Command, args []string) {
if len(args) <= 1 {
stdout("USAGE:./cfs-cli acl aclDel volName ipAddr\n")
return
}
var err error
defer func() {
if err != nil {
errout("Error: %v", err)
}
}()
var aclInfo *proto.AclRsp
if aclInfo, err = client.UserAPI().AclOperation(args[0], args[1], util.AclDelIP); err != nil || !aclInfo.OK {
return
}
stdout("success!\n")
},
}
cmd.Flags().StringVar(&optKeyword, "keyword", "", "Specify keyword of volume name to filter")
return cmd
}
func newAclCheckCmd(client *master.MasterClient) *cobra.Command {
var optKeyword string
var cmd = &cobra.Command{
Use: CliAclCheck,
Short: cmdAclCheckShort,
Aliases: []string{"check"},
Run: func(cmd *cobra.Command, args []string) {
if len(args) <= 1 {
stdout("USAGE:./cfs-cli acl aclCheck volName ipAddr\n")
return
}
var err error
defer func() {
if err != nil {
errout("Error: %v", err)
}
}()
var aclInfo *proto.AclRsp
if aclInfo, err = client.UserAPI().AclOperation(args[0], args[1], util.AclCheckIP); err != nil || !aclInfo.OK {
return
}
stdout("%v\n", volumeAclTableHeader)
for _, info := range aclInfo.List {
stdout("%v\n", formatAclInfoTableRow(info))
}
},
}
cmd.Flags().StringVar(&optKeyword, "keyword", "", "Specify keyword of volume name to filter")
return cmd
}

15
cli/cmd/fmt.go
View File

@ -160,6 +160,21 @@ func formatVolInfoTableRow(vi *proto.VolInfo) string {
formatVolumeStatus(vi.Status), time.Unix(vi.CreateTime, 0).Local().Format(time.RFC1123))
}
var (
volumeVersionPattern = "%-20v %-40v %-8v %-8v"
volumeVersionTableHeader = fmt.Sprintf(volumeVersionPattern, "VER", "CTIME", "STATUS", "OTHER")
)
var (
volumeAclPattern = "%-20v %-40v %-8v"
volumeAclTableHeader = fmt.Sprintf(volumeAclPattern, "IP", "CTIME", "OTHER")
)
func formatAclInfoTableRow(aclInfo *proto.AclIpInfo) string {
return fmt.Sprintf(volumeAclPattern,
aclInfo.Ip, time.Unix(aclInfo.CTime, 0).Format(time.RFC1123), "")
}
var (
dataPartitionTablePattern = "%-8v %-8v %-10v %-10v %-18v %-18v"
dataPartitionTableHeader = fmt.Sprintf(dataPartitionTablePattern,

1
cli/cmd/root.go
View File

@ -61,6 +61,7 @@ func NewRootCmd(client *master.MasterClient) *CubeFSCmd {
newMetaPartitionCmd(client),
newConfigCmd(),
newZoneCmd(client),
newAclCmd(client),
)
return cmd
}

7
client/fuse.go
View File

@ -375,7 +375,6 @@ func main() {
}
registerInterceptedSignal(opt.MountPoint)
if err = checkPermission(opt); err != nil {
err = errors.NewErrorf("check permission failed: %v", err)
syslog.Println(err)
@ -745,7 +744,11 @@ func parseMountOption(cfg *config.Config) (*proto.MountOptions, error) {
func checkPermission(opt *proto.MountOptions) (err error) {
var mc = master.NewMasterClientFromString(opt.Master, false)
localIP, _ := ump.GetLocalIpAddr()
if info, err := mc.UserAPI().AclOperation(opt.Volname, localIP, util.AclCheckIP); err != nil || !info.OK {
syslog.Println(err)
return proto.ErrNoAclPermission
}
// Check user access policy is enabled
if opt.AccessKey != "" {
var userInfo *proto.UserInfo

70
master/api_service.go
View File

@ -283,6 +283,76 @@ func (m *Server) clusterStat(w http.ResponseWriter, r *http.Request) {
sendOkReply(w, r, newSuccessHTTPReply(cs))
}
func (m *Server) aclOperate(w http.ResponseWriter, r *http.Request) {
var (
ip string
err error
volName string
vol *Vol
op uint64
value string
ok, res bool
ipList []*proto.AclIpInfo
)
if volName, err = extractName(r); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
if value = r.FormValue(OperateKey); value == "" {
err = keyNotFound(OperateKey)
sendErrReply(w, r, newErrHTTPReply(err))
}
op, err = strconv.ParseUint(value, 10, 64)
if err != nil {
err = fmt.Errorf("parseUintParam %s-%s is not legal, err %s", OperateKey, value, err.Error())
return
}
if op != util.AclListIP {
if ip = r.FormValue(IPKey); ip == "" {
err = keyNotFound(IPKey)
sendErrReply(w, r, newErrHTTPReply(err))
}
}
log.LogDebugf("aclOperate. name %v op %v ip %v", volName, op, ip)
if vol, err = m.cluster.getVol(volName); err != nil {
log.LogDebugf("aclOperate. name %v not found", volName)
sendErrReply(w, r, newErrHTTPReply(err))
return
}
ok = true
opAclRes := vol.aclMgr.aclOperate(op, ip)
switch op {
case util.AclCheckIP:
if ipList, res = opAclRes.([]*proto.AclIpInfo); !res {
sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("inner error")))
return
}
case util.AclAddIP, util.AclDelIP:
if opAclRes != nil {
if err, res = opAclRes.(error); !res {
sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("inner error")))
return
}
}
case util.AclListIP:
if ipList, res = opAclRes.([]*proto.AclIpInfo); !res {
sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("inner error")))
return
}
}
rsp := &proto.AclRsp{
OK: ok,
List: ipList,
}
_ = sendOkReply(w, r, newSuccessHTTPReply(rsp))
return
}
func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminGetCluster))
defer func() {

5
master/cluster.go
View File

@ -487,6 +487,9 @@ func (c *Cluster) scheduleToCheckHeartbeat() {
}
}()
}
func (c *Cluster) passAclCheck(ip string) {
}
func (c *Cluster) checkLeaderAddr() {
leaderID, _ := c.partition.LeaderTerm()
@ -2707,6 +2710,8 @@ func (c *Cluster) createVol(req *createVolReq) (vol *Vol, err error) {
goto errHandler
}
vol.aclMgr.init(c, vol)
if err = vol.initMetaPartitions(c, req.mpCount); err != nil {
vol.Status = markDelete

5
master/const.go
View File

@ -95,6 +95,9 @@ const (
TimeOut = "timeout"
CountByMeta = "countByMeta"
dpReadOnlyWhenVolFull = "dpReadOnlyWhenVolFull"
PeriodicKey = "periodic"
IPKey = "ip"
OperateKey = "op"
)
const (
@ -219,6 +222,7 @@ const (
opSyncAddDecommissionDisk uint32 = 0x28
opSyncDeleteDecommissionDisk uint32 = 0x29
opSyncUpdateDecommissionDisk uint32 = 0x2A
opSyncAcl uint32 = 0x36
)
const (
@ -261,4 +265,5 @@ const (
volCachePrefix = keySeparator + volNameAcronym + keySeparator
DecommissionDiskAcronym = "dd"
DecommissionDiskPrefix = keySeparator + DecommissionDiskAcronym + keySeparator
AclPrefix = keySeparator + "acl" + keySeparator
)

4
master/http_server.go
View File

@ -160,6 +160,10 @@ func (m *Server) registerAPIRoutes(router *mux.Router) {
router.NewRoute().Methods(http.MethodGet).
Path(proto.AdminGetCluster).
HandlerFunc(m.getCluster)
router.NewRoute().Name(proto.AdminACL).
Methods(http.MethodGet).
Path(proto.AdminACL).
HandlerFunc(m.aclOperate)
router.NewRoute().Methods(http.MethodGet, http.MethodPost).
Path(proto.AdminSetClusterInfo).
HandlerFunc(m.setClusterInfo)

109
master/limiter.go
View File

@ -1,6 +1,7 @@
package master
import (
"encoding/json"
"fmt"
"sync"
"time"
@ -705,3 +706,111 @@ func (vol *Vol) volQosUpdateLimit(c *Cluster, limitArgs *qosArgs) error {
vol.qosManager.volUpdateLimit(limitArgs)
return c.syncUpdateVol(vol)
}
type AclManager struct {
aclIps map[string]*proto.AclIpInfo
c *Cluster
vol *Vol
sync.RWMutex
}
type AclFsm struct {
AclIpArr []*proto.AclIpInfo
}
func (acl *AclManager) init(c *Cluster, vol *Vol) {
acl.c = c
acl.vol = vol
acl.aclIps = make(map[string]*proto.AclIpInfo)
}
func (acl *AclManager) aclOperate(op uint64, ip string) interface{} {
acl.Lock()
defer acl.Unlock()
switch op {
case util.AclAddIP:
return acl.addIp(ip)
case util.AclDelIP:
return acl.removeIp(ip)
case util.AclCheckIP:
return acl.checkIp(ip)
case util.AclListIP:
return acl.listAll()
default:
err := fmt.Errorf("aclOperate op %v not found", op)
return err
}
}
func (acl *AclManager) listAll() (val []*proto.AclIpInfo) {
log.LogDebugf("vol %v listAll", acl.vol.Name)
for ip, info := range acl.aclIps {
log.LogDebugf("vol %v listAll ip %v", ip, acl.vol.Name)
val = append(val, info)
}
return
}
func (acl *AclManager) checkIp(ip string) (val []*proto.AclIpInfo) {
log.LogDebugf("checkIp %v", ip)
if info, ok := acl.aclIps[ip]; ok {
log.LogDebugf("vol %v checkIp ip %v", ip, acl.vol.Name)
val = append(val, info)
}
return
}
func (acl *AclManager) addIp(ip string) (err error) {
log.LogDebugf("addIp %v", ip)
if _, ok := acl.aclIps[ip]; ok {
return
}
acl.aclIps[ip] = &proto.AclIpInfo{
Ip: ip,
CTime: time.Now().Unix(),
}
return acl.persist()
}
func (acl *AclManager) removeIp(ip string) (err error) {
log.LogDebugf("removeIp %v", ip)
delete(acl.aclIps, ip)
return acl.persist()
}
func (acl *AclManager) persist() (err error) {
log.LogDebugf("persist")
var aclFsm AclFsm
for _, t := range acl.aclIps {
aclFsm.AclIpArr = append(aclFsm.AclIpArr, t)
}
var val []byte
if val, err = json.Marshal(aclFsm); err != nil {
log.LogErrorf("vol %v acl persist error %v", acl.vol.Name, err)
return
}
if err = acl.c.syncAclList(acl.vol, val); err != nil {
log.LogErrorf("vol %v acl persist syncAclList error %v", acl.vol.Name, err)
return
}
return
}
func (acl *AclManager) load(c *Cluster, val []byte) (err error) {
log.LogDebugf("load")
acl.c = c
aclFsm := &AclFsm{}
if err = json.Unmarshal(val, aclFsm); err != nil {
log.LogErrorf("vol %v acl load %v", acl.vol.Name)
return
}
for _, info := range aclFsm.AclIpArr {
acl.aclIps[info.Ip] = info
log.LogDebugf("vol %v acl load %v", acl.vol.Name, info.Ip)
}
return
}

32
master/metadata_fsm_op.go
View File

@ -642,6 +642,33 @@ func (c *Cluster) syncPutVolInfo(opType uint32, vol *Vol) (err error) {
return c.submit(metadata)
}
func (c *Cluster) syncAclList(vol *Vol, val []byte) (err error) {
log.LogDebugf("syncAclList vol %v vallen %v", vol.Name, len(val))
metadata := new(RaftCmd)
metadata.Op = opSyncAcl
metadata.K = AclPrefix + strconv.FormatUint(vol.ID, 10)
metadata.V = val
return c.submit(metadata)
}
func (c *Cluster) loadAclList(vol *Vol) (err error) {
key := AclPrefix + strconv.FormatUint(vol.ID, 10)
result, err := c.fsm.store.SeekForPrefix([]byte(key))
if err != nil {
log.LogErrorf("action[loadAclList] err %v", err)
return
}
log.LogDebugf("loadAclList vol %v rocksdb value count %v", vol.Name, len(result))
vol.aclMgr.init(c, vol)
for _, value := range result {
return vol.aclMgr.load(c, value)
}
return
}
// key=#mp#volID#metaPartitionID,value=json.Marshal(metaPartitionValue)
func (c *Cluster) syncAddMetaPartition(mp *MetaPartition) (err error) {
return c.putMetaPartitionInfo(opSyncAddMetaPartition, mp)
@ -1190,6 +1217,11 @@ func (c *Cluster) loadVols() (err error) {
}
vol := newVolFromVolValue(vv)
vol.Status = vv.Status
if err = c.loadAclList(vol); err != nil {
log.LogInfof("action[loadVols],vol[%v] load ver manager error %v", vol.Name, err)
continue
}
c.putVol(vol)
log.LogInfof("action[loadVols],vol[%v]", vol.Name)
}

4
master/vol.go
View File

@ -89,8 +89,8 @@ type Vol struct {
domainId uint64
qosManager *QosCtrlManager
DpReadOnlyWhenVolFull bool
volLock sync.RWMutex
aclMgr AclManager
volLock sync.RWMutex
}
func newVol(vv volValue) (vol *Vol) {

17
metanode/partition_fsmop.go
View File

@ -71,13 +71,16 @@ func (mp *metaPartition) fsmUpdatePartition(end uint64) (status uint8,
status = proto.OpOk
oldEnd := mp.config.End
mp.config.End = end
defer func() {
if err != nil {
mp.config.End = oldEnd
status = proto.OpDiskErr
}
}()
err = mp.PersistMetadata()
if end < mp.config.Cursor {
status = proto.OpAgain
mp.config.End = oldEnd
return
}
if err = mp.PersistMetadata(); err != nil {
status = proto.OpDiskErr
mp.config.End = oldEnd
}
return
}

17
proto/admin_proto.go
View File

@ -19,6 +19,7 @@ import "github.com/cubefs/cubefs/util"
// api
const (
// Admin APIs
AdminGetMasterApiList = "/admin/getMasterApiList"
AdminSetApiQpsLimit = "/admin/setApiQpsLimit"
AdminGetApiQpsLimit = "/admin/getApiQpsLimit"
@ -71,6 +72,7 @@ const (
AdminSetClusterUuidEnable = "/admin/setClusterUuidEnable"
AdminGetClusterUuid = "/admin/getClusterUuid"
AdminGenerateClusterUuid = "/admin/generateClusterUuid"
//graphql master api
AdminClusterAPI = "/api/cluster"
AdminUserAPI = "/api/user"
@ -102,6 +104,9 @@ const (
QosUpload = "/admin/qosUpload"
QosUpdateMasterLimit = "/qos/masterLimit"
// acl api
AdminACL = "/admin/aclOp"
//raft node APIs
AddRaftNode = "/raftNode/add"
RemoveRaftNode = "/raftNode/remove"
@ -286,6 +291,18 @@ type RegisterMetaNodeResp struct {
ID uint64
}
type AclIpInfo struct {
Ip string
CTime int64
}
type AclRsp struct {
Info string
OK bool
List []*AclIpInfo
Reserve string
}
// ClusterInfo defines the cluster infomation.
type ClusterInfo struct {
Cluster string

1
proto/errors.go
View File

@ -87,6 +87,7 @@ var (
ErrNoNodeSetToDecommission = errors.New("no node set available to decommission ")
ErrVolNoAvailableSpace = errors.New("vol has no available space")
ErrVolNoCacheAndRule = errors.New("vol has no cache and rule")
ErrNoAclPermission = errors.New("acl no permission")
)
// http response error code and error message definitions

14
sdk/master/api_admin.go
View File

@ -410,6 +410,20 @@ func (api *AdminAPI) GetVolumeSimpleInfoWithFlowInfo(volName string) (vv *proto.
return
}
// access control list
func (api *AdminAPI) CheckACL() (ci *proto.ClusterInfo, err error) {
var request = newAPIRequest(http.MethodGet, proto.AdminACL)
var buf []byte
if buf, err = api.mc.serveRequest(request); err != nil {
return
}
ci = &proto.ClusterInfo{}
if err = json.Unmarshal(buf, &ci); err != nil {
return
}
return
}
func (api *AdminAPI) GetClusterInfo() (ci *proto.ClusterInfo, err error) {
var request = newAPIRequest(http.MethodGet, proto.AdminGetIP)
var buf []byte

25
sdk/master/api_user.go
View File

@ -2,7 +2,11 @@ package master
import (
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/util/ump"
"net/http"
"os"
"strconv"
"github.com/cubefs/cubefs/proto"
)
@ -57,8 +61,10 @@ func (api *UserAPI) UpdateUser(param *proto.UserUpdateParam) (userInfo *proto.Us
}
func (api *UserAPI) GetAKInfo(accesskey string) (userInfo *proto.UserInfo, err error) {
localIP, _ := ump.GetLocalIpAddr()
var request = newAPIRequest(http.MethodGet, proto.UserGetAKInfo)
request.addParam("ak", accesskey)
request.addParam("ip", localIP)
var data []byte
if data, err = api.mc.serveRequest(request); err != nil {
return
@ -70,6 +76,25 @@ func (api *UserAPI) GetAKInfo(accesskey string) (userInfo *proto.UserInfo, err e
return
}
func (api *UserAPI) AclOperation(volName string, localIP string, op uint32) (aclInfo *proto.AclRsp, err error) {
var request = newAPIRequest(http.MethodGet, proto.AdminACL)
request.addParam("name", volName)
request.addParam("ip", localIP)
request.addParam("op", strconv.Itoa(int(op)))
var data []byte
if data, err = api.mc.serveRequest(request); err != nil {
fmt.Fprintf(os.Stdout, "AclOperation serveRequest err %v\n", err)
return
}
aclInfo = &proto.AclRsp{}
if err = json.Unmarshal(data, aclInfo); err != nil {
fmt.Fprintf(os.Stdout, "AclOperation Unmarshal err %v\n", err)
return
}
return
}
func (api *UserAPI) GetUserInfo(userID string) (userInfo *proto.UserInfo, err error) {
var request = newAPIRequest(http.MethodGet, proto.UserGetInfo)
request.addParam("user", userID)

6
util/unit.go
View File

@ -45,6 +45,12 @@ const (
SyscallTryMaxTimes = 3
)
const (
AclListIP = 0
AclAddIP = 1
AclDelIP = 2
AclCheckIP = 3
)
const (
DefaultTinySizeLimit = 1 * MB // TODO explain tiny extent?
)