AnonymousTopicBasedPubSubCommunicationforMicroblogging/leader/leader.go

package main

//#cgo CFLAGS: -fopenmp -O2
//#cgo LDFLAGS: -lcrypto -lm -fopenmp
//#include "../c/dpf.h"
//#include "../c/okv.h"
//#include "../c/dpf.c"
//#include "../c/okv.c"
import "C"

//ssssssssssssss

import (
	"crypto/rand"
	"crypto/rsa"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"crypto/x509/pkix"
	"encoding/pem"
	"fmt"
	"math"
	"math/big"
	"net"
	"strconv"
	"sync"
	"time"

	lib "2PPS/lib"
	"unsafe"

	"golang.org/x/crypto/nacl/box"
)

//this stores all neccessary information for each client
type clientKeys struct {
	roundsParticipating int
	PublicKey           *[32]byte
	SharedSecret        [32]byte
	PirQuery            [][]byte
}

var clientData = make(map[net.Addr]clientKeys)

var leaderPrivateKey *[32]byte

var leaderPublicKey *[32]byte

var followerPublicKey *[32]byte

// every roundsBeforeUpdate the client updates his pirQuery, the sharedSecrets are updated locally
const roundsBeforeUpdate int = 1

const follower string = "127.0.0.1:4443"

const maxNumberOfClients = 1000

//needs to be changed at leader/follower/client at the same time
const neededSubscriptions = 1

var topicList []byte
var topicAmount int
var archiveTopicAmount int

//works on my machine
var numThreads = 12

//has to be dividable by 32
var dataLength int = 32

//this needs to be adjusted peridodically
var dbWriteSize int = 2

//counts the number of rounds
var round int = 1

var startTime time.Time

//adjust this for follower aswell
var maxTimePerRound time.Duration = 5 * time.Second

//channel for goroutine communication with clients
var phase1Channel = make(chan net.Conn, maxNumberOfClients)
var phase3Channel = make(chan net.Conn, maxNumberOfClients)

//variable for calculating the dbWrite size
var publisherNumber float64

