Implements the frontend logic for gNSI Pathz

Author: niranjanivivek

Makefile

@@ -113,7 +113,7 @@ endif
 
 # download and apply patch for gnmi client, which will break advancetls
 # backup crypto and gnxi
-	mkdir backup_crypto
+	mkdir -p backup_crypto
 	cp -r vendor/golang.org/x/crypto/* backup_crypto/
 
 # download and patch crypto and gnxi
@@ -126,6 +126,17 @@ endif
 	git apply patches/0001-Updated-to-filter-and-write-to-file.patch
 	git apply patches/0003-Fix-client-json-parsing-issue.patch
 
+# Manually adding patched client packages and their dependencies
+# to vendor/modules.txt. This satisfies 'go install -mod=vendor' lookup checks,
+# which are required after manual patching/copying of gnxi and gnmi-cli code.
+	echo "github.com/google/gnxi v0.0.0-20181220173256-89f51f0ce1e2" >> vendor/modules.txt
+	echo "github.com/google/gnxi/gnmi_get" >> vendor/modules.txt
+	echo "github.com/google/gnxi/gnmi_set" >> vendor/modules.txt
+	echo "github.com/openconfig/gnmi/cli" >> vendor/modules.txt
+	echo "github.com/openconfig/gnmi/client/flags" >> vendor/modules.txt
+	echo "golang.org/x/crypto/ssh/terminal" >> vendor/modules.txt
+	echo "github.com/openconfig/gnmi/cmd/gnmi_cli" >> vendor/modules.txt
+
 ifeq ($(CROSS_BUILD_ENVIRON),y)
 	$(GO) build -o ${GOBIN}/gnmi_get -mod=vendor github.com/google/gnxi/gnmi_get
 	$(GO) build -o ${GOBIN}/gnmi_set -mod=vendor github.com/google/gnxi/gnmi_set
@@ -212,6 +223,7 @@ check_gotest: $(DBCONFG) $(ENVFILE)
 	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(GO) test -race -coverprofile=coverage-telemetry.txt -covermode=atomic -mod=vendor -v github.com/sonic-net/sonic-gnmi/telemetry
 	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(GO) test -race -coverprofile=coverage-config.txt -covermode=atomic -v github.com/sonic-net/sonic-gnmi/sonic_db_config
 	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(TESTENV) $(GO) test -race -timeout 20m -coverprofile=coverage-gnmi.txt -covermode=atomic -mod=vendor $(BLD_FLAGS) -v github.com/sonic-net/sonic-gnmi/gnmi_server -coverpkg ../...
+	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(TESTENV) $(GO) test -race -timeout 20m -coverprofile=coverage-pathz_authorizer.txt -covermode=atomic -mod=vendor $(BLD_FLAGS) -v github.com/sonic-net/sonic-gnmi/pathz_authorizer -coverpkg ../...
 ifneq ($(ENABLE_DIALOUT_VALUE),0)
 	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(TESTENV) $(GO) test -coverprofile=coverage-dialout.txt -covermode=atomic -mod=vendor $(BLD_FLAGS) -v github.com/sonic-net/sonic-gnmi/dialout/dialout_client
 endif

gnmi_server/gnsi_pathz.go

@@ -0,0 +1,339 @@
+package gnmi
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"strconv"
+	"sync"
+
+	lvl "github.com/sonic-net/sonic-gnmi/gnmi_server/log"
+	"github.com/sonic-net/sonic-gnmi/pathz_authorizer"
+
+	log "github.com/golang/glog"
+	"github.com/golang/protobuf/proto"
+	"github.com/openconfig/gnsi/pathz"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"path/filepath"
+)
+
+var (
+	pathzMu sync.Mutex
+)
+
+const (
+	pathzTbl          string        = "PATHZ_POLICY|"
+	pathzVersionFld   string        = "pathz_version"
+	pathzCreatedOnFld string        = "pathz_created_on"
+	pathzPolicyActive pathzInstance = "ACTIVE"
+	// support for sandbox not yet implemented
+	pathzPolicySandbox pathzInstance = "SANDBOX"
+)
+
+type pathzInstance string
+type PathzMetadata struct {
+	PathzVersion   string `json:"pathz_version"`
+	PathzCreatedOn string `json:"pathz_created_on"`
+}
+
+type GNSIPathzServer struct {
+	*Server
+	pathzProcessor      pathz_authorizer.GnmiAuthzProcessorInterface
+	pathzMetadata       *PathzMetadata
+	pathzMetadataCopy   *PathzMetadata
+	policyCopy          *pathz.AuthorizationPolicy
+	policyUpdated       bool
+	pathzV1Policy       string
+	pathzV1PolicyBackup string
+	pathz.UnimplementedPathzServer
+}
+
+func NewPathzMetadata() *PathzMetadata {
+	return &PathzMetadata{
+		PathzVersion:   "unknown",
+		PathzCreatedOn: "0",
+	}
+}
+
+func (srv *GNSIPathzServer) Probe(context.Context, *pathz.ProbeRequest) (*pathz.ProbeResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Probe not implemented")
+}
+
+func (srv *GNSIPathzServer) Get(context.Context, *pathz.GetRequest) (*pathz.GetResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Get not implemented")
+}
+
+func NewGNSIPathzServer(srv *Server) *GNSIPathzServer {
+	ret := &GNSIPathzServer{
+		Server:              srv,
+		pathzProcessor:      &pathz_authorizer.GnmiAuthzProcessor{},
+		pathzMetadata:       NewPathzMetadata(),
+		pathzV1Policy:       srv.config.PathzPolicyFile,
+		pathzV1PolicyBackup: srv.config.PathzPolicyFile + ".backup",
+	}
+	if err := ret.loadPathzFreshness(srv.config.PathzMetaFile); err != nil {
+		log.V(lvl.ERROR).Info(err)
+	}
+	ret.writePathzMetadataToDB(pathzPolicyActive)
+	if srv.config.PathzPolicy {
+		if err := ret.pathzProcessor.UpdatePolicyFromFile(ret.pathzV1Policy); err != nil {
+			log.V(lvl.ERROR).Infof("Failed to load gNMI pathz file %s: %v", ret.pathzV1Policy, err)
+		}
+	}
+	return ret
+}
+
+func (srv *GNSIPathzServer) savePathzFileFreshess(path string) error {
+	log.V(lvl.INFO).Infof("Saving pathz metadata to file: %s", path)
+	buf := new(bytes.Buffer)
+	enc := json.NewEncoder(buf)
+	if err := enc.Encode(*srv.pathzMetadata); err != nil {
+		log.V(lvl.ERROR).Info(err)
+		return err
+	}
+	return attemptWrite(path, buf.Bytes(), 0o644)
+}
+
+func (srv *GNSIPathzServer) loadPathzFreshness(path string) error {
+	bytes, err := os.ReadFile(path)
+	if err != nil {
+		return err
+	}
+	return json.Unmarshal(bytes, srv.pathzMetadata)
+}
+
+func (srv *GNSIPathzServer) savePathzPolicyToFile(p *pathz.AuthorizationPolicy) (string, error) {
+	content := proto.MarshalTextString(p)
+	log.V(lvl.DEBUG).Infof("Saving pathz policy to file: %s", srv.pathzV1Policy)
+	return content, attemptWrite(srv.pathzV1Policy, []byte(content), 0o644)
+}
+
+func (srv *GNSIPathzServer) verifyPathzFile(c string) error {
+	content, err := os.ReadFile(srv.pathzV1Policy)
+	if err != nil {
+		return err
+	}
+	if c != string(content) {
+		return fmt.Errorf("Pathz file %s contains error.", srv.pathzV1Policy)
+	}
+	return nil
+}
+
+func (srv *GNSIPathzServer) writePathzMetadataToDB(instance pathzInstance) error {
+	id := string(instance)
+	log.V(lvl.INFO).Infof("Writing pathz metadata to DB: %s Version: %s CreatedOn: %s", id, srv.pathzMetadata.PathzVersion, srv.pathzMetadata.PathzCreatedOn)
+	if err := writeCredentialsMetadataToDB(pathzTbl+id, "", pathzVersionFld, srv.pathzMetadata.PathzVersion); err != nil {
+		return err
+	}
+	return writeCredentialsMetadataToDB(pathzTbl+id, "", pathzCreatedOnFld, srv.pathzMetadata.PathzCreatedOn)
+}
+
+func (srv *GNSIPathzServer) updatePolicy(p *pathz.AuthorizationPolicy) error {
+	log.V(lvl.INFO).Info("Updating gNMI pathz policy")
+	log.V(lvl.DEBUG).Infof("Policy: %v", p.String())
+	c, err := srv.savePathzPolicyToFile(p)
+	if err != nil {
+		return err
+	}
+	if err := srv.verifyPathzFile(c); err != nil {
+		log.V(lvl.ERROR).Infof("Failed to verify gNMI pathz policy: %v", err)
+		return err
+	}
+	err = srv.pathzProcessor.UpdatePolicyFromProto(p)
+	if err != nil {
+		log.V(lvl.ERROR).Infof("Failed to update gNMI pathz policy: %v", err)
+	}
+	return err
+}
+
+func (srv *GNSIPathzServer) createCheckpoint() error {
+	log.V(lvl.INFO).Info("Creating gNMI pathz policy checkpoint")
+	srv.policyCopy = srv.pathzProcessor.GetPolicy()
+	srv.policyUpdated = false
+	srv.pathzMetadataCopy = srv.pathzMetadata
+	return copyFile(srv.pathzV1Policy, srv.pathzV1PolicyBackup)
+}
+
+func (srv *GNSIPathzServer) revertPolicy() error {
+	log.V(lvl.INFO).Info("Reverting gNMI pathz policy")
+	if srv.policyUpdated {
+		srv.policyUpdated = false
+		if err := srv.pathzProcessor.UpdatePolicyFromProto(srv.policyCopy); err != nil {
+			log.V(lvl.ERROR).Infof("Failed to revert gNMI pathz policy: %v", err)
+			os.Remove(srv.pathzV1PolicyBackup)
+			return err
+		}
+	}
+	srv.pathzMetadata = srv.pathzMetadataCopy
+	return os.Rename(srv.pathzV1PolicyBackup, srv.pathzV1Policy)
+}
+
+func (srv *GNSIPathzServer) commitChanges() error {
+	log.V(lvl.INFO).Info("Committing gNMI pathz policy changes")
+	if err := srv.writePathzMetadataToDB(pathzPolicyActive); err != nil {
+		return err
+	}
+	return srv.savePathzFileFreshess(srv.config.PathzMetaFile)
+}
+
+// Rotate implements the gNSI.pathz.Rotate RPC.
+func (srv *GNSIPathzServer) Rotate(stream pathz.Pathz_RotateServer) error {
+	// Reject while NSF Freeze is ongoing
+	log.V(lvl.INFO).Info("gNSI pathz Rotate RPC")
+	// TODO(b/344081417) Enable code once NSF Freeze is implemented
+	// Reject while NSF Freeze is ongoing
+	// if srv.Server.WarmRestartHelper.FetchFreezeStatus() {
+	// 	log.V(lvl.ERROR).Info("gNSI pathz Rotate RPC disabled since NSF is ongoing!")
+	// 	return status.Errorf(codes.Unavailable, "gNSI pathz Rotate RPC disabled since NSF is ongoing!")
+	// }
+	ctx := stream.Context()
+	ctx, err := authenticate(srv.config, ctx, "gnoi", false)
+	if err != nil {
+		return err
+	}
+	// Concurrent Pathz RPCs are not allowed.
+	if !pathzMu.TryLock() {
+		log.V(lvl.ERROR).Infoln("Concurrent Pathz RPCs are not allowed")
+		return status.Errorf(codes.Aborted, "Concurrent Pathz RPCs are not allowed")
+	}
+	defer pathzMu.Unlock()
+	if err := fileCheck(srv.pathzV1Policy); err != nil {
+		log.V(lvl.ERROR).Infof("Error in reading file %s: %v", srv.pathzV1Policy, err)
+		return status.Errorf(codes.NotFound, "Error in reading file %s: %v", srv.pathzV1Policy, err)
+	}
+	if err := srv.createCheckpoint(); err != nil {
+		log.V(lvl.ERROR).Infof("Error in creating checkpoint: %v", err)
+		return status.Errorf(codes.Aborted, "Error in creating checkpoint: %v", err)
+	}
+	for {
+		req, err := stream.Recv()
+		log.V(lvl.DEBUG).Infof("Received a Rotate request message: %v", req.String())
+		if err == io.EOF {
+			log.V(lvl.ERROR).Infoln("Received EOF instead of a UploadRequest/Finalize request! Reverting to last good state")
+			// Connection closed without Finalize message. Revert all changes made until now.
+			if err := srv.revertPolicy(); err != nil {
+				return status.Errorf(codes.Aborted, "Error in reverting policy: %v", err)
+			}
+			return status.Errorf(codes.Aborted, "No Finalize message")
+		}
+		if err != nil {
+			log.V(lvl.ERROR).Infof("Reverting to last good state Received error: %v", err)
+			// Connection closed without Finalize message. Revert all changes made until now.
+			srv.revertPolicy()
+			return status.Errorf(codes.Aborted, err.Error())
+		}
+		if endReq := req.GetFinalizeRotation(); endReq != nil {
+			// This is the last message. All changes are final.
+			log.V(lvl.INFO).Infof("Received a Finalize request message: %v", endReq)
+			if !srv.policyUpdated {
+				log.V(lvl.ERROR).Infoln("Received finalize message without successful rotation")
+				srv.revertPolicy()
+				return status.Errorf(codes.Aborted, "Received finalize message without successful rotation")
+			}
+			if err := srv.commitChanges(); err != nil {
+				// Revert won't be called if the final commit fails.
+				return status.Errorf(codes.Aborted, "Final policy commit fails: %v", err)
+			}
+			os.Remove(srv.pathzV1PolicyBackup)
+			return nil
+		}
+		resp, err := srv.processRotateRequest(req)
+		if err != nil {
+			log.V(lvl.ERROR).Infof("Reverting to last good state; While processing a rotate request got error: %v", err)
+			// Connection closed without Finalize message. Revert all changes made until now.
+			srv.revertPolicy()
+			return err
+		}
+		if err := stream.Send(resp); err != nil {
+			log.V(lvl.ERROR).Infof("Reverting to last good state; While sending a confirmation got error: %v", err)
+			// Connection closed without Finalize message. Revert all changes made until now.
+			srv.revertPolicy()
+			return status.Errorf(codes.Aborted, err.Error())
+		}
+	}
+}
+
+func (srv *GNSIPathzServer) processRotateRequest(req *pathz.RotateRequest) (*pathz.RotateResponse, error) {
+	policyReq := req.GetUploadRequest()
+	if policyReq == nil {
+		return nil, status.Errorf(codes.Aborted, "Unknown request: %v", req)
+	}
+	log.V(lvl.INFO).Infof("Received a gNSI.Pathz UploadRequest request message")
+	if len(policyReq.GetVersion()) == 0 {
+		return nil, status.Errorf(codes.Aborted, "Pathz policy version cannot be empty")
+	}
+	if srv.pathzMetadata.PathzVersion == policyReq.GetVersion() && !req.GetForceOverwrite() {
+		return nil, status.Errorf(codes.AlreadyExists, "Pathz with version `%v` already exists", policyReq.GetVersion())
+	}
+	srv.pathzMetadata.PathzVersion = policyReq.GetVersion()
+	srv.pathzMetadata.PathzCreatedOn = strconv.FormatUint(policyReq.GetCreatedOn(), 10)
+	if err := srv.updatePolicy(policyReq.GetPolicy()); err != nil {
+		return nil, status.Errorf(codes.Aborted, err.Error())
+	}
+	srv.policyUpdated = true
+	resp := &pathz.RotateResponse{
+		Response: &pathz.RotateResponse_Upload{},
+	}
+	return resp, nil
+}
+func attemptWrite(name string, data []byte, perm os.FileMode) error {
+	log.V(lvl.INFO).Infof("Writing: %s", name)
+	err := os.WriteFile(name, data, perm)
+	if err != nil {
+		if e := os.Remove(name); e != nil {
+			err = fmt.Errorf("Write %s failed: %w; Cleanup failed", name, err)
+		}
+	}
+	return err
+}
+func fileCheck(f string) error {
+	srcStat, err := os.Stat(f)
+	if err != nil {
+		return err
+	}
+	if !srcStat.Mode().IsRegular() {
+		return fmt.Errorf("%s is not a regular file", f)
+	}
+	return nil
+}
+func copyFile(srcPath, dstPath string) error {
+	srcStat, err := os.Stat(srcPath)
+	if err != nil {
+		return err
+	}
+	if !srcStat.Mode().IsRegular() {
+		return fmt.Errorf("%s is not a regular file", srcPath)
+	}
+	src, err := os.Open(srcPath)
+	if err != nil {
+		return err
+	}
+	defer src.Close()
+	tmpDst, err := os.CreateTemp(filepath.Dir(dstPath), filepath.Base(dstPath))
+	if err != nil {
+		return err
+	}
+	if _, err := io.Copy(tmpDst, src); err != nil {
+		if e := os.Remove(tmpDst.Name()); e != nil {
+			log.V(2).Infof("Failed to cleanup file: %v: %v", tmpDst.Name(), e)
+		}
+		return err
+	}
+	if err := tmpDst.Close(); err != nil {
+		if e := os.Remove(tmpDst.Name()); e != nil {
+			log.V(2).Infof("Failed to cleanup file: %v: %v", tmpDst.Name(), e)
+		}
+		return err
+	}
+	if err := os.Rename(tmpDst.Name(), dstPath); err != nil {
+		if e := os.Remove(tmpDst.Name()); e != nil {
+			log.V(2).Infof("Failed to cleanup file: %v: %v", tmpDst.Name(), e)
+		}
+		return err
+	}
+	return os.Chmod(dstPath, 0600)
+}