ziti/router/handler_link/bind.go

package handler_link

import (
	"crypto/x509"
	"time"

	"github.com/michaelquigley/pfxlog"
	"github.com/openziti/channel/v5"
	"github.com/openziti/channel/v5/latency"
	"github.com/openziti/channel/v5/protobufs"
	"github.com/openziti/foundation/v2/concurrenz"
	nfpem "github.com/openziti/foundation/v2/pem"
	"github.com/openziti/metrics"
	"github.com/openziti/sdk-golang/v2/xgress"
	"github.com/openziti/ziti/v2/common/pb/ctrl_pb"
	"github.com/openziti/ziti/v2/common/trace"
	"github.com/openziti/ziti/v2/router/env"
	"github.com/openziti/ziti/v2/router/forwarder"
	metrics2 "github.com/openziti/ziti/v2/router/metrics"
	"github.com/openziti/ziti/v2/router/xlink"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

func NewBindHandlerFactory(c env.NetworkControllers, f *forwarder.Forwarder, hbo *channel.HeartbeatOptions, mr metrics.Registry, registry xlink.Registry) *bindHandlerFactory {
	return &bindHandlerFactory{
		ctrl:             c,
		forwarder:        f,
		metricsRegistry:  mr,
		xlinkRegistry:    registry,
		heartbeatOptions: hbo,
	}
}

type bindHandlerFactory struct {
	ctrl             env.NetworkControllers
	forwarder        *forwarder.Forwarder
	metricsRegistry  metrics.Registry
	xlinkRegistry    xlink.Registry
	heartbeatOptions *channel.HeartbeatOptions
}

func (self *bindHandlerFactory) NewBindHandler(link xlink.Xlink, latency bool, listenerSide bool) channel.BindHandler {
	return &bindHandler{
		bindHandlerFactory: self,
		xlink:              link,
		trackLatency:       latency,
		listenerSide:       listenerSide,
	}
}

type bindHandler struct {
	*bindHandlerFactory
	xlink        xlink.Xlink
	trackLatency bool
	listenerSide bool
}

func (self *bindHandler) BindChannel(binding channel.Binding) error {
	ch := binding.GetChannel()
	if self.listenerSide {
		if err := self.verifyRouter(self.xlink, ch); err != nil {
			return err
		}
	}

	log := pfxlog.Logger().WithFields(map[string]interface{}{
		"linkId":        self.xlink.Id(),
		"routerId":      self.xlink.DestinationId(),
		"routerVersion": self.xlink.DestVersion(),
		"iteration":     self.xlink.Iteration(),
		"dialed":        self.xlink.IsDialed(),
	})

	binding.GetChannel().SetLogicalName("l/" + self.xlink.Id())
	binding.SetUserData(self.xlink.Id())
	binding.AddCloseHandler(newCloseHandler(self.xlink, self.forwarder, self.xlinkRegistry))
	binding.AddErrorHandler(newErrorHandler(self.xlink, self.ctrl))
	channel.AddReceiveHandlers(binding, newPayloadHandler(self.xlink, self.forwarder))
	channel.AddReceiveHandlers(binding, newAckHandler(self.xlink, self.forwarder))
	binding.AddReceiveHandler(channel.ContentTypeLatencyType, &latency.LatencyHandler{})
	channel.AddReceiveHandlers(binding, newControlHandler(self.xlink, self.forwarder))
	binding.AddPeekHandler(metrics2.NewChannelPeekHandler(self.xlink.Id(), self.forwarder.MetricsRegistry()))
	binding.AddPeekHandler(trace.NewChannelPeekHandler(self.xlink.Id(), ch, self.forwarder.TraceController()))
	if self.xlink.LinkProtocol() == "dtls" {
		binding.AddTransformHandler(xgress.PayloadTransformer{})
	}
	if err := self.xlink.Init(self.forwarder.MetricsRegistry()); err != nil {
		return err
	}

	latencyMetric := self.metricsRegistry.Histogram("link." + self.xlink.Id() + ".latency")
	queueTimeMetric := self.metricsRegistry.Histogram("link." + self.xlink.Id() + ".queue_time")
	binding.AddCloseHandler(channel.CloseHandlerF(func(ch channel.Channel) {
		latencyMetric.Dispose()
		queueTimeMetric.Dispose()
	}))

	log.Info("link destination support heartbeats")
	cb := &heartbeatCallback{
		linkId:           self.xlink.Id(),
		latencyMetric:    latencyMetric,
		queueTimeMetric:  queueTimeMetric,
		ch:               binding.GetChannel(),
		heartbeatOptions: self.heartbeatOptions,
		latencySemaphore: concurrenz.NewSemaphore(2),
		lastResponse:     time.Now().Add(self.heartbeatOptions.CloseUnresponsiveTimeout * 2).UnixMilli(),
	}
	channel.ConfigureHeartbeat(binding, 10*time.Second, time.Second, cb)

	return nil
}

func (self *bindHandler) verifyRouter(l xlink.Xlink, ch channel.Channel) error {
	// Fingerprint only the leaf certificate whose key the TLS handshake proved possession of
	// (certs[0]). Fingerprinting the rest of the presented chain would let a peer present a victim
	// router's certificate as filler and be admitted under the victim's id, since the controller
	// accepts the link if any presented fingerprint matches the claimed router's enrolled one.
	fingerprint, err := leafFingerprint(ch.Certificates())
	if err != nil {
		return errors.Wrapf(err, "unable to verify router for link %v", l.Id())
	}

	verifyLink := &ctrl_pb.VerifyRouter{
		RouterId:     l.DestinationId(),
		Fingerprints: []string{fingerprint},
	}

	ctrlCh := self.ctrl.AnyChannel()
	if ctrlCh == nil {
		return errors.Errorf("unable to verify link %v, no controller available", l.Id())
	}

	reply, err := protobufs.MarshalTyped(verifyLink).WithTimeout(10 * time.Second).SendForReply(ctrlCh)
	if err != nil {
		return errors.Wrapf(err, "unable to verify router %v for link %v", l.DestinationId(), l.Id())
	}

	if reply.ContentType != channel.ContentTypeResultType {
		return errors.Errorf("unexpected response type to verify link: %v", reply.ContentType)
	}

	result := channel.UnmarshalResult(reply)
	if result.Success {
		logrus.WithField("linkId", l.Id()).
			WithField("routerId", l.DestinationId()).
			Info("successfully verified router for link")
		return nil
	}

	return errors.Errorf("unable to verify link [%v]", result.Message)
}

// leafFingerprint returns the fingerprint of the leaf certificate whose key the TLS handshake proved
// possession of (certs[0]). Only the leaf is fingerprinted: fingerprinting the rest of the presented
// chain would let a peer present a victim router's certificate as filler and be admitted under the
// victim's id.
func leafFingerprint(certs []*x509.Certificate) (string, error) {
	if len(certs) == 0 {
		return "", errors.New("no certificates presented")
	}
	return nfpem.FingerprintFromCertificate(certs[0]), nil
}

type heartbeatCallback struct {
	linkId           string
	latencyMetric    metrics.Histogram
	queueTimeMetric  metrics.Histogram
	lastResponse     int64
	heartbeatOptions *channel.HeartbeatOptions
	ch               channel.Channel
	latencySemaphore concurrenz.Semaphore
}

func (self *heartbeatCallback) HeartbeatTx(int64) {}

func (self *heartbeatCallback) HeartbeatRx(int64) {}

func (self *heartbeatCallback) HeartbeatRespTx(int64) {}

func (self *heartbeatCallback) HeartbeatRespRx(ts int64) {
	now := time.Now()
	self.lastResponse = now.UnixMilli()
	self.latencyMetric.Update(now.UnixNano() - ts)
}

func (self *heartbeatCallback) CheckHeartBeat() {
	log := pfxlog.Logger().WithField("channelId", self.ch.Label())
	now := time.Now().UnixMilli()
	if delta := now - self.lastResponse; delta > 30000 {
		log.Warn("heartbeat not received in time, link may be unhealthy")
		self.latencyMetric.Clear()
		self.latencyMetric.Update(8888888888888)

		if delta > self.heartbeatOptions.CloseUnresponsiveTimeout.Milliseconds() {
			log.Error("heartbeat not received in time, closing router link connection")
			if err := self.ch.Close(); err != nil {
				log.WithError(err).Error("error while closing router link connection")
			}
		}
	}

	go self.checkQueueTime()
}

func (self *heartbeatCallback) checkQueueTime() {
	log := pfxlog.Logger().WithField("linkId", self.linkId)
	if !self.latencySemaphore.TryAcquire() {
		log.Warn("unable to check queue time, too many check already running")
		return
	}

	defer self.latencySemaphore.Release()

	sendTracker := &latency.SendTimeTracker{
		Handler: func(latencyType latency.Type, latency time.Duration) {
			self.queueTimeMetric.Update(latency.Nanoseconds())
		},
		StartTime: time.Now(),
	}
	if err := self.ch.Send(sendTracker); err != nil && !self.ch.IsClosed() {
		log.WithError(err).Error("unable to send queue time tracer")
	}
}