func main() {

	generatedPublicKey, generatedPrivateKey, err := box.GenerateKey(rand.Reader)
	if err != nil {
		panic(err)
	}
	//why is this neccessary?
	leaderPrivateKey = generatedPrivateKey
	leaderPublicKey = generatedPublicKey

	/*
		if len(os.Args) != 4 {
			fmt.Println("try again with: numThreads, dataLength, numRows")
			return
		}
		numThreads, _ = strconv.Atoi(os.Args[2])
		dataLength, _ = strconv.Atoi(os.Args[3])
		numRows, _ = strconv.Atoi(os.Args[4])
	*/

	C.initializeServer(C.int(numThreads))

	//calls follower for setup
	conf := &tls.Config{
		InsecureSkipVerify: true,
	}
	followerConnection, err := tls.Dial("tcp", follower, conf)
	if err != nil {
		panic(err)
	}
	followerConnection.SetDeadline(time.Time{})

	//receives follower publicKey
	var tmpFollowerPubKey [32]byte
	_, err = followerConnection.Read(tmpFollowerPubKey[:])
	if err != nil {
		panic(err)
	}
	followerPublicKey = &tmpFollowerPubKey

	//send publicKey to follower
	_, err = followerConnection.Write(leaderPublicKey[:])
	if err != nil {
		panic(err)
	}

	//goroutine for accepting new clients
	go func() {
		leaderConnectionPrivateKey, err := rsa.GenerateKey(rand.Reader, 2048)
		if err != nil {
			panic(err)
		}

		// Generate a pem block with the private key
		keyPem := pem.EncodeToMemory(&pem.Block{
			Type:  "RSA PRIVATE KEY",
			Bytes: x509.MarshalPKCS1PrivateKey(leaderConnectionPrivateKey),
		})

		tml := x509.Certificate{
			// you can add any attr that you need
			NotBefore: time.Now(),
			NotAfter:  time.Now().AddDate(5, 0, 0),
			// you have to generate a different serial number each execution
			SerialNumber: big.NewInt(123123),
			Subject: pkix.Name{
				CommonName:   "New Name",
				Organization: []string{"New Org."},
			},
			BasicConstraintsValid: true,
		}
		cert, err := x509.CreateCertificate(rand.Reader, &tml, &tml, &leaderConnectionPrivateKey.PublicKey, leaderConnectionPrivateKey)
		if err != nil {
			panic(err)
		}

		// Generate a pem block with the certificate
		certPem := pem.EncodeToMemory(&pem.Block{
			Type:  "CERTIFICATE",
			Bytes: cert,
		})

		tlsCert, err := tls.X509KeyPair(certPem, keyPem)
		if err != nil {
			panic(err)
		}

		config := &tls.Config{Certificates: []tls.Certificate{tlsCert}}

		//listens for clients
		lnClients, err := tls.Listen("tcp", ":4441", config)
		if err != nil {
			panic(err)
		}
		defer lnClients.Close()
		for {
			clientConnection, err := lnClients.Accept()
			if err != nil {
				panic(err)
			}
			clientConnection.SetDeadline(time.Time{})

			//sends topicList so client can participate in phase 3 asap
			sendTopicLists(clientConnection)

			//send leader publicKey
			_, err = clientConnection.Write(leaderPublicKey[:])
			if err != nil {
				panic(err)
			}

			//send follower publicKey
			_, err = clientConnection.Write(followerPublicKey[:])
			if err != nil {
				panic(err)
			}

			var clientPublicKey *[32]byte
			var tmpClientPublicKey [32]byte
			//gets publicKey from client
			_, err = clientConnection.Read(tmpClientPublicKey[:])
			if err != nil {
				panic(err)
			}

			clientPublicKey = &tmpClientPublicKey

			//this is the key for map(client data)
			remoteAddress := clientConnection.RemoteAddr()
			//pirQuery will be added in phase 3
			//bs! only want to set roundsParticipating and answerAmount to 0, mb there is a better way
			//will work for now
			var emptyArray [32]byte
			var emptyByteArray [][]byte
			keys := clientKeys{0, clientPublicKey, emptyArray, emptyByteArray}
			clientData[remoteAddress] = keys

			phase1Channel <- clientConnection
		}
	}()

	wg := &sync.WaitGroup{}

	//the current phase
	phase := make([]byte, 1)

	//locks access to DB
	var m sync.Mutex

	for {

		//phase1
		startTime = time.Now()
		phase[0] = 1

		fmt.Println("phase1")

		//creates a new write Db for this round
		for i := 0; i < dbWriteSize; i++ {
			C.createDb(C.int(1), C.int(dataLength))
		}
		for id := 0; id < numThreads; id++ {
			wg.Add(1)

			followerConnection, err := tls.Dial("tcp", follower, conf)
			if err != nil {
				panic(err)
			}
			followerConnection.SetDeadline(time.Time{})

			go phase1(id, phase, followerConnection, wg, m, startTime)
		}

		wg.Wait()

		fmt.Println("phase2")

		//phase2

		followerConnection, err := tls.Dial("tcp", follower, conf)
		if err != nil {
			panic(err)
		}
		followerConnection.SetDeadline(time.Time{})

		phase2(followerConnection)

		//phase3

		fmt.Println("phase3")

		if round == 1 {
			//addTestTweets()
		}

		//no tweets -> continue to phase 1 and mb get tweets
		topicList, topicAmount = lib.GetTopicList(0)
		if len(topicList) == 0 {
			continue
		}

		phase[0] = 3
		startTime = time.Now()

		for id := 0; id < numThreads; id++ {
			wg.Add(1)
			followerConnection, err := tls.Dial("tcp", follower, conf)
			if err != nil {
				panic(err)
			}
			followerConnection.SetDeadline(time.Time{})

			go phase3(id, phase, followerConnection, wg, startTime)
		}

		wg.Wait()

		lib.CleanUpdbR(round)

		round++
	}
}

