mirror of
https://github.com/openziti/ziti.git
synced 2026-09-10 08:45:41 +00:00
Paul Lorenz
809 lines
26 KiB
Go
809 lines
26 KiB
Go
package handler_edge_ctrl
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"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.WithField("token", sessionCtx.GetSessionToken())
|
|
}
|
|
|
|
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.WithField("token", sessionCtx.GetSessionToken())
|
|
}
|
|
|
|
if err != nil {
|
|
logger.WithError(err).Error("operation failed")
|
|
} else {
|
|
logger.Debug("operation success")
|
|
}
|
|
}
|
|
|
|
type requestContext interface {
|
|
GetHandler() requestHandler
|
|
GetMessage() *channel.Message
|
|
}
|
|
|
|
type sessionRequestContext interface {
|
|
requestContext
|
|
GetSessionToken() string
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
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 ""
|
|
}
|