core.go 26.2 KB
Newer Older
1 2 3 4
/*
Package core implements the IpfsNode object and related methods.

Packages underneath core/ provide a (relatively) stable, low-level API
5 6 7 8
to carry out most IPFS-related tasks.  For more details on the other
interfaces and how core/... fits into the bigger IPFS picture, see:

  $ godoc github.com/ipfs/go-ipfs
9
*/
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
10 11
package core

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
12
import (
Jakub Sztandera's avatar
Jakub Sztandera committed
13
	"bytes"
14
	"context"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
15
	"errors"
Juan Batiz-Benet's avatar
go fmt  
Juan Batiz-Benet committed
16
	"fmt"
17
	"io"
18 19 20
	"io/ioutil"
	"os"
	"strings"
21
	"time"
22

23
	version "github.com/ipfs/go-ipfs"
24
	rp "github.com/ipfs/go-ipfs/exchange/reprovide"
25
	filestore "github.com/ipfs/go-ipfs/filestore"
26 27 28
	mount "github.com/ipfs/go-ipfs/fuse/mount"
	namesys "github.com/ipfs/go-ipfs/namesys"
	ipnsrp "github.com/ipfs/go-ipfs/namesys/republisher"
Łukasz Magiera's avatar
Łukasz Magiera committed
29
	p2p "github.com/ipfs/go-ipfs/p2p"
30
	pin "github.com/ipfs/go-ipfs/pin"
31
	provider "github.com/ipfs/go-ipfs/provider"
32
	repo "github.com/ipfs/go-ipfs/repo"
33

Jakub Sztandera's avatar
Jakub Sztandera committed
34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79
	bitswap "github.com/ipfs/go-bitswap"
	bsnet "github.com/ipfs/go-bitswap/network"
	bserv "github.com/ipfs/go-blockservice"
	cid "github.com/ipfs/go-cid"
	ds "github.com/ipfs/go-datastore"
	bstore "github.com/ipfs/go-ipfs-blockstore"
	config "github.com/ipfs/go-ipfs-config"
	exchange "github.com/ipfs/go-ipfs-exchange-interface"
	nilrouting "github.com/ipfs/go-ipfs-routing/none"
	u "github.com/ipfs/go-ipfs-util"
	ipld "github.com/ipfs/go-ipld-format"
	logging "github.com/ipfs/go-log"
	merkledag "github.com/ipfs/go-merkledag"
	mfs "github.com/ipfs/go-mfs"
	resolver "github.com/ipfs/go-path/resolver"
	ft "github.com/ipfs/go-unixfs"
	goprocess "github.com/jbenet/goprocess"
	libp2p "github.com/libp2p/go-libp2p"
	autonat "github.com/libp2p/go-libp2p-autonat-svc"
	circuit "github.com/libp2p/go-libp2p-circuit"
	connmgr "github.com/libp2p/go-libp2p-connmgr"
	ic "github.com/libp2p/go-libp2p-crypto"
	p2phost "github.com/libp2p/go-libp2p-host"
	ifconnmgr "github.com/libp2p/go-libp2p-interface-connmgr"
	dht "github.com/libp2p/go-libp2p-kad-dht"
	dhtopts "github.com/libp2p/go-libp2p-kad-dht/opts"
	metrics "github.com/libp2p/go-libp2p-metrics"
	peer "github.com/libp2p/go-libp2p-peer"
	pstore "github.com/libp2p/go-libp2p-peerstore"
	pnet "github.com/libp2p/go-libp2p-pnet"
	pubsub "github.com/libp2p/go-libp2p-pubsub"
	psrouter "github.com/libp2p/go-libp2p-pubsub-router"
	quic "github.com/libp2p/go-libp2p-quic-transport"
	record "github.com/libp2p/go-libp2p-record"
	routing "github.com/libp2p/go-libp2p-routing"
	rhelpers "github.com/libp2p/go-libp2p-routing-helpers"
	discovery "github.com/libp2p/go-libp2p/p2p/discovery"
	p2pbhost "github.com/libp2p/go-libp2p/p2p/host/basic"
	rhost "github.com/libp2p/go-libp2p/p2p/host/routed"
	identify "github.com/libp2p/go-libp2p/p2p/protocol/identify"
	mafilter "github.com/libp2p/go-maddr-filter"
	smux "github.com/libp2p/go-stream-muxer"
	ma "github.com/multiformats/go-multiaddr"
	mplex "github.com/whyrusleeping/go-smux-multiplex"
	yamux "github.com/whyrusleeping/go-smux-yamux"
	mamask "github.com/whyrusleeping/multiaddr-filter"
Łukasz Magiera's avatar
Łukasz Magiera committed
80
)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
81

Jeromy's avatar
Jeromy committed
82
const IpnsValidatorTag = "ipns"
83

84
const kReprovideFrequency = time.Hour * 12
85
const discoveryConnTimeout = time.Second * 30
86
const DefaultIpnsCacheSize = 128
Jeromy's avatar
Jeromy committed
87

Jeromy's avatar
Jeromy committed
88
var log = logging.Logger("core")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
89

90
func init() {
91
	identify.ClientVersion = "go-ipfs/" + version.CurrentVersionNumber + "/" + version.CurrentCommit
92 93
}