func phase1(id int, phase []byte, followerConnection net.Conn, wg *sync.WaitGroup, m sync.Mutex, startTime time.Time) {

	roundAsBytes := intToByte(round)
	gotClient := make([]byte, 1)
	gotClient[0] = 0

	//wait until time is up
	for len(phase1Channel) == 0 {
		if time.Since(startTime) > maxTimePerRound {
			//tells follower that this worker is done
			_, err := followerConnection.Write(gotClient)
			if err != nil {
				panic(err)
			}
			wg.Done()
			return
		}
		time.Sleep(1 * time.Second)
	}

	for clientConnection := range phase1Channel {

		gotClient[0] = 1
		//tells follower that this worker got a clientConnection
		_, err := followerConnection.Write(gotClient)
		if err != nil {
			panic(err)
		}

		//sends clients publicKey to follower
		clientPublicKey := clientData[clientConnection.RemoteAddr()].PublicKey
		_, err = followerConnection.Write(clientPublicKey[:])
		if err != nil {
			panic(err)
		}

		//setup the worker-specific db
		dbSize := int(C.dbSize)
		db := make([][]byte, dbSize)
		for i := 0; i < dbSize; i++ {
			db[i] = make([]byte, int(C.db[i].dataSize))
		}

		//tells client that phase 1 has begun
		_, err = clientConnection.Write(phase)
		if err != nil {
			panic(err)
		}

		//tells client current dbWriteSize
		_, err = clientConnection.Write(intToByte(dbWriteSize))
		if err != nil {
			panic(err)
		}

		fmt.Println("SizeToClient", dbWriteSize)

		//tells client current round
		_, err = clientConnection.Write(roundAsBytes)
		if err != nil {
			panic(err)
		}

		//accept dpfQuery from client
		dpfLengthBytes := make([]byte, 4)
		_, err = clientConnection.Read(dpfLengthBytes)
		if err != nil {
			panic(err)
		}
		dpfLength := byteToInt(dpfLengthBytes)

		dpfQueryAEncrypted := make([]byte, dpfLength)
		dpfQueryBEncrypted := make([]byte, dpfLength)

		_, err = clientConnection.Read(dpfQueryAEncrypted)
		if err != nil {
			panic(err)
		}

		_, err = clientConnection.Read(dpfQueryBEncrypted)
		if err != nil {
			panic(err)
		}

		_, err = followerConnection.Write(dpfLengthBytes)
		if err != nil {
			panic(err)
		}

		_, err = followerConnection.Write(dpfQueryBEncrypted)
		if err != nil {
			panic(err)
		}

		//auditing starts here

		//generate seed
		var seed [16]byte
		_, err = rand.Read(seed[:])
		if err != nil {
			panic(err)
		}

		//send seed to client
		_, err = clientConnection.Write(seed[:])
		if err != nil {
			panic(err)
		}

		//send seed to follower
		_, err = followerConnection.Write(seed[:])
		if err != nil {
			panic(err)
		}

		//receive proofs
		auditLengthBytes := make([]byte, 4)

		_, err = clientConnection.Read(auditLengthBytes)
		if err != nil {
			panic(err)
		}

		auditLength := byteToInt(auditLengthBytes)

		clientAuditA := make([]byte, auditLength)
		clientAuditB := make([]byte, auditLength)

		_, err = clientConnection.Read(clientAuditA)
		if err != nil {
			panic(err)
		}

		_, err = clientConnection.Read(clientAuditB)
		if err != nil {
			panic(err)
		}

		//send client audit to follower
		_, err = followerConnection.Write(auditLengthBytes)
		if err != nil {
			panic(err)
		}

		_, err = followerConnection.Write(clientAuditB)
		if err != nil {
			panic(err)
		}

		//decrypts the client audit
		var decryptNonce [24]byte
		copy(decryptNonce[:], clientAuditA[:24])

		clientAuditA, ok := box.Open(nil, clientAuditA[24:], &decryptNonce, clientPublicKey, leaderPrivateKey)
		if !ok {
			panic("clientAudit decryption not ok")
		}

		//decrypt dpfQueryA for sorting into db
		copy(decryptNonce[:], dpfQueryAEncrypted[:24])
		dpfQueryA, ok := box.Open(nil, dpfQueryAEncrypted[24:], &decryptNonce, clientPublicKey, leaderPrivateKey)
		if !ok {
			panic("dpfQueryA decryption not ok")
		}

		vector := make([]byte, dbSize*16)
		//var str string
		//run dpf, xor into local db
		for i := 0; i < dbSize; i++ {
			ds := int(C.db[i].dataSize)
			dataShare := make([]byte, ds)
			pos := C.getUint128_t(C.int(i))
			v := C.evalDPF(C.ctx[id], (*C.uchar)(&dpfQueryA[0]), pos, C.int(ds), (*C.uchar)(&dataShare[0]))
			copy(vector[i*16:(i+1)*16], C.GoBytes(unsafe.Pointer(&v), 16))
			for j := 0; j < ds; j++ {
				db[i][j] = db[i][j] ^ dataShare[j]
			}
		}

		//prepare for audit
		mVal := make([]byte, 16)
		cVal := make([]byte, 16)
		C.serverSetupProof(C.ctx[id], (*C.uchar)(&seed[0]), C.dbSize, (*C.uchar)(&vector[0]), (*C.uchar)(&mVal[0]), (*C.uchar)(&cVal[0]))

		//compute audit query
		auditResultA := make([]byte, 96)
		C.serverComputeQuery(C.ctx[id], (*C.uchar)(&seed[0]), (*C.uchar)(&mVal[0]), (*C.uchar)(&cVal[0]), (*C.uchar)(&clientAuditA[0]), (*C.uchar)(&auditResultA[0]))

		//encrypt leader audit result
		var nonce [24]byte
		//fill nonce with randomness
		_, err = rand.Read(nonce[:])
		if err != nil {
			panic("couldn't get randomness for nonce!")
		}

		auditResultAEncrypted := box.Seal(nonce[:], auditResultA, &nonce, followerPublicKey, leaderPrivateKey)

		encryptedAuditResultALengthBytes := intToByte(len(auditResultAEncrypted))

		//send audit result to follower
		_, err = followerConnection.Write(encryptedAuditResultALengthBytes)
		if err != nil {
			panic(err)
		}

		_, err = followerConnection.Write(auditResultAEncrypted)
		if err != nil {
			panic(err)
		}

		//receive follower audit result
		auditResultBEncryptedLengthBytes := make([]byte, 4)
		_, err = followerConnection.Read(auditResultBEncryptedLengthBytes)
		if err != nil {
			panic(err)
		}
		auditResultBEncryptedLength := byteToInt(auditResultBEncryptedLengthBytes)

		auditResultBEncrypted := make([]byte, auditResultBEncryptedLength)
		_, err = followerConnection.Read(auditResultBEncrypted)
		if err != nil {
			panic(err)
		}

		//decrypts the audit result from follower
		copy(decryptNonce[:], auditResultBEncrypted[:24])
		auditResultB, ok := box.Open(nil, auditResultBEncrypted[24:], &decryptNonce, followerPublicKey, leaderPrivateKey)
		if !ok {
			panic("auditResultB decryption not ok")
		}

		//compute audit result
		auditResult := int(C.serverVerifyProof((*C.uchar)(&auditResultA[0]), (*C.uchar)(&auditResultB[0])))

		if byteToInt(auditResultB) == 0 || auditResult == 0 {
			//fmt.Println("audit failed")
		} /*else {
			fmt.Println("audit passed")
		}
		*/

		//xor the worker's DB into the main DB
		for i := 0; i < dbSize; i++ {
			m.Lock()
			C.xorIn(C.int(i), (*C.uchar)(&db[i][0]))
			m.Unlock()
		}
		phase3Channel <- clientConnection

		//loop that waits for new client or leaves phase1 if time is up
		for {
			if time.Since(startTime) < maxTimePerRound {
				//this worker handles the next client
				if len(phase1Channel) > 0 {
					break
					//this worker waits for next client
				} else {
					time.Sleep(1 * time.Second)
				}
				//times up
			} else {
				//tells follower that this worker is done
				gotClient[0] = 0
				_, err := followerConnection.Write(gotClient)
				if err != nil {
					panic(err)
				}
				wg.Done()
				return
			}
		}
	}
}

