cwtch/event/bridge/pipeBridge.go

// +build !windows

package bridge

import (
	"cwtch.im/cwtch/event"
	"cwtch.im/cwtch/protocol/connections"
	"encoding/base64"
	"encoding/binary"
	"encoding/json"
	"git.openprivacy.ca/openprivacy/libricochet-go/log"
	"os"
	"sync"
	"syscall"
	"time"
)

/* pipeBridge creates a pair of named pipes
   Needs a call to new client and service to fully successfully open
*/

const maxBufferSize = 1000

const serviceName = "service"
const clientName = "client"

const syn = "SYN"
const synack = "SYNACK"
const ack = "ACK"

type pipeBridge struct {
	infile, outfile string
	in, out         *os.File
	read            chan event.IPCMessage
	write           *InfiniteChannel
	closedChan      chan bool
	state           connections.ConnectionState
	lock            sync.Mutex
	threeShake      func() bool

	// For logging / debugging purposes
	name string
}

func newPipeBridge(inFilename, outFilename string) *pipeBridge {
	syscall.Mkfifo(inFilename, 0600)
	syscall.Mkfifo(outFilename, 0600)
	pb := &pipeBridge{infile: inFilename, outfile: outFilename, state: connections.DISCONNECTED}
	pb.read = make(chan event.IPCMessage, maxBufferSize)
	pb.write = newInfiniteChannel() //make(chan event.IPCMessage, maxBufferSize)
	return pb
}

// NewPipeBridgeClient returns a pipe backed IPCBridge for a client
func NewPipeBridgeClient(inFilename, outFilename string) event.IPCBridge {
	log.Debugf("Making new PipeBridge Client...\n")
	pb := newPipeBridge(inFilename, outFilename)
	pb.name = clientName
	pb.threeShake = pb.threeShakeClient
	go pb.connectionManager()

	return pb
}

// NewPipeBridgeService returns a pipe backed IPCBridge for a service
func NewPipeBridgeService(inFilename, outFilename string) event.IPCBridge {
	log.Debugf("Making new PipeBridge Service...\n")
	pb := newPipeBridge(inFilename, outFilename)
	pb.name = serviceName
	pb.threeShake = pb.threeShakeService

	go pb.connectionManager()

	log.Debugf("Successfully created new PipeBridge Service!\n")
	return pb
}

func (pb *pipeBridge) setState(state connections.ConnectionState) {
	pb.lock.Lock()
	defer pb.lock.Unlock()

	pb.state = state
}

func (pb *pipeBridge) getState() connections.ConnectionState {
	pb.lock.Lock()
	defer pb.lock.Unlock()

	return pb.state
}

func (pb *pipeBridge) connectionManager() {
	for pb.getState() != connections.KILLED {
		log.Debugf("clientConnManager loop start init\n")
		pb.setState(connections.CONNECTING)

		var err error
		log.Debugf("%v open file infile\n", pb.name)
		pb.in, err = os.OpenFile(pb.infile, os.O_RDWR, 0600)
		if err != nil {
			pb.setState(connections.DISCONNECTED)
			continue
		}

		log.Debugf("%v open file outfile\n", pb.name)
		pb.out, err = os.OpenFile(pb.outfile, os.O_RDWR, 0600)
		if err != nil {
			pb.setState(connections.DISCONNECTED)
			continue
		}

		log.Debugf("Successfully connected PipeBridge %v!\n", pb.name)

		pb.handleConns()
	}
	log.Debugf("exiting %v ConnectionManager\n", pb.name)

}

// threeShake performs a 3way handshake sync up
func (pb *pipeBridge) threeShakeService() bool {
	synacked := false

	for {
		resp, err := pb.readString()
		if err != nil {
			return false
		}

		if string(resp) == syn {
			if !synacked {
				err = pb.writeString([]byte(synack))
				if err != nil {
					return false
				}
				synacked = true
			}
		} else if string(resp) == ack {
			return true
		}
	}
}

func (pb *pipeBridge) synLoop(stop chan bool) {
	delay := time.Duration(0)
	for {
		select {
		case <-time.After(delay):
			err := pb.writeString([]byte(syn))
			if err != nil {
				return
			}
			delay = time.Second
		case <-stop:
			return
		}
	}
}

func (pb *pipeBridge) threeShakeClient() bool {
	stop := make(chan bool)
	go pb.synLoop(stop)
	for {
		resp, err := pb.readString()
		if err != nil {
			return false
		}

		if string(resp) == synack {
			stop <- true
			err := pb.writeString([]byte(ack))
			if err != nil {
				return false
			}
			return true
		}
	}
}

func (pb *pipeBridge) handleConns() {

	if !pb.threeShake() {
		pb.setState(connections.FAILED)
		pb.closeReset()
		return
	}

	pb.setState(connections.AUTHENTICATED)

	pb.closedChan = make(chan bool, 5)

	log.Debugf("handleConns authed, %v 2xgo\n", pb.name)

	go pb.handleRead()
	go pb.handleWrite()

	<-pb.closedChan
	log.Debugf("handleConns <-closedChan (%v)\n", pb.name)
	if pb.getState() != connections.KILLED {
		pb.setState(connections.FAILED)
	}
	pb.closeReset()
	log.Debugf("handleConns done for %v, exit\n", pb.name)
}

