security: sign and verify redirect URLs (part six); dpq

* cluster-key:
  - skip signing requests when csk version is zero (ie., not generated)
  - can legally happen since primary keeps driving certain startup
    action _after_ datapath is already enabled
-----
* datapath query (dpq):
  - errors to include: tag, key=value
  - remove count; refactor
-----
* prev. commit: f616e23

Signed-off-by: Alex Aizman <alex.aizman@gmail.com>

Author: alex-aizman

ais/csk.go

@@ -17,10 +17,12 @@ import (
 	"strconv"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/cmn/debug"
 	"github.com/NVIDIA/aistore/cmn/mono"
+	"github.com/NVIDIA/aistore/cmn/nlog"
 	"github.com/NVIDIA/aistore/memsys"
 )
 
@@ -46,18 +48,21 @@ import (
 //   and compares it against the provided signature. Any mismatch results in a 401.
 
 // TODO: sign.verify()
+// - check that csk is always initialized across (enable/disable; lifecycle events)
+//   (see related sign.warn())
 // - anti-replay logic (sliding time window, per-sender nonce tracking, etc.)
 // - validate pid: a) != "" (weak) or b) in Smap (strong)
-// - stricter handling for legacy unsigned redirects
 // - optionally, extend HMAC payload to cover assorted query parameters
 
 const (
-	cskTag         = "csk"
-	cskKeyLen      = 16  // len(secret)
-	cskSigLen      = 43  // = base64.RawURLEncoding.EncodedLen(sha256.Size)
-	cskSepa        = 0   // to separate strings in HMAC payload
-	cskBase        = 36  // base for all signed int64/uint64 fields
-	cskURLOverhead = 160 // pid+utm+vpams+x+u qparams (~145 worst-case)
+	cskTag = "csk"
+
+	cskKeyLen      = 16               // len(secret)
+	cskSigLen      = 43               // = base64.RawURLEncoding.EncodedLen(sha256.Size)
+	cskSepa        = 0                // to separate strings in HMAC payload
+	cskBase        = 36               // base for all signed int64/uint64 fields
+	cskURLOverhead = 160              // pid+utm+vpams+x+u qparams (~145 worst-case)
+	cskUptime      = 10 * time.Second // cluster uptime after which we start warning if CSK version remains zero
 )
 
 type (
@@ -157,10 +162,16 @@ func (sign *signer) compute(pid string) {
 	binary.BigEndian.PutUint64(b8[:], sign.nonce)
 	sb.WriteBytes(b8[:])
 
+	// load key
+	k := sign.h.owner.csk.load()
+	if k.ver == 0 {
+		sign.warn()
+		return
+	}
+	debug.Assert(len(k.secret) > 0, k.String())
+
 	// hash
 	hb := handAlloc()
-	k := sign.h.owner.csk.load()
-	debug.Assert(k.ver > 0 && len(k.secret) > 0)
 	if hb.h == nil || hb.ver != k.ver {
 		hb.h = hmac.New(sha256.New, k.secret)
 		hb.ver = k.ver
@@ -178,6 +189,19 @@ func (sign *signer) compute(pid string) {
 	handFree(hb)
 }
 
+func (sign *signer) warn() {
+	const skip = "skip HMAC sign (cluster-key version is zero)"
+	uptime := sign.h.keepalive.cluUptime(mono.NanoTime())
+	switch {
+	case uptime < cskUptime:
+		// expected during early startup, stay quiet
+	case uptime < cskUptime<<1:
+		nlog.Warningln(sign.h.String(), skip)
+	case uptime < min(time.Minute, cskUptime<<2):
+		nlog.Errorln(sign.h.String(), skip)
+	}
+}
+
 func (sign *signer) buildURL(nodeURL string, now int64) string {
 	var (
 		h  = sign.h

ais/dpq.go

@@ -49,6 +49,8 @@ import (
 // Note: it's always a good idea to explicitly add each and every new query into one of the
 // non-default categories, potentially 3, 4, or 5.
 
+const dpqTag = "dpq"
+
 type (
 	cskgrp struct {
 		nonce   uint64 // QparamNonce
@@ -73,8 +75,6 @@ type (
 		}
 		csk cskgrp // csk envelope (group CSK/HMAC)
 
-		count int
-
 		// boolean fields
 		skipVC        bool // QparamSkipVC (skip loading existing object's metadata)
 		isGFN         bool // QparamIsGFNRequest
@@ -123,7 +123,7 @@ func dpqAlloc() (d *dpq) {
 }
 
 func dpqFree(d *dpq) {
-	c := d.count
+	c := len(d.m)
 	m := d.m
 
 	clear(m)
@@ -143,7 +143,6 @@ func (dpq *dpq) parse(rawQuery string) error {
 		query = rawQuery // r.URL.RawQuery
 		iters int
 	)
-	dpq.count = 0
 	for query != "" && iters < maxDpqCap {
 		var (
 			err   error
@@ -210,7 +209,6 @@ func (dpq *dpq) parse(rawQuery string) error {
 		case apc.QparamMptUploadID, apc.QparamMptPartNo, apc.QparamFltPresence, apc.QparamBinfoWithOrWithoutRemote,
 			apc.QparamAppendType, apc.QparamETLName, apc.QparamETLTransformArgs:
 			dpq.m[key] = value
-			dpq.count++
 
 		// Finally, assorted named exceptions that we simply skip, and b) all the rest parameters
 		default:
@@ -224,17 +222,19 @@ func (dpq *dpq) parse(rawQuery string) error {
 
 			// Default: unescape and store in map
 			dpq.m[key], err = _unescape(value)
-			dpq.count++
 		}
 
 		// common error return
 		if err != nil {
-			return err
+			if errors.Is(err, strconv.ErrSyntax) {
+				err = strconv.ErrSyntax
+			}
+			return fmt.Errorf("%s: '%s=%s', err: %v", dpqTag, key, value, err)
 		}
 		iters++
 	}
 	if iters >= maxDpqCap {
-		return errors.New("exceeded max number of dpq iterations: " + strconv.Itoa(iters))
+		return fmt.Errorf("%s: exceeded maximum number of iterations (%d)", dpqTag, iters)
 	}
 
 	return nil
@@ -281,7 +281,7 @@ func (dpq *dpq) _arch(key, val string) (err error) {
 	}
 	// either/or
 	if dpq.arch.path != "" && dpq.arch.mmode != "" { // (empty arch.regx is fine - is EmptyMatchAny)
-		err = fmt.Errorf("query parameters archpath=%q (match one) and archregx=%q (match many) are mutually exclusive",
+		err = fmt.Errorf("%s: archpath=%q (match one) and archregx=%q (match many) are mutually exclusive", dpqTag,
 			apc.QparamArchpath, apc.QparamArchregx)
 	}
 	return err
@@ -309,24 +309,25 @@ func (dpq *dpq) _archstr() string {
 //
 
 func cskFromQ(q url.Values) (*cskgrp, error) {
+	const tag = "from-q"
 	sig := q.Get(apc.QparamHMAC)
 	if sig == "" {
 		return nil, nil
 	}
 	if len(sig) != cskSigLen {
-		return nil, fmt.Errorf("invalid signature length: %d", len(sig))
+		return nil, fmt.Errorf("%s: invalid signature length: %d", tag, len(sig))
 	}
 
 	s := q.Get(apc.QparamNonce)
 	nonce, err := strconv.ParseUint(s, cskBase, 64)
 	if err != nil {
-		return nil, fmt.Errorf("invalid nonce: %v", err)
+		return nil, fmt.Errorf("%s: invalid nonce '%s=%s': %v", tag, apc.QparamNonce, s, err)
 	}
 
 	s = q.Get(apc.QparamSmapVer)
 	smapVer, err := strconv.ParseInt(s, cskBase, 64)
 	if err != nil {
-		return nil, fmt.Errorf("invalid smap version: %v", err)
+		return nil, fmt.Errorf("%s: invalid Smap version %q: %v", tag, s, err)
 	}
 	return &cskgrp{nonce: nonce, smapVer: smapVer, hmacSig: sig}, nil
 }

ais/earlystart.go

@@ -336,7 +336,7 @@ func (p *proxy) primaryStartup(loadedSmap *smapX, config *cmn.Config, ntargets i
 		cos.ExitLog(err)
 	}
 
-	// 11. metasync (smap, config, etl & bmd) and startup as primary
+	// 11. metasync (all cluster-level meta) and startup as primary
 	smap = p.owner.smap.get()
 	var (
 		actMsgExt = p.newAmsgStr(metaction2, bmd)
@@ -345,7 +345,7 @@ func (p *proxy) primaryStartup(loadedSmap *smapX, config *cmn.Config, ntargets i
 	if cluConfig.Auth.CSKEnabled() {
 		k := p.owner.csk.gen(smap.Version)
 		pairs = append(pairs, revsPair{k, actMsgExt})
-	}
+	} // note: can do signed requests after this point
 
 	wg := p.metasyncer.sync(pairs...)
 	wg.Wait()