Official peer implementation.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

382 lines
16 KiB

package testing
import (
// Import SQL Cipher
app2 ""
_ ""
mrand "math/rand"
var (
aliceLines = []string{"Hello, I'm Alice", "bye"}
bobLines = []string{"Hi, my name is Bob.", "toodles", "welcome"}
carolLines = []string{"Howdy, thanks!"}
func waitForConnection(t *testing.T, peer peer.CwtchPeer, addr string, target connections.ConnectionState) {
peerName, _ := peer.GetScopedZonedAttribute(attr.LocalScope, attr.ProfileZone, constants.Name)
for {
log.Infof("%v checking connection...\n", peerName)
state := peer.GetPeerState(addr)
log.Infof("Waiting for Peer %v to %v - state: %v\n", peerName, addr, state)
if state == connections.FAILED {
t.Fatalf("%v could not connect to %v", peer.GetOnion(), addr)
if state != target {
log.Infof("peer %v %v waiting connect %v, currently: %v\n", peerName, peer.GetOnion(), addr, connections.ConnectionStateName[state])
time.Sleep(time.Second * 5)
} else {
log.Infof("peer %v %v CONNECTED to %v\n", peerName, peer.GetOnion(), addr)
func TestCwtchPeerIntegration(t *testing.T) {
// Goroutine Monitoring Start..
numGoRoutinesStart := runtime.NumGoroutine()
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...
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 {
torDataDir := ""
if torDataDir, err = ioutil.TempDir(dataDir, "data-dir-"); err != nil {
t.Fatalf("could not create data dir")
acn, err := tor.NewTorACNWithAuth("./tordir", path.Join("..", "tor"), torDataDir, controlPort, tor.HashedPasswordAuthenticator{Password: base64.StdEncoding.EncodeToString(key)})
if err != nil {
t.Fatalf("Could not start Tor: %v", err)
defer acn.Close()
// ***** 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)
numGoRoutinesPostAppStart := runtime.NumGoroutine()
// ***** cwtchPeer setup *****
log.Infoln("Creating Alice...")
app.CreateTaggedPeer("Alice", "asdfasdf", "test")
log.Infoln("Creating Bob...")
app.CreateTaggedPeer("Bob", "asdfasdf", "test")
log.Infoln("Creating Carol...")
app.CreateTaggedPeer("Carol", "asdfasdf", "test")
alice := app2.WaitGetPeer(app, "Alice")
log.Infoln("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 := app2.WaitGetPeer(app, "Bob")
log.Infoln("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 := app2.WaitGetPeer(app, "Carol")
log.Infoln("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})
waitTime := time.Duration(60) * time.Second
log.Infof("** Waiting for Alice, Bob, and Carol to connect with onion network... (%v)\n", waitTime)
numGoRoutinesPostPeerStart := runtime.NumGoroutine()
log.Infof("** Wait Done!")
// ***** Peering, server joining, group creation / invite *****
log.Infoln("Alice peering with Bob...")
// Simulate Alice Adding Bob
alice2bobConversationID, err := alice.NewContactConversation(bob.GetOnion(), model.DefaultP2PAccessControl(), true)
if err != nil {
t.Fatalf("error adding conversaiton %v", alice2bobConversationID)
bob2aliceConversationID, err := bob.NewContactConversation(alice.GetOnion(), model.DefaultP2PAccessControl(), true)
if err != nil {
t.Fatalf("error adding conversaiton %v", bob2aliceConversationID)
log.Infof("Alice peering with Carol...")
// Simulate Alice Adding Carol
alice2carolConversationID, err := alice.NewContactConversation(carol.GetOnion(), model.DefaultP2PAccessControl(), true)
if err != nil {
t.Fatalf("error adding conversaiton %v", alice2carolConversationID)
carol2aliceConversationID, err := carol.NewContactConversation(alice.GetOnion(), model.DefaultP2PAccessControl(), true)
if err != nil {
t.Fatalf("error adding conversaiton %v", carol2aliceConversationID)
// Test that we can rekey alice without issues...
err = alice.ChangePassword("asdfasdf", "password 1 2 3", "password 1 2 3")
if err != nil {
t.Fatalf("error changing password for Alice: %v", err)
if !alice.CheckPassword("password 1 2 3") {
t.Fatalf("Alice password did not change...")
waitForConnection(t, alice, bob.GetOnion(), connections.AUTHENTICATED)
waitForConnection(t, alice, carol.GetOnion(), connections.AUTHENTICATED)
waitForConnection(t, bob, alice.GetOnion(), connections.AUTHENTICATED)
waitForConnection(t, carol, alice.GetOnion(), connections.AUTHENTICATED)
log.Infof("Alice and Bob getVal")
alice.SendScopedZonedGetValToContact(alice2bobConversationID, attr.PublicScope, attr.ProfileZone, constants.Name)
bob.SendScopedZonedGetValToContact(bob2aliceConversationID, attr.PublicScope, attr.ProfileZone, constants.Name)
alice.SendScopedZonedGetValToContact(alice2carolConversationID, attr.PublicScope, attr.ProfileZone, constants.Name)
carol.SendScopedZonedGetValToContact(carol2aliceConversationID, 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, err := bob.GetConversationAttribute(bob2aliceConversationID, attr.PublicScope.ConstructScopedZonedPath(attr.ProfileZone.ConstructZonedPath(constants.Name)))
if err != nil || aliceName != "Alice" {
t.Fatalf("Bob: alice GetKeyVal error on alice %v: %v\n", aliceName, err)
log.Infof("Bob has alice's name as '%v'\n", aliceName)
bobName, err := alice.GetConversationAttribute(alice2bobConversationID, attr.PublicScope.ConstructScopedZonedPath(attr.ProfileZone.ConstructZonedPath(constants.Name)))
if err != nil || bobName != "Bob" {
t.Fatalf("Alice: bob GetKeyVal error on bob %v: %v \n", bobName, err)
log.Infof("Alice has bob's name as '%v'\n", bobName)
aliceName, err = carol.GetConversationAttribute(carol2aliceConversationID, attr.PublicScope.ConstructScopedZonedPath(attr.ProfileZone.ConstructZonedPath(constants.Name)))
if err != nil || aliceName != "Alice" {
t.Fatalf("carol GetKeyVal error for alice %v: %v\n", aliceName, err)
carolName, err := alice.GetConversationAttribute(alice2carolConversationID, attr.PublicScope.ConstructScopedZonedPath(attr.ProfileZone.ConstructZonedPath(constants.Name)))
if err != nil || carolName != "Carol" {
t.Fatalf("alice GetKeyVal error, carol %v: %v\n", carolName, err)
log.Infof("Alice has carol's name as '%v'\n", carolName)
// Group Testing
// Simulate Alice Creating a Group
log.Infoln("Alice joining server...")
if _, err := alice.AddServer(string(serverKeyBundle)); err != nil {
t.Fatalf("Failed to Add Server Bundle %v", err)
log.Infof("Waiting for alice to join server...")
waitForConnection(t, alice, ServerAddr, connections.SYNCED)
// Creating a Group
log.Infof("Creating group on %v...", ServerAddr)
aliceGroupConversationID, err := alice.StartGroup("Our Cool Testing Group", ServerAddr)
log.Infof("Created group: %v!\n", aliceGroupConversationID)
if err != nil {
t.Errorf("Failed to init group: %v", err)
// Invites
log.Infoln("Alice inviting Bob to group...")
_, err = alice.SendInviteToConversation(alice2bobConversationID, aliceGroupConversationID)
if err != nil {
t.Fatalf("Error for Alice inviting Bob to group: %v", err)
time.Sleep(time.Second * 5)
// Alice invites Bob to the Group...
message, _, err := bob.GetChannelMessage(bob2aliceConversationID, 0, 1)
log.Infof("Alice message to Bob %v %v", message, err)
var overlayMessage model.MessageWrapper
json.Unmarshal([]byte(message), &overlayMessage)
log.Infof("Parsed Overlay Message: %v", overlayMessage)
err = bob.ImportBundle(overlayMessage.Data)
log.Infof("Result of Bob Importing the Bundle from Alice: %v", err)
log.Infof("Waiting for Bob to join connect to group server...")
waitForConnection(t, bob, ServerAddr, connections.SYNCED)
// 1 = Alice
// 2 = Server
// 3 = Group...
bobGroupConversationID := 3
numGoRoutinesPostServerConnect := runtime.NumGoroutine()
// ***** Conversation *****
log.Infof("Starting conversation in group...")
checkSendMessageToGroup(t, alice, aliceGroupConversationID, aliceLines[0])
checkSendMessageToGroup(t, bob, bobGroupConversationID, bobLines[0])
checkSendMessageToGroup(t, alice, aliceGroupConversationID, aliceLines[1])
checkSendMessageToGroup(t, bob, bobGroupConversationID, bobLines[1])
// Pretend that Carol Aquires the Overlay Message through some other means...
json.Unmarshal([]byte(message), &overlayMessage)
log.Infof("Parsed Overlay Message: %v", overlayMessage)
err = carol.ImportBundle(overlayMessage.Data)
log.Infof("Result of Carol Importing the Bundle from Alice: %v", err)
log.Infof("Waiting for Carol to join connect to group server...")
carolGroupConversationID := 3
waitForConnection(t, carol, ServerAddr, connections.SYNCED)
numGoRoutinesPostCarolConnect := runtime.NumGoroutine()
// Check Alice Timeline
log.Infof("Checking Alice's Timeline...")
checkMessage(t, alice, aliceGroupConversationID, 1, aliceLines[0])
checkMessage(t, alice, aliceGroupConversationID, 2, bobLines[0])
checkMessage(t, alice, aliceGroupConversationID, 3, aliceLines[1])
checkMessage(t, alice, aliceGroupConversationID, 4, bobLines[1])
log.Infof("Shutting down Alice...")
time.Sleep(time.Second * 3)
numGoRoutinesPostAlice := runtime.NumGoroutine()
checkSendMessageToGroup(t, carol, carolGroupConversationID, carolLines[0])
checkSendMessageToGroup(t, bob, bobGroupConversationID, bobLines[2])
// Time to Sync
time.Sleep(time.Second * 10)
// Check Bob Timeline
log.Infof("Checking Bob's Timeline...")
checkMessage(t, bob, bobGroupConversationID, 1, aliceLines[0])
checkMessage(t, bob, bobGroupConversationID, 2, bobLines[0])
checkMessage(t, bob, bobGroupConversationID, 3, aliceLines[1])
checkMessage(t, bob, bobGroupConversationID, 4, bobLines[1])
checkMessage(t, bob, bobGroupConversationID, 5, carolLines[0])
checkMessage(t, bob, bobGroupConversationID, 6, bobLines[2])
// Check Carol Timeline
log.Infof("Checking Carols's Timeline...")
checkMessage(t, carol, carolGroupConversationID, 1, aliceLines[0])
checkMessage(t, carol, carolGroupConversationID, 2, bobLines[0])
checkMessage(t, carol, carolGroupConversationID, 3, aliceLines[1])
checkMessage(t, carol, carolGroupConversationID, 4, bobLines[1])
checkMessage(t, carol, carolGroupConversationID, 5, carolLines[0])
checkMessage(t, carol, carolGroupConversationID, 6, bobLines[2])
log.Infof("Shutting down Bob...")
time.Sleep(time.Second * 3)
numGoRoutinesPostBob := runtime.NumGoroutine()
log.Infof("Shutting down Carol...")
time.Sleep(time.Second * 3)
numGoRoutinesPostCarol := runtime.NumGoroutine()
log.Infof("Shutting down apps...")
log.Infof("app Shutdown: %v\n", runtime.NumGoroutine())
time.Sleep(2 * time.Second)
log.Infof("Done shutdown: %v\n", runtime.NumGoroutine())
log.Infof("Shutting down ACN...")
time.Sleep(time.Second * 30) // the network status plugin might keep goroutines alive for a minute before killing them
numGoRoutinesPostAppShutdown := runtime.NumGoroutine()
// Printing out the current goroutines
// Very useful if we are leaking any.
pprof.Lookup("goroutine").WriteTo(os.Stdout, 1)
log.Infof("numGoRoutinesStart: %v\nnumGoRoutinesPostAppStart: %v\nnumGoRoutinesPostPeerStart: %v\nnumGoRoutinesPostPeerAndServerConnect: %v\n"+
"numGoRoutinesPostAlice: %v\nnumGoRoutinesPostCarolConnect: %v\nnumGoRoutinesPostBob: %v\nnumGoRoutinesPostCarol: %v\nnumGoRoutinesPostAppShutdown: %v",
numGoRoutinesStart, numGoRoutinesPostAppStart, numGoRoutinesPostPeerStart, numGoRoutinesPostServerConnect,
numGoRoutinesPostAlice, numGoRoutinesPostCarolConnect, numGoRoutinesPostBob, numGoRoutinesPostCarol, numGoRoutinesPostAppShutdown)
if numGoRoutinesStart != numGoRoutinesPostAppShutdown {
t.Errorf("Number of GoRoutines at start (%v) does not match number of goRoutines after cleanup of peers and servers (%v), clean up failed, v detected!", numGoRoutinesStart, numGoRoutinesPostAppShutdown)
// Utility function for sending a message from a peer to a group
func checkSendMessageToGroup(t *testing.T, profile peer.CwtchPeer, id int, message string) {
name, _ := profile.GetScopedZonedAttribute(attr.PublicScope, attr.ProfileZone, constants.Name)
log.Infof("%v> %v\n", name, message)
_, err := profile.SendMessage(id, message)
if err != nil {
t.Fatalf("Alice failed to send a message to the group: %v", err)
time.Sleep(time.Second * 10)
// Utility function for testing that a message in a conversation is as expected
func checkMessage(t *testing.T, profile peer.CwtchPeer, id int, messageID int, expected string) {
message, _, err := profile.GetChannelMessage(id, 0, messageID)
if err != nil {
t.Fatalf("unexpected message %v expected: %v got error: %v", profile.GetOnion(), expected, err)
if message != expected {
t.Fatalf("unexpected message %v expected: %v got: [%v]", profile.GetOnion(), expected, message)