aboutsummaryrefslogtreecommitdiff
path: root/src/Database.cpp
blob: 226aa05473dac6b3e65f8547a2b60aafd08aebb6 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
#include "../include/odhtdb/Database.hpp"
#include "../include/odhtdb/Group.hpp"
#include "../include/odhtdb/LocalUser.hpp"
#include "../include/odhtdb/RemoteUser.hpp"
#include "../include/odhtdb/Encryption.hpp"
#include "../include/odhtdb/DhtKey.hpp"
#include "../include/odhtdb/bin2hex.hpp"
#include "../include/odhtdb/Log.hpp"
#include <boost/uuid/uuid_generators.hpp>
#include <opendht.h>
#include <fmt/format.h>
#include <sodium/randombytes.h>
#include <thread>
#include <chrono>
#include <sibs/SafeSerializer.hpp>
#include <sibs/SafeDeserializer.hpp>
#include <cassert>

using namespace dht;
using namespace std;
using namespace chrono_literals;

static int databaseCount = 0;
// TODO: Verify time_t is always signed
static time_t timeOffset = 0; // Updated by comparing local time with ntp server
static thread *ntpThread = nullptr;
static bool timestampSynced = false;
static InfoHash CREATE_DATA_HASH = InfoHash::get("__odhtdb__.create_data");
static InfoHash ADD_DATA_HASH = InfoHash::get("__odhtdb__.add_data");

const int OPENDHT_INFOHASH_LEN = 20;

namespace odhtdb
{
    const u16 DATABASE_CREATE_PACKET_STRUCTURE_VERSION = 1;
    const u16 DATABASE_ADD_PACKET_STRUCTURE_VERSION = 1;
    
    class RequestQuarantineException : public runtime_error
    {
    public:
        RequestQuarantineException() : runtime_error("Request quarantine, will be processed later (can be real of fake request)") {}
    };
    
    DataView combine(sibs::SafeSerializer &headerSerializer, const Encryption &encryptedData)
    {
        usize allocationSize = headerSerializer.getBuffer().size() + encryptedData.getNonce().size + encryptedData.getCipherText().size;
        char *result = new char[allocationSize];
        memcpy(result, headerSerializer.getBuffer().data(), headerSerializer.getBuffer().size());
        memcpy(result + headerSerializer.getBuffer().size(), encryptedData.getNonce().data, encryptedData.getNonce().size);
        memcpy(result + headerSerializer.getBuffer().size() + encryptedData.getNonce().size, encryptedData.getCipherText().data, encryptedData.getCipherText().size);
        return DataView(result, allocationSize);
    }
    
    DataView combine(const Signature::PublicKey &publicKey, const string &signedEncryptedData)
    {
        usize allocationSize = publicKey.getSize() + signedEncryptedData.size();
        char *result = new char[allocationSize];
        memcpy(result, publicKey.getData(), publicKey.getSize());
        memcpy(result + publicKey.getSize(), signedEncryptedData.data(), signedEncryptedData.size());
        return DataView(result, allocationSize);
    }
    
    DatabaseCreateResponse::DatabaseCreateResponse(LocalUser *_nodeAdminUser, const shared_ptr<OwnedMemory> &_key, const shared_ptr<Hash> &_hash) : 
        nodeAdminUser(_nodeAdminUser),
        key(_key),
        hash(_hash)
    {
        
    }
    
    const LocalUser* DatabaseCreateResponse::getNodeAdminUser() const
    {
        return nodeAdminUser;
    }
        
    const shared_ptr<OwnedMemory> DatabaseCreateResponse::getNodeEncryptionKey() const
    {
        return key;
    }
    
    const shared_ptr<Hash> DatabaseCreateResponse::getRequestHash() const
    {
        return hash;
    }
    
