| 1 | package cli |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "fmt" |
| 6 | "io" |
| 7 | "net" |
| 8 | "net/http" |
| 9 | "os/exec" |
| 10 | "slices" |
| 11 | "syscall" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/ipfs/kubo/core/commands" |
| 16 | "github.com/ipfs/kubo/test/cli/harness" |
| 17 | "github.com/stretchr/testify/require" |
| 18 | ) |
| 19 | |
| 20 | // waitForListenerCount waits until the node has exactly the expected number of listeners. |
| 21 | func waitForListenerCount(t *testing.T, node *harness.Node, expectedCount int) { |
| 22 | t.Helper() |
| 23 | require.Eventually(t, func() bool { |
| 24 | lsOut := node.IPFS("p2p", "ls", "--enc=json") |
| 25 | var lsResult commands.P2PLsOutput |
| 26 | if err := json.Unmarshal(lsOut.Stdout.Bytes(), &lsResult); err != nil { |
| 27 | return false |
| 28 | } |
| 29 | return len(lsResult.Listeners) == expectedCount |
| 30 | }, 5*time.Second, 100*time.Millisecond, "expected %d listeners", expectedCount) |
| 31 | } |
| 32 | |
| 33 | // waitForListenerProtocol waits until the node has a listener with the given protocol. |
| 34 | func waitForListenerProtocol(t *testing.T, node *harness.Node, protocol string) { |
| 35 | t.Helper() |
| 36 | require.Eventually(t, func() bool { |
| 37 | lsOut := node.IPFS("p2p", "ls", "--enc=json") |
| 38 | var lsResult commands.P2PLsOutput |
| 39 | if err := json.Unmarshal(lsOut.Stdout.Bytes(), &lsResult); err != nil { |
| 40 | return false |
| 41 | } |
| 42 | return slices.ContainsFunc(lsResult.Listeners, func(l commands.P2PListenerInfoOutput) bool { |
| 43 | return l.Protocol == protocol |
| 44 | }) |
| 45 | }, 5*time.Second, 100*time.Millisecond, "expected listener with protocol %s", protocol) |
| 46 | } |
| 47 | |
| 48 | func TestP2PForeground(t *testing.T) { |
| 49 | t.Parallel() |
| 50 | |
| 51 | t.Run("listen foreground creates listener and removes on interrupt", func(t *testing.T) { |
| 52 | t.Parallel() |
| 53 | node := harness.NewT(t).NewNode().Init() |
| 54 | node.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 55 | node.StartDaemon() |
| 56 | |
| 57 | listenPort := harness.NewRandPort() |
| 58 | |
| 59 | // Start foreground listener asynchronously |
| 60 | res := node.Runner.Run(harness.RunRequest{ |
| 61 | Path: node.IPFSBin, |
| 62 | Args: []string{"p2p", "listen", "--foreground", "/x/fgtest", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", listenPort)}, |
| 63 | RunFunc: (*exec.Cmd).Start, |
| 64 | }) |
| 65 | require.NoError(t, res.Err) |
| 66 | |
| 67 | // Wait for listener to be created |
| 68 | waitForListenerProtocol(t, node, "/x/fgtest") |
| 69 | |
| 70 | // Send SIGTERM |
| 71 | _ = res.Cmd.Process.Signal(syscall.SIGTERM) |
| 72 | _ = res.Cmd.Wait() |
| 73 | |
| 74 | // Wait for listener to be removed |
| 75 | waitForListenerCount(t, node, 0) |
| 76 | }) |
| 77 | |
| 78 | t.Run("listen foreground text output on SIGTERM", func(t *testing.T) { |
| 79 | t.Parallel() |
| 80 | node := harness.NewT(t).NewNode().Init() |
| 81 | node.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 82 | node.StartDaemon() |
| 83 | |
| 84 | listenPort := harness.NewRandPort() |
| 85 | |
| 86 | // Run without --enc=json to test actual text output users see |
| 87 | res := node.Runner.Run(harness.RunRequest{ |
| 88 | Path: node.IPFSBin, |
| 89 | Args: []string{"p2p", "listen", "--foreground", "/x/sigterm", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", listenPort)}, |
| 90 | RunFunc: (*exec.Cmd).Start, |
| 91 | }) |
| 92 | require.NoError(t, res.Err) |
| 93 | |
| 94 | waitForListenerProtocol(t, node, "/x/sigterm") |
| 95 | |
| 96 | _ = res.Cmd.Process.Signal(syscall.SIGTERM) |
| 97 | _ = res.Cmd.Wait() |
| 98 | |
| 99 | // Verify stdout shows "waiting for interrupt" message |
| 100 | stdout := res.Stdout.String() |
| 101 | require.Contains(t, stdout, "waiting for interrupt") |
| 102 | |
| 103 | // Note: "Received interrupt, removing listener" message is NOT visible to CLI on SIGTERM |
| 104 | // because the command runs in the daemon via RPC and the response stream closes before |
| 105 | // the message can be emitted. The important behavior is verified in the first test: |
| 106 | // the listener IS removed when SIGTERM is sent. |
| 107 | }) |
| 108 | |
| 109 | t.Run("forward foreground creates forwarder and removes on interrupt", func(t *testing.T) { |
| 110 | t.Parallel() |
| 111 | nodes := harness.NewT(t).NewNodes(2).Init() |
| 112 | nodes.ForEachPar(func(n *harness.Node) { |
| 113 | n.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 114 | }) |
| 115 | nodes.StartDaemons().Connect() |
| 116 | |
| 117 | forwardPort := harness.NewRandPort() |
| 118 | |
| 119 | // Start foreground forwarder asynchronously on node 0 |
| 120 | res := nodes[0].Runner.Run(harness.RunRequest{ |
| 121 | Path: nodes[0].IPFSBin, |
| 122 | Args: []string{"p2p", "forward", "--foreground", "/x/fgfwd", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", forwardPort), "/p2p/" + nodes[1].PeerID().String()}, |
| 123 | RunFunc: (*exec.Cmd).Start, |
| 124 | }) |
| 125 | require.NoError(t, res.Err) |
| 126 | |
| 127 | // Wait for forwarder to be created |
| 128 | waitForListenerCount(t, nodes[0], 1) |
| 129 | |
| 130 | // Send SIGTERM |
| 131 | _ = res.Cmd.Process.Signal(syscall.SIGTERM) |
| 132 | _ = res.Cmd.Wait() |
| 133 | |
| 134 | // Wait for forwarder to be removed |
| 135 | waitForListenerCount(t, nodes[0], 0) |
| 136 | }) |
| 137 | |
| 138 | t.Run("forward foreground text output on SIGTERM", func(t *testing.T) { |
| 139 | t.Parallel() |
| 140 | nodes := harness.NewT(t).NewNodes(2).Init() |
| 141 | nodes.ForEachPar(func(n *harness.Node) { |
| 142 | n.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 143 | }) |
| 144 | nodes.StartDaemons().Connect() |
| 145 | |
| 146 | forwardPort := harness.NewRandPort() |
| 147 | |
| 148 | // Run without --enc=json to test actual text output users see |
| 149 | res := nodes[0].Runner.Run(harness.RunRequest{ |
| 150 | Path: nodes[0].IPFSBin, |
| 151 | Args: []string{"p2p", "forward", "--foreground", "/x/fwdsigterm", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", forwardPort), "/p2p/" + nodes[1].PeerID().String()}, |
| 152 | RunFunc: (*exec.Cmd).Start, |
| 153 | }) |
| 154 | require.NoError(t, res.Err) |
| 155 | |
| 156 | waitForListenerCount(t, nodes[0], 1) |
| 157 | |
| 158 | _ = res.Cmd.Process.Signal(syscall.SIGTERM) |
| 159 | _ = res.Cmd.Wait() |
| 160 | |
| 161 | // Verify stdout shows "waiting for interrupt" message |
| 162 | stdout := res.Stdout.String() |
| 163 | require.Contains(t, stdout, "waiting for interrupt") |
| 164 | |
| 165 | // Note: "Received interrupt, removing forwarder" message is NOT visible to CLI on SIGTERM |
| 166 | // because the response stream closes before the message can be emitted. |
| 167 | }) |
| 168 | |
| 169 | t.Run("listen without foreground returns immediately and persists", func(t *testing.T) { |
| 170 | t.Parallel() |
| 171 | node := harness.NewT(t).NewNode().Init() |
| 172 | node.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 173 | node.StartDaemon() |
| 174 | |
| 175 | listenPort := harness.NewRandPort() |
| 176 | |
| 177 | // This should return immediately (not block) |
| 178 | node.IPFS("p2p", "listen", "/x/nofg", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", listenPort)) |
| 179 | |
| 180 | // Listener should still exist |
| 181 | waitForListenerProtocol(t, node, "/x/nofg") |
| 182 | |
| 183 | // Clean up |
| 184 | node.IPFS("p2p", "close", "-p", "/x/nofg") |
| 185 | }) |
| 186 | |
| 187 | t.Run("listen foreground text output on p2p close", func(t *testing.T) { |
| 188 | t.Parallel() |
| 189 | node := harness.NewT(t).NewNode().Init() |
| 190 | node.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 191 | node.StartDaemon() |
| 192 | |
| 193 | listenPort := harness.NewRandPort() |
| 194 | |
| 195 | // Run without --enc=json to test actual text output users see |
| 196 | res := node.Runner.Run(harness.RunRequest{ |
| 197 | Path: node.IPFSBin, |
| 198 | Args: []string{"p2p", "listen", "--foreground", "/x/closetest", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", listenPort)}, |
| 199 | RunFunc: (*exec.Cmd).Start, |
| 200 | }) |
| 201 | require.NoError(t, res.Err) |
| 202 | |
| 203 | // Wait for listener to be created |
| 204 | waitForListenerProtocol(t, node, "/x/closetest") |
| 205 | |
| 206 | // Close the listener via ipfs p2p close command |
| 207 | node.IPFS("p2p", "close", "-p", "/x/closetest") |
| 208 | |
| 209 | // Wait for foreground command to exit (it should exit quickly after close) |
| 210 | done := make(chan error, 1) |
| 211 | go func() { |
| 212 | done <- res.Cmd.Wait() |
| 213 | }() |
| 214 | |
| 215 | select { |
| 216 | case <-done: |
| 217 | // Good - command exited |
| 218 | case <-time.After(5 * time.Second): |
| 219 | _ = res.Cmd.Process.Kill() |
| 220 | t.Fatal("foreground command did not exit after listener was closed via ipfs p2p close") |
| 221 | } |
| 222 | |
| 223 | // Wait for listener to be removed |
| 224 | waitForListenerCount(t, node, 0) |
| 225 | |
| 226 | // Verify text output shows BOTH messages when closed via p2p close |
| 227 | // (unlike SIGTERM, the stream is still open so "Received interrupt" is emitted) |
| 228 | out := res.Stdout.String() |
| 229 | require.Contains(t, out, "waiting for interrupt") |
| 230 | require.Contains(t, out, "Received interrupt, removing listener") |
| 231 | }) |
| 232 | |
| 233 | t.Run("forward foreground text output on p2p close", func(t *testing.T) { |
| 234 | t.Parallel() |
| 235 | nodes := harness.NewT(t).NewNodes(2).Init() |
| 236 | nodes.ForEachPar(func(n *harness.Node) { |
| 237 | n.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 238 | }) |
| 239 | nodes.StartDaemons().Connect() |
| 240 | |
| 241 | forwardPort := harness.NewRandPort() |
| 242 | |
| 243 | // Run without --enc=json to test actual text output users see |
| 244 | res := nodes[0].Runner.Run(harness.RunRequest{ |
| 245 | Path: nodes[0].IPFSBin, |
| 246 | Args: []string{"p2p", "forward", "--foreground", "/x/fwdclose", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", forwardPort), "/p2p/" + nodes[1].PeerID().String()}, |
| 247 | RunFunc: (*exec.Cmd).Start, |
| 248 | }) |
| 249 | require.NoError(t, res.Err) |
| 250 | |
| 251 | // Wait for forwarder to be created |
| 252 | waitForListenerCount(t, nodes[0], 1) |
| 253 | |
| 254 | // Close the forwarder via ipfs p2p close command |
| 255 | nodes[0].IPFS("p2p", "close", "-a") |
| 256 | |
| 257 | // Wait for foreground command to exit |
| 258 | done := make(chan error, 1) |
| 259 | go func() { |
| 260 | done <- res.Cmd.Wait() |
| 261 | }() |
| 262 | |
| 263 | select { |
| 264 | case <-done: |
| 265 | // Good - command exited |
| 266 | case <-time.After(5 * time.Second): |
| 267 | _ = res.Cmd.Process.Kill() |
| 268 | t.Fatal("foreground command did not exit after forwarder was closed via ipfs p2p close") |
| 269 | } |
| 270 | |
| 271 | // Wait for forwarder to be removed |
| 272 | waitForListenerCount(t, nodes[0], 0) |
| 273 | |
| 274 | // Verify text output shows BOTH messages when closed via p2p close |
| 275 | out := res.Stdout.String() |
| 276 | require.Contains(t, out, "waiting for interrupt") |
| 277 | require.Contains(t, out, "Received interrupt, removing forwarder") |
| 278 | }) |
| 279 | |
| 280 | t.Run("listen foreground tunnel transfers data and cleans up on SIGTERM", func(t *testing.T) { |
| 281 | t.Parallel() |
| 282 | nodes := harness.NewT(t).NewNodes(2).Init() |
| 283 | nodes.ForEachPar(func(n *harness.Node) { |
| 284 | n.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 285 | }) |
| 286 | nodes.StartDaemons().Connect() |
| 287 | |
| 288 | httpServerPort := harness.NewRandPort() |
| 289 | forwardPort := harness.NewRandPort() |
| 290 | |
| 291 | // Start HTTP server |
| 292 | expectedBody := "Hello from p2p tunnel!" |
| 293 | httpServer := &http.Server{ |
| 294 | Addr: fmt.Sprintf("127.0.0.1:%d", httpServerPort), |
| 295 | Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { |
| 296 | _, _ = w.Write([]byte(expectedBody)) |
| 297 | }), |
| 298 | } |
| 299 | listener, err := net.Listen("tcp", httpServer.Addr) |
| 300 | require.NoError(t, err) |
| 301 | go func() { _ = httpServer.Serve(listener) }() |
| 302 | defer httpServer.Close() |
| 303 | |
| 304 | // Node 0: listen --foreground |
| 305 | listenRes := nodes[0].Runner.Run(harness.RunRequest{ |
| 306 | Path: nodes[0].IPFSBin, |
| 307 | Args: []string{"p2p", "listen", "--foreground", "/x/httptest", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", httpServerPort)}, |
| 308 | RunFunc: (*exec.Cmd).Start, |
| 309 | }) |
| 310 | require.NoError(t, listenRes.Err) |
| 311 | |
| 312 | // Wait for listener to be created |
| 313 | waitForListenerProtocol(t, nodes[0], "/x/httptest") |
| 314 | |
| 315 | // Node 1: forward (non-foreground) |
| 316 | nodes[1].IPFS("p2p", "forward", "/x/httptest", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", forwardPort), "/p2p/"+nodes[0].PeerID().String()) |
| 317 | |
| 318 | // Verify data flows through tunnel |
| 319 | resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/", forwardPort)) |
| 320 | require.NoError(t, err) |
| 321 | body, err := io.ReadAll(resp.Body) |
| 322 | resp.Body.Close() |
| 323 | require.NoError(t, err) |
| 324 | require.Equal(t, expectedBody, string(body)) |
| 325 | |
| 326 | // Clean up forwarder on node 1 |
| 327 | nodes[1].IPFS("p2p", "close", "-a") |
| 328 | |
| 329 | // SIGTERM the listen --foreground command |
| 330 | _ = listenRes.Cmd.Process.Signal(syscall.SIGTERM) |
| 331 | _ = listenRes.Cmd.Wait() |
| 332 | |
| 333 | // Wait for listener to be removed on node 0 |
| 334 | waitForListenerCount(t, nodes[0], 0) |
| 335 | }) |
| 336 | |
| 337 | t.Run("forward foreground tunnel transfers data and cleans up on SIGTERM", func(t *testing.T) { |
| 338 | t.Parallel() |
| 339 | nodes := harness.NewT(t).NewNodes(2).Init() |
| 340 | nodes.ForEachPar(func(n *harness.Node) { |
| 341 | n.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 342 | }) |
| 343 | nodes.StartDaemons().Connect() |
| 344 | |
| 345 | httpServerPort := harness.NewRandPort() |
| 346 | forwardPort := harness.NewRandPort() |
| 347 | |
| 348 | // Start HTTP server |
| 349 | expectedBody := "Hello from forward foreground tunnel!" |
| 350 | httpServer := &http.Server{ |
| 351 | Addr: fmt.Sprintf("127.0.0.1:%d", httpServerPort), |
| 352 | Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { |
| 353 | _, _ = w.Write([]byte(expectedBody)) |
| 354 | }), |
| 355 | } |
| 356 | listener, err := net.Listen("tcp", httpServer.Addr) |
| 357 | require.NoError(t, err) |
| 358 | go func() { _ = httpServer.Serve(listener) }() |
| 359 | defer httpServer.Close() |
| 360 | |
| 361 | // Node 0: listen (non-foreground) |
| 362 | nodes[0].IPFS("p2p", "listen", "/x/httptest", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", httpServerPort)) |
| 363 | |
| 364 | // Node 1: forward --foreground |
| 365 | forwardRes := nodes[1].Runner.Run(harness.RunRequest{ |
| 366 | Path: nodes[1].IPFSBin, |
| 367 | Args: []string{"p2p", "forward", "--foreground", "/x/httptest", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", forwardPort), "/p2p/" + nodes[0].PeerID().String()}, |
| 368 | RunFunc: (*exec.Cmd).Start, |
| 369 | }) |
| 370 | require.NoError(t, forwardRes.Err) |
| 371 | |
| 372 | // Wait for forwarder to be created |
| 373 | waitForListenerCount(t, nodes[1], 1) |
| 374 | |
| 375 | // Verify data flows through tunnel |
| 376 | resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/", forwardPort)) |
| 377 | require.NoError(t, err) |
| 378 | body, err := io.ReadAll(resp.Body) |
| 379 | resp.Body.Close() |
| 380 | require.NoError(t, err) |
| 381 | require.Equal(t, expectedBody, string(body)) |
| 382 | |
| 383 | // SIGTERM the forward --foreground command |
| 384 | _ = forwardRes.Cmd.Process.Signal(syscall.SIGTERM) |
| 385 | _ = forwardRes.Cmd.Wait() |
| 386 | |
| 387 | // Wait for forwarder to be removed on node 1 |
| 388 | waitForListenerCount(t, nodes[1], 0) |
| 389 | |
| 390 | // Clean up listener on node 0 |
| 391 | nodes[0].IPFS("p2p", "close", "-a") |
| 392 | }) |
| 393 | |
| 394 | t.Run("foreground command exits when daemon shuts down", func(t *testing.T) { |
| 395 | t.Parallel() |
| 396 | node := harness.NewT(t).NewNode().Init() |
| 397 | node.IPFS("config", "--json", "Experimental.Libp2pStreamMounting", "true") |
| 398 | node.StartDaemon() |
| 399 | |
| 400 | listenPort := harness.NewRandPort() |
| 401 | |
| 402 | // Start foreground listener |
| 403 | res := node.Runner.Run(harness.RunRequest{ |
| 404 | Path: node.IPFSBin, |
| 405 | Args: []string{"p2p", "listen", "--foreground", "/x/daemontest", fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", listenPort)}, |
| 406 | RunFunc: (*exec.Cmd).Start, |
| 407 | }) |
| 408 | require.NoError(t, res.Err) |
| 409 | |
| 410 | // Wait for listener to be created |
| 411 | waitForListenerProtocol(t, node, "/x/daemontest") |
| 412 | |
| 413 | // Stop the daemon |
| 414 | node.StopDaemon() |
| 415 | |
| 416 | // Wait for foreground command to exit |
| 417 | done := make(chan error, 1) |
| 418 | go func() { |
| 419 | done <- res.Cmd.Wait() |
| 420 | }() |
| 421 | |
| 422 | select { |
| 423 | case <-done: |
| 424 | // Good - foreground command exited when daemon stopped |
| 425 | case <-time.After(5 * time.Second): |
| 426 | _ = res.Cmd.Process.Kill() |
| 427 | t.Fatal("foreground command did not exit when daemon was stopped") |
| 428 | } |
| 429 | }) |
| 430 | } |