#include "../include/odhtdb/Database.hpp" #include "../include/odhtdb/Group.hpp" #include "../include/odhtdb/Encryption.hpp" #include "../include/odhtdb/DhtKey.hpp" #include "../include/odhtdb/bin2hex.hpp" #include "../include/odhtdb/Log.hpp" #include #include #include #include #include #include #include #include #include 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 odhtdb::u64 timeOffsetFraction = 0; static thread *ntpThread = nullptr; static bool timestampSynced = false; const int OPENDHT_INFOHASH_LEN = 20; namespace odhtdb { static boost::uuids::random_generator uuidGen; const u16 DATABASE_CREATE_PACKET_STRUCTURE_VERSION = 1; const u16 DATABASE_ADD_PACKET_STRUCTURE_VERSION = 1; const u16 DATABASE_REQUEST_OLD_DATA_STRUCTURE_VERSION = 1; class RequestQuarantineException : public runtime_error { public: RequestQuarantineException() : runtime_error("Request quarantine, will be processed later (can be real of fake request)") {} }; OwnedMemory 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 OwnedMemory(result, allocationSize); } OwnedMemory 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 OwnedMemory(result, allocationSize); } DatabaseCreateResponse::DatabaseCreateResponse(std::shared_ptr _nodeAdminKeyPair, std::shared_ptr _nodeAdminGroupId, shared_ptr _key, shared_ptr _hash) : nodeAdminKeyPair(_nodeAdminKeyPair), nodeAdminGroupId(_nodeAdminGroupId), key(_key), hash(_hash) { } const shared_ptr DatabaseCreateResponse::getNodeAdminKeyPair() const { return nodeAdminKeyPair; } const shared_ptr DatabaseCreateResponse::getNodeAdminGroupId() const { return nodeAdminGroupId; } const shared_ptr DatabaseCreateResponse::getNodeEncryptionKey() const { return key; } const shared_ptr DatabaseCreateResponse::getRequestHash() const { return hash; } Database::Database(const char *bootstrapNodeAddr, u16 port, const boost::filesystem::path &storageDir, DatabaseCallbackFuncs callbackFuncs) : onCreateNodeCallbackFunc(callbackFuncs.createNodeCallbackFunc), onAddNodeCallbackFunc(callbackFuncs.addNodeCallbackFunc), onAddUserCallbackFunc(callbackFuncs.addUserCallbackFunc), databaseStorage(this, storageDir), shuttingDown(false) { node.run(port , { /*.dht_config = */{ /*.node_config = */{ /*.node_id = */{}, /*.network = */0, /*.is_bootstrap = */true, /*.maintain_storage*/false }, /*.id = */databaseStorage.getIdentity() }, /*.threaded = */true, /*.proxy_server = */"", /*.push_node_id = */"" }); auto portStr = to_string(port); node.bootstrap(bootstrapNodeAddr, portStr.c_str()); const auto &remoteNodes = databaseStorage.getRemoteNodes(); if(!remoteNodes.empty()) node.bootstrap(remoteNodes); Log::debug("Connecting to bootstrap node (%s) and %u other known nodes that we have connected to previously", bootstrapNodeAddr, remoteNodes.size()); // TODO: Make this work for multiple threads initializing database at same time ++databaseCount; if(databaseCount == 1) { if(ntpThread) delete ntpThread; const int ntpFetchTimestampRetries = 5; ntpThread = new thread([]() { ntp::NtpClient ntpClient("pool.ntp.org", 10); while(databaseCount > 0) { int fetchRetryCounter = 0; while(fetchRetryCounter < ntpFetchTimestampRetries) { try { ntp::NtpTimestamp ntpTimestamp = ntpClient.getTimestamp(); struct timeval currentLocalTime; gettimeofday(¤tLocalTime, NULL); timeOffset = currentLocalTime.tv_sec - ntpTimestamp.seconds; timeOffsetFraction = currentLocalTime.tv_usec - ntpTimestamp.fractions; timestampSynced = true; break; } catch(ntp::NtpClientException &e) { Log::warn("Failed to sync clock with ntp server, reason: %s. Try #%d", e.what(), fetchRetryCounter); this_thread::sleep_for(500ms); } ++fetchRetryCounter; } if(fetchRetryCounter == ntpFetchTimestampRetries) throw ntp::NtpClientException("Failed to retrieve ntp timestamp after several retries"); this_thread::sleep_for(60s); } timestampSynced = false; }); ntpThread->detach(); } remoteNodesSaveThread = thread([this]() { int saveIntervalMs = 5000; // 5 sec const int sleepDurationMs = 200; while(!shuttingDown) { for(int i = 0; i < saveIntervalMs / sleepDurationMs; ++i) { this_thread::sleep_for(chrono::milliseconds(sleepDurationMs)); if(shuttingDown) return; } databaseStorage.setRemoteNodes(node.exportNodes()); saveIntervalMs = 30000; // 30 sec } }); } Database::~Database() { // TODO: Make this work for multiple threads removing database object at same time --databaseCount; shuttingDown = true; remoteNodesSaveThread.join(); node.join(); } struct ActionGap { u64 start; u64 range; }; void Database::seed(const DatabaseNode &nodeToSeed, DatabaseFetchOrder fetchOrder) { if(seedInfoMap.find(*nodeToSeed.getRequestHash()) != seedInfoMap.end()) { Log::warn("You are already seeding node %s, ignoring...", nodeToSeed.getRequestHash()->toString().c_str()); return; } DatabaseSeedInfo newSeedInfo; // 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). databaseStorage.setNodeDecryptionKey(*nodeToSeed.getRequestHash(), DataView(nodeToSeed.getNodeEncryptionKey()->data, nodeToSeed.getNodeEncryptionKey()->size)); Log::debug("Seeding key: %s", nodeToSeed.getRequestHash()->toString().c_str()); DhtKey dhtKey(*nodeToSeed.getRequestHash()); auto newDataListenerFuture = node.listen(dhtKey.getNewDataListenerKey(), [this, nodeToSeed](const shared_ptr &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()); }); newSeedInfo.newDataListenerFuture = make_shared>(move(newDataListenerFuture)); u8 responseKey[OPENDHT_INFOHASH_LEN]; randombytes_buf(responseKey, OPENDHT_INFOHASH_LEN); shared_ptr responseKeyShared = make_shared(responseKey, OPENDHT_INFOHASH_LEN);; newSeedInfo.reponseKeyInfoHash = responseKeyShared; // TODO: If this response key is spammed, generate a new one. auto responseKeyFuture = node.listen(*responseKeyShared, [this, nodeToSeed](const shared_ptr &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()); }); newSeedInfo.responseKeyFuture = make_shared>(move(responseKeyFuture)); // 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. auto requestOldDataListenerFuture = node.listen(dhtKey.getRequestOldDataKey(), [this, nodeToSeed, responseKeyShared](const shared_ptr &value) { Log::debug("Request: Got request to send old data"); try { static_assert(HASH_LEN == OPENDHT_INFOHASH_LEN, "Wrong hashlen size, did it change with opendht upgrade?"); sibs::SafeDeserializer deserializer(value->data.data(), value->data.size()); u16 requestStructureVersion = deserializer.extract(); if(requestStructureVersion != DATABASE_REQUEST_OLD_DATA_STRUCTURE_VERSION) { Log::warn("Request: structure is version %d but we are at version %d, ignoring request", requestStructureVersion, DATABASE_REQUEST_OLD_DATA_STRUCTURE_VERSION); return true; } InfoHash requestResponseInfoHash; deserializer.extract(requestResponseInfoHash.data(), OPENDHT_INFOHASH_LEN); if(*responseKeyShared == requestResponseInfoHash) return true; // We sent the request, ignore our own requests bool userWantsCreateNode = deserializer.extract() == 1; DatabaseFetchOrder fetchOrder = deserializer.extract(); if(userWantsCreateNode) { databaseStorage.fetchNodeRaw(*nodeToSeed.getRequestHash(), [this, requestResponseInfoHash](const DataView rawData) { Log::debug("Request: Sent create packet to requesting peer"); Value value((u8*)rawData.data, rawData.size); node.put(requestResponseInfoHash, move(value), [](bool ok) { if(!ok) Log::error("Failed to put response for old data for 'create' data"); }); }); } // TODO(Performance improvement): Sort actions by gap start and do a binary search to check if raw data is the packet the peer wants DataViewMap> actionGaps; while(!deserializer.empty()) { u8 userPublicKeyRaw[PUBLIC_KEY_NUM_BYTES]; deserializer.extract(userPublicKeyRaw, PUBLIC_KEY_NUM_BYTES); u64 actionGapStart = deserializer.extract(); u64 actionGapRange = deserializer.extract(); DataView userPublicKey(userPublicKeyRaw, PUBLIC_KEY_NUM_BYTES); actionGaps[userPublicKey].push_back({ actionGapStart, actionGapRange }); } // TODO(Performance improvement): Instead of sending several packets, combine them into one databaseStorage.fetchNodeAddDataRaw(*nodeToSeed.getRequestHash(), [this, requestResponseInfoHash, &actionGaps](const DataView rawData, const DataView creatorPublicKey, u64 actionCounter) { bool sendData = false; auto actionGapsIt = actionGaps.find(creatorPublicKey); if(actionGapsIt == actionGaps.end()) sendData = true; else { for(const auto &userActionGaps : actionGapsIt->second) { if(actionCounter >= userActionGaps.start && actionCounter <= userActionGaps.start + userActionGaps.range) { sendData = true; break; } } } if(!sendData) return; Value value((u8*)rawData.data, rawData.size); node.put(requestResponseInfoHash, move(value), [](bool ok) { if(!ok) Log::error("Failed to put response for old data for 'add' data"); }); }, fetchOrder); } catch (std::exception &e) { Log::warn("Failed while serving peer, error: %s", e.what()); } return true; }); newSeedInfo.requestOldDataListenerFuture = make_shared>(move(requestOldDataListenerFuture)); seedInfoMap[*nodeToSeed.getRequestHash()] = newSeedInfo; sibs::SafeSerializer serializer; serializer.add(DATABASE_REQUEST_OLD_DATA_STRUCTURE_VERSION); serializer.add(responseKey, OPENDHT_INFOHASH_LEN); bool iHaveCreateNode = databaseStorage.doesNodeExist(*nodeToSeed.getRequestHash()); serializer.add(iHaveCreateNode ? (u8)0 : (u8)1); serializer.add(fetchOrder); DataViewMap userLatestActionCounter; databaseStorage.fetchNodeUserActionGaps(*nodeToSeed.getRequestHash(), [&serializer, &userLatestActionCounter](const DataView userPublicKey, u64 actionGapStart, u64 actionGapRange) { serializer.add((const u8*)userPublicKey.data, PUBLIC_KEY_NUM_BYTES); serializer.add(actionGapStart); serializer.add(actionGapRange); userLatestActionCounter[userPublicKey] = std::max(userLatestActionCounter[userPublicKey], actionGapStart + actionGapRange); }); databaseStorage.fetchNodeUserLatestActionCounter(*nodeToSeed.getRequestHash(), [&userLatestActionCounter](const DataView userPublicKey, u64 latestActionCounter) { userLatestActionCounter[userPublicKey] = std::max(userLatestActionCounter[userPublicKey], latestActionCounter); }); for(auto userLatestActionCounterData : userLatestActionCounter) { // Public key serializer.add((const u8*)userLatestActionCounterData.first.data, PUBLIC_KEY_NUM_BYTES); // Latest action counter start serializer.add(userLatestActionCounterData.second); // Latest action counter range (infinite range, meaning we want all packets older than start (latest known packet by user)) serializer.add(~(u64)0ULL - userLatestActionCounterData.second); } Value requestValue(move(serializer.getBuffer())); node.put(dhtKey.getRequestOldDataKey(), move(requestValue), [](bool ok) { if(!ok) Log::warn("Failed to put request to get old data"); }); } void Database::stopSeeding(const Hash &nodeHash) { auto seedInfoIt = seedInfoMap.find(nodeHash); if(seedInfoIt != seedInfoMap.end()) { // TODO: Verify if doing get on listener future stalls program forever... Opendht documentation is not clear on this DhtKey dhtKey(nodeHash); node.cancelListen(dhtKey.getNewDataListenerKey(), seedInfoIt->second.newDataListenerFuture->get()); node.cancelListen(dhtKey.getRequestOldDataKey(), seedInfoIt->second.requestOldDataListenerFuture->get()); node.cancelListen(*seedInfoIt->second.reponseKeyInfoHash, seedInfoIt->second.responseKeyFuture->get()); seedInfoMap.erase(seedInfoIt); } } void Database::loadNode(const Hash &nodeHash, DatabaseLoadOrder loadOrder) { databaseStorage.loadNode(nodeHash, loadOrder); } unique_ptr Database::create() { shared_ptr creatorKeyPair = make_shared(); auto adminGroupId = uuidGen(); assert(adminGroupId.size() == GROUP_ID_LENGTH); // Header sibs::SafeSerializer serializer; serializer.add(DATABASE_CREATE_PACKET_STRUCTURE_VERSION); u64 timestampCombined = getSyncedTimestampUtc().getCombined(); serializer.add(timestampCombined); serializer.add((u8*)creatorKeyPair->getPublicKey().getData(), PUBLIC_KEY_NUM_BYTES); serializer.add(adminGroupId.data, adminGroupId.size()); try { unsigned char *encryptionKeyRaw = new unsigned char[ENCRYPTION_KEY_BYTE_SIZE]; Encryption::generateKey(encryptionKeyRaw); shared_ptr encryptionKey = make_shared(encryptionKeyRaw, ENCRYPTION_KEY_BYTE_SIZE); shared_ptr hashRequestKey = make_shared(serializer.getBuffer().data(), serializer.getBuffer().size()); databaseStorage.setNodeDecryptionKey(*hashRequestKey, DataView(encryptionKey->data, encryptionKey->size)); databaseStorage.createStorage(*hashRequestKey, creatorKeyPair->getPublicKey(), DataView(adminGroupId.data, adminGroupId.size()), timestampCombined, (const u8*)serializer.getBuffer().data(), serializer.getBuffer().size()); DhtKey dhtKey(*hashRequestKey); Value createDataValue(move(serializer.getBuffer())); 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?", "Database::create"); }); shared_ptr adminGroupIdResponse = make_shared(new u8[GROUP_ID_LENGTH], GROUP_ID_LENGTH); memcpy(adminGroupIdResponse->data, adminGroupId.data, GROUP_ID_LENGTH); return make_unique(creatorKeyPair, adminGroupIdResponse, encryptionKey, hashRequestKey); } catch (EncryptionException &e) { throw DatabaseCreateException("Failed to encrypt data for 'create' request"); } } void Database::addData(const DatabaseNode &nodeInfo, const Signature::KeyPair &userToPerformActionWith, DataView dataToAdd) { if(!databaseStorage.isUserAllowedToAddDataInNode(*nodeInfo.getRequestHash(), userToPerformActionWith.getPublicKey())) { // 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 += userToPerformActionWith.getPublicKey().toString(); errMsg += " is not allowed to perform the operation: ADD_USER"; throw PermissionDeniedException(errMsg); } sibs::SafeSerializer serializer; serializer.add(DATABASE_ADD_PACKET_STRUCTURE_VERSION); u64 timestampCombined = getSyncedTimestampUtc().getCombined(); serializer.add(timestampCombined); serializer.add(DatabaseOperation::ADD_DATA); u64 newActionCounter = databaseStorage.getUserActionCounter(*nodeInfo.getRequestHash(), userToPerformActionWith.getPublicKey()) + 1; serializer.add(newActionCounter); DataView encryptionKey(nodeInfo.getNodeEncryptionKey()->data, ENCRYPTION_KEY_BYTE_SIZE); Encryption encryptedBody(dataToAdd, DataView(), encryptionKey); OwnedMemory requestData = combine(serializer, encryptedBody); string signedRequestData = userToPerformActionWith.getPrivateKey().sign(requestData.getView()); OwnedMemory stagedAddObject = combine(userToPerformActionWith.getPublicKey(), signedRequestData); Hash requestDataHash(stagedAddObject.data, stagedAddObject.size); DataView encryptedDataView((char*)requestData.data + serializer.getBuffer().size(), requestData.size - serializer.getBuffer().size()); databaseStorage.appendStorage(*nodeInfo.getRequestHash(), requestDataHash, DatabaseOperation::ADD_DATA, newActionCounter, userToPerformActionWith.getPublicKey(), timestampCombined, (u8*)stagedAddObject.data, stagedAddObject.size, encryptedDataView); DhtKey dhtKey(requestDataHash); Value addDataValue((u8*)stagedAddObject.data, stagedAddObject.size); node.put(dhtKey.getNewDataListenerKey(), move(addDataValue), [](bool ok) { // TODO: Handle failure to put data if(!ok) Log::warn("Failed to put: %s, what to do?", "Database::addData"); }); } void Database::addUser(const DatabaseNode &nodeInfo, const Signature::KeyPair &userToPerformActionWith, const Signature::PublicKey &userToAddPublicKey, const DataView &groupToAddUserTo) { sibs::SafeSerializer serializer; serializer.add(DATABASE_ADD_PACKET_STRUCTURE_VERSION); u64 timestampCombined = getSyncedTimestampUtc().getCombined(); serializer.add(timestampCombined); serializer.add(DatabaseOperation::ADD_USER); u64 newActionCounter = databaseStorage.getUserActionCounter(*nodeInfo.getRequestHash(), userToPerformActionWith.getPublicKey()) + 1; serializer.add(newActionCounter); usize additionalDataOffset = serializer.getBuffer().size(); serializer.add((u8*)userToAddPublicKey.getData(), PUBLIC_KEY_NUM_BYTES); serializer.add((uint8_t*)groupToAddUserTo.data, groupToAddUserTo.size); auto padding = uuidGen(); assert(padding.size() == 16); serializer.add(padding.data, padding.size()); DataView requestData { serializer.getBuffer().data(), serializer.getBuffer().size() }; string signedRequestData = userToPerformActionWith.getPrivateKey().sign(requestData); OwnedMemory stagedAddObject = combine(userToPerformActionWith.getPublicKey(), signedRequestData); Hash requestDataHash(stagedAddObject.data, stagedAddObject.size); DataView additionalDataView((void*)(static_cast(requestData.data) + additionalDataOffset), requestData.size - additionalDataOffset); databaseStorage.appendStorage(*nodeInfo.getRequestHash(), requestDataHash, DatabaseOperation::ADD_USER, newActionCounter, userToPerformActionWith.getPublicKey(), timestampCombined, (u8*)stagedAddObject.data, stagedAddObject.size, additionalDataView); DhtKey dhtKey(requestDataHash); Value addDataValue((u8*)stagedAddObject.data, stagedAddObject.size); node.put(dhtKey.getNewDataListenerKey(), move(addDataValue), [](bool ok) { // TODO: Handle failure to put data if(!ok) Log::warn("Failed to put: %s, what to do?", "Database::addUser"); }); } ntp::NtpTimestamp Database::getSyncedTimestampUtc() const { while(!timestampSynced) { this_thread::sleep_for(10ms); } struct timeval currentLocalTime; gettimeofday(¤tLocalTime, NULL); ntp::NtpTimestamp timestamp; timestamp.seconds = currentLocalTime.tv_sec - timeOffset; timestamp.fractions = currentLocalTime.tv_usec - timeOffsetFraction; return timestamp; } void Database::deserializeCreateRequest(const shared_ptr &value, const Hash &hash, const shared_ptr encryptionKey) { sibs::SafeDeserializer deserializer(value->data.data(), value->data.size()); u16 packetStructureVersion = deserializer.extract(); 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(); /* // TODO: This doesn't seem to work right now, fix it auto currentTimestamp = getSyncedTimestampUtc(); if(creationDate > currentTimestamp.getCombined()) { auto creationDateTimestamp = ntp::NtpTimestamp::fromCombined(creationDate); string errMsg = "Packet is from the future. Packet creation time: "; errMsg += to_string((double)creationDateTimestamp.seconds + creationDateTimestamp.getFractionAsSeconds()); errMsg += ", current time: "; errMsg += to_string((double)currentTimestamp.seconds + currentTimestamp.getFractionAsSeconds()); throw sibs::DeserializeException(errMsg); } */ 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[GROUP_ID_LENGTH]; deserializer.extract(adminGroupId, GROUP_ID_LENGTH); if(deserializer.getSize() < ENCRYPTION_NONCE_BYTE_SIZE) throw sibs::DeserializeException("Unsigned encrypted body is too small (unable to extract nonce)"); databaseStorage.createStorage(hash, userPublicKey, DataView(adminGroupId, GROUP_ID_LENGTH), creationDate, value->data.data(), value->data.size()); } void Database::deserializeAddRequest(const shared_ptr &value, const Hash &requestDataHash, const std::shared_ptr &nodeHash, const shared_ptr 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(); 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(); auto currentTimestamp = getSyncedTimestampUtc(); /* // TODO: This doesn't seem to work right now, fix it if(creationDate > currentTimestamp.getCombined()) { auto creationDateTimestamp = ntp::NtpTimestamp::fromCombined(creationDate); string errMsg = "Packet is from the future. Packet creation time: "; errMsg += to_string((double)creationDateTimestamp.seconds + creationDateTimestamp.getFractionAsSeconds()); errMsg += ", current time: "; errMsg += to_string((double)currentTimestamp.seconds + currentTimestamp.getFractionAsSeconds()); throw sibs::DeserializeException(errMsg); } */ DatabaseOperation operation = deserializerUnsigned.extract(); u64 newActionCounter = deserializerUnsigned.extract(); DataView additionalDataView((void*)deserializerUnsigned.getBuffer(), deserializerUnsigned.getSize()); databaseStorage.appendStorage(*nodeHash, requestDataHash, operation, newActionCounter, creatorPublicKey, creationDate, value->data.data(), value->data.size(), additionalDataView); } bool Database::listenCreateData(shared_ptr value, const Hash &hash, const shared_ptr encryptionKey) { Log::debug("Got create data"); try { // This check is here to reduce processing, it doesn't matter much if the packet bypasses this, // the database has constraint to deal with this in multi-threaded way if(databaseStorage.doesNodeExist(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 value, const Hash &requestDataHash, const std::shared_ptr nodeHash, const shared_ptr encryptionKey) { Log::debug("Got add data"); try { // This check is here to reduce processing, it doesn't matter much if the packet bypasses this, // the database has constraint to deal with this in multi-threaded way if(databaseStorage.doesDataExist(requestDataHash)) throw DatabaseStorageAlreadyExists("Add data request hash is equal to hash already in storage (duplicate data?)"); deserializeAddRequest(value, requestDataHash, nodeHash, encryptionKey); } 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; } bool Database::doesStoredUserExist(const string &username) const { return databaseStorage.doesStoredUserExist(username); } void Database::storeUserWithoutNodes(const string &username, const string &password) { return databaseStorage.storeUserWithoutNodes(username, password); } void Database::storeUserPasswordEncrypted(const Hash &nodeHash, const string &username, const string &password, const Signature::KeyPair &keyPair) { return databaseStorage.storeUserPasswordEncrypted(nodeHash, username, password, keyPair); } vector Database::getStoredUserNodeDataDecrypted(const string &username, const string &password) { return databaseStorage.getStoredUserNodeDataDecrypted(username, password); } vector Database::getUserGroups(const Hash &nodeHash, const Signature::PublicKey &userPublicKey) const { return databaseStorage.getUserGroups(nodeHash, userPublicKey); } }