lldp.go

// +build linux

/*
 * Copyright (C) 2018 Red Hat, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy ofthe License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specificlanguage governing permissions and
 * limitations under the License.
 *
 */

package lldp

import (
	"bytes"
	"fmt"
	"net"
	"strings"
	"sync"
	"syscall"
	"time"
	"unsafe"

	"github.com/google/gopacket"
	"github.com/google/gopacket/layers"
	"github.com/safchain/ethtool"
	"golang.org/x/sys/unix"

	"github.com/skydive-project/skydive/common"
	"github.com/skydive-project/skydive/flow/probes"
	"github.com/skydive-project/skydive/graffiti/graph"
	"github.com/skydive-project/skydive/probe"
	"github.com/skydive-project/skydive/topology"
	tp "github.com/skydive-project/skydive/topology/probes"
)

/*
#include <linux/if_packet.h>  // packet_mreq
#include <linux/if_ether.h>  // ETH_ALEN
*/
import "C"

const lldpBPFFilter = `ether[0] & 1 = 1 and
  ether proto 0x88cc and
  !(ether src %s) and
  (ether dst 01:80:c2:00:00:0e or
   ether dst 01:80:c2:00:00:03 or
   ether dst 01:80:c2:00:00:00)`

// Capture 8192 bytes so that we have the full Ethernet frame
const lldpSnapLen = 8192

const (
	SwitchUP           string = "UP"
	SwitchInaccessible string = "INACCESSIBLE"
	SwitchDOWN         string = "DOWN"
)

const (
	LinkActive   string = "ACTIVE"
	LinkInactive string = "INACTIVE"
)

// Probe describes the probe that is in charge of listening for
// LLDP packets on interfaces and create the corresponding chassis and port nodes
type Probe struct {
	sync.RWMutex
	graph.DefaultGraphListener
	Ctx           tp.Context
	interfaceMap  map[string]*probes.GoPacketProbe // map interface names to the packet probes
	state         common.ServiceState              // state of the probe (running or stopped)
	wg            sync.WaitGroup                   // capture goroutines wait group
	autoDiscovery bool                             // capture LLDP traffic on all capable interfaces
	ignoreDCBX    bool                             // ignore dcbx packets
	linkWatcher   *Watcher                         // remember the last time each link is updated
}

type Watcher struct {
	sync.Mutex
	elements map[graph.Identifier]time.Time // mapping from element's ID and the last moment it was updated
	interval time.Duration                  // watcher interval
	timeout  time.Duration                  // update element after timeout
	quit     chan bool                      // stop watcher
}

func (w *Watcher) Start(g *graph.Graph, wg *sync.WaitGroup) {
	ticker := time.NewTicker(w.interval)
	go func() {
		defer func() {
			ticker.Stop()
			wg.Done()
		}()
		for {
			select {
			case <-ticker.C:
				now := time.Now().UTC()
				g.Lock()
				for id, last := range w.elements {
					if now.Sub(last) < w.timeout {
						continue
					}
					if link := g.GetEdge(id); link != nil {
						g.AddMetadata(link, "State", LinkInactive)
						w.DropElement(id)
					}
				}
				g.Unlock()
			case <-w.quit:
				return
			}
		}
	}()
}

func (w *Watcher) Stop() {
	w.quit <- true
}

func (w *Watcher) UpdateElement(id graph.Identifier) {
	w.Lock()
	defer w.Unlock()
	w.elements[id] = time.Now().UTC()
}

func (w *Watcher) DropElement(id graph.Identifier) {
	w.Lock()
	defer w.Unlock()
	delete(w.elements, id)
}

type ifreq struct {
	ifrName   [ethtool.IFNAMSIZ]byte
	ifrHwaddr syscall.RawSockaddr
}

type lldpCapture struct{}

func (c *lldpCapture) OnStarted(*probes.CaptureMetadata) {
}

func (c *lldpCapture) OnStopped() {
}

func (c *lldpCapture) OnError(err error) {
}

