WIP: Add libp2p-go

cmd/dendrite-demo-libp2p/main.go

@@ -0,0 +1,203 @@
+// Copyright 2017 Vector Creations Ltd
+// Copyright 2018 New Vector Ltd
+// Copyright 2019-2020 The Matrix.org Foundation C.I.C.
+//
+// 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 in writing, software
+// distributed under the License is distributed on an "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.
+
+package main
+
+import (
+	"crypto/ed25519"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"time"
+
+	gostream "github.com/libp2p/go-libp2p-gostream"
+	p2phttp "github.com/libp2p/go-libp2p-http"
+	p2pdisc "github.com/libp2p/go-libp2p/p2p/discovery"
+	"github.com/matrix-org/dendrite/appservice"
+	"github.com/matrix-org/dendrite/clientapi"
+	"github.com/matrix-org/dendrite/clientapi/producers"
+	"github.com/matrix-org/dendrite/cmd/dendrite-demo-libp2p/storage"
+	"github.com/matrix-org/dendrite/common"
+	"github.com/matrix-org/dendrite/common/config"
+	"github.com/matrix-org/dendrite/common/keydb"
+	"github.com/matrix-org/dendrite/common/transactions"
+	"github.com/matrix-org/dendrite/eduserver"
+	"github.com/matrix-org/dendrite/federationapi"
+	"github.com/matrix-org/dendrite/federationsender"
+	"github.com/matrix-org/dendrite/mediaapi"
+	"github.com/matrix-org/dendrite/publicroomsapi"
+	"github.com/matrix-org/dendrite/roomserver"
+	"github.com/matrix-org/dendrite/syncapi"
+	"github.com/matrix-org/gomatrixserverlib"
+
+	"github.com/matrix-org/dendrite/eduserver/cache"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/sirupsen/logrus"
+)
+
+func createKeyDB(
+	base *P2PDendrite,
+) keydb.Database {
+	db, err := keydb.NewDatabase(
+		string(base.Base.Cfg.Database.ServerKey),
+		base.Base.Cfg.Matrix.ServerName,
+		base.Base.Cfg.Matrix.PrivateKey.Public().(ed25519.PublicKey),
+		base.Base.Cfg.Matrix.KeyID,
+	)
+	if err != nil {
+		logrus.WithError(err).Panicf("failed to connect to keys db")
+	}
+	mdns := mDNSListener{
+		host:  base.LibP2P,
+		keydb: db,
+	}
+	serv, err := p2pdisc.NewMdnsService(
+		base.LibP2PContext,
+		base.LibP2P,
+		time.Second*10,
+		"_matrix-dendrite-p2p._tcp",
+	)
+	if err != nil {
+		panic(err)
+	}
+	serv.RegisterNotifee(&mdns)
+	return db
+}
+
+func createFederationClient(
+	base *P2PDendrite,
+) *gomatrixserverlib.FederationClient {
+	fmt.Println("Running in libp2p federation mode")
+	fmt.Println("Warning: Federation with non-libp2p homeservers will not work in this mode yet!")
+	tr := &http.Transport{}
+	tr.RegisterProtocol(
+		"matrix",
+		p2phttp.NewTransport(base.LibP2P, p2phttp.ProtocolOption("/matrix")),
+	)
+	return gomatrixserverlib.NewFederationClientWithTransport(
+		base.Base.Cfg.Matrix.ServerName, base.Base.Cfg.Matrix.KeyID, base.Base.Cfg.Matrix.PrivateKey, tr,
+	)
+}
+
+func main() {
+	instanceName := flag.String("name", "dendrite-p2p", "the name of this P2P demo instance")
+	instancePort := flag.Int("port", 8080, "the port that the client API will listen on")
+	flag.Parse()
+
+	filename := fmt.Sprintf("%s-private.key", *instanceName)
+	_, err := os.Stat(filename)
+	var privKey ed25519.PrivateKey
+	if os.IsNotExist(err) {
+		_, privKey, _ = ed25519.GenerateKey(nil)
+		if err = ioutil.WriteFile(filename, privKey, 0600); err != nil {
+			fmt.Printf("Couldn't write private key to file '%s': %s\n", filename, err)
+		}
+	} else {
+		privKey, err = ioutil.ReadFile(filename)
+		if err != nil {
+			fmt.Printf("Couldn't read private key from file '%s': %s\n", filename, err)
+			_, privKey, _ = ed25519.GenerateKey(nil)
+		}
+	}
+
+	cfg := config.Dendrite{}
+	cfg.Matrix.ServerName = "p2p"
+	cfg.Matrix.PrivateKey = privKey
+	cfg.Matrix.KeyID = gomatrixserverlib.KeyID(fmt.Sprintf("ed25519:%s", *instanceName))
+	cfg.Kafka.UseNaffka = true
+	cfg.Kafka.Topics.OutputRoomEvent = "roomserverOutput"
+	cfg.Kafka.Topics.OutputClientData = "clientapiOutput"
+	cfg.Kafka.Topics.OutputTypingEvent = "typingServerOutput"
+	cfg.Kafka.Topics.UserUpdates = "userUpdates"
+	cfg.Database.Account = config.DataSource(fmt.Sprintf("file:%s-account.db", *instanceName))
+	cfg.Database.Device = config.DataSource(fmt.Sprintf("file:%s-device.db", *instanceName))
+	cfg.Database.MediaAPI = config.DataSource(fmt.Sprintf("file:%s-mediaapi.db", *instanceName))
+	cfg.Database.SyncAPI = config.DataSource(fmt.Sprintf("file:%s-syncapi.db", *instanceName))
+	cfg.Database.RoomServer = config.DataSource(fmt.Sprintf("file:%s-roomserver.db", *instanceName))
+	cfg.Database.ServerKey = config.DataSource(fmt.Sprintf("file:%s-serverkey.db", *instanceName))
+	cfg.Database.FederationSender = config.DataSource(fmt.Sprintf("file:%s-federationsender.db", *instanceName))
+	cfg.Database.AppService = config.DataSource(fmt.Sprintf("file:%s-appservice.db", *instanceName))
+	cfg.Database.PublicRoomsAPI = config.DataSource(fmt.Sprintf("file:%s-publicroomsa.db", *instanceName))
+	cfg.Database.Naffka = config.DataSource(fmt.Sprintf("file:%s-naffka.db", *instanceName))
+	if err = cfg.Derive(); err != nil {
+		panic(err)
+	}
+
+	base := NewP2PDendrite(&cfg, "Monolith")
+	defer base.Base.Close() // nolint: errcheck
+
+	accountDB := base.Base.CreateAccountsDB()
+	deviceDB := base.Base.CreateDeviceDB()
+	keyDB := createKeyDB(base)
+	federation := createFederationClient(base)
+	keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
+
+	alias, input, query := roomserver.SetupRoomServerComponent(&base.Base)
+	eduInputAPI := eduserver.SetupEDUServerComponent(&base.Base, cache.New())
+	asQuery := appservice.SetupAppServiceAPIComponent(
+		&base.Base, accountDB, deviceDB, federation, alias, query, transactions.New(),
+	)
+	fedSenderAPI := federationsender.SetupFederationSenderComponent(&base.Base, federation, query)
+
+	clientapi.SetupClientAPIComponent(
+		&base.Base, deviceDB, accountDB,
+		federation, &keyRing, alias, input, query,
+		eduInputAPI, asQuery, transactions.New(), fedSenderAPI,
+	)
+	eduProducer := producers.NewEDUServerProducer(eduInputAPI)
+	federationapi.SetupFederationAPIComponent(&base.Base, accountDB, deviceDB, federation, &keyRing, alias, input, query, asQuery, fedSenderAPI, eduProducer)
+	mediaapi.SetupMediaAPIComponent(&base.Base, deviceDB)
+	publicRoomsDB, err := storage.NewPublicRoomsServerDatabaseWithPubSub(string(base.Base.Cfg.Database.PublicRoomsAPI), base.LibP2PPubsub)
+	if err != nil {
+		logrus.WithError(err).Panicf("failed to connect to public rooms db")
+	}
+	publicroomsapi.SetupPublicRoomsAPIComponent(&base.Base, deviceDB, publicRoomsDB, query, federation, nil) // Check this later
+	syncapi.SetupSyncAPIComponent(&base.Base, deviceDB, accountDB, query, federation, &cfg)
+
+	httpHandler := common.WrapHandlerInCORS(base.Base.APIMux)
+
+	// Set up the API endpoints we handle. /metrics is for prometheus, and is
+	// not wrapped by CORS, while everything else is
+	http.Handle("/metrics", promhttp.Handler())
+	http.Handle("/", httpHandler)
+
+	// Expose the matrix APIs directly rather than putting them under a /api path.
+	go func() {
+		httpBindAddr := fmt.Sprintf(":%d", *instancePort)
+		logrus.Info("Listening on ", httpBindAddr)
+		logrus.Fatal(http.ListenAndServe(httpBindAddr, nil))
+	}()
+	// Expose the matrix APIs also via libp2p
+	if base.LibP2P != nil {
+		go func() {
+			logrus.Info("Listening on libp2p host ID ", base.LibP2P.ID())
+			listener, err := gostream.Listen(base.LibP2P, "/matrix")
+			if err != nil {
+				panic(err)
+			}
+			defer func() {
+				logrus.Fatal(listener.Close())
+			}()
+			logrus.Fatal(http.Serve(listener, nil))
+		}()
+	}
+
+	// We want to block forever to let the HTTP and HTTPS handler serve the APIs
+	select {}
+}

