tapir/networks/tor/BaseOnionService.go

326 lignes
10 KiB
Go

package tor
import (
"context"
"crypto/rand"
"encoding/base64"
"errors"
"git.openprivacy.ca/cwtch.im/tapir"
"git.openprivacy.ca/cwtch.im/tapir/primitives"
"git.openprivacy.ca/openprivacy/connectivity"
"git.openprivacy.ca/openprivacy/log"
"golang.org/x/crypto/ed25519"
"sync"
"time"
)
// BaseOnionService is a concrete implementation of the service interface over Tor onion services.
type BaseOnionService struct {
connections sync.Map
acn connectivity.ACN
id *primitives.Identity
privateKey ed25519.PrivateKey
ls connectivity.ListenService
lock sync.Mutex
port int
shutdownChannel chan bool
}
// Metrics provides a report of useful information about the status of the service e.g. the number of active
// connections
func (s *BaseOnionService) Metrics() tapir.ServiceMetrics {
s.lock.Lock()
defer s.lock.Unlock()
count := 0
s.connections.Range(func(key, value interface{}) bool {
connection := value.(tapir.Connection)
if !connection.IsClosed() {
count++
}
return true
})
return tapir.ServiceMetrics{
ConnectionCount: count,
}
}
// Init initializes a BaseOnionService with a given private key and identity
// The private key is needed to initialize the Onion listen socket, ideally we could just pass an Identity in here.
func (s *BaseOnionService) Init(acn connectivity.ACN, sk ed25519.PrivateKey, id *primitives.Identity) {
// run add onion
// get listen context
s.acn = acn
s.id = id
s.privateKey = sk
s.port = 9878
// blocking so we can wait on shutdown for closing of this goroutine
s.shutdownChannel = make(chan bool)
go func() {
for s.waitOrTimeout() {
s.GarbageCollect()
}
log.Debugf("closing down garbage collection goroutine")
}()
}
func (s *BaseOnionService) waitOrTimeout() bool {
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
select {
case <-s.shutdownChannel:
return false
case <-ctx.Done():
return true
}
}
// SetPort configures the port that the service uses.
func (s *BaseOnionService) SetPort(port int) {
s.port = port
}
// WaitForCapabilityOrClose blocks until the connection has the given capability or the underlying connection is closed
// (through error or user action)
func (s *BaseOnionService) WaitForCapabilityOrClose(cid string, name tapir.Capability) (tapir.Connection, error) {
attempts := 0
for {
if attempts > 4 {
s.connections.Range(func(key, value interface{}) bool {
connection := value.(tapir.Connection)
if connection.Hostname() == cid {
if !connection.IsClosed() {
connection.Close()
s.connections.Delete(key)
}
}
return true
})
log.Debugf("WaitForCapabilityOrClose attempts exceeded for %v, all connections closed", cid)
return nil, errors.New("failed to acquire capability after multiple attempts, forcibly closing all connections with the peer")
}
if attempts > 0 {
// Allow connections to be torn down / closed before checking again
// There is no point in busy looping...
time.Sleep(time.Second * time.Duration(attempts))
}
// Increment Attempts
attempts++
log.Debugf("Lookup up a connection %v...", cid)
// Lookup the connection...
conn, err := s.GetConnection(cid)
// If there are no active connections then return an error...
if conn == nil {
log.Debugf("no active connection found for %v", cid)
return nil, err
}
if err == nil {
// if we have only one connection and it has the desired capability then return the connection with
// no error...
if conn.HasCapability(name) {
return conn, nil
}
log.Debugf("Found 1 connections for %v, but it lacks the desired capability %v", cid, name)
continue
}
// If We have 2 connections for the same hostname...
if err != nil {
log.Debugf("found duplicate connections for %v <-> %v %v", s.id.Hostname(), cid, err)
inboundCount := 0
// By convention the lowest lexicographical hostname purges all their outbounds to the higher
// hostname
// Which should only leave a single connection remaining (as we dedupe on connect too)
// This does allow people to attempt to guarantee being the outbound connection but due to the bidirectional
// authentication this should never result in an advantage in the protocol.
// Close all outbound connections to connection
s.connections.Range(func(key, value interface{}) bool {
connection := value.(tapir.Connection)
if connection.Hostname() == cid {
if !connection.IsClosed() && connection.IsOutbound() && s.id.Hostname() < cid {
connection.Close()
s.connections.Delete(key)
}
if !connection.IsClosed() && !connection.IsOutbound() {
inboundCount++
}
}
return true
})
// If we have more than 1 inbound count then forcibly close all connections...
// This shouldn't happen honestly, but if it does then it can cause an infinite check here
if inboundCount > 1 {
s.connections.Range(func(key, value interface{}) bool {
connection := value.(tapir.Connection)
if connection.Hostname() == cid {
if !connection.IsClosed() {
connection.Close()
s.connections.Delete(key)
}
}
return true
})
return nil, errors.New("multiple inbound connections found and closed; the only resolution to this is to close them all and try connecting again")
}
}
}
}
// GetConnection returns a connection for a given hostname.
func (s *BaseOnionService) GetConnection(hostname string) (tapir.Connection, error) {
conn := make([]tapir.Connection, 0)
s.connections.Range(func(key, value interface{}) bool {
connection := value.(tapir.Connection)
if connection.Hostname() == hostname {
if !connection.IsClosed() {
conn = append(conn, connection)
} else {
// Delete this Closed Connection
s.connections.Delete(key)
}
}
return true
})
if len(conn) == 0 {
return nil, errors.New("no connection found")
}
if len(conn) > 1 {
// If there are multiple connections we return the first one but also notify the user (this should be a
// temporary edgecase (see Connect))
return conn[0], errors.New("multiple connections found")
}
return conn[0], nil
}
// GarbageCollect iterates through the connection pool and cleans up any connections that are closed
// that haven't been removed from the map.
func (s *BaseOnionService) GarbageCollect() {
log.Debugf("running garbage collection...")
s.connections.Range(func(key, value interface{}) bool {
connection := value.(tapir.Connection)
if connection.IsClosed() {
// Delete this Closed Connection
s.connections.Delete(key)
}
return true
})
}
// Connect initializes a new outbound connection to the given peer, using the defined Application
func (s *BaseOnionService) Connect(hostname string, app tapir.Application) (bool, error) {
currconn, _ := s.GetConnection(hostname)
// We already have a connection
if currconn != nil {
// Note: This check is not 100% reliable. And we may end up with two connections between peers
// This can happen when a client connects to a server as the server is connecting to the client
// Because at the start of the connection the server cannot derive the true hostname of the client until it
// has auth'd
// We mitigate this by performing multiple checks when Connect'ing
return true, errors.New("already connected to " + hostname)
}
// connects to a remote server
// spins off to a connection struct
log.Debugf("Connecting to %v", hostname)
conn, _, err := s.acn.Open(hostname)
if err == nil {
connectionID := s.getNewConnectionID()
// Second check. If we didn't catch a double connection attempt before the Open we *should* catch it now because
// the auth protocol is quick and Open over onion connections can take some time.
// Again this isn't 100% reliable.
tconn, _ := s.GetConnection(hostname)
if tconn != nil {
conn.Close()
return true, errors.New("already connected to " + hostname)
}
log.Debugf("Connected to %v [%v]", hostname, connectionID)
s.connections.Store(connectionID, tapir.NewConnection(s, s.id, hostname, true, conn, app.NewInstance()))
return true, nil
}
log.Debugf("Error connecting to %v %v", hostname, err)
return false, err
}
func (s *BaseOnionService) getNewConnectionID() string {
id := make([]byte, 10)
rand.Read(id)
connectionID := "connection-" + base64.StdEncoding.EncodeToString(id)
return connectionID
}
// Listen starts a blocking routine that waits for incoming connections and initializes connections with them based
// on the given Application.
func (s *BaseOnionService) Listen(app tapir.Application) error {
// accepts a new connection
// spins off to a connection struct
s.lock.Lock()
ls, err := s.acn.Listen(s.privateKey, s.port)
s.ls = ls
s.lock.Unlock()
if err == nil {
log.Debugf("Starting a service on %v ", s.ls.AddressFull())
for {
conn, err := s.ls.Accept()
if err == nil {
tempHostname := s.getNewConnectionID()
log.Debugf("Accepted connection from %v", tempHostname)
s.connections.Store(tempHostname, tapir.NewConnection(s, s.id, tempHostname, false, conn, app.NewInstance()))
} else {
log.Debugf("Error accepting connection %v", err)
return err
}
}
}
log.Debugf("Error listening to connection %v", err)
return err
}
// Shutdown closes the service and ensures that any connections are closed.
func (s *BaseOnionService) Shutdown() {
s.lock.Lock()
defer s.lock.Unlock()
if s.ls != nil {
s.ls.Close()
}
// close all existing connections manually
s.connections.Range(func(key, value interface{}) bool {
connection := value.(tapir.Connection)
connection.Close()
return true
})
// wait for the return of our garbage collection goroutine
s.shutdownChannel <- true
}
// Broadcast sends a message to all connections who possess the given capability
func (s *BaseOnionService) Broadcast(message []byte, capability tapir.Capability) error {
s.lock.Lock()
defer s.lock.Unlock()
s.connections.Range(func(key, value interface{}) bool {
connection := value.(tapir.Connection)
if connection.HasCapability(capability) {
connection.Send(message)
}
return true
})
return nil
}