// addMulticastAddr adds a multicast address to an interface using an ioctl call
func addMulticastAddr(intf string, addr string) error {
	fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_DGRAM, 0)
	if err != nil {
		return err
	}
	defer syscall.Close(fd)

	var name [ethtool.IFNAMSIZ]byte
	copy(name[:], []byte(intf))

	mac, _ := net.ParseMAC(addr)

	var sockaddr syscall.RawSockaddr
	// RawSockaddr.Data is []int8 on amd64 and []uint8 on ppc64le
	switch data := interface{}(sockaddr.Data).(type) {
	case [14]int8:
		for i, n := range mac {
			data[i] = int8(n)
		}
	case [14]uint8:
		for i, n := range mac {
			data[i] = uint8(n)
		}
	}

	ifr := &ifreq{
		ifrName:   name,
		ifrHwaddr: sockaddr,
	}

	_, _, ep := unix.Syscall(unix.SYS_IOCTL, uintptr(fd),
		unix.SIOCADDMULTI, uintptr(unsafe.Pointer(ifr)))

	if ep != 0 {
		return syscall.Errno(ep)
	}
	return nil
}

func (p *Probe) handlePacket(n *graph.Node, ifName string, packet gopacket.Packet) {
	lldpLayer := packet.Layer(layers.LayerTypeLinkLayerDiscovery)
	if lldpLayer != nil {
		lldpLayer := lldpLayer.(*layers.LinkLayerDiscovery)

		bytesToString := func(b []byte) string {
			return string(bytes.Trim(b, "\x00"))
		}

		chassisLLDPMetadata := &Metadata{
			ChassisIDType: lldpLayer.ChassisID.Subtype.String(),
		}

		chassisMetadata := graph.Metadata{
			"LLDP":  chassisLLDPMetadata,
			"Type":  "switch",
			"State": SwitchUP,
			"Probe": "lldp",
		}

		var chassisID string
		var chassisDiscriminators []string
		switch lldpLayer.ChassisID.Subtype {
		case layers.LLDPChassisIDSubTypeMACAddr:
			chassisID = net.HardwareAddr(lldpLayer.ChassisID.ID).String()
		default:
			chassisID = bytesToString(lldpLayer.ChassisID.ID)
		}
		chassisLLDPMetadata.ChassisID = chassisID
		chassisMetadata.SetField("Name", chassisID)

		portLLDPMetadata := &Metadata{
			PortIDType: lldpLayer.PortID.Subtype.String(),
		}
		portMetadata := graph.Metadata{
			"LLDP":  portLLDPMetadata,
			"Type":  "switchport",
			"Probe": "lldp",
		}

		var portID string
		switch lldpLayer.PortID.Subtype {
		case layers.LLDPPortIDSubtypeMACAddr:
			portID = net.HardwareAddr(lldpLayer.PortID.ID).String()
		default:
			portID = bytesToString(lldpLayer.PortID.ID)
		}
		portLLDPMetadata.PortID = portID
		portMetadata.SetField("Name", portID)

		if lldpLayerInfo := packet.Layer(layers.LayerTypeLinkLayerDiscoveryInfo); lldpLayerInfo != nil {
			lldpLayerInfo := lldpLayerInfo.(*layers.LinkLayerDiscoveryInfo)

			// Stop early on DCBX packets if ignoreDCBX equals true
			if p.ignoreDCBX {
				for _, tlv := range lldpLayerInfo.OrgTLVs {
					if tlv.OUI == layers.IEEEOUIDCBX {
						return
					}
				}
			}

			if portIDSubtype := lldpLayer.PortID.Subtype; portIDSubtype == layers.LLDPPortIDSubtypeIfaceName {
				portMetadata["Name"] = bytesToString(lldpLayer.PortID.ID)
			} else {
				portDescription := lldpLayerInfo.PortDescription
				portMetadata["Name"] = bytesToString([]byte(portDescription))
			}

			if portDescription := lldpLayerInfo.PortDescription; portDescription != "" {
				portLLDPMetadata.Description = portDescription
			}

			if lldpLayerInfo.SysDescription != "" {
				chassisLLDPMetadata.Description = bytesToString([]byte(lldpLayerInfo.SysDescription))
			}

			if sysName := bytesToString([]byte(lldpLayerInfo.SysName)); sysName != "" {
				chassisDiscriminators = append(chassisDiscriminators, sysName, "SysName")
				chassisLLDPMetadata.SysName = sysName
				chassisMetadata["Name"] = sysName
			}

			if mgmtAddress := lldpLayerInfo.MgmtAddress; len(mgmtAddress.Address) > 0 {
				var addr string
				switch mgmtAddress.Subtype {
				case layers.IANAAddressFamilyIPV4, layers.IANAAddressFamilyIPV6:
					addr = net.IP(mgmtAddress.Address).String()
				case layers.IANAAddressFamilyDistname:
					addr = bytesToString(mgmtAddress.Address)
				}

				if addr != "" {
					chassisDiscriminators = append(chassisDiscriminators, addr, "MgmtAddress")
					chassisLLDPMetadata.MgmtAddress = addr
				}
			}

			if lldp8201Q, err := lldpLayerInfo.Decode8021(); err == nil {
				if lldp8201Q.LinkAggregation.Supported {
					portLLDPMetadata.LinkAggregation = &LinkAggregationMetadata{
						Enabled:   lldp8201Q.LinkAggregation.Enabled,
						PortID:    int64(lldp8201Q.LinkAggregation.PortID),
						Supported: lldp8201Q.LinkAggregation.Supported,
					}
				}

				if lldp8201Q.PVID != 0 {
					portLLDPMetadata.PVID = int64(lldp8201Q.PVID)
				}

				if lldp8201Q.VIDUsageDigest != 0 {
					portLLDPMetadata.VIDUsageDigest = int64(lldp8201Q.VIDUsageDigest)
				}

				if lldp8201Q.ManagementVID != 0 {
					portLLDPMetadata.ManagementVID = int64(lldp8201Q.ManagementVID)
				}

				if len(lldp8201Q.VLANNames) != 0 {
					portLLDPMetadata.VLANNames = make([]VLANNameMetadata, len(lldp8201Q.VLANNames))
					for i, vlan := range lldp8201Q.VLANNames {
						portLLDPMetadata.VLANNames[i].ID = int64(vlan.ID)
						portLLDPMetadata.VLANNames[i].Name = bytesToString([]byte(vlan.Name))
					}
				}

				if len(lldp8201Q.PPVIDs) != 0 {
					portLLDPMetadata.PPVIDs = make([]PPVIDMetadata, len(lldp8201Q.PPVIDs))
					for i, ppvid := range lldp8201Q.PPVIDs {
						portLLDPMetadata.PPVIDs[i].Enabled = ppvid.Enabled
						portLLDPMetadata.PPVIDs[i].ID = int64(ppvid.ID)
						portLLDPMetadata.PPVIDs[i].Supported = ppvid.Supported
					}
				}
			}

			if lldp8023, err := lldpLayerInfo.Decode8023(); err == nil {
				if lldp8023.MTU != 0 {
					portMetadata["MTU"] = int64(lldp8023.MTU)
				}
			}
		}

		p.Ctx.Graph.Lock()

		// Create a node for the sending chassis with a predictable ID
		// Some switches - such as Cisco Nexus - sends a different chassis ID
		// for each port, so you use SysName and MgmtAddress if present and
		// fallback to chassis ID otherwise.
		if len(chassisDiscriminators) == 0 {
			chassisDiscriminators = append(chassisDiscriminators, chassisID, lldpLayer.ChassisID.Subtype.String())
		}

		chassisNodeID := graph.GenID(chassisDiscriminators...)
		chassis := p.getOrCreate(chassisNodeID, chassisMetadata)

		// Delete nodes with the same mgmt address
		// Reason: MgmtAddress must be unique between switches
		if mgmtAddress, _ := chassisMetadata.GetFieldString("LLDP.MgmtAddress"); len(mgmtAddress) > 0 {
			dupSwitches := p.Ctx.Graph.GetNodes(graph.Metadata{"LLDP.MgmtAddress": mgmtAddress})
			for _, sw := range dupSwitches {
				if sw != chassis {
					for _, pt := range p.Ctx.Graph.LookupChildren(sw, nil, topology.OwnershipMetadata()) {
						p.Ctx.Graph.DelNode(pt)
					}
					p.Ctx.Graph.DelNode(sw)
				}
			}
		}

		// Create a port with a predicatable ID
		port := p.getOrCreate(graph.GenID(
			string(chassisNodeID),
			portID, lldpLayer.PortID.Subtype.String(),
		), portMetadata)

		if !topology.HaveOwnershipLink(p.Ctx.Graph, chassis, port) {
			topology.AddOwnershipLink(p.Ctx.Graph, chassis, port, nil)
		}

		var err error
		link := p.Ctx.Graph.GetFirstLink(port, n, topology.Layer2Metadata())
		if link == nil {
			link, err = topology.AddLayer2Link(p.Ctx.Graph, port, n, nil)
		}
		if err == nil {
			p.linkWatcher.UpdateElement(link.ID)
			p.Ctx.Graph.AddMetadata(link, "State", LinkActive)
		}
		p.Ctx.Graph.Unlock()
	}
}

