|
| 1 | +package gnmi |
| 2 | + |
| 3 | +import ( |
| 4 | + "bytes" |
| 5 | + "context" |
| 6 | + "encoding/json" |
| 7 | + "fmt" |
| 8 | + "io" |
| 9 | + "os" |
| 10 | + "strconv" |
| 11 | + "sync" |
| 12 | + |
| 13 | + "github.com/sonic-net/sonic-gnmi/pathz_authorizer" |
| 14 | + |
| 15 | + log "github.com/golang/glog" |
| 16 | + "github.com/golang/protobuf/proto" |
| 17 | + "github.com/openconfig/gnsi/pathz" |
| 18 | + "google.golang.org/grpc/codes" |
| 19 | + "google.golang.org/grpc/status" |
| 20 | +) |
| 21 | + |
| 22 | +var ( |
| 23 | + pathzMu sync.Mutex |
| 24 | +) |
| 25 | + |
| 26 | +const ( |
| 27 | + pathzTbl string = "PATHZ_POLICY|" |
| 28 | + pathzVersionFld string = "pathz_version" |
| 29 | + pathzCreatedOnFld string = "pathz_created_on" |
| 30 | + pathzPolicyActive pathzInstance = "ACTIVE" |
| 31 | + // support for sandbox not yet implemented |
| 32 | + pathzPolicySandbox pathzInstance = "SANDBOX" |
| 33 | +) |
| 34 | + |
| 35 | +type pathzInstance string |
| 36 | +type PathzMetadata struct { |
| 37 | + PathzVersion string `json:"pathz_version"` |
| 38 | + PathzCreatedOn string `json:"pathz_created_on"` |
| 39 | +} |
| 40 | + |
| 41 | +type GNSIPathzServer struct { |
| 42 | + *Server |
| 43 | + pathzProcessor pathz_authorizer.GnmiAuthzProcessorInterface |
| 44 | + pathzMetadata *PathzMetadata |
| 45 | + pathzMetadataCopy *PathzMetadata |
| 46 | + policyCopy *pathz.AuthorizationPolicy |
| 47 | + policyUpdated bool |
| 48 | + pathzV1Policy string |
| 49 | + pathzV1PolicyBackup string |
| 50 | + pathz.UnimplementedPathzServer |
| 51 | +} |
| 52 | + |
| 53 | +func NewPathzMetadata() *PathzMetadata { |
| 54 | + return &PathzMetadata{ |
| 55 | + PathzVersion: "unknown", |
| 56 | + PathzCreatedOn: "0", |
| 57 | + } |
| 58 | +} |
| 59 | + |
| 60 | +func (srv *GNSIPathzServer) Probe(context.Context, *pathz.ProbeRequest) (*pathz.ProbeResponse, error) { |
| 61 | + return nil, status.Errorf(codes.Unimplemented, "method Probe not implemented") |
| 62 | +} |
| 63 | + |
| 64 | +func (srv *GNSIPathzServer) Get(context.Context, *pathz.GetRequest) (*pathz.GetResponse, error) { |
| 65 | + return nil, status.Errorf(codes.Unimplemented, "method Get not implemented") |
| 66 | +} |
| 67 | +func NewGNSIPathzServer(srv *Server) *GNSIPathzServer { |
| 68 | + ret := &GNSIPathzServer{ |
| 69 | + Server: srv, |
| 70 | + pathzProcessor: &pathz_authorizer.GnmiAuthzProcessor{}, |
| 71 | + pathzMetadata: NewPathzMetadata(), |
| 72 | + pathzV1Policy: srv.config.PathzPolicyFile, |
| 73 | + pathzV1PolicyBackup: srv.config.PathzPolicyFile + ".backup", |
| 74 | + } |
| 75 | + if err := ret.loadPathzFreshness(srv.config.PathzMetaFile); err != nil { |
| 76 | + log.V(0).Info(err) |
| 77 | + } |
| 78 | + ret.writePathzMetadataToDB(pathzPolicyActive) |
| 79 | + if srv.config.PathzPolicy { |
| 80 | + if err := ret.pathzProcessor.UpdatePolicyFromFile(ret.pathzV1Policy); err != nil { |
| 81 | + log.V(0).Infof("Failed to load gNMI pathz file %s: %v", ret.pathzV1Policy, err) |
| 82 | + } |
| 83 | + } |
| 84 | + return ret |
| 85 | +} |
| 86 | + |
| 87 | +func (srv *GNSIPathzServer) savePathzFileFreshess(path string) error { |
| 88 | + log.V(2).Infof("Saving pathz metadata to file: %s", path) |
| 89 | + buf := new(bytes.Buffer) |
| 90 | + enc := json.NewEncoder(buf) |
| 91 | + if err := enc.Encode(*srv.pathzMetadata); err != nil { |
| 92 | + log.V(0).Info(err) |
| 93 | + return err |
| 94 | + } |
| 95 | + return attemptWrite(path, buf.Bytes(), 0o644) |
| 96 | +} |
| 97 | + |
| 98 | +func (srv *GNSIPathzServer) loadPathzFreshness(path string) error { |
| 99 | + bytes, err := os.ReadFile(path) |
| 100 | + if err != nil { |
| 101 | + return err |
| 102 | + } |
| 103 | + return json.Unmarshal(bytes, srv.pathzMetadata) |
| 104 | +} |
| 105 | + |
| 106 | +func (srv *GNSIPathzServer) savePathzPolicyToFile(p *pathz.AuthorizationPolicy) (string, error) { |
| 107 | + content := proto.MarshalTextString(p) |
| 108 | + log.V(3).Infof("Saving pathz policy to file: %s", srv.pathzV1Policy) |
| 109 | + return content, attemptWrite(srv.pathzV1Policy, []byte(content), 0o644) |
| 110 | +} |
| 111 | + |
| 112 | +func (srv *GNSIPathzServer) verifyPathzFile(c string) error { |
| 113 | + content, err := os.ReadFile(srv.pathzV1Policy) |
| 114 | + if err != nil { |
| 115 | + return err |
| 116 | + } |
| 117 | + if c != string(content) { |
| 118 | + return fmt.Errorf("Pathz file %s contains error.", srv.pathzV1Policy) |
| 119 | + } |
| 120 | + return nil |
| 121 | +} |
| 122 | + |
| 123 | +func (srv *GNSIPathzServer) writePathzMetadataToDB(instance pathzInstance) error { |
| 124 | + id := string(instance) |
| 125 | + log.V(2).Infof("Writing pathz metadata to DB: %s Version: %s CreatedOn: %s", id, srv.pathzMetadata.PathzVersion, srv.pathzMetadata.PathzCreatedOn) |
| 126 | + if err := writeCredentialsMetadataToDB(pathzTbl+id, "", pathzVersionFld, srv.pathzMetadata.PathzVersion); err != nil { |
| 127 | + return err |
| 128 | + } |
| 129 | + return writeCredentialsMetadataToDB(pathzTbl+id, "", pathzCreatedOnFld, srv.pathzMetadata.PathzCreatedOn) |
| 130 | +} |
| 131 | + |
| 132 | +func (srv *GNSIPathzServer) updatePolicy(p *pathz.AuthorizationPolicy) error { |
| 133 | + log.V(2).Info("Updating gNMI pathz policy") |
| 134 | + log.V(3).Infof("Policy: %v", p.String()) |
| 135 | + c, err := srv.savePathzPolicyToFile(p) |
| 136 | + if err != nil { |
| 137 | + return err |
| 138 | + } |
| 139 | + if err := srv.verifyPathzFile(c); err != nil { |
| 140 | + log.V(0).Infof("Failed to verify gNMI pathz policy: %v", err) |
| 141 | + return err |
| 142 | + } |
| 143 | + err = srv.pathzProcessor.UpdatePolicyFromProto(p) |
| 144 | + if err != nil { |
| 145 | + log.V(0).Infof("Failed to update gNMI pathz policy: %v", err) |
| 146 | + } |
| 147 | + return err |
| 148 | +} |
| 149 | + |
| 150 | +func (srv *GNSIPathzServer) createCheckpoint() error { |
| 151 | + log.V(2).Info("Creating gNMI pathz policy checkpoint") |
| 152 | + srv.policyCopy = srv.pathzProcessor.GetPolicy() |
| 153 | + srv.policyUpdated = false |
| 154 | + srv.pathzMetadataCopy = srv.pathzMetadata |
| 155 | + return copyFile(srv.pathzV1Policy, srv.pathzV1PolicyBackup) |
| 156 | +} |
| 157 | + |
| 158 | +func (srv *GNSIPathzServer) revertPolicy() error { |
| 159 | + log.V(2).Info("Reverting gNMI pathz policy") |
| 160 | + if srv.policyUpdated { |
| 161 | + srv.policyUpdated = false |
| 162 | + if err := srv.pathzProcessor.UpdatePolicyFromProto(srv.policyCopy); err != nil { |
| 163 | + log.V(0).Infof("Failed to revert gNMI pathz policy: %v", err) |
| 164 | + os.Remove(srv.pathzV1PolicyBackup) |
| 165 | + return err |
| 166 | + } |
| 167 | + } |
| 168 | + srv.pathzMetadata = srv.pathzMetadataCopy |
| 169 | + return os.Rename(srv.pathzV1PolicyBackup, srv.pathzV1Policy) |
| 170 | +} |
| 171 | + |
| 172 | +func (srv *GNSIPathzServer) commitChanges() error { |
| 173 | + log.V(2).Info("Committing gNMI pathz policy changes") |
| 174 | + if err := srv.writePathzMetadataToDB(pathzPolicyActive); err != nil { |
| 175 | + return err |
| 176 | + } |
| 177 | + return srv.savePathzFileFreshess(srv.config.PathzMetaFile) |
| 178 | +} |
| 179 | + |
| 180 | +// Rotate implements the gNSI.pathz.Rotate RPC. |
| 181 | +func (srv *GNSIPathzServer) Rotate(stream pathz.Pathz_RotateServer) error { |
| 182 | + log.V(2).Info("gNSI pathz Rotate RPC") |
| 183 | + ctx := stream.Context() |
| 184 | + ctx, err := authenticate(srv.config, ctx, "gnoi", false) |
| 185 | + if err != nil { |
| 186 | + return err |
| 187 | + } |
| 188 | + // Concurrent Pathz RPCs are not allowed. |
| 189 | + if !pathzMu.TryLock() { |
| 190 | + log.V(0).Infoln("Concurrent Pathz RPCs are not allowed") |
| 191 | + return status.Errorf(codes.Aborted, "Concurrent Pathz RPCs are not allowed") |
| 192 | + } |
| 193 | + defer pathzMu.Unlock() |
| 194 | + if err := fileCheck(srv.pathzV1Policy); err != nil { |
| 195 | + log.V(0).Infof("Error in reading file %s: %v", srv.pathzV1Policy, err) |
| 196 | + return status.Errorf(codes.NotFound, "Error in reading file %s: %v", srv.pathzV1Policy, err) |
| 197 | + } |
| 198 | + if err := srv.createCheckpoint(); err != nil { |
| 199 | + log.V(0).Infof("Error in creating checkpoint: %v", err) |
| 200 | + return status.Errorf(codes.Aborted, "Error in creating checkpoint: %v", err) |
| 201 | + } |
| 202 | + for { |
| 203 | + req, err := stream.Recv() |
| 204 | + log.V(3).Infof("Received a Rotate request message: %v", req.String()) |
| 205 | + if err == io.EOF { |
| 206 | + log.V(0).Infoln("Received EOF instead of a UploadRequest/Finalize request! Reverting to last good state") |
| 207 | + // Connection closed without Finalize message. Revert all changes made until now. |
| 208 | + if err := srv.revertPolicy(); err != nil { |
| 209 | + return status.Errorf(codes.Aborted, "Error in reverting policy: %v", err) |
| 210 | + } |
| 211 | + return status.Errorf(codes.Aborted, "No Finalize message") |
| 212 | + } |
| 213 | + if err != nil { |
| 214 | + log.V(0).Infof("Reverting to last good state Received error: %v", err) |
| 215 | + // Connection closed without Finalize message. Revert all changes made until now. |
| 216 | + srv.revertPolicy() |
| 217 | + return status.Errorf(codes.Aborted, err.Error()) |
| 218 | + } |
| 219 | + if endReq := req.GetFinalizeRotation(); endReq != nil { |
| 220 | + // This is the last message. All changes are final. |
| 221 | + log.V(2).Infof("Received a Finalize request message: %v", endReq) |
| 222 | + if !srv.policyUpdated { |
| 223 | + log.V(0).Infoln("Received finalize message without successful rotation") |
| 224 | + srv.revertPolicy() |
| 225 | + return status.Errorf(codes.Aborted, "Received finalize message without successful rotation") |
| 226 | + } |
| 227 | + if err := srv.commitChanges(); err != nil { |
| 228 | + // Revert won't be called if the final commit fails. |
| 229 | + return status.Errorf(codes.Aborted, "Final policy commit fails: %v", err) |
| 230 | + } |
| 231 | + os.Remove(srv.pathzV1PolicyBackup) |
| 232 | + return nil |
| 233 | + } |
| 234 | + resp, err := srv.processRotateRequest(req) |
| 235 | + if err != nil { |
| 236 | + log.V(0).Infof("Reverting to last good state; While processing a rotate request got error: %v", err) |
| 237 | + // Connection closed without Finalize message. Revert all changes made until now. |
| 238 | + srv.revertPolicy() |
| 239 | + return err |
| 240 | + } |
| 241 | + if err := stream.Send(resp); err != nil { |
| 242 | + log.V(0).Infof("Reverting to last good state; While sending a confirmation got error: %v", err) |
| 243 | + // Connection closed without Finalize message. Revert all changes made until now. |
| 244 | + srv.revertPolicy() |
| 245 | + return status.Errorf(codes.Aborted, err.Error()) |
| 246 | + } |
| 247 | + } |
| 248 | +} |
| 249 | + |
| 250 | +func (srv *GNSIPathzServer) processRotateRequest(req *pathz.RotateRequest) (*pathz.RotateResponse, error) { |
| 251 | + policyReq := req.GetUploadRequest() |
| 252 | + if policyReq == nil { |
| 253 | + return nil, status.Errorf(codes.Aborted, "Unknown request: %v", req) |
| 254 | + } |
| 255 | + log.V(2).Infof("Received a gNSI.Pathz UploadRequest request message") |
| 256 | + if len(policyReq.GetVersion()) == 0 { |
| 257 | + return nil, status.Errorf(codes.Aborted, "Pathz policy version cannot be empty") |
| 258 | + } |
| 259 | + if srv.pathzMetadata.PathzVersion == policyReq.GetVersion() && !req.GetForceOverwrite() { |
| 260 | + return nil, status.Errorf(codes.AlreadyExists, "Pathz with version `%v` already exists", policyReq.GetVersion()) |
| 261 | + } |
| 262 | + srv.pathzMetadata.PathzVersion = policyReq.GetVersion() |
| 263 | + srv.pathzMetadata.PathzCreatedOn = strconv.FormatUint(policyReq.GetCreatedOn(), 10) |
| 264 | + if err := srv.updatePolicy(policyReq.GetPolicy()); err != nil { |
| 265 | + return nil, status.Errorf(codes.Aborted, err.Error()) |
| 266 | + } |
| 267 | + srv.policyUpdated = true |
| 268 | + resp := &pathz.RotateResponse{ |
| 269 | + Response: &pathz.RotateResponse_Upload{}, |
| 270 | + } |
| 271 | + return resp, nil |
| 272 | +} |
| 273 | +func attemptWrite(name string, data []byte, perm os.FileMode) error { |
| 274 | + log.V(2).Infof("Writing: %s", name) |
| 275 | + err := os.WriteFile(name, data, perm) |
| 276 | + if err != nil { |
| 277 | + if e := os.Remove(name); e != nil { |
| 278 | + err = fmt.Errorf("Write %s failed: %w; Cleanup failed", name, err) |
| 279 | + } |
| 280 | + } |
| 281 | + return err |
| 282 | +} |
0 commit comments