Juan Batiz-Benet's avatar
go lint  
Juan Batiz-Benet committed
94
// IpfsNode is IPFS Core module. It represents an IPFS instance.
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
95 96
type IpfsNode struct {

97
	// Self
98
	Identity peer.ID // the local node's identity
99

100
	Repo repo.Repo
101 102

	// Local node
103 104 105 106
	Pinning         pin.Pinner // the pinning manager
	Mounts          Mounts     // current mount state, if any.
	PrivateKey      ic.PrivKey // the local node's private Key
	PNetFingerprint []byte     // fingerprint of private network
107 108

	// Services
109 110 111 112 113 114 115 116 117 118 119 120
	Peerstore       pstore.Peerstore     // storage for other Peer instances
	Blockstore      bstore.GCBlockstore  // the block store (lower level)
	Filestore       *filestore.Filestore // the filestore blockstore
	BaseBlocks      bstore.Blockstore    // the raw blockstore, no filestore wrapping
	GCLocker        bstore.GCLocker      // the locker used to protect the blockstore during gc
	Blocks          bserv.BlockService   // the block service, get/add blocks.
	DAG             ipld.DAGService      // the merkle dag service, get/add objects.
	Resolver        *resolver.Resolver   // the path resolution system
	Reporter        metrics.Reporter
	Discovery       discovery.Service
	FilesRoot       *mfs.Root
	RecordValidator record.Validator
121 122

	// Online
123 124 125 126 127
	PeerHost     p2phost.Host        // the network host (server+client)
	Bootstrapper io.Closer           // the periodic bootstrapper
	Routing      routing.IpfsRouting // the routing system. recommend ipfs-dht
	Exchange     exchange.Interface  // the block exchange + strategy (bitswap)
	Namesys      namesys.NameSystem  // the name system, resolves paths to hashes
128
	Provider     provider.Provider   // the value provider system
129
	Reprovider   *rp.Reprovider      // the value reprovider system
Jeromy's avatar
Jeromy committed
130
	IpnsRepub    *ipnsrp.Republisher
131

vyzo's avatar
vyzo committed
132
	AutoNAT  *autonat.AutoNATService
Steven Allen's avatar
Steven Allen committed
133
	PubSub   *pubsub.PubSub
134
	PSRouter *psrouter.PubsubValueStore
135
	DHT      *dht.IpfsDHT
Łukasz Magiera's avatar
Łukasz Magiera committed
136
	P2P      *p2p.P2P
Jeromy's avatar
Jeromy committed
137

138 139
	proc goprocess.Process
	ctx  context.Context
140

141 142 143
	// Flags
	IsOnline bool // Online is set when networking is enabled.
	IsDaemon bool // Daemon is set when running on a long-running daemon.
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
144 145
}

146 147 148 149 150 151 152 153
// Mounts defines what the node's mount state is. This should
// perhaps be moved to the daemon or mount. It's here because
// it needs to be accessible across daemon requests.
type Mounts struct {
	Ipfs mount.Mount
	Ipns mount.Mount
}

