follower.go 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805
  1. package main
  2. //#cgo CFLAGS: -fopenmp -O2
  3. //#cgo LDFLAGS: -lcrypto -lm -fopenmp
  4. //#include "../c/dpf.h"
  5. //#include "../c/okv.h"
  6. //#include "../c/dpf.c"
  7. //#include "../c/okv.c"
  8. import "C"
  9. import (
  10. "2PPS/lib"
  11. "crypto/rand"
  12. "crypto/rsa"
  13. "crypto/sha256"
  14. "crypto/tls"
  15. "crypto/x509"
  16. "crypto/x509/pkix"
  17. "encoding/pem"
  18. "fmt"
  19. "math/big"
  20. "net"
  21. "strconv"
  22. "strings"
  23. "sync"
  24. "time"
  25. "unsafe"
  26. "golang.org/x/crypto/nacl/box"
  27. )
  28. //this stores all neccessary information for each client
  29. type clientKeys struct {
  30. SharedSecret [32]byte
  31. PirQuery [][]byte
  32. }
  33. //uses clients publicKey as key
  34. var clientData = make(map[[32]byte]clientKeys)
  35. var topicList []byte
  36. var topicAmount int
  37. var archiveTopicAmount int
  38. var followerPrivateKey *[32]byte
  39. var followerPublicKey *[32]byte
  40. var leaderPublicKey *[32]byte
  41. //needs to be changed at leader/follower/client at the same time
  42. const dataLength = 256
  43. const numThreads = 12
  44. //Maximum Transport Unit
  45. const mtu int = 1100
  46. var dbWriteSize int = 10000
  47. var neededSubscriptions int
  48. var round int = 0
  49. var startTime time.Time
  50. func main() {
  51. generatedPublicKey, generatedPrivateKey, err := box.GenerateKey(rand.Reader)
  52. if err != nil {
  53. panic(err)
  54. }
  55. followerPrivateKey = generatedPrivateKey
  56. followerPublicKey = generatedPublicKey
  57. C.initializeServer(C.int(numThreads))
  58. followerConnectionPrivateKey, err := rsa.GenerateKey(rand.Reader, 2048)
  59. if err != nil {
  60. panic(err)
  61. }
  62. // Generate a pem block with the private key
  63. keyPem := pem.EncodeToMemory(&pem.Block{
  64. Type: "RSA PRIVATE KEY",
  65. Bytes: x509.MarshalPKCS1PrivateKey(followerConnectionPrivateKey),
  66. })
  67. tml := x509.Certificate{
  68. // you can add any attr that you need
  69. NotBefore: time.Now(),
  70. NotAfter: time.Now().AddDate(5, 0, 0),
  71. // you have to generate a different serial number each execution
  72. SerialNumber: big.NewInt(123123),
  73. Subject: pkix.Name{
  74. CommonName: "New Name",
  75. Organization: []string{"New Org."},
  76. },
  77. BasicConstraintsValid: true,
  78. }
  79. cert, err := x509.CreateCertificate(rand.Reader, &tml, &tml, &followerConnectionPrivateKey.PublicKey, followerConnectionPrivateKey)
  80. if err != nil {
  81. panic(err)
  82. }
  83. // Generate a pem block with the certificate
  84. certPem := pem.EncodeToMemory(&pem.Block{
  85. Type: "CERTIFICATE",
  86. Bytes: cert,
  87. })
  88. tlsCert, err := tls.X509KeyPair(certPem, keyPem)
  89. if err != nil {
  90. panic(err)
  91. }
  92. config := &tls.Config{Certificates: []tls.Certificate{tlsCert}}
  93. fmt.Println("start leader")
  94. //listens for leader
  95. lnLeader, err := tls.Listen("tcp", ":4442", config)
  96. if err != nil {
  97. panic(err)
  98. }
  99. defer lnLeader.Close()
  100. leaderConnection, err := lnLeader.Accept()
  101. if err != nil {
  102. panic(err)
  103. }
  104. //send publicKey to leader
  105. writeTo(leaderConnection, followerPublicKey[:])
  106. //receives leader PublicKey
  107. var tmpLeaderPubKey [32]byte
  108. _, err = leaderConnection.Read(tmpLeaderPubKey[:])
  109. if err != nil {
  110. panic(err)
  111. }
  112. neededSubscriptionsBytes := readFrom(leaderConnection, 4)
  113. neededSubscriptions = byteToInt(neededSubscriptionsBytes)
  114. leaderPublicKey = &tmpLeaderPubKey
  115. //setup ends here
  116. //locks access to DB
  117. m := &sync.RWMutex{}
  118. wg := &sync.WaitGroup{}
  119. for {
  120. round++
  121. fmt.Println("Phase 1 Round", round)
  122. //create write db for this round
  123. for i := 0; i < dbWriteSize; i++ {
  124. C.createDb(C.int(0), C.int(dataLength))
  125. }
  126. //receives the virtualAddresses
  127. virtualAddresses := make([]int, dbWriteSize+1)
  128. for i := 0; i <= dbWriteSize; i++ {
  129. virtualAddress := readFrom(leaderConnection, 4)
  130. virtualAddresses[i] = byteToInt(virtualAddress)
  131. }
  132. for i := 0; i < numThreads; i++ {
  133. wg.Add(1)
  134. leaderConnection, err := lnLeader.Accept()
  135. if err != nil {
  136. panic(err)
  137. }
  138. leaderConnection.SetDeadline(time.Time{})
  139. startTime = time.Now()
  140. go phase1(i, leaderConnection, m, wg, virtualAddresses)
  141. }
  142. wg.Wait()
  143. //Phase 2
  144. leaderConnection, err := lnLeader.Accept()
  145. if err != nil {
  146. panic(err)
  147. }
  148. leaderConnection.SetDeadline(time.Time{})
  149. phase2(leaderConnection)
  150. //Phase 3
  151. if round == 1 {
  152. //addTestTweets()
  153. }
  154. //no tweets -> continue to phase 1 and mb get tweets
  155. topicList, topicAmount = lib.GetTopicList(0)
  156. if len(topicList) == 0 {
  157. continue
  158. }
  159. for i := 0; i < numThreads; i++ {
  160. wg.Add(1)
  161. leaderConnection, err := lnLeader.Accept()
  162. if err != nil {
  163. panic(err)
  164. }
  165. leaderConnection.SetDeadline(time.Time{})
  166. startTime = time.Now()
  167. go phase3(leaderConnection, wg, m)
  168. }
  169. wg.Wait()
  170. lib.CleanUpdbR(round)
  171. }
  172. }
  173. func phase1(id int, leaderWorkerConnection net.Conn, m *sync.RWMutex, wg *sync.WaitGroup, virtualAddresses []int) {
  174. for {
  175. gotClient := readFrom(leaderWorkerConnection, 1)
  176. //this worker is done
  177. if gotClient[0] == 0 {
  178. wg.Done()
  179. return
  180. }
  181. //setup the worker-specific db
  182. dbSize := int(C.dbSize)
  183. db := make([][]byte, dbSize)
  184. for i := 0; i < dbSize; i++ {
  185. db[i] = make([]byte, int(C.db[i].dataSize))
  186. }
  187. //gets clients publicKey
  188. var clientPublicKey *[32]byte
  189. var tmpClientPublicKey [32]byte
  190. _, err := leaderWorkerConnection.Read(tmpClientPublicKey[:])
  191. if err != nil {
  192. fmt.Println("no error handling")
  193. panic(err)
  194. }
  195. clientPublicKey = &tmpClientPublicKey
  196. m.RLock()
  197. clientKeys := clientData[tmpClientPublicKey]
  198. m.RUnlock()
  199. clientKeys, pirQuery, errorBool := handlePirQuery(clientKeys, leaderWorkerConnection, 0, tmpClientPublicKey, true)
  200. if errorBool {
  201. continue
  202. }
  203. getSendVirtualAddress(pirQuery[0], virtualAddresses, clientKeys.SharedSecret, leaderWorkerConnection)
  204. m.Lock()
  205. clientData[*clientPublicKey] = clientKeys
  206. m.Unlock()
  207. //gets dpfQuery from leader
  208. dpfLengthBytes, errorBool := readFromWError(leaderWorkerConnection, 4)
  209. if errorBool {
  210. continue
  211. }
  212. dpfLength := byteToInt(dpfLengthBytes)
  213. dpfQueryBEncrypted, errorBool := readFromWError(leaderWorkerConnection, dpfLength)
  214. if errorBool {
  215. continue
  216. }
  217. //decrypt dpfQueryB for sorting into db
  218. var decryptNonce [24]byte
  219. copy(decryptNonce[:], dpfQueryBEncrypted[:24])
  220. dpfQueryB, ok := box.Open(nil, dpfQueryBEncrypted[24:], &decryptNonce, clientPublicKey, followerPrivateKey)
  221. if !ok {
  222. panic("dpfQueryB decryption not ok")
  223. }
  224. ds := int(C.db[0].dataSize)
  225. dataShareFollower := make([]byte, ds)
  226. pos := C.getUint128_t(C.int(virtualAddresses[dbWriteSize]))
  227. C.evalDPF(C.ctx[id], (*C.uchar)(&dpfQueryB[0]), pos, C.int(ds), (*C.uchar)(&dataShareFollower[0]))
  228. writeTo(leaderWorkerConnection, dataShareFollower)
  229. dataShareLeader, errorBool := readFromWError(leaderWorkerConnection, ds)
  230. if errorBool {
  231. continue
  232. }
  233. auditXOR := make([]byte, ds)
  234. passedAudit := true
  235. for i := 0; i < ds; i++ {
  236. auditXOR[i] = dataShareLeader[i] ^ dataShareFollower[i]
  237. //client tried to write to a position that is not a virtuallAddress
  238. if auditXOR[i] != 0 {
  239. passedAudit = false
  240. }
  241. }
  242. if passedAudit {
  243. //run dpf, xor into local db
  244. for i := 0; i < dbSize; i++ {
  245. ds := int(C.db[i].dataSize)
  246. dataShare := make([]byte, ds)
  247. pos := C.getUint128_t(C.int(virtualAddresses[i]))
  248. C.evalDPF(C.ctx[id], (*C.uchar)(&dpfQueryB[0]), pos, C.int(ds), (*C.uchar)(&dataShare[0]))
  249. for j := 0; j < ds; j++ {
  250. db[i][j] = db[i][j] ^ dataShare[j]
  251. }
  252. }
  253. //xor the worker's DB into the main DB
  254. for i := 0; i < dbSize; i++ {
  255. m.Lock()
  256. C.xorIn(C.int(i), (*C.uchar)(&db[i][0]))
  257. m.Unlock()
  258. }
  259. }
  260. }
  261. }
  262. func phase2(leaderWorkerConnection net.Conn) {
  263. //gets current seed
  264. seedFollower := make([]byte, 16)
  265. C.readSeed((*C.uchar)(&seedFollower[0]))
  266. //get data
  267. dbSize := int(C.dbSize)
  268. tmpdbFollower := make([][]byte, dbSize)
  269. for i := range tmpdbFollower {
  270. tmpdbFollower[i] = make([]byte, dataLength)
  271. }
  272. for i := 0; i < dbSize; i++ {
  273. C.readData(C.int(i), (*C.uchar)(&tmpdbFollower[i][0]))
  274. }
  275. //receive seed from leader
  276. seedLeader := readFrom(leaderWorkerConnection, 16)
  277. //receive data from leader
  278. tmpdbLeader := make([][]byte, dbSize)
  279. for i := range tmpdbLeader {
  280. tmpdbLeader[i] = make([]byte, dataLength)
  281. }
  282. for i := 0; i < dbSize; i++ {
  283. tmpdbLeader[i] = readFrom(leaderWorkerConnection, dataLength)
  284. }
  285. //writes seed to leader
  286. writeTo(leaderWorkerConnection, seedFollower)
  287. //write data to leader
  288. for i := 0; i < dbSize; i++ {
  289. writeTo(leaderWorkerConnection, tmpdbFollower[i])
  290. }
  291. //put together the db
  292. tmpdb := make([][]byte, dbSize)
  293. for i := range tmpdb {
  294. tmpdb[i] = make([]byte, dataLength)
  295. }
  296. //get own Ciphers
  297. ciphersFollowers := make([]*C.uchar, dbSize)
  298. for i := 0; i < dbSize; i++ {
  299. ciphersFollowers[i] = (*C.uchar)(C.malloc(16))
  300. }
  301. for i := 0; i < dbSize; i++ {
  302. C.getCipher(0, C.int(i), ciphersFollowers[i])
  303. }
  304. //receive ciphers from leader
  305. ciphersLeader := make([]byte, dbSize*16)
  306. for i := 0; i < dbSize; i++ {
  307. _, err := leaderWorkerConnection.Read(ciphersLeader[i*16:])
  308. if err != nil {
  309. panic(err)
  310. }
  311. }
  312. //send own Ciphers to leader
  313. for i := 0; i < dbSize; i++ {
  314. writeTo(leaderWorkerConnection, C.GoBytes(unsafe.Pointer(ciphersFollowers[i]), 16))
  315. }
  316. //put in ciphers from leader
  317. for i := 0; i < dbSize; i++ {
  318. C.putCipher(0, C.int(i), (*C.uchar)(&ciphersLeader[i*16]))
  319. }
  320. for i := 0; i < dbSize; i++ {
  321. 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]))
  322. }
  323. var tweets []lib.Tweet
  324. for i := 0; i < dbSize; i++ {
  325. //discard cover message
  326. if tmpdb[i][1] == 0 {
  327. continue
  328. } else if -1 == strings.Index(string(tmpdb[i]), ";;") {
  329. continue
  330. } else {
  331. //reconstruct tweet
  332. var position int = 0
  333. var topics []string
  334. var topic string
  335. var text string
  336. for _, letter := range tmpdb[i] {
  337. if string(letter) == ";" {
  338. if topic != "" {
  339. topics = append(topics, topic)
  340. topic = ""
  341. }
  342. position++
  343. } else {
  344. if position == 0 {
  345. if string(letter) == "," {
  346. topics = append(topics, topic)
  347. topic = ""
  348. } else {
  349. //if topics are
  350. //int
  351. //topic = topic + fmt.Sprint(int(letter))
  352. //string
  353. topic = topic + string(letter)
  354. }
  355. } else if position == 1 {
  356. text = text + string(letter)
  357. }
  358. }
  359. }
  360. tweet := lib.Tweet{"", -1, topics, text, round}
  361. if text != "" {
  362. tweets = append(tweets, tweet)
  363. }
  364. }
  365. }
  366. //fmt.Println("tweets recovered: ", tweets)
  367. //sort into read db
  368. lib.NewEntries(tweets, 0)
  369. //reset write db after the tweets were moved to read db
  370. C.resetDb()
  371. //gets current dbWriteSize from leader
  372. dbWriteSizeBytes := readFrom(leaderWorkerConnection, 4)
  373. dbWriteSize = byteToInt(dbWriteSizeBytes)
  374. lib.CleanUpdbR(round)
  375. }
  376. func addTestTweets() {
  377. //creates test tweets
  378. tweets := make([]lib.Tweet, 5)
  379. for i := range tweets {
  380. j := i
  381. if i == 1 {
  382. j = 0
  383. }
  384. text := "Text " + strconv.Itoa(i)
  385. var topics []string
  386. topics = append(topics, "Topic "+strconv.Itoa(j))
  387. tweets[i] = lib.Tweet{"", -1, topics, text, i}
  388. }
  389. lib.NewEntries(tweets, 0)
  390. }
  391. func phase3(leaderWorkerConnection net.Conn, wg *sync.WaitGroup, m *sync.RWMutex) {
  392. for {
  393. gotClient, errorBool := readFromWError(leaderWorkerConnection, 1)
  394. if errorBool {
  395. continue
  396. }
  397. //this worker is done
  398. if gotClient[0] == 0 {
  399. wg.Done()
  400. return
  401. }
  402. subPhase, errorBool := readFromWError(leaderWorkerConnection, 1)
  403. if errorBool {
  404. continue
  405. }
  406. var clientPublicKey [32]byte
  407. _, err := leaderWorkerConnection.Read(clientPublicKey[:])
  408. if err != nil {
  409. fmt.Println("no error handling")
  410. panic(err)
  411. }
  412. //gets the client data
  413. m.RLock()
  414. clientKeys := clientData[clientPublicKey]
  415. m.RUnlock()
  416. if subPhase[0] == 0 || subPhase[0] == 1 {
  417. clientKeys, _, errorBool = handlePirQuery(clientKeys, leaderWorkerConnection, int(subPhase[0]), clientPublicKey, false)
  418. if errorBool {
  419. continue
  420. }
  421. }
  422. getSendTweets(clientKeys, nil, leaderWorkerConnection, clientPublicKey)
  423. wantsArchive, errorBool := readFromWError(leaderWorkerConnection, 1)
  424. if errorBool {
  425. continue
  426. }
  427. if wantsArchive[0] == 1 {
  428. _, archiveQuerys, errorBool := handlePirQuery(clientKeys, leaderWorkerConnection, -1, clientPublicKey, false)
  429. if errorBool {
  430. continue
  431. }
  432. getSendTweets(clientKeys, archiveQuerys, leaderWorkerConnection, clientPublicKey)
  433. }
  434. //saves clientKeys
  435. m.Lock()
  436. clientData[clientPublicKey] = clientKeys
  437. m.Unlock()
  438. }
  439. }
  440. //gets tweet from db and sends them to leader
  441. func getSendTweets(clientKeys clientKeys, archiveQuerys [][]byte, leaderWorkerConnection net.Conn, pubKey [32]byte) {
  442. tmpNeededSubscriptions := neededSubscriptions
  443. if tmpNeededSubscriptions > topicAmount {
  444. tmpNeededSubscriptions = topicAmount
  445. }
  446. if archiveQuerys != nil {
  447. tmpNeededSubscriptions = len(archiveQuerys)
  448. if tmpNeededSubscriptions > archiveTopicAmount {
  449. tmpNeededSubscriptions = archiveTopicAmount
  450. }
  451. }
  452. for i := 0; i < tmpNeededSubscriptions; i++ {
  453. //gets all requested tweets
  454. var tweets []byte
  455. if archiveQuerys == nil {
  456. tweets = lib.GetTweets(clientKeys.PirQuery[i], dataLength, 0, pubKey)
  457. } else {
  458. tweets = lib.GetTweets(archiveQuerys[i], dataLength, 1, pubKey)
  459. }
  460. //expand sharedSecret so it is of right length
  461. expandBy := len(tweets) / 32
  462. var expandedSharedSecret []byte
  463. for i := 0; i < expandBy; i++ {
  464. expandedSharedSecret = append(expandedSharedSecret, clientKeys.SharedSecret[:]...)
  465. }
  466. //Xor's sharedSecret with all tweets
  467. lib.Xor(expandedSharedSecret[:], tweets)
  468. if archiveQuerys != nil {
  469. //fmt.Println("length", len(tweets))
  470. }
  471. //sends tweets to leader
  472. writeTo(leaderWorkerConnection, tweets)
  473. }
  474. }
  475. //returns true if client connection is lost
  476. func handlePirQuery(clientKeys clientKeys, leaderWorkerConnection net.Conn, subPhase int, clientPublicKey [32]byte, doAuditing bool) (clientKeys, [][]byte, bool) {
  477. archiveNeededSubscriptions := make([]byte, 4)
  478. if subPhase == -1 {
  479. var errorBool bool
  480. archiveNeededSubscriptions, errorBool = readFromWError(leaderWorkerConnection, 4)
  481. if errorBool {
  482. return clientKeys, nil, true
  483. }
  484. }
  485. //gets the msg length
  486. msgLengthBytes, errorBool := readFromWError(leaderWorkerConnection, 4)
  487. if errorBool {
  488. return clientKeys, nil, true
  489. }
  490. msgLength := byteToInt(msgLengthBytes)
  491. //gets the message
  492. message, errorBool := readFromWError(leaderWorkerConnection, msgLength)
  493. if errorBool {
  494. return clientKeys, nil, true
  495. }
  496. var decryptNonce [24]byte
  497. copy(decryptNonce[:], message[:24])
  498. decrypted, ok := box.Open(nil, message[24:], &decryptNonce, &clientPublicKey, followerPrivateKey)
  499. if !ok {
  500. fmt.Println("pirQuery decryption not ok")
  501. return clientKeys, nil, true
  502. }
  503. //gets sharedSecret
  504. if subPhase == 0 {
  505. var newSharedSecret [32]byte
  506. for index := 0; index < 32; index++ {
  507. newSharedSecret[index] = decrypted[index]
  508. }
  509. clientKeys.SharedSecret = newSharedSecret
  510. decrypted = decrypted[32:]
  511. if doAuditing {
  512. result := make([][]byte, 1)
  513. result[0] = decrypted
  514. return clientKeys, result, false
  515. }
  516. //follower updates sharedSecret
  517. } else if subPhase == 1 {
  518. sharedSecret := clientKeys.SharedSecret
  519. sharedSecret = sha256.Sum256(sharedSecret[:])
  520. clientKeys.SharedSecret = sharedSecret
  521. }
  522. //follower expects pirQuery
  523. //transforms byteArray to ints of wanted topics
  524. tmpNeededSubscriptions := neededSubscriptions
  525. if tmpNeededSubscriptions > topicAmount {
  526. tmpNeededSubscriptions = topicAmount
  527. }
  528. tmpTopicAmount := topicAmount
  529. if subPhase == -1 {
  530. tmpNeededSubscriptions = byteToInt(archiveNeededSubscriptions)
  531. _, archiveTopicAmount = lib.GetTopicList(1)
  532. if tmpNeededSubscriptions > archiveTopicAmount {
  533. tmpNeededSubscriptions = archiveTopicAmount
  534. }
  535. tmpTopicAmount = archiveTopicAmount
  536. }
  537. pirQueryFlattened := decrypted
  538. pirQuerys := make([][]byte, tmpNeededSubscriptions)
  539. for i := range pirQuerys {
  540. pirQuerys[i] = make([]byte, tmpTopicAmount)
  541. }
  542. for i := 0; i < tmpNeededSubscriptions; i++ {
  543. pirQuerys[i] = pirQueryFlattened[i*tmpTopicAmount : (i+1)*tmpTopicAmount]
  544. }
  545. //sets the pirQuery for the client in case whe are not archiving
  546. if subPhase != -1 {
  547. clientKeys.PirQuery = pirQuerys
  548. }
  549. if subPhase == -1 {
  550. //fmt.Println("query", pirQuerys)
  551. }
  552. return clientKeys, pirQuerys, false
  553. }
  554. func getSendVirtualAddress(pirQuery []byte, virtualAddresses []int, sharedSecret [32]byte, leaderWorkerConnection net.Conn) {
  555. //xores all requested addresses into virtuallAddress
  556. virtualAddress := make([]byte, 4)
  557. for index, num := range pirQuery {
  558. if num == 1 {
  559. currentAddress := intToByte(virtualAddresses[index])
  560. for i := 0; i < 4; i++ {
  561. virtualAddress[i] = virtualAddress[i] ^ currentAddress[i]
  562. }
  563. }
  564. }
  565. //xores the sharedSecret
  566. for i := 0; i < 4; i++ {
  567. virtualAddress[i] = virtualAddress[i] ^ sharedSecret[i]
  568. }
  569. writeTo(leaderWorkerConnection, virtualAddress)
  570. }
  571. //sends the array to the connection
  572. func writeTo(connection net.Conn, array []byte) {
  573. remainingLength := len(array)
  574. for remainingLength > 0 {
  575. if remainingLength >= mtu {
  576. _, err := connection.Write(array[:mtu])
  577. if err != nil {
  578. panic(err)
  579. }
  580. array = array[mtu:]
  581. remainingLength -= mtu
  582. } else {
  583. _, err := connection.Write(array)
  584. if err != nil {
  585. panic(err)
  586. }
  587. remainingLength = 0
  588. }
  589. }
  590. }
  591. //reads an array which is returned and of size "size" from the connection
  592. //returned bool is one if connection to client was lost
  593. func readFrom(connection net.Conn, size int) []byte {
  594. var array []byte
  595. remainingSize := size
  596. for remainingSize > 0 {
  597. var err error
  598. toAppend := make([]byte, mtu)
  599. if remainingSize > mtu {
  600. _, err = connection.Read(toAppend)
  601. array = append(array, toAppend...)
  602. remainingSize -= mtu
  603. } else {
  604. _, err = connection.Read(toAppend[:remainingSize])
  605. array = append(array, toAppend[:remainingSize]...)
  606. remainingSize = 0
  607. }
  608. if err != nil {
  609. panic(err)
  610. }
  611. }
  612. return array
  613. }
  614. //reads an array which is returned and of size "size" from the connection
  615. //returns true if connection to client is lost
  616. func readFromWError(connection net.Conn, size int) ([]byte, bool) {
  617. var array []byte
  618. remainingSize := size + 1
  619. for remainingSize > 0 {
  620. var err error
  621. toAppend := make([]byte, mtu)
  622. if remainingSize > mtu {
  623. _, err = connection.Read(toAppend)
  624. array = append(array, toAppend...)
  625. remainingSize -= mtu
  626. } else {
  627. _, err = connection.Read(toAppend[:remainingSize])
  628. array = append(array, toAppend[:remainingSize]...)
  629. remainingSize = 0
  630. }
  631. if err != nil {
  632. panic(err)
  633. }
  634. }
  635. if array[0] == 1 {
  636. return nil, true
  637. }
  638. return array[1:], false
  639. /*
  640. array := make([]byte, size+1)
  641. _, err := connection.Read(array)
  642. if err != nil {
  643. panic(err)
  644. }
  645. if array[0] == 1 {
  646. return nil, true
  647. }
  648. return array[1:], false
  649. */
  650. }
  651. func transformBytesToStringArray(topicsAsBytes []byte) []string {
  652. var topics []string
  653. var topic string
  654. var position int = 0
  655. for _, letter := range topicsAsBytes {
  656. if string(letter) == "," {
  657. topics[position] = topic
  658. topic = ""
  659. position++
  660. } else {
  661. topic = topic + string(letter)
  662. }
  663. }
  664. return topics
  665. }
  666. func byteToInt(myBytes []byte) (x int) {
  667. x = int(myBytes[3])<<24 + int(myBytes[2])<<16 + int(myBytes[1])<<8 + int(myBytes[0])
  668. return
  669. }
  670. func intToByte(myInt int) (retBytes []byte) {
  671. retBytes = make([]byte, 4)
  672. retBytes[3] = byte((myInt >> 24) & 0xff)
  673. retBytes[2] = byte((myInt >> 16) & 0xff)
  674. retBytes[1] = byte((myInt >> 8) & 0xff)
  675. retBytes[0] = byte(myInt & 0xff)
  676. return
  677. }