Eden to evetest: rewrite networking, security, apps and lps tests

evetest/Dockerfile.evetest

@@ -5,7 +5,7 @@ ARG EVETEST_VERSION=dev
 ARG ALPINE_VERSION=3.21
 ARG GOLANG_VERSION=1.25
 ARG EVETEST_ADAM_REPO=lfedge/adam
-ARG EVETEST_ADAM_VERSION=0.0.75
+ARG EVETEST_ADAM_VERSION=0.0.81
 
 ###########################
 # Pull adam binary stage

evetest/Makefile

@@ -4,7 +4,7 @@
 EVETEST_VERSION := $(shell grep -v '^\#' VERSION | head -n1)
 EVETEST_ORG ?= lfedge
 
-EVETEST_ADAM_VERSION ?= 0.0.75
+EVETEST_ADAM_VERSION ?= 0.0.81
 EVETEST_ADAM_REPO ?= lfedge/adam
 
 EVETEST_IMAGE := $(EVETEST_ORG)/evetest:$(EVETEST_VERSION)

evetest/README.md

@@ -750,8 +750,8 @@ same terminal session beforehand.
 |----------|-------------|---------|
 | `EVETEST_ORG` | Docker Hub organization for evetest and evetest-broker images | `lfedge` |
 | `EVETEST_EVE_REPO` | EVE image repository | `lfedge/eve` |
-| `EVETEST_ADAM_VERSION` | Adam controller version *(build-time only, see note below)* | `0.0.75` |
-| `EVETEST_SDN_VERSION` | SDN emulator version | `1.0` |
+| `EVETEST_ADAM_VERSION` | Adam controller version *(build-time only, see note below)* | `0.0.81` |
+| `EVETEST_SDN_VERSION` | SDN emulator version | `1.1` |
 
 > **`EVETEST_ADAM_VERSION` requires evetest container rebuild.** Unlike the other
 > variables above, Adam's binary is baked into the evetest image at build time

evetest/VERSION

@@ -1,2 +1,2 @@
 # Evetest version. Increment this manually whenever changes are made to the evetest framework.
-1.0
+1.1

evetest/cli/evecmd.go

@@ -468,22 +468,27 @@ func eveAppFlowLogsCmd() *cobra.Command {
 				if err != nil {
 					return fmt.Errorf("stream error: %w", err)
 				}
-				var lines []string
+				var newEntries []string
 				for _, ipFlow := range resp.IpFlows {
-					lines = append(lines, fmt.Sprintf("IP flow: %s", ipFlow.String()))
+					newEntries = append(newEntries,
+						fmt.Sprintf("IP flow: %s", ipFlow.String()))
 				}
 				for _, dnsReq := range resp.DnsRequests {
-					lines = append(lines, fmt.Sprintf("DNS request: %s", dnsReq.String()))
+					newEntries = append(newEntries,
+						fmt.Sprintf("DNS request: %s", dnsReq.String()))
 				}
-				entry := strings.Join(lines, "\n")
 				if tail > 0 {
-					entries = append(entries, entry)
+					entries = append(entries, newEntries...)
 				} else {
-					fmt.Println(entry)
+					for _, entry := range newEntries {
+						fmt.Println(entry)
+						fmt.Println()
+					}
 				}
 			}
 			for _, e := range tailEntries(entries, tail) {
 				fmt.Println(e)
+				fmt.Println()
 			}
 			return nil
 		},

evetest/constants/config.go