vyzo's avatar
vyzo committed
154
func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption RoutingOption, hostOption HostOption, do DiscoveryOption, pubsub, ipnsps, mplex bool) error {
155
	if n.PeerHost != nil { // already online.
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
156
		return errors.New("node already online")
157 158
	}

159 160 161 162
	if n.PrivateKey == nil {
		return fmt.Errorf("private key not available")
	}

Jeromy's avatar
Jeromy committed
163
	// get undialable addrs from config
164 165 166 167
	cfg, err := n.Repo.Config()
	if err != nil {
		return err
	}
Steven Allen's avatar
Steven Allen committed
168 169

	var libp2pOpts []libp2p.Option
170
	for _, s := range cfg.Swarm.AddrFilters {
Jeromy's avatar
Jeromy committed
171 172
		f, err := mamask.NewMask(s)
		if err != nil {
173
			return fmt.Errorf("incorrectly formatted address filter in config: %s", s)
Jeromy's avatar
Jeromy committed
174
		}
Steven Allen's avatar
Steven Allen committed
175
		libp2pOpts = append(libp2pOpts, libp2p.FilterAddresses(f))
Jeromy's avatar
Jeromy committed
176 177
	}

178 179 180
	if !cfg.Swarm.DisableBandwidthMetrics {
		// Set reporter
		n.Reporter = metrics.NewBandwidthCounter()
Steven Allen's avatar
Steven Allen committed
181
		libp2pOpts = append(libp2pOpts, libp2p.BandwidthReporter(n.Reporter))
182 183
	}

Jakub Sztandera's avatar
Jakub Sztandera committed
184 185 186 187 188 189
	swarmkey, err := n.Repo.SwarmKey()
	if err != nil {
		return err
	}

	if swarmkey != nil {
Steven Allen's avatar
Steven Allen committed
190
		protec, err := pnet.NewProtector(bytes.NewReader(swarmkey))
Jakub Sztandera's avatar
Jakub Sztandera committed
191
		if err != nil {
192
			return fmt.Errorf("failed to configure private network: %s", err)
Jakub Sztandera's avatar
Jakub Sztandera committed
193
		}
194
		n.PNetFingerprint = protec.Fingerprint()
195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212
		go func() {
			t := time.NewTicker(30 * time.Second)
			<-t.C // swallow one tick
			for {
				select {
				case <-t.C:
					if ph := n.PeerHost; ph != nil {
						if len(ph.Network().Peers()) == 0 {
							log.Warning("We are in private network and have no peers.")
							log.Warning("This might be configuration mistake.")
						}
					}
				case <-n.Process().Closing():
					t.Stop()
					return
				}
			}
		}()
Steven Allen's avatar
Steven Allen committed
213 214

		libp2pOpts = append(libp2pOpts, libp2p.PrivateNetwork(protec))
Jakub Sztandera's avatar
Jakub Sztandera committed
215 216
	}

217 218 219 220
	addrsFactory, err := makeAddrsFactory(cfg.Addresses)
	if err != nil {
		return err
	}
Steven Allen's avatar
Steven Allen committed
221 222 223 224
	if !cfg.Swarm.DisableRelay {
		addrsFactory = composeAddrsFactory(addrsFactory, filterRelayAddrs)
	}
	libp2pOpts = append(libp2pOpts, libp2p.AddrsFactory(addrsFactory))
225

Steven Allen's avatar
Steven Allen committed
226
	connm, err := constructConnMgr(cfg.Swarm.ConnMgr)
Jeromy's avatar
Jeromy committed
227 228 229
	if err != nil {
		return err
	}
Steven Allen's avatar
Steven Allen committed
230 231 232
	libp2pOpts = append(libp2pOpts, libp2p.ConnectionManager(connm))

	libp2pOpts = append(libp2pOpts, makeSmuxTransportOption(mplex))
Jeromy's avatar
Jeromy committed
233

Steven Allen's avatar
Steven Allen committed
234 235
	if !cfg.Swarm.DisableNatPortMap {
		libp2pOpts = append(libp2pOpts, libp2p.NATPortMap())
236
	}
Steven Allen's avatar
Steven Allen committed
237

238 239 240
	// disable the default listen addrs
	libp2pOpts = append(libp2pOpts, libp2p.NoListenAddrs)

241 242 243 244 245 246 247 248 249 250 251
	if cfg.Swarm.DisableRelay {
		// Enabled by default.
		libp2pOpts = append(libp2pOpts, libp2p.DisableRelay())
	} else {
		relayOpts := []circuit.RelayOpt{circuit.OptDiscovery}
		if cfg.Swarm.EnableRelayHop {
			relayOpts = append(relayOpts, circuit.OptHop)
		}
		libp2pOpts = append(libp2pOpts, libp2p.EnableRelay(relayOpts...))
	}

252 253 254
	// explicitly enable the default transports
	libp2pOpts = append(libp2pOpts, libp2p.DefaultTransports)

Marten Seemann's avatar
Marten Seemann committed
255 256 257 258
	if cfg.Experimental.QUIC {
		libp2pOpts = append(libp2pOpts, libp2p.Transport(quic.NewTransport))
	}

vyzo's avatar
vyzo committed
259 260 261 262 263 264 265 266
	// enable routing
	libp2pOpts = append(libp2pOpts, libp2p.Routing(func(h p2phost.Host) (routing.PeerRouting, error) {
		r, err := routingOption(ctx, h, n.Repo.Datastore(), n.RecordValidator)
		n.Routing = r
		return r, err
	}))

	// enable autorelay
vyzo's avatar
vyzo committed
267
	if cfg.Swarm.EnableAutoRelay {
vyzo's avatar
vyzo committed
268
		libp2pOpts = append(libp2pOpts, libp2p.EnableAutoRelay())
vyzo's avatar
vyzo committed
269
	}
270

Steven Allen's avatar
Steven Allen committed
271
	peerhost, err := hostOption(ctx, n.Identity, n.Peerstore, libp2pOpts...)
vyzo's avatar
vyzo committed
272

273
	if err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
274
		return err
275 276
	}

277 278
	n.PeerHost = peerhost

279
	if err := n.startOnlineServicesWithHost(ctx, routingOption, pubsub, ipnsps); err != nil {
280
		return err
281 282 283
	}

	// Ok, now we're ready to listen.
Łukasz Magiera's avatar
Łukasz Magiera committed
284
	if err := startListening(n.PeerHost, cfg); err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
285
		return err
286
	}
287

Łukasz Magiera's avatar
Łukasz Magiera committed
288
	n.P2P = p2p.NewP2P(n.Identity, n.PeerHost, n.Peerstore)
289

290
	// setup local discovery
Jeromy's avatar
Jeromy committed
291
	if do != nil {
Jeromy's avatar
Jeromy committed
292
		service, err := do(ctx, n.PeerHost)
Jeromy's avatar
Jeromy committed
293
		if err != nil {
Jeromy's avatar
Jeromy committed
294 295 296 297
			log.Error("mdns error: ", err)
		} else {
			service.RegisterNotifee(n)
			n.Discovery = service
Jeromy's avatar
Jeromy committed
298
		}
299 300
	}

301
	return n.Bootstrap(DefaultBootstrapConfig)
302 303
}

Jeromy's avatar
Jeromy committed
304 305
func constructConnMgr(cfg config.ConnMgr) (ifconnmgr.ConnManager, error) {
	switch cfg.Type {
306 307 308 309
	case "":
		// 'default' value is the basic connection manager
		return connmgr.NewConnManager(config.DefaultConnMgrLowWater, config.DefaultConnMgrHighWater, config.DefaultConnMgrGracePeriod), nil
	case "none":
Jeromy's avatar
Jeromy committed
310 311 312 313 314 315 316 317 318 319 320 321 322
		return nil, nil
	case "basic":
		grace, err := time.ParseDuration(cfg.GracePeriod)
		if err != nil {
			return nil, fmt.Errorf("parsing Swarm.ConnMgr.GracePeriod: %s", err)
		}

		return connmgr.NewConnManager(cfg.LowWater, cfg.HighWater, grace), nil
	default:
		return nil, fmt.Errorf("unrecognized ConnMgr.Type: %q", cfg.Type)
	}
}

