follower.go 19 KB

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