mirror of
https://github.com/openziti/ziti.git
synced 2026-10-26 21:47:05 +00:00
Andrew Martinez
8415a7dbf6
* The controller and the router wrote the complete service session token, an OIDC service session JWT or a legacy session token, into structured log fields during edge control operations. Several of the sites logged at Error or Info, so the values reached logs at default verbosity during ordinary operation. * Adds LoggingFields to the session and api session types in common, controller/model and router/state, so a logging site asks for the session fields or the api session fields instead of picking values by hand. Ids, type, identity, service and the token's validity window go in; the token does not. Empty values are dropped, so the sets merge over each other. * AddLoggingFields in router/state now delegates to LoggingFields. Its output moves from a nested field to flat prefixed keys. * The edge control handlers log the session field set on the shared returnError and logResult paths, on circuit setup and on terminator update. The terminator removal paths log the terminator id, since they never load a session. * Adds unit tests over returnError and logResult that fail if a session token reaches a log entry, covering the loaded-session, failed-to-load and JWT-claims cases.
837 lines
27 KiB
Go
837 lines
27 KiB
Go
package handler_edge_ctrl
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"maps"
|
|
"math"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/michaelquigley/pfxlog"
|
|
"github.com/openziti/channel/v5"
|
|
"github.com/openziti/identity"
|
|
"github.com/openziti/sdk-golang/v2/ziti/edge"
|
|
"github.com/openziti/ziti/v2/common"
|
|
"github.com/openziti/ziti/v2/common/ctrl_msg"
|
|
"github.com/openziti/ziti/v2/common/logcontext"
|
|
"github.com/openziti/ziti/v2/common/pb/edge_ctrl_pb"
|
|
"github.com/openziti/ziti/v2/controller/change"
|
|
"github.com/openziti/ziti/v2/controller/db"
|
|
"github.com/openziti/ziti/v2/controller/env"
|
|
"github.com/openziti/ziti/v2/controller/fields"
|
|
"github.com/openziti/ziti/v2/controller/model"
|
|
"github.com/openziti/ziti/v2/controller/models"
|
|
"github.com/openziti/ziti/v2/controller/network"
|
|
"github.com/openziti/ziti/v2/controller/oidc_auth"
|
|
"github.com/openziti/ziti/v2/controller/storage/boltz"
|
|
"github.com/openziti/ziti/v2/controller/xt"
|
|
"github.com/sirupsen/logrus"
|
|
)
|
|
|
|
type requestHandler interface {
|
|
getAppEnv() *env.AppEnv
|
|
getNetwork() *network.Network
|
|
getChannel() channel.Channel
|
|
Label() string
|
|
}
|
|
|
|
type UpdateTerminatorRequest interface {
|
|
GetCost() uint32
|
|
GetUpdateCost() bool
|
|
GetPrecedence() edge_ctrl_pb.TerminatorPrecedence
|
|
GetUpdatePrecedence() bool
|
|
}
|
|
|
|
type baseRequestHandler struct {
|
|
ch channel.Channel
|
|
appEnv *env.AppEnv
|
|
}
|
|
|
|
// responseMutatingRequestContext is implemented by request contexts that need to
|
|
// modify error responses before they are sent (e.g. to attach a circuit ID).
|
|
type responseMutatingRequestContext interface {
|
|
UpdateResponse(m *channel.Message)
|
|
}
|
|
|
|
func (self *baseRequestHandler) getNetwork() *network.Network {
|
|
return self.appEnv.GetHostController().GetNetwork()
|
|
}
|
|
|
|
func (self *baseRequestHandler) getAppEnv() *env.AppEnv {
|
|
return self.appEnv
|
|
}
|
|
|
|
func (self *baseRequestHandler) getChannel() channel.Channel {
|
|
return self.ch
|
|
}
|
|
|
|
func (self *baseRequestHandler) returnError(ctx requestContext, err controllerError) {
|
|
responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), []byte(err.Error()))
|
|
responseMsg.PutUint32Header(edge.ErrorCodeHeader, err.ErrorCode())
|
|
responseMsg.PutUint32Header(ctrl_msg.ErrorRetryHintHeader, uint32(err.GetRetryHint()))
|
|
responseMsg.ReplyTo(ctx.GetMessage())
|
|
logger := pfxlog.
|
|
ContextLogger(self.ch.Label()).
|
|
WithError(err).
|
|
WithField("routerId", ctx.GetHandler().getChannel().Id()).
|
|
WithField("operation", ctx.GetHandler().Label())
|
|
|
|
if sessionCtx, ok := ctx.(sessionRequestContext); ok {
|
|
logger = logger.WithFields(sessionCtx.sessionLoggingFields())
|
|
}
|
|
|
|
if mutator, ok := ctx.(responseMutatingRequestContext); ok {
|
|
mutator.UpdateResponse(responseMsg)
|
|
}
|
|
|
|
if sendErr := self.ch.Send(responseMsg); sendErr != nil {
|
|
logger.WithError(err).WithField("sendError", sendErr).Error("failed to send error response")
|
|
} else {
|
|
logger.WithError(err).Error("responded with error")
|
|
}
|
|
}
|
|
|
|
func (self *baseRequestHandler) logResult(ctx requestContext, err error) {
|
|
logger := logrus.
|
|
WithField("routerId", ctx.GetHandler().getChannel().Id()).
|
|
WithField("operation", ctx.GetHandler().Label())
|
|
|
|
if sessionCtx, ok := ctx.(sessionRequestContext); ok {
|
|
logger = logger.WithFields(sessionCtx.sessionLoggingFields())
|
|
}
|
|
|
|
if err != nil {
|
|
logger.WithError(err).Error("operation failed")
|
|
} else {
|
|
logger.Debug("operation success")
|
|
}
|
|
}
|
|
|
|
type requestContext interface {
|
|
GetHandler() requestHandler
|
|
GetMessage() *channel.Message
|
|
}
|
|
|
|
// sessionRequestContext is implemented by request contexts that can describe the service
|
|
// session a request was made against. The description is built from ids and the token's
|
|
// validity window, never from the token itself, which is a credential.
|
|
type sessionRequestContext interface {
|
|
requestContext
|
|
sessionLoggingFields() logrus.Fields
|
|
}
|
|
|
|
type baseSessionRequestContext struct {
|
|
handler requestHandler
|
|
msg *channel.Message
|
|
err controllerError
|
|
sourceRouter *model.Router
|
|
session *model.Session
|
|
apiSession *model.ApiSession
|
|
service *model.EdgeService
|
|
newSession bool
|
|
logContext logcontext.Context
|
|
env model.Env
|
|
accessClaims *common.AccessClaims
|
|
serviceAccessClaims *common.ServiceAccessClaims
|
|
}
|
|
|
|
// sessionLoggingFields returns the log fields describing the service session this request was
|
|
// made against, empty when no session has been loaded. A JWT service session contributes the
|
|
// token's validity window on top of what the stored session carries. The session token itself
|
|
// is a credential and is never among the fields.
|
|
func (self *baseSessionRequestContext) sessionLoggingFields() logrus.Fields {
|
|
fields := self.session.LoggingFields()
|
|
maps.Copy(fields, self.serviceAccessClaims.LoggingFields())
|
|
return fields
|
|
}
|
|
|
|
// apiSessionLoggingFields returns the log fields describing the api session this request
|
|
// authenticated with, empty when no api session has been loaded. The api session token is a
|
|
// credential and is never among the fields.
|
|
func (self *baseSessionRequestContext) apiSessionLoggingFields() logrus.Fields {
|
|
fields := self.apiSession.LoggingFields()
|
|
maps.Copy(fields, self.accessClaims.LoggingFields())
|
|
return fields
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) getApiSessionId() string {
|
|
if self.apiSession != nil {
|
|
return self.apiSession.Id
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) newChangeContext() *change.Context {
|
|
result := change.New().SetSourceType(change.SourceTypeControlChannel).
|
|
SetSourceMethod(self.handler.Label()).
|
|
SetSourceLocal(self.handler.getChannel().Underlay().GetLocalAddr().String()).
|
|
SetSourceRemote(self.handler.getChannel().Underlay().GetRemoteAddr().String())
|
|
if self.session != nil {
|
|
result.
|
|
SetChangeAuthorType(change.AuthorTypeIdentity).
|
|
SetChangeAuthorId(self.session.IdentityId)
|
|
if self.apiSession != nil && self.apiSession.Identity != nil {
|
|
result.SetChangeAuthorName(self.apiSession.Identity.Name)
|
|
} else if authorIdentity, _ := self.handler.getAppEnv().Managers.Identity.Read(self.session.IdentityId); authorIdentity != nil {
|
|
result.SetChangeAuthorName(authorIdentity.Name)
|
|
}
|
|
} else if self.sourceRouter != nil {
|
|
result.
|
|
SetChangeAuthorType(change.AuthorTypeRouter).
|
|
SetChangeAuthorId(self.sourceRouter.Id).
|
|
SetChangeAuthorName(self.sourceRouter.Name)
|
|
}
|
|
return result
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) newTunnelChangeContext() *change.Context {
|
|
return change.New().SetSourceType(change.SourceTypeControlChannel).
|
|
SetSourceMethod(self.handler.Label()).
|
|
SetSourceLocal(self.handler.getChannel().Underlay().GetLocalAddr().String()).
|
|
SetSourceRemote(self.handler.getChannel().Underlay().GetRemoteAddr().String()).
|
|
SetChangeAuthorType(change.AuthorTypeRouter).
|
|
SetChangeAuthorId(self.sourceRouter.Id).
|
|
SetChangeAuthorName(self.sourceRouter.Name)
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) GetMessage() *channel.Message {
|
|
return self.msg
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) GetHandler() requestHandler {
|
|
return self.handler
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) loadRouter() bool {
|
|
routerId := self.handler.getChannel().Id()
|
|
var err error
|
|
self.sourceRouter, err = self.handler.getNetwork().GetRouter(routerId)
|
|
if err != nil {
|
|
self.err = internalError(err)
|
|
logrus.
|
|
WithField("router", routerId).
|
|
WithField("operation", self.handler.Label()).
|
|
WithError(self.err).Errorf("could not find router closing channel")
|
|
_ = self.handler.getChannel().Close()
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) loadSession(sessionToken string, apiSessionToken string) {
|
|
if strings.HasPrefix(apiSessionToken, oidc_auth.JwtTokenPrefix) {
|
|
self.loadFromTokens(sessionToken, apiSessionToken)
|
|
} else {
|
|
self.loadFromBolt(sessionToken, apiSessionToken)
|
|
}
|
|
|
|
if self.err != nil {
|
|
return
|
|
}
|
|
|
|
if self.session == nil {
|
|
self.err = internalError(errors.New("session was not found after load"))
|
|
return
|
|
}
|
|
|
|
if self.apiSession == nil {
|
|
self.err = internalError(errors.New("api session was not found after load"))
|
|
return
|
|
}
|
|
|
|
self.logContext = logcontext.NewContext()
|
|
traceSpec := self.handler.getAppEnv().TraceManager.GetIdentityTrace(self.apiSession.IdentityId)
|
|
traceEnabled := traceSpec != nil && time.Now().Before(traceSpec.Until)
|
|
if traceEnabled {
|
|
self.logContext.SetChannelsMask(traceSpec.ChannelMask)
|
|
self.logContext.WithField("traceId", traceSpec.TraceId)
|
|
}
|
|
self.logContext.WithField("sessionId", self.session.Id)
|
|
self.logContext.WithField("apiSessionId", self.apiSession.Id)
|
|
|
|
if traceEnabled {
|
|
pfxlog.ChannelLogger(logcontext.EstablishPath).
|
|
Wire(self.logContext).
|
|
Debug("tracing enabled for this session")
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) loadFromTokens(sessionToken, apiSessionToken string) {
|
|
if self.err != nil {
|
|
return
|
|
}
|
|
|
|
var err error
|
|
self.accessClaims, err = self.env.ValidateAccessToken(apiSessionToken)
|
|
|
|
if err != nil {
|
|
self.err = internalError(err)
|
|
return
|
|
}
|
|
|
|
serviceAccessClaims, err := self.env.ValidateServiceAccessToken(sessionToken, &self.accessClaims.ApiSessionId)
|
|
|
|
if err != nil {
|
|
self.err = internalError(err)
|
|
return
|
|
}
|
|
|
|
self.serviceAccessClaims = serviceAccessClaims
|
|
|
|
if self.accessClaims.Subject != serviceAccessClaims.IdentityId {
|
|
self.err = internalError(fmt.Errorf("access and service tokens do not match, got access identity id %s and service identity id %s", self.accessClaims.Subject, serviceAccessClaims.IdentityId))
|
|
return
|
|
}
|
|
|
|
self.session = &model.Session{
|
|
BaseEntity: models.BaseEntity{
|
|
Id: serviceAccessClaims.ID,
|
|
CreatedAt: serviceAccessClaims.IssuedAt.Time,
|
|
UpdatedAt: serviceAccessClaims.IssuedAt.Time,
|
|
},
|
|
Token: sessionToken,
|
|
IdentityId: serviceAccessClaims.IdentityId,
|
|
ApiSessionId: serviceAccessClaims.ApiSessionId,
|
|
ServiceId: serviceAccessClaims.Subject,
|
|
Type: serviceAccessClaims.Type,
|
|
}
|
|
|
|
tokenIdentity, err := self.env.GetManagers().Identity.Read(self.accessClaims.Subject)
|
|
|
|
if err != nil {
|
|
self.err = internalError(err)
|
|
return
|
|
}
|
|
|
|
self.apiSession = &model.ApiSession{
|
|
BaseEntity: models.BaseEntity{
|
|
Id: serviceAccessClaims.ApiSessionId,
|
|
},
|
|
Token: apiSessionToken,
|
|
IdentityId: serviceAccessClaims.IdentityId,
|
|
Identity: tokenIdentity,
|
|
IPAddress: self.accessClaims.RemoteAddress,
|
|
ConfigTypes: self.accessClaims.ConfigTypesAsMap(),
|
|
TotpComplete: false,
|
|
TotpRequired: false,
|
|
ExpiresAt: time.Time{},
|
|
ExpirationDuration: 0,
|
|
LastActivityAt: time.Time{},
|
|
AuthenticatorId: "",
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) loadFromBolt(sessionToken string, apiSessionToken string) {
|
|
if self.err != nil {
|
|
return
|
|
}
|
|
|
|
var err error
|
|
self.apiSession, err = self.handler.getAppEnv().Managers.ApiSession.ReadByToken(apiSessionToken)
|
|
|
|
if err != nil {
|
|
if boltz.IsErrNotFoundErr(err) {
|
|
self.err = InvalidApiSessionError{}
|
|
} else {
|
|
self.err = internalError(err)
|
|
}
|
|
logrus.
|
|
WithField("operation", self.handler.Label()).
|
|
WithError(self.err).Errorf("invalid api-session")
|
|
return
|
|
}
|
|
|
|
if !strings.HasPrefix(sessionToken, oidc_auth.JwtTokenPrefix) {
|
|
self.session, err = self.handler.getAppEnv().Managers.Session.ReadByToken(sessionToken)
|
|
|
|
if err != nil {
|
|
if boltz.IsErrNotFoundErr(err) {
|
|
self.err = InvalidSessionError{}
|
|
} else {
|
|
self.err = internalError(err)
|
|
}
|
|
logrus.
|
|
WithField("operation", self.handler.Label()).
|
|
WithError(self.err).Errorf("invalid session")
|
|
return
|
|
}
|
|
|
|
if self.session.ApiSessionId != self.apiSession.Id {
|
|
self.err = InvalidSessionError{}
|
|
logrus.
|
|
WithField("operation", self.handler.Label()).
|
|
WithField("sessionId", self.session.Id).
|
|
WithField("sessionApiSessionId", self.session.ApiSessionId).
|
|
WithField("apiSessionId", self.apiSession.Id).
|
|
WithError(self.err).Error("session does not belong to api session")
|
|
}
|
|
return
|
|
}
|
|
|
|
serviceAccessClaims, err := self.env.ValidateServiceAccessToken(sessionToken, &self.apiSession.Id)
|
|
|
|
if err != nil {
|
|
// A token-level failure (bad/expired/mismatched/revoked token) means the client should discard
|
|
// and re-create its session, so return InvalidSession. An infrastructure failure (e.g. a
|
|
// revocation datastore read) must stay an internalError, otherwise a transient controller fault
|
|
// would make clients discard valid sessions and trigger a reauthentication storm.
|
|
var invalidToken *common.InvalidTokenError
|
|
if errors.As(err, &invalidToken) {
|
|
self.err = InvalidSessionError{}
|
|
logrus.
|
|
WithField("operation", self.handler.Label()).
|
|
WithError(err).Error("service access token invalid; treating as invalid session")
|
|
} else {
|
|
self.err = internalError(err)
|
|
}
|
|
return
|
|
}
|
|
|
|
self.serviceAccessClaims = serviceAccessClaims
|
|
|
|
if !serviceAccessClaims.IsLegacy {
|
|
self.err = internalError(errors.New("legacy api session detected, but session token is not durable"))
|
|
return
|
|
}
|
|
|
|
self.session, err = self.handler.getAppEnv().Managers.Session.Read(serviceAccessClaims.ID)
|
|
|
|
if err != nil {
|
|
if boltz.IsErrNotFoundErr(err) {
|
|
self.err = InvalidSessionError{}
|
|
} else {
|
|
self.err = internalError(err)
|
|
}
|
|
logrus.
|
|
WithField("operation", self.handler.Label()).
|
|
WithError(self.err).Errorf("invalid session")
|
|
return
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) checkSessionType(sessionType string) {
|
|
if self.err == nil {
|
|
if self.session.Type != sessionType {
|
|
self.err = WrongSessionTypeError{}
|
|
logrus.
|
|
WithField("sessionId", self.session.Id).
|
|
WithField("operation", self.handler.Label()).
|
|
WithError(self.err).Errorf("wrong session type")
|
|
}
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) verifyIdentityEdgeRouterAccess() {
|
|
if self.err == nil {
|
|
self.verifyEdgeRouterAccess(self.session.IdentityId, self.session.ServiceId,
|
|
func(string, model.EdgeRouterAccess) controllerError {
|
|
return InvalidEdgeRouterForSessionError{}
|
|
})
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) verifyEdgeRouterServiceBindAccess() {
|
|
if self.err == nil {
|
|
self.verifyServiceBindAccess(self.sourceRouter.Id, self.service.Id)
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) verifyServiceBindAccess(identityId string, serviceId string) {
|
|
if self.err == nil {
|
|
// validate edge router
|
|
result, err := self.handler.getAppEnv().Managers.EdgeService.IsBindableByIdentity(serviceId, identityId)
|
|
if err != nil {
|
|
self.err = internalError(err)
|
|
logrus.
|
|
WithField("routerId", self.sourceRouter.Id).
|
|
WithField("identityId", identityId).
|
|
WithField("serviceId", serviceId).
|
|
WithField("operation", self.handler.Label()).
|
|
WithError(err).Error("unable to verify edge router access to bind service")
|
|
return
|
|
} else if !result {
|
|
self.err = InvalidServiceError{}
|
|
}
|
|
}
|
|
}
|
|
|
|
// verifyRouterEdgeRouterAccess checks that the requesting router may act on the service on its
|
|
// own behalf, as an edge router tunneler does. No edge session is involved, so denials are
|
|
// reported with the missing policy link rather than as a session error.
|
|
func (self *baseSessionRequestContext) verifyRouterEdgeRouterAccess() {
|
|
if self.err == nil {
|
|
self.verifyEdgeRouterAccess(self.sourceRouter.Id, self.service.Id, self.newEdgeRouterAccessDeniedError)
|
|
}
|
|
}
|
|
|
|
// newEdgeRouterAccessDeniedError builds a denial message naming the policy that would need to be
|
|
// added to permit the identity to reach the service through this edge router.
|
|
func (self *baseSessionRequestContext) newEdgeRouterAccessDeniedError(identityId string, access model.EdgeRouterAccess) controllerError {
|
|
var reason string
|
|
if !access.IdentityAllowed && !access.ServiceAllowed {
|
|
reason = "no edge router policy links the identity to the edge router and no service edge router policy links the service to the edge router"
|
|
} else if !access.IdentityAllowed {
|
|
reason = "no edge router policy links the identity to the edge router"
|
|
} else {
|
|
reason = "no service edge router policy links the service to the edge router"
|
|
}
|
|
|
|
return edgeRouterAccessDenied(fmt.Sprintf("edge router %s may not be used for service %s by identity %s: %s",
|
|
self.sourceRouter.Name, self.service.Name, identityId, reason))
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) verifyEdgeRouterAccess(identityId string, serviceId string, newDeniedError func(identityId string, access model.EdgeRouterAccess) controllerError) {
|
|
if self.err == nil {
|
|
// validate edge router
|
|
erMgr := self.handler.getAppEnv().Managers.EdgeRouter
|
|
access, err := erMgr.GetEdgeRouterAccess(identityId, serviceId, self.sourceRouter.Id)
|
|
if err != nil {
|
|
self.err = internalError(err)
|
|
logrus.
|
|
WithField("routerId", self.sourceRouter.Id).
|
|
WithField("identityId", identityId).
|
|
WithField("serviceId", serviceId).
|
|
WithField("operation", self.handler.Label()).
|
|
WithError(err).Error("unable to verify edge router access")
|
|
return
|
|
}
|
|
|
|
if !access.IsAllowed() {
|
|
self.err = newDeniedError(identityId, access)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) loadService() {
|
|
if self.err == nil {
|
|
var err error
|
|
self.service, err = self.handler.getAppEnv().Managers.EdgeService.Read(self.session.ServiceId)
|
|
|
|
if err != nil {
|
|
if boltz.IsErrNotFoundErr(err) {
|
|
self.err = InvalidServiceError{}
|
|
} else {
|
|
self.err = internalError(err)
|
|
}
|
|
logrus.
|
|
WithField("sessionId", self.session.Id).
|
|
WithField("operation", self.handler.Label()).
|
|
WithField("serviceId", self.session.ServiceId).
|
|
WithError(self.err).
|
|
Error("service not found")
|
|
}
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) verifyTerminator(terminatorId string, binding string) *model.Terminator {
|
|
if self.err == nil {
|
|
var terminator *model.Terminator
|
|
var err error
|
|
terminator, err = self.handler.getNetwork().Terminator.Read(terminatorId)
|
|
|
|
if err != nil {
|
|
if boltz.IsErrNotFoundErr(err) {
|
|
self.err = invalidTerminator("invalid terminator: not found")
|
|
} else {
|
|
self.err = internalError(err)
|
|
}
|
|
log := logrus.
|
|
WithField("operation", self.handler.Label()).
|
|
WithField("terminatorId", terminatorId).
|
|
WithError(self.err)
|
|
if self.session != nil {
|
|
log = log.WithField("sessionId", self.session.Id)
|
|
}
|
|
log.Error("terminator not found")
|
|
return nil
|
|
}
|
|
|
|
if terminator != nil && terminator.Router != self.sourceRouter.Id {
|
|
self.err = invalidTerminator(fmt.Sprintf("%v request for terminator %v on router %v came from router %v",
|
|
self.handler.Label(), terminatorId, terminator.Router, self.sourceRouter.Id))
|
|
|
|
log := logrus.
|
|
WithField("operation", self.handler.Label()).
|
|
WithField("sourceRouter", self.sourceRouter.Id).
|
|
WithField("terminatorId", terminatorId).
|
|
WithField("terminatorRouter", terminator.Router).
|
|
WithError(self.err)
|
|
if self.session != nil {
|
|
log = log.WithField("sessionId", self.session.Id)
|
|
}
|
|
log.Error("not allowed to operate on terminators on other routers")
|
|
return nil
|
|
}
|
|
|
|
if terminator != nil && terminator.Binding != binding {
|
|
self.err = invalidTerminator(fmt.Sprintf("can't operate on terminator %v with wrong binding, expected binding %v, was %v ",
|
|
terminatorId, binding, terminator.Binding))
|
|
|
|
log := logrus.
|
|
WithField("operation", self.handler.Label()).
|
|
WithField("sourceRouter", self.sourceRouter.Id).
|
|
WithField("terminatorId", terminatorId).
|
|
WithField("terminatorRouter", terminator.Router).
|
|
WithField("binding", terminator.Binding).
|
|
WithField("expectedBinding", binding).
|
|
WithError(self.err)
|
|
if self.session != nil {
|
|
log = log.WithField("sessionId", self.session.Id)
|
|
}
|
|
log.Error("incorrect binding")
|
|
return nil
|
|
}
|
|
|
|
return terminator
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) verifyTerminatorId(id string) {
|
|
if self.err == nil {
|
|
if id == "" {
|
|
self.err = invalidTerminator("provided terminator id is blank")
|
|
}
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) updateTerminator(terminator *model.Terminator, request UpdateTerminatorRequest, ctx *change.Context) {
|
|
if self.err == nil {
|
|
checker := fields.UpdatedFieldsMap{}
|
|
|
|
if request.GetUpdateCost() {
|
|
if request.GetCost() > math.MaxUint16 {
|
|
self.err = invalidCost(fmt.Sprintf("invalid cost %v. cost must be between 0 and %v inclusive", request.GetCost(), math.MaxUint16))
|
|
return
|
|
}
|
|
terminator.Cost = uint16(request.GetCost())
|
|
checker[db.FieldTerminatorCost] = struct{}{}
|
|
}
|
|
|
|
if request.GetUpdatePrecedence() {
|
|
if request.GetPrecedence() == edge_ctrl_pb.TerminatorPrecedence_Default {
|
|
terminator.Precedence = xt.Precedences.Default
|
|
} else if request.GetPrecedence() == edge_ctrl_pb.TerminatorPrecedence_Required {
|
|
terminator.Precedence = xt.Precedences.Required
|
|
} else if request.GetPrecedence() == edge_ctrl_pb.TerminatorPrecedence_Failed {
|
|
terminator.Precedence = xt.Precedences.Failed
|
|
} else {
|
|
self.err = invalidPrecedence(fmt.Sprintf("invalid precedence: %v", request.GetPrecedence()))
|
|
return
|
|
}
|
|
|
|
checker[db.FieldTerminatorPrecedence] = struct{}{}
|
|
}
|
|
|
|
self.err = internalError(self.handler.getNetwork().Terminator.Update(terminator, checker, ctx))
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) newCircuitCreateParms(serviceId string, peerData map[uint32][]byte) model.CreateCircuitParams {
|
|
return &sessionCircuitParams{
|
|
serviceId: serviceId,
|
|
sourceRouter: self.sourceRouter,
|
|
clientId: &identity.TokenId{Token: self.session.Id, Data: peerData},
|
|
logCtx: self.logContext,
|
|
deadline: time.Now().Add(self.handler.getAppEnv().GetHostController().GetNetwork().GetOptions().RouteTimeout),
|
|
reqCtx: self,
|
|
}
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) newTunnelCircuitCreateParms(serviceId string, peerData map[uint32][]byte) model.CreateCircuitParams {
|
|
return &tunnelCircuitParams{
|
|
serviceId: serviceId,
|
|
sourceRouter: self.sourceRouter,
|
|
clientId: &identity.TokenId{Token: self.sourceRouter.Id, Data: peerData},
|
|
logCtx: self.logContext,
|
|
deadline: time.Now().Add(self.handler.getAppEnv().GetHostController().GetNetwork().GetOptions().RouteTimeout),
|
|
reqCtx: self,
|
|
}
|
|
}
|
|
|
|
type circuitParamsFactory = func(serviceId string, peerData map[uint32][]byte) model.CreateCircuitParams
|
|
|
|
func (self *baseSessionRequestContext) createCircuit(terminatorInstanceId string, peerData map[uint32][]byte, paramsFactory circuitParamsFactory) (*model.Circuit, map[uint32][]byte) {
|
|
var circuit *model.Circuit
|
|
returnPeerData := map[uint32][]byte{}
|
|
|
|
if self.err == nil {
|
|
if self.service.EncryptionRequired && peerData[uint32(edge.PublicKeyHeader)] == nil {
|
|
self.err = encryptionDataMissing("encryption required on service, initiator did not send public header")
|
|
return nil, nil
|
|
}
|
|
|
|
serviceId := self.service.Id
|
|
if terminatorInstanceId != "" {
|
|
serviceId = terminatorInstanceId + "@" + serviceId
|
|
}
|
|
|
|
n := self.handler.getAppEnv().GetHostController().GetNetwork()
|
|
params := paramsFactory(serviceId, peerData)
|
|
var err error
|
|
circuit, err = n.CreateCircuit(params)
|
|
if err != nil {
|
|
self.err = internalError(err)
|
|
}
|
|
|
|
if circuit != nil && err == nil {
|
|
//static terminator peer data
|
|
for k, v := range circuit.Terminator.GetPeerData() {
|
|
returnPeerData[k] = v
|
|
}
|
|
|
|
//runtime peer data
|
|
for k, v := range circuit.PeerData {
|
|
returnPeerData[k] = v
|
|
}
|
|
|
|
if self.service.EncryptionRequired && returnPeerData[uint32(edge.PublicKeyHeader)] == nil {
|
|
self.err = encryptionDataMissing("encryption required on service, terminator did not send public header")
|
|
if err = n.RemoveCircuit(circuit.Id, true); err != nil {
|
|
logrus.
|
|
WithField("operation", self.handler.Label()).
|
|
WithField("sourceRouter", self.sourceRouter.Id).
|
|
WithError(err).
|
|
Error("failed to remove circuit")
|
|
}
|
|
return circuit, nil
|
|
}
|
|
}
|
|
}
|
|
return circuit, returnPeerData
|
|
}
|
|
|
|
func (self *baseSessionRequestContext) validateExistingTerminator(terminator *model.Terminator, identityId string, binding string, log *logrus.Entry) controllerError {
|
|
if self.err != nil {
|
|
return self.err
|
|
}
|
|
|
|
if terminator.Binding != binding {
|
|
log.WithField("binding", binding).
|
|
WithField("conflictingBinding", terminator.Binding).
|
|
Error("selected terminator address conflicts with a terminator for a different binding")
|
|
return nonRetriableError(errors.New("selected id conflicts with terminator for different binding"))
|
|
}
|
|
|
|
if terminator.Service != self.service.Id {
|
|
log.WithField("conflictingService", terminator.Service).
|
|
Error("selected terminator address conflicts with a terminator for a different service")
|
|
return nonRetriableError(errors.New("selected id conflicts with terminator for different service"))
|
|
}
|
|
|
|
if terminator.Router != self.sourceRouter.Id {
|
|
log.WithField("conflictingRouter", terminator.Router).
|
|
Error("selected terminator address conflicts with a terminator for a different router")
|
|
return nonRetriableError(errors.New("selected id conflicts with terminator for different router"))
|
|
}
|
|
|
|
if terminator.HostId != identityId {
|
|
log.WithField("identityId", identityId).
|
|
WithField("conflictingIdentity", terminator.HostId).
|
|
Error("selected terminator address conflicts with a terminator for a different identity")
|
|
return nonRetriableError(errors.New("selected id conflicts with terminator for different identity"))
|
|
}
|
|
|
|
log.Info("terminator already exists")
|
|
return nil
|
|
}
|
|
|
|
type sessionCircuitParams struct {
|
|
serviceId string
|
|
sourceRouter *model.Router
|
|
clientId *identity.TokenId
|
|
logCtx logcontext.Context
|
|
deadline time.Time
|
|
reqCtx *baseSessionRequestContext
|
|
}
|
|
|
|
func (self *sessionCircuitParams) GetServiceId() string {
|
|
return self.serviceId
|
|
}
|
|
|
|
func (self *sessionCircuitParams) GetSourceRouter() *model.Router {
|
|
return self.sourceRouter
|
|
}
|
|
|
|
func (self *sessionCircuitParams) GetClientId() *identity.TokenId {
|
|
return self.clientId
|
|
}
|
|
|
|
func (self *sessionCircuitParams) GetCircuitTags(t xt.CostedTerminator) map[string]string {
|
|
if t == nil {
|
|
return map[string]string{
|
|
"serviceId": self.serviceId,
|
|
"clientId": self.reqCtx.session.IdentityId,
|
|
}
|
|
}
|
|
|
|
hostId := t.GetHostId()
|
|
return map[string]string{
|
|
"serviceId": self.serviceId,
|
|
"clientId": self.reqCtx.session.IdentityId,
|
|
"hostId": hostId,
|
|
}
|
|
}
|
|
|
|
func (self *sessionCircuitParams) GetLogContext() logcontext.Context {
|
|
return self.logCtx
|
|
}
|
|
|
|
func (self *sessionCircuitParams) GetDeadline() time.Time {
|
|
return self.deadline
|
|
}
|
|
|
|
func (self *sessionCircuitParams) GetCircuitId() string {
|
|
return ""
|
|
}
|
|
|
|
type tunnelCircuitParams struct {
|
|
serviceId string
|
|
sourceRouter *model.Router
|
|
clientId *identity.TokenId
|
|
logCtx logcontext.Context
|
|
deadline time.Time
|
|
reqCtx *baseSessionRequestContext
|
|
}
|
|
|
|
func (self *tunnelCircuitParams) GetServiceId() string {
|
|
return self.serviceId
|
|
}
|
|
|
|
func (self *tunnelCircuitParams) GetSourceRouter() *model.Router {
|
|
return self.sourceRouter
|
|
}
|
|
|
|
func (self *tunnelCircuitParams) GetClientId() *identity.TokenId {
|
|
return self.clientId
|
|
}
|
|
|
|
func (self *tunnelCircuitParams) GetCircuitTags(t xt.CostedTerminator) map[string]string {
|
|
if t == nil {
|
|
return map[string]string{
|
|
"serviceId": self.serviceId,
|
|
"clientId": self.sourceRouter.Id,
|
|
}
|
|
}
|
|
|
|
hostId := t.GetHostId()
|
|
return map[string]string{
|
|
"serviceId": self.serviceId,
|
|
"clientId": self.sourceRouter.Id,
|
|
"hostId": hostId,
|
|
}
|
|
}
|
|
|
|
func (self *tunnelCircuitParams) GetLogContext() logcontext.Context {
|
|
return self.logCtx
|
|
}
|
|
|
|
func (self *tunnelCircuitParams) GetDeadline() time.Time {
|
|
return self.deadline
|
|
}
|
|
|
|
func (self *tunnelCircuitParams) GetCircuitId() string {
|
|
return ""
|
|
}
|