Merge branch 'main' of github.com:gopatchy/tendrils

cmd/tendrils/main.go

@@ -15,6 +15,7 @@ func main() {
 	noIGMP := flag.Bool("no-igmp", false, "disable IGMP querier")
 	noMDNS := flag.Bool("no-mdns", false, "disable mDNS discovery")
 	noArtNet := flag.Bool("no-artnet", false, "disable Art-Net discovery")
+	noSACN := flag.Bool("no-sacn", false, "disable sACN discovery")
 	noDante := flag.Bool("no-dante", false, "disable Dante discovery")
 	noBMD := flag.Bool("no-bmd", false, "disable Blackmagic discovery")
 	noShure := flag.Bool("no-shure", false, "disable Shure discovery")
@@ -27,6 +28,7 @@ func main() {
 	debugIGMP := flag.Bool("debug-igmp", false, "debug IGMP querier")
 	debugMDNS := flag.Bool("debug-mdns", false, "debug mDNS discovery")
 	debugArtNet := flag.Bool("debug-artnet", false, "debug Art-Net discovery")
+	debugSACN := flag.Bool("debug-sacn", false, "debug sACN discovery")
 	debugDante := flag.Bool("debug-dante", false, "debug Dante discovery")
 	debugBMD := flag.Bool("debug-bmd", false, "debug Blackmagic discovery")
 	debugShure := flag.Bool("debug-shure", false, "debug Shure discovery")
@@ -43,6 +45,7 @@ func main() {
 	t.DisableIGMP = *noIGMP
 	t.DisableMDNS = *noMDNS
 	t.DisableArtNet = *noArtNet
+	t.DisableSACN = *noSACN
 	t.DisableDante = *noDante
 	t.DisableBMD = *noBMD
 	t.DisableShure = *noShure
@@ -55,6 +58,7 @@ func main() {
 	t.DebugIGMP = *debugIGMP
 	t.DebugMDNS = *debugMDNS
 	t.DebugArtNet = *debugArtNet
+	t.DebugSACN = *debugSACN
 	t.DebugDante = *debugDante
 	t.DebugBMD = *debugBMD
 	t.DebugShure = *debugShure

http.go

@@ -31,6 +31,7 @@ type StatusResponse struct {
 	MulticastGroups  []*MulticastGroupMembers `json:"multicast_groups"`
 	ArtNetNodes      []*ArtNetNode            `json:"artnet_nodes"`
 	SACNNodes        []*SACNNode              `json:"sacn_nodes"`
+	SACNSources      []*SACNSource            `json:"sacn_sources"`
 	DanteFlows       []*DanteFlow             `json:"dante_flows"`
 	PortErrors       []*PortError             `json:"port_errors"`
 	UnreachableNodes []string                 `json:"unreachable_nodes"`
@@ -141,6 +142,7 @@ func (t *Tendrils) GetStatus() *StatusResponse {
 		MulticastGroups:  t.getMulticastGroups(),
 		ArtNetNodes:      t.getArtNetNodes(),
 		SACNNodes:        t.getSACNNodes(),
+		SACNSources:      t.getSACNSources(),
 		DanteFlows:       t.getDanteFlows(),
 		PortErrors:       t.errors.GetErrors(),
 		UnreachableNodes: t.errors.GetUnreachableNodes(),

sacn_discovery.go

@@ -0,0 +1,225 @@
+package tendrils
+
+import (
+	"context"
+	"encoding/binary"
+	"log"
+	"net"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/fvbommel/sortorder"
+	"golang.org/x/net/ipv4"
+)
+
+const (
+	sacnPort                    = 5568
+	vectorRootE131Extended      = 0x00000008
+	vectorE131Discovery         = 0x00000002
+	vectorUniverseDiscovery     = 0x00000001
+)
+
+var sacnDiscoveryAddr = net.IPv4(239, 255, 250, 214)
+
+var sacnPacketIdentifier = [12]byte{
+	0x41, 0x53, 0x43, 0x2d, 0x45, 0x31, 0x2e, 0x31, 0x37, 0x00, 0x00, 0x00,
+}
+
+type SACNSource struct {
+	TypeID     string    `json:"typeid"`
+	Node       *Node     `json:"node"`
+	SourceName string    `json:"source_name"`
+	CID        string    `json:"cid"`
+	Universes  []int     `json:"universes"`
+	LastSeen   time.Time `json:"last_seen"`
+}
+
+type SACNSources struct {
+	mu      sync.RWMutex
+	sources map[string]*SACNSource
+}
+
+func NewSACNSources() *SACNSources {
+	return &SACNSources{
+		sources: map[string]*SACNSource{},
+	}
+}
+
+func (s *SACNSources) Update(cid [16]byte, sourceName string, universes []int, srcIP net.IP) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	cidStr := formatCID(cid)
+	existing, exists := s.sources[cidStr]
+	if exists {
+		existing.SourceName = sourceName
+		existing.Universes = universes
+		existing.LastSeen = time.Now()
+	} else {
+		s.sources[cidStr] = &SACNSource{
+			TypeID:     newTypeID("sacnsource"),
+			SourceName: sourceName,
+			CID:        cidStr,
+			Universes:  universes,
+			LastSeen:   time.Now(),
+		}
+	}
+}
+
+func (s *SACNSources) SetNode(cid string, node *Node) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if source, exists := s.sources[cid]; exists {
+		source.Node = node
+	}
+}
+
+func (s *SACNSources) Expire() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	expireTime := time.Now().Add(-60 * time.Second)
+	for cid, source := range s.sources {
+		if source.LastSeen.Before(expireTime) {
+			delete(s.sources, cid)
+		}
+	}
+}
+
+func (s *SACNSources) GetAll() []*SACNSource {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	result := make([]*SACNSource, 0, len(s.sources))
+	for _, source := range s.sources {
+		result = append(result, source)
+	}
+	return result
+}
+
+func formatCID(cid [16]byte) string {
+	return strings.ToLower(formatUUID(cid))
+}
+
+func formatUUID(b [16]byte) string {
+	return strings.ToUpper(
+		strings.Join([]string{
+			encodeHex(b[0:4]),
+			encodeHex(b[4:6]),
+			encodeHex(b[6:8]),
+			encodeHex(b[8:10]),
+			encodeHex(b[10:16]),
+		}, "-"),
+	)
+}
+
+func encodeHex(b []byte) string {
+	const hexChars = "0123456789ABCDEF"
+	result := make([]byte, len(b)*2)
+	for i, v := range b {
+		result[i*2] = hexChars[v>>4]
+		result[i*2+1] = hexChars[v&0x0f]
+	}
+	return string(result)
+}
+
+func (t *Tendrils) startSACNDiscoveryListener(ctx context.Context, iface net.Interface) {
+	c, err := net.ListenPacket("udp4", ":5568")
+	if err != nil {
+		log.Printf("[ERROR] failed to listen sacn discovery: %v", err)
+		return
+	}
+	defer c.Close()
+
+	p := ipv4.NewPacketConn(c)
+
+	if err := p.JoinGroup(&iface, &net.UDPAddr{IP: sacnDiscoveryAddr}); err != nil {
+		log.Printf("[ERROR] failed to join sacn discovery multicast on %s: %v", iface.Name, err)
+		return
+	}
+
+	if t.DebugSACN {
+		log.Printf("[sacn] listening for discovery on %s", iface.Name)
+	}
+
+	buf := make([]byte, 1500)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		c.SetReadDeadline(time.Now().Add(1 * time.Second))
+		n, _, err := c.ReadFrom(buf)
+		if err != nil {
+			if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+				continue
+			}
+			continue
+		}
+
+		t.handleSACNDiscoveryPacket(buf[:n])
+	}
+}
+
+func (t *Tendrils) handleSACNDiscoveryPacket(data []byte) {
+	if len(data) < 120 {
+		return
+	}
+
+	if data[4] != sacnPacketIdentifier[0] || data[5] != sacnPacketIdentifier[1] ||
+		data[6] != sacnPacketIdentifier[2] || data[7] != sacnPacketIdentifier[3] {
+		return
+	}
+
+	rootVector := binary.BigEndian.Uint32(data[18:22])
+	if rootVector != vectorRootE131Extended {
+		return
+	}
+
+	framingVector := binary.BigEndian.Uint32(data[40:44])
+	if framingVector != vectorE131Discovery {
+		return
+	}
+
+	var cid [16]byte
+	copy(cid[:], data[22:38])
+
+	sourceName := strings.TrimRight(string(data[44:108]), "\x00")
+
+	discoveryVector := binary.BigEndian.Uint32(data[114:118])
+	if discoveryVector != vectorUniverseDiscovery {
+		return
+	}
+
+	universeCount := (len(data) - 120) / 2
+	universes := make([]int, 0, universeCount)
+	for i := 0; i < universeCount; i++ {
+		u := binary.BigEndian.Uint16(data[120+i*2 : 122+i*2])
+		if u >= 1 && u <= 63999 {
+			universes = append(universes, int(u))
+		}
+	}
+
+	if t.DebugSACN {
+		log.Printf("[sacn] discovery from %q cid=%s universes=%v", sourceName, formatCID(cid), universes)
+	}
+
+	t.sacnSources.Update(cid, sourceName, universes, nil)
+	t.NotifyUpdate()
+}
+
+func (t *Tendrils) getSACNSources() []*SACNSource {
+	t.sacnSources.Expire()
+
+	sources := t.sacnSources.GetAll()
+	sort.Slice(sources, func(i, j int) bool {
+		return sortorder.NaturalLess(sources[i].SourceName, sources[j].SourceName)
+	})
+
+	return sources
+}

tendrils.go

@@ -35,6 +35,7 @@ type Tendrils struct {
 	nodes            *Nodes
 	artnet           *ArtNetNodes
 	artnetConn       *net.UDPConn
+	sacnSources      *SACNSources
 	danteFlows       *DanteFlows
 	errors           *ErrorTracker
 	ping             *PingManager
@@ -53,6 +54,7 @@ type Tendrils struct {
 	DisableIGMP    bool
 	DisableMDNS    bool
 	DisableArtNet  bool
+	DisableSACN    bool
 	DisableDante   bool
 	DisableBMD     bool
 	DisableShure   bool
@@ -65,6 +67,7 @@ type Tendrils struct {
 	DebugIGMP      bool
 	DebugMDNS      bool
 	DebugArtNet    bool
+	DebugSACN      bool
 	DebugDante     bool
 	DebugBMD       bool
 	DebugShure     bool
@@ -76,6 +79,7 @@ func New() *Tendrils {
 	t := &Tendrils{
 		activeInterfaces: map[string]context.CancelFunc{},
 		artnet:           NewArtNetNodes(),
+		sacnSources:      NewSACNSources(),
 		danteFlows:       NewDanteFlows(),
 		ping:             NewPingManager(),
 		sseSubs:          map[int]chan struct{}{},
@@ -302,6 +306,9 @@ func (t *Tendrils) startInterface(ctx context.Context, iface net.Interface) {
 	if !t.DisableArtNet {
 		go t.startArtNetPoller(ctx, iface)
 	}
+	if !t.DisableSACN {
+		go t.startSACNDiscoveryListener(ctx, iface)
+	}
 	if !t.DisableDante {
 		go t.listenDante(ctx, iface)
 	}