func phase2(followerConnection net.Conn) {

	//gets current seed
	seedLeader := make([]byte, 16)
	C.readSeed((*C.uchar)(&seedLeader[0]))

	//get data
	dbSize := int(C.dbSize)
	tmpdbLeader := make([][]byte, dbSize)
	for i := range tmpdbLeader {
		tmpdbLeader[i] = make([]byte, dataLength)
	}
	for i := 0; i < dbSize; i++ {
		C.readData(C.int(i), (*C.uchar)(&tmpdbLeader[i][0]))
	}

	//writes seed to follower
	_, err := followerConnection.Write(seedLeader)
	if err != nil {
		panic(err)
	}

	//write data to follower
	//this is surely inefficent
	for i := 0; i < dbSize; i++ {
		_, err = followerConnection.Write(tmpdbLeader[i])
		if err != nil {
			panic(err)
		}
	}

	//receive seed from follower
	seedFollower := make([]byte, 16)
	_, err = followerConnection.Read(seedFollower)
	if err != nil {
		panic(err)
	}

	//receive data from follower
	tmpdbFollower := make([][]byte, dbSize)
	for i := range tmpdbFollower {
		tmpdbFollower[i] = make([]byte, dataLength)
	}
	for i := 0; i < dbSize; i++ {
		_, err = followerConnection.Read(tmpdbFollower[i])
		if err != nil {
			panic(err)
		}
	}

	//put together the db
	tmpdb := make([][]byte, dbSize)
	for i := range tmpdb {
		tmpdb[i] = make([]byte, dataLength)
	}

	//get own Ciphers
	ciphersLeader := make([]*C.uchar, dbSize)
	for i := 0; i < dbSize; i++ {
		ciphersLeader[i] = (*C.uchar)(C.malloc(16))
	}
	for i := 0; i < dbSize; i++ {
		C.getCipher(1, C.int(i), ciphersLeader[i])
	}

	//send own Ciphers to follower
	for i := 0; i < dbSize; i++ {
		_, err = followerConnection.Write(C.GoBytes(unsafe.Pointer(ciphersLeader[i]), 16))
		if err != nil {
			panic(err)
		}
	}

	//receive ciphers from follower
	ciphersFollower := make([]byte, dbSize*16)
	for i := 0; i < dbSize; i++ {
		_, err = followerConnection.Read(ciphersFollower[i*16:])
		if err != nil {
			panic(err)
		}
	}

	//put in ciphers from follower
	for i := 0; i < dbSize; i++ {
		C.putCipher(1, C.int(i), (*C.uchar)(&ciphersFollower[i*16]))
	}

	//decrypt each row
	for i := 0; i < dbSize; i++ {
		C.decryptRow(C.int(i), (*C.uchar)(&tmpdb[i][0]), (*C.uchar)(&tmpdbLeader[i][0]), (*C.uchar)(&tmpdbFollower[i][0]), (*C.uchar)(&seedLeader[0]), (*C.uchar)(&seedFollower[0]))
	}

	var tweets []lib.Tweet
	publisherNumber = 0
	for i := 0; i < dbSize; i++ {
		//discard cover message
		if tmpdb[i][0] == 0 {
			continue
		} else {
			publisherNumber++
			//reconstruct tweet
			var position int = 0
			var topics []string
			var topic string
			var text string
			for _, letter := range tmpdb[i] {
				if string(letter) == ";" {
					if topic != "" {
						topics = append(topics, topic)
						topic = ""
					}
					position++
				} else {
					if position == 0 {
						if string(letter) == "," {
							topics = append(topics, topic)
							topic = ""
						} else {
							topic = topic + string(letter)
						}
					} else if position == 1 {
						text = text + string(letter)
					}
				}
			}

			tweet := lib.Tweet{"", -1, topics, text, round}

			tweets = append(tweets, tweet)
		}
	}

	//fmt.Println("tweets recovered: ", tweets)

	//sort into read db
	lib.NewEntries(tweets, 0)

	C.resetDb()

	//calculates the dbWriteSize for this round
	if publisherNumber > 0 {
		dbWriteSize = int(math.Ceil(19.5 * publisherNumber))
	} else {
		//default, handles 100 clients
		dbWriteSize = 1950
	}

	//writes dbWriteSize of current round to follower
	_, err = followerConnection.Write(intToByte(dbWriteSize))
	if err != nil {
		panic(err)
	}
}

