First version of coda_net2, the libp2p-integrated coda networki… (#2450)

* First version of coda_net2, the libp2p-integrated coda networking library

* TODO the three functions that aren't done yet

* go mod tidy

* nixify the libp2p_helper build

* use real logger, process startup code from membership.ml

* less Or_error; Keypair.to_peerid

* first test

* update go deps, fix stream EOF handling, add some notes about generate_methodidx to a README

* use coda_base58

* guard against unconfigured state

* add listening_addrs method

* add me method

* reformat, add pubsub test

* add multiaddr comment

* PeerID.t

* note about multiple validators

* harmonize pk/sk

* reformat, remove random_peers, comment.

* TODO note about connection limits

* remove binary (oops)

* review feedback

* Have subscribe return a pipe

Previously you could specify which Strict_pipe behavior
you want. This is less annoying, and hopefully Strict_pipe
will be gone?

* type the helper RPCs

* straggling details

* WIP [goupdate, integrate subscription and validation]

* integrate validation with subscription

the justification I used for keeping them separate is kinda
weak. Topic validation is no longer optional!

* some bugfix, more docs, json logs, rework validation

* deriving yojson for upcall messages

* Fix pipe leak in subscription error paths

* Don't handle protocols we're already handling

* Fix duplicate subscription detection

Previously it would fail to actually look for other subscriptions for
that topic...

* Connection management note

* Move the warning attribute

* Thanks for the types @deepthiskumar!

* Note some pipe leakage

* Rename safe_secret

The "safe" bit was that it was JSON-string-safe

* Flip around duplication protocol handler condition

* process->peer

* Flesh out peers TODO

* graceful shutdown TODO love

* Do not close the pipes in reset

* Swap around filter_inplace bool

* make some control flow more obvious

* Nicer error message

* Note about stale lockfile

* yeah how did that happen

* remove extraneous sleeps

* remove callipyge dep

ended up doing keygen in the helper

* flesh out the sender todo (only needed for trust)

* pid file handling more notes

* WIP

* Use pinned nixpkgs to build libp2p_helper

* Finish up stream_state stuff

* reformat

* mutex one line at a time

* Process RPCs serially

I'm 95% sure it's safe to do these concurrently minus one edge case
regarding stream closure. I'm not sure how to handle the edge case,
so just process them serially.

* Make reformat

* Attempt to install libp2p in unit tests

* Nix try again

* Trying nix again

* Use sudo HOME too

* Better yaml

* Sudo the world

* Better better sudo

* Better sudo on the libp2p install

* Assume this system is like my computer

root sudo the nix-build

* Attempt with chmod 777 on the dir

* Qualify path to libp2p_helper

* Typo on path name

* Try disabiling coda_net2 tests and libp2p nix hack

Author: Corey Richardson

README.md

@@ -0,0 +1,5 @@
+# libp2p_helper hints
+
+If you are adding new `methodIdx` values, edit `generate_methodidx/main.go`
+(search for `TypesAndValues`) with the names of the new values. Then, run `go
+run generate_methodidx/main.go > libp2p_helper/methodidx_jsonenum.go`.

default.nix

@@ -0,0 +1,11 @@
+((import (builtins.fetchTarball {
+  name = "nixpkgs-unstable-2019-03-18";
+  url = https://github.com/nixos/nixpkgs/archive/0125544e2a0552590c87dca1583768b49ba911c0.tar.gz;
+  sha256 = "04xvlqw3zbq91zkfa506b2k1ajmj7pqh3nvdh9maabw6m5jhm5rl";
+  })) {}).buildGoModule rec {
+  name = "libp2p_helper-${version}";
+  version = "0.1";
+  src = ./src;
+  modSha256 = "1spndcx0z50cmpfxfd0971nj9n0v77fghxl36hr1pvs6kv0ra5c3";
+}
+

src/codanet.go

@@ -0,0 +1,178 @@
+package codanet
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	dsb "github.com/ipfs/go-ds-badger"
+	p2p "github.com/libp2p/go-libp2p"
+	crypto "github.com/libp2p/go-libp2p-core/crypto"
+	host "github.com/libp2p/go-libp2p-core/host"
+	routing "github.com/libp2p/go-libp2p-core/routing"
+	discovery "github.com/libp2p/go-libp2p-discovery"
+	kad "github.com/libp2p/go-libp2p-kad-dht"
+	kadopts "github.com/libp2p/go-libp2p-kad-dht/opts"
+	"github.com/libp2p/go-libp2p-peerstore"
+	"github.com/libp2p/go-libp2p-peerstore/pstoreds"
+	pnet "github.com/libp2p/go-libp2p-pnet"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/libp2p/go-libp2p-record"
+	secio "github.com/libp2p/go-libp2p-secio"
+	p2pconfig "github.com/libp2p/go-libp2p/config"
+	mdns "github.com/libp2p/go-libp2p/p2p/discovery"
+	tcp "github.com/libp2p/go-tcp-transport"
+	ws "github.com/libp2p/go-ws-transport"
+	"github.com/multiformats/go-multiaddr"
+	"golang.org/x/crypto/blake2b"
+	"log"
+	"os"
+	"path"
+	"strconv"
+	"time"
+)
+
+// Helper contains all the daemon state
+type Helper struct {
+	Host   host.Host
+	Mdns   mdns.Service
+	Dht    *kad.IpfsDHT
+	Ctx    context.Context
+	Pubsub *pubsub.PubSub
+}
+
+type mdnsListener struct {
+	FoundPeer chan peerstore.PeerInfo
+}
+
+func (l *mdnsListener) HandlePeerFound(info peerstore.PeerInfo) {
+	l.FoundPeer <- info
+}
+
+type customValidator struct {
+	Base record.Validator
+}
+
+func (cv customValidator) Validate(key string, value []byte) error {
+	log.Printf("DHT Validating: %s = %s", key, value)
+	return cv.Base.Validate(key, value)
+}
+
+func (cv customValidator) Select(key string, values [][]byte) (int, error) {
+	log.Printf("DHT Selecting Among: %s = %s", key, bytes.Join(values, []byte("; ")))
+	return cv.Base.Select(key, values)
+}
+
+// MakeHelper does all the initialization to run one host
+func MakeHelper(ctx context.Context, listenOn []multiaddr.Multiaddr, statedir string, pk crypto.PrivKey, networkID string) (*Helper, error) {
+	dso := dsb.DefaultOptions
+
+	bp := path.Join(statedir, strconv.Itoa(os.Getpid()))
+	os.MkdirAll(bp, 0700)
+
+	ds, err := dsb.NewDatastore(path.Join(statedir, strconv.Itoa(os.Getpid()), "libp2p-peerstore-v0"), &dso)
+	if err != nil {
+		return nil, err
+	}
+
+	dsoDht := dsb.DefaultOptions
+	dsDht, err := dsb.NewDatastore(path.Join(statedir, strconv.Itoa(os.Getpid()), "libp2p-dht-v0"), &dsoDht)
+	if err != nil {
+		return nil, err
+	}
+
+	ps, err := pstoreds.NewPeerstore(ctx, ds, pstoreds.DefaultOpts())
+	if err != nil {
+		return nil, err
+	}
+
+	pnetKey := blake2b.Sum256([]byte("/coda/0.0.1"))
+	prot, err := pnet.NewV1ProtectorFromBytes(&pnetKey)
+	if err != nil {
+		return nil, err
+	}
+
+	rv := customValidator{Base: record.NamespacedValidator{"pk": record.PublicKeyValidator{}}}
+
+	kadch := make(chan *kad.IpfsDHT)
+
+	// Make sure this doesn't get too out of sync with the defaults,
+	// NewWithoutDefaults is considered unstable.
+	host, err := p2p.NewWithoutDefaults(ctx,
+		p2p.Transport(tcp.NewTCPTransport),
+		p2p.Transport(ws.New),
+		p2p.Muxer("/mplex/6.7.0", DefaultMplexTransport),
+		p2p.Security(secio.ID, secio.New),
+		p2p.Identity(pk),
+		p2p.Peerstore(ps),
+		p2p.DisableRelay(),
+		p2p.ListenAddrs(listenOn...),
+		p2p.Routing(
+			p2pconfig.RoutingC(func(host host.Host) (routing.PeerRouting, error) {
+				kad, err := kad.New(ctx, host, kadopts.Datastore(dsDht), kadopts.Validator(rv))
+				go func() { kadch <- kad }()
+				return kad, err
+			})),
+		p2p.PrivateNetwork(prot))
+
+	if err != nil {
+		return nil, err
+	}
+
+	rendezvousString := fmt.Sprintf("/coda/0.0.1/%s", networkID)
+
+	mdns, err := mdns.NewMdnsService(ctx, host, time.Minute, "_coda-discovery._udp.local")
+	if err != nil {
+		return nil, err
+	}
+	l := &mdnsListener{FoundPeer: make(chan peerstore.PeerInfo)}
+	mdns.RegisterNotifee(l)
+
+	kad := <-kadch
+
+	if err = kad.Bootstrap(ctx); err != nil {
+		return nil, err
+	}
+	routingDiscovery := discovery.NewRoutingDiscovery(kad)
+
+	log.Println("Announcing ourselves for", rendezvousString)
+	discovery.Advertise(ctx, routingDiscovery, rendezvousString)
+
+	// try and find some peers for this chain
+	//dhtpeers, err := routingDiscovery.FindPeers(ctx, rendezvousString, discovery.Limit(16))
+	//if err != nil {
+	//	return nil, err
+	//}
+
+	foundPeer := func(info peerstore.PeerInfo, source string) {
+		if info.ID != "" {
+			ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
+			defer cancel()
+			if err := host.Connect(ctx, info); err != nil {
+				log.Printf("Warn: couldn't connect to %s peer %v (different chain?): %v", source, info.Loggable(), err)
+			} else {
+				log.Printf("Found a %s peer: %s", source, info.Loggable())
+				host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.PermanentAddrTTL)
+			}
+		}
+	}
+
+	go func() {
+		for {
+			info := <-l.FoundPeer
+			foundPeer(info, "local")
+		}
+	}()
+
+	pubsub, err := pubsub.NewFloodSub(ctx, host, pubsub.WithStrictSignatureVerification(true), pubsub.WithMessageSigning(true))
+	if err != nil {
+		return nil, err
+	}
+
+	return &Helper{
+		Host:   host,
+		Ctx:    ctx,
+		Mdns:   mdns,
+		Dht:    kad,
+		Pubsub: pubsub,
+	}, nil
+}