    Database::Database(const char *bootstrapNodeAddr, u16 port, const boost::filesystem::path &storageDir) : 
        onCreateNodeCallbackFunc(nullptr),
        onAddNodeCallbackFunc(nullptr),
        onAddUserCallbackFunc(nullptr),
        databaseStorage(storageDir)
    {
        // TODO: Cache this in storage. It takes pretty long time to generate new identity
        auto identity = dht::crypto::generateIdentity();
        node.run(port , {
            /*.dht_config = */{
                /*.node_config = */{
                    /*.node_id = */{},
                    /*.network = */0,
                    /*.is_bootstrap = */false,
                    /*.maintain_storage*/false
                },
                /*.id = */identity
            },
            /*.threaded = */true,
            /*.proxy_server = */"",
            /*.push_node_id = */""
        });
        fmt::MemoryWriter portStr;
        portStr << port;
        node.bootstrap(bootstrapNodeAddr, portStr.c_str());

        // TODO: Make this work for multiple threads initializing database at same time
        ++databaseCount;
        if(databaseCount == 1)
        {
            if(ntpThread)
                delete ntpThread;

            ntpThread = new thread([]()
            {
                ntp::NtpClient ntpClient("pool.ntp.org");
                while(databaseCount > 0)
                {
                    ntp::NtpTimestamp ntpTimestamp = ntpClient.getTimestamp();
                    timeOffset = time(nullptr) - ntpTimestamp.seconds;
                    timestampSynced = true;
                    // TODO: Also use timestamp fraction (milliseconds)
                    this_thread::sleep_for(60s);
                }
                timestampSynced = false;
            });

            // TODO: Catch std::system_error instead of this if-statement
            if(ntpThread->joinable())
                ntpThread->detach();
        }

        while(!timestampSynced)
        {
            this_thread::sleep_for(10ms);
        }
    }

    Database::~Database()
    {
        // TODO: Make this work for multiple threads removing database object at same time
        --databaseCount;
        node.join();
    }

    void Database::seed(const DatabaseNode &nodeToSeed)
    {
        // TODO: Use cached files and seed those. If none exists, request new files to seed.
        // If nobody requests my cached files in a long time, request new files to seed and remove cached files
        // (only if there are plenty of other seeders for the cached files. This could also cause race issue 
        // where all nodes with a cached file delete it at same time)
        
        Log::debug("Seeding key: %s", nodeToSeed.getRequestHash()->toString().c_str());
        DhtKey dhtKey(*nodeToSeed.getRequestHash());
        
        node.listen(dhtKey.getNewDataListenerKey(), [this, nodeToSeed](const shared_ptr<Value> &value)
        {
            Log::debug("Seed: New data listener received data...");
            const Hash requestHash(value->data.data(), value->data.size());
            if(requestHash == *nodeToSeed.getRequestHash())
                return true;
                //return listenCreateData(value, requestHash, encryptionKey);
            else
                return listenAddData(value, requestHash, nodeToSeed.getRequestHash(), nodeToSeed.getNodeEncryptionKey());
        });
        
        u8 responseKey[OPENDHT_INFOHASH_LEN];
        randombytes_buf(responseKey, OPENDHT_INFOHASH_LEN);
        
        // TODO: If this response key is spammed, generate a new one.
        node.listen(InfoHash(responseKey, OPENDHT_INFOHASH_LEN), [this, nodeToSeed](const shared_ptr<Value> &value)
        {
            const Hash requestHash(value->data.data(), value->data.size());
            if(requestHash == *nodeToSeed.getRequestHash())
                return listenCreateData(value, requestHash, nodeToSeed.getNodeEncryptionKey());
            else
                return listenAddData(value, requestHash, nodeToSeed.getRequestHash(), nodeToSeed.getNodeEncryptionKey());
        });
        
        // TODO: Before listening on this key, we should check how many remote peers are also providing this data.
        // This is to prevent too many peers from responding to a request to get old data.
        node.listen(dhtKey.getRequestOldDataKey(), [this, nodeToSeed](const shared_ptr<Value> &value)
        {
            Log::debug("Request: Got request to send old data");
            try
            {
                sibs::SafeDeserializer deserializer(value->data.data(), value->data.size());
                u64 dataStartTimestamp = deserializer.extract<u64>();
                u8 requestResponseKey[OPENDHT_INFOHASH_LEN];
                deserializer.extract(requestResponseKey, OPENDHT_INFOHASH_LEN);
                
                auto requestedData = databaseStorage.getStorage(*nodeToSeed.getRequestHash());
                if(!requestedData)
                {
                    Log::warn("No data found for hash %s, unable to serve peer", nodeToSeed.getRequestHash()->toString().c_str());
                    return true;
                }
                
                InfoHash requestResponseInfoHash(requestResponseKey, OPENDHT_INFOHASH_LEN);
                
                if(dataStartTimestamp == 0)
                {
                    Log::debug("Request: Sent create packet to requesting peer");
                    node.put(requestResponseInfoHash, Value((u8*)requestedData->data.data, requestedData->data.size), [](bool ok)
                    {
                        if(!ok)
                            Log::warn("Failed to put response for old data for 'create' data");
                    });
                }
                
                for(auto requestedObject : requestedData->objects)
                {
                    node.put(requestResponseInfoHash, Value((u8*)requestedObject->data.data, requestedObject->data.size), [](bool ok)
                    {
                        if(!ok)
                            Log::warn("Failed to put response for old data for 'add' data");
                    });
                }
            }
            catch (sibs::DeserializeException &e)
            {
                Log::warn("Failed to deserialize 'get old data' request: %s", e.what());
            }
            return true;
        });
        
        sibs::SafeSerializer serializer;
        serializer.add((u64)0); // Timestamp in microseconds, fetch data newer than this. // TODO: Get timestamp from database storage
        serializer.add(responseKey, OPENDHT_INFOHASH_LEN);
        node.put(dhtKey.getRequestOldDataKey(), Value(serializer.getBuffer().data(), serializer.getBuffer().size()), [](bool ok)
        {
            if(!ok)
                Log::warn("Failed to put request to get old data");
        });

        //node.listen(CREATE_DATA_HASH, bind(&Database::listenCreateData, this, _1));
        //node.listen(ADD_DATA_HASH, bind(&Database::listenAddData, this, _1));
    }

