cwtch/peer/cwtch_peer.go

package peer

import (
	"cwtch.im/cwtch/model"
	"cwtch.im/cwtch/peer/connections"
	"cwtch.im/cwtch/peer/peer"
	"cwtch.im/cwtch/protocol"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/s-rah/go-ricochet/application"
	"github.com/s-rah/go-ricochet/channels"
	"github.com/s-rah/go-ricochet/connection"
	"io/ioutil"
	"log"
	"sync"
)

// CwtchPeer manages incoming and outgoing connections and all processing for a Cwtch Peer
type CwtchPeer struct {
	connection.AutoConnectionHandler
	Profile            *model.Profile
	mutex              sync.Mutex
	Log                chan string `json:"-"`
	connectionsManager *connections.Manager
	profilefile        string
}

func (cp *CwtchPeer) setup() {
	cp.Log = make(chan string)
	cp.connectionsManager = connections.NewConnectionsManager()
	cp.Init()

	go cp.connectionsManager.AttemptReconnections()

	for onion, profile := range cp.Profile.Contacts {
		if profile.Trusted && !profile.Blocked {
			cp.PeerWithOnion(onion)
		}
	}

	for _, group := range cp.Profile.Groups {
		if group.Accepted || group.Owner == "self" {
			cp.JoinServer(group.GroupServer)
		}
	}
}

// NewCwtchPeer creates and returns a new CwtchPeer with the given name.
func NewCwtchPeer(name string) *CwtchPeer {
	cp := new(CwtchPeer)
	cp.Profile = model.GenerateNewProfile(name)
	cp.setup()
	return cp
}

// Save saves the CwtchPeer profile state to a file.
func (cp *CwtchPeer) Save(profilefile string) error {
	cp.mutex.Lock()
	bytes, _ := json.Marshal(cp)
	err := ioutil.WriteFile(profilefile, bytes, 0600)
	cp.profilefile = profilefile
	cp.mutex.Unlock()
	return err
}

// LoadCwtchPeer loads an existing CwtchPeer from a file.
func LoadCwtchPeer(profilefile string) (*CwtchPeer, error) {
	bytes, _ := ioutil.ReadFile(profilefile)
	cp := new(CwtchPeer)
	err := json.Unmarshal(bytes, &cp)
	cp.setup()
	cp.profilefile = profilefile
	return cp, err
}

// PeerWithOnion is the entry point for CwtchPeer relationships
func (cp *CwtchPeer) PeerWithOnion(onion string) {
	cp.connectionsManager.ManagePeerConnection(onion, cp.Profile)
}

// InviteOnionToGroup kicks off the invite process
func (cp *CwtchPeer) InviteOnionToGroup(onion string, groupid string) error {

	group := cp.Profile.GetGroupByGroupID(groupid)
	if group != nil {
		fmt.Printf("Constructing invite for group: %v\n", group)
		invite, err := group.Invite()
		if err != nil {
			return err
		}
		ppc := cp.connectionsManager.GetPeerPeerConnectionForOnion(onion)
		if ppc == nil {
			return errors.New("peer connection not setup for onion. peers must be trusted before sending")
		}
		if ppc.GetState() == connections.AUTHENTICATED {
			fmt.Printf("Got connection for group: %v - Sending Invite\n", ppc)
			ppc.SendGroupInvite(invite)
		} else {
			return errors.New("cannot send invite to onion: peer connection is not ready")
		}
		return nil
	}
	return errors.New("group id could not be found")
}

// ReceiveGroupMessage is a callback function that processes GroupMessages from a given server
func (cp *CwtchPeer) ReceiveGroupMessage(server string, gm *protocol.GroupMessage) {
	cp.Profile.AttemptDecryption(gm.Ciphertext)
}

// JoinServer manages a new server connection with the given onion address
func (cp *CwtchPeer) JoinServer(onion string) {
	cp.connectionsManager.ManageServerConnection(onion, cp.ReceiveGroupMessage)
}

// SendMessageToGroup attemps to sent the given message to the given group id.
func (cp *CwtchPeer) SendMessageToGroup(groupid string, message string) error {
	group := cp.Profile.GetGroupByGroupID(groupid)
	if group == nil {
		return errors.New("group does not exit")
	}
	psc := cp.connectionsManager.GetPeerServerConnectionForOnion(group.GroupServer)
	if psc == nil {
		return errors.New("could not find server connection to send message to")
	}
	ct, err := cp.Profile.EncryptMessageToGroup(message, groupid)
	if err != nil {
		return err
	}
	gm := &protocol.GroupMessage{
		Ciphertext: ct,
	}
	err = psc.SendGroupMessage(gm)
	return err
}