cmd/dendrite-demo-libp2p/mdnslistener.go

@@ -0,0 +1,63 @@
+// Copyright 2020 The Matrix.org Foundation C.I.C.
+//
+// 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 in writing, software
+// distributed under the License is distributed on an "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.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"math"
+
+	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/matrix-org/dendrite/common/keydb"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+type mDNSListener struct {
+	keydb keydb.Database
+	host  host.Host
+}
+
+func (n *mDNSListener) HandlePeerFound(p peer.AddrInfo) {
+	if err := n.host.Connect(context.Background(), p); err != nil {
+		fmt.Println("Error adding peer", p.ID.String(), "via mDNS:", err)
+	}
+	if pubkey, err := p.ID.ExtractPublicKey(); err == nil {
+		raw, _ := pubkey.Raw()
+		if err := n.keydb.StoreKeys(
+			context.Background(),
+			map[gomatrixserverlib.PublicKeyLookupRequest]gomatrixserverlib.PublicKeyLookupResult{
+				{
+					ServerName: gomatrixserverlib.ServerName(p.ID.String()),
+					KeyID:      "ed25519:p2pdemo",
+				}: {
+					VerifyKey: gomatrixserverlib.VerifyKey{
+						Key: gomatrixserverlib.Base64String(raw),
+					},
+					ValidUntilTS: math.MaxUint64 >> 1,
+					ExpiredTS:    gomatrixserverlib.PublicKeyNotExpired,
+				},
+			},
+		); err != nil {
+			fmt.Println("Failed to store keys:", err)
+		}
+	}
+	fmt.Println("Discovered", len(n.host.Peerstore().Peers())-1, "other libp2p peer(s):")
+	for _, peer := range n.host.Peerstore().Peers() {
+		if peer != n.host.ID() {
+			fmt.Println("-", peer)
+		}
+	}
+}