    unique_ptr<DatabaseCreateResponse> Database::create(const string &ownerName, const std::string &ownerPlainPassword, const string &nodeName)
    {
        // TODO: Should this be declared static? is there any difference in behavior/performance?
        boost::uuids::random_generator uuidGen;
        auto adminGroupId = uuidGen();
        auto adminGroup = new Group("administrator", adminGroupId.data, ADMIN_PERMISSION);
        LocalUser *nodeAdminUser = LocalUser::create(Signature::KeyPair(), ownerName, adminGroup, ownerPlainPassword);
        
        // Header
        sibs::SafeSerializer serializer;
        serializer.add(DATABASE_CREATE_PACKET_STRUCTURE_VERSION); // Packet structure version
        // TODO: Append fractions to get real microseconds time
        u64 timestampMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull;
        serializer.add(timestampMicroseconds);
        serializer.add((u8*)nodeAdminUser->getPublicKey().getData(), PUBLIC_KEY_NUM_BYTES);
        serializer.add(adminGroupId.data, adminGroupId.size());
        
        // Encrypted body
        sibs::SafeSerializer encryptedSerializer;
        assert(nodeAdminUser->getName().size() <= 255);
        encryptedSerializer.add((u8)nodeAdminUser->getName().size());
        encryptedSerializer.add((u8*)nodeAdminUser->getName().data(), nodeAdminUser->getName().size());
        assert(nodeName.size() <= 255);
        encryptedSerializer.add((u8)nodeName.size());
        encryptedSerializer.add((u8*)nodeName.data(), nodeName.size());
        
        try
        {
            Encryption encryptedBody(DataView(encryptedSerializer.getBuffer().data(), encryptedSerializer.getBuffer().size()));
            DataView requestData = combine(serializer, encryptedBody);
            shared_ptr<Hash> hashRequestKey = make_shared<Hash>(requestData.data, requestData.size);
            databaseStorage.createStorage(*hashRequestKey, adminGroup, timestampMicroseconds, (const u8*)requestData.data, requestData.size);

            string nodeNameCopy(nodeName);
            DatabaseCreateNodeRequest createNodeRequest(hashRequestKey.get(), timestampMicroseconds, nodeAdminUser, move(nodeNameCopy));
            if(onCreateNodeCallbackFunc)
                onCreateNodeCallbackFunc(createNodeRequest);

            stagedCreateObjects.emplace_back(make_unique<StagedObject>(requestData, hashRequestKey));
            
            assert(encryptedBody.getKey().size == ENCRYPTION_KEY_BYTE_SIZE);
            auto key = make_shared<OwnedMemory>(new char[encryptedBody.getKey().size], encryptedBody.getKey().size);
            memcpy(key->data, encryptedBody.getKey().data, encryptedBody.getKey().size);
            return make_unique<DatabaseCreateResponse>(nodeAdminUser, move(key), hashRequestKey);
        }
        catch (EncryptionException &e)
        {
            throw DatabaseCreateException("Failed to encrypt data for 'create' request");
        }
    }