// GetPeers returns a list of peer connections.
func (cp *CwtchPeer) GetPeers() map[string]connections.ConnectionState {
	return cp.connectionsManager.GetPeers()
}

// GetServers returns a list of server connections
func (cp *CwtchPeer) GetServers() map[string]connections.ConnectionState {
	return cp.connectionsManager.GetServers()
}

// TrustPeer sets an existing peer relationship to trusted
func (cp *CwtchPeer) TrustPeer(peer string) error {
	_, ok := cp.Profile.Contacts[peer]
	if !ok {
		return errors.New("peer does not exist")
	}
	cp.Profile.Contacts[peer].Trusted = true
	cp.PeerWithOnion(peer)
	return nil
}

// BlockPeer blocks an existing peer relationship.
func (cp *CwtchPeer) BlockPeer(peer string) error {
	_, ok := cp.Profile.Contacts[peer]
	if !ok {
		return errors.New("peer does not exist")
	}
	cp.Profile.Contacts[peer].Blocked = true
	return nil
}

// AcceptInvite accepts a given existing group invite
func (cp *CwtchPeer) AcceptInvite(groupID string) error {
	g := cp.Profile.GetGroupByGroupID(groupID)
	if g == nil {
		return errors.New("group invite does not exit")
	}
	g.Accepted = true
	return nil
}

// RejectInvite rejects a given group invite.
func (cp *CwtchPeer) RejectInvite(groupID string) error {
	g := cp.Profile.GetGroupByGroupID(groupID)
	if g == nil {
		return errors.New("group invite does not exit")
	}
	g.Accepted = false
	// TODO delete group from Profile
	return nil
}

// Listen sets up an onion listener to process incoming cwtch messages
func (cp *CwtchPeer) Listen() error {
	cwtchpeer := new(application.RicochetApplication)
	l, err := application.SetupOnion("127.0.0.1:9051", "tcp4", "", cp.Profile.OnionPrivateKey, 9878)

	if err != nil {
		return err
	}

	af := application.ApplicationInstanceFactory{}
	af.Init()
	af.AddHandler("im.cwtch.peer", func(rai *application.ApplicationInstance) func() channels.Handler {
		cpi := new(CwtchPeerInstance)
		cpi.Init(rai, cwtchpeer)
		return func() channels.Handler {
			cpc := new(peer.CwtchPeerChannel)
			cpc.Handler = &CwtchPeerHandler{Onion: rai.RemoteHostname, Peer: cp}
			return cpc
		}
	})

	cwtchpeer.Init(cp.Profile.Name, cp.Profile.OnionPrivateKey, af, new(application.AcceptAllContactManager))
	log.Printf("Running cwtch peer on %v", l.Addr().String())
	cwtchpeer.Run(l)
	return nil
}

// CwtchPeerInstance encapsulates incoming peer connections
type CwtchPeerInstance struct {
	rai *application.ApplicationInstance
	ra  *application.RicochetApplication
}

// Init sets up a CwtchPeerInstance
func (cpi *CwtchPeerInstance) Init(rai *application.ApplicationInstance, ra *application.RicochetApplication) {
	cpi.rai = rai
	cpi.ra = ra
}

// CwtchPeerHandler encapsulates handling of incoming CwtchPackets
type CwtchPeerHandler struct {
	Onion string
	Peer  *CwtchPeer
}

// ClientIdentity handles incoming ClientIdentity packets
func (cph *CwtchPeerHandler) ClientIdentity(ci *protocol.CwtchIdentity) {
	log.Printf("Received Client Identity from %v %v\n", cph.Onion, ci.String())
	cph.Peer.Profile.AddCwtchIdentity(cph.Onion, ci)
	cph.Peer.Save(cph.Peer.profilefile)
}

// HandleGroupInvite handles incoming GroupInvites
func (cph *CwtchPeerHandler) HandleGroupInvite(gci *protocol.GroupChatInvite) {
	log.Printf("Received GroupID from %v %v\n", cph.Onion, gci.String())
	cph.Peer.Profile.ProcessInvite(gci, cph.Onion)
}

// GetClientIdentityPacket returns our ClientIdentity packet so it can be sent to the connected peer.
func (cph *CwtchPeerHandler) GetClientIdentityPacket() []byte {
	return cph.Peer.Profile.GetCwtchIdentityPacket()
}