From fdbf00ca76e3a2d7dc0086201fa110065a7cb926 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mattias=20Walstr=C3=B6m?= Date: Tue, 31 Mar 2026 15:31:39 +0200 Subject: [PATCH] Fix routing and ip addresses --- package/yangerd/yangerd.mk | 1 + src/statd/statd.c | 6 +- src/yangerd/internal/ipbatch/ipbatch.go | 46 +- src/yangerd/internal/monitor/monitor.go | 115 +++- src/yangerd/internal/zapi/zapi.go | 540 ++++++++++++++++++ src/yangerd/internal/zapi/zapi_test.go | 350 ++++++++++++ .../internal/zapiwatcher/zapiwatcher.go | 245 +++----- .../internal/zapiwatcher/zapiwatcher_test.go | 115 ++-- 8 files changed, 1166 insertions(+), 252 deletions(-) create mode 100644 src/yangerd/internal/zapi/zapi.go create mode 100644 src/yangerd/internal/zapi/zapi_test.go diff --git a/package/yangerd/yangerd.mk b/package/yangerd/yangerd.mk index 5cf6019a..b12ad4b5 100644 --- a/package/yangerd/yangerd.mk +++ b/package/yangerd/yangerd.mk @@ -28,6 +28,7 @@ define YANGERD_INSTALL_EXTRA >> $(TARGET_DIR)/etc/default/yangerd echo 'YANGERD_ENABLE_GPS=$(if $(BR2_PACKAGE_GPSD),true,false)' \ >> $(TARGET_DIR)/etc/default/yangerd + echo 'YANGERD_LOG_LEVEL=deug' >> $(TARGET_DIR)/etc/default/yangerd endef YANGERD_POST_INSTALL_TARGET_HOOKS += YANGERD_INSTALL_EXTRA diff --git a/src/statd/statd.c b/src/statd/statd.c index 77db0b5c..fe140acf 100644 --- a/src/statd/statd.c +++ b/src/statd/statd.c @@ -81,6 +81,8 @@ static int ly_add_yangerd_data(const struct ly_ctx *ctx, struct lyd_node **paren return SR_ERR_SYS; } + NOTE("yangerd: got %zu bytes JSON for %s", len, path); + err = lyd_parse_data_mem(ctx, json, LYD_JSON, LYD_PARSE_ONLY, 0, parent); if (err) ERROR("Error, parsing yanger data (%d): %s", err, ly_errmsg(ctx)); @@ -142,11 +144,11 @@ static int sr_iface_cb(sr_session_ctx_t *session, uint32_t, const char *, err = ly_add_yangerd_data(ctx, parent, "ietf-interfaces:interfaces"); if (err) - ERROR("Error adding interface data"); + ERROR("Error adding interface data (err %d)", err); sr_release_context(con); - return SR_ERR_OK; + return err ? SR_ERR_INTERNAL : SR_ERR_OK; } static int sr_generic_cb(sr_session_ctx_t *session, uint32_t, const char *, diff --git a/src/yangerd/internal/ipbatch/ipbatch.go b/src/yangerd/internal/ipbatch/ipbatch.go index 2e862c29..dcb06e07 100644 --- a/src/yangerd/internal/ipbatch/ipbatch.go +++ b/src/yangerd/internal/ipbatch/ipbatch.go @@ -1,9 +1,16 @@ -// Package ipbatch manages a persistent `ip -json -s -d -force -batch -` +// Package ipbatch manages a persistent `ip -json [-s] [-d] -force -batch -` // subprocess. Commands sent via Query are serialized by a mutex and // paired with the single JSON-array line the subprocess writes to -// stdout. The -s and -d global flags ensure link queries include -// statistics and details. On subprocess death the manager enters a -// dead state and attempts automatic restart with exponential backoff. +// stdout. The caller chooses global flags via functional options: +// WithStats adds -s (statistics) and WithDetails adds -d (details). +// +// IMPORTANT: When -s is present, `link show` commands produce multiple +// lines of output — breaking the one-command-one-line protocol used by +// Query. Address queries must therefore use a separate IPBatch instance +// that omits -s (use WithDetails only). +// +// On subprocess death the manager enters a dead state and attempts +// automatic restart with exponential backoff. package ipbatch import ( @@ -33,6 +40,15 @@ const ( reconnectFactor = 2.0 ) +// Option configures an IPBatch instance. +type Option func(*IPBatch) + +// WithStats adds -s (statistics) to the ip command. +func WithStats() Option { return func(b *IPBatch) { b.stats = true } } + +// WithDetails adds -d (details) to the ip command. +func WithDetails() Option { return func(b *IPBatch) { b.details = true } } + // IPBatch wraps a persistent `ip -json -force -batch -` subprocess. type IPBatch struct { cmd *exec.Cmd @@ -44,17 +60,23 @@ type IPBatch struct { log *slog.Logger ctx context.Context cancel context.CancelFunc + + stats bool + details bool } // New spawns the ip batch subprocess. The returned IPBatch is ready // for Query calls. A background goroutine drains stderr. -func New(ctx context.Context, log *slog.Logger) (*IPBatch, error) { +func New(ctx context.Context, log *slog.Logger, opts ...Option) (*IPBatch, error) { ctx, cancel := context.WithCancel(ctx) b := &IPBatch{ log: log, ctx: ctx, cancel: cancel, } + for _, o := range opts { + o(b) + } if err := b.start(); err != nil { cancel() return nil, err @@ -64,7 +86,16 @@ func New(ctx context.Context, log *slog.Logger) (*IPBatch, error) { } func (b *IPBatch) start() error { - cmd := exec.CommandContext(b.ctx, "ip", "-json", "-s", "-d", "-force", "-batch", "-") + args := []string{"-json"} + if b.stats { + args = append(args, "-s") + } + if b.details { + args = append(args, "-d") + } + args = append(args, "-force", "-batch", "-") + + cmd := exec.CommandContext(b.ctx, "ip", args...) stdin, err := cmd.StdinPipe() if err != nil { return fmt.Errorf("stdin pipe: %w", err) @@ -119,6 +150,9 @@ func (b *IPBatch) Query(command string) (json.RawMessage, error) { } raw := make([]byte, len(b.stdout.Bytes())) copy(raw, b.stdout.Bytes()) + + b.log.Debug("ipbatch query", "cmd", command, "respLen", len(raw)) + return json.RawMessage(raw), nil } diff --git a/src/yangerd/internal/monitor/monitor.go b/src/yangerd/internal/monitor/monitor.go index 43c10bdb..e8738831 100644 --- a/src/yangerd/internal/monitor/monitor.go +++ b/src/yangerd/internal/monitor/monitor.go @@ -29,18 +29,22 @@ const treeKey = "ietf-interfaces:interfaces" // with ethernet/wifi/bridge data before being stored as a single // complete YANG document. type NLMonitor struct { - batch *ipbatch.IPBatch + linkBatch *ipbatch.IPBatch + addrBatch *ipbatch.IPBatch brBatch *bridgebatch.BridgeBatch tree *tree.Tree ethRefresh func(string) log *slog.Logger fc iface.FileChecker + // initDone is closed after the first initialDump completes. + initDone chan struct{} + // staging holds raw ip-json data used as input to iface.Transform(). // Protected by mu. mu sync.Mutex links json.RawMessage // ip -json -s -d link show (includes stats+details) - addrs json.RawMessage // ip -json addr show + addrs json.RawMessage // ip -json -d addr show (details only, no stats) fdb map[string]json.RawMessage mdb map[string]json.RawMessage ethernet map[string]json.RawMessage // ifname → ethtool JSON @@ -50,13 +54,17 @@ type NLMonitor struct { } // New creates a netlink monitor backed by ip/bridge batch query workers. -func New(batch *ipbatch.IPBatch, brBatch *bridgebatch.BridgeBatch, t *tree.Tree, fc iface.FileChecker, log *slog.Logger) *NLMonitor { +// linkBatch should include -s -d flags; addrBatch should include -d only +// (no -s, which causes multi-line output for link commands). +func New(linkBatch, addrBatch *ipbatch.IPBatch, brBatch *bridgebatch.BridgeBatch, t *tree.Tree, fc iface.FileChecker, log *slog.Logger) *NLMonitor { return &NLMonitor{ - batch: batch, + linkBatch: linkBatch, + addrBatch: addrBatch, brBatch: brBatch, tree: t, fc: fc, log: log, + initDone: make(chan struct{}), fdb: make(map[string]json.RawMessage), mdb: make(map[string]json.RawMessage), ethernet: make(map[string]json.RawMessage), @@ -71,6 +79,11 @@ func (m *NLMonitor) SetEthRefresh(fn func(string)) { m.ethRefresh = fn } +// WaitReady returns a channel that is closed after initialDump completes. +func (m *NLMonitor) WaitReady() <-chan struct{} { + return m.initDone +} + // SetEthernetData updates the staged ethernet data for an interface // and triggers a full rebuild of the YANG document. func (m *NLMonitor) SetEthernetData(ifname string, data json.RawMessage) { @@ -133,6 +146,7 @@ func (m *NLMonitor) Run(ctx context.Context) error { if err := m.initialDump(); err != nil { m.log.Error("initial dump failed", "err", err) } + close(m.initDone) for { select { @@ -166,15 +180,18 @@ func (m *NLMonitor) Run(ctx context.Context) error { } func (m *NLMonitor) initialDump() error { - linkRaw, err := m.queryIP("link show") + linkRaw, err := m.queryLink("link show") if err != nil { return err } - addrRaw, err := m.queryIP("addr show") + addrRaw, err := m.queryAddr("addr show") if err != nil { return err } + m.log.Debug("initialDump", "linkBytes", len(linkRaw), "addrBytes", len(addrRaw)) + m.validateAddrData("initialDump", addrRaw) + m.mu.Lock() m.links = linkRaw m.addrs = addrRaw @@ -209,8 +226,15 @@ func (m *NLMonitor) handleAddrUpdate(update netlink.AddrUpdate) { return } - raw, err := m.queryIP("addr show dev " + ifname) + m.log.Debug("handleAddrUpdate", "ifname", ifname) + raw, err := m.queryAddr("addr show dev " + ifname) if err != nil { + m.log.Error("handleAddrUpdate queryAddr failed", "ifname", ifname, "err", err) + return + } + + if !m.validateAddrData("handleAddrUpdate/"+ifname, raw) { + m.log.Error("handleAddrUpdate: REFUSING to store invalid addr data", "ifname", ifname) return } @@ -260,15 +284,19 @@ func (m *NLMonitor) handleMDBUpdate() { } func (m *NLMonitor) refreshInterface(name string) { - linkRaw, err := m.queryIP("link show dev " + name) + linkRaw, err := m.queryLink("link show dev " + name) if err != nil { return } - addrRaw, err := m.queryIP("addr show dev " + name) + addrRaw, err := m.queryAddr("addr show dev " + name) if err != nil { addrRaw = nil } + if addrRaw != nil && !m.validateAddrData("refreshInterface/"+name, addrRaw) { + m.log.Error("refreshInterface: REFUSING to store invalid addr data", "ifname", name) + addrRaw = nil + } m.mu.Lock() m.updateOperStatus(name, linkRaw) @@ -406,14 +434,27 @@ func (m *NLMonitor) updateOperStatus(ifname string, raw json.RawMessage) { } } -func (m *NLMonitor) queryIP(command string) (json.RawMessage, error) { - raw, err := m.batch.Query(command) +func (m *NLMonitor) queryLink(command string) (json.RawMessage, error) { + raw, err := m.linkBatch.Query(command) if err != nil { if errors.Is(err, ipbatch.ErrBatchDead) { - m.log.Warn("ip batch dead", "command", command, "err", err) + m.log.Warn("link batch dead", "command", command, "err", err) return nil, err } - m.log.Error("ip batch query failed", "command", command, "err", err) + m.log.Error("link batch query failed", "command", command, "err", err) + return nil, err + } + return raw, nil +} + +func (m *NLMonitor) queryAddr(command string) (json.RawMessage, error) { + raw, err := m.addrBatch.Query(command) + if err != nil { + if errors.Is(err, ipbatch.ErrBatchDead) { + m.log.Warn("addr batch dead", "command", command, "err", err) + return nil, err + } + m.log.Error("addr batch query failed", "command", command, "err", err) return nil, err } return raw, nil @@ -527,6 +568,54 @@ func bridgeNameFromNeigh(update netlink.NeighUpdate) (string, bool) { return "", false } +// validateAddrData checks whether a JSON response from "addr show" contains +// addr_info entries. "ip -json addr show" always includes an "addr_info" +// array for every interface object; its absence means we got link-format +// data instead. Returns true if the data looks valid (has addr_info). +func (m *NLMonitor) validateAddrData(caller string, raw json.RawMessage) bool { + if len(raw) == 0 { + m.log.Error("addr data is EMPTY", "caller", caller) + return false + } + + var rows []map[string]json.RawMessage + if err := json.Unmarshal(raw, &rows); err != nil { + m.log.Error("addr data unmarshal failed", "caller", caller, "err", err, "raw", string(raw)) + return false + } + + if len(rows) == 0 { + // Empty array is valid — interface exists but has no addresses. + return true + } + + for _, row := range rows { + ifnRaw, _ := row["ifname"] + var ifn string + json.Unmarshal(ifnRaw, &ifn) + + if _, ok := row["addr_info"]; !ok { + m.log.Error("addr data MISSING addr_info — got link-format data", + "caller", caller, + "ifname", ifn, + "keys", mapKeys(row), + "raw", string(raw), + ) + return false + } + } + return true +} + +// mapKeys returns the JSON object keys from a map for diagnostic logging. +func mapKeys(m map[string]json.RawMessage) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + return keys +} + func extractOperStatus(raw json.RawMessage) (string, bool) { var rows []map[string]json.RawMessage if err := json.Unmarshal(raw, &rows); err != nil || len(rows) == 0 { diff --git a/src/yangerd/internal/zapi/zapi.go b/src/yangerd/internal/zapi/zapi.go new file mode 100644 index 00000000..a1993fba --- /dev/null +++ b/src/yangerd/internal/zapi/zapi.go @@ -0,0 +1,540 @@ +// Package zapi implements a minimal ZAPI v6 client for FRR 10.5. +// +// It speaks only the subset of the Zebra wire protocol needed by +// yangerd: Hello, RouterIDAdd, RedistributeAdd, and decoding of +// RedistributeRouteAdd/Del messages. +package zapi + +import ( + "encoding/binary" + "fmt" + "io" + "net" + "syscall" +) + +// Wire constants for ZAPI v6. +const ( + HeaderSize = 10 + HeaderMarker = 0xFE + HeaderVersion = 6 + + DefaultVrf uint32 = 0 +) + +// Command IDs for FRR 10.5 ZAPI v6 (from lib/zclient.h). +type Command uint16 + +const ( + CmdInterfaceAdd Command = 0 + CmdInterfaceDelete Command = 1 + CmdInterfaceAddrAdd Command = 2 + CmdInterfaceAddrDelete Command = 3 + CmdInterfaceUp Command = 4 + CmdInterfaceDown Command = 5 + CmdInterfaceSetMaster Command = 6 + CmdInterfaceSetARP Command = 7 // new in FRR 10.x + CmdInterfaceSetProtodown Command = 8 + CmdRouteAdd Command = 9 + CmdRouteDelete Command = 10 + CmdRouteNotifyOwner Command = 11 + CmdRedistributeAdd Command = 12 + CmdRedistributeDelete Command = 13 + CmdRedistDefaultAdd Command = 14 + CmdRedistDefaultDelete Command = 15 + CmdRouterIDAdd Command = 16 + CmdRouterIDDelete Command = 17 + CmdRouterIDUpdate Command = 18 + CmdHello Command = 19 + CmdCapabilities Command = 20 + CmdNexthopRegister Command = 21 + CmdNexthopUnregister Command = 22 + CmdNexthopUpdate Command = 23 + + CmdRedistRouteAdd Command = 31 + CmdRedistRouteDel Command = 32 +) + +// RouteType identifies the source protocol of a route. +type RouteType uint8 + +const ( + RouteSystem RouteType = 0 + RouteKernel RouteType = 1 + RouteConnect RouteType = 2 + RouteStatic RouteType = 3 + RouteRIP RouteType = 4 + RouteRIPNG RouteType = 5 + RouteOSPF RouteType = 6 + RouteOSPF6 RouteType = 7 + RouteISIS RouteType = 8 + RouteBGP RouteType = 9 +) + +// AFI values. +const ( + AFIIPv4 uint8 = 1 + AFIIPv6 uint8 = 2 +) + +// Message flags (from struct zapi_route.message). +type MsgFlag uint32 + +const ( + MsgNexthop MsgFlag = 0x01 + MsgDistance MsgFlag = 0x02 + MsgMetric MsgFlag = 0x04 + MsgTag MsgFlag = 0x08 + MsgMTU MsgFlag = 0x10 + MsgSrcPfx MsgFlag = 0x20 + MsgBackupNH MsgFlag = 0x40 + MsgNHG MsgFlag = 0x80 + MsgTableID MsgFlag = 0x100 + MsgSRTE MsgFlag = 0x200 + MsgOpaque MsgFlag = 0x400 +) + +// Nexthop type values (from lib/nexthop.h). +type NHType uint8 + +const ( + NHIFIndex NHType = 1 + NHIPv4 NHType = 2 + NHIPv4IFIndex NHType = 3 + NHIPv6 NHType = 4 + NHIPv6IFIndex NHType = 5 + NHBlackhole NHType = 6 +) + +// Nexthop flags (from lib/zclient.h: ZAPI_NEXTHOP_FLAG_*). +const ( + nhFlagLabel uint8 = 0x02 + nhFlagWeight uint8 = 0x04 + nhFlagHasBackup uint8 = 0x08 + nhFlagSeg6 uint8 = 0x10 + nhFlagSeg6Local uint8 = 0x20 +) + +// Header is a ZAPI v6 message header. +type Header struct { + Length uint16 + Marker uint8 + Version uint8 + VrfID uint32 + Command Command +} + +// Nexthop holds one entry from the nexthop list in a route message. +type Nexthop struct { + Type NHType + Gate net.IP + Ifindex uint32 +} + +// Route is the decoded content of a RedistributeRouteAdd/Del message. +type Route struct { + Type RouteType + Flags uint32 + Message MsgFlag + Prefix net.IPNet + Nexthops []Nexthop + Distance uint8 + Metric uint32 + Tag uint32 + MTU uint32 +} + +// Message is a decoded ZAPI message received from zebra. +type Message struct { + Header Header + Route *Route // non-nil only for route messages +} + +// EncodeHeader serializes a ZAPI v6 header. +func EncodeHeader(length uint16, vrfID uint32, cmd Command) []byte { + buf := make([]byte, HeaderSize) + binary.BigEndian.PutUint16(buf[0:2], length) + buf[2] = HeaderMarker + buf[3] = HeaderVersion + binary.LittleEndian.PutUint32(buf[4:8], vrfID) + binary.BigEndian.PutUint16(buf[8:10], uint16(cmd)) + return buf +} + +// DecodeHeader parses a ZAPI v6 header from exactly HeaderSize bytes. +func DecodeHeader(data []byte) (Header, error) { + if len(data) < HeaderSize { + return Header{}, fmt.Errorf("header too short: %d bytes", len(data)) + } + h := Header{ + Length: binary.BigEndian.Uint16(data[0:2]), + Marker: data[2], + Version: data[3], + VrfID: binary.LittleEndian.Uint32(data[4:8]), + Command: Command(binary.BigEndian.Uint16(data[8:10])), + } + if h.Marker != HeaderMarker { + return Header{}, fmt.Errorf("bad marker: 0x%02x", h.Marker) + } + if h.Version != HeaderVersion { + return Header{}, fmt.Errorf("unsupported version: %d", h.Version) + } + return h, nil +} + +// EncodeHello builds a Hello message body. +// Fields: redistDefault(1), instance(2), sessionID(4), receiveNotify(1), synchronous(1), flags(4), ... +// We send zeros for everything (redistDefault=0 means "no default route type"). +func EncodeHello() []byte { + body := make([]byte, 12) // redistDefault + instance + sessionID + receiveNotify + synchronous + flags + return body +} + +// EncodeRouterIDAdd builds a RouterIDAdd message body. +// Body is just the AFI value (1 byte). +func EncodeRouterIDAdd(afi uint8) []byte { + return []byte{afi} +} + +// EncodeRedistributeAdd builds a RedistributeAdd body. +// Body: afi(1), routeType(1), instance(2). +func EncodeRedistributeAdd(afi uint8, rt RouteType) []byte { + buf := make([]byte, 4) + buf[0] = afi + buf[1] = uint8(rt) + // instance = 0 (already zeroed) + return buf +} + +// BuildMessage constructs a complete wire message from command and body. +func BuildMessage(cmd Command, vrfID uint32, body []byte) []byte { + length := uint16(HeaderSize + len(body)) + hdr := EncodeHeader(length, vrfID, cmd) + return append(hdr, body...) +} + +// ReadMessage reads one complete ZAPI message from the connection. +// It returns the header and the raw body bytes. +func ReadMessage(r io.Reader) (Header, []byte, error) { + hdrBuf := make([]byte, HeaderSize) + if _, err := io.ReadFull(r, hdrBuf); err != nil { + return Header{}, nil, fmt.Errorf("read header: %w", err) + } + + hdr, err := DecodeHeader(hdrBuf) + if err != nil { + return Header{}, nil, err + } + + bodyLen := int(hdr.Length) - HeaderSize + if bodyLen < 0 { + return Header{}, nil, fmt.Errorf("invalid message length: %d", hdr.Length) + } + if bodyLen == 0 { + return hdr, nil, nil + } + + body := make([]byte, bodyLen) + if _, err := io.ReadFull(r, body); err != nil { + return Header{}, nil, fmt.Errorf("read body: %w", err) + } + + return hdr, body, nil +} + +// DecodeRoute parses an IPRouteBody from the given body bytes. +// This handles the FRR 10.5 / ZAPI v6 (frr >= 7.5) format: +// +// type(1) instance(2) flags(4) message(4) safi(1) family(1) prefixlen(1) prefix(var) ... +func DecodeRoute(body []byte) (*Route, error) { + if len(body) < 10 { + return nil, fmt.Errorf("route body too short: %d bytes", len(body)) + } + + r := &Route{} + pos := 0 + + // type(1) + r.Type = RouteType(body[pos]) + pos++ + + // instance(2) + pos += 2 + + // flags(4) + r.Flags = binary.BigEndian.Uint32(body[pos : pos+4]) + pos += 4 + + // message(4) — FRR >= 7.5 uses 4-byte message field + r.Message = MsgFlag(binary.BigEndian.Uint32(body[pos : pos+4])) + pos += 4 + + // safi(1) + pos++ + + // family(1) + if pos >= len(body) { + return nil, fmt.Errorf("truncated at family") + } + family := body[pos] + pos++ + + addrLen, err := addressByteLen(family) + if err != nil { + return nil, err + } + + // prefixlen(1) + if pos >= len(body) { + return nil, fmt.Errorf("truncated at prefixlen") + } + prefixLen := body[pos] + pos++ + + // prefix (ceil(prefixLen/8) bytes) + byteLen := int((prefixLen + 7) / 8) + if pos+byteLen > len(body) { + return nil, fmt.Errorf("truncated at prefix data") + } + ipBuf := make([]byte, addrLen) + copy(ipBuf, body[pos:pos+byteLen]) + pos += byteLen + + var ip net.IP + if family == syscall.AF_INET { + ip = net.IP(ipBuf).To4() + } else { + ip = net.IP(ipBuf).To16() + } + mask := net.CIDRMask(int(prefixLen), addrLen*8) + r.Prefix = net.IPNet{IP: ip, Mask: mask} + + // source prefix (if MsgSrcPfx set) + if r.Message&MsgSrcPfx != 0 { + if pos >= len(body) { + return nil, fmt.Errorf("truncated at src prefix") + } + srcPfxLen := body[pos] + pos++ + srcByteLen := int((srcPfxLen + 7) / 8) + if pos+srcByteLen > len(body) { + return nil, fmt.Errorf("truncated at src prefix data") + } + pos += srcByteLen + } + + // NHG (if MsgNHG set, frr >= 8) + if r.Message&MsgNHG != 0 { + if pos+4 > len(body) { + return nil, fmt.Errorf("truncated at nhg") + } + pos += 4 // skip nhgid + } + + // Nexthops + if r.Message&MsgNexthop != 0 { + if pos+2 > len(body) { + return nil, fmt.Errorf("truncated at nexthop count") + } + numNH := binary.BigEndian.Uint16(body[pos : pos+2]) + pos += 2 + r.Nexthops = make([]Nexthop, 0, numNH) + + for i := uint16(0); i < numNH; i++ { + nh, n, err := decodeNexthop(body[pos:], family) + if err != nil { + return nil, fmt.Errorf("nexthop %d: %w", i, err) + } + r.Nexthops = append(r.Nexthops, nh) + pos += n + } + } + + // Backup nexthops (skip) + if r.Message&MsgBackupNH != 0 { + if pos+2 > len(body) { + return nil, fmt.Errorf("truncated at backup nexthop count") + } + numBackup := binary.BigEndian.Uint16(body[pos : pos+2]) + pos += 2 + for i := uint16(0); i < numBackup; i++ { + _, n, err := decodeNexthop(body[pos:], family) + if err != nil { + return nil, fmt.Errorf("backup nexthop %d: %w", i, err) + } + pos += n + } + } + + // Distance + if r.Message&MsgDistance != 0 { + if pos >= len(body) { + return nil, fmt.Errorf("truncated at distance") + } + r.Distance = body[pos] + pos++ + } + + // Metric + if r.Message&MsgMetric != 0 { + if pos+4 > len(body) { + return nil, fmt.Errorf("truncated at metric") + } + r.Metric = binary.BigEndian.Uint32(body[pos : pos+4]) + pos += 4 + } + + // Tag + if r.Message&MsgTag != 0 { + if pos+4 > len(body) { + return nil, fmt.Errorf("truncated at tag") + } + r.Tag = binary.BigEndian.Uint32(body[pos : pos+4]) + pos += 4 + } + + // MTU + if r.Message&MsgMTU != 0 { + if pos+4 > len(body) { + return nil, fmt.Errorf("truncated at mtu") + } + r.MTU = binary.BigEndian.Uint32(body[pos : pos+4]) + pos += 4 + } + + // Remaining fields (tableID, SRTE, opaque) are not needed. + return r, nil +} + +// decodeNexthop decodes one nexthop from the body. +// FRR >= 7.3 / ZAPI v6 format: vrfID(4) type(1) flags(1) [gate] [ifindex] ... +func decodeNexthop(data []byte, family uint8) (Nexthop, int, error) { + nh := Nexthop{} + pos := 0 + + // vrfID(4) + if pos+4 > len(data) { + return nh, 0, fmt.Errorf("truncated at vrf_id") + } + pos += 4 // skip vrfID + + // type(1) + if pos >= len(data) { + return nh, 0, fmt.Errorf("truncated at nh type") + } + nh.Type = NHType(data[pos]) + pos++ + + // flags(1) — FRR >= 7.3 + if pos >= len(data) { + return nh, 0, fmt.Errorf("truncated at nh flags") + } + flags := data[pos] + pos++ + + // For FRR >= 7.3, IPv4 and IPv6 are treated as IPv4IFIndex and IPv6IFIndex + // respectively (nexthopProcessIPToIPIFindex). We do the same mapping. + nhType := nh.Type + switch nhType { + case NHIPv4: + nhType = NHIPv4IFIndex + case NHIPv6: + nhType = NHIPv6IFIndex + } + + // Decode gate address + switch nhType { + case NHIPv4IFIndex: + if pos+4 > len(data) { + return nh, 0, fmt.Errorf("truncated at ipv4 gate") + } + nh.Gate = net.IP(data[pos : pos+4]).To4() + pos += 4 + case NHIPv6IFIndex: + if pos+16 > len(data) { + return nh, 0, fmt.Errorf("truncated at ipv6 gate") + } + nh.Gate = net.IP(data[pos : pos+16]).To16() + pos += 16 + } + + // Decode ifindex + switch nhType { + case NHIFIndex, NHIPv4IFIndex, NHIPv6IFIndex: + if pos+4 > len(data) { + return nh, 0, fmt.Errorf("truncated at ifindex") + } + nh.Ifindex = binary.BigEndian.Uint32(data[pos : pos+4]) + pos += 4 + } + + // Blackhole type + if nhType == NHBlackhole { + if pos >= len(data) { + return nh, 0, fmt.Errorf("truncated at blackhole type") + } + pos++ // skip blackhole type byte + } + + // Labels (if flag set) + if flags&nhFlagLabel != 0 { + if pos >= len(data) { + return nh, 0, fmt.Errorf("truncated at label count") + } + labelNum := int(data[pos]) + pos++ + if labelNum > 16 { + labelNum = 16 + } + pos += labelNum * 4 // skip label data + } + + // Weight (if flag set) + if flags&nhFlagWeight != 0 { + if pos+4 > len(data) { + return nh, 0, fmt.Errorf("truncated at weight") + } + pos += 4 + } + + // SRTE color — present when message has MsgSRTE flag. + // We can't check message flags here; instead rely on the frr >=7.5 + // behavior: SRTE color is only present if the SRTE message flag was + // set on the route. The caller handles this by not passing SRTE routes + // to us (we only subscribe to simple route types). For safety, we skip + // nothing here — SRTE is handled at the route level if needed. + + // Backup nexthop indices (if flag set) + if flags&nhFlagHasBackup != 0 { + if pos >= len(data) { + return nh, 0, fmt.Errorf("truncated at backup count") + } + backupNum := int(data[pos]) + pos++ + pos += backupNum // skip backup indices + } + + // SEG6 (if flag set) + if flags&nhFlagSeg6 != 0 { + // seg6local_action(4) + seg6local_context(4+16+4 = 24) + pos += 28 + } + + // SEG6 local (if flag set) + if flags&nhFlagSeg6Local != 0 { + pos += 16 // struct in6_addr + } + + return nh, pos, nil +} + +func addressByteLen(family uint8) (int, error) { + switch family { + case syscall.AF_INET: + return 4, nil + case syscall.AF_INET6: + return 16, nil + default: + return 0, fmt.Errorf("unsupported address family: %d", family) + } +} diff --git a/src/yangerd/internal/zapi/zapi_test.go b/src/yangerd/internal/zapi/zapi_test.go new file mode 100644 index 00000000..165636b8 --- /dev/null +++ b/src/yangerd/internal/zapi/zapi_test.go @@ -0,0 +1,350 @@ +package zapi + +import ( + "bytes" + "encoding/binary" + "net" + "syscall" + "testing" +) + +func TestEncodeDecodeHeader(t *testing.T) { + raw := EncodeHeader(42, 0, CmdHello) + hdr, err := DecodeHeader(raw) + if err != nil { + t.Fatal(err) + } + if hdr.Length != 42 { + t.Errorf("Length = %d, want 42", hdr.Length) + } + if hdr.Command != CmdHello { + t.Errorf("Command = %d, want %d", hdr.Command, CmdHello) + } + if hdr.VrfID != 0 { + t.Errorf("VrfID = %d, want 0", hdr.VrfID) + } +} + +func TestDecodeHeaderBadMarker(t *testing.T) { + raw := EncodeHeader(10, 0, CmdHello) + raw[2] = 0x00 + _, err := DecodeHeader(raw) + if err == nil { + t.Fatal("expected error for bad marker") + } +} + +func TestDecodeHeaderBadVersion(t *testing.T) { + raw := EncodeHeader(10, 0, CmdHello) + raw[3] = 5 + _, err := DecodeHeader(raw) + if err == nil { + t.Fatal("expected error for bad version") + } +} + +func TestBuildMessage(t *testing.T) { + body := EncodeHello() + msg := BuildMessage(CmdHello, DefaultVrf, body) + if len(msg) != HeaderSize+len(body) { + t.Errorf("message len = %d, want %d", len(msg), HeaderSize+len(body)) + } + hdr, err := DecodeHeader(msg[:HeaderSize]) + if err != nil { + t.Fatal(err) + } + if hdr.Command != CmdHello { + t.Errorf("Command = %d, want %d", hdr.Command, CmdHello) + } + if int(hdr.Length) != len(msg) { + t.Errorf("Length = %d, want %d", hdr.Length, len(msg)) + } +} + +func TestEncodeRedistributeAdd(t *testing.T) { + body := EncodeRedistributeAdd(AFIIPv4, RouteStatic) + if len(body) != 4 { + t.Fatalf("body len = %d, want 4", len(body)) + } + if body[0] != AFIIPv4 { + t.Errorf("afi = %d, want %d", body[0], AFIIPv4) + } + if body[1] != uint8(RouteStatic) { + t.Errorf("routeType = %d, want %d", body[1], RouteStatic) + } +} + +func TestReadMessage(t *testing.T) { + body := []byte{0x01, 0x02, 0x03} + msg := BuildMessage(CmdRouterIDUpdate, DefaultVrf, body) + r := bytes.NewReader(msg) + + hdr, gotBody, err := ReadMessage(r) + if err != nil { + t.Fatal(err) + } + if hdr.Command != CmdRouterIDUpdate { + t.Errorf("Command = %d, want %d", hdr.Command, CmdRouterIDUpdate) + } + if !bytes.Equal(gotBody, body) { + t.Errorf("body = %v, want %v", gotBody, body) + } +} + +func TestDecodeRouteIPv4(t *testing.T) { + body := buildRouteBody(t, syscall.AF_INET, + net.ParseIP("10.0.0.0").To4(), 24, + uint32(MsgNexthop|MsgDistance|MsgMetric), + RouteConnect, 0, + []testNexthop{ + {nhType: NHIPv4IFIndex, gate: net.ParseIP("10.0.0.1").To4(), ifindex: 2}, + }, + 10, 100, + ) + + route, err := DecodeRoute(body) + if err != nil { + t.Fatal(err) + } + + if route.Type != RouteConnect { + t.Errorf("Type = %d, want %d", route.Type, RouteConnect) + } + want := "10.0.0.0/24" + if route.Prefix.String() != want { + t.Errorf("Prefix = %s, want %s", route.Prefix.String(), want) + } + if len(route.Nexthops) != 1 { + t.Fatalf("Nexthops = %d, want 1", len(route.Nexthops)) + } + if route.Nexthops[0].Gate.String() != "10.0.0.1" { + t.Errorf("Gate = %s, want 10.0.0.1", route.Nexthops[0].Gate) + } + if route.Nexthops[0].Ifindex != 2 { + t.Errorf("Ifindex = %d, want 2", route.Nexthops[0].Ifindex) + } + if route.Distance != 10 { + t.Errorf("Distance = %d, want 10", route.Distance) + } + if route.Metric != 100 { + t.Errorf("Metric = %d, want 100", route.Metric) + } +} + +func TestDecodeRouteIPv6(t *testing.T) { + body := buildRouteBody(t, syscall.AF_INET6, + net.ParseIP("2001:db8::").To16(), 48, + uint32(MsgNexthop|MsgMetric), + RouteOSPF, 0, + []testNexthop{ + {nhType: NHIPv6IFIndex, gate: net.ParseIP("2001:db8::1").To16(), ifindex: 3}, + }, + 0, 200, + ) + + route, err := DecodeRoute(body) + if err != nil { + t.Fatal(err) + } + + want := "2001:db8::/48" + if route.Prefix.String() != want { + t.Errorf("Prefix = %s, want %s", route.Prefix.String(), want) + } + if route.Metric != 200 { + t.Errorf("Metric = %d, want 200", route.Metric) + } + if len(route.Nexthops) != 1 { + t.Fatalf("Nexthops = %d, want 1", len(route.Nexthops)) + } + if route.Nexthops[0].Gate.String() != "2001:db8::1" { + t.Errorf("Gate = %s, want 2001:db8::1", route.Nexthops[0].Gate) + } +} + +func TestDecodeRouteNoNexthops(t *testing.T) { + body := buildRouteBody(t, syscall.AF_INET, + net.ParseIP("192.168.1.0").To4(), 24, + uint32(MsgMetric), + RouteKernel, 0, + nil, + 0, 50, + ) + + route, err := DecodeRoute(body) + if err != nil { + t.Fatal(err) + } + if len(route.Nexthops) != 0 { + t.Errorf("Nexthops = %d, want 0", len(route.Nexthops)) + } + if route.Metric != 50 { + t.Errorf("Metric = %d, want 50", route.Metric) + } +} + +func TestDecodeRouteMultipleNexthops(t *testing.T) { + body := buildRouteBody(t, syscall.AF_INET, + net.ParseIP("10.0.0.0").To4(), 8, + uint32(MsgNexthop), + RouteStatic, 0, + []testNexthop{ + {nhType: NHIPv4IFIndex, gate: net.ParseIP("10.0.0.1").To4(), ifindex: 1}, + {nhType: NHIPv4IFIndex, gate: net.ParseIP("10.0.0.2").To4(), ifindex: 2}, + }, + 0, 0, + ) + + route, err := DecodeRoute(body) + if err != nil { + t.Fatal(err) + } + if len(route.Nexthops) != 2 { + t.Fatalf("Nexthops = %d, want 2", len(route.Nexthops)) + } + if route.Nexthops[1].Gate.String() != "10.0.0.2" { + t.Errorf("Gate[1] = %s, want 10.0.0.2", route.Nexthops[1].Gate) + } +} + +func TestDecodeRouteDefaultRoute(t *testing.T) { + body := buildRouteBody(t, syscall.AF_INET, + net.ParseIP("0.0.0.0").To4(), 0, + uint32(MsgNexthop|MsgMetric), + RouteStatic, 0, + []testNexthop{ + {nhType: NHIPv4IFIndex, gate: net.ParseIP("192.168.1.1").To4(), ifindex: 5}, + }, + 0, 0, + ) + + route, err := DecodeRoute(body) + if err != nil { + t.Fatal(err) + } + if route.Prefix.String() != "0.0.0.0/0" { + t.Errorf("Prefix = %s, want 0.0.0.0/0", route.Prefix.String()) + } +} + +func TestDecodeRouteIFIndexOnly(t *testing.T) { + body := buildRouteBody(t, syscall.AF_INET, + net.ParseIP("10.0.0.0").To4(), 24, + uint32(MsgNexthop), + RouteConnect, 0, + []testNexthop{ + {nhType: NHIFIndex, ifindex: 7}, + }, + 0, 0, + ) + + route, err := DecodeRoute(body) + if err != nil { + t.Fatal(err) + } + if len(route.Nexthops) != 1 { + t.Fatalf("Nexthops = %d, want 1", len(route.Nexthops)) + } + if route.Nexthops[0].Ifindex != 7 { + t.Errorf("Ifindex = %d, want 7", route.Nexthops[0].Ifindex) + } +} + +type testNexthop struct { + nhType NHType + gate net.IP + ifindex uint32 +} + +func buildRouteBody(t *testing.T, family uint8, prefix net.IP, prefixLen uint8, msgFlags uint32, routeType RouteType, flags uint32, nexthops []testNexthop, distance uint8, metric uint32) []byte { + t.Helper() + var buf []byte + + buf = append(buf, uint8(routeType)) + + // instance(2) + buf = append(buf, 0, 0) + + // flags(4) + tmp := make([]byte, 4) + binary.BigEndian.PutUint32(tmp, flags) + buf = append(buf, tmp...) + + // message(4) + binary.BigEndian.PutUint32(tmp, msgFlags) + buf = append(buf, tmp...) + + // safi(1) = unicast + buf = append(buf, 1) + + // family(1) + buf = append(buf, family) + + // prefixlen(1) + buf = append(buf, prefixLen) + + // prefix bytes + byteLen := int((prefixLen + 7) / 8) + buf = append(buf, prefix[:byteLen]...) + + // nexthops + if MsgFlag(msgFlags)&MsgNexthop != 0 { + nhCount := make([]byte, 2) + binary.BigEndian.PutUint16(nhCount, uint16(len(nexthops))) + buf = append(buf, nhCount...) + + for _, nh := range nexthops { + buf = append(buf, encodeTestNexthop(nh)...) + } + } + + // distance + if MsgFlag(msgFlags)&MsgDistance != 0 { + buf = append(buf, distance) + } + + // metric + if MsgFlag(msgFlags)&MsgMetric != 0 { + binary.BigEndian.PutUint32(tmp, metric) + buf = append(buf, tmp...) + } + + return buf +} + +func encodeTestNexthop(nh testNexthop) []byte { + var buf []byte + + // vrfID(4) = 0 + buf = append(buf, 0, 0, 0, 0) + + // type(1) + buf = append(buf, uint8(nh.nhType)) + + // flags(1) = 0 + buf = append(buf, 0) + + nhType := nh.nhType + switch nhType { + case NHIPv4: + nhType = NHIPv4IFIndex + case NHIPv6: + nhType = NHIPv6IFIndex + } + + switch nhType { + case NHIPv4IFIndex: + buf = append(buf, nh.gate.To4()...) + case NHIPv6IFIndex: + buf = append(buf, nh.gate.To16()...) + } + + switch nhType { + case NHIFIndex, NHIPv4IFIndex, NHIPv6IFIndex: + tmp := make([]byte, 4) + binary.BigEndian.PutUint32(tmp, nh.ifindex) + buf = append(buf, tmp...) + } + + return buf +} diff --git a/src/yangerd/internal/zapiwatcher/zapiwatcher.go b/src/yangerd/internal/zapiwatcher/zapiwatcher.go index 9850a6d7..fc10e891 100644 --- a/src/yangerd/internal/zapiwatcher/zapiwatcher.go +++ b/src/yangerd/internal/zapiwatcher/zapiwatcher.go @@ -1,11 +1,8 @@ -// Package zapiwatcher subscribes to FRR zebra zserv route redistribution -// events and mirrors routing data into the operational YANG tree. package zapiwatcher import ( "context" "encoding/json" - "errors" "fmt" "log/slog" "math" @@ -16,45 +13,33 @@ import ( "time" "github.com/kernelkit/infix/src/yangerd/internal/tree" - "github.com/osrg/gobgp/v3/pkg/log" - "github.com/osrg/gobgp/v3/pkg/zebra" + "github.com/kernelkit/infix/src/yangerd/internal/zapi" ) const ( zapiSocketPath = "/var/run/frr/zserv.api" - zapiVersion = 6 - zapiSoftware = "frr10.5" reconnectInitial = 100 * time.Millisecond reconnectMax = 30 * time.Second reconnectFactor = 2.0 ) -const ( - RouteKernel zebra.RouteType = 1 - RouteConnect zebra.RouteType = 2 - RouteStatic zebra.RouteType = 3 - RouteRIP zebra.RouteType = 4 - RouteOSPF zebra.RouteType = 6 -) - -var subscribeTypes = []zebra.RouteType{ - RouteKernel, - RouteConnect, - RouteStatic, - RouteRIP, - RouteOSPF, +var subscribeTypes = []zapi.RouteType{ + zapi.RouteKernel, + zapi.RouteConnect, + zapi.RouteStatic, + zapi.RouteRIP, + zapi.RouteOSPF, } -var routeTypeToProtocol = map[zebra.RouteType]string{ - RouteKernel: "infix-routing:kernel", - RouteConnect: "ietf-routing:direct", - RouteStatic: "ietf-routing:static", - RouteOSPF: "ietf-ospf:ospfv2", - RouteRIP: "ietf-rip:ripv2", +var routeTypeToProtocol = map[zapi.RouteType]string{ + zapi.RouteKernel: "infix-routing:kernel", + zapi.RouteConnect: "ietf-routing:direct", + zapi.RouteStatic: "ietf-routing:static", + zapi.RouteOSPF: "ietf-ospf:ospfv2", + zapi.RouteRIP: "ietf-rip:ripv2", } -// ZAPIWatcher keeps the routing subtree in sync with zebra route updates. type ZAPIWatcher struct { tree *tree.Tree log *slog.Logger @@ -64,29 +49,6 @@ type ZAPIWatcher struct { const routingTreeKey = "ietf-routing:routing" -// slogAdapter wraps slog.Logger to implement gobgp v3 log.Logger interface. -type slogAdapter struct { - l *slog.Logger -} - -func (a *slogAdapter) Panic(msg string, fields log.Fields) { a.l.Error(msg, toAttrs(fields)...) } -func (a *slogAdapter) Fatal(msg string, fields log.Fields) { a.l.Error(msg, toAttrs(fields)...) } -func (a *slogAdapter) Error(msg string, fields log.Fields) { a.l.Error(msg, toAttrs(fields)...) } -func (a *slogAdapter) Warn(msg string, fields log.Fields) { a.l.Warn(msg, toAttrs(fields)...) } -func (a *slogAdapter) Info(msg string, fields log.Fields) { a.l.Info(msg, toAttrs(fields)...) } -func (a *slogAdapter) Debug(msg string, fields log.Fields) { a.l.Debug(msg, toAttrs(fields)...) } -func (a *slogAdapter) SetLevel(level log.LogLevel) {} -func (a *slogAdapter) GetLevel() log.LogLevel { return log.LogLevel(0) } - -func toAttrs(fields log.Fields) []any { - attrs := make([]any, 0, len(fields)*2) - for k, v := range fields { - attrs = append(attrs, k, v) - } - return attrs -} - -// New creates a ZAPIWatcher bound to the provided operational tree and logger. func New(t *tree.Tree, log *slog.Logger) *ZAPIWatcher { if log == nil { log = slog.Default() @@ -94,15 +56,11 @@ func New(t *tree.Tree, log *slog.Logger) *ZAPIWatcher { return &ZAPIWatcher{tree: t, log: log, routes: make(map[string]json.RawMessage)} } -// Run starts the zebra watcher loop with exponential reconnect backoff. -// -// On successful connection it processes incoming route messages until -// disconnect or context cancellation. On disconnect, cached routes are cleared. func (w *ZAPIWatcher) Run(ctx context.Context) error { delay := reconnectInitial for { - cli, err := w.connect(ctx) + conn, err := w.connect(ctx) if err != nil { if ctx.Err() != nil { return ctx.Err() @@ -119,9 +77,10 @@ func (w *ZAPIWatcher) Run(ctx context.Context) error { } delay = reconnectInitial - w.log.Info("zapi watcher: connected", "socket", zapiSocketPath, "version", zapiVersion, "software", zapiSoftware) + w.log.Info("zapi watcher: connected", "socket", zapiSocketPath) - err = w.processMessages(ctx, cli) + err = w.processMessages(ctx, conn) + _ = conn.Close() if ctx.Err() != nil { return ctx.Err() } @@ -131,71 +90,79 @@ func (w *ZAPIWatcher) Run(ctx context.Context) error { } } -func (w *ZAPIWatcher) connect(ctx context.Context) (*zebra.Client, error) { +func (w *ZAPIWatcher) connect(ctx context.Context) (net.Conn, error) { d := net.Dialer{} conn, err := d.DialContext(ctx, "unix", zapiSocketPath) if err != nil { - return nil, fmt.Errorf("dial zserv socket: %w", err) - } - _ = conn.Close() - - software := zebra.NewSoftware(zapiVersion, zapiSoftware) - zebra.MaxSoftware = software - - cli, err := zebra.NewClient(&slogAdapter{l: w.log}, "unix", zapiSocketPath, zebra.RouteType(0), zapiVersion, software, 0) - if err != nil { - return nil, fmt.Errorf("new zapi client: %w", err) + return nil, fmt.Errorf("dial zserv: %w", err) } - cli.SendHello() - cli.SendRouterIDAdd() - for _, typ := range subscribeTypes { - cli.SendRedistribute(typ, zebra.DefaultVrf) + hello := zapi.BuildMessage(zapi.CmdHello, zapi.DefaultVrf, zapi.EncodeHello()) + if _, err := conn.Write(hello); err != nil { + conn.Close() + return nil, fmt.Errorf("send hello: %w", err) } - return cli, nil + for _, afi := range []uint8{zapi.AFIIPv4, zapi.AFIIPv6} { + msg := zapi.BuildMessage(zapi.CmdRouterIDAdd, zapi.DefaultVrf, zapi.EncodeRouterIDAdd(afi)) + if _, err := conn.Write(msg); err != nil { + conn.Close() + return nil, fmt.Errorf("send router-id-add: %w", err) + } + } + + for _, rt := range subscribeTypes { + for _, afi := range []uint8{zapi.AFIIPv4, zapi.AFIIPv6} { + msg := zapi.BuildMessage(zapi.CmdRedistributeAdd, zapi.DefaultVrf, zapi.EncodeRedistributeAdd(afi, rt)) + if _, err := conn.Write(msg); err != nil { + conn.Close() + return nil, fmt.Errorf("send redistribute-add: %w", err) + } + } + } + + return conn, nil } -func (w *ZAPIWatcher) processMessages(ctx context.Context, cli *zebra.Client) error { +func (w *ZAPIWatcher) processMessages(ctx context.Context, conn net.Conn) error { for { select { case <-ctx.Done(): return ctx.Err() - case msg, ok := <-cli.Receive(): - if !ok { - return errors.New("zapi receive channel closed") - } - if msg == nil { - continue - } - w.handleMessage(msg) + default: } + + hdr, body, err := zapi.ReadMessage(conn) + if err != nil { + return fmt.Errorf("read message: %w", err) + } + + w.handleMessage(hdr, body) } } -func (w *ZAPIWatcher) handleMessage(msg *zebra.Message) { - ipr, ok := msg.Body.(*zebra.IPRouteBody) - if !ok || ipr == nil { - return - } - - cmd := msg.Header.Command.ToCommon(zapiVersion, zebra.NewSoftware(zapiVersion, zapiSoftware)) - switch cmd { - case zebra.RedistributeRouteAdd: - w.addRoute(ipr) - case zebra.RedistributeRouteDel: - w.deleteRoute(ipr) +func (w *ZAPIWatcher) handleMessage(hdr zapi.Header, body []byte) { + switch hdr.Command { + case zapi.CmdRedistRouteAdd: + route, err := zapi.DecodeRoute(body) + if err != nil { + w.log.Warn("zapi watcher: decode route add", "err", err) + return + } + w.addRoute(route) + case zapi.CmdRedistRouteDel: + route, err := zapi.DecodeRoute(body) + if err != nil { + w.log.Warn("zapi watcher: decode route del", "err", err) + return + } + w.deleteRoute(route) } } -func (w *ZAPIWatcher) addRoute(route *zebra.IPRouteBody) { - pfx, ok := ipNetFromPrefix(route.Prefix) - if !ok { - w.log.Debug("zapi watcher: ignore route add with invalid prefix") - return - } - rib := ribName(pfx) - key := rib + ":" + pfx.String() + ":" + routeProtocol(route.Type) +func (w *ZAPIWatcher) addRoute(route *zapi.Route) { + rib := ribName(route.Prefix) + key := rib + ":" + route.Prefix.String() + ":" + routeProtocol(route.Type) w.mu.Lock() w.routes[key] = transformRoute(route) @@ -204,13 +171,9 @@ func (w *ZAPIWatcher) addRoute(route *zebra.IPRouteBody) { w.writeRibs() } -func (w *ZAPIWatcher) deleteRoute(route *zebra.IPRouteBody) { - pfx, ok := ipNetFromPrefix(route.Prefix) - if !ok { - return - } - rib := ribName(pfx) - key := rib + ":" + pfx.String() + ":" + routeProtocol(route.Type) +func (w *ZAPIWatcher) deleteRoute(route *zapi.Route) { + rib := ribName(route.Prefix) + key := rib + ":" + route.Prefix.String() + ":" + routeProtocol(route.Type) w.mu.Lock() delete(w.routes, key) @@ -234,7 +197,7 @@ func (w *ZAPIWatcher) writeRibs() { ipv6Routes := make([]json.RawMessage, 0) for key, routeData := range w.routes { - if strings.HasPrefix(key, "ipv4-master:") { + if strings.HasPrefix(key, "ipv4:") { ipv4Routes = append(ipv4Routes, routeData) } else { ipv6Routes = append(ipv6Routes, routeData) @@ -245,14 +208,14 @@ func (w *ZAPIWatcher) writeRibs() { ribs := map[string]any{ "rib": []map[string]any{ { - "name": "ipv4-master", + "name": "ipv4", "address-family": "ietf-routing:ipv4", "routes": map[string]any{ "route": ipv4Routes, }, }, { - "name": "ipv6-master", + "name": "ipv6", "address-family": "ietf-routing:ipv6", "routes": map[string]any{ "route": ipv6Routes, @@ -272,15 +235,19 @@ func (w *ZAPIWatcher) writeRibs() { func ribName(prefix net.IPNet) string { if prefix.IP.To4() != nil { - return "ipv4-master" + return "ipv4" } - return "ipv6-master" + return "ipv6" } -func transformRoute(route *zebra.IPRouteBody) json.RawMessage { - pfx, ok := ipNetFromPrefix(route.Prefix) - if !ok { - return json.RawMessage(`{}`) +func transformRoute(route *zapi.Route) json.RawMessage { + isIPv4 := route.Prefix.IP.To4() != nil + + addrKey := "ietf-ipv6-unicast-routing:address" + destKey := "ietf-ipv6-unicast-routing:destination-prefix" + if isIPv4 { + addrKey = "ietf-ipv4-unicast-routing:address" + destKey = "ietf-ipv4-unicast-routing:destination-prefix" } nextHops := make([]map[string]any, 0, len(route.Nexthops)) @@ -288,7 +255,7 @@ func transformRoute(route *zebra.IPRouteBody) json.RawMessage { hop := make(map[string]any) if len(nh.Gate) > 0 && !nh.Gate.IsUnspecified() { - hop["next-hop-address"] = nh.Gate.String() + hop[addrKey] = nh.Gate.String() } if nh.Ifindex > 0 { @@ -306,11 +273,13 @@ func transformRoute(route *zebra.IPRouteBody) json.RawMessage { } routeNode := map[string]any{ - "destination-prefix": pfx.String(), - "source-protocol": routeProtocol(route.Type), - "metric": route.Metric, - "next-hop-list": map[string]any{ - "next-hop": nextHops, + destKey: route.Prefix.String(), + "source-protocol": routeProtocol(route.Type), + "route-preference": route.Metric, + "next-hop": map[string]any{ + "next-hop-list": map[string]any{ + "next-hop": nextHops, + }, }, } @@ -322,33 +291,9 @@ func transformRoute(route *zebra.IPRouteBody) json.RawMessage { return json.RawMessage(encoded) } -func routeProtocol(rt zebra.RouteType) string { +func routeProtocol(rt zapi.RouteType) string { if protocol, ok := routeTypeToProtocol[rt]; ok { return protocol } return "infix-routing:kernel" } - -func ipNetFromPrefix(prefix zebra.Prefix) (net.IPNet, bool) { - ip := prefix.Prefix - if len(ip) == 0 { - return net.IPNet{}, false - } - - bits := 128 - if v4 := ip.To4(); v4 != nil { - ip = v4 - bits = 32 - } - - prefixLen := int(prefix.PrefixLen) - if prefixLen < 0 { - prefixLen = 0 - } - if prefixLen > bits { - prefixLen = bits - } - - mask := net.CIDRMask(prefixLen, bits) - return net.IPNet{IP: ip.Mask(mask), Mask: mask}, true -} diff --git a/src/yangerd/internal/zapiwatcher/zapiwatcher_test.go b/src/yangerd/internal/zapiwatcher/zapiwatcher_test.go index 78f85942..4741773a 100644 --- a/src/yangerd/internal/zapiwatcher/zapiwatcher_test.go +++ b/src/yangerd/internal/zapiwatcher/zapiwatcher_test.go @@ -5,22 +5,22 @@ import ( "net" "testing" - "github.com/osrg/gobgp/v3/pkg/zebra" + "github.com/kernelkit/infix/src/yangerd/internal/zapi" ) func TestRouteProtocol(t *testing.T) { tests := []struct { name string - rt zebra.RouteType + rt zapi.RouteType expected string }{ - {"kernel", RouteKernel, "infix-routing:kernel"}, - {"connect", RouteConnect, "ietf-routing:direct"}, - {"static", RouteStatic, "ietf-routing:static"}, - {"ospf", RouteOSPF, "ietf-ospf:ospfv2"}, - {"rip", RouteRIP, "ietf-rip:ripv2"}, - {"unknown defaults to kernel", zebra.RouteType(99), "infix-routing:kernel"}, - {"zero defaults to kernel", zebra.RouteType(0), "infix-routing:kernel"}, + {"kernel", zapi.RouteKernel, "infix-routing:kernel"}, + {"connect", zapi.RouteConnect, "ietf-routing:direct"}, + {"static", zapi.RouteStatic, "ietf-routing:static"}, + {"ospf", zapi.RouteOSPF, "ietf-ospf:ospfv2"}, + {"rip", zapi.RouteRIP, "ietf-rip:ripv2"}, + {"unknown defaults to kernel", zapi.RouteType(99), "infix-routing:kernel"}, + {"zero defaults to kernel", zapi.RouteType(0), "infix-routing:kernel"}, } for _, tc := range tests { @@ -42,17 +42,17 @@ func TestRibName(t *testing.T) { { "ipv4", net.IPNet{IP: net.ParseIP("10.0.0.0").To4(), Mask: net.CIDRMask(24, 32)}, - "ipv4-master", + "ipv4", }, { "ipv6", net.IPNet{IP: net.ParseIP("2001:db8::"), Mask: net.CIDRMask(64, 128)}, - "ipv6-master", + "ipv6", }, { "loopback v4", net.IPNet{IP: net.ParseIP("127.0.0.0").To4(), Mask: net.CIDRMask(8, 32)}, - "ipv4-master", + "ipv4", }, } @@ -67,12 +67,14 @@ func TestRibName(t *testing.T) { } func TestTransformRoute(t *testing.T) { - route := &zebra.IPRouteBody{ - Type: RouteStatic, + route := &zapi.Route{ + Type: zapi.RouteStatic, Metric: 100, + Prefix: net.IPNet{ + IP: net.ParseIP("10.0.0.0").To4(), + Mask: net.CIDRMask(24, 32), + }, } - route.Prefix.Prefix = net.ParseIP("10.0.0.0") - route.Prefix.PrefixLen = 24 result := transformRoute(route) @@ -81,7 +83,7 @@ func TestTransformRoute(t *testing.T) { t.Fatalf("unmarshal transformRoute result: %v", err) } - if got := parsed["destination-prefix"]; got != "10.0.0.0/24" { + if got := parsed["ietf-ipv4-unicast-routing:destination-prefix"]; got != "10.0.0.0/24" { t.Errorf("destination-prefix = %v, want %q", got, "10.0.0.0/24") } @@ -89,13 +91,17 @@ func TestTransformRoute(t *testing.T) { t.Errorf("source-protocol = %v, want %q", got, "ietf-routing:static") } - if got, ok := parsed["metric"].(float64); !ok || int(got) != 100 { - t.Errorf("metric = %v, want 100", parsed["metric"]) + if got, ok := parsed["route-preference"].(float64); !ok || int(got) != 100 { + t.Errorf("route-preference = %v, want 100", parsed["route-preference"]) } - nhList, ok := parsed["next-hop-list"].(map[string]any) + nhContainer, ok := parsed["next-hop"].(map[string]any) if !ok { - t.Fatalf("next-hop-list not a map: %T", parsed["next-hop-list"]) + t.Fatalf("next-hop not a map: %T", parsed["next-hop"]) + } + nhList, ok := nhContainer["next-hop-list"].(map[string]any) + if !ok { + t.Fatalf("next-hop-list not a map: %T", nhContainer["next-hop-list"]) } hops, ok := nhList["next-hop"].([]any) if !ok { @@ -107,12 +113,14 @@ func TestTransformRoute(t *testing.T) { } func TestTransformRouteIPv6(t *testing.T) { - route := &zebra.IPRouteBody{ - Type: RouteOSPF, + route := &zapi.Route{ + Type: zapi.RouteOSPF, Metric: 10, + Prefix: net.IPNet{ + IP: net.ParseIP("2001:db8::").To16(), + Mask: net.CIDRMask(48, 128), + }, } - route.Prefix.Prefix = net.ParseIP("2001:db8::") - route.Prefix.PrefixLen = 48 result := transformRoute(route) @@ -121,65 +129,10 @@ func TestTransformRouteIPv6(t *testing.T) { t.Fatalf("unmarshal: %v", err) } - if got := parsed["destination-prefix"]; got != "2001:db8::/48" { + if got := parsed["ietf-ipv6-unicast-routing:destination-prefix"]; got != "2001:db8::/48" { t.Errorf("destination-prefix = %v, want %q", got, "2001:db8::/48") } if got := parsed["source-protocol"]; got != "ietf-ospf:ospfv2" { t.Errorf("source-protocol = %v, want %q", got, "ietf-ospf:ospfv2") } } - -func TestIpNetFromPrefix(t *testing.T) { - tests := []struct { - name string - prefix zebra.Prefix - wantCIDR string - wantOK bool - }{ - { - "valid ipv4", - zebra.Prefix{Prefix: net.ParseIP("192.168.1.0"), PrefixLen: 24}, - "192.168.1.0/24", - true, - }, - { - "valid ipv6", - zebra.Prefix{Prefix: net.ParseIP("2001:db8::"), PrefixLen: 64}, - "2001:db8::/64", - true, - }, - { - "host route ipv4", - zebra.Prefix{Prefix: net.ParseIP("10.0.0.1"), PrefixLen: 32}, - "10.0.0.1/32", - true, - }, - { - "zero prefix length", - zebra.Prefix{Prefix: net.ParseIP("0.0.0.0"), PrefixLen: 0}, - "0.0.0.0/0", - true, - }, - { - "prefix masking applied", - zebra.Prefix{Prefix: net.ParseIP("10.0.0.5"), PrefixLen: 24}, - "10.0.0.0/24", - true, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - got, ok := ipNetFromPrefix(tc.prefix) - if ok != tc.wantOK { - t.Fatalf("ipNetFromPrefix ok = %v, want %v", ok, tc.wantOK) - } - if !ok { - return - } - if got.String() != tc.wantCIDR { - t.Errorf("ipNetFromPrefix = %q, want %q", got.String(), tc.wantCIDR) - } - }) - } -}