    void Database::addData(const DatabaseNode &nodeInfo, LocalUser *userToPerformActionWith, DataView dataToAdd)
    {
        sibs::SafeSerializer serializer;
        serializer.add(DATABASE_ADD_PACKET_STRUCTURE_VERSION);
        // TODO: Append fractions to get real microseconds time
        u64 timestampMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull;
        serializer.add(timestampMicroseconds);
        serializer.add(DatabaseOperation::ADD_DATA);
        
        DataView encryptionKey(nodeInfo.getNodeEncryptionKey()->data, ENCRYPTION_KEY_BYTE_SIZE);
        Encryption encryptedBody(dataToAdd, DataView(), encryptionKey);
        DataView requestData = combine(serializer, encryptedBody);
        string signedRequestData = userToPerformActionWith->getPrivateKey().sign(requestData);
        free(requestData.data);
        DataView stagedAddObject = combine(userToPerformActionWith->getPublicKey(), signedRequestData);
        Hash requestDataHash(stagedAddObject.data, stagedAddObject.size);
        databaseStorage.appendStorage(*nodeInfo.getRequestHash(), requestDataHash, userToPerformActionWith, timestampMicroseconds, (u8*)stagedAddObject.data, stagedAddObject.size);

        DatabaseAddNodeRequest addNodeRequest(&*nodeInfo.getRequestHash(), &requestDataHash, timestampMicroseconds, userToPerformActionWith, dataToAdd);
        if(onAddNodeCallbackFunc)
            onAddNodeCallbackFunc(addNodeRequest);

        stagedAddObjects.emplace_back(make_unique<StagedObject>(stagedAddObject, nodeInfo.getRequestHash()));
    }
    
    Group* getGroupWithRightsToAddUserToGroup(const vector<Group*> &groups, Group *groupToAddUserTo)
    {
        for(auto group : groups)
        {
            const auto &groupPermission = group->getPermission();
            if(groupPermission.getFlag(PermissionType::ADD_USER_LOWER_LEVEL) && groupPermission.getPermissionLevel() < groupToAddUserTo->getPermission().getPermissionLevel())
            {
                return group;
            }
            else if(groupPermission.getFlag(PermissionType::ADD_USER_SAME_LEVEL) && groupPermission.getPermissionLevel() == groupToAddUserTo->getPermission().getPermissionLevel())
            {
                return group;
            }
        }
        return nullptr;
    }
    
