master
go 468 lines 16.5 KB
Raw
1 package node
2
3 import (
4 "context"
5 "errors"
6 "fmt"
7 "regexp"
8 "strings"
9 "time"
10
11 blockstore "github.com/ipfs/boxo/blockstore"
12 offline "github.com/ipfs/boxo/exchange/offline"
13 uio "github.com/ipfs/boxo/ipld/unixfs/io"
14 util "github.com/ipfs/boxo/util"
15 "github.com/ipfs/go-log/v2"
16 "github.com/ipfs/kubo/config"
17 "github.com/ipfs/kubo/core/node/libp2p"
18 "github.com/ipfs/kubo/p2p"
19 pubsub "github.com/libp2p/go-libp2p-pubsub"
20 "github.com/libp2p/go-libp2p-pubsub/timecache"
21 "github.com/libp2p/go-libp2p/core/peer"
22 rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
23 "go.uber.org/fx"
24 )
25
26 var logger = log.Logger("core:constructor")
27
28 var BaseLibP2P = fx.Options(
29 fx.Provide(libp2p.PNet),
30 fx.Provide(libp2p.ConnectionManager),
31 fx.Provide(libp2p.Host),
32 fx.Provide(libp2p.MultiaddrResolver),
33
34 fx.Provide(libp2p.DiscoveryHandler),
35
36 fx.Invoke(libp2p.PNetChecker),
37 )
38
39 func LibP2P(bcfg *BuildCfg, cfg *config.Config, userResourceOverrides rcmgr.PartialLimitConfig) fx.Option {
40 var connmgr fx.Option
41
42 // set connmgr based on Swarm.ConnMgr.Type
43 connMgrType := cfg.Swarm.ConnMgr.Type.WithDefault(config.DefaultConnMgrType)
44 switch connMgrType {
45 case "none":
46 connmgr = fx.Options() // noop
47 case "", "basic":
48 grace := cfg.Swarm.ConnMgr.GracePeriod.WithDefault(config.DefaultConnMgrGracePeriod)
49 low := int(cfg.Swarm.ConnMgr.LowWater.WithDefault(config.DefaultConnMgrLowWater))
50 high := int(cfg.Swarm.ConnMgr.HighWater.WithDefault(config.DefaultConnMgrHighWater))
51 silence := cfg.Swarm.ConnMgr.SilencePeriod.WithDefault(config.DefaultConnMgrSilencePeriod)
52 connmgr = fx.Provide(libp2p.ConnectionManager(low, high, grace, silence))
53
54 default:
55 return fx.Error(fmt.Errorf("unrecognized Swarm.ConnMgr.Type: %q", connMgrType))
56 }
57
58 // parse PubSub config
59
60 ps, disc := fx.Options(), fx.Options()
61 if bcfg.getOpt("pubsub") || bcfg.getOpt("ipnsps") {
62 disc = fx.Provide(libp2p.TopicDiscovery())
63
64 var pubsubOptions []pubsub.Option
65 pubsubOptions = append(
66 pubsubOptions,
67 pubsub.WithMessageSigning(!cfg.Pubsub.DisableSigning),
68 pubsub.WithSeenMessagesTTL(cfg.Pubsub.SeenMessagesTTL.WithDefault(pubsub.TimeCacheDuration)),
69 )
70
71 var seenMessagesStrategy timecache.Strategy
72 configSeenMessagesStrategy := cfg.Pubsub.SeenMessagesStrategy.WithDefault(config.DefaultSeenMessagesStrategy)
73 switch configSeenMessagesStrategy {
74 case config.LastSeenMessagesStrategy:
75 seenMessagesStrategy = timecache.Strategy_LastSeen
76 case config.FirstSeenMessagesStrategy:
77 seenMessagesStrategy = timecache.Strategy_FirstSeen
78 default:
79 return fx.Error(fmt.Errorf("unsupported Pubsub.SeenMessagesStrategy %q", configSeenMessagesStrategy))
80 }
81 pubsubOptions = append(pubsubOptions, pubsub.WithSeenMessagesStrategy(seenMessagesStrategy))
82
83 switch cfg.Pubsub.Router {
84 case "":
85 fallthrough
86 case "gossipsub":
87 ps = fx.Provide(libp2p.GossipSub(pubsubOptions...))
88 case "floodsub":
89 ps = fx.Provide(libp2p.FloodSub(pubsubOptions...))
90 default:
91 return fx.Error(fmt.Errorf("unknown pubsub router %s", cfg.Pubsub.Router))
92 }
93 }
94
95 autonat := fx.Options()
96
97 switch cfg.AutoNAT.ServiceMode {
98 default:
99 panic("BUG: unhandled autonat service mode")
100 case config.AutoNATServiceDisabled:
101 case config.AutoNATServiceUnset:
102 // TODO
103 //
104 // We're enabling the AutoNAT service by default on _all_ nodes
105 // for the moment.
106 //
107 // We should consider disabling it by default if the dht is set
108 // to dhtclient.
109 fallthrough
110 case config.AutoNATServiceEnabled:
111 autonat = fx.Provide(libp2p.AutoNATService(cfg.AutoNAT.Throttle, false))
112 case config.AutoNATServiceEnabledV1Only:
113 autonat = fx.Provide(libp2p.AutoNATService(cfg.AutoNAT.Throttle, true))
114 }
115
116 enableTCPTransport := cfg.Swarm.Transports.Network.TCP.WithDefault(true)
117 enableWebsocketTransport := cfg.Swarm.Transports.Network.Websocket.WithDefault(true)
118 enableRelayTransport := cfg.Swarm.Transports.Network.Relay.WithDefault(true) // nolint
119 enableRelayService := cfg.Swarm.RelayService.Enabled.WithDefault(enableRelayTransport)
120 enableRelayClient := cfg.Swarm.RelayClient.Enabled.WithDefault(enableRelayTransport)
121 enableAutoTLS := cfg.AutoTLS.Enabled.WithDefault(config.DefaultAutoTLSEnabled)
122 enableAutoWSS := cfg.AutoTLS.AutoWSS.WithDefault(config.DefaultAutoWSS)
123 atlsLog := log.Logger("autotls")
124
125 // Log error when relay subsystem could not be initialized due to missing dependency
126 if !enableRelayTransport {
127 if enableRelayService {
128 logger.Fatal("Failed to enable `Swarm.RelayService`, it requires `Swarm.Transports.Network.Relay` to be true.")
129 }
130 if enableRelayClient {
131 logger.Fatal("Failed to enable `Swarm.RelayClient`, it requires `Swarm.Transports.Network.Relay` to be true.")
132 }
133 }
134
135 switch {
136 case enableAutoTLS && enableTCPTransport && enableWebsocketTransport:
137 // AutoTLS for Secure WebSockets: ensure WSS listeners are in place (manual or automatic)
138 wssWildcard := fmt.Sprintf("/tls/sni/*.%s/ws", cfg.AutoTLS.DomainSuffix.WithDefault(config.DefaultDomainSuffix))
139 wssWildcardPresent := false
140 customWsPresent := false
141 customWsRegex := regexp.MustCompile(`/wss?$`)
142 tcpRegex := regexp.MustCompile(`/tcp/\d+$`)
143
144 // inspect listeners defined in config at Addresses.Swarm
145 var tcpListeners []string
146 for _, listener := range cfg.Addresses.Swarm {
147 // detect if user manually added /tls/sni/.../ws listener matching AutoTLS.DomainSuffix
148 if strings.Contains(listener, wssWildcard) {
149 atlsLog.Infof("found compatible wildcard listener in Addresses.Swarm. AutoTLS will be used on %s", listener)
150 wssWildcardPresent = true
151 break
152 }
153 // detect if user manually added own /ws or /wss listener that is
154 // not related to AutoTLS feature
155 if customWsRegex.MatchString(listener) {
156 atlsLog.Infof("found custom /ws listener set by user in Addresses.Swarm. AutoTLS will not be used on %s.", listener)
157 customWsPresent = true
158 break
159 }
160 // else, remember /tcp listeners that can be reused for /tls/sni/../ws
161 if tcpRegex.MatchString(listener) {
162 tcpListeners = append(tcpListeners, listener)
163 }
164 }
165
166 // Append AutoTLS's wildcard listener
167 // if no manual /ws listener was set by the user
168 if enableAutoWSS && !wssWildcardPresent && !customWsPresent {
169 if len(tcpListeners) == 0 {
170 logger.Error("Invalid configuration, AutoTLS will be disabled: AutoTLS.AutoWSS=true requires at least one /tcp listener present in Addresses.Swarm, see https://github.com/ipfs/kubo/blob/master/docs/config.md#autotls")
171 enableAutoTLS = false
172 }
173 for _, tcpListener := range tcpListeners {
174 wssListener := tcpListener + wssWildcard
175 cfg.Addresses.Swarm = append(cfg.Addresses.Swarm, wssListener)
176 atlsLog.Infof("appended AutoWSS listener: %s", wssListener)
177 }
178 }
179
180 if !wssWildcardPresent && !enableAutoWSS {
181 logger.Error(fmt.Sprintf("Invalid configuration, AutoTLS will be disabled: AutoTLS.Enabled=true requires a /tcp listener ending with %q to be present in Addresses.Swarm or AutoTLS.AutoWSS=true, see https://github.com/ipfs/kubo/blob/master/docs/config.md#autotls", wssWildcard))
182 enableAutoTLS = false
183 }
184 case enableAutoTLS && !enableTCPTransport:
185 logger.Error("Invalid configuration: AutoTLS.Enabled=true requires Swarm.Transports.Network.TCP to be true as well. AutoTLS will be disabled.")
186 enableAutoTLS = false
187 case enableAutoTLS && !enableWebsocketTransport:
188 logger.Error("Invalid configuration: AutoTLS.Enabled=true requires Swarm.Transports.Network.Websocket to be true as well. AutoTLS will be disabled.")
189 enableAutoTLS = false
190 }
191
192 // Gather all the options
193 opts := fx.Options(
194 BaseLibP2P,
195
196 // identify's AgentVersion (incl. optional agent-version-suffix)
197 fx.Provide(libp2p.UserAgent()),
198
199 // Services (resource management)
200 fx.Provide(libp2p.ResourceManager(bcfg.Repo.Path(), cfg.Swarm, userResourceOverrides)),
201 maybeProvide(libp2p.P2PForgeCertMgr(bcfg.Repo.Path(), cfg.AutoTLS, atlsLog), enableAutoTLS),
202 maybeInvoke(libp2p.StartP2PAutoTLS, enableAutoTLS),
203 fx.Provide(libp2p.AddrFilters(cfg.Swarm.AddrFilters)),
204 fx.Invoke(libp2p.MonitorDeadListeners(cfg.Swarm.AddrFilters, cfg.Addresses.NoAnnounce)),
205 fx.Provide(libp2p.AddrsFactory(cfg.Addresses.Announce, cfg.Addresses.AppendAnnounce, cfg.Addresses.NoAnnounce)),
206 fx.Provide(libp2p.SmuxTransport(cfg.Swarm.Transports)),
207 fx.Provide(libp2p.RelayTransport(enableRelayTransport)),
208 fx.Provide(libp2p.RelayService(enableRelayService, cfg.Swarm.RelayService)),
209 fx.Provide(libp2p.Transports(cfg.Swarm.Transports)),
210 fx.Provide(libp2p.ListenOn(cfg.Addresses.Swarm)),
211 fx.Invoke(libp2p.SetupDiscovery(cfg.Discovery.MDNS.Enabled)),
212 fx.Provide(libp2p.ForceReachability(cfg.Internal.Libp2pForceReachability)),
213 fx.Provide(libp2p.HolePunching(cfg.Swarm.EnableHolePunching, enableRelayClient)),
214
215 fx.Provide(libp2p.Security(!bcfg.DisableEncryptedConnections, cfg.Swarm.Transports)),
216
217 fx.Provide(libp2p.Routing),
218 fx.Provide(libp2p.ContentRouting),
219 fx.Provide(libp2p.ContentDiscovery),
220
221 fx.Provide(libp2p.BaseRouting(cfg)),
222 maybeProvide(libp2p.PubsubRouter, bcfg.getOpt("ipnsps")),
223
224 maybeProvide(libp2p.BandwidthCounter, !cfg.Swarm.DisableBandwidthMetrics),
225 maybeProvide(libp2p.NatPortMap, !cfg.Swarm.DisableNatPortMap),
226 libp2p.MaybeAutoRelay(cfg.Swarm.RelayClient.StaticRelays, cfg.Peering, enableRelayClient),
227 autonat,
228 connmgr,
229 ps,
230 disc,
231 )
232
233 return opts
234 }
235
236 // Storage groups units which setup datastore based persistence and blockstore layers
237 func Storage(bcfg *BuildCfg, cfg *config.Config) fx.Option {
238 cacheOpts := blockstore.DefaultCacheOpts()
239 cacheOpts.HasBloomFilterSize = cfg.Datastore.BloomFilterSize
240 cacheOpts.HasTwoQueueCacheSize = int(cfg.Datastore.BlockKeyCacheSize.WithDefault(config.DefaultBlockKeyCacheSize))
241 if !bcfg.Permanent {
242 cacheOpts.HasBloomFilterSize = 0
243 }
244
245 finalBstore := fx.Provide(GcBlockstoreCtor)
246 if cfg.Experimental.FilestoreEnabled || cfg.Experimental.UrlstoreEnabled {
247 finalBstore = fx.Provide(FilestoreBlockstoreCtor(
248 cfg.Provide.Strategy.WithDefault(config.DefaultProvideStrategy),
249 ))
250 }
251
252 return fx.Options(
253 fx.Provide(RepoConfig),
254 fx.Provide(Datastore),
255 fx.Provide(BaseBlockstoreCtor(
256 cacheOpts,
257 cfg.Datastore.HashOnRead,
258 cfg.Datastore.WriteThrough.WithDefault(config.DefaultWriteThrough),
259 cfg.Provide.Strategy.WithDefault(config.DefaultProvideStrategy),
260 )),
261 finalBstore,
262 )
263 }
264
265 // Identity groups units providing cryptographic identity
266 func Identity(cfg *config.Config) fx.Option {
267 // PeerID
268
269 cid := cfg.Identity.PeerID
270 if cid == "" {
271 return fx.Error(errors.New("identity was not set in config (was 'ipfs init' run?)"))
272 }
273 if len(cid) == 0 {
274 return fx.Error(errors.New("no peer ID in config! (was 'ipfs init' run?)"))
275 }
276
277 id, err := peer.Decode(cid)
278 if err != nil {
279 return fx.Error(fmt.Errorf("peer ID invalid: %s", err))
280 }
281
282 // Private Key
283
284 if cfg.Identity.PrivKey == "" {
285 return fx.Options( // No PK (usually in tests)
286 fx.Provide(PeerID(id)),
287 fx.Provide(libp2p.Peerstore),
288 )
289 }
290
291 sk, err := cfg.Identity.DecodePrivateKey("passphrase todo!")
292 if err != nil {
293 return fx.Error(err)
294 }
295
296 return fx.Options( // Full identity
297 fx.Provide(PeerID(id)),
298 fx.Provide(PrivateKey(sk)),
299 fx.Provide(libp2p.Peerstore),
300
301 fx.Invoke(libp2p.PstoreAddSelfKeys),
302 )
303 }
304
305 // IPNS groups namesys related units
306 var IPNS = fx.Options(
307 fx.Provide(RecordValidator),
308 )
309
310 // Online groups online-only units
311 func Online(bcfg *BuildCfg, cfg *config.Config, userResourceOverrides rcmgr.PartialLimitConfig) fx.Option {
312 // Namesys params
313
314 ipnsCacheSize := cfg.Ipns.ResolveCacheSize
315 if ipnsCacheSize == 0 {
316 ipnsCacheSize = DefaultIpnsCacheSize
317 }
318 if ipnsCacheSize < 0 {
319 return fx.Error(errors.New("cannot specify negative resolve cache size"))
320 }
321
322 // Republisher params
323
324 var repubPeriod, recordLifetime time.Duration
325
326 if cfg.Ipns.RepublishPeriod != "" {
327 d, err := time.ParseDuration(cfg.Ipns.RepublishPeriod)
328 if err != nil {
329 return fx.Error(fmt.Errorf("failure to parse config setting IPNS.RepublishPeriod: %s", err))
330 }
331
332 if !util.Debug && (d < time.Minute || d > (time.Hour*24)) {
333 return fx.Error(fmt.Errorf("config setting IPNS.RepublishPeriod is not between 1min and 1day: %s", d))
334 }
335
336 repubPeriod = d
337 }
338
339 if cfg.Ipns.RecordLifetime != "" {
340 d, err := time.ParseDuration(cfg.Ipns.RecordLifetime)
341 if err != nil {
342 return fx.Error(fmt.Errorf("failure to parse config setting IPNS.RecordLifetime: %s", err))
343 }
344
345 recordLifetime = d
346 }
347
348 isBitswapLibp2pEnabled := cfg.Bitswap.Libp2pEnabled.WithDefault(config.DefaultBitswapLibp2pEnabled)
349 isBitswapServerEnabled := cfg.Bitswap.ServerEnabled.WithDefault(config.DefaultBitswapServerEnabled)
350 isHTTPRetrievalEnabled := cfg.HTTPRetrieval.Enabled.WithDefault(config.DefaultHTTPRetrievalEnabled)
351
352 // The Provide system handles both new CID announcements and periodic
353 // re-announcements. Provide.Enabled=false fully disables it.
354 // Provide.DHT.Interval=0 disables only the periodic reprovide schedule;
355 // new CIDs still announce via fast-provide-root and 'ipfs provide once'.
356 isProviderEnabled := cfg.Provide.Enabled.WithDefault(config.DefaultProvideEnabled)
357
358 return fx.Options(
359 fx.Provide(BitswapOptions(cfg)),
360 fx.Provide(Bitswap(isBitswapServerEnabled, isBitswapLibp2pEnabled, isHTTPRetrievalEnabled)),
361 fx.Provide(OnlineExchange(isBitswapLibp2pEnabled)),
362 fx.Provide(DNSResolver),
363 fx.Provide(Namesys(ipnsCacheSize, cfg.Ipns.MaxCacheTTL.WithDefault(config.DefaultIpnsMaxCacheTTL))),
364 fx.Provide(Peering),
365 PeerWith(cfg.Peering.Peers...),
366
367 fx.Invoke(IpnsRepublisher(repubPeriod, recordLifetime)),
368
369 fx.Provide(p2p.New),
370
371 LibP2P(bcfg, cfg, userResourceOverrides),
372 OnlineProviders(isProviderEnabled, cfg),
373 )
374 }
375
376 // Offline groups offline alternatives to Online units
377 func Offline(cfg *config.Config) fx.Option {
378 return fx.Options(
379 fx.Provide(offline.Exchange),
380 fx.Provide(DNSResolver),
381 fx.Provide(Namesys(0, 0)),
382 fx.Provide(libp2p.Routing),
383 fx.Provide(libp2p.ContentRouting),
384 fx.Provide(libp2p.OfflineRouting),
385 fx.Provide(libp2p.ContentDiscovery),
386 OfflineProviders(),
387 )
388 }
389
390 // Core groups basic IPFS services
391 var Core = fx.Options(
392 fx.Provide(Dag),
393 fx.Provide(FetcherConfig),
394 fx.Provide(PathResolverConfig),
395 )
396
397 func Networked(bcfg *BuildCfg, cfg *config.Config, userResourceOverrides rcmgr.PartialLimitConfig) fx.Option {
398 if bcfg.Online {
399 return Online(bcfg, cfg, userResourceOverrides)
400 }
401 return Offline(cfg)
402 }
403
404 // IPFS builds a group of fx Options based on the passed BuildCfg
405 func IPFS(ctx context.Context, bcfg *BuildCfg) fx.Option {
406 if bcfg == nil {
407 bcfg = new(BuildCfg)
408 }
409
410 bcfgOpts, cfg := bcfg.options(ctx)
411 if cfg == nil {
412 return bcfgOpts // error
413 }
414
415 userResourceOverrides, err := bcfg.Repo.UserResourceOverrides()
416 if err != nil {
417 return fx.Error(err)
418 }
419
420 // Migrate users of deprecated Experimental.ShardingEnabled flag
421 if cfg.Experimental.ShardingEnabled {
422 logger.Fatal("The `Experimental.ShardingEnabled` field is no longer used, please remove it from the config. Use Import.UnixFSHAMTDirectorySizeThreshold instead.")
423 }
424 if !cfg.Internal.UnixFSShardingSizeThreshold.IsDefault() {
425 msg := "The `Internal.UnixFSShardingSizeThreshold` field was renamed to `Import.UnixFSHAMTDirectorySizeThreshold`. Please update your config.\n"
426 if !cfg.Import.UnixFSHAMTDirectorySizeThreshold.IsDefault() {
427 logger.Fatal(msg) // conflicting values, hard fail
428 }
429 logger.Error(msg)
430 // Migrate the old OptionalString value to the new OptionalBytes field.
431 // Since OptionalBytes embeds OptionalString, we can construct it directly
432 // with the old value, preserving the user's original string (e.g., "256KiB").
433 cfg.Import.UnixFSHAMTDirectorySizeThreshold = config.OptionalBytes{OptionalString: *cfg.Internal.UnixFSShardingSizeThreshold}
434 }
435
436 // Validate Import configuration
437 if err := config.ValidateImportConfig(&cfg.Import); err != nil {
438 return fx.Error(err)
439 }
440
441 // Validate Provide configuration
442 if err := config.ValidateProvideConfig(&cfg.Provide); err != nil {
443 return fx.Error(err)
444 }
445
446 // Directory sharding settings from Import config.
447 // These globals affect both `ipfs add` and MFS (`ipfs files` API).
448 shardSizeThreshold := cfg.Import.UnixFSHAMTDirectorySizeThreshold.WithDefault(config.DefaultUnixFSHAMTDirectorySizeThreshold)
449 shardMaxFanout := cfg.Import.UnixFSHAMTDirectoryMaxFanout.WithDefault(config.DefaultUnixFSHAMTDirectoryMaxFanout)
450 uio.HAMTShardingSize = int(shardSizeThreshold)
451 uio.DefaultShardWidth = int(shardMaxFanout)
452 uio.HAMTSizeEstimation = cfg.Import.HAMTSizeEstimationMode()
453
454 providerStrategy := cfg.Provide.Strategy.WithDefault(config.DefaultProvideStrategy)
455
456 return fx.Options(
457 bcfgOpts,
458
459 Storage(bcfg, cfg),
460 Identity(cfg),
461 IPNS,
462 Networked(bcfg, cfg, userResourceOverrides),
463 fx.Provide(BlockService(cfg)),
464 fx.Provide(Pinning(providerStrategy)),
465 fx.Provide(Files(providerStrategy)),
466 Core,
467 )
468 }