func addTestTweets() {
	//creates test tweets
	tweets := make([]lib.Tweet, 5)

	for i := range tweets {
		j := i
		if i == 1 {
			j = 0
		}
		text := "Text " + strconv.Itoa(i)
		var topics []string
		topics = append(topics, "Topic "+strconv.Itoa(j))
		tweets[i] = lib.Tweet{"", -1, topics, text, i}
	}

	lib.NewEntries(tweets, 0)
}

//opti! mb it is quicker to send updated topicLists to clients first so pirQuerys are ready
func phase3(id int, phase []byte, followerConnection net.Conn, wg *sync.WaitGroup, startTime time.Time) {

	gotClient := make([]byte, 1)
	gotClient[0] = 0

	//wait until time is up
	for len(phase3Channel) == 0 {
		if time.Since(startTime) > maxTimePerRound {
			//tells follower that this worker is done
			_, err := followerConnection.Write(gotClient)
			if err != nil {
				panic(err)
			}
			wg.Done()
			return
		}
		time.Sleep(1 * time.Second)
	}

	for clientConnection := range phase3Channel {

		gotClient[0] = 1
		//tells follower that this worker got a clientConnection
		_, err := followerConnection.Write(gotClient)
		if err != nil {
			panic(err)
		}
		_, err = clientConnection.Write(phase)
		if err != nil {
			panic(err)
		}

		/*
			possible Values
			0 : new client
				leader expects sharedSecrets, expects pirQuery
			1 : update needed
				leader sends topicList, performs local update of sharedSecret, expects pirQuery
			2 : no update needed
				nothing
		*/
		subPhase := make([]byte, 1)

		//gets the data for the current client
		var clientKeys = clientData[clientConnection.RemoteAddr()]
		var roundsParticipating = clientKeys.roundsParticipating
		//client participates for the first time
		if roundsParticipating == 0 {
			subPhase[0] = 0
		} else if roundsParticipating%roundsBeforeUpdate == 0 {
			subPhase[0] = 1
		} else {
			subPhase[0] = 2
		}

		//tells client what leader expects
		_, err = clientConnection.Write(subPhase)
		if err != nil {
			panic(err)
		}

		//tells follower what will happen
		_, err = followerConnection.Write(subPhase)
		if err != nil {
			panic(err)
		}

		//sends clients publicKey so follower knows which client is being served
		_, err = followerConnection.Write(clientKeys.PublicKey[:])
		if err != nil {
			panic(err)
		}

		//increases rounds participating for client
		clientKeys.roundsParticipating = roundsParticipating + 1

		//declaring variables here to prevent dupclicates later
		var sharedSecret [32]byte = clientData[clientConnection.RemoteAddr()].SharedSecret

		if subPhase[0] == 0 {
			sendTopicLists(clientConnection)
			clientKeys, _ = handlePirQuery(clientKeys, clientConnection, followerConnection, int(subPhase[0]))
		} else if subPhase[0] == 1 {
			sendTopicLists(clientConnection)

			//updates sharedSecret
			sharedSecret = sha256.Sum256(sharedSecret[:])
			clientKeys.SharedSecret = sharedSecret

			clientKeys, _ = handlePirQuery(clientKeys, clientConnection, followerConnection, int(subPhase[0]))
		}

		getSendTweets(clientKeys, nil, clientConnection, followerConnection)

		wantsArchive := make([]byte, 1)
		_, err = clientConnection.Read(wantsArchive)
		if err != nil {
			panic(err)
		}

		followerConnection.Write(wantsArchive)
		if err != nil {
			panic(err)
		}

		if wantsArchive[0] == 1 && archiveTopicAmount > 0 {
			_, archiveQuerys := handlePirQuery(clientKeys, clientConnection, followerConnection, -1)
			getSendTweets(clientKeys, archiveQuerys, clientConnection, followerConnection)
		}

		//saves all changes for client
		clientData[clientConnection.RemoteAddr()] = clientKeys

		phase1Channel <- clientConnection

		for {
			if time.Since(startTime) < maxTimePerRound {
				//this worker handles the next client
				if len(phase3Channel) > 0 {
					break
					//this worker waits for next client
				} else {
					time.Sleep(1 * time.Second)
				}
				//times up
			} else {
				//tells follower that this worker is done
				gotClient[0] = 0
				_, err := followerConnection.Write(gotClient)
				if err != nil {
					panic(err)
				}
				wg.Done()
				return
			}
		}
	}
}