    void Database::addUser(const DatabaseNode &nodeInfo, LocalUser *userToPerformActionWith, const string &userToAddName, const Signature::PublicKey &userToAddPublicKey, Group *groupToAddUserTo)
    {
        auto groupWithAddUserRights = getGroupWithRightsToAddUserToGroup(userToPerformActionWith->getGroups(), groupToAddUserTo);
        if(!groupWithAddUserRights)
        {
            string errMsg = "The user ";
            errMsg += userToPerformActionWith->getName();
            errMsg += " does not belong to any group that is allowed to add an user to the group ";
            errMsg += groupToAddUserTo->getName();
            throw PermissionDeniedException(errMsg);
        }
        
        sibs::SafeSerializer serializer;
        serializer.add(DATABASE_ADD_PACKET_STRUCTURE_VERSION);
        // TODO: Append fractions to get real microseconds time
        u64 timestampMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull;
        serializer.add(timestampMicroseconds);
        serializer.add(DatabaseOperation::ADD_USER);
        
        assert(userToAddName.size() <= 255);
        serializer.add((u8)userToAddName.size());
        serializer.add((u8*)userToAddName.data(), userToAddName.size());
        serializer.add((u8*)userToAddPublicKey.getData(), PUBLIC_KEY_NUM_BYTES);
        serializer.add((uint8_t*)groupToAddUserTo->getId().data, groupToAddUserTo->getId().size);
        
        DataView requestData { serializer.getBuffer().data(), serializer.getBuffer().size() };
        string signedRequestData = userToPerformActionWith->getPrivateKey().sign(requestData);
        DataView stagedAddObject = combine(userToPerformActionWith->getPublicKey(), signedRequestData);
        Hash requestDataHash(stagedAddObject.data, stagedAddObject.size);
        databaseStorage.appendStorage(*nodeInfo.getRequestHash(), requestDataHash, userToPerformActionWith, timestampMicroseconds, (u8*)stagedAddObject.data, stagedAddObject.size);
        auto userToAdd = RemoteUser::create(userToAddPublicKey, userToAddName, groupToAddUserTo);
        databaseStorage.addUser(*nodeInfo.getRequestHash(), userToAdd);

        DatabaseAddUserRequest addUserRequest(&*nodeInfo.getRequestHash(), &requestDataHash, timestampMicroseconds, userToPerformActionWith, userToAdd, groupToAddUserTo);
        if(onAddUserCallbackFunc)
            onAddUserCallbackFunc(addUserRequest);

        stagedAddObjects.emplace_back(make_unique<StagedObject>(stagedAddObject, nodeInfo.getRequestHash()));
    }

    void Database::commit()
    {
        // TODO: Combine staged objects into one object for efficiency.

        try
        {
            Log::debug("Num objects to create: %zu", stagedCreateObjects.size());
            for(const auto &stagedObject : stagedCreateObjects)
            {
                commitStagedCreateObject(stagedObject);
            }
            
            Log::debug("Num objects to add: %zu", stagedAddObjects.size());
            for(const auto &stagedObject : stagedAddObjects)
            {
                commitStagedAddObject(stagedObject);
            }
        }
        catch (exception &e)
        {
            // TODO: Add rollback
            Log::error("Failed to commit, reason: %s", e.what());
        }
        
        for(const auto &stagedObject : stagedCreateObjects)
        {
            free(stagedObject->data.data);
        }
        stagedCreateObjects.clear();
        
        for(const auto &stagedObject : stagedAddObjects)
        {
            free(stagedObject->data.data);
        }
        stagedAddObjects.clear();
        
        // TODO: Add node.listen here to get notified when remote peers got the commit, then we can say we can return
    }

    void Database::commitStagedCreateObject(const unique_ptr<StagedObject> &stagedObject)
    {
        DhtKey dhtKey(*stagedObject->requestKey);
        Value createDataValue((u8*)stagedObject->data.data, stagedObject->data.size);
        node.put(dhtKey.getNewDataListenerKey(), move(createDataValue), [](bool ok)
        {
            // TODO: Handle failure to put data
            if(!ok)
                Log::warn("Failed to put: %s, what to do?", "commitStagedCreateObject");
        }/* TODO: How to make this work?, time_point(), false*/);
    }

    void Database::commitStagedAddObject(const unique_ptr<StagedObject> &stagedObject)
    {
        DhtKey dhtKey(*stagedObject->requestKey);
        Value createDataValue((u8*)stagedObject->data.data, stagedObject->data.size);
        node.put(dhtKey.getNewDataListenerKey(), move(createDataValue), [](bool ok)
        {
            // TODO: Handle failure to put data
            if(!ok)
                Log::warn("Failed to put: %s, what to do?", "commitStagedAddObject");
        }/* TODO: How to make this work?, time_point(), false*/);
    }

    ntp::NtpTimestamp Database::getSyncedTimestampUtc() const
    {
        assert(timestampSynced);
        ntp::NtpTimestamp timestamp;
        timestamp.seconds = time(nullptr) - timeOffset;
        timestamp.fractions = 0; // TODO: Set this
        return timestamp;
    }

