Fix routing and ip addresses

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
 

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 *,

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
 }
 

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 {

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)
+	}
+}

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
+}

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
-}

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)
-			}
-		})
-	}
-}