master
go 389 lines 14 KB
Raw
1 package libp2p
2
3 import (
4 "context"
5 "fmt"
6 "os"
7 "slices"
8 "strings"
9 "time"
10
11 "github.com/ipfs/boxo/autoconf"
12 "github.com/ipfs/go-datastore"
13 "github.com/ipfs/kubo/config"
14 irouting "github.com/ipfs/kubo/routing"
15 dht "github.com/libp2p/go-libp2p-kad-dht"
16 dual "github.com/libp2p/go-libp2p-kad-dht/dual"
17 record "github.com/libp2p/go-libp2p-record"
18 routinghelpers "github.com/libp2p/go-libp2p-routing-helpers"
19 host "github.com/libp2p/go-libp2p/core/host"
20 "github.com/libp2p/go-libp2p/core/peer"
21 routing "github.com/libp2p/go-libp2p/core/routing"
22 basichost "github.com/libp2p/go-libp2p/p2p/host/basic"
23 ma "github.com/multiformats/go-multiaddr"
24 )
25
26 type RoutingOptionArgs struct {
27 Ctx context.Context
28 Host host.Host
29 Datastore datastore.Batching
30 Validator record.Validator
31 BootstrapPeers []peer.AddrInfo
32 OptimisticProvide bool
33 OptimisticProvideJobsPoolSize int
34 LoopbackAddressesOnLanDHT bool
35 }
36
37 type RoutingOption func(args RoutingOptionArgs) (routing.Routing, error)
38
39 var noopRouter = routinghelpers.Null{}
40
41 // EndpointSource tracks where a URL came from to determine appropriate capabilities
42 type EndpointSource struct {
43 URL string
44 SupportsRead bool // came from DelegatedRoutersWithAutoConf (Read operations)
45 SupportsWrite bool // came from DelegatedPublishersWithAutoConf (Write operations)
46 }
47
48 // determineCapabilities determines endpoint capabilities based on URL path and source
49 func determineCapabilities(endpoint EndpointSource) (string, autoconf.EndpointCapabilities, error) {
50 parsed, err := autoconf.DetermineKnownCapabilities(endpoint.URL, endpoint.SupportsRead, endpoint.SupportsWrite)
51 if err != nil {
52 log.Debugf("Skipping endpoint %q: %v", endpoint.URL, err)
53 return "", autoconf.EndpointCapabilities{}, nil // Return empty caps, not error
54 }
55
56 return parsed.BaseURL, parsed.Capabilities, nil
57 }
58
59 // collectAllEndpoints gathers URLs from both router and publisher sources
60 func collectAllEndpoints(cfg *config.Config) []EndpointSource {
61 var endpoints []EndpointSource
62
63 // Get router URLs (Read operations)
64 var routerURLs []string
65 if envRouters := os.Getenv(config.EnvHTTPRouters); envRouters != "" {
66 // Use environment variable override if set (space or comma separated)
67 splitFunc := func(r rune) bool { return r == ',' || r == ' ' }
68 routerURLs = strings.FieldsFunc(envRouters, splitFunc)
69 log.Warnf("Using HTTP routers from %s environment variable instead of config/autoconf: %v", config.EnvHTTPRouters, routerURLs)
70 } else {
71 // Use delegated routers from autoconf
72 routerURLs = cfg.DelegatedRoutersWithAutoConf()
73 // No fallback - if autoconf doesn't provide endpoints, use empty list
74 // This exposes any autoconf issues rather than masking them with hardcoded defaults
75 }
76
77 // Add router URLs to collection
78 for _, url := range routerURLs {
79 endpoints = append(endpoints, EndpointSource{
80 URL: url,
81 SupportsRead: true,
82 SupportsWrite: false,
83 })
84 }
85
86 // Get publisher URLs (Write operations)
87 publisherURLs := cfg.DelegatedPublishersWithAutoConf()
88
89 // Add publisher URLs, merging with existing router URLs if they match
90 for _, url := range publisherURLs {
91 found := false
92 for i, existing := range endpoints {
93 if existing.URL == url {
94 endpoints[i].SupportsWrite = true
95 found = true
96 break
97 }
98 }
99 if !found {
100 endpoints = append(endpoints, EndpointSource{
101 URL: url,
102 SupportsRead: false,
103 SupportsWrite: true,
104 })
105 }
106 }
107
108 return endpoints
109 }
110
111 func constructDefaultHTTPRouters(cfg *config.Config, addrFunc func() []ma.Multiaddr) ([]*routinghelpers.ParallelRouter, error) {
112 var routers []*routinghelpers.ParallelRouter
113 httpRetrievalEnabled := cfg.HTTPRetrieval.Enabled.WithDefault(config.DefaultHTTPRetrievalEnabled)
114
115 // Collect URLs from both router and publisher sources
116 endpoints := collectAllEndpoints(cfg)
117
118 // Group endpoints by origin (base URL) and aggregate capabilities
119 originCapabilities := make(map[string]autoconf.EndpointCapabilities)
120 for _, endpoint := range endpoints {
121 // Parse endpoint and determine capabilities based on source
122 baseURL, capabilities, err := determineCapabilities(endpoint)
123 if err != nil {
124 return nil, fmt.Errorf("failed to parse endpoint %q: %w", endpoint.URL, err)
125 }
126
127 // Aggregate capabilities for this origin
128 existing := originCapabilities[baseURL]
129 existing.Merge(capabilities)
130 originCapabilities[baseURL] = existing
131 }
132
133 // Create single HTTP router and composer per origin
134 for baseURL, capabilities := range originCapabilities {
135 // Construct HTTP router using base URL (without path)
136 httpRouter, err := irouting.ConstructHTTPRouter(baseURL, cfg.Identity.PeerID, addrFunc, cfg.Identity.PrivKey, httpRetrievalEnabled)
137 if err != nil {
138 return nil, err
139 }
140
141 // Configure router operations based on aggregated capabilities
142 // https://specs.ipfs.tech/routing/http-routing-v1/
143 composer := &irouting.Composer{
144 GetValueRouter: noopRouter, // Default disabled, enabled below based on capabilities
145 PutValueRouter: noopRouter, // Default disabled, enabled below based on capabilities
146 ProvideRouter: noopRouter, // we don't have spec for sending provides to /routing/v1 (revisit once https://github.com/ipfs/specs/pull/378 or similar is ratified)
147 FindPeersRouter: noopRouter, // Default disabled, enabled below based on capabilities
148 FindProvidersRouter: noopRouter, // Default disabled, enabled below based on capabilities
149 }
150
151 // Enable specific capabilities
152 if capabilities.IPNSGet {
153 composer.GetValueRouter = httpRouter // GET /routing/v1/ipns for IPNS resolution
154 }
155 if capabilities.IPNSPut {
156 composer.PutValueRouter = httpRouter // PUT /routing/v1/ipns for IPNS publishing
157 }
158 if capabilities.Peers {
159 composer.FindPeersRouter = httpRouter // GET /routing/v1/peers
160 }
161 if capabilities.Providers {
162 composer.FindProvidersRouter = httpRouter // GET /routing/v1/providers
163 }
164
165 // Handle special cases and backward compatibility
166 if baseURL == config.CidContactRoutingURL {
167 // Special-case: cid.contact only supports /routing/v1/providers/cid endpoint
168 // Override any capabilities detected from URL path to ensure only providers is enabled
169 // TODO: Consider moving this to configuration or removing once cid.contact adds more capabilities
170 composer.GetValueRouter = noopRouter
171 composer.PutValueRouter = noopRouter
172 composer.ProvideRouter = noopRouter
173 composer.FindPeersRouter = noopRouter
174 composer.FindProvidersRouter = httpRouter // Only providers supported
175 }
176
177 routers = append(routers, &routinghelpers.ParallelRouter{
178 Router: composer,
179 IgnoreError: true, // https://github.com/ipfs/kubo/pull/9475#discussion_r1042507387
180 Timeout: 15 * time.Second, // 5x server value from https://github.com/ipfs/kubo/pull/9475#discussion_r1042428529
181 DoNotWaitForSearchValue: true,
182 ExecuteAfter: 0,
183 })
184 }
185 return routers, nil
186 }
187
188 // ConstructDelegatedOnlyRouting returns routers used when Routing.Type is set to "delegated"
189 // This provides HTTP-only routing without DHT, using only delegated routers and IPNS publishers.
190 // Useful for environments where DHT connectivity is not available or desired
191 func ConstructDelegatedOnlyRouting(cfg *config.Config) RoutingOption {
192 return func(args RoutingOptionArgs) (routing.Routing, error) {
193 // Use only HTTP routers (includes both read and write capabilities) - no DHT
194 var routers []*routinghelpers.ParallelRouter
195
196 // Add HTTP delegated routers (includes both router and publisher capabilities)
197 addrFunc := httpRouterAddrFunc(args.Host, cfg.Addresses)
198 httpRouters, err := constructDefaultHTTPRouters(cfg, addrFunc)
199 if err != nil {
200 return nil, err
201 }
202 routers = append(routers, httpRouters...)
203
204 // Validate that we have at least one router configured
205 if len(routers) == 0 {
206 return nil, fmt.Errorf("no delegated routers or publishers configured for 'delegated' routing mode")
207 }
208
209 routing := routinghelpers.NewComposableParallel(routers)
210 return routing, nil
211 }
212 }
213
214 // ConstructDefaultRouting returns routers used when Routing.Type is unset or set to "auto"
215 func ConstructDefaultRouting(cfg *config.Config, routingOpt RoutingOption) RoutingOption {
216 return func(args RoutingOptionArgs) (routing.Routing, error) {
217 // Defined routers will be queried in parallel (optimizing for response speed)
218 // Different trade-offs can be made by setting Routing.Type = "custom" with own Routing.Routers
219 var routers []*routinghelpers.ParallelRouter
220
221 dhtRouting, err := routingOpt(args)
222 if err != nil {
223 return nil, err
224 }
225 routers = append(routers, &routinghelpers.ParallelRouter{
226 Router: dhtRouting,
227 IgnoreError: false,
228 DoNotWaitForSearchValue: true,
229 ExecuteAfter: 0,
230 })
231
232 addrFunc := httpRouterAddrFunc(args.Host, cfg.Addresses)
233 httpRouters, err := constructDefaultHTTPRouters(cfg, addrFunc)
234 if err != nil {
235 return nil, err
236 }
237
238 routers = append(routers, httpRouters...)
239
240 routing := routinghelpers.NewComposableParallel(routers)
241 return routing, nil
242 }
243 }
244
245 // constructDHTRouting is used when Routing.Type = "dht"
246 func constructDHTRouting(mode dht.ModeOpt) RoutingOption {
247 return func(args RoutingOptionArgs) (routing.Routing, error) {
248 dhtOpts := []dht.Option{
249 dht.Concurrency(10),
250 dht.Mode(mode),
251 dht.Datastore(args.Datastore),
252 dht.Validator(args.Validator),
253 }
254 if args.OptimisticProvide {
255 dhtOpts = append(dhtOpts, dht.EnableOptimisticProvide())
256 }
257 if args.OptimisticProvideJobsPoolSize != 0 {
258 dhtOpts = append(dhtOpts, dht.OptimisticProvideJobsPoolSize(args.OptimisticProvideJobsPoolSize))
259 }
260 wanOptions := []dht.Option{
261 dht.BootstrapPeers(args.BootstrapPeers...),
262 }
263 // In stub mode, allow loopback peers in the WAN routing
264 // table so Provide/PutValue work with ephemeral test peers.
265 if os.Getenv("TEST_DHT_STUB") != "" {
266 wanOptions = append(wanOptions,
267 dht.AddressFilter(nil),
268 dht.QueryFilter(func(_ any, _ peer.AddrInfo) bool { return true }),
269 dht.RoutingTableFilter(func(_ any, _ peer.ID) bool { return true }),
270 dht.RoutingTablePeerDiversityFilter(nil),
271 )
272 }
273 lanOptions := []dht.Option{}
274 if args.LoopbackAddressesOnLanDHT {
275 lanOptions = append(lanOptions, dht.AddressFilter(nil))
276 }
277 d, err := dual.New(
278 args.Ctx, args.Host,
279 dual.DHTOption(dhtOpts...),
280 dual.WanDHTOption(wanOptions...),
281 dual.LanDHTOption(lanOptions...),
282 )
283 if err != nil {
284 return nil, err
285 }
286 return d, nil
287 }
288 }
289
290 // ConstructDelegatedRouting is used when Routing.Type = "custom"
291 func ConstructDelegatedRouting(routers config.Routers, methods config.Methods, peerID string, addrs config.Addresses, privKey string, httpRetrieval bool) RoutingOption {
292 return func(args RoutingOptionArgs) (routing.Routing, error) {
293 addrFunc := httpRouterAddrFunc(args.Host, addrs)
294 return irouting.Parse(routers, methods,
295 &irouting.ExtraDHTParams{
296 BootstrapPeers: args.BootstrapPeers,
297 Host: args.Host,
298 Validator: args.Validator,
299 Datastore: args.Datastore,
300 Context: args.Ctx,
301 },
302 &irouting.ExtraHTTPParams{
303 PeerID: peerID,
304 AddrFunc: addrFunc,
305 PrivKeyB64: privKey,
306 HTTPRetrieval: httpRetrieval,
307 },
308 )
309 }
310 }
311
312 func constructNilRouting(_ RoutingOptionArgs) (routing.Routing, error) {
313 return routinghelpers.Null{}, nil
314 }
315
316 var (
317 DHTOption RoutingOption = constructDHTRouting(dht.ModeAuto)
318 DHTClientOption = constructDHTRouting(dht.ModeClient)
319 DHTServerOption = constructDHTRouting(dht.ModeServer)
320 NilRouterOption = constructNilRouting
321 )
322
323 // confirmedAddrsHost matches libp2p hosts that support AutoNAT V2 address confirmation.
324 type confirmedAddrsHost interface {
325 ConfirmedAddrs() (reachable, unreachable, unknown []ma.Multiaddr)
326 }
327
328 // Compile-time check: BasicHost must satisfy confirmedAddrsHost.
329 // ConfirmedAddrs is not part of the core host.Host interface and is marked
330 // experimental in go-libp2p. If BasicHost ever drops or changes this method,
331 // this assertion will fail at build time. In that case, update
332 // httpRouterAddrFunc (this file) and the swarm autonat command
333 // (core/commands/swarm_addrs_autonat.go) which both type-assert to this
334 // interface.
335 var _ confirmedAddrsHost = (*basichost.BasicHost)(nil)
336
337 // httpRouterAddrFunc returns a function that resolves provider addresses for
338 // HTTP routers at provide-time.
339 //
340 // Resolution logic:
341 // - If Announce is set, use it as a static override (no dynamic resolution).
342 // - Otherwise, prefer AutoNAT V2 confirmed reachable addresses when available,
343 // falling back to host.Addrs() which resolves 0.0.0.0/:: Swarm binds to
344 // concrete interface addresses and applies the libp2p AddrsFactory
345 // (Addresses.NoAnnounce CIDR filters and Swarm.AddrFilters).
346 // - AppendAnnounce addresses are always appended.
347 func httpRouterAddrFunc(h host.Host, cfgAddrs config.Addresses) func() []ma.Multiaddr {
348 appendAddrs := parseMultiaddrs(cfgAddrs.AppendAnnounce)
349
350 // If Announce is explicitly set, use it as a static override.
351 if len(cfgAddrs.Announce) > 0 {
352 staticAddrs := slices.Concat(parseMultiaddrs(cfgAddrs.Announce), appendAddrs)
353 return func() []ma.Multiaddr { return staticAddrs }
354 }
355
356 ch, hasConfirmed := h.(confirmedAddrsHost)
357 return func() []ma.Multiaddr {
358 if hasConfirmed {
359 reachable, _, _ := ch.ConfirmedAddrs()
360 if len(reachable) > 0 {
361 if len(appendAddrs) == 0 {
362 return reachable
363 }
364 return slices.Concat(reachable, appendAddrs)
365 }
366 }
367 // Fallback: host.Addrs() resolves wildcard binds (0.0.0.0, ::) to
368 // concrete interface addresses and applies the libp2p AddrsFactory,
369 // which is where Addresses.NoAnnounce CIDR filtering happens.
370 hostAddrs := h.Addrs()
371 if len(appendAddrs) == 0 {
372 return hostAddrs
373 }
374 return slices.Concat(hostAddrs, appendAddrs)
375 }
376 }
377
378 func parseMultiaddrs(strs []string) []ma.Multiaddr {
379 addrs := make([]ma.Multiaddr, 0, len(strs))
380 for _, s := range strs {
381 a, err := ma.NewMultiaddr(s)
382 if err != nil {
383 log.Errorf("ignoring invalid multiaddr %q: %s", s, err)
384 continue
385 }
386 addrs = append(addrs, a)
387 }
388 return addrs
389 }