func (p *Probe) startCapture(ifName, mac string, n *graph.Node) error {
	lldpPrefix := "01:80:c2:00:00"
	for _, lastByte := range []byte{0x00, 0x03, 0x0e} {
		// Add multicast address so that the kernel does not discard it
		if err := addMulticastAddr(ifName, fmt.Sprintf("%s:%02X", lldpPrefix, lastByte)); err != nil {
			return fmt.Errorf("Failed to add multicast address: %s", err)
		}
	}

	// Set BPF filter to only capture LLDP packets
	bpfFilter := fmt.Sprintf(lldpBPFFilter, mac)

	ctx := probes.Context{
		Config: p.Ctx.Config,
		Logger: p.Ctx.Logger,
		Graph:  p.Ctx.Graph,
	}

	packetProbe, err := probes.NewGoPacketProbe(ctx, n, probes.AFPacket, bpfFilter, lldpSnapLen)
	if err != nil {
		return err
	}

	// Set capturing state to true
	p.interfaceMap[ifName] = packetProbe

	p.wg.Add(1)

	go func() {
		defer func() {
			p.Ctx.Logger.Infof("Stopping LLDP capture on %s", ifName)

			p.Lock()
			p.interfaceMap[ifName] = nil
			p.Unlock()

			p.wg.Done()
		}()

		if err := packetProbe.Run(func(packet gopacket.Packet) {
			p.handlePacket(n, ifName, packet)
		}, &lldpCapture{}); err != nil {
			p.Ctx.Logger.Errorf("LLDP capture error on %s", ifName)
		}
	}()

	return err
}