Łukasz Magiera's avatar
Łukasz Magiera committed
323 324 325 326 327 328
func (n *IpfsNode) startLateOnlineServices(ctx context.Context) error {
	cfg, err := n.Repo.Config()
	if err != nil {
		return err
	}

329 330 331 332 333 334
	// Provider

	n.Provider.Run()

	// Reprovider

335
	var keyProvider rp.KeyChanFunc
Łukasz Magiera's avatar
Łukasz Magiera committed
336 337 338 339 340 341 342 343 344 345 346

	switch cfg.Reprovider.Strategy {
	case "all":
		fallthrough
	case "":
		keyProvider = rp.NewBlockstoreProvider(n.Blockstore)
	case "roots":
		keyProvider = rp.NewPinnedProvider(n.Pinning, n.DAG, true)
	case "pinned":
		keyProvider = rp.NewPinnedProvider(n.Pinning, n.DAG, false)
	default:
347
		return fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy)
Łukasz Magiera's avatar
Łukasz Magiera committed
348
	}
349
	n.Reprovider = rp.NewReprovider(ctx, n.Routing, keyProvider)
Łukasz Magiera's avatar
Łukasz Magiera committed
350

351 352 353 354 355
	reproviderInterval := kReprovideFrequency
	if cfg.Reprovider.Interval != "" {
		dur, err := time.ParseDuration(cfg.Reprovider.Interval)
		if err != nil {
			return err
Łukasz Magiera's avatar
Łukasz Magiera committed
356 357
		}

358
		reproviderInterval = dur
Łukasz Magiera's avatar
Łukasz Magiera committed
359 360
	}

361 362
	go n.Reprovider.Run(reproviderInterval)

Łukasz Magiera's avatar
Łukasz Magiera committed
363 364 365
	return nil
}

366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401
func makeAddrsFactory(cfg config.Addresses) (p2pbhost.AddrsFactory, error) {
	var annAddrs []ma.Multiaddr
	for _, addr := range cfg.Announce {
		maddr, err := ma.NewMultiaddr(addr)
		if err != nil {
			return nil, err
		}
		annAddrs = append(annAddrs, maddr)
	}

	filters := mafilter.NewFilters()
	noAnnAddrs := map[string]bool{}
	for _, addr := range cfg.NoAnnounce {
		f, err := mamask.NewMask(addr)
		if err == nil {
			filters.AddDialFilter(f)
			continue
		}
		maddr, err := ma.NewMultiaddr(addr)
		if err != nil {
			return nil, err
		}
		noAnnAddrs[maddr.String()] = true
	}

	return func(allAddrs []ma.Multiaddr) []ma.Multiaddr {
		var addrs []ma.Multiaddr
		if len(annAddrs) > 0 {
			addrs = annAddrs
		} else {
			addrs = allAddrs
		}

		var out []ma.Multiaddr
		for _, maddr := range addrs {
			// check for exact matches
402
			ok := noAnnAddrs[maddr.String()]
403 404 405 406 407 408 409 410 411
			// check for /ipcidr matches
			if !ok && !filters.AddrBlocked(maddr) {
				out = append(out, maddr)
			}
		}
		return out
	}, nil
}

Steven Allen's avatar
Steven Allen committed
412 413 414
func makeSmuxTransportOption(mplexExp bool) libp2p.Option {
	const yamuxID = "/yamux/1.0.0"
	const mplexID = "/mplex/6.7.0"
415 416

	ymxtpt := &yamux.Transport{
Steven Allen's avatar
Steven Allen committed
417
		AcceptBacklog:          512,
418 419 420
		ConnectionWriteTimeout: time.Second * 10,
		KeepAliveInterval:      time.Second * 30,
		EnableKeepAlive:        true,
421
		MaxStreamWindowSize:    uint32(16 * 1024 * 1024), // 16MiB
422 423 424
		LogOutput:              ioutil.Discard,
	}

425 426 427 428
	if os.Getenv("YAMUX_DEBUG") != "" {
		ymxtpt.LogOutput = os.Stderr
	}

Steven Allen's avatar
Steven Allen committed
429
	muxers := map[string]smux.Transport{yamuxID: ymxtpt}
430
	if mplexExp {
Steven Allen's avatar
Steven Allen committed
431
		muxers[mplexID] = mplex.DefaultTransport
432 433 434
	}

	// Allow muxer preference order overriding
Steven Allen's avatar
Steven Allen committed
435
	order := []string{yamuxID, mplexID}
436
	if prefs := os.Getenv("LIBP2P_MUX_PREFS"); prefs != "" {
Steven Allen's avatar
Steven Allen committed
437 438 439 440 441 442 443 444 445 446 447 448
		order = strings.Fields(prefs)
	}

	opts := make([]libp2p.Option, 0, len(order))
	for _, id := range order {
		tpt, ok := muxers[id]
		if !ok {
			log.Warning("unknown or duplicate muxer in LIBP2P_MUX_PREFS: %s", id)
			continue
		}
		delete(muxers, id)
		opts = append(opts, libp2p.Muxer(id, tpt))
449 450
	}

Steven Allen's avatar
Steven Allen committed
451
	return libp2p.ChainOptions(opts...)
452 453
}

Jeromy's avatar
Jeromy committed
454 455
func setupDiscoveryOption(d config.Discovery) DiscoveryOption {
	if d.MDNS.Enabled {
Jeromy's avatar
Jeromy committed
456
		return func(ctx context.Context, h p2phost.Host) (discovery.Service, error) {
Jeromy's avatar
Jeromy committed
457 458 459
			if d.MDNS.Interval == 0 {
				d.MDNS.Interval = 5
			}
Jeromy's avatar
Jeromy committed
460
			return discovery.NewMdnsService(ctx, h, time.Duration(d.MDNS.Interval)*time.Second, discovery.ServiceTag)
Jeromy's avatar
Jeromy committed
461 462 463 464 465
		}
	}
	return nil
}

