ricochet-go/core/contact.go

569 lines
16 KiB
Go

package core
import (
"fmt"
"github.com/ricochet-im/ricochet-go/core/utils"
"github.com/ricochet-im/ricochet-go/rpc"
protocol "github.com/s-rah/go-ricochet"
"golang.org/x/net/context"
"log"
"strconv"
"sync"
"time"
)
// XXX There is generally a lot of duplication and boilerplate between
// Contact, ConfigContact, and rpc.Contact. This should be reduced somehow.
// XXX Consider replacing the config contact with the protobuf structure,
// and extending the protobuf structure for everything it needs.
type Contact struct {
core *Ricochet
id int
data ConfigContact
status ricochet.Contact_Status
mutex sync.Mutex
events *utils.Publisher
connEnabled bool
connection *protocol.OpenConnection
connChannel chan *protocol.OpenConnection
connClosedChannel chan struct{}
connStopped chan struct{}
outboundConnAuthKnown bool
conversation *Conversation
}
func ContactFromConfig(core *Ricochet, id int, data ConfigContact, events *utils.Publisher) (*Contact, error) {
contact := &Contact{
core: core,
id: id,
data: data,
events: events,
}
if id < 0 {
return nil, fmt.Errorf("Invalid contact ID '%d'", id)
} else if !IsOnionValid(data.Hostname) {
return nil, fmt.Errorf("Invalid contact hostname '%s", data.Hostname)
}
if data.Request.Pending {
if data.Request.WhenRejected != "" {
contact.status = ricochet.Contact_REJECTED
} else {
contact.status = ricochet.Contact_REQUEST
}
}
return contact, nil
}
func (c *Contact) Id() int {
return c.id
}
func (c *Contact) Nickname() string {
c.mutex.Lock()
defer c.mutex.Unlock()
return c.data.Nickname
}
func (c *Contact) Address() string {
c.mutex.Lock()
defer c.mutex.Unlock()
address, _ := AddressFromOnion(c.data.Hostname)
return address
}
func (c *Contact) Hostname() string {
c.mutex.Lock()
defer c.mutex.Unlock()
return c.data.Hostname
}
func (c *Contact) LastConnected() time.Time {
c.mutex.Lock()
defer c.mutex.Unlock()
time, _ := time.Parse(time.RFC3339, c.data.LastConnected)
return time
}
func (c *Contact) WhenCreated() time.Time {
c.mutex.Lock()
defer c.mutex.Unlock()
time, _ := time.Parse(time.RFC3339, c.data.WhenCreated)
return time
}
func (c *Contact) Status() ricochet.Contact_Status {
c.mutex.Lock()
defer c.mutex.Unlock()
return c.status
}
func (c *Contact) Data() *ricochet.Contact {
c.mutex.Lock()
defer c.mutex.Unlock()
address, _ := AddressFromOnion(c.data.Hostname)
data := &ricochet.Contact{
Id: int32(c.id),
Address: address,
Nickname: c.data.Nickname,
WhenCreated: c.data.WhenCreated,
LastConnected: c.data.LastConnected,
Status: c.status,
}
if c.data.Request.Pending {
data.Request = &ricochet.ContactRequest{
Direction: ricochet.ContactRequest_OUTBOUND,
Address: data.Address,
Nickname: data.Nickname,
Text: c.data.Request.Message,
FromNickname: c.data.Request.MyNickname,
}
}
return data
}
func (c *Contact) Conversation() *Conversation {
c.mutex.Lock()
defer c.mutex.Unlock()
if c.conversation == nil {
address, _ := AddressFromOnion(c.data.Hostname)
entity := &ricochet.Entity{
ContactId: int32(c.id),
Address: address,
}
c.conversation = NewConversation(c, entity, c.core.Identity.ConversationStream)
}
return c.conversation
}
// XXX Thread safety disaster
func (c *Contact) Connection() *protocol.OpenConnection {
c.mutex.Lock()
defer c.mutex.Unlock()
return c.connection
}
func (c *Contact) StartConnection() {
c.mutex.Lock()
defer c.mutex.Unlock()
if c.connEnabled {
return
}
c.connEnabled = true
c.connChannel = make(chan *protocol.OpenConnection)
c.connClosedChannel = make(chan struct{})
c.connStopped = make(chan struct{})
go c.contactConnection()
}
func (c *Contact) StopConnection() {
c.mutex.Lock()
if !c.connEnabled {
c.mutex.Unlock()
return
}
stopped := c.connStopped
close(c.connChannel)
c.connChannel = nil
c.connClosedChannel = nil
c.connStopped = nil
c.mutex.Unlock()
<-stopped
}
func (c *Contact) shouldMakeOutboundConnections() bool {
c.mutex.Lock()
defer c.mutex.Unlock()
// Don't make connections to contacts in the REJECTED state
if c.status == ricochet.Contact_REJECTED {
return false
}
return c.connEnabled
}
// Goroutine to handle the protocol connection for a contact.
// Responsible for making outbound connections and taking over authenticated
// inbound connections, running protocol handlers on the active connection, and
// reacting to connection loss. Nothing else may write Contact.connection.
func (c *Contact) contactConnection() {
// XXX Should the protocol continue handling its own goroutines?
// I'm thinking the protocol API design I want is:
// - "handler" assigned per-connection
// - each inbound listener has its own handler also, assigns for conns
// - so, non-authed connection has a handler that _can't_ do anything other than auth
// - inbound contact req connection gets a special handler for that case
// - authenticated contact conns get handler changed here
// It's probably more sensible to never break the conn read loop, because of buffering etc
// So likely we don't want to change that goroutine. Could still use a channel to pass it
// to handler for parsing, which could let it go on any goroutine we want, if it's desirable
// to put it on e.g. this routine. Hmm.
connChannel := c.connChannel
connClosedChannel := c.connClosedChannel
stopped := c.connStopped
connectionLoop:
for {
// If there is no active connection, spawn an outbound connector.
// A successful connection is returned via connChannel; otherwise, it will keep trying.
var outboundCtx context.Context
outboundCancel := func() {}
if c.connection == nil && c.shouldMakeOutboundConnections() {
outboundCtx, outboundCancel = context.WithCancel(context.Background())
go c.connectOutbound(outboundCtx, connChannel)
}
select {
case conn, ok := <-connChannel:
outboundCancel()
if !ok {
// Closing connChannel exits this connection routine, for contact
// deletion, exit, or some other case.
break connectionLoop
} else if conn == nil {
// Signal used to restart outbound connection attempts
continue
}
// Received a new connection. The connection is already authenticated and ready to use.
// This connection may be inbound or outbound. setConnection will decide whether to
// replace an existing connection.
// XXX Tweak setConnection logic if needed
// Logic when keeping connection, need to make sure protocol is going, etc...
if err := c.setConnection(conn); err != nil {
log.Printf("Discarded new contact %s connection: %s", c.Address(), err)
if !conn.Closed && conn != c.connection {
conn.Close()
}
continue
}
case <-connClosedChannel:
outboundCancel()
c.clearConnection(nil)
}
}
log.Printf("Exiting contact connection loop for %s", c.Address())
c.clearConnection(nil)
close(stopped)
}
// Attempt an outbound connection to the contact, retrying automatically using OnionConnector.
// This function _must_ send something to connChannel before returning, unless the context has
// been cancelled.
func (c *Contact) connectOutbound(ctx context.Context, connChannel chan *protocol.OpenConnection) {
c.mutex.Lock()
connector := OnionConnector{
Network: c.core.Network,
NeverGiveUp: true,
}
hostname := c.data.Hostname
c.mutex.Unlock()
for {
conn, err := connector.Connect(hostname+":9878", ctx)
if err != nil {
// The only failure here should be context, because NeverGiveUp
// is set, but be robust anyway.
if ctx.Err() != nil {
return
}
log.Printf("Contact connection failure: %s", err)
continue
}
log.Printf("Successful outbound connection to contact %s", hostname)
oc, err := protocol.Open(conn, hostname[0:16])
if err != nil {
log.Printf("Contact connection protocol failure: %s", err)
if oc != nil {
oc.Close()
}
if err := connector.Backoff(ctx); err != nil {
return
}
continue
} else {
log.Printf("Protocol connection open: %v", oc)
// XXX Protocol API needs to be reworked; see notes in
// contactConnection. Ideally we should authenticate here and
// pass the conneciton back, but for now this can do nothing:
// the connection will either succeed and come in via the
// protocol handler, or will be closed and signalled via
// OnConnectionClosed. Alternatively, it will break because this
// is fragile and dumb.
// XXX BUG: This means no backoff for authentication failure
handler := &ProtocolConnection{
Conn: oc,
Contact: c,
MyHostname: c.core.Identity.Address()[9:],
PrivateKey: c.core.Identity.PrivateKey(),
}
go oc.Process(handler)
return
}
}
}
func (c *Contact) setConnection(conn *protocol.OpenConnection) error {
if conn.Client {
log.Printf("Contact %s has a new outbound connection", c.Address())
} else {
log.Printf("Contact %s has a new inbound connection", c.Address())
}
c.mutex.Lock()
if conn == c.connection {
c.mutex.Unlock()
return fmt.Errorf("Duplicate assignment of connection %v to contact %v", conn, c)
}
if !conn.IsAuthed || conn.Closed {
c.mutex.Unlock()
conn.Close()
return fmt.Errorf("Connection %v is not in a valid state to assign to contact %v", conn, c)
}
if c.data.Hostname[0:16] != conn.OtherHostname {
c.mutex.Unlock()
conn.Close()
return fmt.Errorf("Connection hostname %s doesn't match contact hostname %s when assigning connection", conn.OtherHostname, c.data.Hostname[0:16])
}
if conn.Client && !c.outboundConnAuthKnown && !c.data.Request.Pending {
log.Printf("Outbound connection to contact says we are not a known contact for %v", c)
// XXX Should move to rejected status, stop attempting connections.
c.mutex.Unlock()
conn.Close()
return fmt.Errorf("Outbound connection says we are not a known contact")
}
if c.connection != nil {
if c.shouldReplaceConnection(conn) {
// XXX Signal state change for connection loss?
c.connection.Close()
c.connection = nil
} else {
c.mutex.Unlock()
conn.Close()
return fmt.Errorf("Using existing connection")
}
}
// If this connection is inbound and there's an outbound attempt, keep this
// connection and cancel outbound if we haven't sent authentication yet, or
// if the outbound connection will lose the fallback comparison above.
// XXX implement this
c.connection = conn
log.Printf("Assigned connection %v to contact %v", c.connection, c)
if c.data.Request.Pending {
if conn.Client && !c.outboundConnAuthKnown {
// Outbound connection for contact request; send request message
// XXX hardcoded channel ID
log.Printf("Sending outbound contact request to %v", c)
conn.SendContactRequest(5, c.data.Request.MyNickname, c.data.Request.Message)
} else {
// Inbound connection or outbound connection with a positive
// 'isKnownContact' response implicitly accepts the contact request
// and can continue as a contact
log.Printf("Contact request implicitly accepted by contact %v", c)
c.updateContactRequest("Accepted")
}
} else {
c.status = ricochet.Contact_ONLINE
}
// Update LastConnected time
config := c.core.Config.OpenWrite()
c.data.LastConnected = time.Now().Format(time.RFC3339)
config.Contacts[strconv.Itoa(c.id)] = c.data
config.Save()
c.mutex.Unlock()
event := ricochet.ContactEvent{
Type: ricochet.ContactEvent_UPDATE,
Subject: &ricochet.ContactEvent_Contact{
Contact: c.Data(),
},
}
c.events.Publish(event)
// Send any queued messages
sent := c.Conversation().SendQueuedMessages()
if sent > 0 {
log.Printf("Sent %d queued messages to contact", sent)
}
return nil
}
// Close and clear state related to the active contact connection.
// If ifConn is non-nil, the active connection is only cleared if
// it is the same as ifConn. Returns true if cleared.
func (c *Contact) clearConnection(ifConn *protocol.OpenConnection) bool {
c.mutex.Lock()
if c.connection == nil || (ifConn != nil && c.connection != ifConn) {
c.mutex.Unlock()
return false
}
conn := c.connection
c.connection = nil
c.status = ricochet.Contact_OFFLINE
// XXX eww, and also potentially deadlockable?
config := c.core.Config.OpenWrite()
c.data.LastConnected = time.Now().Format(time.RFC3339)
config.Contacts[strconv.Itoa(c.id)] = c.data
config.Save()
c.mutex.Unlock()
if conn != nil && !conn.Closed {
conn.Close()
}
event := ricochet.ContactEvent{
Type: ricochet.ContactEvent_UPDATE,
Subject: &ricochet.ContactEvent_Contact{
Contact: c.Data(),
},
}
c.events.Publish(event)
return true
}
// Decide whether to replace the existing connection with conn.
// Assumes mutex is held.
func (c *Contact) shouldReplaceConnection(conn *protocol.OpenConnection) bool {
if c.connection == nil {
return true
} else if c.connection.Closed {
log.Printf("Replacing dead connection %v for contact %v", c.connection, c)
return true
} else if c.connection.Client == conn.Client {
// If the existing connection is in the same direction, always use the new one
log.Printf("Replacing existing same-direction connection %v with new connection %v for contact %v", c.connection, conn, c)
return true
} else if false {
// If the existing connection is more than 30 seconds old, use the new one
// XXX implement this
} else if preferOutbound := conn.MyHostname < conn.OtherHostname; preferOutbound == conn.Client {
// Fall back to string comparison of hostnames for a stable resolution
// New connection wins
log.Printf("Replacing existing connection %v with new connection %v for contact %v according to fallback order", c.connection, conn, c)
return true
} else {
// Old connection wins fallback
log.Printf("Keeping existing connection %v instead of new connection %v for contact %v according to fallback order", c.connection, conn, c)
return false
}
return false
}
// Update the status of a contact request from a protocol event. Returns
// true if the contact request channel should remain open.
func (c *Contact) UpdateContactRequest(status string) bool {
c.mutex.Lock()
defer c.mutex.Unlock()
if !c.data.Request.Pending {
return false
}
re := c.updateContactRequest(status)
event := ricochet.ContactEvent{
Type: ricochet.ContactEvent_UPDATE,
Subject: &ricochet.ContactEvent_Contact{
Contact: c.Data(),
},
}
c.events.Publish(event)
return re
}
// Same as above, but assumes the mutex is already held and that the caller
// will send an UPDATE event
func (c *Contact) updateContactRequest(status string) bool {
config := c.core.Config.OpenWrite()
now := time.Now().Format(time.RFC3339)
// Whether to keep the channel open
var re bool
switch status {
case "Pending":
c.data.Request.WhenDelivered = now
re = true
case "Accepted":
c.data.Request = ConfigContactRequest{}
if c.connection != nil {
c.status = ricochet.Contact_ONLINE
} else {
c.status = ricochet.Contact_UNKNOWN
}
case "Rejected":
c.data.Request.WhenRejected = now
case "Error":
c.data.Request.WhenRejected = now
c.data.Request.RemoteError = "error occurred"
default:
log.Printf("Unknown contact request status '%s'", status)
}
config.Contacts[strconv.Itoa(c.id)] = c.data
config.Save()
return re
}
// XXX also will go away during protocol API rework
func (c *Contact) OnConnectionAuthenticated(conn *protocol.OpenConnection, knownContact bool) {
c.mutex.Lock()
if c.connChannel == nil {
log.Printf("Inbound connection from contact, but connections are not enabled for contact %v", c)
c.mutex.Unlock()
conn.Close()
}
// XXX this is ugly
if conn.Client {
c.outboundConnAuthKnown = knownContact
}
c.connChannel <- conn
c.mutex.Unlock()
}
// XXX rework connection close to have a proper notification instead of this "find contact" mess.
func (c *Contact) OnConnectionClosed(conn *protocol.OpenConnection) {
c.mutex.Lock()
if c.connection != conn || c.connClosedChannel == nil {
c.mutex.Unlock()
return
}
c.connClosedChannel <- struct{}{}
c.mutex.Unlock()
}