    void Database::deserializeCreateRequest(const shared_ptr<dht::Value> &value, const Hash &hash, const shared_ptr<OwnedMemory> encryptionKey)
    {
        sibs::SafeDeserializer deserializer(value->data.data(), value->data.size());
        u16 packetStructureVersion = deserializer.extract<u16>();
        if(packetStructureVersion != DATABASE_CREATE_PACKET_STRUCTURE_VERSION)
        {
            string errMsg = "Received 'create' request with packet structure version ";
            errMsg += to_string(packetStructureVersion);
            errMsg += ", but our packet structure version is ";
            errMsg += to_string(DATABASE_CREATE_PACKET_STRUCTURE_VERSION);
            throw sibs::DeserializeException(errMsg);
        }
        
        u64 creationDate = deserializer.extract<u64>();
        // TODO: Append fractions to get real microseconds time
        u64 timestampMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull;
        if(creationDate > timestampMicroseconds)
            throw sibs::DeserializeException("Packet is from the future");
        
        char creatorPublicKeyRaw[PUBLIC_KEY_NUM_BYTES];
        deserializer.extract((u8*)creatorPublicKeyRaw, PUBLIC_KEY_NUM_BYTES);
        Signature::PublicKey userPublicKey(creatorPublicKeyRaw, PUBLIC_KEY_NUM_BYTES);
        
        uint8_t adminGroupId[16];
        deserializer.extract(adminGroupId, 16);
        
        if(deserializer.getSize() < ENCRYPTION_NONCE_BYTE_SIZE)
            throw sibs::DeserializeException("Unsigned encrypted body is too small (unable to extract nonce)");
        
        auto adminGroup = new Group("administrator", adminGroupId, ADMIN_PERMISSION);
        // TODO: Username is encrypted, we dont know it... unless we have encryption key, in which case we should modify the user name and set it
        auto creatorUser = RemoteUser::create(userPublicKey, "ENCRYPTED USER NAME", adminGroup);
        databaseStorage.createStorage(hash, adminGroup, creationDate, value->data.data(), value->data.size());
        
        u8 nonce[ENCRYPTION_NONCE_BYTE_SIZE];
        deserializer.extract(nonce, ENCRYPTION_NONCE_BYTE_SIZE);
        
        DataView dataToDecrypt((void*)deserializer.getBuffer(), deserializer.getSize());
        Decryption decryptedBody(dataToDecrypt, DataView(nonce, ENCRYPTION_NONCE_BYTE_SIZE), DataView(encryptionKey->data, ENCRYPTION_KEY_BYTE_SIZE));
        sibs::SafeDeserializer bodyDeserializer((const u8*)decryptedBody.getDecryptedText().data, decryptedBody.getDecryptedText().size);
        
        u8 creatorNameLength = bodyDeserializer.extract<u8>();
        string creatorName; // TODO: Add this user name to storage added above
        creatorName.resize(creatorNameLength);
        bodyDeserializer.extract((u8*)&creatorName[0], creatorNameLength);
        
        u8 nameLength = bodyDeserializer.extract<u8>();
        string name;
        name.resize(nameLength);
        bodyDeserializer.extract((u8*)&name[0], nameLength);
        
        Log::debug("Got create object, name: %s", name.c_str());
        DatabaseCreateNodeRequest createNodeRequest(&hash, creationDate, creatorUser, move(name));
        if(onCreateNodeCallbackFunc)
            onCreateNodeCallbackFunc(createNodeRequest);
    }
    
    bool isUserAllowedToAddData(const User *user)
    {
        for(Group *group : user->getGroups())
        {
            if(group->getPermission().getFlag(PermissionType::ADD_DATA))
                return true;
        }
        return false;
    }