func getSendTweets(clientKeys clientKeys, archiveQuerys [][]byte, clientConnection, followerConnection net.Conn) {
	tmpNeededSubscriptions := neededSubscriptions
	if archiveQuerys != nil {
		tmpNeededSubscriptions = len(archiveQuerys)
	}
	for i := 0; i < tmpNeededSubscriptions; i++ {
		//gets all requested tweets
		var tweets []byte
		if archiveQuerys == nil {
			tweets = lib.GetTweets(clientKeys.PirQuery[i], dataLength, 0)
		} else {
			tweets = lib.GetTweets(archiveQuerys[i], dataLength, 1)
		}

		//expand sharedSecret so it is of right length
		expandBy := len(tweets) / 32
		var expandedSharedSecret []byte
		for i := 0; i < expandBy; i++ {
			expandedSharedSecret = append(expandedSharedSecret, clientKeys.SharedSecret[:]...)
		}

		//Xor's sharedSecret with all tweets
		lib.Xor(expandedSharedSecret[:], tweets)

		//receives tweets from follower and Xor's them in
		tweetsLengthBytes := make([]byte, 4)

		_, err := followerConnection.Read(tweetsLengthBytes)
		if err != nil {
			panic(err)
		}
		tweetsReceivedLength := byteToInt(tweetsLengthBytes)

		receivedTweets := make([]byte, tweetsReceivedLength)
		_, err = followerConnection.Read(receivedTweets)
		if err != nil {
			panic(err)
		}

		lib.Xor(receivedTweets, tweets)

		//sends tweets to client
		tweetsLengthBytes = intToByte(len(tweets))
		_, err = clientConnection.Write(tweetsLengthBytes)
		if err != nil {
			panic(err)
		}
		_, err = clientConnection.Write(tweets)
		if err != nil {
			panic(err)
		}
	}
}

