access control: disconnected-backend(x)

* 'disconnected-backend' is now gone - removed
* reserve the bit for the (TBD) 'object-update' permission
* with refactoring

Signed-off-by: Alex Aizman <[email protected]>

ais/prxauth.go

@@ -244,14 +244,13 @@ func aceErrToCode(err error) (status int) {
 	default:
 		status = http.StatusForbidden
 	}
-	return
+	return status
 }
 
-func (p *proxy) access(hdr http.Header, bck *meta.Bck, ace apc.AccessAttrs) error {
+func (p *proxy) access(hdr http.Header, bck *meta.Bck, ace apc.AccessAttrs) (err error) {
 	var (
 		tk     *tok.Token
 		bucket *cmn.Bck
-		err    error
 		cfg    = cmn.GCO.Get()
 	)
 	if p.isIntraCall(hdr, false /*from primary*/) == nil {
@@ -278,18 +277,17 @@ func (p *proxy) access(hdr http.Header, bck *meta.Bck, ace apc.AccessAttrs) erro
 		// cluster ACL: create/list buckets, node management, etc.
 		return nil
 	}
-	if !cfg.Auth.Enabled || tk.IsAdmin {
-		// PATCH and ACL are always allowed in two cases:
-		// - a user is a superuser
-		// - AuthN is disabled
+
+	// bucket access conventions:
+	// - PATCH and ACL are always allowed if: a) user is a superuser, or b) AuthN is disabled
+	// - without AuthN, read-only access is also always permitted
+	if !cfg.Auth.Enabled {
+		ace &^= (apc.AcePATCH | apc.AceBckSetACL | apc.AccessRO)
+	} else if tk.IsAdmin {
 		ace &^= (apc.AcePATCH | apc.AceBckSetACL)
 	}
 	if ace == 0 {
 		return nil
 	}
-	if !cfg.Auth.Enabled {
-		// Without AuthN, read-only access is always OK
-		ace &^= apc.AccessRO
-	}
 	return bck.Allow(ace)
 }

ais/prxbck.go

@@ -117,8 +117,8 @@ func (args *bckInitArgs) init() (errCode int, err error) {
 		}
 	}
 
-	if err = args._checkRemoteBckPermissions(); err != nil {
-		errCode = http.StatusBadRequest
+	if err = args.accessSupported(); err != nil {
+		errCode = http.StatusMethodNotAllowed
 		return
 	}
 	if args.skipBackend {
@@ -144,46 +144,48 @@ func (args *bckInitArgs) init() (errCode int, err error) {
 		}
 		args.perms = dtor.Access
 	}
-	errCode, err = args.access(bck)
+	errCode, err = args.accessAllowed(bck)
 	return
 }
 
-func (args *bckInitArgs) _checkRemoteBckPermissions() (err error) {
-	var op string
+// returns true when operation requires the 'perm' type access
+func (args *bckInitArgs) _perm(perm apc.AccessAttrs) bool { return (args.perms & perm) == perm }
+
+// (compare w/ accessAllowed)
+func (args *bckInitArgs) accessSupported() error {
 	if !args.bck.IsRemote() {
-		return
+		return nil
 	}
-	if args._requiresPermission(apc.AceMoveBucket) {
+
+	var op string
+	if args._perm(apc.AceMoveBucket) {
 		op = "rename/move remote bucket"
-		goto retErr
+		goto rerr
 	}
-	// HDFS buckets are allowed to be deleted.
+	// accept rename (check!) HDFS buckets are fine across the board
 	if args.bck.IsHDFS() {
-		return
+		return nil
 	}
-	// HTTP buckets should fail on PUT and bucket rename operations
-	if args.bck.IsHTTP() && args._requiresPermission(apc.AcePUT) {
-		op = "PUT => HTTP bucket"
-		goto retErr
+	// HTTP buckets are not writeable
+	if args.bck.IsHTTP() && args._perm(apc.AcePUT) {
+		op = "write to HTTP bucket"
+		goto rerr
 	}
-	// Destroy and Rename/Move are not permitted.
-	if args.bck.IsCloud() && args._requiresPermission(apc.AceDestroyBucket) && args.msg.Action == apc.ActDestroyBck {
-		op = "destroy " + args.bck.Provider + " (cloud) bucket"
-		goto retErr
+	// Cloud bucket: destroy op. not allowed, and not supported yet
+	// (have no separate perm for eviction, that's why an extra check)
+	if rmb := args.bck.IsCloud() && args._perm(apc.AceDestroyBucket) && args.msg.Action == apc.ActDestroyBck; !rmb {
+		return nil
 	}
-	return
-retErr:
-	return cmn.NewErrUnsupp(op, args.bck.String())
+	op = "destroy cloud bucket"
+rerr:
+	return cmn.NewErrUnsupp(op, args.bck.Cname(""))
 }
 
-func (args *bckInitArgs) _requiresPermission(perm apc.AccessAttrs) bool {
-	return (args.perms & perm) == perm
-}
-
-func (args *bckInitArgs) access(bck *meta.Bck) (errCode int, err error) {
+// (compare w/ accessSupported)
+func (args *bckInitArgs) accessAllowed(bck *meta.Bck) (errCode int, err error) {
 	err = args.p.access(args.r.Header, bck, args.perms)
 	errCode = aceErrToCode(err)
-	return
+	return errCode, err
 }
 
 // initAndTry initializes the bucket (proxy-only, as the filename implies).

ais/target.go

@@ -772,18 +772,9 @@ func (t *target) httpobjput(w http.ResponseWriter, r *http.Request, apireq *apiR
 	}
 
 	// load (maybe)
-	skipVC := cmn.Rom.Features().IsSet(feat.SkipVC) || cos.IsParseBool(apireq.dpq.skipVC) // apc.QparamSkipVC
-	if lom.Bck().IsRemote() {
-		var errdb error
-		if skipVC {
-			errdb = lom.CheckRemoteBackend(false)
-		} else if lom.Load(true, false) == nil {
-			errdb = lom.CheckRemoteBackend(true)
-		}
-		if errdb != nil {
-			t.writeErr(w, r, errdb)
-			return
-		}
+	skipVC := cmn.Rom.Features().IsSet(feat.SkipVC) || cos.IsParseBool(apireq.dpq.skipVC)
+	if !skipVC {
+		_ = lom.Load(true, false)
 	}
 
 	// do
@@ -833,7 +824,7 @@ func (t *target) httpobjput(w http.ResponseWriter, r *http.Request, apireq *apiR
 			poi.t = t
 			poi.lom = lom
 			poi.config = config
-			poi.skipVC = skipVC
+			poi.skipVC = skipVC // feat.SkipVC || apc.QparamSkipVC
 			poi.restful = true
 			poi.t2t = t2tput
 		}

ais/test/bucket_test.go

@@ -3003,8 +3003,9 @@ func TestBackendBucket(t *testing.T) {
 		aisObjList.Entries, cachedObjName,
 	)
 
-	// Disconnect backend bucket while denying (the default) apc.AceDisconnectedBackend permission
-	aattrs := apc.AccessAll &^ apc.AceDisconnectedBackend
+	// Disallow PUT (TODO: use apc.AceObjUpdate instead, when/if supported)
+
+	aattrs := apc.AccessAll &^ apc.AcePUT
 	_, err = api.SetBucketProps(baseParams, aisBck, &cmn.BpropsToSet{
 		BackendBck: &cmn.BackendBckToSet{
 			Name:     apc.String(""),
@@ -3015,9 +3016,9 @@ func TestBackendBucket(t *testing.T) {
 	tassert.CheckFatal(t, err)
 	p, err = api.HeadBucket(baseParams, aisBck, true /* don't add */)
 	tassert.CheckFatal(t, err)
-	tassert.Fatalf(t, p.BackendBck.IsEmpty(), "backend bucket isn't empty")
+	tassert.Fatalf(t, p.BackendBck.IsEmpty(), "backend bucket is still configured: %s", p.BackendBck.String())
 
-	// Check if we can still get object and list objects.
+	// Check that we can still GET and list-objects
 	_, err = api.GetObject(baseParams, aisBck, cachedObjName, nil)
 	tassert.CheckFatal(t, err)
 
@@ -3029,13 +3030,15 @@ func TestBackendBucket(t *testing.T) {
 		aisObjList.Entries, cachedObjName,
 	)
 
-	// Check that we cannot do cold gets anymore.
+	// Check that we cannot cold GET anymore - no backend
+	tlog.Logln("Trying to cold-GET when there's no backend anymore (expecting to fail)")
 	_, err = api.GetObject(baseParams, aisBck, remoteObjList.Entries[1].Name, nil)
 	tassert.Fatalf(t, err != nil, "expected error (object should not exist)")
 
-	// Check that we cannot do put anymore.
+	// Check that we cannot do PUT anymore.
+	tlog.Logln("Trying to PUT 2nd version (expecting to fail)")
 	err = tools.PutObjRR(baseParams, aisBck, cachedObjName, 256, cos.ChecksumNone)
-	tassert.Errorf(t, err != nil, "expected err!=nil (put should not be allowed with objSrc!=BackendBck )")
+	tassert.Errorf(t, err != nil, "expected err != nil")
 }
 
 //

ais/test/downloader_test.go

@@ -951,17 +951,21 @@ func TestDownloadOverrideObject(t *testing.T) {
 
 	clearDownloadList(t)
 
-	// disallow updating downloaded objects
-	aattrs := apc.AccessAll &^ apc.AceDisconnectedBackend
-	props := &cmn.BpropsToSet{Access: apc.AccAttrs(aattrs)}
-	tools.CreateBucket(t, proxyURL, bck, props, true /*cleanup*/)
+	tools.CreateBucket(t, proxyURL, bck, nil, true /*cleanup*/)
 
 	downloadObject(t, bck, objName, link, false /*expectedSkipped*/, true /*bucket exists*/)
 	oldProps := verifyProps(t, bck, objName, expectedSize, "1")
 
-	// Update the file
+	// Disallow PUT (TODO: use apc.AceObjUpdate instead when/if supported)
+	aattrs := apc.AccessAll &^ apc.AcePUT
+	_, err := api.SetBucketProps(baseParams, bck, &cmn.BpropsToSet{
+		Access: apc.AccAttrs(aattrs),
+	})
+	tassert.CheckFatal(t, err)
+
+	tlog.Logln("Trying to update the object (expecting to fail)")
 	r, _ := readers.NewRand(10, p.Cksum.Type)
-	_, err := api.PutObject(&api.PutArgs{
+	_, err = api.PutObject(&api.PutArgs{
 		BaseParams: baseParams,
 		Bck:        bck,
 		ObjName:    objName,
@@ -971,6 +975,14 @@ func TestDownloadOverrideObject(t *testing.T) {
 	tassert.Fatalf(t, err != nil, "expected: err!=nil, got: nil")
 	verifyProps(t, bck, objName, expectedSize, "1")
 
+	// Allow PUT back again (TODO: ditto)
+	tlog.Logln("Allow updates back again, and download " + link)
+	aattrs = apc.AccessAll
+	_, err = api.SetBucketProps(baseParams, bck, &cmn.BpropsToSet{
+		Access: apc.AccAttrs(aattrs),
+	})
+	tassert.CheckFatal(t, err)
+
 	downloadObject(t, bck, objName, link, true /*expectedSkipped*/, true /*bucket exists*/)
 	newProps := verifyProps(t, bck, objName, expectedSize, "1")
 	tassert.Errorf(

ais/tgtobj.go

@@ -15,6 +15,7 @@ import (
 	"net/http"
 	"os"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -228,21 +229,26 @@ rerr:
 	return
 }
 
+// verbose only
 func (poi *putOI) loghdr() string {
-	s := poi.owt.String() + ", " + poi.lom.String()
-	if poi.xctn != nil { // may not be showing remote xaction (see doPut)
-		s += ", " + poi.xctn.String()
+	sb := strings.Builder{}
+	sb.WriteString(poi.owt.String())
+	sb.WriteString(", ")
+	sb.WriteString(poi.lom.String())
+	if poi.xctn != nil {
+		sb.WriteString(", ")
+		sb.WriteString(poi.xctn.String())
 	}
 	if poi.skipVC {
-		s += ", skip-vc"
+		sb.WriteString(", skip-vc")
 	}
 	if poi.coldGET {
-		s += ", cold-get"
+		sb.WriteString(", cold-get")
 	}
 	if poi.t2t {
-		s += ", t2t"
+		sb.WriteString(", t2t")
 	}
-	return s
+	return sb.String()
 }
 
 func (poi *putOI) finalize() (errCode int, err error) {
@@ -305,7 +311,7 @@ func (poi *putOI) fini() (errCode int, err error) {
 	case cmn.OwtGetPrefetchLock:
 		if !lom.TryLock(true) {
 			if poi.config.FastV(4, cos.SmoduleAIS) {
-				nlog.Warningf("(%s) is busy", poi.loghdr())
+				nlog.Warningln(poi.loghdr(), "is busy")
 			}
 			return 0, cmn.ErrSkip // e.g. prefetch can skip it and keep on going
 		}
@@ -323,7 +329,7 @@ func (poi *putOI) fini() (errCode int, err error) {
 		if poi.owt == cmn.OwtPut || poi.owt == cmn.OwtFinalize || poi.owt == cmn.OwtPromote {
 			if poi.skipVC {
 				err = lom.IncVersion()
-				debug.Assert(err == nil)
+				debug.AssertNoErr(err)
 			} else if remSrc, ok := lom.GetCustomKey(cmn.SourceObjMD); !ok || remSrc == "" {
 				if err = lom.IncVersion(); err != nil {
 					nlog.Errorln(err)
@@ -580,10 +586,6 @@ do:
 			return
 		}
 		if !eq {
-			if err = goi.lom.CheckRemoteBackend(true /*loaded*/); err != nil {
-				goi.unlocked = true
-				return
-			}
 			cold, goi.verchanged = true, true
 		}
 		goi.lom.Lock(false)

api/apc/access.go

@@ -22,10 +22,8 @@ const (
 	AceObjDELETE
 	AceObjMOVE
 	AcePromote
-	// permission to overwrite objects that were previously read from:
-	// a) any remote backend that is currently not configured as the bucket's backend
-	// b) HTPP ("ht://") since it's not writable
-	AceDisconnectedBackend
+	// TODO: reserving the only perm that must be checked on the target side (NIY)
+	AceObjUpdate
 	// bucket metadata
 	AceBckHEAD   // get bucket props and ACL
 	AceObjLIST   // list objects in a bucket
@@ -45,14 +43,14 @@ const (
 // access => operation
 var accessOp = map[AccessAttrs]string{
 	// object
-	AceGET:                 "GET",
-	AceObjHEAD:             "HEAD-OBJECT",
-	AcePUT:                 "PUT",
-	AceAPPEND:              "APPEND",
-	AceObjDELETE:           "DELETE-OBJECT",
-	AceObjMOVE:             "MOVE-OBJECT",
-	AcePromote:             "PROMOTE",
-	AceDisconnectedBackend: "DISCONNECTED-BACKEND",
+	AceGET:       "GET",
+	AceObjHEAD:   "HEAD-OBJECT",
+	AcePUT:       "PUT",
+	AceAPPEND:    "APPEND",
+	AceObjDELETE: "DELETE-OBJECT",
+	AceObjMOVE:   "MOVE-OBJECT",
+	AcePromote:   "PROMOTE",
+	AceObjUpdate: "UPDATE-OBJECT",
 	// bucket
 	AceBckHEAD:   "HEAD-BUCKET",
 	AceObjLIST:   "LIST-OBJECTS",
@@ -99,9 +97,9 @@ func SupportedPermissions() []string {
 func (a AccessAttrs) Has(perms AccessAttrs) bool { return a&perms == perms }
 func (a AccessAttrs) String() string             { return strconv.FormatUint(uint64(a), 10) }
 
-func (a AccessAttrs) Describe() string {
+func (a AccessAttrs) Describe(all bool) string {
 	if a == 0 {
-		return "No access"
+		return "none"
 	}
 	accList := make([]string, 0, 24)
 	if a.Has(AceGET) {
@@ -125,8 +123,8 @@ func (a AccessAttrs) Describe() string {
 	if a.Has(AcePromote) {
 		accList = append(accList, accessOp[AcePromote])
 	}
-	if a.Has(AceDisconnectedBackend) {
-		accList = append(accList, accessOp[AceDisconnectedBackend])
+	if a.Has(AceObjUpdate) {
+		accList = append(accList, accessOp[AceObjUpdate])
 	}
 	//
 	if a.Has(AceBckHEAD) {
@@ -160,7 +158,13 @@ func (a AccessAttrs) Describe() string {
 	if a.Has(AceAdmin) {
 		accList = append(accList, accessOp[AceAdmin])
 	}
-	return strings.Join(accList, ",")
+
+	// return
+	if all || len(accList) <= 4 {
+		return strings.Join(accList, ",")
+	}
+	accList = accList[:4]
+	return strings.Join(accList, ",") + ", ..."
 }
 
 func AccessOp(access AccessAttrs) string {