466 467
// HandlePeerFound attempts to connect to peer from `PeerInfo`, if it fails
// logs a warning log.
Jeromy's avatar
Jeromy committed
468
func (n *IpfsNode) HandlePeerFound(p pstore.PeerInfo) {
469
	log.Warning("trying peer info: ", p)
470
	ctx, cancel := context.WithTimeout(n.Context(), discoveryConnTimeout)
rht's avatar
rht committed
471
	defer cancel()
472
	if err := n.PeerHost.Connect(ctx, p); err != nil {
473 474 475 476
		log.Warning("Failed to connect to peer found by discovery: ", err)
	}
}

477 478
// startOnlineServicesWithHost  is the set of services which need to be
// initialized with the host and _before_ we start listening.
479
func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, routingOption RoutingOption, enablePubsub bool, enableIpnsps bool) error {
vyzo's avatar
vyzo committed
480 481 482 483 484 485
	cfg, err := n.Repo.Config()
	if err != nil {
		return err
	}

	if cfg.Swarm.EnableAutoNATService {
vyzo's avatar
vyzo committed
486 487 488 489 490
		var opts []libp2p.Option
		if cfg.Experimental.QUIC {
			opts = append(opts, libp2p.DefaultTransports, libp2p.Transport(quic.NewTransport))
		}

491
		svc, err := autonat.NewAutoNATService(ctx, n.PeerHost, opts...)
492 493 494
		if err != nil {
			return err
		}
vyzo's avatar
vyzo committed
495 496
		n.AutoNAT = svc
	}
497

vyzo's avatar
vyzo committed
498
	if enablePubsub || enableIpnsps {
Steven Allen's avatar
Steven Allen committed
499
		var service *pubsub.PubSub
500

Steven Allen's avatar
Steven Allen committed
501 502 503 504 505 506 507 508 509
		var pubsubOptions []pubsub.Option
		if cfg.Pubsub.DisableSigning {
			pubsubOptions = append(pubsubOptions, pubsub.WithMessageSigning(false))
		}

		if cfg.Pubsub.StrictSignatureVerification {
			pubsubOptions = append(pubsubOptions, pubsub.WithStrictSignatureVerification(true))
		}

510 511 512 513
		switch cfg.Pubsub.Router {
		case "":
			fallthrough
		case "floodsub":
514
			service, err = pubsub.NewFloodSub(ctx, n.PeerHost, pubsubOptions...)
515 516

		case "gossipsub":
517
			service, err = pubsub.NewGossipSub(ctx, n.PeerHost, pubsubOptions...)
518 519 520 521 522

		default:
			err = fmt.Errorf("Unknown pubsub router %s", cfg.Pubsub.Router)
		}

523 524 525
		if err != nil {
			return err
		}
Steven Allen's avatar
Steven Allen committed
526
		n.PubSub = service
527 528
	}

529 530
	// this code is necessary just for tests: mock network constructions
	// ignore the libp2p constructor options that actually construct the routing!
531
	if n.Routing == nil {
532
		r, err := routingOption(ctx, n.PeerHost, n.Repo.Datastore(), n.RecordValidator)
533 534 535 536
		if err != nil {
			return err
		}
		n.Routing = r
537
		n.PeerHost = rhost.Wrap(n.PeerHost, n.Routing)
538 539
	}

540 541 542 543 544 545 546 547 548 549 550 551 552
	// TODO: I'm not a fan of type assertions like this but the
	// `RoutingOption` system doesn't currently provide access to the
	// IpfsNode.
	//
	// Ideally, we'd do something like:
	//
	// 1. Add some fancy method to introspect into tiered routers to extract
	//    things like the pubsub router or the DHT (complicated, messy,
	//    probably not worth it).
	// 2. Pass the IpfsNode into the RoutingOption (would also remove the
	//    PSRouter case below.
	// 3. Introduce some kind of service manager? (my personal favorite but
	//    that requires a fair amount of work).
553
	if dht, ok := n.Routing.(*dht.IpfsDHT); ok {
554 555 556
		n.DHT = dht
	}

Steven Allen's avatar
Steven Allen committed
557
	if enableIpnsps {
558 559
		n.PSRouter = psrouter.NewPubsubValueStore(
			ctx,
560
			n.PeerHost,
561
			n.Routing,
Steven Allen's avatar
Steven Allen committed
562
			n.PubSub,
563
			n.RecordValidator,
564 565
		)
		n.Routing = rhelpers.Tiered{
Łukasz Magiera's avatar
Łukasz Magiera committed
566 567 568 569 570 571 572
			Routers: []routing.IpfsRouting{
				// Always check pubsub first.
				&rhelpers.Compose{
					ValueStore: &rhelpers.LimitedValueStore{
						ValueStore: n.PSRouter,
						Namespaces: []string{"ipns"},
					},
573
				},
Łukasz Magiera's avatar
Łukasz Magiera committed
574
				n.Routing,
575
			},
Łukasz Magiera's avatar
Łukasz Magiera committed
576
			Validator: n.RecordValidator,
577 578 579
		}
	}

580
	// setup exchange service
581
	bitswapNetwork := bsnet.NewFromIpfsHost(n.PeerHost, n.Routing)
Łukasz Magiera's avatar
Łukasz Magiera committed
582
	n.Exchange = bitswap.New(ctx, bitswapNetwork, n.Blockstore)
583

584 585 586 587 588
	size, err := n.getCacheSize()
	if err != nil {
		return err
	}

589
	// setup name system
590
	n.Namesys = namesys.NewNameSystem(n.Routing, n.Repo.Datastore(), size)
591

Jeromy's avatar
Jeromy committed
592
	// setup ipns republishing
593
	return n.setupIpnsRepublisher()
594 595
}

