@cryptotaxi247 / kubo / commits / 25ebab9da

feat(p2p): add --foreground flag to listen and forward commands (#11099)

* feat(p2p): add --foreground flag to listen and forward commands adds `-f/--foreground` option that keeps the command running until interrupted (SIGTERM/Ctrl+C) or closed via `ipfs p2p close`. the listener/forwarder is automatically removed when the command exits. useful for systemd services and scripts that need cleanup on exit. * docs: add p2p-tunnels.md with systemd examples - add dedicated docs/p2p-tunnels.md covering: - why p2p tunnels (NAT traversal, no public IP needed) - quick start with netcat - background and foreground modes - systemd integration with path-based activation - security considerations and troubleshooting - document Experimental.Libp2pStreamMounting in docs/config.md - simplify docs/experimental-features.md, link to new doc - add "Learn more" links to ipfs p2p listen/forward --help - update changelog entry with doc link - add cross-reference in misc/README.md * chore: reference kubo#5460 for p2p config Ref. https://github.com/ipfs/kubo/issues/5460 * fix(daemon): write api/gateway files only after HTTP server is ready fixes race condition where $IPFS_PATH/api and $IPFS_PATH/gateway files were written before the HTTP servers were ready to accept connections. this caused issues for tools like systemd path units that immediately try to connect when these files appear. changes: - add corehttp.ServeWithReady() that signals when server is ready - wait for ready signal before writing address files - use sync.WaitGroup.Go() (Go 1.25) for cleaner goroutine management - add TestAddressFileReady to verify both api and gateway files * fix(daemon): buffer errc channel and wait for all listeners - buffer error channel with len(listeners) to prevent deadlock when multiple servers write errors simultaneously - wait for ALL listeners to be ready before writing api/gateway file, not just the first one Feedback-from: https://github.com/ipfs/kubo/pull/11099#pullrequestreview-3593885839 * docs(changelog): improve p2p tunnel section clarity reframe to lead with user benefit and add example output * docs(p2p): remove obsolete race condition caveat the "First launch fails but restarts work" troubleshooting section described a race where the api file was written before the daemon was ready. this was fixed in 80b703a which ensures api/gateway files are only written after HTTP servers are ready to accept connections. --------- Co-authored-by: Andrew Gillis <11790789+gammazero@users.noreply.github.com>

Marcin Rataj committed Jan 9, 2026 at 19:22 UTC 25ebab9dae9a564f2c6f2424961feb5b7642ea31
13 files changed +1048 -134
cmd/ipfs/kubo/daemon.go
+53 -22
@@ -883,23 +883,38 @@ func serveHTTPApi(req *cmds.Request, cctx *oldcmds.Context) (<-chan error, error
883 return nil, fmt.Errorf("serveHTTPApi: ConstructNode() failed: %s", err)
884 }
885
886 + // Buffer channel to prevent deadlock when multiple servers write errors simultaneously
887 + errc := make(chan error, len(listeners))
888 + var wg sync.WaitGroup
889 +
890 + // Start all servers and wait for them to be ready before writing api file.
891 + // This prevents race conditions where external tools (like systemd path units)
892 + // see the file and try to connect before servers can accept connections.
893 if len(listeners) > 0 {
887 - // Only add an api file if the API is running.
894 + readyChannels := make([]chan struct{}, len(listeners))
895 + for i, lis := range listeners {
896 + readyChannels[i] = make(chan struct{})
897 + ready := readyChannels[i]
898 + wg.Go(func() {
899 + errc <- corehttp.ServeWithReady(node, manet.NetListener(lis), ready, opts...)
900 + })
901 + }
902 +
903 + // Wait for all listeners to be ready or any to fail
904 + for _, ready := range readyChannels {
905 + select {
906 + case <-ready:
907 + // This listener is ready
908 + case err := <-errc:
909 + return nil, fmt.Errorf("serveHTTPApi: %w", err)
910 + }
911 + }
912 +
913 if err := node.Repo.SetAPIAddr(rewriteMaddrToUseLocalhostIfItsAny(listeners[0].Multiaddr())); err != nil {
914 return nil, fmt.Errorf("serveHTTPApi: SetAPIAddr() failed: %w", err)
915 }
916 }
917
893 - errc := make(chan error)
894 - var wg sync.WaitGroup
895 - for _, apiLis := range listeners {
896 - wg.Add(1)
897 - go func(lis manet.Listener) {
898 - defer wg.Done()
899 - errc <- corehttp.Serve(node, manet.NetListener(lis), opts...)
900 - }(apiLis)
901 - }
902 -
918 go func() {
919 wg.Wait()
920 close(errc)
@@ -1058,26 +1073,42 @@ func serveHTTPGateway(req *cmds.Request, cctx *oldcmds.Context) (<-chan error, e
1073 return nil, fmt.Errorf("serveHTTPGateway: ConstructNode() failed: %s", err)
1074 }
1075
1076 + // Buffer channel to prevent deadlock when multiple servers write errors simultaneously
1077 + errc := make(chan error, len(listeners))
1078 + var wg sync.WaitGroup
1079 +
1080 + // Start all servers and wait for them to be ready before writing gateway file.
1081 + // This prevents race conditions where external tools (like systemd path units)
1082 + // see the file and try to connect before servers can accept connections.
1083 if len(listeners) > 0 {
1084 + readyChannels := make([]chan struct{}, len(listeners))
1085 + for i, lis := range listeners {
1086 + readyChannels[i] = make(chan struct{})
1087 + ready := readyChannels[i]
1088 + wg.Go(func() {
1089 + errc <- corehttp.ServeWithReady(node, manet.NetListener(lis), ready, opts...)
1090 + })
1091 + }
1092 +
1093 + // Wait for all listeners to be ready or any to fail
1094 + for _, ready := range readyChannels {
1095 + select {
1096 + case <-ready:
1097 + // This listener is ready
1098 + case err := <-errc:
1099 + return nil, fmt.Errorf("serveHTTPGateway: %w", err)
1100 + }
1101 + }
1102 +
1103 addr, err := manet.ToNetAddr(rewriteMaddrToUseLocalhostIfItsAny(listeners[0].Multiaddr()))
1104 if err != nil {
1064 - return nil, fmt.Errorf("serveHTTPGateway: manet.ToIP() failed: %w", err)
1105 + return nil, fmt.Errorf("serveHTTPGateway: manet.ToNetAddr() failed: %w", err)
1106 }
1107 if err := node.Repo.SetGatewayAddr(addr); err != nil {
1108 return nil, fmt.Errorf("serveHTTPGateway: SetGatewayAddr() failed: %w", err)
1109 }
1110 }
1111
1071 - errc := make(chan error)
1072 - var wg sync.WaitGroup
1073 - for _, lis := range listeners {
1074 - wg.Add(1)
1075 - go func(lis manet.Listener) {
1076 - defer wg.Done()
1077 - errc <- corehttp.Serve(node, manet.NetListener(lis), opts...)
1078 - }(lis)
1079 - }
1080 -
1112 go func() {
1113 wg.Wait()
1114 close(errc)
core/commands/p2p.go
+161 -18
@@ -50,9 +50,17 @@ type P2PStreamsOutput struct {
50 Streams []P2PStreamInfoOutput
51 }
52
53 +// P2PForegroundOutput is output type for foreground mode status messages
54 +type P2PForegroundOutput struct {
55 + Status string // "active" or "closing"
56 + Protocol string
57 + Address string
58 +}
59 +
60 const (
61 allowCustomProtocolOptionName = "allow-custom-protocol"
62 reportPeerIDOptionName = "report-peer-id"
63 + foregroundOptionName = "foreground"
64 )
65
66 var resolveTimeout = 10 * time.Second
@@ -83,15 +91,37 @@ var p2pForwardCmd = &cmds.Command{
91 Helptext: cmds.HelpText{
92 Tagline: "Forward connections to libp2p service.",
93 ShortDescription: `
86 -Forward connections made to <listen-address> to <target-address>.
94 +Forward connections made to <listen-address> to <target-address> via libp2p.
95 +
96 +Creates a local TCP listener that tunnels connections through libp2p to a
97 +remote peer's p2p listener. Similar to SSH port forwarding (-L flag).
98 +
99 +ARGUMENTS:
100 +
101 + <protocol> Protocol name (must start with '` + P2PProtoPrefix + `')
102 + <listen-address> Local multiaddr (e.g., /ip4/127.0.0.1/tcp/3000)
103 + <target-address> Remote peer multiaddr (e.g., /p2p/PeerID)
104 +
105 +FOREGROUND MODE (--foreground, -f):
106 +
107 + By default, the forwarder runs in the daemon and the command returns
108 + immediately. Use --foreground to block until interrupted:
109
88 -<protocol> specifies the libp2p protocol name to use for libp2p
89 -connections and/or handlers. It must be prefixed with '` + P2PProtoPrefix + `'.
110 + - Ctrl+C or SIGTERM: Removes the forwarder and exits
111 + - 'ipfs p2p close': Removes the forwarder and exits
112 + - Daemon shutdown: Forwarder is automatically removed
113
91 -Example:
92 - ipfs p2p forward ` + P2PProtoPrefix + `myproto /ip4/127.0.0.1/tcp/4567 /p2p/QmPeer
93 - - Forward connections to 127.0.0.1:4567 to '` + P2PProtoPrefix + `myproto' service on /p2p/QmPeer
114 + Useful for systemd services or scripts that need cleanup on exit.
115
116 +EXAMPLES:
117 +
118 + # Persistent forwarder (command returns immediately)
119 + ipfs p2p forward /x/myapp /ip4/127.0.0.1/tcp/3000 /p2p/PeerID
120 +
121 + # Temporary forwarder (removed when command exits)
122 + ipfs p2p forward -f /x/myapp /ip4/127.0.0.1/tcp/3000 /p2p/PeerID
123 +
124 +Learn more: https://github.com/ipfs/kubo/blob/master/docs/p2p-tunnels.md
125 `,
126 },
127 Arguments: []cmds.Argument{
@@ -101,6 +131,7 @@ Example:
131 },
132 Options: []cmds.Option{
133 cmds.BoolOption(allowCustomProtocolOptionName, "Don't require /x/ prefix"),
134 + cmds.BoolOption(foregroundOptionName, "f", "Run in foreground; forwarder is removed when command exits"),
135 },
136 Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
137 n, err := p2pGetNode(env)
@@ -130,7 +161,51 @@ Example:
161 return errors.New("protocol name must be within '" + P2PProtoPrefix + "' namespace")
162 }
163
133 - return forwardLocal(n.Context(), n.P2P, n.Peerstore, proto, listen, targets)
164 + listener, err := forwardLocal(n.Context(), n.P2P, n.Peerstore, proto, listen, targets)
165 + if err != nil {
166 + return err
167 + }
168 +
169 + foreground, _ := req.Options[foregroundOptionName].(bool)
170 + if foreground {
171 + if err := res.Emit(&P2PForegroundOutput{
172 + Status: "active",
173 + Protocol: protoOpt,
174 + Address: listenOpt,
175 + }); err != nil {
176 + return err
177 + }
178 + // Wait for either context cancellation (Ctrl+C/daemon shutdown)
179 + // or listener removal (ipfs p2p close)
180 + select {
181 + case <-req.Context.Done():
182 + // SIGTERM/Ctrl+C - cleanup silently (CLI stream already closing)
183 + n.P2P.ListenersLocal.Close(func(l p2p.Listener) bool {
184 + return l == listener
185 + })
186 + return nil
187 + case <-listener.Done():
188 + // Closed via "ipfs p2p close" - emit closing message
189 + return res.Emit(&P2PForegroundOutput{
190 + Status: "closing",
191 + Protocol: protoOpt,
192 + Address: listenOpt,
193 + })
194 + }
195 + }
196 +
197 + return nil
198 + },
199 + Type: P2PForegroundOutput{},
200 + Encoders: cmds.EncoderMap{
201 + cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *P2PForegroundOutput) error {
202 + if out.Status == "active" {
203 + fmt.Fprintf(w, "Forwarding %s to %s, waiting for interrupt...\n", out.Protocol, out.Address)
204 + } else if out.Status == "closing" {
205 + fmt.Fprintf(w, "Received interrupt, removing forwarder for %s\n", out.Protocol)
206 + }
207 + return nil
208 + }),
209 },
210 }
211
@@ -185,14 +260,40 @@ var p2pListenCmd = &cmds.Command{
260 Helptext: cmds.HelpText{
261 Tagline: "Create libp2p service.",
262 ShortDescription: `
188 -Create libp2p service and forward connections made to <target-address>.
263 +Create a libp2p protocol handler that forwards incoming connections to
264 +<target-address>.
265
190 -<protocol> specifies the libp2p handler name. It must be prefixed with '` + P2PProtoPrefix + `'.
266 +When a remote peer connects using 'ipfs p2p forward', the connection is
267 +forwarded to your local service. Similar to SSH port forwarding (server side).
268
192 -Example:
193 - ipfs p2p listen ` + P2PProtoPrefix + `myproto /ip4/127.0.0.1/tcp/1234
194 - - Forward connections to 'myproto' libp2p service to 127.0.0.1:1234
269 +ARGUMENTS:
270
271 + <protocol> Protocol name (must start with '` + P2PProtoPrefix + `')
272 + <target-address> Local multiaddr (e.g., /ip4/127.0.0.1/tcp/3000)
273 +
274 +FOREGROUND MODE (--foreground, -f):
275 +
276 + By default, the listener runs in the daemon and the command returns
277 + immediately. Use --foreground to block until interrupted:
278 +
279 + - Ctrl+C or SIGTERM: Removes the listener and exits
280 + - 'ipfs p2p close': Removes the listener and exits
281 + - Daemon shutdown: Listener is automatically removed
282 +
283 + Useful for systemd services or scripts that need cleanup on exit.
284 +
285 +EXAMPLES:
286 +
287 + # Persistent listener (command returns immediately)
288 + ipfs p2p listen /x/myapp /ip4/127.0.0.1/tcp/3000
289 +
290 + # Temporary listener (removed when command exits)
291 + ipfs p2p listen -f /x/myapp /ip4/127.0.0.1/tcp/3000
292 +
293 + # Report connecting peer ID to the target application
294 + ipfs p2p listen -r /x/myapp /ip4/127.0.0.1/tcp/3000
295 +
296 +Learn more: https://github.com/ipfs/kubo/blob/master/docs/p2p-tunnels.md
297 `,
298 },
299 Arguments: []cmds.Argument{
@@ -202,6 +303,7 @@ Example:
303 Options: []cmds.Option{
304 cmds.BoolOption(allowCustomProtocolOptionName, "Don't require /x/ prefix"),
305 cmds.BoolOption(reportPeerIDOptionName, "r", "Send remote base58 peerid to target when a new connection is established"),
306 + cmds.BoolOption(foregroundOptionName, "f", "Run in foreground; listener is removed when command exits"),
307 },
308 Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
309 n, err := p2pGetNode(env)
@@ -231,8 +333,51 @@ Example:
333 return errors.New("protocol name must be within '" + P2PProtoPrefix + "' namespace")
334 }
335
234 - _, err = n.P2P.ForwardRemote(n.Context(), proto, target, reportPeerID)
235 - return err
336 + listener, err := n.P2P.ForwardRemote(n.Context(), proto, target, reportPeerID)
337 + if err != nil {
338 + return err
339 + }
340 +
341 + foreground, _ := req.Options[foregroundOptionName].(bool)
342 + if foreground {
343 + if err := res.Emit(&P2PForegroundOutput{
344 + Status: "active",
345 + Protocol: protoOpt,
346 + Address: targetOpt,
347 + }); err != nil {
348 + return err
349 + }
350 + // Wait for either context cancellation (Ctrl+C/daemon shutdown)
351 + // or listener removal (ipfs p2p close)
352 + select {
353 + case <-req.Context.Done():
354 + // SIGTERM/Ctrl+C - cleanup silently (CLI stream already closing)
355 + n.P2P.ListenersP2P.Close(func(l p2p.Listener) bool {
356 + return l == listener
357 + })
358 + return nil
359 + case <-listener.Done():
360 + // Closed via "ipfs p2p close" - emit closing message
361 + return res.Emit(&P2PForegroundOutput{
362 + Status: "closing",
363 + Protocol: protoOpt,
364 + Address: targetOpt,
365 + })
366 + }
367 + }
368 +
369 + return nil
370 + },
371 + Type: P2PForegroundOutput{},
372 + Encoders: cmds.EncoderMap{
373 + cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *P2PForegroundOutput) error {
374 + if out.Status == "active" {
375 + fmt.Fprintf(w, "Listening on %s, forwarding to %s, waiting for interrupt...\n", out.Protocol, out.Address)
376 + } else if out.Status == "closing" {
377 + fmt.Fprintf(w, "Received interrupt, removing listener for %s\n", out.Protocol)
378 + }
379 + return nil
380 + }),
381 },
382 }
383
@@ -271,11 +416,9 @@ func checkPort(target ma.Multiaddr) error {
416 }
417
418 // forwardLocal forwards local connections to a libp2p service
274 -func forwardLocal(ctx context.Context, p *p2p.P2P, ps pstore.Peerstore, proto protocol.ID, bindAddr ma.Multiaddr, addr *peer.AddrInfo) error {
419 +func forwardLocal(ctx context.Context, p *p2p.P2P, ps pstore.Peerstore, proto protocol.ID, bindAddr ma.Multiaddr, addr *peer.AddrInfo) (p2p.Listener, error) {
420 ps.AddAddrs(addr.ID, addr.Addrs, pstore.TempAddrTTL)
276 - // TODO: return some info
277 - _, err := p.ForwardLocal(ctx, addr.ID, proto, bindAddr)
278 - return err
421 + return p.ForwardLocal(ctx, addr.ID, proto, bindAddr)
422 }
423
424 const (
core/corehttp/corehttp.go
+18 -1
@@ -78,9 +78,23 @@ func ListenAndServe(n *core.IpfsNode, listeningMultiAddr string, options ...Serv
78 return Serve(n, manet.NetListener(list), options...)
79 }
80
81 -// Serve accepts incoming HTTP connections on the listener and pass them
81 +// Serve accepts incoming HTTP connections on the listener and passes them
82 // to ServeOption handlers.
83 func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error {
84 + return ServeWithReady(node, lis, nil, options...)
85 +}
86 +
87 +// ServeWithReady is like Serve but signals on the ready channel when the
88 +// server is about to accept connections. The channel is closed right before
89 +// server.Serve() is called.
90 +//
91 +// This is useful for callers that need to perform actions (like writing
92 +// address files) only after the server is guaranteed to be accepting
93 +// connections, avoiding race conditions where clients see the file before
94 +// the server is ready.
95 +//
96 +// Passing nil for ready is equivalent to calling Serve().
97 +func ServeWithReady(node *core.IpfsNode, lis net.Listener, ready chan<- struct{}, options ...ServeOption) error {
98 // make sure we close this no matter what.
99 defer lis.Close()
100
@@ -107,6 +121,9 @@ func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error
121 var serverError error
122 serverClosed := make(chan struct{})
123 go func() {
124 + if ready != nil {
125 + close(ready)
126 + }
127 serverError = server.Serve(lis)
128 close(serverClosed)
129 }()
docs/changelogs/v0.40.md
+18
@@ -12,6 +12,7 @@ This release was brought to you by the [Shipyard](https://ipshipyard.com/) team.
12 - [🔦 Highlights](#-highlights)
13 - [Routing V1 HTTP API now exposed by default](#routing-v1-http-api-now-exposed-by-default)
14 - [Track total size when adding pins](#track-total-size-when-adding-pins)
15 + - [🚇 Improved `ipfs p2p` tunnels with foreground mode](#-improved-ipfs-p2p-tunnels-with-foreground-mode)
16 - [Improved `ipfs dag stat` output](#improved-ipfs-dag-stat-output)
17 - [Skip bad keys when listing](#skip_bad_keys_when_listing)
18 - [📦️ Dependency updates](#-dependency-updates)
@@ -35,6 +36,23 @@ Example output:
36 Fetched/Processed 336 nodes (83 MB)
37 ```
38
39 +#### 🚇 Improved `ipfs p2p` tunnels with foreground mode
40 +
41 +P2P tunnels can now run like SSH port forwarding: start a tunnel, use it, and it cleans up automatically when you're done.
42 +
43 +The new `--foreground` (`-f`) flag for `ipfs p2p listen` and `ipfs p2p forward` keeps the command running until interrupted. When you Ctrl+C, send SIGTERM, or stop the service, the tunnel is removed automatically:
44 +
45 +```console
46 +$ ipfs p2p listen /x/ssh /ip4/127.0.0.1/tcp/22 --foreground
47 +Listening on /x/ssh, forwarding to /ip4/127.0.0.1/tcp/22, waiting for interrupt...
48 +^C
49 +Received interrupt, removing listener for /x/ssh
50 +```
51 +
52 +Without `--foreground`, commands return immediately and tunnels persist until explicitly closed (existing behavior).
53 +
54 +See [docs/p2p-tunnels.md](https://github.com/ipfs/kubo/blob/master/docs/p2p-tunnels.md) for usage examples.
55 +
56 #### Improved `ipfs dag stat` output
57
58 The `ipfs dag stat` command has been improved for better terminal UX:
docs/config.md
+12
@@ -59,6 +59,7 @@ config file at runtime.
59 - [`Discovery.MDNS.Enabled`](#discoverymdnsenabled)
60 - [`Discovery.MDNS.Interval`](#discoverymdnsinterval)
61 - [`Experimental`](#experimental)
62 + - [`Experimental.Libp2pStreamMounting`](#experimentallibp2pstreammounting)
63 - [`Gateway`](#gateway)
64 - [`Gateway.NoFetch`](#gatewaynofetch)
65 - [`Gateway.NoDNSLink`](#gatewaynodnslink)
@@ -1069,6 +1070,17 @@ in the [new mDNS implementation](https://github.com/libp2p/zeroconf#readme).
1070
1071 Toggle and configure experimental features of Kubo. Experimental features are listed [here](./experimental-features.md).
1072
1073 +### `Experimental.Libp2pStreamMounting`
1074 +
1075 +Enables the `ipfs p2p` commands for tunneling TCP connections through libp2p
1076 +streams, similar to SSH port forwarding.
1077 +
1078 +See [docs/p2p-tunnels.md](p2p-tunnels.md) for usage examples.
1079 +
1080 +Default: `false`
1081 +
1082 +Type: `bool`
1083 +
1084 ## `Gateway`
1085
1086 Options for the HTTP gateway.
docs/experimental-features.md
+7 -86
@@ -199,9 +199,8 @@ configured, the daemon will fail to start.
199
200 ## ipfs p2p
201
202 -Allows tunneling of TCP connections through Libp2p streams. If you've ever used
203 -port forwarding with SSH (the `-L` option in OpenSSH), this feature is quite
204 -similar.
202 +Allows tunneling of TCP connections through libp2p streams, similar to SSH port
203 +forwarding (`ssh -L`).
204
205 ### State
206
@@ -220,98 +219,20 @@ Experimental, will be stabilized in 0.6.0
219 > If you enable this and plan to expose CLI or HTTP RPC to other users or machines,
220 > secure RPC API using [`API.Authorizations`](https://github.com/ipfs/kubo/blob/master/docs/config.md#apiauthorizations) or custom auth middleware.
221
223 -The `p2p` command needs to be enabled in the config:
224 -
222 ```sh
223 > ipfs config --json Experimental.Libp2pStreamMounting true
224 ```
225
226 ### How to use
227
231 -**Netcat example:**
232 -
233 -First, pick a protocol name for your application. Think of the protocol name as
234 -a port number, just significantly more user-friendly. In this example, we're
235 -going to use `/x/kickass/1.0`.
236 -
237 -***Setup:***
238 -
239 -1. A "server" node with peer ID `$SERVER_ID`
240 -2. A "client" node.
241 -
242 -***On the "server" node:***
243 -
244 -First, start your application and have it listen for TCP connections on
245 -port `$APP_PORT`.
246 -
247 -Then, configure the p2p listener by running:
248 -
249 -```sh
250 -> ipfs p2p listen /x/kickass/1.0 /ip4/127.0.0.1/tcp/$APP_PORT
251 -```
252 -
253 -This will configure IPFS to forward all incoming `/x/kickass/1.0` streams to
254 -`127.0.0.1:$APP_PORT` (opening a new connection to `127.0.0.1:$APP_PORT` per
255 -incoming stream.
256 -
257 -***On the "client" node:***
258 -
259 -First, configure the client p2p dialer, so that it forwards all inbound
260 -connections on `127.0.0.1:SOME_PORT` to the server node listening
261 -on `/x/kickass/1.0`.
262 -
263 -```sh
264 -> ipfs p2p forward /x/kickass/1.0 /ip4/127.0.0.1/tcp/$SOME_PORT /p2p/$SERVER_ID
265 -```
266 -
267 -Next, have your application open a connection to `127.0.0.1:$SOME_PORT`. This
268 -connection will be forwarded to the service running on `127.0.0.1:$APP_PORT` on
269 -the remote machine. You can test it with netcat:
270 -
271 -***On "server" node:***
272 -```sh
273 -> nc -v -l -p $APP_PORT
274 -```
275 -
276 -***On "client" node:***
277 -```sh
278 -> nc -v 127.0.0.1 $SOME_PORT
279 -```
280 -
281 -You should now see that a connection has been established and be able to
282 -exchange messages between netcat instances.
283 -
284 -(note that depending on your netcat version you may need to drop the `-v` flag)
285 -
286 -**SSH example**
287 -
288 -**Setup:**
289 -
290 -1. A "server" node with peer ID `$SERVER_ID` and running ssh server on the
291 - default port.
292 -2. A "client" node.
293 -
294 -_you can get `$SERVER_ID` by running `ipfs id -f "<id>\n"`_
295 -
296 -***First, on the "server" node:***
297 -
298 -```sh
299 -ipfs p2p listen /x/ssh /ip4/127.0.0.1/tcp/22
300 -```
301 -
302 -***Then, on "client" node:***
303 -
304 -```sh
305 -ipfs p2p forward /x/ssh /ip4/127.0.0.1/tcp/2222 /p2p/$SERVER_ID
306 -```
307 -
308 -You should now be able to connect to your ssh server through a libp2p connection
309 -with `ssh [user]@127.0.0.1 -p 2222`.
310 -
228 +See [docs/p2p-tunnels.md](p2p-tunnels.md) for usage examples, foreground mode,
229 +and systemd integration.
230
231 ### Road to being a real feature
232
314 -- [ ] More documentation
233 +- [x] More documentation
234 +- [x] `ipfs p2p forward` mode
235 +- [ ] Ability to define tunnels via JSON config, similar to [`Peering.Peers`](https://github.com/ipfs/kubo/blob/master/docs/config.md#peeringpeers), see [kubo#5460](https://github.com/ipfs/kubo/issues/5460)
236
237 ## p2p http proxy
238
docs/p2p-tunnels.md new
+214
@@ -0,0 +1,214 @@
1 +# P2P Tunnels
2 +
3 +Kubo supports tunneling TCP connections through libp2p streams, similar to SSH
4 +port forwarding (`ssh -L`). This allows exposing local services to remote peers
5 +and forwarding remote services to local ports.
6 +
7 +- [Why P2P Tunnels?](#why-p2p-tunnels)
8 +- [Quick Start](#quick-start)
9 +- [Background Mode](#background-mode)
10 +- [Foreground Mode](#foreground-mode)
11 + - [systemd Integration](#systemd-integration)
12 +- [Security Considerations](#security-considerations)
13 +- [Troubleshooting](#troubleshooting)
14 +
15 +## Why P2P Tunnels?
16 +
17 +Unlike traditional SSH tunnels, libp2p-based tunnels do not require:
18 +
19 +- **No public IP or open ports**: The server does not need a static IP address
20 + or port forwarding configured on the router. Connectivity to peers behind NAT
21 + is facilitated by [Direct Connection Upgrade through Relay (DCUtR)](https://github.com/libp2p/specs/blob/master/relay/DCUtR.md),
22 + which enables NAT hole-punching.
23 +
24 +- **No DNS or IP address management**: All you need is the server's PeerID and
25 + an agreed-upon protocol name (e.g., `/x/ssh`). Kubo handles peer discovery
26 + and routing via the [Amino DHT](https://specs.ipfs.tech/routing/kad-dht/).
27 +
28 +- **Simplified firewall rules**: Since connections are established through
29 + libp2p's existing swarm connections, no additional firewall configuration is
30 + needed beyond what Kubo already requires.
31 +
32 +This makes p2p tunnels useful for connecting to machines on home networks,
33 +behind corporate firewalls, or in environments where traditional port forwarding
34 +is not available.
35 +
36 +## Quick Start
37 +
38 +Enable the experimental feature:
39 +
40 +```console
41 +$ ipfs config --json Experimental.Libp2pStreamMounting true
42 +```
43 +
44 +Test with netcat (`nc`) - no services required:
45 +
46 +**On the server:**
47 +
48 +```console
49 +$ ipfs p2p listen /x/test /ip4/127.0.0.1/tcp/9999
50 +$ nc -l -p 9999
51 +```
52 +
53 +**On the client:**
54 +
55 +Replace `$SERVER_ID` with the server's peer ID (get it with `ipfs id -f "<id>\n"`
56 +on the server).
57 +
58 +```console
59 +$ ipfs p2p forward /x/test /ip4/127.0.0.1/tcp/9998 /p2p/$SERVER_ID
60 +$ nc 127.0.0.1 9998
61 +```
62 +
63 +Type in either terminal and the text appears in the other. Use Ctrl+C to exit.
64 +
65 +## Background Mode
66 +
67 +By default, `ipfs p2p listen` and `ipfs p2p forward` register the tunnel with
68 +the daemon and return immediately. The tunnel persists until explicitly closed
69 +with `ipfs p2p close` or the daemon shuts down.
70 +
71 +This example exposes a local SSH server (listening on `localhost:22`) to a
72 +remote peer. The same pattern works for any TCP service.
73 +
74 +**On the server** (the machine running SSH):
75 +
76 +Register a p2p listener that forwards incoming connections to the local SSH
77 +server. The protocol name `/x/ssh` is an arbitrary identifier that both peers
78 +must agree on (the `/x/` prefix is required for custom protocols).
79 +
80 +```console
81 +$ ipfs p2p listen /x/ssh /ip4/127.0.0.1/tcp/22
82 +```
83 +
84 +**On the client:**
85 +
86 +Create a local port (`2222`) that tunnels through libp2p to the server's SSH
87 +service.
88 +
89 +```console
90 +$ ipfs p2p forward /x/ssh /ip4/127.0.0.1/tcp/2222 /p2p/$SERVER_ID
91 +```
92 +
93 +Now connect to SSH through the tunnel:
94 +
95 +```console
96 +$ ssh user@127.0.0.1 -p 2222
97 +```
98 +
99 +**Other services:** To tunnel a different service, change the port and protocol
100 +name. For example, to expose a web server on port 8080, use `/x/mywebapp` and
101 +`/ip4/127.0.0.1/tcp/8080`.
102 +
103 +## Foreground Mode
104 +
105 +Use `--foreground` (`-f`) to block until interrupted. The tunnel is
106 +automatically removed when the command exits:
107 +
108 +```console
109 +$ ipfs p2p listen /x/ssh /ip4/127.0.0.1/tcp/22 --foreground
110 +Listening on /x/ssh, forwarding to /ip4/127.0.0.1/tcp/22, waiting for interrupt...
111 +^C
112 +Received interrupt, removing listener for /x/ssh
113 +```
114 +
115 +The listener/forwarder is automatically removed when:
116 +
117 +- The command receives Ctrl+C or SIGTERM
118 +- `ipfs p2p close` is called
119 +- The daemon shuts down
120 +
121 +This mode is useful for systemd services and scripts that need cleanup on exit.
122 +
123 +### systemd Integration
124 +
125 +The `--foreground` flag enables clean integration with systemd. The examples
126 +below show how to run `ipfs p2p listen` as a user service that starts
127 +automatically when the IPFS daemon is ready.
128 +
129 +Ensure IPFS daemon runs as a systemd user service. See
130 +[misc/README.md](https://github.com/ipfs/kubo/blob/master/misc/README.md#systemd)
131 +for setup instructions and where to place unit files.
132 +
133 +#### P2P listener with path-based activation
134 +
135 +Use a `.path` unit to wait for the daemon's RPC API to be ready before starting
136 +the p2p listener.
137 +
138 +**`ipfs-p2p-tunnel.path`**:
139 +
140 +```systemd
141 +[Unit]
142 +Description=Monitor for IPFS daemon startup
143 +After=ipfs.service
144 +Requires=ipfs.service
145 +
146 +[Path]
147 +PathExists=%h/.ipfs/api
148 +Unit=ipfs-p2p-tunnel.service
149 +
150 +[Install]
151 +WantedBy=default.target
152 +```
153 +
154 +The `%h` specifier expands to the user's home directory. If you use a custom
155 +`IPFS_PATH`, adjust accordingly.
156 +
157 +**`ipfs-p2p-tunnel.service`**:
158 +
159 +```systemd
160 +[Unit]
161 +Description=IPFS p2p tunnel
162 +Requires=ipfs.service
163 +
164 +[Service]
165 +ExecStart=ipfs p2p listen /x/ssh /ip4/127.0.0.1/tcp/22 -f
166 +Restart=on-failure
167 +RestartSec=10
168 +
169 +[Install]
170 +WantedBy=default.target
171 +```
172 +
173 +#### Enabling the services
174 +
175 +```console
176 +$ systemctl --user enable ipfs.service
177 +$ systemctl --user enable ipfs-p2p-tunnel.path
178 +$ systemctl --user start ipfs.service
179 +```
180 +
181 +The path unit monitors `~/.ipfs/api` and starts `ipfs-p2p-tunnel.service`
182 +once the file exists.
183 +
184 +## Security Considerations
185 +
186 +> [!WARNING]
187 +> This feature provides CLI and HTTP RPC users with the ability to set up port
188 +> forwarding for localhost and LAN ports. If you enable this and plan to expose
189 +> CLI or HTTP RPC to other users or machines, secure the RPC API using
190 +> [`API.Authorizations`](https://github.com/ipfs/kubo/blob/master/docs/config.md#apiauthorizations)
191 +> or custom auth middleware.
192 +
193 +## Troubleshooting
194 +
195 +### Foreground listener stops when terminal closes
196 +
197 +When using `--foreground`, the listener stops if the terminal closes. For
198 +persistent foreground listeners, use a systemd service, `nohup`, `tmux`, or
199 +`screen`. Without `--foreground`, the listener persists in the daemon regardless
200 +of terminal state.
201 +
202 +### Connection refused errors
203 +
204 +Verify:
205 +
206 +1. The experimental feature is enabled: `ipfs config Experimental.Libp2pStreamMounting`
207 +2. The listener is active: `ipfs p2p ls`
208 +3. Both peers can connect: `ipfs swarm connect /p2p/$PEER_ID`
209 +
210 +### Persistent tunnel configuration
211 +
212 +There is currently no way to define tunnels in the Kubo JSON config file. Use
213 +`--foreground` mode with a systemd service for persistent tunnels. Support for
214 +configuring tunnels via JSON config may be added in the future (see [kubo#5460](https://github.com/ipfs/kubo/issues/5460) - PRs welcome!).
misc/README.md
+6
@@ -39,6 +39,12 @@ To run this in your user session, save it as `~/.config/systemd/user/ipfs.servic
39 ```
40 Read more about `--user` services here: [wiki.archlinux.org:Systemd ](https://wiki.archlinux.org/index.php/Systemd/User#Automatic_start-up_of_systemd_user_instances)
41
42 +#### P2P tunnel services
43 +
44 +For running `ipfs p2p listen` or `ipfs p2p forward` as systemd services,
45 +see [docs/p2p-tunnels.md](../docs/p2p-tunnels.md) for examples using the
46 +`--foreground` flag and path-based activation.
47 +
48 ### initd
49
50 - Here is a full-featured sample service file: https://github.com/dylanPowers/ipfs-linux-service/blob/master/init.d/ipfs
p2p/listener.go
+8 -6
@@ -20,6 +20,10 @@ type Listener interface {
20
21 // close closes the listener. Does not affect child streams
22 close()
23 +
24 + // Done returns a channel that is closed when the listener is closed.
25 + // This allows callers to detect when a listener has been removed.
26 + Done() <-chan struct{}
27 }
28
29 // Listeners manages a group of Listener implementations,
@@ -73,15 +77,13 @@ func (r *Listeners) Register(l Listener) error {
77 return nil
78 }
79
80 +// Close removes and closes all listeners for which matchFunc returns true.
81 +// Returns the number of listeners closed.
82 func (r *Listeners) Close(matchFunc func(listener Listener) bool) int {
77 - todo := make([]Listener, 0)
83 + var todo []Listener
84 r.Lock()
85 for _, l := range r.Listeners {
80 - if !matchFunc(l) {
81 - continue
82 - }
83 -
84 - if _, ok := r.Listeners[l.key()]; ok {
86 + if matchFunc(l) {
87 delete(r.Listeners, l.key())
88 todo = append(todo, l)
89 }
p2p/local.go
+7
@@ -23,6 +23,7 @@ type localListener struct {
23 peer peer.ID
24
25 listener manet.Listener
26 + done chan struct{}
27 }
28
29 // ForwardLocal creates new P2P stream to a remote listener.
@@ -32,6 +33,7 @@ func (p2p *P2P) ForwardLocal(ctx context.Context, peer peer.ID, proto protocol.I
33 p2p: p2p,
34 proto: proto,
35 peer: peer,
36 + done: make(chan struct{}),
37 }
38
39 maListener, err := manet.Listen(bindAddr)
@@ -98,6 +100,11 @@ func (l *localListener) setupStream(local manet.Conn) {
100
101 func (l *localListener) close() {
102 l.listener.Close()
103 + close(l.done)
104 +}
105 +
106 +func (l *localListener) Done() <-chan struct{} {
107 + return l.done
108 }
109
110 func (l *localListener) Protocol() protocol.ID {
p2p/remote.go
+10 -1
@@ -25,6 +25,8 @@ type remoteListener struct {
25 // reportRemote if set to true makes the handler send '<base58 remote peerid>\n'
26 // to target before any data is forwarded
27 reportRemote bool
28 +
29 + done chan struct{}
30 }
31
32 // ForwardRemote creates new p2p listener.
@@ -36,6 +38,7 @@ func (p2p *P2P) ForwardRemote(ctx context.Context, proto protocol.ID, addr ma.Mu
38 addr: addr,
39
40 reportRemote: reportRemote,
41 + done: make(chan struct{}),
42 }
43
44 if err := p2p.ListenersP2P.Register(listener); err != nil {
@@ -99,7 +102,13 @@ func (l *remoteListener) TargetAddress() ma.Multiaddr {
102 return l.addr
103 }
104
102 -func (l *remoteListener) close() {}
105 +func (l *remoteListener) close() {
106 + close(l.done)
107 +}
108 +
109 +func (l *remoteListener) Done() <-chan struct{} {
110 + return l.done
111 +}
112
113 func (l *remoteListener) key() protocol.ID {
114 return l.proto
test/cli/api_file_test.go new
+104
@@ -0,0 +1,104 @@
1 +package cli
2 +
3 +import (
4 + "net/http"
5 + "os"
6 + "os/exec"
7 + "path/filepath"
8 + "strings"
9 + "testing"
10 + "time"
11 +
12 + "github.com/ipfs/kubo/test/cli/harness"
13 + "github.com/stretchr/testify/require"
14 +)
15 +
16 +// TestAddressFileReady verifies that when address files ($IPFS_PATH/api and
17 +// $IPFS_PATH/gateway) are created, the corresponding HTTP servers are ready
18 +// to accept connections immediately. This prevents race conditions for tools
19 +// like systemd path units that start services when these files appear.
20 +func TestAddressFileReady(t *testing.T) {
21 + t.Parallel()
22 +
23 + t.Run("api file", func(t *testing.T) {
24 + t.Parallel()
25 + h := harness.NewT(t)
26 + node := h.NewNode().Init()
27 +
28 + // Start daemon in background (don't use StartDaemon which waits for API)
29 + res := node.Runner.MustRun(harness.RunRequest{
30 + Path: node.IPFSBin,
31 + Args: []string{"daemon"},
32 + RunFunc: (*exec.Cmd).Start,
33 + })
34 + node.Daemon = res
35 + defer node.StopDaemon()
36 +
37 + // Poll for api file to appear
38 + apiFile := filepath.Join(node.Dir, "api")
39 + var fileExists bool
40 + for i := 0; i < 100; i++ {
41 + if _, err := os.Stat(apiFile); err == nil {
42 + fileExists = true
43 + break
44 + }
45 + time.Sleep(100 * time.Millisecond)
46 + }
47 + require.True(t, fileExists, "api file should be created")
48 +
49 + // Read the api file to get the address
50 + apiAddr, err := node.TryAPIAddr()
51 + require.NoError(t, err)
52 +
53 + // Extract IP and port from multiaddr
54 + ip, err := apiAddr.ValueForProtocol(4) // P_IP4
55 + require.NoError(t, err)
56 + port, err := apiAddr.ValueForProtocol(6) // P_TCP
57 + require.NoError(t, err)
58 +
59 + // Immediately try to use the API - should work on first attempt
60 + url := "http://" + ip + ":" + port + "/api/v0/id"
61 + resp, err := http.Post(url, "", nil)
62 + require.NoError(t, err, "RPC API should be ready immediately when api file exists")
63 + defer resp.Body.Close()
64 + require.Equal(t, http.StatusOK, resp.StatusCode)
65 + })
66 +
67 + t.Run("gateway file", func(t *testing.T) {
68 + t.Parallel()
69 + h := harness.NewT(t)
70 + node := h.NewNode().Init()
71 +
72 + // Start daemon in background
73 + res := node.Runner.MustRun(harness.RunRequest{
74 + Path: node.IPFSBin,
75 + Args: []string{"daemon"},
76 + RunFunc: (*exec.Cmd).Start,
77 + })
78 + node.Daemon = res
79 + defer node.StopDaemon()
80 +
81 + // Poll for gateway file to appear
82 + gatewayFile := filepath.Join(node.Dir, "gateway")
83 + var fileExists bool
84 + for i := 0; i < 100; i++ {
85 + if _, err := os.Stat(gatewayFile); err == nil {
86 + fileExists = true
87 + break
88 + }
89 + time.Sleep(100 * time.Millisecond)
90 + }
91 + require.True(t, fileExists, "gateway file should be created")
92 +
93 + // Read the gateway file to get the URL (already includes http:// prefix)
94 + gatewayURL, err := os.ReadFile(gatewayFile)
95 + require.NoError(t, err)
96 +
97 + // Immediately try to use the Gateway - should work on first attempt
98 + url := strings.TrimSpace(string(gatewayURL)) + "/ipfs/bafkqaaa" // empty file CID
99 + resp, err := http.Get(url)
100 + require.NoError(t, err, "Gateway should be ready immediately when gateway file exists")
101 + defer resp.Body.Close()
102 + require.Equal(t, http.StatusOK, resp.StatusCode)
103 + })
104 +}
test/cli/p2p_test.go new
+430
@@ -0,0 +1,430 @@
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 +}