func (p *Probe) getOrCreate(id graph.Identifier, m graph.Metadata) *graph.Node {
	node := p.Ctx.Graph.GetNode(id)
	if node == nil {
		var err error

		node, err = p.Ctx.Graph.NewNode(id, m)
		if err != nil {
			p.Ctx.Logger.Error(err)
		}
	} else {
		tr := p.Ctx.Graph.StartMetadataTransaction(node)
		for k, v := range m {
			// Do not update chassis ID field on cisco nexus switches
			if k == "LLDP" {
				cid, err := node.GetFieldString("LLDP.ChassisID")
				cdesc, _ := node.GetFieldString("LLDP.Description")
				if err == nil && strings.Contains(strings.ToLower(cdesc), "cisco nexus") {
					v1 := v.(*Metadata)
					v1.ChassisID = cid
					v = v1
				}
			}
			tr.AddMetadata(k, v)
		}
		tr.Commit()
	}
	return node
}

// handleNode checks if LLDP capture should be started on the passed interface node, ie:
// - when no LLDP capture is running for this interface
// - when its first packet layer is Ethernet and it has a MAC address
// - when the interface is listed in the configuration file or we are in auto discovery mode
func (p *Probe) handleNode(n *graph.Node) {
	if state, _ := n.GetFieldString("State"); state != "UP" {
		return
	}

	firstLayerType, _ := probes.GoPacketFirstLayerType(n)
	mac, _ := n.GetFieldString("BondSlave.PermMAC")
	if mac == "" {
		mac, _ = n.GetFieldString("MAC")
	}
	name, _ := n.GetFieldString("Name")

	if name != "" && mac != "" && firstLayerType == layers.LayerTypeEthernet {
		if activeProbe, found := p.interfaceMap[name]; (found || p.autoDiscovery) && activeProbe == nil {
			p.Ctx.Logger.Infof("Starting LLDP capture on %s (MAC: %s)", name, mac)
			if err := p.startCapture(name, mac, n); err != nil {
				p.Ctx.Logger.Error(err)
			}
		}
	}
}