cmd/dendrite-demo-libp2p/p2pdendrite.go

@@ -0,0 +1,126 @@
+// Copyright 2020 The Matrix.org Foundation C.I.C.
+//
+// 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 in writing, software
+// distributed under the License is distributed on an "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.
+
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"errors"
+
+	pstore "github.com/libp2p/go-libp2p-core/peerstore"
+	record "github.com/libp2p/go-libp2p-record"
+	"github.com/matrix-org/dendrite/common/basecomponent"
+
+	"github.com/libp2p/go-libp2p"
+	circuit "github.com/libp2p/go-libp2p-circuit"
+	crypto "github.com/libp2p/go-libp2p-core/crypto"
+	routing "github.com/libp2p/go-libp2p-core/routing"
+
+	host "github.com/libp2p/go-libp2p-core/host"
+	dht "github.com/libp2p/go-libp2p-kad-dht"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/matrix-org/gomatrixserverlib"
+
+	"github.com/matrix-org/dendrite/common/config"
+)
+
+// P2PDendrite is a Peer-to-Peer variant of BaseDendrite.
+type P2PDendrite struct {
+	Base basecomponent.BaseDendrite
+
+	// Store our libp2p object so that we can make outgoing connections from it
+	// later
+	LibP2P        host.Host
+	LibP2PContext context.Context
+	LibP2PCancel  context.CancelFunc
+	LibP2PDHT     *dht.IpfsDHT
+	LibP2PPubsub  *pubsub.PubSub
+}
+
+// NewP2PDendrite creates a new instance to be used by a component.
+// The componentName is used for logging purposes, and should be a friendly name
+// of the component running, e.g. SyncAPI.
+func NewP2PDendrite(cfg *config.Dendrite, componentName string) *P2PDendrite {
+	baseDendrite := basecomponent.NewBaseDendrite(cfg, componentName)
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	privKey, err := crypto.UnmarshalEd25519PrivateKey(cfg.Matrix.PrivateKey[:])
+	if err != nil {
+		panic(err)
+	}
+
+	//defaultIP6ListenAddr, _ := multiaddr.NewMultiaddr("/ip6/::/tcp/0")
+	var libp2pdht *dht.IpfsDHT
+	libp2p, err := libp2p.New(ctx,
+		libp2p.Identity(privKey),
+		libp2p.DefaultListenAddrs,
+		//libp2p.ListenAddrs(defaultIP6ListenAddr),
+		libp2p.DefaultTransports,
+		libp2p.Routing(func(h host.Host) (r routing.PeerRouting, err error) {
+			libp2pdht, err = dht.New(ctx, h)
+			if err != nil {
+				return nil, err
+			}
+			libp2pdht.Validator = libP2PValidator{}
+			r = libp2pdht
+			return
+		}),
+		libp2p.EnableAutoRelay(),
+		libp2p.EnableRelay(circuit.OptHop),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	libp2ppubsub, err := pubsub.NewFloodSub(context.Background(), libp2p, []pubsub.Option{
+		pubsub.WithMessageSigning(true),
+	}...)
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Println("Our public key:", privKey.GetPublic())
+	fmt.Println("Our node ID:", libp2p.ID())
+	fmt.Println("Our addresses:", libp2p.Addrs())
+
+	cfg.Matrix.ServerName = gomatrixserverlib.ServerName(libp2p.ID().String())
+
+	return &P2PDendrite{
+		Base:          *baseDendrite,
+		LibP2P:        libp2p,
+		LibP2PContext: ctx,
+		LibP2PCancel:  cancel,
+		LibP2PDHT:     libp2pdht,
+		LibP2PPubsub:  libp2ppubsub,
+	}
+}
+
+type libP2PValidator struct {
+	KeyBook pstore.KeyBook
+}
+
+func (v libP2PValidator) Validate(key string, value []byte) error {
+	ns, _, err := record.SplitKey(key)
+	if err != nil || ns != "matrix" {
+		return errors.New("not Matrix path")
+	}
+	return nil
+}
+
+func (v libP2PValidator) Select(k string, vals [][]byte) (int, error) {
+	return 0, nil
+}