/* */ #include "DHTMessageFactoryImpl.h" #include #include #include "LogFactory.h" #include "DlAbortEx.h" #include "Data.h" #include "Dictionary.h" #include "List.h" #include "DHTNode.h" #include "DHTRoutingTable.h" #include "DHTPingMessage.h" #include "DHTPingReplyMessage.h" #include "DHTFindNodeMessage.h" #include "DHTFindNodeReplyMessage.h" #include "DHTGetPeersMessage.h" #include "DHTGetPeersReplyMessage.h" #include "DHTAnnouncePeerMessage.h" #include "DHTAnnouncePeerReplyMessage.h" #include "DHTUnknownMessage.h" #include "DHTConnection.h" #include "DHTMessageDispatcher.h" #include "DHTPeerAnnounceStorage.h" #include "DHTTokenTracker.h" #include "DHTMessageCallback.h" #include "PeerMessageUtil.h" #include "BtRuntime.h" #include "Util.h" #include "Peer.h" #include "Logger.h" #include "StringFormat.h" namespace aria2 { DHTMessageFactoryImpl::DHTMessageFactoryImpl(): _logger(LogFactory::getInstance()) {} DHTMessageFactoryImpl::~DHTMessageFactoryImpl() {} SharedHandle DHTMessageFactoryImpl::getRemoteNode(const unsigned char* id, const std::string& ipaddr, uint16_t port) const { SharedHandle node = _routingTable->getNode(id, ipaddr, port); if(node.isNull()) { node.reset(new DHTNode(id)); node->setIPAddress(ipaddr); node->setPort(port); } return node; } static const Dictionary* getDictionary(const Dictionary* d, const std::string& key) { const Dictionary* c = dynamic_cast(d->get(key)); if(c) { return c; } else { throw DlAbortEx (StringFormat("Malformed DHT message. Missing %s", key.c_str()).str()); } } static const Data* getData(const Dictionary* d, const std::string& key) { const Data* c = dynamic_cast(d->get(key)); if(c) { return c; } else { throw DlAbortEx (StringFormat("Malformed DHT message. Missing %s", key.c_str()).str()); } } static const Data* getData(const List* l, size_t index) { const Data* c = dynamic_cast(l->getList()[index]); if(c) { return c; } else { throw DlAbortEx (StringFormat("Malformed DHT message. element[%u] is not Data.", index).str()); } } static const List* getList(const Dictionary* d, const std::string& key) { const List* l = dynamic_cast(d->get(key)); if(l) { return l; } else { throw DlAbortEx (StringFormat("Malformed DHT message. Missing %s", key.c_str()).str()); } } void DHTMessageFactoryImpl::validateID(const Data* id) const { if(id->getLen() != DHT_ID_LENGTH) { throw DlAbortEx (StringFormat("Malformed DHT message. Invalid ID length. Expected:%d, Actual:%d", DHT_ID_LENGTH, id->getLen()).str()); } } void DHTMessageFactoryImpl::validatePort(const Data* i) const { if(!i->isNumber()) { throw DlAbortEx (StringFormat("Malformed DHT message. Invalid port=%s", Util::toHex(i->toString()).c_str()).str()); } uint32_t port = i->toInt(); if(UINT16_MAX < port) { throw DlAbortEx (StringFormat("Malformed DHT message. Invalid port=%u", port).str()); } } SharedHandle DHTMessageFactoryImpl::createQueryMessage(const Dictionary* d, const std::string& ipaddr, uint16_t port) { const Data* q = getData(d, DHTQueryMessage::Q); const Data* t = getData(d, DHTMessage::T); const Data* y = getData(d, DHTMessage::Y); const Dictionary* a = getDictionary(d, DHTQueryMessage::A); if(y->toString() != DHTQueryMessage::Q) { throw DlAbortEx("Malformed DHT message. y != q"); } const Data* id = getData(getDictionary(d, DHTQueryMessage::A), DHTMessage::ID); validateID(id); SharedHandle remoteNode = getRemoteNode(id->getData(), ipaddr, port); std::string messageType = q->toString(); std::string transactionID = t->toString(); if(messageType == DHTPingMessage::PING) { return createPingMessage(remoteNode, transactionID); } else if(messageType == DHTFindNodeMessage::FIND_NODE) { const Data* targetNodeID = getData(a, DHTFindNodeMessage::TARGET_NODE); validateID(targetNodeID); return createFindNodeMessage(remoteNode, targetNodeID->getData(), transactionID); } else if(messageType == DHTGetPeersMessage::GET_PEERS) { const Data* infoHash = getData(a, DHTGetPeersMessage::INFO_HASH); validateID(infoHash); return createGetPeersMessage(remoteNode, infoHash->getData(), transactionID); } else if(messageType == DHTAnnouncePeerMessage::ANNOUNCE_PEER) { const Data* infoHash = getData(a, DHTAnnouncePeerMessage::INFO_HASH); validateID(infoHash); const Data* port = getData(a, DHTAnnouncePeerMessage::PORT); validatePort(port); const Data* token = getData(a, DHTAnnouncePeerMessage::TOKEN); return createAnnouncePeerMessage(remoteNode, infoHash->getData(), static_cast(port->toInt()), token->toString(), transactionID); } else { throw DlAbortEx (StringFormat("Unsupported message type: %s", messageType.c_str()).str()); } } SharedHandle DHTMessageFactoryImpl::createResponseMessage(const std::string& messageType, const Dictionary* d, const std::string& ipaddr, uint16_t port) { const Data* t = getData(d, DHTMessage::T); const Data* y = getData(d, DHTMessage::Y); if(y->toString() == DHTUnknownMessage::E) { // for now, just report error message arrived and throw exception. const List* e = getList(d, DHTUnknownMessage::E); if(e->getList().size() == 2) { _logger->info("Received Error DHT message. code=%s, msg=%s", Util::urlencode(getData(e, 0)->toString()).c_str(), Util::urlencode(getData(e, 1)->toString()).c_str()); } else { _logger->debug("e doesn't have 2 elements."); } throw DlAbortEx("Received Error DHT message."); } else if(y->toString() != DHTResponseMessage::R) { throw DlAbortEx (StringFormat("Malformed DHT message. y != r: y=%s", Util::urlencode(y->toString()).c_str()).str()); } const Dictionary* r = getDictionary(d, DHTResponseMessage::R); const Data* id = getData(r, DHTMessage::ID); validateID(id); SharedHandle remoteNode = getRemoteNode(id->getData(), ipaddr, port); std::string transactionID = t->toString(); if(messageType == DHTPingReplyMessage::PING) { return createPingReplyMessage(remoteNode, id->getData(), transactionID); } else if(messageType == DHTFindNodeReplyMessage::FIND_NODE) { return createFindNodeReplyMessage(remoteNode, d, transactionID); } else if(messageType == DHTGetPeersReplyMessage::GET_PEERS) { const List* values = dynamic_cast(r->get(DHTGetPeersReplyMessage::VALUES)); if(values) { return createGetPeersReplyMessageWithValues(remoteNode, d, transactionID); } else { const Data* nodes = dynamic_cast(r->get(DHTGetPeersReplyMessage::NODES)); if(nodes) { return createGetPeersReplyMessageWithNodes(remoteNode, d, transactionID); } else { throw DlAbortEx("Malformed DHT message: missing nodes/values"); } } } else if(messageType == DHTAnnouncePeerReplyMessage::ANNOUNCE_PEER) { return createAnnouncePeerReplyMessage(remoteNode, transactionID); } else { throw DlAbortEx (StringFormat("Unsupported message type: %s", messageType.c_str()).str()); } } void DHTMessageFactoryImpl::setCommonProperty(const SharedHandle& m) { m->setConnection(_connection); m->setMessageDispatcher(_dispatcher); m->setRoutingTable(_routingTable); WeakHandle factory(this); m->setMessageFactory(factory); } SharedHandle DHTMessageFactoryImpl::createPingMessage(const SharedHandle& remoteNode, const std::string& transactionID) { SharedHandle m(new DHTPingMessage(_localNode, remoteNode, transactionID)); setCommonProperty(m); return m; } SharedHandle DHTMessageFactoryImpl::createPingReplyMessage(const SharedHandle& remoteNode, const unsigned char* id, const std::string& transactionID) { SharedHandle m(new DHTPingReplyMessage(_localNode, remoteNode, id, transactionID)); setCommonProperty(m); return m; } SharedHandle DHTMessageFactoryImpl::createFindNodeMessage(const SharedHandle& remoteNode, const unsigned char* targetNodeID, const std::string& transactionID) { SharedHandle m(new DHTFindNodeMessage(_localNode, remoteNode, targetNodeID, transactionID)); setCommonProperty(m); return m; } SharedHandle DHTMessageFactoryImpl::createFindNodeReplyMessage(const SharedHandle& remoteNode, const std::deque >& closestKNodes, const std::string& transactionID) { SharedHandle m(new DHTFindNodeReplyMessage(_localNode, remoteNode, transactionID)); m->setClosestKNodes(closestKNodes); setCommonProperty(m); return m; } std::deque > DHTMessageFactoryImpl::extractNodes(const unsigned char* src, size_t length) { if(length%26 != 0) { throw DlAbortEx("Nodes length is not multiple of 26"); } std::deque > nodes; for(size_t offset = 0; offset < length; offset += 26) { SharedHandle node(new DHTNode(src+offset)); std::pair addr = PeerMessageUtil::unpackcompact(src+offset+DHT_ID_LENGTH); if(addr.first.empty()) { continue; } node->setIPAddress(addr.first); node->setPort(addr.second); nodes.push_back(node); } return nodes; } SharedHandle DHTMessageFactoryImpl::createFindNodeReplyMessage(const SharedHandle& remoteNode, const Dictionary* d, const std::string& transactionID) { const Data* nodesData = getData(getDictionary(d, DHTResponseMessage::R), DHTFindNodeReplyMessage::NODES); std::deque > nodes = extractNodes(nodesData->getData(), nodesData->getLen()); return createFindNodeReplyMessage(remoteNode, nodes, transactionID); } SharedHandle DHTMessageFactoryImpl::createGetPeersMessage(const SharedHandle& remoteNode, const unsigned char* infoHash, const std::string& transactionID) { SharedHandle m(new DHTGetPeersMessage(_localNode, remoteNode, infoHash, transactionID)); m->setPeerAnnounceStorage(_peerAnnounceStorage); m->setTokenTracker(_tokenTracker); setCommonProperty(m); return m; } SharedHandle DHTMessageFactoryImpl::createGetPeersReplyMessageWithNodes(const SharedHandle& remoteNode, const Dictionary* d, const std::string& transactionID) { const Dictionary* r = getDictionary(d, DHTResponseMessage::R); const Data* nodesData = getData(r, DHTGetPeersReplyMessage::NODES); std::deque > nodes = extractNodes(nodesData->getData(), nodesData->getLen()); const Data* token = getData(r, DHTGetPeersReplyMessage::TOKEN); return createGetPeersReplyMessage(remoteNode, nodes, token->toString(), transactionID); } SharedHandle DHTMessageFactoryImpl::createGetPeersReplyMessage(const SharedHandle& remoteNode, const std::deque >& closestKNodes, const std::string& token, const std::string& transactionID) { SharedHandle m (new DHTGetPeersReplyMessage(_localNode, remoteNode, token, transactionID)); m->setClosestKNodes(closestKNodes); setCommonProperty(m); return m; } SharedHandle DHTMessageFactoryImpl::createGetPeersReplyMessageWithValues(const SharedHandle& remoteNode, const Dictionary* d, const std::string& transactionID) { const Dictionary* r = getDictionary(d, DHTResponseMessage::R); const List* valuesList = getList(r, DHTGetPeersReplyMessage::VALUES); std::deque > peers; for(std::deque::const_iterator i = valuesList->getList().begin(); i != valuesList->getList().end(); ++i) { const Data* data = dynamic_cast(*i); if(data && data->getLen() == 6) { std::pair addr = PeerMessageUtil::unpackcompact(data->getData()); PeerHandle peer(new Peer(addr.first, addr.second)); peers.push_back(peer); } } const Data* token = getData(r, DHTGetPeersReplyMessage::TOKEN); return createGetPeersReplyMessage(remoteNode, peers, token->toString(), transactionID); } SharedHandle DHTMessageFactoryImpl::createGetPeersReplyMessage(const SharedHandle& remoteNode, const std::deque >& values, const std::string& token, const std::string& transactionID) { SharedHandle m(new DHTGetPeersReplyMessage(_localNode, remoteNode, token, transactionID)); m->setValues(values); setCommonProperty(m); return m; } SharedHandle DHTMessageFactoryImpl::createAnnouncePeerMessage(const SharedHandle& remoteNode, const unsigned char* infoHash, uint16_t tcpPort, const std::string& token, const std::string& transactionID) { SharedHandle m (new DHTAnnouncePeerMessage(_localNode, remoteNode, infoHash, tcpPort, token, transactionID)); m->setPeerAnnounceStorage(_peerAnnounceStorage); m->setTokenTracker(_tokenTracker); setCommonProperty(m); return m; } SharedHandle DHTMessageFactoryImpl::createAnnouncePeerReplyMessage(const SharedHandle& remoteNode, const std::string& transactionID) { SharedHandle m (new DHTAnnouncePeerReplyMessage(_localNode, remoteNode, transactionID)); setCommonProperty(m); return m; } SharedHandle DHTMessageFactoryImpl::createUnknownMessage(const unsigned char* data, size_t length, const std::string& ipaddr, uint16_t port) { SharedHandle m (new DHTUnknownMessage(_localNode, data, length, ipaddr, port)); return m; } void DHTMessageFactoryImpl::setRoutingTable(const WeakHandle& routingTable) { _routingTable = routingTable; } void DHTMessageFactoryImpl::setConnection(const WeakHandle& connection) { _connection = connection; } void DHTMessageFactoryImpl::setMessageDispatcher(const WeakHandle& dispatcher) { _dispatcher = dispatcher; } void DHTMessageFactoryImpl::setPeerAnnounceStorage(const WeakHandle& storage) { _peerAnnounceStorage = storage; } void DHTMessageFactoryImpl::setTokenTracker(const WeakHandle& tokenTracker) { _tokenTracker = tokenTracker; } void DHTMessageFactoryImpl::setLocalNode(const SharedHandle& localNode) { _localNode = localNode; } } // namespace aria2