    void Database::deserializeAddRequest(const shared_ptr<dht::Value> &value, const Hash &requestDataHash, const std::shared_ptr<Hash> &nodeHash, const shared_ptr<OwnedMemory> encryptionKey)
    {
        sibs::SafeDeserializer deserializer(value->data.data(), value->data.size());
        char creatorPublicKeyRaw[PUBLIC_KEY_NUM_BYTES];
        deserializer.extract((u8*)creatorPublicKeyRaw, PUBLIC_KEY_NUM_BYTES);
        Signature::PublicKey creatorPublicKey(creatorPublicKeyRaw, PUBLIC_KEY_NUM_BYTES);
        
        DataView signedData((void*)deserializer.getBuffer(), deserializer.getSize());
        string unsignedData = creatorPublicKey.unsign(signedData);
        sibs::SafeDeserializer deserializerUnsigned((u8*)unsignedData.data(), unsignedData.size());
        
        u16 packetStructureVersion = deserializerUnsigned.extract<u16>();
        if(packetStructureVersion != DATABASE_CREATE_PACKET_STRUCTURE_VERSION)
        {
            string errMsg = "Received 'create' request with packet structure version ";
            errMsg += to_string(packetStructureVersion);
            errMsg += ", but our packet structure version is ";
            errMsg += to_string(DATABASE_CREATE_PACKET_STRUCTURE_VERSION);
            throw sibs::DeserializeException(errMsg);
        }
        
        u64 creationDate = deserializerUnsigned.extract<u64>();
        // TODO: Append fractions to get real microseconds time
        u64 timestampMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull;
        if(creationDate > timestampMicroseconds)
            throw sibs::DeserializeException("Packet is from the future");
        
        DatabaseOperation operation = deserializerUnsigned.extract<DatabaseOperation>();
#if 0
        const Hash *node = databaseStorage.getNodeByUserPublicKey(creatorPublicKey);
        if(!node)
        {
            // The user (public key) could belong to a node but we might not have retrieved the node info yet since data may
            // not be retrieved in order.
            // Data in quarantine is processed when 'create' packet is received or removed after 60 seconds
            databaseStorage.addToQuarantine(requestDataHash, creatorPublicKey, creationDate, value->data.data(), value->data.size());
            throw RequestQuarantineException();
        }
#endif
        auto creatorUser = databaseStorage.getUserByPublicKey(*nodeHash, creatorPublicKey);
        // TODO: Verify there isn't already data with same timestamp for this node. Same for quarantine.
        // TODO: We might receive 'add' data packet before 'create'. If that happens, we should put it in quarantine and process it later.
        databaseStorage.appendStorage(*nodeHash, requestDataHash, creatorUser, creationDate, value->data.data(), value->data.size());
        
        if(operation == DatabaseOperation::ADD_DATA)
        {
            if(deserializerUnsigned.getSize() < ENCRYPTION_NONCE_BYTE_SIZE)
                throw sibs::DeserializeException("Unsigned encrypted body is too small (unable to extract nonce)");
            
            u8 nonce[ENCRYPTION_NONCE_BYTE_SIZE];
            deserializerUnsigned.extract(nonce, ENCRYPTION_NONCE_BYTE_SIZE);
            DataView dataToDecrypt((void*)deserializerUnsigned.getBuffer(), deserializerUnsigned.getSize());
            Decryption decryptedBody(dataToDecrypt, DataView(nonce, ENCRYPTION_NONCE_BYTE_SIZE), DataView(encryptionKey->data, ENCRYPTION_KEY_BYTE_SIZE));
            
            if(!isUserAllowedToAddData(creatorUser))
            {
                // TODO: User might have permission to perform operation, but we haven't got the packet that adds user to the group with the permission,
                // or we haven't received the packet that modifies group with the permission to perform the operation.
                // This also means that an user can be in a group that has permission to perform the operation and then later be removed from it,
                // and remote peers would accept our request to perform operation if they haven't received the operation that removes the user from the group.
                // How to handle this?
                string errMsg = "User ";
                errMsg += creatorUser->getName();
                errMsg += " is not allowed to perform the operation: ";
                errMsg += to_string((u8)operation);
                throw PermissionDeniedException(errMsg);
            }
            
            Log::debug("Got add object, timestamp: %zu, data: %.*s", creationDate, decryptedBody.getDecryptedText().size, decryptedBody.getDecryptedText().data);
            const DatabaseAddNodeRequest addNodeRequest(&*nodeHash, &requestDataHash, creationDate, creatorUser, decryptedBody.getDecryptedText());
            if(onAddNodeCallbackFunc)
                onAddNodeCallbackFunc(addNodeRequest);
        }
        else if(operation == DatabaseOperation::ADD_USER)
        { 
            u8 nameLength = deserializerUnsigned.extract<u8>();
            string name;
            name.resize(nameLength);
            deserializerUnsigned.extract((u8*)&name[0], nameLength);
            
            char userToAddPublicKeyRaw[PUBLIC_KEY_NUM_BYTES];
            deserializerUnsigned.extract((u8*)userToAddPublicKeyRaw, PUBLIC_KEY_NUM_BYTES);
            Signature::PublicKey userToAddPublicKey(userToAddPublicKeyRaw, PUBLIC_KEY_NUM_BYTES);
            
            uint8_t groupId[16];
            deserializerUnsigned.extract(groupId, 16);
            
            auto group = databaseStorage.getGroupById(*nodeHash, groupId);
            if(group)
            {
                auto user = RemoteUser::create(userToAddPublicKey, name, group);
                // TODO: What if we receive packets in wrong order? (maliciously or non-maliciously). You would be able to register a user to a group with given name
                // and further registration would be dropped (even if that is the correct one)
                if(!databaseStorage.addUser(*nodeHash, user)) return;
                
                auto creatorUserGroupWithRights = getGroupWithRightsToAddUserToGroup(creatorUser->getGroups(), group);
                if(!creatorUserGroupWithRights)
                {
                    // TODO: User might have permission to perform operation, but we haven't got the packet that adds user to the group with the permission,
                    // or we haven't received the packet that modifies group with the permission to perform the operation.
                    // This also means that an user can be in a group that has permission to perform the operation and then later be removed from it,
                    // and remote peers would accept our request to perform operation if they haven't received the operation that removes the user from the group.
                    // How to handle this?
                    string errMsg = "User ";
                    errMsg += creatorUser->getName();
                    errMsg += " is not allowed to perform the operation: ";
                    errMsg += to_string((u8)operation);
                    throw PermissionDeniedException(errMsg);
                }
                
                Log::debug("Got add user object, timestamp: %zu, user added: %.*s", creationDate, nameLength, name.c_str());
                DatabaseAddUserRequest addUserRequest(&*nodeHash, &requestDataHash, creationDate, creatorUser, user, group);
                if(onAddUserCallbackFunc)
                    onAddUserCallbackFunc(addUserRequest);
            }
            else
            {
                throw sibs::DeserializeException("TODO: Add to quarantine? You can receive ADD_USER packet before you receive ADD_GROUP");
            }
        }
        else
        {
            string errMsg = "Got unexpected operation: ";
            errMsg += to_string((u8)operation);
            throw sibs::DeserializeException(errMsg);
        }  
    }