func (pb *pipeBridge) closeReset() {
	pb.in.Close()
	pb.out.Close()
	close(pb.read)
	pb.write.Close()

	if pb.getState() != connections.KILLED {
		pb.read = make(chan event.IPCMessage, maxBufferSize)
		pb.write = newInfiniteChannel()
	}
}

func (pb *pipeBridge) handleWrite() {
	log.Debugf("handleWrite() %v\n", pb.name)
	defer log.Debugf("exiting handleWrite() %v\n", pb.name)

	for {
		select {
		case messageInf := <-pb.write.output:
			if messageInf == nil {
				pb.closedChan <- true
				return
			}
			message := messageInf.(event.IPCMessage)
			if message.Message.EventType == event.EncryptedGroupMessage || message.Message.EventType == event.SendMessageToGroup || message.Message.EventType == event.NewMessageFromGroup {
				log.Debugf("handleWrite <- message: %v %v ...\n", message.Dest, message.Message.EventType)
			} else {
				log.Debugf("handleWrite <- message: %v\n", message)
			}
			if pb.getState() == connections.AUTHENTICATED {
				encMessage := &event.IPCMessage{Dest: message.Dest, Message: event.Event{EventType: message.Message.EventType, EventID: message.Message.EventID, Data: make(map[event.Field]string)}}
				for k, v := range message.Message.Data {
					encMessage.Message.Data[k] = base64.StdEncoding.EncodeToString([]byte(v))
				}

				messageJSON, _ := json.Marshal(encMessage)
				err := pb.writeString(messageJSON)
				if err != nil {
					pb.closedChan <- true
					return
				}
			} else {
				return
			}
		}
	}
}

func (pb *pipeBridge) handleRead() {
	log.Debugf("handleRead() %v\n", pb.name)
	defer log.Debugf("exiting handleRead() %v", pb.name)

	for {
		log.Debugf("Waiting to handleRead()...\n")

		buffer, err := pb.readString()
		if err != nil {
			pb.closedChan <- true
			return
		}

		var message event.IPCMessage
		err = json.Unmarshal(buffer, &message)
		if err != nil {
			log.Errorf("Read error: '%v', value: '%v'", err, buffer)
			pb.closedChan <- true
			return // probably new connection trying to initialize
		}
		for k, v := range message.Message.Data {
			val, _ := base64.StdEncoding.DecodeString(v)
			message.Message.Data[k] = string(val)
		}
		if message.Message.EventType == event.EncryptedGroupMessage || message.Message.EventType == event.SendMessageToGroup || message.Message.EventType == event.NewMessageFromGroup {
			log.Debugf("handleRead read<-: %v %v ...\n", message.Dest, message.Message.EventType)
		} else {
			log.Debugf("handleRead read<-: %v\n", message)
		}
		pb.read <- message
		log.Debugf("handleRead wrote\n")
	}
}

func (pb *pipeBridge) Read() (*event.IPCMessage, bool) {
	log.Debugf("Read() %v...\n", pb.name)
	var ok = false
	var message event.IPCMessage
	for !ok && pb.getState() != connections.KILLED {
		message, ok = <-pb.read
		if message.Message.EventType == event.EncryptedGroupMessage || message.Message.EventType == event.SendMessageToGroup || message.Message.EventType == event.NewMessageFromGroup {
			log.Debugf("Read %v: %v %v ...\n", pb.name, message.Dest, message.Message.EventType)
		} else {
			log.Debugf("Read %v: %v\n", pb.name, message)
		}
	}
	return &message, pb.getState() != connections.KILLED
}

func (pb *pipeBridge) Write(message *event.IPCMessage) {
	if message.Message.EventType == event.EncryptedGroupMessage || message.Message.EventType == event.SendMessageToGroup || message.Message.EventType == event.NewMessageFromGroup {
		log.Debugf("Write %v: %v %v ...\n", pb.name, message.Dest, message.Message.EventType)
	} else {
		log.Debugf("Write %v: %v\n", pb.name, message)
	}
	pb.write.input <- *message
	log.Debugf("Wrote\n")
}

func (pb *pipeBridge) Shutdown() {
	log.Debugf("pb.Shutdown() for %v currently in state: %v\n", pb.name, connections.ConnectionStateName[pb.getState()])
	pb.state = connections.KILLED
	pb.closedChan <- true
	log.Debugf("Done Shutdown for %v\n", pb.name)
}

func (pb *pipeBridge) writeString(message []byte) error {
	size := make([]byte, 2)
	binary.LittleEndian.PutUint16(size, uint16(len(message)))
	pb.out.Write(size)

	for pos := 0; pos < len(message); {
		n, err := pb.out.Write(message[pos:])
		if err != nil {
			log.Errorf("Writing out on pipeBridge: %v\n", err)
			return err
		}
		pos += n
	}
	return nil
}

func (pb *pipeBridge) readString() ([]byte, error) {
	var n int
	size := make([]byte, 2)
	var err error

	n, err = pb.in.Read(size)
	if err != nil || n != 2 {
		log.Errorf("Could not read len int from stream: %v\n", err)
		return nil, err
	}

	n = int(binary.LittleEndian.Uint16(size))
	pos := 0
	buffer := make([]byte, n)
	for n > 0 {
		m, err := pb.in.Read(buffer[pos:])
		if err != nil {
			log.Errorf("Reading into buffer from pipe: %v\n", err)
			return nil, err
		}
		n -= m
		pos += m
	}
	return buffer, nil
}