src/gen_keys/main.go

@@ -0,0 +1,15 @@
+package main
+import (
+        pnet "github.com/libp2p/go-libp2p-pnet"
+        "log"
+        "io"
+        "os"
+    )
+
+func main() {
+    key, err := pnet.GenerateV1PSK()
+    if err != nil {
+        log.Fatal("sad %s", err);
+    }
+    io.Copy(os.Stdout, key)
+}

src/generate_methodidx/main.go

@@ -0,0 +1,119 @@
+// Copyright 2017 Google Inc. All rights reserved.
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to writing, software distributed
+// under the License is distributed on a "AS IS" BASIS, WITHOUT WARRANTIES OR
+// CONDITIONS OF ANY KIND, either express or implied.
+//
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Added as a .go file to avoid embedding issues of the template.
+
+package main
+
+import (
+	"bytes"
+	"go/format"
+	"log"
+	"os"
+	"text/template"
+)
+
+var generatedTmpl = template.Must(template.New("generated").Parse(`
+// generated by {{.Command}}; DO NOT EDIT
+
+package {{.PackageName}}
+
+import (
+    "encoding/json"
+    "fmt"
+)
+
+{{range $typename, $values := .TypesAndValues}}
+
+var (
+    _{{$typename}}NameToValue = map[string]{{$typename}} {
+        {{range $values}}"{{.}}": {{.}},
+        {{end}}
+    }
+
+    _{{$typename}}ValueToName = map[{{$typename}}]string {
+        {{range $values}}{{.}}: "{{.}}",
+        {{end}}
+    }
+)
+
+func init() {
+    var v {{$typename}}
+    if _, ok := interface{}(v).(fmt.Stringer); ok {
+        _{{$typename}}NameToValue = map[string]{{$typename}} {
+            {{range $values}}interface{}({{.}}).(fmt.Stringer).String(): {{.}},
+            {{end}}
+        }
+    }
+}
+
+// MarshalJSON is generated so {{$typename}} satisfies json.Marshaler.
+func (r {{$typename}}) MarshalJSON() ([]byte, error) {
+    if s, ok := interface{}(r).(fmt.Stringer); ok {
+        return json.Marshal(s.String())
+    }
+    s, ok := _{{$typename}}ValueToName[r]
+    if !ok {
+        return nil, fmt.Errorf("invalid {{$typename}}: %d", r)
+    }
+    return json.Marshal(s)
+}
+
+// UnmarshalJSON is generated so {{$typename}} satisfies json.Unmarshaler.
+func (r *{{$typename}}) UnmarshalJSON(data []byte) error {
+    var s string
+    if err := json.Unmarshal(data, &s); err != nil {
+        return fmt.Errorf("{{$typename}} should be a string, got %s", data)
+    }
+    v, ok := _{{$typename}}NameToValue[s]
+    if !ok {
+        return fmt.Errorf("invalid {{$typename}} %q", s)
+    }
+    *r = v
+    return nil
+}
+
+{{end}}
+`))
+
+type analysis struct {
+	Command        string
+	PackageName    string
+	TypesAndValues map[string][]string
+}
+
+func main() {
+	var buf bytes.Buffer
+
+	an := analysis{
+		Command:     "generate_methodidx",
+		PackageName: "main",
+		TypesAndValues: map[string][]string{
+			"methodIdx": []string{"configure", "listen", "publish", "subscribe", "unsubscribe",  "validationComplete", "generateKeypair", "openStream", "closeStream", "resetStream", "sendStreamMsg", "removeStreamHandler", "addStreamHandler", "listeningAddrs"},
+		},
+	}
+
+	if err := generatedTmpl.Execute(&buf, an); err != nil {
+		log.Fatalf("generating code: %v", err)
+	}
+
+	src, err := format.Source(buf.Bytes())
+	if err != nil {
+		// Should never happen, but can arise when developing this code.
+		// The user can compile the output to see the error.
+		log.Printf("warning: internal error: invalid Go generated: %s", err)
+		log.Printf("warning: compile the package to analyze the error")
+		src = buf.Bytes()
+	}
+	os.Stdout.Write(src)
+}