Add data-exchange network policy and multi-network stream fix

- Fix SYN-ACK/RST source address for multi-network connections:
  use pkt.Dst instead of d.Addr() so responses match the network-
  specific address the connection was opened with
- Fix policy runner bootstrap: always refresh peers from registry
  on startup, not just for policies with cycle rules
- Add admin-token registry paths for pilotctl network join --node-id,
  member-tags set, and policy set
- Add data-exchange-policy.json: service-node gated connectivity
  with text messaging (port 1000) open and file transfer (port 1001)
  restricted to service-tagged nodes
- Add integration test (6 subtests) covering connect, dial, datagram
  policy enforcement

Author: teovl

cmd/pilotctl/main.go

@@ -4384,15 +4384,30 @@ func cmdNetworkList() {
 
 func cmdNetworkJoin(args []string) {
 	if len(args) < 1 {
-		fatalCode("invalid_argument", "usage: pilotctl network join <network_id> [--token TOKEN]")
+		fatalCode("invalid_argument", "usage: pilotctl network join <network_id> [--token TOKEN] [--node-id N]")
 	}
 	netID := parseUint16(args[0], "network_id")
-	token := ""
-	for i := 1; i < len(args)-1; i++ {
-		if args[i] == "--token" {
-			token = args[i+1]
-			break
+	flags, _ := parseFlags(args[1:])
+	token := flagString(flags, "token", "")
+	nodeIDStr := flagString(flags, "node-id", "")
+
+	// Admin path: --node-id joins a remote node directly via registry
+	if nodeIDStr != "" {
+		nodeID := parseNodeID(nodeIDStr)
+		adminToken := requireAdminToken()
+		rc := connectRegistry()
+		defer rc.Close()
+
+		result, err := rc.JoinNetwork(nodeID, netID, token, 0, adminToken)
+		if err != nil {
+			fatalCode("connection_failed", "network join: %v", err)
 		}
+		if jsonOutput {
+			output(result)
+		} else {
+			fmt.Printf("joined node %d to network %d\n", nodeID, netID)
+		}
+		return
 	}
 
 	d := connectDriver()
@@ -5303,7 +5318,11 @@ func cmdPolicySet(args []string) {
 	reg := connectRegistry()
 	defer reg.Close()
 
-	_, err = reg.SetExprPolicy(netID, policyJSON, flagString(flags, "admin-token", ""))
+	adminToken := flagString(flags, "admin-token", "")
+	if adminToken == "" {
+		adminToken = getAdminToken()
+	}
+	_, err = reg.SetExprPolicy(netID, policyJSON, adminToken)
 	if err != nil {
 		fatalCode("connection_failed", "set policy on registry: %v", err)
 	}
@@ -5488,6 +5507,23 @@ func cmdMemberTagsSet(args []string) {
 
 	tags := strings.Split(tagsStr, ",")
 
+	// If admin token is available, go directly to registry (no daemon needed)
+	if adminToken := getAdminToken(); adminToken != "" {
+		rc := connectRegistry()
+		defer rc.Close()
+
+		result, err := rc.SetMemberTags(netID, uint32(nid), tags, adminToken)
+		if err != nil {
+			fatalCode("connection_failed", "member-tags set: %v", err)
+		}
+		if jsonOutput {
+			output(result)
+			return
+		}
+		fmt.Printf("Member tags set for node %d in network %d: %s\n", uint32(nid), netID, strings.Join(tags, ", "))
+		return
+	}
+
 	d := connectDriver()
 	defer d.Close()
 

data-exchange-policy.json

@@ -0,0 +1,47 @@
+{
+  "version": 1,
+  "rules": [
+    {
+      "name": "service-connect",
+      "on": "connect",
+      "match": "has_tag(local_tags, \"service\") || has_tag(peer_tags, \"service\")",
+      "actions": [{"type": "allow"}]
+    },
+    {
+      "name": "service-dial",
+      "on": "dial",
+      "match": "has_tag(local_tags, \"service\") || has_tag(peer_tags, \"service\")",
+      "actions": [{"type": "allow"}]
+    },
+    {
+      "name": "block-peer-connect",
+      "on": "connect",
+      "match": "true",
+      "actions": [{"type": "deny"}]
+    },
+    {
+      "name": "block-peer-dial",
+      "on": "dial",
+      "match": "true",
+      "actions": [{"type": "deny"}]
+    },
+    {
+      "name": "allow-text",
+      "on": "datagram",
+      "match": "port == 1000",
+      "actions": [{"type": "allow"}]
+    },
+    {
+      "name": "allow-service-files",
+      "on": "datagram",
+      "match": "port == 1001 && ((direction == \"out\" && has_tag(local_tags, \"service\")) || (direction == \"in\" && has_tag(peer_tags, \"service\")))",
+      "actions": [{"type": "allow"}]
+    },
+    {
+      "name": "deny-other-data",
+      "on": "datagram",
+      "match": "true",
+      "actions": [{"type": "deny"}]
+    }
+  ]
+}

pkg/daemon/daemon.go

@@ -1323,7 +1323,7 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
 				Version:  protocol.Version,
 				Flags:    protocol.FlagSYN | protocol.FlagACK,
 				Protocol: protocol.ProtoStream,
-				Src:      d.Addr(),
+				Src:      pkt.Dst,
 				Dst:      pkt.Src,
 				SrcPort:  pkt.DstPort,
 				DstPort:  pkt.SrcPort,
@@ -1396,7 +1396,11 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
 
 		conn := d.ports.NewConnection(pkt.DstPort, pkt.Src, pkt.SrcPort)
 		conn.Mu.Lock()
-		conn.LocalAddr = d.Addr()
+		// Use the destination address from the SYN as our local address.
+		// This ensures the correct network-specific address is used for
+		// multi-network connections (e.g. 1:0001.0000.0003 instead of
+		// the primary 0:0000.0000.0003).
+		conn.LocalAddr = pkt.Dst
 		conn.State = StateSynReceived
 		conn.RecvAck = pkt.Seq + 1
 		conn.ExpectedSeq = pkt.Seq + 1 // first data segment after SYN
@@ -1417,7 +1421,7 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
 			Version:  protocol.Version,
 			Flags:    protocol.FlagSYN | protocol.FlagACK,
 			Protocol: protocol.ProtoStream,
-			Src:      d.Addr(),
+			Src:      pkt.Dst,
 			Dst:      pkt.Src,
 			SrcPort:  pkt.DstPort,
 			DstPort:  pkt.SrcPort,
@@ -1727,7 +1731,7 @@ func (d *Daemon) sendRST(orig *protocol.Packet) {
 		Version:  protocol.Version,
 		Flags:    protocol.FlagRST,
 		Protocol: protocol.ProtoStream,
-		Src:      d.Addr(),
+		Src:      orig.Dst,
 		Dst:      orig.Src,
 		SrcPort:  orig.DstPort,
 		DstPort:  orig.SrcPort,

pkg/daemon/policy_runner.go

@@ -353,16 +353,11 @@ func (pr *PolicyRunner) executeWebhook(d policy.Directive) {
 func (pr *PolicyRunner) cycleLoop() {
 	defer close(pr.done)
 
-	// Bootstrap if no peers and the policy has cycle rules
-	if pr.compiled.HasRulesFor(policy.EventCycle) {
-		pr.mu.RLock()
-		needBootstrap := len(pr.peers) == 0
-		pr.mu.RUnlock()
-		if needBootstrap {
-			if err := pr.bootstrap(); err != nil {
-				slog.Warn("policy: bootstrap failed", "network_id", pr.netID, "err", err)
-			}
-		}
+	// Always bootstrap from registry to refresh peer list and tags.
+	// Persisted state preserves scores/history, but membership and tags
+	// may have changed since last run.
+	if err := pr.bootstrap(); err != nil {
+		slog.Warn("policy: bootstrap failed", "network_id", pr.netID, "err", err)
 	}
 
 	cycleStr, _ := pr.compiled.CycleDuration()