cwtch/testing/cwtch_peer_server_integration_test.go

package testing

import (
	"crypto/rand"
	app2 "cwtch.im/cwtch/app"
	"cwtch.im/cwtch/app/utils"
	"cwtch.im/cwtch/event"
	"cwtch.im/cwtch/event/bridge"
	"cwtch.im/cwtch/model"
	"cwtch.im/cwtch/model/attr"
	"cwtch.im/cwtch/model/constants"
	"cwtch.im/cwtch/peer"
	"cwtch.im/cwtch/protocol/connections"
	"encoding/base64"
	"fmt"
	"git.openprivacy.ca/openprivacy/connectivity/tor"
	"git.openprivacy.ca/openprivacy/log"
	mrand "math/rand"
	"os"
	"os/user"
	"path"
	"runtime"
	"runtime/pprof"
	"strings"
	"testing"
	"time"
)

var (
	aliceLines = []string{"Hello, I'm Alice", "bye"}
	bobLines   = []string{"Hi, my name is Bob.", "toodles", "welcome"}
	carolLines = []string{"Howdy, thanks!"}
)

func printAndCountVerifedTimeline(t *testing.T, timeline []model.Message) int {
	numVerified := 0
	for _, message := range timeline {
		fmt.Printf("%v %v> %s\n", message.Timestamp, message.PeerID, message.Message)
		numVerified++
	}
	return numVerified
}

func waitForPeerGroupConnection(t *testing.T, peer peer.CwtchPeer, groupID string) {
	peerName, _ := peer.GetScopedZonedAttribute(attr.LocalScope, attr.ProfileZone, constants.Name)
	for {
		fmt.Printf("%v checking group connection...\n", peerName)
		state, ok := peer.GetGroupState(groupID)
		if ok {
			fmt.Printf("Waiting for Peer %v to join group %v - state: %v\n", peerName, groupID, state)
			if state == connections.FAILED {
				t.Fatalf("%v could not connect to %v", peer.GetOnion(), groupID)
			}
			if state != connections.SYNCED {
				fmt.Printf("peer %v  %v waiting connect to group %v, currently: %v\n", peerName, peer.GetOnion(), groupID, connections.ConnectionStateName[state])
				time.Sleep(time.Second * 5)
				continue
			} else {
				fmt.Printf("peer %v  %v CONNECTED to group %v\n", peerName, peer.GetOnion(), groupID)
				break
			}
		}
		time.Sleep(time.Second * 2)
	}
	return
}

func waitForPeerPeerConnection(t *testing.T, peera peer.CwtchPeer, peerb peer.CwtchPeer) {
	for {
		state, ok := peera.GetPeerState(peerb.GetOnion())
		if ok {
			//log.Infof("Waiting for Peer %v to peer with peer: %v - state: %v\n", peera.GetProfile().Name, peerb.GetProfile().Name, state)
			if state == connections.FAILED {
				t.Fatalf("%v could not connect to %v", peera.GetOnion(), peerb.GetOnion())
			}
			if state != connections.AUTHENTICATED {
				fmt.Printf("peer %v waiting connect to peer %v, currently: %v\n", peera.GetOnion(), peerb.GetOnion(), connections.ConnectionStateName[state])
				time.Sleep(time.Second * 5)
				continue
			} else {
				peerAName, _ := peera.GetScopedZonedAttribute(attr.LocalScope, attr.ProfileZone, constants.Name)
				peerBName, _ := peerb.GetScopedZonedAttribute(attr.LocalScope, attr.ProfileZone, constants.Name)
				fmt.Printf("%v CONNECTED and AUTHED to %v\n", peerAName, peerBName)
				break
			}
		}
	}
	return
}

