feat(nacos): add access control and namespace administration

2026-10-19 16:16:45 +00:00 · 2026-08-21 01:15:46 +08:00
parent 30f75d8388
commit 7fa94152a8
59 changed files with 4563 additions and 479 deletions
@@ -209,6 +209,16 @@ test('initializes both Nacos versions with the shared administrator credentials'
  }
 });

+test('verifies Nacos 2.5 rejects unauthenticated configuration reads', () => {
+  const recipe = discoverRecipes().find((item) => recipeSelector(item) === 'nacos@2.5');
+  const authenticationStep = recipe.smoke.steps.find((step) => step.name === 'reject unauthenticated configuration reads');
+
+  assert.ok(authenticationStep);
+  assert.match(authenticationStep.command.join(' '), /nacos\/v1\/cs\/configs/);
+  assert.match(authenticationStep.command.join(' '), /403/);
+  assert.equal(authenticationStep.expect, 'unauthenticated access rejected');
+});
+
 test('configures r-nacos with its admin account and console port', () => {
  const recipe = discoverRecipes().find((item) => recipeSelector(item) === 'rnacos@0.8');
  assert.equal(recipe.connection.username, 'admin');
@@ -0,0 +1,767 @@
+// Command nacos-service-seed creates deterministic Nacos services and persistent instances.
+//
+// The defaults match the local environments started by this repository:
+//
+//	make db DB=nacos@2.5
+//	make db DB=nacos@3.2
+//	go run ./scripts/nacos-service-seed.go -target both -serve
+//
+// Nacos 3 registration intentionally uses the 8848 Admin API. The generated
+// instances can then be inspected through a separate 8080 Console API DBX
+// connection to verify the reduced Console capability set.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"os/signal"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+)
+
+const (
+	maxResponseBytes   = 1 << 20
+	serviceOwnerSource = "dbx-nacos-service-seed"
+	catalogPageSize    = 100
+)
+
+type options struct {
+	target           string
+	action           string
+	v2URL            string
+	v3URL            string
+	username         string
+	password         string
+	namespace        string
+	group            string
+	cluster          string
+	prefix           string
+	serviceCount     int
+	instancesPerSvc  int
+	instanceIP       string
+	instanceBasePort int
+	serve            bool
+	forceExisting    bool
+	listenAddress    string
+	timeout          time.Duration
+}
+
+type demoServers struct {
+	servers     []*http.Server
+	addresses   []string
+	serveErrors chan error
+}
+
+type nacosTarget struct {
+	name                 string
+	baseURL              string
+	loginPath            string
+	servicePath          string
+	instancePath         string
+	instanceListPath     string
+	instanceHealthPath   string
+	instanceListCatalog  bool
+	supportsHealthUpdate bool
+	tokenInHeader        bool
+	token                string
+	client               *http.Client
+}
+
+type nacosInstance struct {
+	IP          string
+	Port        int
+	ClusterName string
+	Ephemeral   bool
+}
+
+func main() {
+	opts := parseFlags()
+	if err := validateOptions(opts); err != nil {
+		fmt.Fprintln(os.Stderr, "error:", err)
+		os.Exit(2)
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	var listeners *demoServers
+	if opts.serve {
+		var err error
+		listeners, err = startDemoServers(opts)
+		if err != nil {
+			fmt.Fprintln(os.Stderr, "error:", err)
+			os.Exit(1)
+		}
+		defer func() {
+			shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			if err := listeners.shutdown(shutdownCtx); err != nil {
+				fmt.Fprintln(os.Stderr, "error: stop demo servers:", err)
+			}
+		}()
+		fmt.Printf("[demo] listening on %s\n", strings.Join(listeners.addresses, ", "))
+	}
+
+	for _, target := range selectedTargets(opts) {
+		fmt.Printf("\n[%s] %s %d services x %d instances at %s\n", target.name, opts.action, opts.serviceCount, opts.instancesPerSvc, target.baseURL)
+		if err := target.login(ctx, opts.username, opts.password); err != nil {
+			fmt.Fprintf(os.Stderr, "[%s] login failed: %v\n", target.name, err)
+			os.Exit(1)
+		}
+		var err error
+		if opts.action == "cleanup" {
+			err = target.cleanup(ctx, opts)
+		} else {
+			err = target.seed(ctx, opts)
+		}
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "[%s] %s failed: %v\n", target.name, opts.action, err)
+			os.Exit(1)
+		}
+	}
+	if listeners != nil {
+		fmt.Println("[demo] serving registered instances; press Ctrl+C to stop")
+		if err := listeners.wait(ctx); err != nil {
+			fmt.Fprintln(os.Stderr, "error: demo server:", err)
+			os.Exit(1)
+		}
+	}
+}
+
+func parseFlags() options {
+	var opts options
+	flag.StringVar(&opts.target, "target", "both", "target environment: v2, v3, or both")
+	flag.StringVar(&opts.action, "action", "seed", "action: seed or cleanup")
+	flag.StringVar(&opts.v2URL, "v2-url", "http://127.0.0.1:11000/nacos", "Nacos V2 base URL")
+	flag.StringVar(&opts.v3URL, "v3-url", "http://127.0.0.1:11003/nacos", "Nacos V3 Admin base URL (8848 mapping, not the 8080 Console URL)")
+	flag.StringVar(&opts.username, "username", "nacos", "Nacos username")
+	flag.StringVar(&opts.password, "password", "123456", "Nacos password")
+	flag.StringVar(&opts.namespace, "namespace", "public", "existing namespace ID")
+	flag.StringVar(&opts.group, "group", "DEFAULT_GROUP", "service group")
+	flag.StringVar(&opts.cluster, "cluster", "DEFAULT", "instance cluster")
+	flag.StringVar(&opts.prefix, "prefix", "dbx-test-service", "generated service name prefix")
+	flag.IntVar(&opts.serviceCount, "services", 5, "number of services")
+	flag.IntVar(&opts.instancesPerSvc, "instances", 2, "persistent instances per service")
+	flag.StringVar(&opts.instanceIP, "ip", "127.0.0.1", "registered instance IP")
+	flag.IntVar(&opts.instanceBasePort, "port", 28080, "first registered instance port")
+	flag.BoolVar(&opts.serve, "serve", false, "keep an HTTP demo server listening on every registered instance port")
+	flag.BoolVar(&opts.forceExisting, "force-existing", false, "allow seed or cleanup to modify matching services not created by this script")
+	flag.StringVar(&opts.listenAddress, "listen-address", "127.0.0.1", "address used by -serve listeners")
+	flag.DurationVar(&opts.timeout, "timeout", 10*time.Second, "HTTP request timeout")
+	flag.Parse()
+	return opts
+}
+
+func validateOptions(opts options) error {
+	if opts.target != "v2" && opts.target != "v3" && opts.target != "both" {
+		return fmt.Errorf("invalid -target %q; use v2, v3, or both", opts.target)
+	}
+	if opts.action != "seed" && opts.action != "cleanup" {
+		return fmt.Errorf("invalid -action %q; use seed or cleanup", opts.action)
+	}
+	if opts.serve && opts.action != "seed" {
+		return errors.New("-serve can only be used with -action seed")
+	}
+	if opts.username == "" || opts.password == "" {
+		return errors.New("-username and -password must not be empty")
+	}
+	if opts.namespace == "" || opts.group == "" || opts.cluster == "" || opts.prefix == "" {
+		return errors.New("-namespace, -group, -cluster, and -prefix must not be empty")
+	}
+	if opts.serviceCount < 1 || opts.instancesPerSvc < 1 {
+		return errors.New("-services and -instances must be greater than zero")
+	}
+	lastPort := opts.instanceBasePort + opts.serviceCount*opts.instancesPerSvc - 1
+	if opts.instanceBasePort < 1 || lastPort > 65535 {
+		return fmt.Errorf("generated instance port range %d-%d is invalid", opts.instanceBasePort, lastPort)
+	}
+	for label, rawURL := range map[string]string{"v2-url": opts.v2URL, "v3-url": opts.v3URL} {
+		parsed, err := url.ParseRequestURI(rawURL)
+		if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" {
+			return fmt.Errorf("-%s must be an absolute HTTP URL", label)
+		}
+	}
+	return nil
+}
+
+func startDemoServers(opts options) (*demoServers, error) {
+	set := &demoServers{serveErrors: make(chan error, opts.serviceCount*opts.instancesPerSvc)}
+	for serviceIndex := 0; serviceIndex < opts.serviceCount; serviceIndex++ {
+		serviceName := generatedServiceName(opts.prefix, serviceIndex)
+		for instanceIndex := 0; instanceIndex < opts.instancesPerSvc; instanceIndex++ {
+			port := generatedPort(opts, serviceIndex, instanceIndex)
+			listener, err := net.Listen("tcp", net.JoinHostPort(opts.listenAddress, strconv.Itoa(port)))
+			if err != nil {
+				shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+				_ = set.shutdown(shutdownCtx)
+				cancel()
+				return nil, fmt.Errorf("listen on %s:%d: %w", opts.listenAddress, port, err)
+			}
+			server := &http.Server{Handler: demoHandler(serviceName, port)}
+			set.servers = append(set.servers, server)
+			set.addresses = append(set.addresses, listener.Addr().String())
+			go func(server *http.Server, listener net.Listener) {
+				if err := server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
+					set.serveErrors <- err
+				}
+			}(server, listener)
+		}
+	}
+	return set, nil
+}
+
+func demoHandler(serviceName string, port int) http.Handler {
+	payload, _ := json.Marshal(map[string]any{"service": serviceName, "port": port, "status": "healthy"})
+	return http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) {
+		response.Header().Set("Content-Type", "application/json")
+		_, _ = response.Write(payload)
+	})
+}
+
+func (set *demoServers) wait(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return nil
+	case err := <-set.serveErrors:
+		return err
+	}
+}
+
+func (set *demoServers) shutdown(ctx context.Context) error {
+	var failures []string
+	for _, server := range set.servers {
+		if err := server.Shutdown(ctx); err != nil {
+			failures = append(failures, err.Error())
+		}
+	}
+	if len(failures) > 0 {
+		return errors.New(strings.Join(failures, "; "))
+	}
+	return nil
+}
+
+func selectedTargets(opts options) []*nacosTarget {
+	client := &http.Client{Timeout: opts.timeout}
+	v2 := &nacosTarget{
+		name: "Nacos V2", baseURL: strings.TrimRight(opts.v2URL, "/"), loginPath: "/v1/auth/login",
+		servicePath: "/v1/ns/service", instancePath: "/v1/ns/instance", instanceListPath: "/v1/ns/catalog/instances", instanceHealthPath: "/v1/ns/instance", instanceListCatalog: true, supportsHealthUpdate: true, client: client,
+	}
+	v3 := &nacosTarget{
+		name: "Nacos V3 Admin", baseURL: strings.TrimRight(opts.v3URL, "/"), loginPath: "/v3/auth/user/login",
+		servicePath: "/v3/admin/ns/service", instancePath: "/v3/admin/ns/instance", instanceListPath: "/v3/admin/ns/instance/list", tokenInHeader: true, client: client,
+	}
+	switch opts.target {
+	case "v2":
+		return []*nacosTarget{v2}
+	case "v3":
+		return []*nacosTarget{v3}
+	default:
+		return []*nacosTarget{v2, v3}
+	}
+}
+
+func (target *nacosTarget) login(ctx context.Context, username, password string) error {
+	body, _, err := target.rawRequest(ctx, http.MethodPost, target.loginPath, nil, url.Values{
+		"username": {username},
+		"password": {password},
+	}, false)
+	if err != nil {
+		return err
+	}
+	token, err := accessToken(body)
+	if err != nil {
+		return err
+	}
+	target.token = token
+	return nil
+}
+
+func (target *nacosTarget) seed(ctx context.Context, opts options) error {
+	createdInstances := 0
+	for serviceIndex := 0; serviceIndex < opts.serviceCount; serviceIndex++ {
+		serviceName := generatedServiceName(opts.prefix, serviceIndex)
+		if err := target.ensureService(ctx, opts, serviceName, serviceIndex); err != nil {
+			return fmt.Errorf("service %s: %w", serviceName, err)
+		}
+		for instanceIndex := 0; instanceIndex < opts.instancesPerSvc; instanceIndex++ {
+			port := generatedPort(opts, serviceIndex, instanceIndex)
+			metadata, _ := json.Marshal(map[string]any{
+				"source": serviceOwnerSource, "target": target.name,
+				"serviceIndex": serviceIndex + 1, "instanceIndex": instanceIndex + 1,
+			})
+			form := url.Values{
+				"namespaceId": {opts.namespace}, "groupName": {opts.group}, "serviceName": {serviceName},
+				"clusterName": {opts.cluster}, "ip": {opts.instanceIP}, "port": {strconv.Itoa(port)},
+				"weight":  {strconv.FormatFloat(1+float64(instanceIndex)/10, 'f', 1, 64)},
+				"healthy": {"true"}, "enabled": {"true"}, "ephemeral": {"false"}, "metadata": {string(metadata)},
+			}
+			if err := target.operation(ctx, http.MethodPost, target.instancePath, nil, form, false); err != nil {
+				return fmt.Errorf("register %s %s:%d: %w", serviceName, opts.instanceIP, port, err)
+			}
+			if target.supportsHealthUpdate {
+				if err := target.markInstanceHealthy(ctx, opts, serviceName, port); err != nil {
+					return fmt.Errorf("mark healthy %s %s:%d: %w", serviceName, opts.instanceIP, port, err)
+				}
+			}
+			createdInstances++
+		}
+		fmt.Printf("[%s] ready %-28s instances %d-%d\n", target.name, serviceName, generatedPort(opts, serviceIndex, 0), generatedPort(opts, serviceIndex, opts.instancesPerSvc-1))
+	}
+	fmt.Printf("[%s] complete: %d services, %d persistent instances\n", target.name, opts.serviceCount, createdInstances)
+	return nil
+}
+
+// markInstanceHealthy is only used by Nacos V2. Nacos V3 has no compatible
+// health-update endpoint for persistent instances.
+func (target *nacosTarget) markInstanceHealthy(ctx context.Context, opts options, serviceName string, port int) error {
+	form := url.Values{
+		"namespaceId": {opts.namespace}, "groupName": {opts.group}, "serviceName": {serviceName},
+		"clusterName": {opts.cluster}, "ip": {opts.instanceIP}, "port": {strconv.Itoa(port)},
+		"healthy": {"true"}, "ephemeral": {"false"},
+	}
+	return target.operation(ctx, http.MethodPut, target.instanceHealthPath, nil, form, false)
+}
+
+func (target *nacosTarget) ensureService(ctx context.Context, opts options, serviceName string, serviceIndex int) error {
+	query := url.Values{"namespaceId": {opts.namespace}, "groupName": {opts.group}, "serviceName": {serviceName}}
+	exists, owned, err := target.serviceStatus(ctx, query)
+	if err != nil {
+		return err
+	}
+	if exists && !owned && !opts.forceExisting {
+		return errors.New("matching service already exists without the script ownership marker; choose another -prefix or pass -force-existing to replace it")
+	}
+	metadata, _ := json.Marshal(map[string]any{
+		"source": serviceOwnerSource, "target": target.name, "serviceIndex": serviceIndex + 1,
+	})
+	form := url.Values{
+		"namespaceId": {opts.namespace}, "groupName": {opts.group}, "serviceName": {serviceName},
+		"protectThreshold": {"0"}, "ephemeral": {"false"}, "metadata": {string(metadata)},
+		"selector": {`{"type":"none","contextType":"NONE"}`},
+	}
+	method := http.MethodPost
+	if exists {
+		method = http.MethodPut
+	}
+	return target.operation(ctx, method, target.servicePath, nil, form, false)
+}
+
+func (target *nacosTarget) serviceStatus(ctx context.Context, query url.Values) (bool, bool, error) {
+	body, status, err := target.rawRequest(ctx, http.MethodGet, target.servicePath, query, nil, true)
+	if err != nil {
+		if responseMeansMissing([]byte(err.Error())) {
+			return false, false, nil
+		}
+		return false, false, err
+	}
+	if status == http.StatusNotFound || responseMeansMissing(body) {
+		return false, false, nil
+	}
+	if err := operationResponseError(body); err != nil {
+		if responseMeansMissing([]byte(err.Error())) {
+			return false, false, nil
+		}
+		return false, false, err
+	}
+	return true, serviceOwnedBySeed(body), nil
+}
+
+func serviceOwnedBySeed(body []byte) bool {
+	var payload any
+	if err := json.Unmarshal(body, &payload); err != nil {
+		return false
+	}
+	metadata := serviceMetadata(payload)
+	return metadata != nil && metadata["source"] == serviceOwnerSource
+}
+
+func serviceMetadata(payload any) map[string]any {
+	value, ok := payload.(map[string]any)
+	if !ok {
+		return nil
+	}
+	if metadata, exists := value["metadata"]; exists {
+		switch metadata := metadata.(type) {
+		case map[string]any:
+			return metadata
+		case string:
+			var parsed map[string]any
+			if json.Unmarshal([]byte(metadata), &parsed) == nil {
+				return parsed
+			}
+		}
+	}
+	return serviceMetadata(value["data"])
+}
+
+func (target *nacosTarget) cleanup(ctx context.Context, opts options) error {
+	var failures []string
+	for serviceIndex := opts.serviceCount - 1; serviceIndex >= 0; serviceIndex-- {
+		serviceName := generatedServiceName(opts.prefix, serviceIndex)
+		query := url.Values{"namespaceId": {opts.namespace}, "groupName": {opts.group}, "serviceName": {serviceName}}
+		exists, owned, err := target.serviceStatus(ctx, query)
+		if err != nil {
+			failures = append(failures, fmt.Sprintf("inspect %s: %v", serviceName, err))
+			continue
+		}
+		if !exists {
+			continue
+		}
+		if !owned && !opts.forceExisting {
+			failures = append(failures, fmt.Sprintf("refusing to remove %s because it does not have the script ownership marker; pass -force-existing to override", serviceName))
+			continue
+		}
+		instances, err := target.listInstances(ctx, opts, serviceName)
+		if err != nil {
+			failures = append(failures, fmt.Sprintf("list %s: %v", serviceName, err))
+			continue
+		}
+		for _, instance := range instances {
+			query := url.Values{
+				"namespaceId": {opts.namespace}, "groupName": {opts.group}, "serviceName": {serviceName},
+				"clusterName": {instance.ClusterName}, "ip": {instance.IP},
+				"port": {strconv.Itoa(instance.Port)}, "ephemeral": {strconv.FormatBool(instance.Ephemeral)},
+			}
+			if err := target.operation(ctx, http.MethodDelete, target.instancePath, query, nil, true); err != nil {
+				failures = append(failures, fmt.Sprintf("remove %s %s:%d: %v", serviceName, instance.IP, instance.Port, err))
+			}
+		}
+		if err := target.operation(ctx, http.MethodDelete, target.servicePath, query, nil, true); err != nil {
+			failures = append(failures, err.Error())
+		} else {
+			fmt.Printf("[%s] removed %s\n", target.name, serviceName)
+		}
+	}
+	if len(failures) > 0 {
+		return fmt.Errorf("cleanup completed with %d errors: %s", len(failures), strings.Join(failures, "; "))
+	}
+	return nil
+}
+
+func (target *nacosTarget) listInstances(ctx context.Context, opts options, serviceName string) ([]nacosInstance, error) {
+	if target.instanceListCatalog {
+		return target.listCatalogInstances(ctx, opts, serviceName)
+	}
+	query := url.Values{
+		"namespaceId": {opts.namespace}, "groupName": {opts.group}, "serviceName": {serviceName}, "clusterName": {opts.cluster},
+	}
+	body, _, err := target.rawRequest(ctx, http.MethodGet, target.instanceListPath, query, nil, true)
+	if err != nil {
+		if responseMeansMissing([]byte(err.Error())) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	if err := operationResponseError(body); err != nil {
+		if responseMeansMissing([]byte(err.Error())) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	instances, err := parseInstances(body)
+	if err != nil {
+		return nil, err
+	}
+	return instances, nil
+}
+
+func (target *nacosTarget) listCatalogInstances(ctx context.Context, opts options, serviceName string) ([]nacosInstance, error) {
+	qualifiedServiceName := serviceName
+	if !strings.Contains(serviceName, "@@") && opts.group != "" {
+		qualifiedServiceName = opts.group + "@@" + serviceName
+	}
+	instances := make([]nacosInstance, 0)
+	seen := make(map[string]struct{})
+	for pageNo := 1; ; pageNo++ {
+		query := url.Values{
+			"namespaceId": {opts.namespace}, "serviceName": {qualifiedServiceName}, "clusterName": {opts.cluster},
+			"pageNo": {strconv.Itoa(pageNo)}, "pageSize": {strconv.Itoa(catalogPageSize)},
+		}
+		body, _, err := target.rawRequest(ctx, http.MethodGet, target.instanceListPath, query, nil, true)
+		if err != nil {
+			if responseMeansMissing([]byte(err.Error())) {
+				return instances, nil
+			}
+			return nil, err
+		}
+		if err := operationResponseError(body); err != nil {
+			if responseMeansMissing([]byte(err.Error())) {
+				return instances, nil
+			}
+			return nil, err
+		}
+		page, total, err := parseInstancePage(body)
+		if err != nil {
+			return nil, err
+		}
+		added := 0
+		for _, instance := range page {
+			key := fmt.Sprintf("%s\x00%d\x00%s\x00%t", instance.IP, instance.Port, instance.ClusterName, instance.Ephemeral)
+			if _, exists := seen[key]; exists {
+				continue
+			}
+			seen[key] = struct{}{}
+			instances = append(instances, instance)
+			added++
+		}
+		if total != nil {
+			if len(instances) >= *total || len(page) == 0 {
+				break
+			}
+		} else if len(page) < catalogPageSize {
+			break
+		}
+		if added == 0 {
+			return nil, errors.New("Nacos instance pagination made no progress; the server repeated a page")
+		}
+	}
+	return instances, nil
+}
+
+func parseInstances(body []byte) ([]nacosInstance, error) {
+	instances, _, err := parseInstancePage(body)
+	return instances, err
+}
+
+func parseInstancePage(body []byte) ([]nacosInstance, *int, error) {
+	var payload any
+	if err := json.Unmarshal(body, &payload); err != nil {
+		return nil, nil, fmt.Errorf("invalid instance-list response: %s", compactBody(body))
+	}
+	items := instanceItems(payload)
+	instances := make([]nacosInstance, 0, len(items))
+	for _, item := range items {
+		ip, _ := item["ip"].(string)
+		port := numberValue(item["port"])
+		if ip == "" || port < 1 || port > 65535 {
+			continue
+		}
+		clusterName, _ := item["clusterName"].(string)
+		if clusterName == "" {
+			clusterName = "DEFAULT"
+		}
+		ephemeral, _ := item["ephemeral"].(bool)
+		instances = append(instances, nacosInstance{IP: ip, Port: port, ClusterName: clusterName, Ephemeral: ephemeral})
+	}
+	total, ok := instanceTotal(payload)
+	if !ok {
+		return instances, nil, nil
+	}
+	return instances, &total, nil
+}
+
+func instanceItems(payload any) []map[string]any {
+	switch value := payload.(type) {
+	case []any:
+		items := make([]map[string]any, 0, len(value))
+		for _, item := range value {
+			if instance, ok := item.(map[string]any); ok && instance["ip"] != nil {
+				items = append(items, instance)
+			}
+		}
+		return items
+	case map[string]any:
+		for _, key := range []string{"hosts", "list", "pageItems", "data"} {
+			if items := instanceItems(value[key]); len(items) > 0 {
+				return items
+			}
+		}
+	}
+	return nil
+}
+
+func instanceTotal(payload any) (int, bool) {
+	value, ok := payload.(map[string]any)
+	if !ok {
+		return 0, false
+	}
+	for _, key := range []string{"totalCount", "count"} {
+		if raw, exists := value[key]; exists {
+			if total, valid := nonnegativeNumberValue(raw); valid {
+				return total, true
+			}
+		}
+	}
+	return instanceTotal(value["data"])
+}
+
+func nonnegativeNumberValue(value any) (int, bool) {
+	switch value := value.(type) {
+	case float64:
+		if value < 0 {
+			return 0, false
+		}
+		return int(value), true
+	case string:
+		parsed, err := strconv.Atoi(value)
+		return parsed, err == nil && parsed >= 0
+	default:
+		return 0, false
+	}
+}
+
+func numberValue(value any) int {
+	switch value := value.(type) {
+	case float64:
+		return int(value)
+	case string:
+		parsed, _ := strconv.Atoi(value)
+		return parsed
+	default:
+		return 0
+	}
+}
+
+func generatedServiceName(prefix string, index int) string {
+	return fmt.Sprintf("%s-%02d", prefix, index+1)
+}
+
+func generatedPort(opts options, serviceIndex, instanceIndex int) int {
+	return opts.instanceBasePort + serviceIndex*opts.instancesPerSvc + instanceIndex
+}
+
+func (target *nacosTarget) operation(ctx context.Context, method, path string, query, form url.Values, allowMissing bool) error {
+	body, _, err := target.rawRequest(ctx, method, path, query, form, true)
+	if err != nil {
+		if allowMissing && responseMeansMissing([]byte(err.Error())) {
+			return nil
+		}
+		return err
+	}
+	if err := operationResponseError(body); err != nil {
+		if allowMissing && responseMeansMissing([]byte(err.Error())) {
+			return nil
+		}
+		return err
+	}
+	return nil
+}
+
+func (target *nacosTarget) rawRequest(ctx context.Context, method, path string, query, form url.Values, authenticate bool) ([]byte, int, error) {
+	endpoint, err := url.Parse(target.baseURL + path)
+	if err != nil {
+		return nil, 0, err
+	}
+	values := endpoint.Query()
+	for key, entries := range query {
+		for _, value := range entries {
+			values.Add(key, value)
+		}
+	}
+	if authenticate && target.token != "" && !target.tokenInHeader {
+		values.Set("accessToken", target.token)
+	}
+	endpoint.RawQuery = values.Encode()
+
+	var body io.Reader
+	if form != nil {
+		body = strings.NewReader(form.Encode())
+	}
+	req, err := http.NewRequestWithContext(ctx, method, endpoint.String(), body)
+	if err != nil {
+		return nil, 0, err
+	}
+	if form != nil {
+		req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	}
+	if authenticate && target.token != "" && target.tokenInHeader {
+		req.Header.Set("accessToken", target.token)
+	}
+	resp, err := target.client.Do(req)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer resp.Body.Close()
+	responseBody, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
+	if err != nil {
+		return nil, resp.StatusCode, err
+	}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return responseBody, resp.StatusCode, fmt.Errorf("%s %s returned %s: %s", method, path, resp.Status, compactBody(responseBody))
+	}
+	return responseBody, resp.StatusCode, nil
+}
+
+func accessToken(body []byte) (string, error) {
+	var payload map[string]any
+	if err := json.Unmarshal(body, &payload); err != nil {
+		return "", fmt.Errorf("invalid login response: %s", compactBody(body))
+	}
+	if data, ok := payload["data"].(map[string]any); ok {
+		payload = data
+	}
+	for _, key := range []string{"accessToken", "access_token", "token"} {
+		if token, ok := payload[key].(string); ok && token != "" {
+			return token, nil
+		}
+	}
+	return "", fmt.Errorf("login response did not include an access token: %s", compactBody(body))
+}
+
+func operationResponseError(body []byte) error {
+	trimmed := strings.TrimSpace(string(body))
+	if trimmed == "" || strings.EqualFold(trimmed, "true") || strings.EqualFold(trimmed, "ok") {
+		return nil
+	}
+	if strings.EqualFold(trimmed, "false") {
+		return errors.New("Nacos returned false")
+	}
+	var payload map[string]any
+	if err := json.Unmarshal(body, &payload); err != nil {
+		return nil
+	}
+	message := responseMessage(payload)
+	if code, ok := payload["code"]; ok && !successfulCode(code) {
+		return fmt.Errorf("Nacos returned code %v: %s", code, message)
+	}
+	if success, ok := payload["success"].(bool); ok && !success {
+		return fmt.Errorf("Nacos operation failed: %s", message)
+	}
+	if data, ok := payload["data"].(bool); ok && !data {
+		return fmt.Errorf("Nacos operation returned false: %s", message)
+	}
+	return nil
+}
+
+func successfulCode(code any) bool {
+	switch value := code.(type) {
+	case float64:
+		return value == 0 || value == 200
+	case string:
+		return value == "0" || value == "200"
+	default:
+		return false
+	}
+}
+
+func responseMessage(payload map[string]any) string {
+	for _, key := range []string{"message", "msg", "error"} {
+		if value, ok := payload[key].(string); ok && value != "" {
+			return value
+		}
+	}
+	return "unknown error"
+}
+
+func responseMeansMissing(body []byte) bool {
+	lower := strings.ToLower(string(body))
+	return strings.Contains(lower, "not found") || strings.Contains(lower, "not exist") || strings.Contains(lower, "doesn't exist")
+}
+
+func compactBody(body []byte) string {
+	text := strings.TrimSpace(string(body))
+	if len(text) > 400 {
+		return text[:400] + "..."
+	}
+	return text
+}
@@ -0,0 +1,374 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+)
+
+func TestAccessTokenSupportsV2AndV3Responses(t *testing.T) {
+	for name, body := range map[string]string{
+		"v2": `{"accessToken":"v2-token","tokenTtl":18000}`,
+		"v3": `{"code":0,"message":"success","data":{"accessToken":"v3-token","tokenTtl":18000}}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			token, err := accessToken([]byte(body))
+			if err != nil {
+				t.Fatal(err)
+			}
+			if token != name+"-token" {
+				t.Fatalf("unexpected token %q", token)
+			}
+		})
+	}
+}
+
+func TestOperationResponseErrorRejectsBusinessFailures(t *testing.T) {
+	for _, body := range []string{
+		`false`,
+		`{"code":403,"message":"authorization failed"}`,
+		`{"code":0,"message":"failed","data":false}`,
+	} {
+		if err := operationResponseError([]byte(body)); err == nil {
+			t.Fatalf("expected %s to fail", body)
+		}
+	}
+	for _, body := range []string{`true`, `{"code":0,"message":"success","data":true}`, `{"code":200,"data":"ok"}`} {
+		if err := operationResponseError([]byte(body)); err != nil {
+			t.Fatalf("expected %s to succeed: %v", body, err)
+		}
+	}
+}
+
+func TestSelectedTargetsUseRepositoryPortsAndVersionedAPIs(t *testing.T) {
+	opts := options{target: "both", v2URL: "http://127.0.0.1:11000/nacos", v3URL: "http://127.0.0.1:11003/nacos", timeout: 1}
+	targets := selectedTargets(opts)
+	if len(targets) != 2 {
+		t.Fatalf("expected two targets, got %d", len(targets))
+	}
+	if !strings.Contains(targets[0].servicePath, "/v1/") || targets[0].tokenInHeader {
+		t.Fatalf("unexpected V2 target: %+v", targets[0])
+	}
+	if !strings.Contains(targets[1].servicePath, "/v3/admin/") || !targets[1].tokenInHeader {
+		t.Fatalf("unexpected V3 target: %+v", targets[1])
+	}
+	if !targets[0].supportsHealthUpdate || targets[0].instanceHealthPath != "/v1/ns/instance" || targets[0].instanceListPath != "/v1/ns/catalog/instances" || !targets[0].instanceListCatalog || targets[1].supportsHealthUpdate || targets[1].instanceListPath != "/v3/admin/ns/instance/list" || targets[1].instanceListCatalog {
+		t.Fatalf("unexpected instance health-update support: %+v", targets)
+	}
+}
+
+func TestServiceOwnedBySeedSupportsWrappedAndStringMetadata(t *testing.T) {
+	for name, body := range map[string]string{
+		"wrapped object":   `{"code":0,"data":{"metadata":{"source":"dbx-nacos-service-seed"}}}`,
+		"top-level string": `{"metadata":"{\"source\":\"dbx-nacos-service-seed\"}"}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			if !serviceOwnedBySeed([]byte(body)) {
+				t.Fatalf("expected script ownership for %s", body)
+			}
+		})
+	}
+	for _, body := range []string{
+		`{"code":0,"data":{"metadata":{"source":"another-tool"}}}`,
+		`{"code":0,"data":{"metadata":{}}}`,
+		`{"code":0,"data":true}`,
+	} {
+		if serviceOwnedBySeed([]byte(body)) {
+			t.Fatalf("unexpected script ownership for %s", body)
+		}
+	}
+}
+
+func TestSeedRefusesToReplaceUnownedServiceUnlessForced(t *testing.T) {
+	test := func(force bool) ([]string, error) {
+		var requests []string
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			requests = append(requests, r.Method+" "+r.URL.Path)
+			if r.Method == http.MethodGet {
+				_, _ = w.Write([]byte(`{"code":0,"data":{"metadata":{"source":"another-tool"}}}`))
+				return
+			}
+			_, _ = w.Write([]byte(`{"code":0,"data":true}`))
+		}))
+		defer server.Close()
+
+		target := &nacosTarget{baseURL: server.URL, servicePath: "/v3/admin/ns/service", instancePath: "/v3/admin/ns/instance", client: server.Client(), tokenInHeader: true}
+		opts := options{
+			namespace: "public", group: "DEFAULT_GROUP", cluster: "DEFAULT", prefix: "dbx-demo-service",
+			serviceCount: 1, instancesPerSvc: 1, instanceIP: "127.0.0.1", instanceBasePort: 28080, forceExisting: force,
+		}
+		return requests, target.seed(context.Background(), opts)
+	}
+
+	requests, err := test(false)
+	if err == nil || !strings.Contains(err.Error(), "-force-existing") {
+		t.Fatalf("expected ownership error, got %v", err)
+	}
+	if len(requests) != 1 || requests[0] != "GET /v3/admin/ns/service" {
+		t.Fatalf("unowned service was modified: %#v", requests)
+	}
+
+	requests, err = test(true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(requests) != 3 || requests[1] != "PUT /v3/admin/ns/service" || requests[2] != "POST /v3/admin/ns/instance" {
+		t.Fatalf("forced seed did not replace and register: %#v", requests)
+	}
+}
+
+func TestCleanupRefusesToDeleteUnownedServiceUnlessForced(t *testing.T) {
+	test := func(force bool) ([]string, error) {
+		var requests []string
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			requests = append(requests, r.Method+" "+r.URL.Path)
+			switch {
+			case r.Method == http.MethodGet && r.URL.Path == "/v3/admin/ns/service":
+				_, _ = w.Write([]byte(`{"code":0,"data":{"metadata":{"source":"another-tool"}}}`))
+			case r.Method == http.MethodGet:
+				_, _ = w.Write([]byte(`{"code":0,"data":[]}`))
+			default:
+				_, _ = w.Write([]byte(`{"code":0,"data":true}`))
+			}
+		}))
+		defer server.Close()
+
+		target := &nacosTarget{
+			baseURL: server.URL, servicePath: "/v3/admin/ns/service", instancePath: "/v3/admin/ns/instance", instanceListPath: "/v3/admin/ns/instance/list",
+			client: server.Client(), tokenInHeader: true,
+		}
+		opts := options{namespace: "public", group: "DEFAULT_GROUP", cluster: "DEFAULT", prefix: "dbx-demo-service", serviceCount: 1, instancesPerSvc: 1, forceExisting: force}
+		return requests, target.cleanup(context.Background(), opts)
+	}
+
+	requests, err := test(false)
+	if err == nil || !strings.Contains(err.Error(), "-force-existing") {
+		t.Fatalf("expected ownership error, got %v", err)
+	}
+	if len(requests) != 1 || requests[0] != "GET /v3/admin/ns/service" {
+		t.Fatalf("unowned service was inspected beyond its ownership marker: %#v", requests)
+	}
+
+	requests, err = test(true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(requests) != 3 || requests[1] != "GET /v3/admin/ns/instance/list" || requests[2] != "DELETE /v3/admin/ns/service" {
+		t.Fatalf("forced cleanup did not remove the service: %#v", requests)
+	}
+}
+
+func TestCleanupUsesActualInstanceIdentities(t *testing.T) {
+	type request struct {
+		method string
+		path   string
+		query  url.Values
+	}
+	var requests []request
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests = append(requests, request{method: r.Method, path: r.URL.Path, query: r.URL.Query()})
+		if r.Method == http.MethodGet && r.URL.Path == "/v3/admin/ns/service" {
+			_, _ = w.Write([]byte(`{"code":0,"data":{"metadata":{"source":"dbx-nacos-service-seed"}}}`))
+			return
+		}
+		if r.Method == http.MethodGet && r.URL.Path == "/v3/admin/ns/instance/list" {
+			_, _ = w.Write([]byte(`{"code":0,"data":[{"ip":"10.91.0.99","port":28080,"clusterName":"blue","ephemeral":false}]}`))
+			return
+		}
+		_, _ = w.Write([]byte(`{"code":0,"data":true}`))
+	}))
+	defer server.Close()
+
+	target := &nacosTarget{
+		baseURL: server.URL, servicePath: "/v3/admin/ns/service", instancePath: "/v3/admin/ns/instance", instanceListPath: "/v3/admin/ns/instance/list",
+		client: server.Client(), tokenInHeader: true,
+	}
+	opts := options{namespace: "public", group: "DEFAULT_GROUP", cluster: "DEFAULT", prefix: "dbx-demo-service", serviceCount: 1, instancesPerSvc: 3}
+	if err := target.cleanup(context.Background(), opts); err != nil {
+		t.Fatal(err)
+	}
+	if len(requests) != 4 {
+		t.Fatalf("expected ownership lookup, list, instance delete, and service delete; got %#v", requests)
+	}
+	instanceDelete := requests[2]
+	if instanceDelete.method != http.MethodDelete || instanceDelete.path != "/v3/admin/ns/instance" || instanceDelete.query.Get("ip") != "10.91.0.99" || instanceDelete.query.Get("port") != "28080" || instanceDelete.query.Get("clusterName") != "blue" {
+		t.Fatalf("unexpected instance cleanup request: %#v", instanceDelete)
+	}
+}
+
+func TestV2CleanupUsesPaginatedCatalogAndIncludesDisabledInstances(t *testing.T) {
+	type request struct {
+		method string
+		path   string
+		query  url.Values
+	}
+	var requests []request
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests = append(requests, request{method: r.Method, path: r.URL.Path, query: r.URL.Query()})
+		switch {
+		case r.Method == http.MethodGet && r.URL.Path == "/v1/ns/service":
+			_, _ = w.Write([]byte(`{"metadata":{"source":"dbx-nacos-service-seed"}}`))
+		case r.Method == http.MethodGet && r.URL.Path == "/v1/ns/catalog/instances" && r.URL.Query().Get("pageNo") == "1":
+			_, _ = w.Write([]byte(`{"list":[{"ip":"127.0.0.1","port":28080,"clusterName":"manual","enabled":false,"ephemeral":false}],"count":2}`))
+		case r.Method == http.MethodGet && r.URL.Path == "/v1/ns/catalog/instances" && r.URL.Query().Get("pageNo") == "2":
+			_, _ = w.Write([]byte(`{"list":[{"ip":"127.0.0.1","port":28081,"clusterName":"manual","enabled":true,"ephemeral":false}],"count":2}`))
+		default:
+			_, _ = w.Write([]byte("true"))
+		}
+	}))
+	defer server.Close()
+
+	target := &nacosTarget{
+		baseURL: server.URL, servicePath: "/v1/ns/service", instancePath: "/v1/ns/instance", instanceListPath: "/v1/ns/catalog/instances",
+		instanceListCatalog: true, client: server.Client(),
+	}
+	opts := options{namespace: "public", group: "DEFAULT_GROUP", cluster: "manual", prefix: "dbx-demo-service", serviceCount: 1, instancesPerSvc: 2}
+	if err := target.cleanup(context.Background(), opts); err != nil {
+		t.Fatal(err)
+	}
+	if len(requests) != 6 {
+		t.Fatalf("expected ownership lookup, two catalog pages, two instance deletes, and service delete; got %#v", requests)
+	}
+	for index, pageNo := range []string{"1", "2"} {
+		catalog := requests[index+1]
+		if catalog.path != "/v1/ns/catalog/instances" || catalog.query.Get("serviceName") != "DEFAULT_GROUP@@dbx-demo-service-01" || catalog.query.Get("groupName") != "" || catalog.query.Get("clusterName") != "manual" || catalog.query.Get("pageNo") != pageNo || catalog.query.Get("pageSize") != "100" {
+			t.Fatalf("unexpected catalog request: %#v", catalog)
+		}
+	}
+	deletedPorts := []string{requests[3].query.Get("port"), requests[4].query.Get("port")}
+	if strings.Join(deletedPorts, ",") != "28080,28081" {
+		t.Fatalf("disabled or paginated instance was not deleted: %#v", requests)
+	}
+}
+
+func TestSeedMarksPersistentInstancesHealthyAfterRegistration(t *testing.T) {
+	type request struct {
+		method string
+		path   string
+		form   url.Values
+	}
+	var requests []request
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		form, _ := url.ParseQuery(string(body))
+		requests = append(requests, request{method: r.Method, path: r.URL.Path, form: form})
+		switch {
+		case r.Method == http.MethodGet && r.URL.Path == "/v1/ns/service":
+			http.NotFound(w, r)
+		default:
+			_, _ = w.Write([]byte("true"))
+		}
+	}))
+	defer server.Close()
+
+	target := &nacosTarget{
+		baseURL: server.URL, servicePath: "/v1/ns/service", instancePath: "/v1/ns/instance", instanceHealthPath: "/v1/ns/instance",
+		supportsHealthUpdate: true, client: server.Client(),
+	}
+	opts := options{
+		namespace: "public", group: "DEFAULT_GROUP", cluster: "DEFAULT", prefix: "dbx-demo-service",
+		serviceCount: 1, instancesPerSvc: 1, instanceIP: "127.0.0.1", instanceBasePort: 28080,
+	}
+	if err := target.seed(context.Background(), opts); err != nil {
+		t.Fatal(err)
+	}
+	if len(requests) != 4 {
+		t.Fatalf("expected service lookup, creation, registration, and health update; got %#v", requests)
+	}
+	registration, healthUpdate := requests[2], requests[3]
+	if registration.method != http.MethodPost || registration.path != "/v1/ns/instance" || registration.form.Get("ephemeral") != "false" {
+		t.Fatalf("unexpected registration request: %#v", registration)
+	}
+	if healthUpdate.method != http.MethodPut || healthUpdate.path != "/v1/ns/instance" || healthUpdate.form.Get("healthy") != "true" || healthUpdate.form.Get("ephemeral") != "false" {
+		t.Fatalf("unexpected health update request: %#v", healthUpdate)
+	}
+}
+
+func TestV3SeedSetsHealthDuringRegistrationWithoutAHealthUpdate(t *testing.T) {
+	type request struct {
+		method string
+		path   string
+		form   url.Values
+	}
+	var requests []request
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		form, _ := url.ParseQuery(string(body))
+		requests = append(requests, request{method: r.Method, path: r.URL.Path, form: form})
+		if r.Method == http.MethodGet && r.URL.Path == "/v3/admin/ns/service" {
+			http.NotFound(w, r)
+			return
+		}
+		_, _ = w.Write([]byte(`{"code":0,"data":true}`))
+	}))
+	defer server.Close()
+
+	target := &nacosTarget{
+		baseURL: server.URL, servicePath: "/v3/admin/ns/service", instancePath: "/v3/admin/ns/instance",
+		client: server.Client(), tokenInHeader: true,
+	}
+	opts := options{
+		namespace: "public", group: "DEFAULT_GROUP", cluster: "DEFAULT", prefix: "dbx-demo-service",
+		serviceCount: 1, instancesPerSvc: 1, instanceIP: "127.0.0.1", instanceBasePort: 28080,
+	}
+	if err := target.seed(context.Background(), opts); err != nil {
+		t.Fatal(err)
+	}
+	if len(requests) != 3 {
+		t.Fatalf("expected service lookup, creation, and registration; got %#v", requests)
+	}
+	registration := requests[2]
+	if registration.method != http.MethodPost || registration.path != "/v3/admin/ns/instance" || registration.form.Get("healthy") != "true" {
+		t.Fatalf("unexpected V3 registration request: %#v", registration)
+	}
+}
+
+func TestServeStartsHTTPServerForEveryGeneratedPort(t *testing.T) {
+	opts := options{
+		prefix: "dbx-demo-service", serviceCount: 1, instancesPerSvc: 1,
+		listenAddress: "127.0.0.1", instanceBasePort: 0,
+	}
+	servers, err := startDemoServers(opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := servers.shutdown(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+	}()
+	if len(servers.addresses) != 1 {
+		t.Fatalf("expected one listener, got %v", servers.addresses)
+	}
+
+	response, err := http.Get("http://" + servers.addresses[0] + "/health")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		t.Fatalf("unexpected status: %s", response.Status)
+	}
+	var payload map[string]any
+	if err := json.NewDecoder(response.Body).Decode(&payload); err != nil {
+		t.Fatal(err)
+	}
+	if payload["service"] != "dbx-demo-service-01" || payload["status"] != "healthy" {
+		t.Fatalf("unexpected health response: %#v", payload)
+	}
+}
+
+func TestValidateOptionsRejectsServeDuringCleanup(t *testing.T) {
+	err := validateOptions(options{
+		target: "v2", action: "cleanup", username: "nacos", password: "123456", namespace: "public", group: "DEFAULT_GROUP", cluster: "DEFAULT", prefix: "dbx-demo-service",
+		serviceCount: 1, instancesPerSvc: 1, instanceBasePort: 28080, v2URL: "http://127.0.0.1:11000/nacos", v3URL: "http://127.0.0.1:11003/nacos", serve: true,
+	})
+	if err == nil || !strings.Contains(err.Error(), "-serve") {
+		t.Fatalf("expected -serve validation error, got %v", err)
+	}
+}