123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982 |
- 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)
- //variables for calculating the dbWrite size
- const publisherRounds int = 3
- var publisherAmount float64
- var publisherHistory [publisherRounds]int
- 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 {
- fmt.Println(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 {
- fmt.Println(err)
- clientConnection.Close()
- break
- }
- //send follower publicKey
- _, err = clientConnection.Write(followerPublicKey[:])
- if err != nil {
- fmt.Println(err)
- clientConnection.Close()
- break
- }
- var clientPublicKey *[32]byte
- var tmpClientPublicKey [32]byte
- //gets publicKey from client
- _, err = clientConnection.Read(tmpClientPublicKey[:])
- if err != nil {
- fmt.Println(err)
- clientConnection.Close()
- break
- }
- 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")
- //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)
- }
- //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)
- }
- //decrypt dpfQueryA for sorting into db
- var decryptNonce [24]byte
- 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]
- }
- }
- //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
- var currentPublisherAmount int = 0
- for i := 0; i < dbSize; i++ {
- //discard cover message
- if tmpdb[i][0] == 0 {
- continue
- } else {
- currentPublisherAmount++
- //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 publisherAverage over the last publisherRounds rounds
- index := round % publisherRounds
- publisherHistory[index] = currentPublisherAmount
- var publisherAmount int
- for _, num := range publisherHistory {
- publisherAmount += num
- }
- publisherAverage := 0
- if round < publisherRounds {
- publisherAverage = publisherAmount / round
- } else {
- publisherAverage = publisherAmount / publisherRounds
- }
- //calculates the dbWriteSize for this round
- dbWriteSize = int(math.Ceil(19.5 * float64(publisherAverage)))
- //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
- }
|