func TestCwtchPeerIntegration(t *testing.T) {
	numGoRoutinesStart := runtime.NumGoroutine()

	log.AddEverythingFromPattern("connectivity")
	log.SetLevel(log.LevelDebug)
	log.ExcludeFromPattern("connection/connection")
	log.ExcludeFromPattern("outbound/3dhauthchannel")
	log.ExcludeFromPattern("event/eventmanager")
	log.ExcludeFromPattern("pipeBridge")
	log.ExcludeFromPattern("tapir")
	os.Mkdir("tordir", 0700)
	dataDir := path.Join("tordir", "tor")
	os.MkdirAll(dataDir, 0700)

	// we don't need real randomness for the port, just to avoid a possible conflict...
	mrand.Seed(int64(time.Now().Nanosecond()))
	socksPort := mrand.Intn(1000) + 9051
	controlPort := mrand.Intn(1000) + 9052

	// generate a random password
	key := make([]byte, 64)
	_, err := rand.Read(key)
	if err != nil {
		panic(err)
	}

	tor.NewTorrc().WithSocksPort(socksPort).WithOnionTrafficOnly().WithHashedPassword(base64.StdEncoding.EncodeToString(key)).WithControlPort(controlPort).Build("tordir/tor/torrc")
	acn, err := tor.NewTorACNWithAuth("./tordir", path.Join("..", "tor"), controlPort, tor.HashedPasswordAuthenticator{Password: base64.StdEncoding.EncodeToString(key)})
	if err != nil {
		t.Fatalf("Could not start Tor: %v", err)
	}
	pid, _ := acn.GetPID()
	t.Logf("Tor pid: %v", pid)

	// ***** Cwtch Server management *****

	const ServerKeyBundleBase64 = "eyJLZXlzIjp7ImJ1bGxldGluX2JvYXJkX29uaW9uIjoibmZoeHp2enhpbnJpcGdkaDR0Mm00eGN5M2NyZjZwNGNiaGVjdGdja3VqM2lkc2pzYW90Z293YWQiLCJwcml2YWN5X3Bhc3NfcHVibGljX2tleSI6IjVwd2hQRGJ0c0EvdFI3ZHlUVUkzakpZZnM1L3Jaai9iQ1ZWZEpTc0Jtbk09IiwidG9rZW5fc2VydmljZV9vbmlvbiI6ImVvd25mcTRsNTZxMmU0NWs0bW03MjdsanJod3Z0aDZ5ZWN0dWV1bXB4emJ5cWxnbXVhZm1qdXFkIn0sIlNpZ25hdHVyZSI6IlY5R3NPMHNZWFJ1bGZxdzdmbGdtclVxSTBXS0JlSFIzNjIvR3hGbWZPekpEZjJaRks2ck9jNVRRR1ZxVWIrbXIwV2xId0pwdXh0UW1JRU9KNkplYkNRPT0ifQ=="
	const ServerAddr = "nfhxzvzxinripgdh4t2m4xcy3crf6p4cbhectgckuj3idsjsaotgowad"
	serverKeyBundle, _ := base64.StdEncoding.DecodeString(ServerKeyBundleBase64)

	app := app2.NewApp(acn, "./storage")

	usr, _ := user.Current()
	cwtchDir := path.Join(usr.HomeDir, ".cwtch")
	os.Mkdir(cwtchDir, 0700)
	os.RemoveAll(path.Join(cwtchDir, "testing"))
	os.Mkdir(path.Join(cwtchDir, "testing"), 0700)
	bridgeClient := bridge.NewPipeBridgeClient(path.Join(cwtchDir, "testing/clientPipe"), path.Join(cwtchDir, "testing/servicePipe"))
	bridgeService := bridge.NewPipeBridgeService(path.Join(cwtchDir, "testing/servicePipe"), path.Join(cwtchDir, "testing/clientPipe"))
	appClient := app2.NewAppClient("./storage", bridgeClient)
	appService := app2.NewAppService(acn, "./storage", bridgeService)

	numGoRoutinesPostAppStart := runtime.NumGoroutine()

	// ***** cwtchPeer setup *****

	fmt.Println("Creating Alice...")
	app.CreatePeer("alice", "asdfasdf")

	fmt.Println("Creating Bob...")
	app.CreatePeer("bob", "asdfasdf")

	fmt.Println("Creating Carol...")
	appClient.CreatePeer("carol", "asdfasdf")

	alice := utils.WaitGetPeer(app, "alice")
	fmt.Println("Alice created:", alice.GetOnion())
	alice.SetScopedZonedAttribute(attr.PublicScope, attr.ProfileZone, constants.Name, "Alice")
	alice.AutoHandleEvents([]event.Type{event.PeerStateChange, event.ServerStateChange, event.NewGroupInvite, event.NewRetValMessageFromPeer})

	bob := utils.WaitGetPeer(app, "bob")
	fmt.Println("Bob created:", bob.GetOnion())
	bob.SetScopedZonedAttribute(attr.PublicScope, attr.ProfileZone, constants.Name, "Bob")
	bob.AutoHandleEvents([]event.Type{event.PeerStateChange, event.ServerStateChange, event.NewGroupInvite, event.NewRetValMessageFromPeer})

	carol := utils.WaitGetPeer(appClient, "carol")
	fmt.Println("Carol created:", carol.GetOnion())
	carol.SetScopedZonedAttribute(attr.PublicScope, attr.ProfileZone, constants.Name, "Carol")
	carol.AutoHandleEvents([]event.Type{event.PeerStateChange, event.ServerStateChange, event.NewGroupInvite, event.NewRetValMessageFromPeer})

	app.LaunchPeers()
	appClient.LaunchPeers()

	waitTime := time.Duration(60) * time.Second
	t.Logf("** Waiting for Alice, Bob, and Carol to connect with onion network... (%v)\n", waitTime)
	time.Sleep(waitTime)
	numGoRoutinesPostPeerStart := runtime.NumGoroutine()
	fmt.Println("** Wait Done!")

	// ***** Peering, server joining, group creation / invite *****

	fmt.Println("Alice joining server...")
	if _, err := alice.AddServer(string(serverKeyBundle)); err != nil {
		t.Fatalf("Failed to Add Server Bundle %v", err)
	}
	alice.JoinServer(ServerAddr)

	fmt.Println("Alice peering with Bob...")
	alice.PeerWithOnion(bob.GetOnion())

	fmt.Println("Alice peering with Carol...")
	alice.PeerWithOnion(carol.GetOnion())

	fmt.Println("Creating group on ", ServerAddr, "...")
	groupID, _, err := alice.StartGroup(ServerAddr)
	fmt.Printf("Created group: %v!\n", groupID)
	if err != nil {
		t.Errorf("Failed to init group: %v", err)
		return
	}

	fmt.Println("Waiting for alice to join server...")
	waitForPeerGroupConnection(t, alice, groupID)

	fmt.Println("Waiting for alice and Bob to peer...")
	waitForPeerPeerConnection(t, alice, bob)
	// Need to add contact else SetContactAuth fails on peer peer doesnt exist
	// Normal flow would be Bob app monitors for the new connection (a new connection state change to Auth
	//   and the adds the user to peer, and then approves or blocks it
	bob.AddContact("alice?", alice.GetOnion(), model.AuthApproved)
	bob.AddServer(string(serverKeyBundle))
	bob.SetContactAuthorization(alice.GetOnion(), model.AuthApproved)

	waitForPeerPeerConnection(t, alice, carol)
	carol.AddContact("alice?", alice.GetOnion(), model.AuthApproved)
	carol.AddServer(string(serverKeyBundle))
	carol.SetContactAuthorization(alice.GetOnion(), model.AuthApproved)

	fmt.Println("Alice and Bob getVal public.name...")

	alice.SendScopedZonedGetValToContact(bob.GetOnion(), attr.PublicScope, attr.ProfileZone, constants.Name)
	bob.SendScopedZonedGetValToContact(alice.GetOnion(), attr.PublicScope, attr.ProfileZone, constants.Name)

	alice.SendScopedZonedGetValToContact(carol.GetOnion(), attr.PublicScope, attr.ProfileZone, constants.Name)
	carol.SendScopedZonedGetValToContact(alice.GetOnion(), attr.PublicScope, attr.ProfileZone, constants.Name)

	// This used to be 10, but increasing it to 30 because this is now causing frequent issues
	// Probably related to latency/throughput problems in the underlying tor network.
	time.Sleep(30 * time.Second)

	aliceName, exists := bob.GetContactAttribute(alice.GetOnion(), attr.GetPeerScope(constants.Name))
	if !exists || aliceName != "Alice" {
		t.Fatalf("Bob: alice GetKeyVal error on alice peer.name %v\n", exists)
	}
	fmt.Printf("Bob has alice's name as '%v'\n", aliceName)

	bobName, exists := alice.GetContactAttribute(bob.GetOnion(), attr.GetPeerScope(constants.Name))
	if !exists || bobName != "Bob" {
		t.Fatalf("Alice: bob GetKeyVal error on bob peer.name\n")
	}
	fmt.Printf("Alice has bob's name as '%v'\n", bobName)

	aliceName, exists = carol.GetContactAttribute(alice.GetOnion(), attr.GetPeerScope(constants.Name))
	if !exists || aliceName != "Alice" {
		t.Fatalf("carol GetKeyVal error for alice peer.name %v\n", exists)
	}

	carolName, exists := alice.GetContactAttribute(carol.GetOnion(), attr.GetPeerScope(constants.Name))
	if !exists || carolName != "Carol" {
		t.Fatalf("alice GetKeyVal error, carol peer.name\n")
	}
	fmt.Printf("Alice has carol's name as '%v'\n", carolName)

	fmt.Println("Alice inviting Bob to group...")
	err = alice.InviteOnionToGroup(bob.GetOnion(), groupID)
	if err != nil {
		t.Fatalf("Error for Alice inviting Bob to group: %v", err)
	}

	time.Sleep(time.Second * 5)

	fmt.Println("Bob examining groups and accepting invites...")
	for _, message := range bob.GetContact(alice.GetOnion()).Timeline.GetMessages() {
		fmt.Printf("Found message from Alice: %v", message.Message)
		if strings.HasPrefix(message.Message, "torv3") {
			gid, err := bob.ImportGroup(message.Message)
			if err == nil {
				fmt.Printf("Bob found invite...now accepting %v...", gid)
				bob.AcceptInvite(gid)
			} else {
				t.Fatalf("Bob could not accept invite...%v", gid)
			}
		}
	}

	fmt.Println("Waiting for Bob to join connect to group server...")
	waitForPeerGroupConnection(t, bob, groupID)

	numGoRoutinesPostServerConnect := runtime.NumGoroutine()

	// ***** Conversation *****

	fmt.Println("Starting conversation in group...")
	// Conversation
	fmt.Printf("%v> %v\n", aliceName, aliceLines[0])
	err = alice.SendMessage(groupID, aliceLines[0])
	if err != nil {
		t.Fatalf("Alice failed to send a message to the group: %v", err)
	}
	time.Sleep(time.Second * 10)

	fmt.Printf("%v> %v\n", bobName, bobLines[0])
	err = bob.SendMessage(groupID, bobLines[0])
	if err != nil {
		t.Fatalf("Bob failed to send a message to the group: %v", err)
	}
	time.Sleep(time.Second * 10)

	fmt.Printf("%v> %v\n", aliceName, aliceLines[1])
	alice.SendMessage(groupID, aliceLines[1])
	time.Sleep(time.Second * 10)

	fmt.Printf("%v> %v\n", bobName, bobLines[1])
	bob.SendMessage(groupID, bobLines[1])
	time.Sleep(time.Second * 10)

	fmt.Println("Alice inviting Carol to group...")
	err = alice.InviteOnionToGroup(carol.GetOnion(), groupID)
	if err != nil {
		t.Fatalf("Error for Alice inviting Carol to group: %v", err)
	}
	time.Sleep(time.Second * 60) // Account for some token acquisition in Alice and Bob flows.
	fmt.Println("Carol examining groups and accepting invites...")
	for _, message := range carol.GetContact(alice.GetOnion()).Timeline.GetMessages() {
		fmt.Printf("Found message from Alice: %v", message.Message)
		if strings.HasPrefix(message.Message, "torv3") {
			gid, err := carol.ImportGroup(message.Message)
			if err == nil {
				fmt.Printf("Carol found invite...now accepting %v...", gid)
				carol.AcceptInvite(gid)
			} else {
				t.Fatalf("Carol could not accept invite...%v", gid)
			}
		}
	}

	fmt.Println("Shutting down Alice...")
	app.ShutdownPeer(alice.GetOnion())
	time.Sleep(time.Second * 5)
	numGoRoutinesPostAlice := runtime.NumGoroutine()

	fmt.Println("Carol joining server...")
	carol.JoinServer(ServerAddr)
	waitForPeerGroupConnection(t, carol, groupID)
	numGoRotinesPostCarolConnect := runtime.NumGoroutine()

	fmt.Printf("%v> %v", bobName, bobLines[2])
	bob.SendMessage(groupID, bobLines[2])
	// Bob should have enough tokens so we don't need to account for
	// token acquisition here...

	fmt.Printf("%v> %v", carolName, carolLines[0])
	carol.SendMessage(groupID, carolLines[0])
	time.Sleep(time.Second * 30) // we need to account for spam-based token acquisition, but everything should
	// be warmed-up and delays should be pretty small.

	// ***** Verify Test *****

	fmt.Println("Final syncing time...")
	time.Sleep(time.Second * 30)

	alicesGroup := alice.GetGroup(groupID)
	if alicesGroup == nil {
		t.Error("aliceGroup == nil")
		return
	}

	fmt.Printf("Alice's TimeLine:\n")
	aliceVerified := printAndCountVerifedTimeline(t, alicesGroup.GetTimeline())
	if aliceVerified != 4 {
		t.Errorf("Alice did not have 4 verified messages")
	}

	bobsGroup := bob.GetGroup(groupID)
	if bobsGroup == nil {
		t.Error("bobGroup == nil")
		return
	}
	fmt.Printf("Bob's TimeLine:\n")
	bobVerified := printAndCountVerifedTimeline(t, bobsGroup.GetTimeline())
	if bobVerified != 6 {
		t.Errorf("Bob did not have 6 verified messages")
	}

	carolsGroup := carol.GetGroup(groupID)
	fmt.Printf("Carol's TimeLine:\n")
	carolVerified := printAndCountVerifedTimeline(t, carolsGroup.GetTimeline())
	if carolVerified != 6 {
		t.Errorf("Carol did not have 6 verified messages")
	}

	if len(alicesGroup.GetTimeline()) != 4 {
		t.Errorf("Alice's timeline does not have all messages")
	} else {
		// check message 0,1,2,3
		alicesGroup.Timeline.Sort()
		aliceGroupTimeline := alicesGroup.GetTimeline()
		if aliceGroupTimeline[0].Message != aliceLines[0] || aliceGroupTimeline[1].Message != bobLines[0] ||
			aliceGroupTimeline[2].Message != aliceLines[1] || aliceGroupTimeline[3].Message != bobLines[1] {
			t.Errorf("Some of Alice's timeline messages did not have the expected content!")
		}
	}

	if len(bobsGroup.GetTimeline()) != 6 {
		t.Errorf("Bob's timeline does not have all messages")
	} else {
		// check message 0,1,2,3,4,5
		bobsGroup.Timeline.Sort()
		bobGroupTimeline := bobsGroup.GetTimeline()
		if bobGroupTimeline[0].Message != aliceLines[0] || bobGroupTimeline[1].Message != bobLines[0] ||
			bobGroupTimeline[2].Message != aliceLines[1] || bobGroupTimeline[3].Message != bobLines[1] ||
			bobGroupTimeline[4].Message != bobLines[2] || bobGroupTimeline[5].Message != carolLines[0] {
			t.Errorf("Some of Bob's timeline messages did not have the expected content!")
		}
	}

	if len(carolsGroup.GetTimeline()) != 6 {
		t.Errorf("Carol's timeline does not have all messages")
	} else {
		// check message 0,1,2,3,4,5
		carolsGroup.Timeline.Sort()
		carolGroupTimeline := carolsGroup.GetTimeline()
		if carolGroupTimeline[0].Message != aliceLines[0] || carolGroupTimeline[1].Message != bobLines[0] ||
			carolGroupTimeline[2].Message != aliceLines[1] || carolGroupTimeline[3].Message != bobLines[1] ||
			carolGroupTimeline[4].Message != carolLines[0] || carolGroupTimeline[5].Message != bobLines[2] {
			t.Errorf("Some of Carol's timeline messages did not have the expected content!")
		}
	}

	fmt.Println("Shutting down Bob...")
	app.ShutdownPeer(bob.GetOnion())
	time.Sleep(time.Second * 3)
	numGoRoutinesPostBob := runtime.NumGoroutine()

	fmt.Println("Shutting down Carol...")
	appClient.ShutdownPeer(carol.GetOnion())
	time.Sleep(time.Second * 3)
	numGoRoutinesPostCarol := runtime.NumGoroutine()

	fmt.Println("Shutting down apps...")
	fmt.Printf("app Shutdown: %v\n", runtime.NumGoroutine())
	app.Shutdown()
	fmt.Printf("appClientShutdown: %v\n", runtime.NumGoroutine())
	appClient.Shutdown()
	fmt.Printf("appServiceShutdown: %v\n", runtime.NumGoroutine())
	appService.Shutdown()

	fmt.Printf("bridgeClientShutdown: %v\n", runtime.NumGoroutine())
	bridgeClient.Shutdown()
	time.Sleep(2 * time.Second)

	fmt.Printf("brideServiceShutdown: %v\n", runtime.NumGoroutine())
	bridgeService.Shutdown()
	time.Sleep(2 * time.Second)

	fmt.Printf("Done shutdown: %v\n", runtime.NumGoroutine())
	numGoRoutinesPostAppShutdown := runtime.NumGoroutine()

	fmt.Println("Shutting down ACN...")
	acn.Close()
	time.Sleep(time.Second * 2)  // Server ^^ has a 5 second loop attempting reconnect before exiting
	time.Sleep(time.Second * 30) // the network status plugin might keep goroutines alive for a minute before killing them
	numGoRoutinesPostACN := runtime.NumGoroutine()

	// Printing out the current goroutines
	// Very useful if we are leaking any.
	pprof.Lookup("goroutine").WriteTo(os.Stdout, 1)

	fmt.Printf("numGoRoutinesStart: %v\nnumGoRoutinesPostAppStart: %v\nnumGoRoutinesPostPeerStart: %v\nnumGoRoutinesPostPeerAndServerConnect: %v\n"+
		"numGoRoutinesPostAlice: %v\nnumGoRotinesPostCarolConnect: %v\nnumGoRoutinesPostBob: %v\nnumGoRoutinesPostCarol: %v\nnumGoRoutinesPostAppShutdown: %v\nnumGoRoutinesPostACN: %v\n",
		numGoRoutinesStart, numGoRoutinesPostAppStart, numGoRoutinesPostPeerStart, numGoRoutinesPostServerConnect,
		numGoRoutinesPostAlice, numGoRotinesPostCarolConnect, numGoRoutinesPostBob, numGoRoutinesPostCarol, numGoRoutinesPostAppShutdown, numGoRoutinesPostACN)

	if numGoRoutinesStart != numGoRoutinesPostACN {
		t.Errorf("Number of GoRoutines at start (%v) does not match number of goRoutines after cleanup of peers and servers (%v), clean up failed, leak detected!", numGoRoutinesStart, numGoRoutinesPostACN)
	}
}