func handlePirQuery(clientKeys clientKeys, clientConnection net.Conn, followerConnection net.Conn, subPhase int) (clientKeys, [][]byte) {

	clientPublicKey := clientKeys.PublicKey

	//gets the msg length
	msgLengthBytes := make([]byte, 4)
	_, err := clientConnection.Read(msgLengthBytes)
	if err != nil {
		panic(err)
	}
	msgLength := byteToInt(msgLengthBytes)

	leaderBox := make([]byte, msgLength)
	followerBox := make([]byte, msgLength)

	//gets the leader box
	_, err = clientConnection.Read(leaderBox)
	if err != nil {
		panic(err)
	}

	//gets the follower box
	_, err = clientConnection.Read(followerBox)
	if err != nil {
		panic(err)
	}

	tmpNeededSubscriptions := neededSubscriptions
	tmpTopicAmount := topicAmount
	if subPhase == -1 {
		archiveNeededSubscriptions := make([]byte, 4)
		_, err = clientConnection.Read(archiveNeededSubscriptions)
		if err != nil {
			panic(err)
		}
		_, err = followerConnection.Write(archiveNeededSubscriptions)
		if err != nil {
			panic(err)
		}
		tmpNeededSubscriptions = byteToInt(archiveNeededSubscriptions)
		tmpTopicAmount = archiveTopicAmount
	}

	//send length to follower
	_, err = followerConnection.Write(msgLengthBytes)
	if err != nil {
		panic(err)
	}

	//send box to follower
	_, err = followerConnection.Write(followerBox)
	if err != nil {
		panic(err)
	}

	var decryptNonce [24]byte
	copy(decryptNonce[:], leaderBox[:24])
	decrypted, ok := box.Open(nil, leaderBox[24:], &decryptNonce, clientPublicKey, leaderPrivateKey)
	if !ok {
		panic("pirQuery decryption not ok")
	}

	//if sharedSecret is send
	if subPhase == 0 {
		//bs!
		var tmpSharedSecret [32]byte
		for index := 0; index < 32; index++ {
			tmpSharedSecret[index] = decrypted[index]
		}
		clientKeys.SharedSecret = tmpSharedSecret
		decrypted = decrypted[32:]
	}

	//transforms byteArray to ints of wanted topics
	pirQueryFlattened := decrypted
	pirQuerys := make([][]byte, tmpNeededSubscriptions)
	for i := range pirQuerys {
		pirQuerys[i] = make([]byte, tmpTopicAmount)
	}
	for i := 0; i < tmpNeededSubscriptions; i++ {
		pirQuerys[i] = pirQueryFlattened[i*tmpTopicAmount : (i+1)*tmpTopicAmount]
	}

	//sets the pirQuery for the client in case whe are not archiving
	if subPhase != -1 {
		clientKeys.PirQuery = pirQuerys
	}

	return clientKeys, pirQuerys
}

