Multicast DNS implementation (initial)

net/dns/mdns_client_impl.cc

+// Copyright (c) 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "net/dns/mdns_client_impl.h"
+
+#include "base/bind.h"
+#include "base/message_loop_proxy.h"
+#include "base/stl_util.h"
+#include "base/time/default_clock.h"
+#include "net/base/dns_util.h"
+#include "net/base/net_errors.h"
+#include "net/base/net_log.h"
+#include "net/base/rand_callback.h"
+#include "net/dns/dns_protocol.h"
+#include "net/udp/datagram_socket.h"
+
+namespace net {
+
+static const char kMDnsMulticastGroupIPv4[] = "224.0.0.251";
+static const char kMDnsMulticastGroupIPv6[] = "FF02::FB";
+
+static const unsigned kMDnsTransactionNoResultsSeconds = 3;
+
+MDnsClientImpl::Core::Core(MDnsClientImpl* client,
+                           MDnsConnectionFactory* connection_factory)
+    : client_(client), connection_(connection_factory->CreateConnection(this)) {
+}
+
+MDnsClientImpl::Core::~Core() {
+  cleanup_callback_.Cancel();
+  STLDeleteValues(&listeners_);
+}
+
+bool MDnsClientImpl::Core::Init() {
+  return connection_->Init() == OK;
+}
+
+bool MDnsClientImpl::Core::SendQuery(uint16 rrtype, std::string name) {
+  std::string name_dns;
+  if (!DNSDomainFromDot(name, &name_dns))
+    return false;
+
+  // TODO(noamsml): Remove RD flag from query and add known-answer suppression.
+  DnsQuery query(0, name_dns, rrtype);
+
+  return connection_->Send(query.io_buffer(), query.io_buffer()->size()) == OK;
+}
+
+void MDnsClientImpl::Core::HandlePacket(DnsResponse* response,
+                                        int bytes_read) {
+  unsigned offset;
+
+  if (!response->InitParseWithoutQuery(bytes_read)) {
+    LOG(WARNING) << "Could not understand an mDNS packet.";
+    return;  // Message is unreadable.
+  }
+
+  // TODO(noamsml): duplicate query suppression.
+  if (!(response->flags() & dns_protocol::kFlagResponse)) {
+    return;  // Message is a query. ignore it.
+  }
+  DnsRecordParser parser = response->Parser();
+  unsigned answer_count = response->answer_count() +
+      response->additional_answer_count();
+
+  for (unsigned i = 0; i < answer_count; i++) {
+    offset = parser.GetOffset();
+    scoped_ptr<const RecordParsed> scoped_record = RecordParsed::CreateFrom(
+        &parser, base::Time::Now());
+
+    if (!scoped_record) {
+      LOG(WARNING) << "Could not understand an mDNS record.";
+
+      if (offset == parser.GetOffset()) {
+        LOG(WARNING) << "Abandoned parsing the rest of the packet.";
+        return;  // The parser did not advance, abort reading the packet.
+      } else {
+        continue;  // We may be able to extract other records from the packet.
+      }
+    }
+
+    if ((scoped_record->klass() & dns_protocol::kMDnsClassMask) !=
+        dns_protocol::kClassIN) {
+      LOG(WARNING) << "Received an mDNS record with non-IN class. Ignoring.";
+      continue;  // Ignore all records not in the IN class.
+    }
+
+    // We want to retain a copy of the record pointer for updating listeners
+    // but we are passing ownership to the cache.
+    const RecordParsed* record = scoped_record.get();
+    MDnsCache::UpdateType update = cache_.UpdateDnsRecord(scoped_record.Pass());
+
+    // Cleanup time may have changed.
+    ScheduleCleanup(cache_.next_expiration());
+
+    if (update != MDnsCache::NoChange) {
+      MDnsUpdateType update_external;
+
+      switch (update) {
+        case MDnsCache::RecordAdded:
+          update_external = kMDnsRecordAdded;
+          break;
+        case MDnsCache::RecordChanged:
+          update_external = kMDnsRecordChanged;
+          break;
+        case MDnsCache::NoChange:
+          NOTREACHED();
+          // Dummy assignment to suppress compiler warning.
+          update_external = kMDnsRecordChanged;
+          break;
+      }
+
+      AlertListeners(update_external,
+                     ListenerKey(record->type(), record->name()), record);
+      // Alert listeners listening only for rrtype and not for name.
+      AlertListeners(update_external, ListenerKey(record->type(), ""), record);
+    }
+  }
+}
+
+void MDnsClientImpl::Core::OnConnectionError(int error) {
+  // TODO(noamsml): On connection error, recreate connection and flush cache.
+}
+
+void MDnsClientImpl::Core::AlertListeners(
+    MDnsUpdateType update_type,
+    const ListenerKey& key,
+    const RecordParsed* record) {
+  ListenerMap::iterator listener_map_iterator = listeners_.find(key);
+  if (listener_map_iterator == listeners_.end()) return;
+
+  FOR_EACH_OBSERVER(MDnsListenerImpl, *listener_map_iterator->second,
+                    AlertDelegate(update_type, record));
+}
+
+void MDnsClientImpl::Core::AddListener(
+    MDnsListenerImpl* listener) {
+  ListenerKey key(listener->GetType(), listener->GetName());
+  std::pair<ListenerMap::iterator, bool> observer_insert_result =
+      listeners_.insert(
+          make_pair(key, static_cast<ObserverList<MDnsListenerImpl>*>(NULL)));
+
+  // If an equivalent key does not exist, actually create the observer list.
+  if (observer_insert_result.second) {
+    observer_insert_result.first->second = new ObserverList<MDnsListenerImpl>();
+  }
+
+  ObserverList<MDnsListenerImpl>* observer_list =
+      observer_insert_result.first->second;
+
+  observer_list->AddObserver(listener);
+}
+
+void MDnsClientImpl::Core::RemoveListener(MDnsListenerImpl* listener) {
+  ListenerKey key(listener->GetType(), listener->GetName());
+  ListenerMap::iterator observer_list_iterator = listeners_.find(key);
+
+  DCHECK(observer_list_iterator != listeners_.end());
+  DCHECK(observer_list_iterator->second->HasObserver(listener));
+
+  observer_list_iterator->second->RemoveObserver(listener);
+
+  // Remove the observer list from the map if it is empty
+  if (observer_list_iterator->second->size() == 0) {
+    delete observer_list_iterator->second;
+    listeners_.erase(observer_list_iterator);
+  }
+}
+
+void MDnsClientImpl::Core::ScheduleCleanup(base::Time cleanup) {
+  // Cleanup is already scheduled, no need to do anything.
+  if (cleanup == scheduled_cleanup_) return;
+  scheduled_cleanup_ = cleanup;
+
+  // This line has the effect of cancelling the previously scheduled cleanup.
+  cleanup_callback_.Reset(base::Bind(
+      &MDnsClientImpl::Core::DoCleanup, base::Unretained(this)));
+
+  // cleanup == base::Time means no cleanup necessary.
+  if (cleanup != base::Time()) {
+    base::MessageLoop::current()->PostDelayedTask(
+        FROM_HERE,
+        cleanup_callback_.callback(),
+        cleanup - base::Time::Now());
+  }
+}
+
+void MDnsClientImpl::Core::DoCleanup() {
+  cache_.CleanupRecords(base::Time::Now(), base::Bind(
+      &MDnsClientImpl::Core::OnRecordRemoved, base::Unretained(this)));
+
+  ScheduleCleanup(cache_.next_expiration());
+}
+
+void MDnsClientImpl::Core::OnRecordRemoved(
+    const RecordParsed* record) {
+  AlertListeners(kMDnsRecordRemoved,
+                 ListenerKey(record->type(), record->name()), record);
+  // Alert listeners listening only for rrtype and not for name.
+  AlertListeners(kMDnsRecordRemoved, ListenerKey(record->type(), ""),
+                 record);
+}
+
+void MDnsClientImpl::Core::QueryCache(
+    uint16 rrtype, const std::string& name,
+    std::vector<const RecordParsed*>* records) const {
+  cache_.FindDnsRecords(rrtype, name, records, base::Time::Now());
+}
+
+MDnsClientImpl::MDnsClientImpl()
+    : listen_refs_(0), connection_factory_owned_(
+          new MDnsConnectionImplFactory()) {
+  connection_factory_ = connection_factory_owned_.get();
+}
+
+MDnsClientImpl::MDnsClientImpl(MDnsConnectionFactory* connection_factory)
+    : listen_refs_(0), connection_factory_(connection_factory) {
+}
+
+MDnsClientImpl::~MDnsClientImpl() {
+}
+
+bool MDnsClientImpl::AddListenRef() {
+  if (!core_.get()) {
+    core_.reset(new Core(this, connection_factory_));
+    if (!core_->Init()) {
+      core_.reset();
+      return false;
+    }
+  }
+  listen_refs_++;
+  return true;
+}
+
+void MDnsClientImpl::SubtractListenRef() {
+  listen_refs_--;
+  if (listen_refs_ == 0) {
+    base::MessageLoop::current()->PostTask(FROM_HERE, base::Bind(
+        &MDnsClientImpl::Shutdown, base::Unretained(this)));
+  }
+}
+
+void MDnsClientImpl::Shutdown() {
+  // We need to check that new listeners haven't been created.
+  if (listen_refs_ == 0) {
+    core_.reset();
+  }
+}
+
+bool MDnsClientImpl::IsListeningForTests() {
+  return core_.get() != NULL;
+}
+
+scoped_ptr<MDnsListener> MDnsClientImpl::CreateListener(
+    uint16 rrtype,
+    const std::string& name,
+    MDnsListener::Delegate* delegate) {
+  return scoped_ptr<net::MDnsListener>(
+      new MDnsListenerImpl(rrtype, name, delegate, this));
+}
+
+scoped_ptr<MDnsTransaction> MDnsClientImpl::CreateTransaction(
+    uint16 rrtype,
+    const std::string& name,
+    int flags,
+    const MDnsTransaction::ResultCallback& callback) {
+  return scoped_ptr<MDnsTransaction>(
+      new MDnsTransactionImpl(rrtype, name, flags, callback, this));
+}
+
+MDnsListenerImpl::MDnsListenerImpl(
+    uint16 rrtype,
+    const std::string& name,
+    MDnsListener::Delegate* delegate,
+    MDnsClientImpl* client)
+    : rrtype_(rrtype), name_(name), client_(client), delegate_(delegate),
+      started_(false) {
+}
+
+bool MDnsListenerImpl::Start() {
+  DCHECK(!started_);
+
+  if (!client_->AddListenRef()) return false;
+  started_ = true;
+
+  client_->core()->AddListener(this);
+
+  return true;
+}
+
+MDnsListenerImpl::~MDnsListenerImpl() {
+  if (started_) {
+    client_->core()->RemoveListener(this);
+    client_->SubtractListenRef();
+  }
+}
+
+const std::string& MDnsListenerImpl::GetName() const {
+  return name_;
+}
+
+uint16 MDnsListenerImpl::GetType() const {
+  return rrtype_;
+}
+
+void MDnsListenerImpl::AlertDelegate(MDnsUpdateType update_type,
+                                     const RecordParsed* record) {
+  DCHECK(started_);
+  delegate_->OnRecordUpdate(update_type, record);
+}
+
+MDnsTransactionImpl::MDnsTransactionImpl(
+    uint16 rrtype,
+    const std::string& name,
+    int flags,
+    const MDnsTransaction::ResultCallback& callback,
+    MDnsClientImpl* client)
+    : rrtype_(rrtype), name_(name), callback_(callback), client_(client),
+      started_(false), flags_(flags) {
+  DCHECK((flags_ & kMDnsTransactionFlagMask) == flags_);
+  DCHECK(flags_ & kMDnsTransactionQueryCache ||
+         flags_ & kMDnsTransactionQueryNetwork);
+}
+
+MDnsTransactionImpl::~MDnsTransactionImpl() {
+  timeout_.Cancel();
+}
+
+bool MDnsTransactionImpl::Start() {
+  DCHECK(!started_);
+  started_ = true;
+  std::vector<const RecordParsed*> records;
+  base::WeakPtr<MDnsTransactionImpl> weak_this = AsWeakPtr();
+
+  if (flags_ & kMDnsTransactionQueryCache) {
+    if (client_->core()) {
+      client_->core()->QueryCache(rrtype_, name_, &records);
+      for (std::vector<const RecordParsed*>::iterator i = records.begin();
+           i != records.end() && weak_this; ++i) {
+        weak_this->TriggerCallback(kMDnsTransactionRecord, records.front());
+      }
+    }
+  }
+
+  if (!weak_this) return true;
+
+  if (is_active() && flags_ & kMDnsTransactionQueryNetwork) {
+    listener_ = client_->CreateListener(rrtype_, name_, this);
+    if (!listener_->Start()) return false;
+
+    DCHECK(client_->core());
+    if (!client_->core()->SendQuery(rrtype_, name_)) {
+      return false;
+    }
+
+    timeout_.Reset(base::Bind(&MDnsTransactionImpl::SignalTransactionOver,
+                              weak_this));
+    base::MessageLoop::current()->PostDelayedTask(
+        FROM_HERE,
+        timeout_.callback(),
+        base::TimeDelta::FromSeconds(kMDnsTransactionNoResultsSeconds));
+
+    return listener_.get() != NULL;
+  } else {
+    // If this is a cache only query, signal that the transaction is over
+    // immediately.
+    SignalTransactionOver();
+  }
+
+  return true;
+}
+
+const std::string& MDnsTransactionImpl::GetName() const {
+  return name_;
+}
+
+uint16 MDnsTransactionImpl::GetType() const {
+  return rrtype_;
+}
+
+void MDnsTransactionImpl::CacheRecordFound(const RecordParsed* record) {
+  DCHECK(started_);
+  OnRecordUpdate(kMDnsRecordAdded, record);
+}
+
+void MDnsTransactionImpl::TriggerCallback(MDnsTransactionResult result,
+                                          const RecordParsed* record) {
+  DCHECK(started_);
+  if (!is_active()) return;
+
+  // Ensure callback is run after touching all class state, so that
+  // the callback can delete the transaction.
+  MDnsTransaction::ResultCallback callback = callback_;
+
+  if (flags_ & kMDnsTransactionSingleResult) {
+    Reset();
+  }
+
+  callback.Run(result, record);
+}
+
+void MDnsTransactionImpl::Reset() {
+  callback_.Reset();
+  listener_.reset();
+  timeout_.Cancel();
+}
+
+void MDnsTransactionImpl::OnRecordUpdate(MDnsUpdateType update,
+                                         const RecordParsed* record) {
+  DCHECK(started_);
+  if (update ==  kMDnsRecordAdded || update == kMDnsRecordChanged) {
+    TriggerCallback(kMDnsTransactionRecord, record);
+  }
+}
+
+void MDnsTransactionImpl::SignalTransactionOver() {
+  DCHECK(started_);
+  base::WeakPtr<MDnsTransactionImpl> weak_this = AsWeakPtr();
+
+  if (flags_ & kMDnsTransactionSingleResult) {
+    TriggerCallback(kMDnsTransactionNoResults, NULL);
+  } else {
+    TriggerCallback(kMDnsTransactionDone, NULL);
+  }
+
+  if (weak_this) {
+    weak_this->Reset();
+  }
+}
+
+void MDnsTransactionImpl::OnNsecRecord(const std::string& name, unsigned type) {
+  // TODO(noamsml): NSEC records not yet implemented
+}
+
+void MDnsTransactionImpl::OnCachePurged() {
+  // TODO(noamsml): Cache purge situations not yet implemented
+}
+
+MDnsConnectionImpl::SocketHandler::SocketHandler(
+    MDnsConnectionImpl* connection, const IPEndPoint& multicast_addr)
+    : socket_owned_(new UDPServerSocket(NULL, NetLog::Source())),
+      socket_(socket_owned_.get()), connection_(connection),
+      response_(new DnsResponse(dns_protocol::kMaxMulticastSize)),
+      multicast_addr_(multicast_addr) {
+}
+
+MDnsConnectionImpl::SocketHandler::SocketHandler(
+    DatagramServerSocket* socket,
+    MDnsConnectionImpl* connection, const IPEndPoint& multicast_addr)
+    : socket_(socket), connection_(connection),
+      response_(new DnsResponse(dns_protocol::kMaxMulticastSize)),
+      multicast_addr_(multicast_addr) {
+}
+
+
+MDnsConnectionImpl::SocketHandler::~SocketHandler() {
+}
+
+int MDnsConnectionImpl::SocketHandler::Start() {
+  int rv = BindSocket();
+  if (rv != OK) {
+    return rv;
+  }
+
+  return DoLoop(0);
+}
+
+int MDnsConnectionImpl::SocketHandler::DoLoop(int rv) {
+  do {
+    if (rv > 0) {
+      connection_->OnDatagramReceived(socket_,
+                                      response_.get(), &recv_addr_, rv);
+    }
+
+    rv = socket_->RecvFrom(
+        response_->io_buffer(),
+        response_->io_buffer()->size(),
+        &recv_addr_,
+        base::Bind(&MDnsConnectionImpl::SocketHandler::OnDatagramReceived,
+                   base::Unretained(this)));
+  } while (rv > 0);
+
+  if (rv != ERR_IO_PENDING) {
+    return rv;
+  }
+
+  return OK;
+}
+
+void MDnsConnectionImpl::SocketHandler::OnDatagramReceived(int rv) {
+  if (rv >= OK) {
+    rv = DoLoop(rv);
+  }
+
+  if (rv != OK) {
+    connection_->OnError(this, rv);
+  }
+}
+
+int MDnsConnectionImpl::SocketHandler::Send(IOBuffer* buffer, unsigned size) {
+  return socket_->SendTo(
+      buffer, size, multicast_addr_,
+      base::Bind(&MDnsConnectionImpl::SocketHandler::SendDone,
+                 base::Unretained(this) ));
+}
+
+void MDnsConnectionImpl::SocketHandler::SendDone(int sent) {
+  // TODO(noamsml): Retry logic.
+}
+
+int MDnsConnectionImpl::SocketHandler::BindSocket() {
+  IPAddressNumber address_any;
+  address_any.resize(multicast_addr_.address().size(), 0);
+
+  IPEndPoint bind_endpoint(address_any, multicast_addr_.port());
+
+  socket_->AllowAddressReuse();
+  int rv = socket_->Listen(bind_endpoint);
+
+  if (rv < OK) return rv;
+
+  socket_->SetMulticastLoopbackMode(false);
+
+  return socket_->JoinGroup(multicast_addr_.address());
+}
+
+
+MDnsConnectionImpl::MDnsConnectionImpl(MDnsConnection::Delegate* delegate)
+    : socket_handler_ipv4_(this, GetIPEndPoint(kMDnsMulticastGroupIPv4,
+                                     dns_protocol::kDefaultPortMulticast)),
+      socket_handler_ipv6_(this, GetIPEndPoint(kMDnsMulticastGroupIPv6,
+                                     dns_protocol::kDefaultPortMulticast)),
+  delegate_(delegate) {
+}
+
+MDnsConnectionImpl::MDnsConnectionImpl(DatagramServerSocket* socket_ipv4,
+                                       DatagramServerSocket* socket_ipv6,
+                                       MDnsConnection::Delegate* delegate)
+    : socket_handler_ipv4_(socket_ipv4, this,
+                           GetIPEndPoint(kMDnsMulticastGroupIPv4,
+                                         dns_protocol::kDefaultPortMulticast)),
+      socket_handler_ipv6_(socket_ipv6, this,
+                           GetIPEndPoint(kMDnsMulticastGroupIPv6,
+                                         dns_protocol::kDefaultPortMulticast)),
+  delegate_(delegate) {
+}
+
+MDnsConnectionImpl::~MDnsConnectionImpl() {
+}
+
+int MDnsConnectionImpl::Init() {
+  int rv;
+
+  rv = socket_handler_ipv4_.Start();
+  if (rv != OK) return rv;
+  rv = socket_handler_ipv6_.Start();
+  if (rv != OK) return rv;
+
+  return true;
+}
+
+int MDnsConnectionImpl::Send(IOBuffer* buffer, unsigned size) {
+  int rv;
+
+  rv = socket_handler_ipv4_.Send(buffer, size);
+  if (rv < OK && rv != ERR_IO_PENDING) return rv;
+
+  rv = socket_handler_ipv6_.Send(buffer, size);
+  if (rv < OK && rv != ERR_IO_PENDING) return rv;
+
+  return OK;
+}
+
+void MDnsConnectionImpl::OnError(SocketHandler* loop,
+                                 int error) {
+  // TODO(noamsml): Specific handling of intermittent errors that can be handled
+  // in the connection.
+  delegate_->OnConnectionError(error);
+}
+
+IPEndPoint MDnsConnectionImpl::GetIPEndPoint(const char* address, int port) {
+  IPAddressNumber multicast_group_number;
+  bool success = ParseIPLiteralToNumber(address,
+                                        &multicast_group_number);
+  DCHECK(success);
+  return IPEndPoint(multicast_group_number, port);
+}
+
+void MDnsConnectionImpl::OnDatagramReceived(
+    DatagramServerSocket* socket,
+    DnsResponse* response,
+    IPEndPoint* recv_addr,
+    int bytes_read) {
+  // TODO(noamsml): More sophisticated error handling.
+  DCHECK_GT(bytes_read, 0);
+  delegate_->HandlePacket(response, bytes_read);
+}
+
+MDnsConnectionImplFactory::MDnsConnectionImplFactory() {
+}
+
+MDnsConnectionImplFactory::~MDnsConnectionImplFactory() {
+}
+
+scoped_ptr<MDnsConnection> MDnsConnectionImplFactory::CreateConnection(
+    MDnsConnection::Delegate* delegate) {
+  return scoped_ptr<MDnsConnection>(new MDnsConnectionImpl(delegate));
+}
+
+}  // namespace net