From 8083e6fa5c1a0a5693bd5ce191e797272ea0d36b Mon Sep 17 00:00:00 2001 From: Sarah Jamie Lewis Date: Wed, 21 Nov 2018 13:23:59 -0800 Subject: [PATCH 1/7] Adding Method to Save Messages in p2p Transcript --- model/message.go | 6 +++--- model/profile.go | 20 ++++++++++++++++++++ model/profile_test.go | 27 +++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 3 deletions(-) diff --git a/model/message.go b/model/message.go index 4ce71f0..ded861d 100644 --- a/model/message.go +++ b/model/message.go @@ -55,11 +55,11 @@ func (t *Timeline) Swap(i, j int) { t.Messages[i], t.Messages[j] = t.Messages[j], t.Messages[i] } -// Less checks 2 messages (i and j) in the timeline and returns true if i occcured before j, else false +// Less checks 2 messages (i and j) in the timeline and returns true if i occurred before j, else false func (t *Timeline) Less(i, j int) bool { - if t.Messages[i].Timestamp.Before(t.Messages[j].Timestamp) { - return true + if t.Messages[j].Timestamp.Before(t.Messages[i].Timestamp) { + return false } if compareSignatures(t.Messages[i].PreviousMessageSig, t.SignedGroupID) { diff --git a/model/profile.go b/model/profile.go index 8bb48fe..5fceef9 100644 --- a/model/profile.go +++ b/model/profile.go @@ -22,6 +22,7 @@ type PublicProfile struct { Blocked bool Onion string Attributes map[string]string + Timeline Timeline lock sync.Mutex } @@ -102,6 +103,25 @@ func (p *Profile) RejectInvite(groupID string) { p.lock.Unlock() } +// AddMessageToContactTimeline allows the saving of a message sent via a direct connection chat to the profile. +func (p *Profile) AddMessageToContactTimeline(onion string, fromMe bool, message string, sent time.Time) { + p.lock.Lock() + defer p.lock.Unlock() + contact, ok := p.Contacts[onion] + + // We don't really need a Signature here, but we use it to maintain order + now := time.Now() + sig := p.SignMessage(onion + message + sent.String() + now.String()) + if ok { + if fromMe { + contact.Timeline.Insert(&Message{PeerID: p.Onion, Message: message, Timestamp: sent, Received: now, Signature: sig}) + } else { + contact.Timeline.Insert(&Message{PeerID: onion, Message: message, Timestamp: sent, Received: now, Signature: sig}) + } + } +} + + // AcceptInvite accepts a group invite func (p *Profile) AcceptInvite(groupID string) (err error) { p.lock.Lock() diff --git a/model/profile_test.go b/model/profile_test.go index e8493bc..53e845c 100644 --- a/model/profile_test.go +++ b/model/profile_test.go @@ -4,8 +4,33 @@ import ( "cwtch.im/cwtch/protocol" "github.com/golang/protobuf/proto" "testing" + "time" ) + +func TestP2P(t *testing.T) { + sarah := GenerateNewProfile("Sarah") + alice := GenerateNewProfile("Alice") + + sarah.AddContact(alice.Onion, &alice.PublicProfile) + + sarah.AddMessageToContactTimeline(alice.Onion, false, "hello",time.Now()) + sarah.AddMessageToContactTimeline(alice.Onion, true, "world",time.Now()) + + contact,_ := sarah.GetContact(alice.Onion) + for i,m := range contact.Timeline.GetMessages() { + if i == 0 && (m.Message != "hello" || m.PeerID != alice.Onion) { + t.Fatalf("Timeline is invalid: %v", m) + } + + if i == 1 && (m.Message != "world" || m.PeerID != sarah.Onion) { + t.Fatalf("Timeline is invalid: %v", m) + } + t.Logf("Message: %v", m) + } +} + + func TestProfileIdentity(t *testing.T) { sarah := GenerateNewProfile("Sarah") alice := GenerateNewProfile("Alice") @@ -66,6 +91,8 @@ func TestBlockPeer(t *testing.T) { } } + + func TestAcceptNonExistentGroup(t *testing.T) { sarah := GenerateNewProfile("Sarah") sarah.AcceptInvite("doesnotexist") From dc489398eaf22df3546ebb588dc80e4ae18b10f8 Mon Sep 17 00:00:00 2001 From: Sarah Jamie Lewis Date: Wed, 21 Nov 2018 14:15:43 -0800 Subject: [PATCH 2/7] Checking if a group is setup on a v3 onion server --- model/group.go | 5 +++++ model/group_test.go | 9 ++++++++- model/message_test.go | 4 ++-- model/profile.go | 1 - model/profile_test.go | 22 +++++++++------------ peer/connections/peerpeerconnection_test.go | 2 +- peer/cwtch_peer_test.go | 2 +- 7 files changed, 26 insertions(+), 19 deletions(-) diff --git a/model/group.go b/model/group.go index d28db70..9adf2e8 100644 --- a/model/group.go +++ b/model/group.go @@ -34,6 +34,11 @@ type Group struct { // NewGroup initializes a new group associated with a given CwtchServer func NewGroup(server string) (*Group, error) { group := new(Group) + + if utils.IsValidHostname(server) == false { + return nil, errors.New("Server is not a valid v3 onion") + } + group.GroupServer = server var groupID [16]byte diff --git a/model/group_test.go b/model/group_test.go index 233eaac..00b52ae 100644 --- a/model/group_test.go +++ b/model/group_test.go @@ -8,7 +8,7 @@ import ( ) func TestGroup(t *testing.T) { - g, _ := NewGroup("server.onion") + g, _ := NewGroup("2c3kmoobnyghj2zw6pwv7d57yzld753auo3ugauezzpvfak3ahc4bdyd") dgm := &protocol.DecryptedGroupMessage{ Onion: proto.String("onion"), Text: proto.String("Hello World!"), @@ -30,3 +30,10 @@ func TestGroup(t *testing.T) { } t.Logf("Got message %v", message) } + +func TestGroupErr(t *testing.T) { + _, err := NewGroup("not a real group name") + if err == nil { + t.Errorf("Group Setup Should Have Failed") + } +} diff --git a/model/message_test.go b/model/message_test.go index a639d59..c66d0d6 100644 --- a/model/message_test.go +++ b/model/message_test.go @@ -16,7 +16,7 @@ func TestMessagePadding(t *testing.T) { sarah.AddContact(alice.Onion, &alice.PublicProfile) alice.AddContact(sarah.Onion, &sarah.PublicProfile) - gid, invite, _ := alice.StartGroup("aaa.onion") + gid, invite, _ := alice.StartGroup("2c3kmoobnyghj2zw6pwv7d57yzld753auo3ugauezzpvfak3ahc4bdyd") gci := &protocol.CwtchPeerPacket{} proto.Unmarshal(invite, gci) sarah.ProcessInvite(gci.GetGroupChatInvite(), alice.Onion) @@ -50,7 +50,7 @@ func TestTranscriptConsistency(t *testing.T) { sarah.AddContact(alice.Onion, &alice.PublicProfile) alice.AddContact(sarah.Onion, &sarah.PublicProfile) - gid, invite, _ := alice.StartGroup("aaa.onion") + gid, invite, _ := alice.StartGroup("2c3kmoobnyghj2zw6pwv7d57yzld753auo3ugauezzpvfak3ahc4bdyd") gci := &protocol.CwtchPeerPacket{} proto.Unmarshal(invite, gci) sarah.ProcessInvite(gci.GetGroupChatInvite(), alice.Onion) diff --git a/model/profile.go b/model/profile.go index 5fceef9..9096a42 100644 --- a/model/profile.go +++ b/model/profile.go @@ -121,7 +121,6 @@ func (p *Profile) AddMessageToContactTimeline(onion string, fromMe bool, message } } - // AcceptInvite accepts a group invite func (p *Profile) AcceptInvite(groupID string) (err error) { p.lock.Lock() diff --git a/model/profile_test.go b/model/profile_test.go index 53e845c..ba02b23 100644 --- a/model/profile_test.go +++ b/model/profile_test.go @@ -7,30 +7,28 @@ import ( "time" ) - func TestP2P(t *testing.T) { sarah := GenerateNewProfile("Sarah") alice := GenerateNewProfile("Alice") sarah.AddContact(alice.Onion, &alice.PublicProfile) - sarah.AddMessageToContactTimeline(alice.Onion, false, "hello",time.Now()) - sarah.AddMessageToContactTimeline(alice.Onion, true, "world",time.Now()) + sarah.AddMessageToContactTimeline(alice.Onion, false, "hello", time.Now()) + sarah.AddMessageToContactTimeline(alice.Onion, true, "world", time.Now()) - contact,_ := sarah.GetContact(alice.Onion) - for i,m := range contact.Timeline.GetMessages() { - if i == 0 && (m.Message != "hello" || m.PeerID != alice.Onion) { + contact, _ := sarah.GetContact(alice.Onion) + for i, m := range contact.Timeline.GetMessages() { + if i == 0 && (m.Message != "hello" || m.PeerID != alice.Onion) { t.Fatalf("Timeline is invalid: %v", m) } - if i == 1 && (m.Message != "world" || m.PeerID != sarah.Onion) { + if i == 1 && (m.Message != "world" || m.PeerID != sarah.Onion) { t.Fatalf("Timeline is invalid: %v", m) } t.Logf("Message: %v", m) } } - func TestProfileIdentity(t *testing.T) { sarah := GenerateNewProfile("Sarah") alice := GenerateNewProfile("Alice") @@ -91,8 +89,6 @@ func TestBlockPeer(t *testing.T) { } } - - func TestAcceptNonExistentGroup(t *testing.T) { sarah := GenerateNewProfile("Sarah") sarah.AcceptInvite("doesnotexist") @@ -104,7 +100,7 @@ func TestRejectGroupInvite(t *testing.T) { sarah.AddContact(alice.Onion, &alice.PublicProfile) alice.AddContact(sarah.Onion, &sarah.PublicProfile) - gid, invite, _ := alice.StartGroup("aaa.onion") + gid, invite, _ := alice.StartGroup("2c3kmoobnyghj2zw6pwv7d57yzld753auo3ugauezzpvfak3ahc4bdyd") gci := &protocol.CwtchPeerPacket{} proto.Unmarshal(invite, gci) sarah.ProcessInvite(gci.GetGroupChatInvite(), alice.Onion) @@ -128,7 +124,7 @@ func TestProfileGroup(t *testing.T) { sarah.AddContact(alice.Onion, &alice.PublicProfile) alice.AddContact(sarah.Onion, &sarah.PublicProfile) - gid, invite, _ := alice.StartGroupWithMessage("aaa.onion", []byte("Hello World")) + gid, invite, _ := alice.StartGroupWithMessage("2c3kmoobnyghj2zw6pwv7d57yzld753auo3ugauezzpvfak3ahc4bdyd", []byte("Hello World")) gci := &protocol.CwtchPeerPacket{} proto.Unmarshal(invite, gci) sarah.ProcessInvite(gci.GetGroupChatInvite(), alice.Onion) @@ -141,7 +137,7 @@ func TestProfileGroup(t *testing.T) { c, s1, _ := sarah.EncryptMessageToGroup("Hello World", group.GroupID) alice.AttemptDecryption(c, s1) - gid2, invite2, _ := alice.StartGroup("bbb.onion") + gid2, invite2, _ := alice.StartGroup("2c3kmoobnyghj2zw6pwv7d57yzld753auo3ugauezzpvfak3ahc4bdyd") gci2 := &protocol.CwtchPeerPacket{} proto.Unmarshal(invite2, gci2) sarah.ProcessInvite(gci2.GetGroupChatInvite(), alice.Onion) diff --git a/peer/connections/peerpeerconnection_test.go b/peer/connections/peerpeerconnection_test.go index a9d5827..4901a5e 100644 --- a/peer/connections/peerpeerconnection_test.go +++ b/peer/connections/peerpeerconnection_test.go @@ -77,7 +77,7 @@ func TestPeerPeerConnection(t *testing.T) { if state != AUTHENTICATED { t.Errorf("connection state should be authenticated(3), was instead %v", state) } - _, invite, _ := profile.StartGroup("aaa.onion") + _, invite, _ := profile.StartGroup("2c3kmoobnyghj2zw6pwv7d57yzld753auo3ugauezzpvfak3ahc4bdyd") ppc.SendGroupInvite(invite) time.Sleep(time.Second * 3) if tp.ReceivedGroupInvite == false { diff --git a/peer/cwtch_peer_test.go b/peer/cwtch_peer_test.go index 84325fb..b71a9bd 100644 --- a/peer/cwtch_peer_test.go +++ b/peer/cwtch_peer_test.go @@ -20,7 +20,7 @@ func TestCwtchPeerGenerate(t *testing.T) { } func TestTrustPeer(t *testing.T) { - groupName := "test.server" + groupName := "2c3kmoobnyghj2zw6pwv7d57yzld753auo3ugauezzpvfak3ahc4bdyd" alice := NewCwtchPeer("alice") alice.Init(connectivity.LocalProvider()) bob := NewCwtchPeer("bob") From ff54059111d828af34087145fbec39d9ed258ca8 Mon Sep 17 00:00:00 2001 From: Dan Ballard Date: Wed, 21 Nov 2018 15:10:02 -0800 Subject: [PATCH 3/7] update name --- app/app.go | 4 ++-- peer/connections/connectionsmanager.go | 4 ++-- peer/connections/peerpeerconnection.go | 4 ++-- peer/connections/peerserverconnection.go | 4 ++-- peer/cwtch_peer.go | 6 +++--- server/server.go | 2 +- 6 files changed, 12 insertions(+), 12 deletions(-) diff --git a/app/app.go b/app/app.go index 2a0ced1..381efb0 100644 --- a/app/app.go +++ b/app/app.go @@ -18,7 +18,7 @@ import ( type application struct { peers map[string]peer.CwtchPeer - mn connectivity.Mixnet + mn connectivity.ACN directory string mutex sync.Mutex primaryonion string @@ -41,7 +41,7 @@ type Application interface { } // NewApp creates a new app with some environment awareness and initializes a Tor Manager -func NewApp(mn connectivity.Mixnet, appDirectory string) Application { +func NewApp(mn connectivity.ACN, appDirectory string) Application { log.Printf("NewApp(%v)\n", appDirectory) app := &application{peers: make(map[string]peer.CwtchPeer), storage: make(map[string]storage.ProfileStore), directory: appDirectory, mn: mn} os.Mkdir(path.Join(app.directory, "profiles"), 0700) diff --git a/peer/connections/connectionsmanager.go b/peer/connections/connectionsmanager.go index d1a8de6..8380fed 100644 --- a/peer/connections/connectionsmanager.go +++ b/peer/connections/connectionsmanager.go @@ -15,11 +15,11 @@ type Manager struct { serverConnections map[string]*PeerServerConnection lock sync.Mutex breakChannel chan bool - mn connectivity.Mixnet + mn connectivity.ACN } // NewConnectionsManager creates a new instance of Manager. -func NewConnectionsManager(mn connectivity.Mixnet) *Manager { +func NewConnectionsManager(mn connectivity.ACN) *Manager { m := new(Manager) m.mn = mn m.peerConnections = make(map[string]*PeerPeerConnection) diff --git a/peer/connections/peerpeerconnection.go b/peer/connections/peerpeerconnection.go index 384fb22..7db1cea 100644 --- a/peer/connections/peerpeerconnection.go +++ b/peer/connections/peerpeerconnection.go @@ -23,11 +23,11 @@ type PeerPeerConnection struct { profile *model.Profile dataHandler func(string, []byte) []byte aif application.ApplicationInstanceFactory - mn connectivity.Mixnet + mn connectivity.ACN } // NewPeerPeerConnection creates a new peer connection for the given hostname and profile. -func NewPeerPeerConnection(mn connectivity.Mixnet, peerhostname string, profile *model.Profile, dataHandler func(string, []byte) []byte, aif application.ApplicationInstanceFactory) *PeerPeerConnection { +func NewPeerPeerConnection(mn connectivity.ACN, peerhostname string, profile *model.Profile, dataHandler func(string, []byte) []byte, aif application.ApplicationInstanceFactory) *PeerPeerConnection { ppc := new(PeerPeerConnection) ppc.mn = mn ppc.PeerHostname = peerhostname diff --git a/peer/connections/peerserverconnection.go b/peer/connections/peerserverconnection.go index 0af48d3..baf65d4 100644 --- a/peer/connections/peerserverconnection.go +++ b/peer/connections/peerserverconnection.go @@ -23,13 +23,13 @@ type PeerServerConnection struct { Server string state ConnectionState connection *connection.Connection - mn connectivity.Mixnet + mn connectivity.ACN GroupMessageHandler func(string, *protocol.GroupMessage) } // NewPeerServerConnection creates a new Peer->Server outbound connection -func NewPeerServerConnection(mn connectivity.Mixnet, serverhostname string) *PeerServerConnection { +func NewPeerServerConnection(mn connectivity.ACN, serverhostname string) *PeerServerConnection { psc := new(PeerServerConnection) psc.mn = mn psc.Server = serverhostname diff --git a/peer/cwtch_peer.go b/peer/cwtch_peer.go index 59ac9a1..7af5f29 100644 --- a/peer/cwtch_peer.go +++ b/peer/cwtch_peer.go @@ -28,7 +28,7 @@ type cwtchPeer struct { connection.AutoConnectionHandler Profile *model.Profile app *application.RicochetApplication - mn connectivity.Mixnet + mn connectivity.ACN mutex sync.Mutex connectionsManager *connections.Manager dataHandler func(string, []byte) []byte @@ -40,7 +40,7 @@ type cwtchPeer struct { // CwtchPeer provides us with a way of testing systems built on top of cwtch without having to // directly implement a cwtchPeer. type CwtchPeer interface { - Init(connectivity.Mixnet) + Init(connectivity.ACN) PeerWithOnion(string) *connections.PeerPeerConnection InviteOnionToGroup(string, string) error @@ -90,7 +90,7 @@ func FromProfile(profile *model.Profile) CwtchPeer { } // Init instantiates a cwtchPeer -func (cp *cwtchPeer) Init(mn connectivity.Mixnet) { +func (cp *cwtchPeer) Init(mn connectivity.ACN) { cp.mn = mn cp.connectionsManager = connections.NewConnectionsManager(cp.mn) go cp.connectionsManager.AttemptReconnections() diff --git a/server/server.go b/server/server.go index 9f476c4..16f48cb 100644 --- a/server/server.go +++ b/server/server.go @@ -23,7 +23,7 @@ type Server struct { // TODO: surface errors // TODO: handle HUP/KILL signals to exit and close Tor gracefully // TODO: handle user input to exit -func (s *Server) Run(mn connectivity.Mixnet, serverConfig Config) { +func (s *Server) Run(mn connectivity.ACN, serverConfig Config) { s.config = serverConfig cwtchserver := new(application.RicochetApplication) s.metricsPack.Start(cwtchserver, serverConfig.ConfigDir, s.config.ServerReporting.LogMetricsToFile) From ddbf96e668c3aafd454347601adf485a992f9617 Mon Sep 17 00:00:00 2001 From: erinn Date: Wed, 21 Nov 2018 16:08:47 -0800 Subject: [PATCH 4/7] little fixes --- app/app.go | 17 +++++++++++++++++ peer/cwtch_peer.go | 14 +++++++++++--- storage/file_profile_store.go | 1 + 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/app/app.go b/app/app.go index 381efb0..831ff1a 100644 --- a/app/app.go +++ b/app/app.go @@ -34,6 +34,7 @@ type Application interface { PrimaryIdentity() peer.CwtchPeer GetPeer(onion string) peer.CwtchPeer ListPeers() map[string]string + LaunchPeers() //GetTorStatus() (map[string]string, error) @@ -108,8 +109,10 @@ func (app *application) LoadProfiles(password string) error { log.Printf("Error: profile for onion %v already exists", p.GetProfile().Onion) continue } + p.Init(app.mn) p.Listen() + app.mutex.Lock() app.peers[p.GetProfile().Onion] = p app.storage[p.GetProfile().Onion] = fileStore @@ -121,6 +124,20 @@ func (app *application) LoadProfiles(password string) error { return nil } +func (app *application) LaunchPeers() { + for _, p := range app.peers { + if !p.IsStarted() { + app.startPeer(p) + } + } +} + +func (app *application) startPeer(peer peer.CwtchPeer) { + go func() { + peer.Listen() + }() +} + // ListPeers returns a map of onions to their profile's Name func (app *application) ListPeers() map[string]string { keys := map[string]string{} diff --git a/peer/cwtch_peer.go b/peer/cwtch_peer.go index 7af5f29..783ac44 100644 --- a/peer/cwtch_peer.go +++ b/peer/cwtch_peer.go @@ -32,9 +32,9 @@ type cwtchPeer struct { mutex sync.Mutex connectionsManager *connections.Manager dataHandler func(string, []byte) []byte - //handlers map[string]func(*application.ApplicationInstance) func() channels.Handler - aif application.ApplicationInstanceFactory - shutdown bool + shutdown bool + aif application.ApplicationInstanceFactory + started bool } // CwtchPeer provides us with a way of testing systems built on top of cwtch without having to @@ -71,6 +71,7 @@ type CwtchPeer interface { SetPeerDataHandler(func(string, []byte) []byte) Listen() + IsStarted() bool Shutdown() } @@ -353,7 +354,9 @@ func (cp *cwtchPeer) listenFn() error { ra.Init(cp.mn, cp.Profile.Name, identity.InitializeV3(cp.Profile.Name, &cp.Profile.Ed25519PrivateKey, &cp.Profile.Ed25519PublicKey), af, cp) log.Printf("Running cwtch peer on %v", onionService.AddressFull()) cp.app = ra + cp.started = true ra.Run(onionService) + return nil } @@ -366,6 +369,11 @@ func (cp *cwtchPeer) Shutdown() { } } +// IsStarted returns true if Listen() has successfully been run before on this connection (ever). TODO: we will need to properly unset this flag on error if we want to support resumption in the future +func (cp *cwtchPeer) IsStarted() bool { + return cp.started +} + // CwtchPeerInstance encapsulates incoming peer connections type CwtchPeerInstance struct { rai *application.ApplicationInstance diff --git a/storage/file_profile_store.go b/storage/file_profile_store.go index 10e1433..9dc3d7b 100644 --- a/storage/file_profile_store.go +++ b/storage/file_profile_store.go @@ -103,6 +103,7 @@ func (fps *fileProfileStore) Load() (peer.CwtchPeer, error) { if err == nil { return peer.FromProfile(profile), nil } + return nil, err } return nil, err } From 4814e0b4096348bbfff20dfa0bc1ea25f920669a Mon Sep 17 00:00:00 2001 From: Dan Ballard Date: Thu, 22 Nov 2018 10:08:21 -0800 Subject: [PATCH 5/7] undo use of personal fork, pr got merged --- server/metrics/monitors.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/metrics/monitors.go b/server/metrics/monitors.go index aa4b197..8c8c432 100644 --- a/server/metrics/monitors.go +++ b/server/metrics/monitors.go @@ -4,7 +4,7 @@ import ( "bufio" "fmt" "git.openprivacy.ca/openprivacy/libricochet-go/application" - "github.com/dballard/pidusage" + "github.com/struCoder/pidusage" "log" "os" "path" From 8a0d9c54fd505c027af8003396199a7f64440c65 Mon Sep 17 00:00:00 2001 From: Dan Ballard Date: Thu, 22 Nov 2018 10:01:04 -0800 Subject: [PATCH 6/7] local var rename --- app/app.go | 10 +++++----- app/cli/main.go | 6 +++--- peer/connections/connectionsmanager.go | 10 +++++----- peer/connections/peerpeerconnection.go | 8 ++++---- peer/connections/peerserverconnection.go | 8 ++++---- peer/cwtch_peer.go | 12 ++++++------ server/app/main.go | 6 +++--- server/server.go | 6 +++--- testing/cwtch_peer_server_intergration_test.go | 10 +++++----- 9 files changed, 38 insertions(+), 38 deletions(-) diff --git a/app/app.go b/app/app.go index 831ff1a..7a7486e 100644 --- a/app/app.go +++ b/app/app.go @@ -18,7 +18,7 @@ import ( type application struct { peers map[string]peer.CwtchPeer - mn connectivity.ACN + acn connectivity.ACN directory string mutex sync.Mutex primaryonion string @@ -42,9 +42,9 @@ type Application interface { } // NewApp creates a new app with some environment awareness and initializes a Tor Manager -func NewApp(mn connectivity.ACN, appDirectory string) Application { +func NewApp(acn connectivity.ACN, appDirectory string) Application { log.Printf("NewApp(%v)\n", appDirectory) - app := &application{peers: make(map[string]peer.CwtchPeer), storage: make(map[string]storage.ProfileStore), directory: appDirectory, mn: mn} + app := &application{peers: make(map[string]peer.CwtchPeer), storage: make(map[string]storage.ProfileStore), directory: appDirectory, acn: acn} os.Mkdir(path.Join(app.directory, "profiles"), 0700) return app } @@ -73,7 +73,7 @@ func (app *application) CreatePeer(name string, password string) (peer.CwtchPeer if err != nil { return nil, err } - p.Init(app.mn) + p.Init(app.acn) p.Listen() _, exists := app.peers[p.GetProfile().Onion] if exists { @@ -110,7 +110,7 @@ func (app *application) LoadProfiles(password string) error { continue } - p.Init(app.mn) + p.Init(app.acn) p.Listen() app.mutex.Lock() diff --git a/app/cli/main.go b/app/cli/main.go index 8b693c0..ecd8bd6 100644 --- a/app/cli/main.go +++ b/app/cli/main.go @@ -255,11 +255,11 @@ func main() { log.Fatalf("\nError: could not load current user: %v\n", err) } - mn, err := connectivity.StartTor(path.Join(usr.HomeDir, ".cwtch"), "") + acn, err := connectivity.StartTor(path.Join(usr.HomeDir, ".cwtch"), "") if err != nil { log.Fatalf("\nError connecting to Tor: %v\n", err) } - app = app2.NewApp(mn, path.Join(usr.HomeDir, ".cwtch")) + app = app2.NewApp(acn, path.Join(usr.HomeDir, ".cwtch")) if err != nil { log.Fatalf("Error initializing application: %v", err) } @@ -602,6 +602,6 @@ func main() { } app.Shutdown() - mn.Close() + acn.Close() os.Exit(0) } diff --git a/peer/connections/connectionsmanager.go b/peer/connections/connectionsmanager.go index 8380fed..08b1462 100644 --- a/peer/connections/connectionsmanager.go +++ b/peer/connections/connectionsmanager.go @@ -15,13 +15,13 @@ type Manager struct { serverConnections map[string]*PeerServerConnection lock sync.Mutex breakChannel chan bool - mn connectivity.ACN + acn connectivity.ACN } // NewConnectionsManager creates a new instance of Manager. -func NewConnectionsManager(mn connectivity.ACN) *Manager { +func NewConnectionsManager(acn connectivity.ACN) *Manager { m := new(Manager) - m.mn = mn + m.acn = acn m.peerConnections = make(map[string]*PeerPeerConnection) m.serverConnections = make(map[string]*PeerServerConnection) m.breakChannel = make(chan bool) @@ -35,7 +35,7 @@ func (m *Manager) ManagePeerConnection(host string, profile *model.Profile, data _, exists := m.peerConnections[host] if !exists { - ppc := NewPeerPeerConnection(m.mn, host, profile, dataHandler, aif) + ppc := NewPeerPeerConnection(m.acn, host, profile, dataHandler, aif) go ppc.Run() m.peerConnections[host] = ppc return ppc @@ -49,7 +49,7 @@ func (m *Manager) ManageServerConnection(host string, handler func(string, *prot _, exists := m.serverConnections[host] if !exists { - psc := NewPeerServerConnection(m.mn, host) + psc := NewPeerServerConnection(m.acn, host) go psc.Run() psc.GroupMessageHandler = handler m.serverConnections[host] = psc diff --git a/peer/connections/peerpeerconnection.go b/peer/connections/peerpeerconnection.go index 7db1cea..6e34133 100644 --- a/peer/connections/peerpeerconnection.go +++ b/peer/connections/peerpeerconnection.go @@ -23,13 +23,13 @@ type PeerPeerConnection struct { profile *model.Profile dataHandler func(string, []byte) []byte aif application.ApplicationInstanceFactory - mn connectivity.ACN + acn connectivity.ACN } // NewPeerPeerConnection creates a new peer connection for the given hostname and profile. -func NewPeerPeerConnection(mn connectivity.ACN, peerhostname string, profile *model.Profile, dataHandler func(string, []byte) []byte, aif application.ApplicationInstanceFactory) *PeerPeerConnection { +func NewPeerPeerConnection(acn connectivity.ACN, peerhostname string, profile *model.Profile, dataHandler func(string, []byte) []byte, aif application.ApplicationInstanceFactory) *PeerPeerConnection { ppc := new(PeerPeerConnection) - ppc.mn = mn + ppc.acn = acn ppc.PeerHostname = peerhostname ppc.profile = profile ppc.dataHandler = dataHandler @@ -110,7 +110,7 @@ func (ppc *PeerPeerConnection) WaitTilAuthenticated() { // Run manages the setup and teardown of a peer->peer connection func (ppc *PeerPeerConnection) Run() error { ppc.state = CONNECTING - rc, err := goricochet.Open(ppc.mn, ppc.PeerHostname) + rc, err := goricochet.Open(ppc.acn, ppc.PeerHostname) if err == nil { rc.TraceLog(false) ppc.connection = rc diff --git a/peer/connections/peerserverconnection.go b/peer/connections/peerserverconnection.go index baf65d4..eef8f98 100644 --- a/peer/connections/peerserverconnection.go +++ b/peer/connections/peerserverconnection.go @@ -23,15 +23,15 @@ type PeerServerConnection struct { Server string state ConnectionState connection *connection.Connection - mn connectivity.ACN + acn connectivity.ACN GroupMessageHandler func(string, *protocol.GroupMessage) } // NewPeerServerConnection creates a new Peer->Server outbound connection -func NewPeerServerConnection(mn connectivity.ACN, serverhostname string) *PeerServerConnection { +func NewPeerServerConnection(acn connectivity.ACN, serverhostname string) *PeerServerConnection { psc := new(PeerServerConnection) - psc.mn = mn + psc.acn = acn psc.Server = serverhostname psc.Init() return psc @@ -55,7 +55,7 @@ func (psc *PeerServerConnection) WaitTilAuthenticated() { // Run manages the setup and teardown of a peer server connection func (psc *PeerServerConnection) Run() error { log.Printf("Connecting to %v", psc.Server) - rc, err := goricochet.Open(psc.mn, psc.Server) + rc, err := goricochet.Open(psc.acn, psc.Server) if err == nil { rc.TraceLog(true) psc.connection = rc diff --git a/peer/cwtch_peer.go b/peer/cwtch_peer.go index 783ac44..4bd89d2 100644 --- a/peer/cwtch_peer.go +++ b/peer/cwtch_peer.go @@ -28,7 +28,7 @@ type cwtchPeer struct { connection.AutoConnectionHandler Profile *model.Profile app *application.RicochetApplication - mn connectivity.ACN + acn connectivity.ACN mutex sync.Mutex connectionsManager *connections.Manager dataHandler func(string, []byte) []byte @@ -91,9 +91,9 @@ func FromProfile(profile *model.Profile) CwtchPeer { } // Init instantiates a cwtchPeer -func (cp *cwtchPeer) Init(mn connectivity.ACN) { - cp.mn = mn - cp.connectionsManager = connections.NewConnectionsManager(cp.mn) +func (cp *cwtchPeer) Init(acn connectivity.ACN) { + cp.acn = acn + cp.connectionsManager = connections.NewConnectionsManager(cp.acn) go cp.connectionsManager.AttemptReconnections() } @@ -317,7 +317,7 @@ func (cp *cwtchPeer) Listen() { // Listen sets up an onion listener to process incoming cwtch messages func (cp *cwtchPeer) listenFn() error { ra := new(application.RicochetApplication) - onionService, err := cp.mn.Listen(cp.Profile.Ed25519PrivateKey, application.RicochetPort) + onionService, err := cp.acn.Listen(cp.Profile.Ed25519PrivateKey, application.RicochetPort) if err != nil /*&& fmt.Sprintf("%v", err) != "550 Unspecified Tor error: Onion address collision"*/ { return err } @@ -351,7 +351,7 @@ func (cp *cwtchPeer) listenFn() error { af.AddHandler(handlers[i], cp.aif.GetHandler(handlers[i])) } - ra.Init(cp.mn, cp.Profile.Name, identity.InitializeV3(cp.Profile.Name, &cp.Profile.Ed25519PrivateKey, &cp.Profile.Ed25519PublicKey), af, cp) + ra.Init(cp.acn, cp.Profile.Name, identity.InitializeV3(cp.Profile.Name, &cp.Profile.Ed25519PrivateKey, &cp.Profile.Ed25519PublicKey), af, cp) log.Printf("Running cwtch peer on %v", onionService.AddressFull()) cp.app = ra cp.started = true diff --git a/server/app/main.go b/server/app/main.go index 71c1861..f20da29 100644 --- a/server/app/main.go +++ b/server/app/main.go @@ -17,16 +17,16 @@ func main() { serverConfig := cwtchserver.LoadConfig(configDir, serverConfigFile) - mn, err := connectivity.StartTor(path.Join(configDir, "tor"), "") + acn, err := connectivity.StartTor(path.Join(configDir, "tor"), "") if err != nil { log.Fatalf("\nError connecting to Tor: %v\n", err) } - defer mn.Close() + defer acn.Close() server := new(cwtchserver.Server) log.Printf("starting cwtch server...") // TODO load params from .cwtch/server.conf or command line flag // TODO: respond to HUP so t.Close is gracefully called - server.Run(mn, serverConfig) + server.Run(acn, serverConfig) } diff --git a/server/server.go b/server/server.go index 16f48cb..d1d6bfb 100644 --- a/server/server.go +++ b/server/server.go @@ -23,12 +23,12 @@ type Server struct { // TODO: surface errors // TODO: handle HUP/KILL signals to exit and close Tor gracefully // TODO: handle user input to exit -func (s *Server) Run(mn connectivity.ACN, serverConfig Config) { +func (s *Server) Run(acn connectivity.ACN, serverConfig Config) { s.config = serverConfig cwtchserver := new(application.RicochetApplication) s.metricsPack.Start(cwtchserver, serverConfig.ConfigDir, s.config.ServerReporting.LogMetricsToFile) - listenService, err := mn.Listen(s.config.PrivateKey, application.RicochetPort) + listenService, err := acn.Listen(s.config.PrivateKey, application.RicochetPort) if err != nil { log.Fatalf("error setting up onion service: %v", err) @@ -68,7 +68,7 @@ func (s *Server) Run(mn connectivity.ACN, serverConfig Config) { } }) - cwtchserver.Init(mn, "cwtch server for "+listenService.AddressIdentity(), s.config.Identity(), af, new(application.AcceptAllContactManager)) + cwtchserver.Init(acn, "cwtch server for "+listenService.AddressIdentity(), s.config.Identity(), af, new(application.AcceptAllContactManager)) log.Printf("cwtch server running on cwtch:%s", listenService.AddressFull()) s.app = cwtchserver s.app.Run(listenService) diff --git a/testing/cwtch_peer_server_intergration_test.go b/testing/cwtch_peer_server_intergration_test.go index 0e8bde2..9593f60 100644 --- a/testing/cwtch_peer_server_intergration_test.go +++ b/testing/cwtch_peer_server_intergration_test.go @@ -101,7 +101,7 @@ func TestCwtchPeerIntegration(t *testing.T) { log.SetOutput(ioutil.Discard) numGoRoutinesStart := runtime.NumGoroutine() - mn, err := connectivity.StartTor(".", "") + acn, err := connectivity.StartTor(".", "") if err != nil { t.Fatalf("Could not start Tor: %v", err) } @@ -121,7 +121,7 @@ func TestCwtchPeerIntegration(t *testing.T) { config := cwtchserver.LoadConfig(".", "server-test.json") identity := config.Identity() serverAddr = identity.Hostname() - go server.Run(mn, config) + go server.Run(acn, config) // let tor get established fmt.Printf("Establishing Tor hidden service: %v...\n", serverAddr) @@ -135,19 +135,19 @@ func TestCwtchPeerIntegration(t *testing.T) { fmt.Println("Creating Alice...") alice := peer.NewCwtchPeer("Alice") - alice.Init(mn) + alice.Init(acn) alice.Listen() fmt.Println("Alice created:", alice.GetProfile().Onion) fmt.Println("Creating Bob...") bob := peer.NewCwtchPeer("Bob") - bob.Init(mn) + bob.Init(acn) bob.Listen() fmt.Println("Bob created:", bob.GetProfile().Onion) fmt.Println("Creating Carol...") carol := peer.NewCwtchPeer("Carol") - carol.Init(mn) + carol.Init(acn) carol.Listen() fmt.Println("Carol created:", carol.GetProfile().Onion) From 6f6fa909468b282478e34b633c78f4ae3d36b064 Mon Sep 17 00:00:00 2001 From: Dan Ballard Date: Thu, 22 Nov 2018 14:29:05 -0800 Subject: [PATCH 7/7] add channels to tests using Listen in a thread to sync; add some cleanup to other tests; fix a potential NPE --- peer/connections/peerpeerconnection.go | 4 +++- peer/connections/peerpeerconnection_test.go | 7 +++++-- peer/connections/peerserverconnection_test.go | 7 +++++-- peer/cwtch_peer_test.go | 2 ++ 4 files changed, 15 insertions(+), 5 deletions(-) diff --git a/peer/connections/peerpeerconnection.go b/peer/connections/peerpeerconnection.go index 7db1cea..cd5a674 100644 --- a/peer/connections/peerpeerconnection.go +++ b/peer/connections/peerpeerconnection.go @@ -150,5 +150,7 @@ func (ppc *PeerPeerConnection) Run() error { // Close closes the connection func (ppc *PeerPeerConnection) Close() { ppc.state = KILLED - ppc.connection.Conn.Close() + if ppc.connection != nil { + ppc.connection.Conn.Close() + } } diff --git a/peer/connections/peerpeerconnection_test.go b/peer/connections/peerpeerconnection_test.go index 4901a5e..ec15a1f 100644 --- a/peer/connections/peerpeerconnection_test.go +++ b/peer/connections/peerpeerconnection_test.go @@ -22,8 +22,9 @@ func PeerAuthValid(hostname string, key ed25519.PublicKey) (allowed, known bool) return true, true } -func runtestpeer(t *testing.T, tp *TestPeer, identity identity.Identity) { +func runtestpeer(t *testing.T, tp *TestPeer, identity identity.Identity, listenChan chan bool) { ln, _ := net.Listen("tcp", "127.0.0.1:5452") + listenChan <- true conn, _ := ln.Accept() defer conn.Close() @@ -65,7 +66,9 @@ func TestPeerPeerConnection(t *testing.T) { tp := new(TestPeer) tp.Init() - go runtestpeer(t, tp, identity) + listenChan := make(chan bool) + go runtestpeer(t, tp, identity, listenChan) + <-listenChan state := ppc.GetState() if state != DISCONNECTED { fmt.Println("ERROR state should be disconnected") diff --git a/peer/connections/peerserverconnection_test.go b/peer/connections/peerserverconnection_test.go index 5015025..33f01b2 100644 --- a/peer/connections/peerserverconnection_test.go +++ b/peer/connections/peerserverconnection_test.go @@ -33,8 +33,9 @@ func (ts *TestServer) HandleFetchRequest() []*protocol.GroupMessage { return []*protocol.GroupMessage{{Ciphertext: []byte("hello"), Signature: []byte{}, Spamguard: []byte{}}, {Ciphertext: []byte("hello"), Signature: []byte{}, Spamguard: []byte{}}} } -func runtestserver(t *testing.T, ts *TestServer, identity identity.Identity) { +func runtestserver(t *testing.T, ts *TestServer, identity identity.Identity, listenChan chan bool) { ln, _ := net.Listen("tcp", "127.0.0.1:5451") + listenChan <- true conn, _ := ln.Accept() defer conn.Close() @@ -70,7 +71,9 @@ func TestPeerServerConnection(t *testing.T) { ts := new(TestServer) ts.Init() - go runtestserver(t, ts, identity) + listenChan := make(chan bool) + go runtestserver(t, ts, identity, listenChan) + <-listenChan onionAddr := identity.Hostname() psc := NewPeerServerConnection(connectivity.LocalProvider(), "127.0.0.1:5451|"+onionAddr) diff --git a/peer/cwtch_peer_test.go b/peer/cwtch_peer_test.go index b71a9bd..6f605c4 100644 --- a/peer/cwtch_peer_test.go +++ b/peer/cwtch_peer_test.go @@ -23,8 +23,10 @@ func TestTrustPeer(t *testing.T) { groupName := "2c3kmoobnyghj2zw6pwv7d57yzld753auo3ugauezzpvfak3ahc4bdyd" alice := NewCwtchPeer("alice") alice.Init(connectivity.LocalProvider()) + defer alice.Shutdown() bob := NewCwtchPeer("bob") bob.Init(connectivity.LocalProvider()) + defer bob.Shutdown() bobOnion := bob.GetProfile().Onion aliceOnion := alice.GetProfile().Onion