// OnEdgeAdded is called when a new edge was created on the graph
func (p *Probe) OnEdgeAdded(e *graph.Edge) {
	p.Lock()
	defer p.Unlock()

	// Only consider nodes that are owned by the host node
	if e.Parent == p.Ctx.RootNode.ID {
		if relationType, _ := e.GetFieldString("RelationType"); relationType == topology.OwnershipLink {
			n := p.Ctx.Graph.GetNode(e.Child)
			p.handleNode(n)
		}
	}
}

// OnNodeUpdated is called when a new node was updated on the graph
func (p *Probe) OnNodeUpdated(n *graph.Node) {
	p.Lock()
	defer p.Unlock()

	if p.Ctx.Graph.AreLinked(p.Ctx.RootNode, n, topology.OwnershipMetadata()) {
		p.handleNode(n)
	}
}

// Start capturing LLDP packets
func (p *Probe) Start() {
	p.state.Store(common.RunningState)
	p.Ctx.Graph.AddEventListener(p)

	p.Ctx.Graph.RLock()
	defer p.Ctx.Graph.RUnlock()
	p.Lock()
	defer p.Unlock()

	p.wg.Add(1)
	p.linkWatcher.Start(p.Ctx.Graph, &p.wg)

	// The nodes may have already been created
	children := p.Ctx.Graph.LookupChildren(p.Ctx.RootNode, nil, topology.OwnershipMetadata())
	for _, intfNode := range children {
		p.handleNode(intfNode)
	}
}

// Stop capturing LLDP packets
func (p *Probe) Stop() {
	p.Ctx.Graph.RemoveEventListener(p)
	p.linkWatcher.Stop()
	p.state.Store(common.StoppingState)
	for intf, activeProbe := range p.interfaceMap {
		if activeProbe != nil {
			p.Ctx.Logger.Debugf("Stopping probe on %s", intf)
			activeProbe.Stop()
		}
	}
	p.wg.Wait()
}

// Init initializes a new LLDP probe
func (p *Probe) Init(ctx tp.Context, bundle *probe.Bundle) (probe.Handler, error) {
	var (
		interfaces  = ctx.Config.GetStringSlice("agent.topology.lldp.interfaces")
		ignoreDCBX  = ctx.Config.GetBool("agent.topology.lldp.ignore_dcbx")
		timeoutStr  = ctx.Config.GetString("agent.topology.lldp.timeout")
		intervalStr = ctx.Config.GetString("agent.topology.lldp.interval")
		watcher     = &Watcher{elements: make(map[graph.Identifier]time.Time), quit: make(chan bool)}
	)
	interfaceMap := make(map[string]*probes.GoPacketProbe)
	for _, intf := range interfaces {
		interfaceMap[intf] = nil
	}

	if timeout, err := time.ParseDuration(timeoutStr); err == nil {
		watcher.timeout = timeout
	} else {
		watcher.timeout = 5 * time.Minute
	}

	if interval, err := time.ParseDuration(intervalStr); err == nil {
		watcher.interval = interval
	} else {
		watcher.interval = 10 * time.Minute
	}

	p.Ctx = ctx
	p.interfaceMap = interfaceMap
	p.state = common.StoppedState
	p.autoDiscovery = len(interfaces) == 0
	p.ignoreDCBX = ignoreDCBX
	p.linkWatcher = watcher

	return p, nil
}