596 597 598 599 600 601 602 603 604
// getCacheSize returns cache life and cache size
func (n *IpfsNode) getCacheSize() (int, error) {
	cfg, err := n.Repo.Config()
	if err != nil {
		return 0, err
	}

	cs := cfg.Ipns.ResolveCacheSize
	if cs == 0 {
605
		cs = DefaultIpnsCacheSize
606 607 608 609 610 611 612
	}
	if cs < 0 {
		return 0, fmt.Errorf("cannot specify negative resolve cache size")
	}
	return cs, nil
}

613
func (n *IpfsNode) setupIpnsRepublisher() error {
Jeromy's avatar
Jeromy committed
614 615 616 617
	cfg, err := n.Repo.Config()
	if err != nil {
		return err
	}
618

619
	n.IpnsRepub = ipnsrp.NewRepublisher(n.Namesys, n.Repo.Datastore(), n.PrivateKey, n.Repo.Keystore())
620

Jeromy's avatar
Jeromy committed
621 622 623 624 625 626
	if cfg.Ipns.RepublishPeriod != "" {
		d, err := time.ParseDuration(cfg.Ipns.RepublishPeriod)
		if err != nil {
			return fmt.Errorf("failure to parse config setting IPNS.RepublishPeriod: %s", err)
		}

627
		if !u.Debug && (d < time.Minute || d > (time.Hour*24)) {
Jeromy's avatar
Jeromy committed
628 629 630 631 632 633
			return fmt.Errorf("config setting IPNS.RepublishPeriod is not between 1min and 1day: %s", d)
		}

		n.IpnsRepub.Interval = d
	}

634
	if cfg.Ipns.RecordLifetime != "" {
635
		d, err := time.ParseDuration(cfg.Ipns.RecordLifetime)
636 637 638 639 640 641 642
		if err != nil {
			return fmt.Errorf("failure to parse config setting IPNS.RecordLifetime: %s", err)
		}

		n.IpnsRepub.RecordLifetime = d
	}

Jeromy's avatar
Jeromy committed
643 644
	n.Process().Go(n.IpnsRepub.Run)

645 646 647
	return nil
}

648 649 650 651 652 653 654 655 656 657 658 659
// Process returns the Process object
func (n *IpfsNode) Process() goprocess.Process {
	return n.proc
}

// Close calls Close() on the Process object
func (n *IpfsNode) Close() error {
	return n.proc.Close()
}

// Context returns the IpfsNode context
func (n *IpfsNode) Context() context.Context {
660 661 662
	if n.ctx == nil {
		n.ctx = context.TODO()
	}
663 664 665
	return n.ctx
}

666 667
// teardown closes owned children. If any errors occur, this function returns
// the first error.
Brian Tiger Chow's avatar
Brian Tiger Chow committed
668
func (n *IpfsNode) teardown() error {
669
	log.Debug("core is shutting down...")
670 671
	// owned objects are closed in this teardown to ensure that they're closed
	// regardless of which constructor was used to add them to the node.
Jeromy's avatar
Jeromy committed
672 673
	var closers []io.Closer

674
	// NOTE: The order that objects are added(closed) matters, if an object
Jeromy's avatar
Jeromy committed
675 676 677 678 679
	// needs to use another during its shutdown/cleanup process, it should be
	// closed before that other object

	if n.FilesRoot != nil {
		closers = append(closers, n.FilesRoot)
Jeromy's avatar
Jeromy committed
680
	}
681

682 683 684 685
	if n.Exchange != nil {
		closers = append(closers, n.Exchange)
	}

686
	if n.Mounts.Ipfs != nil && !n.Mounts.Ipfs.IsActive() {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
687 688
		closers = append(closers, mount.Closer(n.Mounts.Ipfs))
	}
689
	if n.Mounts.Ipns != nil && !n.Mounts.Ipns.IsActive() {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
690 691 692
		closers = append(closers, mount.Closer(n.Mounts.Ipns))
	}

693 694
	if n.DHT != nil {
		closers = append(closers, n.DHT.Process())
Jeromy's avatar
Jeromy committed
695 696
	}

Jeromy's avatar
Jeromy committed
697 698 699 700
	if n.Blocks != nil {
		closers = append(closers, n.Blocks)
	}

Jeromy's avatar
Jeromy committed
701 702
	if n.Bootstrapper != nil {
		closers = append(closers, n.Bootstrapper)
703 704
	}

Jeromy's avatar
Jeromy committed
705 706
	if n.PeerHost != nil {
		closers = append(closers, n.PeerHost)
707
	}
708

Jeromy's avatar
Jeromy committed
709 710 711
	// Repo closed last, most things need to preserve state here
	closers = append(closers, n.Repo)

712
	var errs []error
713
	for _, closer := range closers {
714 715
		if err := closer.Close(); err != nil {
			errs = append(errs, err)
716 717 718 719
		}
	}
	if len(errs) > 0 {
		return errs[0]
Brian Tiger Chow's avatar
Brian Tiger Chow committed
720 721
	}
	return nil
Brian Tiger Chow's avatar
Brian Tiger Chow committed
722 723
}