@@ -276,10 +276,10 @@ const (
 	DefaultSDNRepo = "lfedge/evetest-sdn"
 
 	// DefaultAdamVersion specifies the Adam version to use by default.
-	DefaultAdamVersion = "0.0.75"
+	DefaultAdamVersion = "0.0.81"
 
 	// DefaultSDNVersion specifies the SDN version to use by default.
-	DefaultSDNVersion = "1.0"
+	DefaultSDNVersion = "1.1"
 
 	// DefaultSDNUplinkIPv4Subnet species the IPv4 subnet used for SDN uplink
 	// interfaces by default.

evetest/controller/adam.go

@@ -31,6 +31,7 @@ import (
 
 	evecerts "github.com/lf-edge/eve-api/go/certs"
 	eveconfig "github.com/lf-edge/eve-api/go/config"
+	eveflowlog "github.com/lf-edge/eve-api/go/flowlog"
 	eveinfo "github.com/lf-edge/eve-api/go/info"
 	evelogs "github.com/lf-edge/eve-api/go/logs"
 	evemetrics "github.com/lf-edge/eve-api/go/metrics"
@@ -168,6 +169,16 @@ type MetricMsgIterator interface {
 	Iterate(msg *evemetrics.ZMetricMsg) (stop bool, err error)
 }
 
+// FlowMsgIterator iterates over device flow log messages (FlowMessage), each
+// of which carries both flow records and DNS request records for one
+// application VIF. Iterate is called for each message that passes the match
+// filter. Returning stop=true signals that no further messages are needed
+// and iteration should stop cleanly. Returning a non-nil error aborts
+// iteration and propagates the error to the caller.
+type FlowMsgIterator interface {
+	Iterate(msg *eveflowlog.FlowMessage) (stop bool, err error)
+}
+
 // NewAdamClient creates a new AdamClient.
 // The caller is responsible for providing a CA certificate and key used
 // to sign all Adam server certificates.
@@ -1684,6 +1695,220 @@ func (ac *AdamClient) SubscribeToDeviceMetrics(devUUID uuid.UUID,
 	return unsubscribe, nil
 }
 
+// IterateDeviceFlowLogs retrieves flow log messages (FlowMessage) published
+// by the specified device and passes matching messages to iterator. Flow
+// messages are stored and served per-device, not per-app (each one carries
+// a Scope identifying which app/VIF it belongs to), so callers that only
+// care about one application filter by msg.GetScope().GetUuid() in match (or
+// inside iterator) themselves.
+//
+// It first performs a one-shot GET request to fetch all currently available
+// messages. If follow is true, it then subscribes to the streaming endpoint
+// and continues delivering new messages until ctx is canceled.
+//
+// If match is non-nil, only messages for which match(msg) returns true are
+// iterated. If match is nil, all messages are iterated.
+func (ac *AdamClient) IterateDeviceFlowLogs(ctx context.Context, devUUID uuid.UUID,
+	match func(msg *eveflowlog.FlowMessage) bool, iterator FlowMsgIterator,
+	follow bool) error {
+	if err := ac.checkAdamRunning(); err != nil {
+		return err
+	}
+
+	ac.mutex.Lock()
+	_, known := ac.knownDevices[devUUID]
+	ac.mutex.Unlock()
+	if !known {
+		return fmt.Errorf("unknown device UUID %q", devUUID)
+	}
+
+	// -------- Initial GET --------
+
+	url := ac.adminURL("device/" + devUUID.String() + "/flowlogs")
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return fmt.Errorf("failed to create GET %s request: %w", url, err)
+	}
+
+	resp, err := ac.httpClient().Do(req)
+	if err != nil {
+		return fmt.Errorf("GET %s failed: %w", url, err)
+	}
+	defer resp.Body.Close()
+
+	// No flow logs recorded yet for this device.
+	if resp.StatusCode != http.StatusNotFound {
+		if resp.StatusCode != http.StatusOK {
+			return fmt.Errorf("unexpected status from GET %s: %d", url, resp.StatusCode)
+		}
+
+		dec := json.NewDecoder(resp.Body)
+		for {
+			var raw json.RawMessage
+			if err := dec.Decode(&raw); err != nil {
+				if errors.Is(err, io.EOF) {
+					break
+				}
+				return fmt.Errorf("failed to decode flow message JSON: %w", err)
+			}
+
+			msg := &eveflowlog.FlowMessage{}
+			if err := protojson.Unmarshal(raw, msg); err != nil {
+				return fmt.Errorf("failed to proto-unmarshal flow message: %w", err)
+			}
+			if match == nil || match(msg) {
+				stop, iterErr := iterator.Iterate(msg)
+				if iterErr != nil {
+					return fmt.Errorf("failed to iterate flow message: %w", iterErr)
+				}
+				if stop {
+					return nil
+				}
+			}
+		}
+	}
+
+	// -------- Follow mode --------
+
+	if !follow {
+		return nil
+	}
+
+	flowMsgCh := make(chan *eveflowlog.FlowMessage, 100)
+	unsubscribe, err := ac.SubscribeToDeviceFlowLogs(devUUID, match, flowMsgCh)
+	if err != nil {
+		return err
+	}
+	defer unsubscribe()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case msg := <-flowMsgCh:
+			stop, iterErr := iterator.Iterate(msg)
+			if iterErr != nil {
+				return fmt.Errorf("failed to iterate flow message: %w", iterErr)
+			}
+			if stop {
+				return nil
+			}
+		}
+	}
+}
+
+// SubscribeToDeviceFlowLogs subscribes to flow log messages (FlowMessage)
+// emitted by the specified device and delivers matching messages to
+// channel. Flow messages are stored and served per-device, not per-app; see
+// IterateDeviceFlowLogs.
+//
+// If match is non-nil, only messages for which match(msg) returns true are
+// forwarded. If match is nil, all messages are delivered.
+//
+// The streaming connection is opened synchronously: by the time this method
+// returns, Adam has accepted the request and any subsequent flow messages
+// for the device will be delivered. On transient failures after the initial
+// connection, a background goroutine reconnects with a fixed retry delay.
+//
+// The returned unsubscribe function stops the background stream and waits
+// for it to exit. It is safe to call multiple times. The channel is closed
+// when the subscription ends.
+func (ac *AdamClient) SubscribeToDeviceFlowLogs(devUUID uuid.UUID,
+	match func(msg *eveflowlog.FlowMessage) bool,
+	channel chan<- *eveflowlog.FlowMessage) (unsubscribe func(), err error) {
+	const retryDelay = 3 * time.Second
+
+	if err = ac.checkAdamRunning(); err != nil {
+		return nil, err
+	}
+
+	ac.mutex.Lock()
+	_, known := ac.knownDevices[devUUID]
+	ac.mutex.Unlock()
+	if !known {
+		return nil, fmt.Errorf("unknown device UUID %q", devUUID)
+	}
+
+	streamCtx, cancel := context.WithCancel(context.Background())
+	url := ac.adminURL("device/" + devUUID.String() + "/flowlogs")
+
+	resp, err := ac.openStream(streamCtx, url)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		defer close(channel)
+
+		current := resp
+		for {
+			if current == nil {
+				select {
+				case <-time.After(retryDelay):
+				case <-streamCtx.Done():
+					return
+				}
+				r, err := ac.openStream(streamCtx, url)
+				if err != nil {
+					if streamCtx.Err() != nil {
+						return
+					}
+					ac.log.Errorf("failed to reopen flow log stream: %v", err)
+					continue
+				}
+				current = r
+			}
+
+			func() {
+				defer current.Body.Close()
+				dec := json.NewDecoder(current.Body)
+				for {
+					var raw json.RawMessage
+					if err := dec.Decode(&raw); err != nil {
+						if streamCtx.Err() != nil {
+							return
+						}
+						if errors.Is(err, io.EOF) {
+							ac.log.Warn("flow log stream closed by server")
+							return
+						}
+						ac.log.Errorf("failed to decode streamed flow message: %v", err)
+						return
+					}
+					msg := &eveflowlog.FlowMessage{}
+					if err := protojson.Unmarshal(raw, msg); err != nil {
+						ac.log.Errorf(
+							"failed to proto-unmarshal streamed flow message: %v", err)
+						continue
+					}
+					if match != nil && !match(msg) {
+						continue
+					}
+					select {
+					case channel <- msg:
+					case <-streamCtx.Done():
+						return
+					}
+				}
+			}()
+			current = nil
+		}
+	}()
+
+	var once sync.Once
+	unsubscribe = func() {
+		once.Do(func() {
+			cancel()
+			wg.Wait()
+		})
+	}
+	return unsubscribe, nil
+}
+
 // findDeviceUUID searches Adam for a device with certificates/serial matching
 // the given callback and returns its UUID if found.
 func (ac *AdamClient) findDeviceUUID(ctx context.Context, httpClient *http.Client,

evetest/devconfig.go

@@ -871,17 +871,27 @@ func (config SwitchNetworkInstanceConfig) toProto(th *TestHarness,
 
 // ApplicationInstanceConfig wraps configuration for a single application deployed on EVE.
 type ApplicationInstanceConfig struct {
-	DisplayName         string
-	Activate            bool
-	ProfileList         []string
-	Image               ApplicationImageStorage
-	VirtualizationMode  eveconfig.VmMode
-	CPUs                uint
-	MemoryBytes         uint64
-	DiskBytes           uint64
-	EnableVNC           bool
-	VNCDisplay          uint
-	VNCPassword         string
+	DisplayName        string
+	Activate           bool
+	ProfileList        []string
+	Image              ApplicationImageStorage
+	VirtualizationMode eveconfig.VmMode
+	CPUs               uint
+	MemoryBytes        uint64
+	DiskBytes          uint64
+	EnableVNC          bool
+	VNCDisplay         uint
+	VNCPassword        string
+	// RemoteConsole gates VNC access under the Kubevirt hypervisor: unlike
+	// KVM (where EnableVNC/VNCDisplay/VNCPassword directly configure a raw
+	// QEMU VNC socket reachable on the device's uplink IP), Kubevirt exposes
+	// VNC via zedkube's virtctl-based proxy, which only binds the VNC port
+	// on 127.0.0.1 and is gated by this field, not by EnableVNC. VNCDisplay
+	// still selects the port (5900+VNCDisplay); VNCPassword is not enforced
+	// -- see pkg/pillar/docs/vnc-workflows.md. zedkube only starts/stops the
+	// proxy on a *change* of this field, and only one remote-console session
+	// is allowed on the device at a time.
+	RemoteConsole       bool
 	DisableLogs         bool
 	UserData            string
 	NetworkAdapters     []AppNetworkAdapter
@@ -912,6 +922,7 @@ func (config ApplicationInstanceConfig) toProto(th *TestHarness, devName string,
 		Fixedresources: vmConfig,
 		Activate:       config.Activate,
 		ProfileList:    config.ProfileList,
+		RemoteConsole:  config.RemoteConsole,
 	}
 	if volumeUUID != NilUUID {
 		appInstConfig.VolumeRefList = append(appInstConfig.VolumeRefList,