    bool Database::listenCreateData(shared_ptr<dht::Value> value, const Hash &hash, const shared_ptr<OwnedMemory> encryptionKey)
    {
        Log::debug("Got create data");
        try
        {
            if(databaseStorage.getStorage(hash))
                throw DatabaseStorageAlreadyExists("Create request hash is equal to hash already in storage (duplicate data?)");
            deserializeCreateRequest(value, hash, encryptionKey);
        }
        catch (exception &e)
        {
            Log::warn("Failed to deserialize 'create' request: %s", e.what());
        }
        return true;
    }

    bool Database::listenAddData(shared_ptr<dht::Value> value, const Hash &requestDataHash, const std::shared_ptr<Hash> nodeHash, const shared_ptr<OwnedMemory> encryptionKey)
    {
        Log::debug("Got add data");
        try
        {
            deserializeAddRequest(value, requestDataHash, nodeHash, encryptionKey);
            //Log::debug("Got add object, timestamp: %zu", addObject.timestamp);
        }
        catch (RequestQuarantineException &e)
        {
            Log::warn("Request was put in quarantine, will be processed later");
        }
        catch (exception &e)
        {
            Log::warn("Failed to deserialize 'add' request: %s", e.what());
        }
        return true;
    }

    void Database::setOnCreateNodeCallback(function<void(const DatabaseCreateNodeRequest&)> callbackFunc)
    {
        onCreateNodeCallbackFunc = callbackFunc;
    }

    void Database::setOnAddNodeCallback(function<void(const DatabaseAddNodeRequest&)> callbackFunc)
    {
        onAddNodeCallbackFunc = callbackFunc;
    }

    void Database::setOnAddUserCallback(function<void(const DatabaseAddUserRequest&)> callbackFunc)
    {
        onAddUserCallbackFunc = callbackFunc;
    }
    
    DatabaseStorage& Database::getStorage()
    {
        return databaseStorage;
    }
}