724
// Bootstrap will set and call the IpfsNodes bootstrap function.
725
func (n *IpfsNode) Bootstrap(cfg BootstrapConfig) error {
726
	// TODO what should return value be when in offlineMode?
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
727 728 729 730
	if n.Routing == nil {
		return nil
	}

731 732 733 734 735 736 737
	if n.Bootstrapper != nil {
		n.Bootstrapper.Close() // stop previous bootstrap process.
	}

	// if the caller did not specify a bootstrap peer function, get the
	// freshest bootstrap peers from config. this responds to live changes.
	if cfg.BootstrapPeers == nil {
Jeromy's avatar
Jeromy committed
738
		cfg.BootstrapPeers = func() []pstore.PeerInfo {
739
			ps, err := n.loadBootstrapPeers()
740
			if err != nil {
741
				log.Warning("failed to parse bootstrap peers from config")
742 743 744 745 746 747 748 749 750
				return nil
			}
			return ps
		}
	}

	var err error
	n.Bootstrapper, err = Bootstrap(n, cfg)
	return err
751 752
}

753 754
func (n *IpfsNode) loadID() error {
	if n.Identity != "" {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
755
		return errors.New("identity already loaded")
756 757
	}

758 759 760 761 762 763
	cfg, err := n.Repo.Config()
	if err != nil {
		return err
	}

	cid := cfg.Identity.PeerID
764
	if cid == "" {
765
		return errors.New("identity was not set in config (was 'ipfs init' run?)")
766 767
	}
	if len(cid) == 0 {
768
		return errors.New("no peer ID in config! (was 'ipfs init' run?)")
769 770
	}

Steven Allen's avatar
Steven Allen committed
771 772 773 774 775 776
	id, err := peer.IDB58Decode(cid)
	if err != nil {
		return fmt.Errorf("peer ID invalid: %s", err)
	}

	n.Identity = id
777 778
	return nil
}
779

780
// GetKey will return a key from the Keystore with name `name`.
781 782
func (n *IpfsNode) GetKey(name string) (ic.PrivKey, error) {
	if name == "self" {
783 784 785
		if n.PrivateKey == nil {
			return nil, fmt.Errorf("private key not available")
		}
786 787 788 789 790 791
		return n.PrivateKey, nil
	} else {
		return n.Repo.Keystore().Get(name)
	}
}

792 793
// loadPrivateKey loads the private key *if* available
func (n *IpfsNode) loadPrivateKey() error {
794
	if n.Identity == "" || n.Peerstore == nil {
Łukasz Magiera's avatar
Łukasz Magiera committed
795
		return errors.New("loaded private key out of order")
796 797
	}

798
	if n.PrivateKey != nil {
Kejie Zhang's avatar
Kejie Zhang committed
799 800
		log.Warning("private key already loaded")
		return nil
801 802
	}

803 804 805 806 807
	cfg, err := n.Repo.Config()
	if err != nil {
		return err
	}

808 809 810 811
	if cfg.Identity.PrivKey == "" {
		return nil
	}

812
	sk, err := loadPrivateKey(&cfg.Identity, n.Identity)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
813
	if err != nil {
814
		return err
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
815
	}
816

Steven Allen's avatar
Steven Allen committed
817
	if err := n.Peerstore.AddPrivKey(n.Identity, sk); err != nil {
818 819 820 821 822
		return err
	}
	if err := n.Peerstore.AddPubKey(n.Identity, sk.GetPublic()); err != nil {
		return err
	}
Steven Allen's avatar
Steven Allen committed
823
	n.PrivateKey = sk
Jeromy's avatar
Jeromy committed
824 825 826
	return nil
}

Jeromy's avatar
Jeromy committed
827
func (n *IpfsNode) loadBootstrapPeers() ([]pstore.PeerInfo, error) {
828 829 830 831 832 833
	cfg, err := n.Repo.Config()
	if err != nil {
		return nil, err
	}

	parsed, err := cfg.BootstrapPeers()
834 835 836 837 838 839
	if err != nil {
		return nil, err
	}
	return toPeerInfos(parsed), nil
}

Jeromy's avatar
Jeromy committed
840
func (n *IpfsNode) loadFilesRoot() error {
Jeromy's avatar
Jeromy committed
841
	dsk := ds.NewKey("/local/filesroot")
842
	pf := func(ctx context.Context, c cid.Cid) error {
Jeromy's avatar
Jeromy committed
843
		return n.Repo.Datastore().Put(dsk, c.Bytes())
Jeromy's avatar
Jeromy committed
844 845
	}

846
	var nd *merkledag.ProtoNode
Jeromy's avatar
Jeromy committed
847 848 849 850
	val, err := n.Repo.Datastore().Get(dsk)

	switch {
	case err == ds.ErrNotFound || val == nil:
851
		nd = ft.EmptyDirNode()
852
		err := n.DAG.Add(n.Context(), nd)
Jeromy's avatar
Jeromy committed
853 854 855 856
		if err != nil {
			return fmt.Errorf("failure writing to dagstore: %s", err)
		}
	case err == nil:
857
		c, err := cid.Cast(val)
Jeromy's avatar
Jeromy committed
858 859 860 861
		if err != nil {
			return err
		}

862
		rnd, err := n.DAG.Get(n.Context(), c)
Jeromy's avatar
Jeromy committed
863 864 865
		if err != nil {
			return fmt.Errorf("error loading filesroot from DAG: %s", err)
		}
866 867 868 869 870 871 872

		pbnd, ok := rnd.(*merkledag.ProtoNode)
		if !ok {
			return merkledag.ErrNotProtobuf
		}

		nd = pbnd
Jeromy's avatar
Jeromy committed
873 874 875 876 877 878 879 880 881 882 883 884 885
	default:
		return err
	}

	mr, err := mfs.NewRoot(n.Context(), n.DAG, nd, pf)
	if err != nil {
		return err
	}

	n.FilesRoot = mr
	return nil
}

