internal/uevent: consume kernel-originated events

These are not possible to cover outside integration vm. Extreme care is required when dealing with this method, so keep it simple.

Signed-off-by: Ophestra <cat@gensokyo.uk>

internal/uevent/uevent.go

@@ -4,6 +4,9 @@
 package uevent
 
 import (
+	"context"
+	"errors"
+	"strconv"
 	"sync/atomic"
 	"syscall"
 
@@ -48,3 +51,55 @@ func Dial() (*Conn, error) {
 	}
 	return &Conn{conn: c}, err
 }
+
+var (
+	// ErrBadSocket is returned by [Conn.Consume] for a reply from a
+	// syscall.Sockaddr with unexpected concrete type.
+	ErrBadSocket = errors.New("unexpected socket address")
+)
+
+// BadPortError is returned by [Conn.Consume] upon receiving a message that did
+// not come from the kernel.
+type BadPortError syscall.SockaddrNetlink
+
+var _ Recoverable = new(BadPortError)
+
+func (*BadPortError) recoverable() {}
+func (e *BadPortError) Error() string {
+	return "unexpected message from port id " + strconv.Itoa(int(e.Pid)) +
+		" on NETLINK_KOBJECT_UEVENT"
+}
+
+// Consume continuously receives and parses events from the kernel. It returns
+// the first error it encounters.
+//
+// Callers must not restart event processing after a non-nil error that does not
+// satisfy [Recoverable] is returned.
+func (c *Conn) Consume(ctx context.Context, events chan<- *Message) error {
+	if err := c.enterExcl(); err != nil {
+		return err
+	}
+	defer c.exitExcl()
+
+	for {
+		data, from, err := c.conn.Recvfrom(ctx, 0)
+		if err != nil {
+			return err
+		}
+
+		// lib/kobject_uevent.c:
+		// set portid 0 to inform userspace message comes from kernel
+		if v, ok := from.(*syscall.SockaddrNetlink); !ok {
+			return ErrBadSocket
+		} else if v.Pid != 0 {
+			return (*BadPortError)(v)
+
+		}
+
+		var msg Message
+		if err = msg.UnmarshalBinary(data); err != nil {
+			return err
+		}
+		events <- &msg
+	}
+}

internal/uevent/uevent_test.go

@@ -138,6 +138,10 @@ func TestErrors(t *testing.T) {
 			Data: "\x00",
 			Kind: 0xbad,
 		}, `section "" is invalid`},
+
+		{"BadPortError", &uevent.BadPortError{
+			Pid: 1,
+		}, "unexpected message from port id 1 on NETLINK_KOBJECT_UEVENT"},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {