Skip to content

Commit 11e3521

Browse files
teovlteovl
committed
Merge branch 'enterprise/phase-1'
2 parents 2d7346c + d52c256 commit 11e3521

11 files changed

Lines changed: 358 additions & 37 deletions

File tree

‎.gitignore‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,8 @@ web/.wrangler/
9090

9191
# Internal tooling
9292
cmd/spoof/
93+
cmd/pilot-admin/
94+
tests/admin_cli_test.go
9395

9496
# Blog messaging (internal)
9597
web/MESSAGING.md

‎Makefile‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ build:
2020
go build -o $(BINDIR)/pilotctl ./cmd/pilotctl
2121
go build -o $(BINDIR)/nameserver ./cmd/nameserver
2222
go build -o $(BINDIR)/gateway ./cmd/gateway
23+
@test -d cmd/pilot-admin && go build -o $(BINDIR)/pilot-admin ./cmd/pilot-admin || true
2324
go build -o $(BINDIR)/webserver ./examples/go/webserver
2425
go build -o $(BINDIR)/client ./examples/go/client
2526
go build -o $(BINDIR)/echo ./examples/go/echo
@@ -70,6 +71,7 @@ build-linux:
7071
GOOS=linux GOARCH=amd64 go build -o $(BINDIR)/pilotctl-linux ./cmd/pilotctl
7172
GOOS=linux GOARCH=amd64 go build -o $(BINDIR)/nameserver-linux ./cmd/nameserver
7273
GOOS=linux GOARCH=amd64 go build -o $(BINDIR)/gateway-linux ./cmd/gateway
74+
@test -d cmd/pilot-admin && GOOS=linux GOARCH=amd64 go build -o $(BINDIR)/pilot-admin-linux ./cmd/pilot-admin || true
7375
GOOS=linux GOARCH=amd64 go build -o $(BINDIR)/echo-linux ./examples/go/echo
7476
GOOS=linux GOARCH=amd64 go build -o $(BINDIR)/client-linux ./examples/go/client
7577
GOOS=linux GOARCH=amd64 go build -o $(BINDIR)/webserver-linux ./examples/go/webserver

‎cmd/rendezvous/main.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ func main() {
2929
httpAddr := flag.String("http", "", "HTTP dashboard listen address (e.g. :3000)")
3030
logLevel := flag.String("log-level", "info", "log level (debug, info, warn, error)")
3131
logFormat := flag.String("log-format", "text", "log format (text, json)")
32+
adminToken := flag.String("admin-token", "", "admin token for network creation (empty = creation disabled)")
3233
flag.Parse()
3334

3435
if *configPath != "" {
@@ -53,6 +54,9 @@ func main() {
5354

5455
// Start registry
5556
r := registry.NewWithStore(*beaconAddr, *storePath)
57+
if *adminToken != "" {
58+
r.SetAdminToken(*adminToken)
59+
}
5660
if *enableTLS {
5761
if err := r.SetTLS(*tlsCert, *tlsKey); err != nil {
5862
log.Fatalf("TLS setup: %v", err)

‎pkg/daemon/daemon.go‎

Lines changed: 46 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -741,6 +741,26 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
741741
return
742742
}
743743

744+
// Trust gate: private nodes only accept SYN from trusted or same-network peers.
745+
// Runs before rate limiting so untrusted sources cannot waste rate-limit tokens.
746+
if !d.config.Public {
747+
srcNode := pkt.Src.Node
748+
trusted := d.handshakes.IsTrusted(srcNode)
749+
if !trusted && d.regConn != nil {
750+
// Fall back to registry trust check (covers admin-set trust pairs + shared networks)
751+
trusted, _ = d.regConn.CheckTrust(d.NodeID(), srcNode)
752+
}
753+
if !trusted {
754+
slog.Warn("SYN rejected: untrusted source", "src_node", srcNode, "src_addr", pkt.Src, "dst_port", pkt.DstPort)
755+
d.webhook.Emit("syn.rejected", map[string]interface{}{
756+
"src_node_id": srcNode,
757+
"src_addr": pkt.Src.String(),
758+
"dst_port": pkt.DstPort,
759+
})
760+
return // silent drop — no RST to avoid leaking node existence
761+
}
762+
}
763+
744764
// SYN rate limiting
745765
if !d.allowSYN() {
746766
slog.Warn("SYN rate limit exceeded", "src_addr", pkt.Src, "src_port", pkt.SrcPort)
@@ -768,26 +788,6 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
768788
return
769789
}
770790

771-
// Trust gate: private nodes only accept SYN from trusted or same-network peers
772-
if !d.config.Public {
773-
srcNode := pkt.Src.Node
774-
trusted := d.handshakes.IsTrusted(srcNode)
775-
if !trusted && d.regConn != nil {
776-
// Fall back to registry trust check (covers admin-set trust pairs + shared networks)
777-
trusted, _ = d.regConn.CheckTrust(d.NodeID(), srcNode)
778-
}
779-
if !trusted {
780-
slog.Warn("SYN rejected: untrusted source", "src_node", srcNode, "src_addr", pkt.Src, "dst_port", pkt.DstPort)
781-
d.webhook.Emit("syn.rejected", map[string]interface{}{
782-
"src_node_id": srcNode,
783-
"src_addr": pkt.Src.String(),
784-
"dst_port": pkt.DstPort,
785-
})
786-
d.sendRST(pkt)
787-
return
788-
}
789-
}
790-
791791
conn := d.ports.NewConnection(pkt.DstPort, pkt.Src, pkt.SrcPort)
792792
conn.Mu.Lock()
793793
conn.LocalAddr = d.Addr()
@@ -1056,13 +1056,33 @@ func (d *Daemon) sendDelayedACK(conn *Connection) {
10561056
}
10571057

10581058
func (d *Daemon) handleDatagramPacket(pkt *protocol.Packet) {
1059-
if len(pkt.Payload) > 0 {
1060-
d.webhook.Emit("data.datagram", map[string]interface{}{
1061-
"src_addr": pkt.Src.String(), "src_port": pkt.SrcPort,
1062-
"dst_port": pkt.DstPort, "size": len(pkt.Payload),
1063-
})
1064-
d.ipc.DeliverDatagram(pkt.Src, pkt.SrcPort, pkt.DstPort, pkt.Payload)
1059+
if len(pkt.Payload) == 0 {
1060+
return
10651061
}
1062+
1063+
// Trust gate: private nodes only accept datagrams from trusted or same-network peers
1064+
if !d.config.Public {
1065+
srcNode := pkt.Src.Node
1066+
trusted := d.handshakes.IsTrusted(srcNode)
1067+
if !trusted && d.regConn != nil {
1068+
trusted, _ = d.regConn.CheckTrust(d.NodeID(), srcNode)
1069+
}
1070+
if !trusted {
1071+
slog.Warn("datagram rejected: untrusted source", "src_node", srcNode, "src_addr", pkt.Src, "dst_port", pkt.DstPort)
1072+
d.webhook.Emit("datagram.rejected", map[string]interface{}{
1073+
"src_node_id": srcNode,
1074+
"src_addr": pkt.Src.String(),
1075+
"dst_port": pkt.DstPort,
1076+
})
1077+
return
1078+
}
1079+
}
1080+
1081+
d.webhook.Emit("data.datagram", map[string]interface{}{
1082+
"src_addr": pkt.Src.String(), "src_port": pkt.SrcPort,
1083+
"dst_port": pkt.DstPort, "size": len(pkt.Payload),
1084+
})
1085+
d.ipc.DeliverDatagram(pkt.Src, pkt.SrcPort, pkt.DstPort, pkt.Payload)
10661086
}
10671087

10681088
func (d *Daemon) handleControlPacket(pkt *protocol.Packet) {

‎pkg/registry/client.go‎

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -310,17 +310,32 @@ func (c *Client) LeaveNetwork(nodeID uint32, networkID uint16, adminToken string
310310
return c.Send(msg)
311311
}
312312

313+
func (c *Client) DeleteNetwork(networkID uint16, adminToken string) (map[string]interface{}, error) {
314+
msg := map[string]interface{}{
315+
"type": "delete_network",
316+
"network_id": networkID,
317+
}
318+
if adminToken != "" {
319+
msg["admin_token"] = adminToken
320+
}
321+
return c.Send(msg)
322+
}
323+
313324
func (c *Client) ListNetworks() (map[string]interface{}, error) {
314325
return c.Send(map[string]interface{}{
315326
"type": "list_networks",
316327
})
317328
}
318329

319-
func (c *Client) ListNodes(networkID uint16) (map[string]interface{}, error) {
320-
return c.Send(map[string]interface{}{
330+
func (c *Client) ListNodes(networkID uint16, adminToken ...string) (map[string]interface{}, error) {
331+
msg := map[string]interface{}{
321332
"type": "list_nodes",
322333
"network_id": networkID,
323-
})
334+
}
335+
if len(adminToken) > 0 && adminToken[0] != "" {
336+
msg["admin_token"] = adminToken[0]
337+
}
338+
return c.Send(msg)
324339
}
325340

326341
func (c *Client) Deregister(nodeID uint32) (map[string]interface{}, error) {

‎pkg/registry/server.go‎

Lines changed: 76 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -718,6 +718,8 @@ func (s *Server) handleMessage(msg map[string]interface{}, remoteAddr string) (r
718718
return s.handleJoinNetwork(msg)
719719
case "leave_network":
720720
return s.handleLeaveNetwork(msg)
721+
case "delete_network":
722+
return s.handleDeleteNetwork(msg)
721723
case "lookup":
722724
return s.handleLookup(msg)
723725
case "resolve":
@@ -1327,6 +1329,50 @@ func (s *Server) handleLeaveNetwork(msg map[string]interface{}) (map[string]inte
13271329
}, nil
13281330
}
13291331

1332+
func (s *Server) handleDeleteNetwork(msg map[string]interface{}) (map[string]interface{}, error) {
1333+
if err := s.requireAdminToken(msg); err != nil {
1334+
return nil, err
1335+
}
1336+
1337+
netID := jsonUint16(msg, "network_id")
1338+
1339+
// Cannot delete backbone
1340+
if netID == 0 {
1341+
return nil, fmt.Errorf("cannot delete the backbone network")
1342+
}
1343+
1344+
s.mu.Lock()
1345+
defer s.mu.Unlock()
1346+
1347+
network, ok := s.networks[netID]
1348+
if !ok {
1349+
return nil, fmt.Errorf("network %d: %w", netID, protocol.ErrNetworkNotFound)
1350+
}
1351+
1352+
// Remove network from all member nodes
1353+
for _, memberID := range network.Members {
1354+
if node, ok := s.nodes[memberID]; ok {
1355+
for i, n := range node.Networks {
1356+
if n == netID {
1357+
node.Networks = append(node.Networks[:i], node.Networks[i+1:]...)
1358+
break
1359+
}
1360+
}
1361+
}
1362+
}
1363+
1364+
name := network.Name
1365+
delete(s.networks, netID)
1366+
s.save()
1367+
1368+
slog.Info("deleted network", "network_id", netID, "name", name)
1369+
1370+
return map[string]interface{}{
1371+
"type": "delete_network_ok",
1372+
"network_id": netID,
1373+
}, nil
1374+
}
1375+
13301376
func (s *Server) handleLookup(msg map[string]interface{}) (map[string]interface{}, error) {
13311377
nodeID := jsonUint32(msg, "node_id")
13321378

@@ -1941,7 +1987,7 @@ func (s *Server) handleResolveHostname(msg map[string]interface{}) (map[string]i
19411987
}
19421988

19431989
if !allowed {
1944-
return nil, fmt.Errorf("resolve denied: hostname %q belongs to a private node", hostname)
1990+
return nil, fmt.Errorf("hostname %q not found", hostname) // same error as non-existent to prevent enumeration
19451991
}
19461992
}
19471993

@@ -2028,9 +2074,36 @@ func (s *Server) handleListNetworks() (map[string]interface{}, error) {
20282074
func (s *Server) handleListNodes(msg map[string]interface{}) (map[string]interface{}, error) {
20292075
netID := jsonUint16(msg, "network_id")
20302076

2031-
// Backbone (network 0) node listing is restricted to prevent enumeration
2077+
// Backbone (network 0) node listing is restricted to prevent enumeration.
2078+
// Admin token bypasses this restriction.
20322079
if netID == 0 {
2033-
return nil, fmt.Errorf("listing backbone nodes is not permitted (use lookup with a specific node_id)")
2080+
if err := s.requireAdminToken(msg); err != nil {
2081+
return nil, fmt.Errorf("listing backbone nodes is not permitted (use lookup with a specific node_id)")
2082+
}
2083+
// Admin-authenticated: list all registered nodes
2084+
s.mu.RLock()
2085+
defer s.mu.RUnlock()
2086+
nodes := make([]map[string]interface{}, 0, len(s.nodes))
2087+
for _, node := range s.nodes {
2088+
entry := map[string]interface{}{
2089+
"node_id": node.ID,
2090+
"address": protocol.Addr{Network: 0, Node: node.ID}.String(),
2091+
}
2092+
if node.Hostname != "" {
2093+
entry["hostname"] = node.Hostname
2094+
}
2095+
if node.TaskExec {
2096+
entry["task_exec"] = true
2097+
}
2098+
if node.Public {
2099+
entry["real_addr"] = node.RealAddr
2100+
}
2101+
nodes = append(nodes, entry)
2102+
}
2103+
return map[string]interface{}{
2104+
"type": "list_nodes_ok",
2105+
"nodes": nodes,
2106+
}, nil
20342107
}
20352108

20362109
s.mu.RLock()

‎tests/datagram_test.go‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,6 @@ func TestUnicastDatagram(t *testing.T) {
5757

5858
// TestBroadcastExcludesSender verifies the broadcaster does not receive its own broadcast.
5959
func TestBroadcastExcludesSender(t *testing.T) {
60-
t.Skip("custom networks are WIP")
6160
t.Parallel()
6261
env := NewTestEnv(t)
6362

‎tests/fuzz_registry_server_test.go‎

Lines changed: 60 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -724,11 +724,69 @@ func TestRegistryClientNetworkCreateNoToken(t *testing.T) {
724724
}
725725

726726
func TestRegistryClientNetworkCreateWithToken(t *testing.T) {
727-
t.Skip("custom networks are WIP")
727+
s := startTestServer(t)
728+
s.SetAdminToken(TestAdminToken)
729+
defer s.Close()
730+
731+
addr := s.Addr().(*net.TCPAddr)
732+
c, err := registry.Dial(addr.String())
733+
if err != nil {
734+
t.Fatalf("Dial: %v", err)
735+
}
736+
defer c.Close()
737+
738+
nodeID := regTestNodeWithKey(t, c, "127.0.0.1:4000")
739+
740+
resp, err := c.CreateNetwork(nodeID, "tokennet", "open", "", TestAdminToken)
741+
if err != nil {
742+
t.Fatalf("CreateNetwork with token: %v", err)
743+
}
744+
netID, ok := resp["network_id"].(float64)
745+
if !ok || netID < 1 {
746+
t.Fatalf("expected valid network_id, got %v", resp["network_id"])
747+
}
748+
t.Logf("created network %d", int(netID))
728749
}
729750

730751
func TestRegistryClientNetworkJoinLeave(t *testing.T) {
731-
t.Skip("custom networks are WIP")
752+
s := startTestServer(t)
753+
s.SetAdminToken(TestAdminToken)
754+
defer s.Close()
755+
756+
addr := s.Addr().(*net.TCPAddr)
757+
c1, err := registry.Dial(addr.String())
758+
if err != nil {
759+
t.Fatalf("Dial c1: %v", err)
760+
}
761+
defer c1.Close()
762+
c2, err := registry.Dial(addr.String())
763+
if err != nil {
764+
t.Fatalf("Dial c2: %v", err)
765+
}
766+
defer c2.Close()
767+
768+
nodeID1 := regTestNodeWithKey(t, c1, "127.0.0.1:4001")
769+
nodeID2 := regTestNodeWithKey(t, c2, "127.0.0.1:4002")
770+
771+
// Create network with node1
772+
resp, err := c1.CreateNetwork(nodeID1, "joinleave", "open", "", TestAdminToken)
773+
if err != nil {
774+
t.Fatalf("CreateNetwork: %v", err)
775+
}
776+
netID := uint16(resp["network_id"].(float64))
777+
778+
// Node2 joins
779+
_, err = c2.JoinNetwork(nodeID2, netID, "", 0, TestAdminToken)
780+
if err != nil {
781+
t.Fatalf("JoinNetwork: %v", err)
782+
}
783+
784+
// Node2 leaves
785+
_, err = c2.LeaveNetwork(nodeID2, netID, TestAdminToken)
786+
if err != nil {
787+
t.Fatalf("LeaveNetwork: %v", err)
788+
}
789+
t.Logf("node %d joined and left network %d", nodeID2, netID)
732790
}
733791

734792
func TestRegistryClientListNetworks(t *testing.T) {

‎tests/persistence_test.go‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ import (
1111
)
1212

1313
func TestRegistryPersistence(t *testing.T) {
14-
t.Skip("custom networks are WIP")
1514
t.Parallel()
1615
tmpDir, err := os.MkdirTemp("/tmp", "w4-persist-")
1716
if err != nil {

‎tests/replication_test.go‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@ import (
1616
// 3. Verify standby has the same data
1717
// 4. Verify standby rejects writes
1818
func TestRegistryReplication(t *testing.T) {
19-
t.Skip("custom networks are WIP")
2019
t.Parallel()
2120
tmpDir, err := os.MkdirTemp("/tmp", "w4-repl-")
2221
if err != nil {

0 commit comments

Comments
 (0)