886 887
func loadPrivateKey(cfg *config.Identity, id peer.ID) (ic.PrivKey, error) {
	sk, err := cfg.DecodePrivateKey("passphrase todo!")
888 889 890
	if err != nil {
		return nil, err
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
891

892 893 894 895
	id2, err := peer.IDFromPrivateKey(sk)
	if err != nil {
		return nil, err
	}
896

897 898
	if id2 != id {
		return nil, fmt.Errorf("private key in config does not match id: %s != %s", id, id2)
899 900
	}

901
	return sk, nil
902
}
903

904
func listenAddresses(cfg *config.Config) ([]ma.Multiaddr, error) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
905 906 907
	var listen []ma.Multiaddr
	for _, addr := range cfg.Addresses.Swarm {
		maddr, err := ma.NewMultiaddr(addr)
908
		if err != nil {
Łukasz Magiera's avatar
Łukasz Magiera committed
909
			return nil, fmt.Errorf("failure to parse config.Addresses.Swarm: %s", cfg.Addresses.Swarm)
910
		}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
911
		listen = append(listen, maddr)
912 913 914 915
	}

	return listen, nil
}
916

Kevin Atkinson's avatar
Kevin Atkinson committed
917
type ConstructPeerHostOpts struct {
918
	AddrsFactory      p2pbhost.AddrsFactory
vyzo's avatar
vyzo committed
919 920 921
	DisableNatPortMap bool
	DisableRelay      bool
	EnableRelayHop    bool
Jeromy's avatar
Jeromy committed
922
	ConnectionManager ifconnmgr.ConnManager
Kevin Atkinson's avatar
Kevin Atkinson committed
923 924
}

Steven Allen's avatar
Steven Allen committed
925
type HostOption func(ctx context.Context, id peer.ID, ps pstore.Peerstore, options ...libp2p.Option) (p2phost.Host, error)
Jeromy's avatar
Jeromy committed
926 927 928

var DefaultHostOption HostOption = constructPeerHost

929
// isolates the complex initialization steps
Steven Allen's avatar
Steven Allen committed
930 931 932 933
func constructPeerHost(ctx context.Context, id peer.ID, ps pstore.Peerstore, options ...libp2p.Option) (p2phost.Host, error) {
	pkey := ps.PrivKey(id)
	if pkey == nil {
		return nil, fmt.Errorf("missing private key for node ID: %s", id.Pretty())
934
	}
Steven Allen's avatar
Steven Allen committed
935 936
	options = append([]libp2p.Option{libp2p.Identity(pkey), libp2p.Peerstore(ps)}, options...)
	return libp2p.New(ctx, options...)
937 938
}

939 940 941 942 943 944 945 946 947 948 949 950
func filterRelayAddrs(addrs []ma.Multiaddr) []ma.Multiaddr {
	var raddrs []ma.Multiaddr
	for _, addr := range addrs {
		_, err := addr.ValueForProtocol(circuit.P_CIRCUIT)
		if err == nil {
			continue
		}
		raddrs = append(raddrs, addr)
	}
	return raddrs
}

951 952 953 954 955 956
func composeAddrsFactory(f, g p2pbhost.AddrsFactory) p2pbhost.AddrsFactory {
	return func(addrs []ma.Multiaddr) []ma.Multiaddr {
		return f(g(addrs))
	}
}

957
// startListening on the network addresses
Łukasz Magiera's avatar
Łukasz Magiera committed
958
func startListening(host p2phost.Host, cfg *config.Config) error {
959 960
	listenAddrs, err := listenAddresses(cfg)
	if err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
961
		return err
962 963 964
	}

	// Actually start listening:
Steven Allen's avatar
Steven Allen committed
965
	if err := host.Network().Listen(listenAddrs...); err != nil {
966
		return err
967 968
	}

969
	// list out our addresses
970
	addrs, err := host.Network().InterfaceListenAddresses()
971
	if err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
972
		return err
973
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
974
	log.Infof("Swarm listening at: %s", addrs)
975
	return nil
976
}
977

978 979 980 981 982 983
func constructDHTRouting(ctx context.Context, host p2phost.Host, dstore ds.Batching, validator record.Validator) (routing.IpfsRouting, error) {
	return dht.New(
		ctx, host,
		dhtopts.Datastore(dstore),
		dhtopts.Validator(validator),
	)
984
}
Jeromy's avatar
Jeromy committed
985

986 987 988 989 990 991 992
func constructClientDHTRouting(ctx context.Context, host p2phost.Host, dstore ds.Batching, validator record.Validator) (routing.IpfsRouting, error) {
	return dht.New(
		ctx, host,
		dhtopts.Client(true),
		dhtopts.Datastore(dstore),
		dhtopts.Validator(validator),
	)
Jeromy's avatar
Jeromy committed
993 994
}

995
type RoutingOption func(context.Context, p2phost.Host, ds.Batching, record.Validator) (routing.IpfsRouting, error)
Jeromy's avatar
Jeromy committed
996

Jeromy's avatar
Jeromy committed
997
type DiscoveryOption func(context.Context, p2phost.Host) (discovery.Service, error)
Jeromy's avatar
Jeromy committed
998

999
var DHTOption RoutingOption = constructDHTRouting
Jeromy's avatar
Jeromy committed
1000
var DHTClientOption RoutingOption = constructClientDHTRouting
1001
var NilRouterOption RoutingOption = nilrouting.ConstructNilRouting