func transformBytesToStringArray(topicsAsBytes []byte) []string {
	var topics []string
	var topic string
	var position int = 0
	for _, letter := range topicsAsBytes {
		if string(letter) == "," {
			topics[position] = topic
			topic = ""
			position++
		} else {
			topic = topic + string(letter)
		}
	}
	return topics
}

func byteToInt(myBytes []byte) (x int) {
	x = int(myBytes[3])<<24 + int(myBytes[2])<<16 + int(myBytes[1])<<8 + int(myBytes[0])
	return
}

func sendTopicLists(clientConnection net.Conn) {
	for i := 0; i < 2; i++ {
		var topicList []byte
		if i == 0 {
			topicList, topicAmount = lib.GetTopicList(i)
		} else {
			topicList, archiveTopicAmount = lib.GetTopicList(i)
		}
		topicListLengthBytes := intToByte(len(topicList))

		_, err := clientConnection.Write(topicListLengthBytes)
		if err != nil {
			panic(err)
		}

		_, err = clientConnection.Write(topicList)
		if err != nil {
			panic(err)
		}
	}
}

func intToByte(myInt int) (retBytes []byte) {
	retBytes = make([]byte, 4)
	retBytes[3] = byte((myInt >> 24) & 0xff)
	retBytes[2] = byte((myInt >> 16) & 0xff)
	retBytes[1] = byte((myInt >> 8) & 0xff)
	retBytes[0] = byte(myInt & 0xff)
	return
}