multicast.go

package main

import (
	"errors"
	"io"
	"net/netip"

	"github.com/sirupsen/logrus"
	"github.com/slackhq/nebula"
	"github.com/slackhq/nebula/config"
	"github.com/slackhq/nebula/overlay"
)

const multicastQueueSize = 1024

type multicastDevice struct {
	overlay.Device
	l       *logrus.Logger
	targets func() []netip.Addr
	queue   chan []byte
}

func newMulticastDeviceFactory(c *config.C, ctrl **nebula.Control) (overlay.DeviceFactory, error) {
	multicastGroups := c.GetStringSlice("multicast.groups", nil)
	if tags := c.GetStringSlice("multicast.tags", nil); len(multicastGroups) == 0 && len(tags) > 0 {
		multicastGroups = tags
	}
	if group := c.GetString("multicast.group", ""); group != "" && len(multicastGroups) == 0 {
		multicastGroups = []string{group}
	}
	if tag := c.GetString("multicast.tag", ""); tag != "" && len(multicastGroups) == 0 {
		multicastGroups = []string{tag}
	}

	multicastEnabled := c.GetBool("multicast.enabled", len(multicastGroups) > 0)
	if !multicastEnabled {
		return nil, nil
	}

	groupSet := make(map[string]struct{}, len(multicastGroups))
	for _, group := range multicastGroups {
		if group != "" {
			groupSet[group] = struct{}{}
		}
	}
	if len(groupSet) == 0 {
		return nil, errors.New("multicast.enabled requires multicast.groups, multicast.group, multicast.tags, or multicast.tag")
	}

	return func(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, routines int) (overlay.Device, error) {
		device, err := overlay.NewDeviceFromConfig(c, l, vpnNetworks, routines)
		if err != nil {
			return nil, err
		}

		return newMulticastDevice(device, l, func() []netip.Addr {
			if *ctrl == nil {
				return nil
			}

			return multicastTargets(*ctrl, groupSet)
		}), nil
	}, nil
}

func newMulticastDevice(device overlay.Device, l *logrus.Logger, targets func() []netip.Addr) *multicastDevice {
	return &multicastDevice{
		Device:  device,
		l:       l,
		targets: targets,
		queue:   make(chan []byte, multicastQueueSize),
	}
}

func (d *multicastDevice) Read(p []byte) (int, error) {
	for {
		select {
		case packet := <-d.queue:
			return copy(p, packet), nil
		default:
		}

		n, err := d.Device.Read(p)
		if err != nil {
			return n, err
		}

		if !isMulticastPacket(p[:n]) {
			return n, nil
		}

		forwarded := d.forwardMulticast(p[:n])
		if forwarded == 0 {
			d.l.Debug("No multicast targets matched")
		}
	}
}

func (d *multicastDevice) SupportsMultiqueue() bool {
	return false
}

func (d *multicastDevice) NewMultiQueueReader() (io.ReadWriteCloser, error) {
	return nil, errors.New("multicast wrapper does not support multiqueue")
}

func (d *multicastDevice) forwardMulticast(packet []byte) int {
	var forwarded int
	for _, target := range d.targets() {
		clone, ok := cloneAsUnicast(packet, target)
		if !ok {
			continue
		}

		select {
		case d.queue <- clone:
			forwarded++
		default:
			d.l.Warn("Multicast queue full, dropping forwarded packet")
			return forwarded
		}
	}

	if forwarded > 0 {
		d.l.WithField("targets", forwarded).Debug("Forwarded multicast packet")
	}
	return forwarded
}

func multicastTargets(ctrl *nebula.Control, groups map[string]struct{}) []netip.Addr {
	return multicastTargetAddrs(ctrl.ListHostmapHosts(false), groups)
}

func multicastTargetAddrs(hosts []nebula.ControlHostInfo, groups map[string]struct{}) []netip.Addr {
	targets := make([]netip.Addr, 0, len(hosts))
	seen := map[netip.Addr]struct{}{}

	for _, host := range hosts {
		if host.Cert == nil || !hasAnyGroup(host.Cert.Groups(), groups) {
			continue
		}

		for _, addr := range host.VpnAddrs {
			if _, ok := seen[addr]; ok {
				continue
			}
			seen[addr] = struct{}{}
			targets = append(targets, addr)
		}
	}

	return targets
}

func hasAnyGroup(hostGroups []string, groups map[string]struct{}) bool {
	for _, group := range hostGroups {
		if _, ok := groups[group]; ok {
			return true
		}
	}
	return false
}

var _ overlay.Device = (*multicastDevice)(nil)