[Sync] Treat XMPP auth errors as sync auth errors

Add controls for XMPP authentication to XMPP test server.

Add integration test for XMPP auth errors.

Add some DVLOGs to help debugging.

BUG=38091
TEST=


Review URL: https://chromiumcodereview.appspot.com/10703035

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@171172 0039d316-1c4b-4281-b951-d872f2087c98

17 files changed, 263 insertions, 37 deletions

diff --git a/chrome/browser/sync/glue/sync_backend_host.cc b/chrome/browser/sync/glue/sync_backend_host.cc
index 315603c..2e275a2 100644
--- a/chrome/browser/sync/glue/sync_backend_host.cc
+++ b/chrome/browser/sync/glue/sync_backend_host.cc
@@ -1585,6 +1585,8 @@ void SyncBackendHost::HandleConnectionStatusChangeOnFrontendLoop(
 
   DCHECK_EQ(MessageLoop::current(), frontend_loop_);
 
+  DVLOG(1) << "Connection status changed: "
+           << syncer::ConnectionStatusToString(status);
   frontend_->OnConnectionStatusChange(status);
 }
 
diff --git a/chrome/browser/sync/profile_sync_service.cc b/chrome/browser/sync/profile_sync_service.cc
index d107d98..f84c02e 100644
--- a/chrome/browser/sync/profile_sync_service.cc
+++ b/chrome/browser/sync/profile_sync_service.cc
@@ -70,6 +70,7 @@
 #include "sync/js/js_arg_list.h"
 #include "sync/js/js_event_details.h"
 #include "sync/notifier/invalidator_registrar.h"
+#include "sync/notifier/invalidator_state.h"
 #include "sync/util/cryptographer.h"
 #include "ui/base/l10n/l10n_util.h"
 
@@ -938,7 +939,10 @@ AuthError ConnectionStatusToAuthError(
 
 void ProfileSyncService::OnConnectionStatusChange(
     syncer::ConnectionStatus status) {
-  UpdateAuthErrorState(ConnectionStatusToAuthError(status));
+  const GoogleServiceAuthError auth_error =
+      ConnectionStatusToAuthError(status);
+  DVLOG(1) << "Connection status change: " << auth_error.ToString();
+  UpdateAuthErrorState(auth_error);
 }
 
 void ProfileSyncService::OnStopSyncingPermanently() {
@@ -1876,6 +1880,10 @@ void ProfileSyncService::UpdateInvalidatorRegistrarState() {
   const syncer::InvalidatorState effective_state =
       backend_initialized_ ?
       invalidator_state_ : syncer::TRANSIENT_INVALIDATION_ERROR;
+  DVLOG(1) << "New invalidator state: "
+           << syncer::InvalidatorStateToString(invalidator_state_)
+           << ", effective state: "
+           << syncer::InvalidatorStateToString(effective_state);
   invalidator_registrar_->UpdateInvalidatorState(effective_state);
 }
 
diff --git a/chrome/browser/sync/profile_sync_service_harness.cc b/chrome/browser/sync/profile_sync_service_harness.cc
index 59e01a2..45cd86e 100644
--- a/chrome/browser/sync/profile_sync_service_harness.cc
+++ b/chrome/browser/sync/profile_sync_service_harness.cc
@@ -186,6 +186,19 @@ bool ProfileSyncServiceHarness::SetupSync(
     return false;
   }
 
+  // Make sure that initial sync wasn't blocked by a missing passphrase.
+  if (wait_state_ == SET_PASSPHRASE_FAILED) {
+    LOG(ERROR) << "A passphrase is required for decryption. Sync cannot proceed"
+                  " until SetDecryptionPassphrase is called.";
+    return false;
+  }
+
+  // Make sure that initial sync wasn't blocked by rejected credentials.
+  if (wait_state_ == CREDENTIALS_REJECTED) {
+    LOG(ERROR) << "Credentials were rejected. Sync cannot proceed.";
+    return false;
+  }
+
   // Choose the datatypes to be synced. If all datatypes are to be synced,
   // set sync_everything to true; otherwise, set it to false.
   bool sync_everything =
@@ -202,13 +215,6 @@ bool ProfileSyncServiceHarness::SetupSync(
     return false;
   }
 
-  // Make sure that a partner client hasn't already set an explicit passphrase.
-  if (wait_state_ == SET_PASSPHRASE_FAILED) {
-    LOG(ERROR) << "A passphrase is required for decryption. Sync cannot proceed"
-                  " until SetDecryptionPassphrase is called.";
-    return false;
-  }
-
   // Set an implicit passphrase for encryption if an explicit one hasn't already
   // been set. If an explicit passphrase has been set, immediately return false,
   // since a decryption passphrase is required.
@@ -237,6 +243,12 @@ bool ProfileSyncServiceHarness::SetupSync(
     return false;
   }
 
+  // Make sure that initial sync wasn't blocked by rejected credentials.
+  if (wait_state_ == CREDENTIALS_REJECTED) {
+    LOG(ERROR) << "Credentials were rejected. Sync cannot proceed.";
+    return false;
+  }
+
   // Indicate to the browser that sync setup is complete.
   service()->SetSyncSetupCompleted();
 
@@ -261,7 +273,7 @@ void ProfileSyncServiceHarness::SignalStateCompleteWithNextState(
 
 void ProfileSyncServiceHarness::SignalStateComplete() {
   if (waiting_for_status_change_)
-    MessageLoop::current()->Quit();
+    MessageLoop::current()->QuitWhenIdle();
 }
 
 bool ProfileSyncServiceHarness::RunStateChangeMachine() {
@@ -289,6 +301,12 @@ bool ProfileSyncServiceHarness::RunStateChangeMachine() {
         SignalStateCompleteWithNextState(SET_PASSPHRASE_FAILED);
         break;
       }
+      if (service()->GetAuthError().state() ==
+          GoogleServiceAuthError::INVALID_GAIA_CREDENTIALS) {
+        // Our credentials were rejected. Do not wait any more.
+        SignalStateCompleteWithNextState(CREDENTIALS_REJECTED);
+        break;
+      }
       break;
     }
     case WAITING_FOR_FULL_SYNC: {
diff --git a/chrome/browser/sync/profile_sync_service_harness.h b/chrome/browser/sync/profile_sync_service_harness.h
index 06421e9..83989a3 100644
--- a/chrome/browser/sync/profile_sync_service_harness.h
+++ b/chrome/browser/sync/profile_sync_service_harness.h
@@ -256,6 +256,9 @@ class ProfileSyncServiceHarness
     // The sync client needs a passphrase in order to decrypt data.
     SET_PASSPHRASE_FAILED,
 
+    // The sync client's credentials were rejected.
+    CREDENTIALS_REJECTED,
+
     // The sync client cannot reach the server.
     SERVER_UNREACHABLE,
 
diff --git a/chrome/browser/sync/test/integration/sync_errors_test.cc b/chrome/browser/sync/test/integration/sync_errors_test.cc
index 9dde47d..403ccb0 100644
--- a/chrome/browser/sync/test/integration/sync_errors_test.cc
+++ b/chrome/browser/sync/test/integration/sync_errors_test.cc
@@ -140,6 +140,8 @@ IN_PROC_BROWSER_TEST_F(SyncErrorTest,
       protocol_error.error_description);
 }
 
+// Trigger an auth error and make sure the sync client detects it when
+// trying to commit.
 IN_PROC_BROWSER_TEST_F(SyncErrorTest, AuthErrorTest) {
   ASSERT_TRUE(SetupSync()) << "SetupSync() failed.";
 
@@ -156,6 +158,19 @@ IN_PROC_BROWSER_TEST_F(SyncErrorTest, AuthErrorTest) {
             GetClient(0)->service()->GetAuthError().state());
 }
 
+// Trigger an XMPP auth error, and make sure sync treats it like any
+// other auth error.
+IN_PROC_BROWSER_TEST_F(SyncErrorTest, XmppAuthErrorTest) {
+  ASSERT_TRUE(SetupClients()) << "SetupClients() failed.";
+
+  TriggerXmppAuthError();
+
+  ASSERT_FALSE(GetClient(0)->SetupSync());
+
+  ASSERT_EQ(GoogleServiceAuthError::INVALID_GAIA_CREDENTIALS,
+            GetClient(0)->service()->GetAuthError().state());
+}
+
 // TODO(lipalani): Fix the typed_url dtc so this test case can pass.
 IN_PROC_BROWSER_TEST_F(SyncErrorTest, DISABLED_DisableDatatypeWhileRunning) {
   ASSERT_TRUE(SetupSync()) << "SetupSync() failed.";
diff --git a/chrome/browser/sync/test/integration/sync_test.cc b/chrome/browser/sync/test/integration/sync_test.cc
index 911d580..dedd759 100644
--- a/chrome/browser/sync/test/integration/sync_test.cc
+++ b/chrome/browser/sync/test/integration/sync_test.cc
@@ -734,6 +734,12 @@ void SyncTest::TriggerAuthError() {
   ui_test_utils::NavigateToURL(browser(), sync_server_.GetURL(path));
 }
 
+void SyncTest::TriggerXmppAuthError() {
+  ASSERT_TRUE(ServerSupportsErrorTriggering());
+  std::string path = "chromiumsync/xmppcred";
+  ui_test_utils::NavigateToURL(browser(), sync_server_.GetURL(path));
+}
+
 namespace {
 
 sync_pb::SyncEnums::ErrorType
diff --git a/chrome/browser/sync/test/integration/sync_test.h b/chrome/browser/sync/test/integration/sync_test.h
index 5702d78..e925106 100644
--- a/chrome/browser/sync/test/integration/sync_test.h
+++ b/chrome/browser/sync/test/integration/sync_test.h
@@ -194,6 +194,10 @@ class SyncTest : public InProcessBrowserTest {
   // this state until shut down.
   void TriggerAuthError();
 
+  // Triggers an XMPP auth error on the server.  Note the server will
+  // stay in this state until shut down.
+  void TriggerXmppAuthError();
+
   // Triggers a sync error on the server.
   //   error: The error the server is expected to return.
   //   frequency: Frequency with which the error is returned.
diff --git a/jingle/notifier/communicator/login.cc b/jingle/notifier/communicator/login.cc
index afb0152..8350d2f 100644
--- a/jingle/notifier/communicator/login.cc
+++ b/jingle/notifier/communicator/login.cc
@@ -59,6 +59,7 @@ void Login::StartConnection() {
 }
 
 void Login::UpdateXmppSettings(const buzz::XmppClientSettings& user_settings) {
+  DVLOG(1) << "XMPP settings updated";
   login_settings_.set_user_settings(user_settings);
 }
 
@@ -68,11 +69,13 @@ void Login::UpdateXmppSettings(const buzz::XmppClientSettings& user_settings) {
 // TODO(akalin): Add unit tests to enforce the behavior above.
 
 void Login::OnConnect(base::WeakPtr<buzz::XmppTaskParentInterface> base_task) {
+  DVLOG(1) << "Connected";
   ResetReconnectState();
   delegate_->OnConnect(base_task);
 }
 
 void Login::OnRedirect(const ServerInformation& redirect_server) {
+  DVLOG(1) << "Redirected";
   login_settings_.SetRedirectServer(redirect_server);
   // Drop the current connection, and start the login process again.
   StartConnection();
@@ -80,28 +83,30 @@ void Login::OnRedirect(const ServerInformation& redirect_server) {
 }
 
 void Login::OnCredentialsRejected() {
+  DVLOG(1) << "Credentials rejected";
   TryReconnect();
   delegate_->OnCredentialsRejected();
 }
 
 void Login::OnSettingsExhausted() {
+  DVLOG(1) << "Settings exhausted";
   TryReconnect();
   delegate_->OnTransientDisconnection();
 }
 
 void Login::OnIPAddressChanged() {
-  DVLOG(1) << "Detected IP address change";
+  DVLOG(1) << "IP address changed";
   OnNetworkEvent();
 }
 
 void Login::OnConnectionTypeChanged(
     net::NetworkChangeNotifier::ConnectionType type) {
-  DVLOG(1) << "Detected connection type change";
+  DVLOG(1) << "Connection type changed";
   OnNetworkEvent();
 }
 
 void Login::OnDNSChanged() {
-  DVLOG(1) << "Detected DNS change";
+  DVLOG(1) << "DNS changed";
   OnNetworkEvent();
 }
 
diff --git a/jingle/notifier/communicator/single_login_attempt.cc b/jingle/notifier/communicator/single_login_attempt.cc
index c2405fe..16d8d42 100644
--- a/jingle/notifier/communicator/single_login_attempt.cc
+++ b/jingle/notifier/communicator/single_login_attempt.cc
@@ -137,6 +137,7 @@ void SingleLoginAttempt::OnError(buzz::XmppEngine::Error error, int subcode,
   }
 
   if (error == buzz::XmppEngine::ERROR_UNAUTHORIZED) {
+    DVLOG(1) << "Credentials rejected";
     delegate_->OnCredentialsRejected();
     return;
   }
@@ -148,7 +149,7 @@ void SingleLoginAttempt::OnError(buzz::XmppEngine::Error error, int subcode,
 
   ++current_settings_;
   if (current_settings_ == settings_list_.end()) {
-    VLOG(1) << "Could not connect to any XMPP server";
+    DVLOG(1) << "Could not connect to any XMPP server";
     delegate_->OnSettingsExhausted();
     return;
   }
diff --git a/jingle/notifier/listener/xmpp_push_client.cc b/jingle/notifier/listener/xmpp_push_client.cc
index 76ee4ee..2227650 100644
--- a/jingle/notifier/listener/xmpp_push_client.cc
+++ b/jingle/notifier/listener/xmpp_push_client.cc
@@ -61,6 +61,7 @@ void XmppPushClient::OnConnect(
 
 void XmppPushClient::OnTransientDisconnection() {
   DCHECK(thread_checker_.CalledOnValidThread());
+  DVLOG(1) << "Push: Transient disconnection";
   base_task_.reset();
   FOR_EACH_OBSERVER(PushClientObserver, observers_,
                     OnNotificationsDisabled(TRANSIENT_NOTIFICATION_ERROR));
@@ -68,6 +69,7 @@ void XmppPushClient::OnTransientDisconnection() {
 
 void XmppPushClient::OnCredentialsRejected() {
   DCHECK(thread_checker_.CalledOnValidThread());
+  DVLOG(1) << "Push: Credentials rejected";
   base_task_.reset();
   FOR_EACH_OBSERVER(
       PushClientObserver, observers_,
diff --git a/net/tools/testserver/testserver.py b/net/tools/testserver/testserver.py
index 4f97587..eefc448 100755
--- a/net/tools/testserver/testserver.py
+++ b/net/tools/testserver/testserver.py
@@ -1788,6 +1788,7 @@ class SyncPageHandler(BasePageHandler):
     get_handlers = [self.ChromiumSyncTimeHandler,
                     self.ChromiumSyncMigrationOpHandler,
                     self.ChromiumSyncCredHandler,
+                    self.ChromiumSyncXmppCredHandler,
                     self.ChromiumSyncDisableNotificationsOpHandler,
                     self.ChromiumSyncEnableNotificationsOpHandler,
                     self.ChromiumSyncSendNotificationOpHandler,
@@ -1894,6 +1895,30 @@ class SyncPageHandler(BasePageHandler):
     self.wfile.write(raw_reply)
     return True
 
+  def ChromiumSyncXmppCredHandler(self):
+    test_name = "/chromiumsync/xmppcred"
+    if not self._ShouldHandleRequest(test_name):
+      return False
+    xmpp_server = self.server.GetXmppServer()
+    try:
+      query = urlparse.urlparse(self.path)[4]
+      cred_valid = urlparse.parse_qs(query)['valid']
+      if cred_valid[0] == 'True':
+        xmpp_server.SetAuthenticated(True)
+      else:
+        xmpp_server.SetAuthenticated(False)
+    except:
+      xmpp_server.SetAuthenticated(False)
+
+    http_response = 200
+    raw_reply = 'XMPP Authenticated: %s ' % xmpp_server.GetAuthenticated()
+    self.send_response(http_response)
+    self.send_header('Content-Type', 'text/html')
+    self.send_header('Content-Length', len(raw_reply))
+    self.end_headers()
+    self.wfile.write(raw_reply)
+    return True
+
   def ChromiumSyncDisableNotificationsOpHandler(self):
     test_name = "/chromiumsync/disablenotifications"
     if not self._ShouldHandleRequest(test_name):
diff --git a/net/tools/testserver/xmppserver.py b/net/tools/testserver/xmppserver.py
index 6952a99..996da96 100644
--- a/net/tools/testserver/xmppserver.py
+++ b/net/tools/testserver/xmppserver.py
@@ -220,6 +220,10 @@ class HandshakeTask(object):
   _AUTH_SUCCESS_STANZA = ParseXml(
     '<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>')
 
+  # Used when in the _AUTH_NEEDED state.
+  _AUTH_FAILURE_STANZA = ParseXml(
+    '<failure xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>')
+
   # Used when in the _AUTH_STREAM_NEEDED state.
   _BIND_STANZA = ParseXml(
     '<stream:features xmlns:stream="http://etherx.jabber.org/streams">'
@@ -242,12 +246,13 @@ class HandshakeTask(object):
   # The id attribute is filled in later.
   _IQ_RESPONSE_STANZA = ParseXml('<iq id="" type="result"/>')
 
-  def __init__(self, connection, resource_prefix):
+  def __init__(self, connection, resource_prefix, authenticated):
     self._connection = connection
     self._id_generator = IdGenerator(resource_prefix)
     self._username = ''
     self._domain = ''
     self._jid = None
+    self._authenticated = authenticated
     self._resource_prefix = resource_prefix
     self._state = self._INITIAL_STREAM_NEEDED
 
@@ -304,6 +309,10 @@ class HandshakeTask(object):
         domain = self._domain
       return (username, domain)
 
+    def Finish():
+      self._state = self._FINISHED
+      self._connection.HandshakeDone(self._jid)
+
     if self._state == self._INITIAL_STREAM_NEEDED:
       HandleStream(stanza)
       self._connection.SendStanza(self._AUTH_STANZA, False)
@@ -312,8 +321,12 @@ class HandshakeTask(object):
     elif self._state == self._AUTH_NEEDED:
       ExpectStanza(stanza, 'auth')
       (self._username, self._domain) = GetUserDomain(stanza)
-      self._connection.SendStanza(self._AUTH_SUCCESS_STANZA, False)
-      self._state = self._AUTH_STREAM_NEEDED
+      if self._authenticated:
+        self._connection.SendStanza(self._AUTH_SUCCESS_STANZA, False)
+        self._state = self._AUTH_STREAM_NEEDED
+      else:
+        self._connection.SendStanza(self._AUTH_FAILURE_STANZA, False)
+        Finish()
 
     elif self._state == self._AUTH_STREAM_NEEDED:
       HandleStream(stanza)
@@ -340,8 +353,7 @@ class HandshakeTask(object):
       xml = CloneXml(self._IQ_RESPONSE_STANZA)
       xml.setAttribute('id', stanza_id)
       self._connection.SendStanza(xml)
-      self._state = self._FINISHED
-      self._connection.HandshakeDone(self._jid)
+      Finish()
 
 
 def AddrString(addr):
@@ -361,7 +373,7 @@ class XmppConnection(asynchat.async_chat):
   # The from and id attributes are filled in later.
   _IQ_RESPONSE_STANZA = ParseXml('<iq from="" id="" type="result"/>')
 
-  def __init__(self, sock, socket_map, delegate, addr):
+  def __init__(self, sock, socket_map, delegate, addr, authenticated):
     """Starts up the xmpp connection.
 
     Args:
@@ -389,7 +401,7 @@ class XmppConnection(asynchat.async_chat):
 
     self._addr = addr
     addr_str = AddrString(self._addr)
-    self._handshake_task = HandshakeTask(self, addr_str)
+    self._handshake_task = HandshakeTask(self, addr_str, authenticated)
     print 'Starting connection to %s' % self
 
   def __str__(self):
@@ -427,10 +439,14 @@ class XmppConnection(asynchat.async_chat):
 
   # Called by self._handshake_task.
   def HandshakeDone(self, jid):
-    self._jid = jid
-    self._handshake_task = None
-    self._delegate.OnXmppHandshakeDone(self)
-    print "Handshake done for %s" % self
+    if jid:
+      self._jid = jid
+      self._handshake_task = None
+      self._delegate.OnXmppHandshakeDone(self)
+      print "Handshake done for %s" % self
+    else:
+      print "Handshake failed for %s" % self
+      self.close()
 
   def _HandlePushCommand(self, stanza):
     if stanza.tagName == 'iq' and stanza.firstChild.tagName == 'subscribe':
@@ -505,10 +521,12 @@ class XmppServer(asyncore.dispatcher):
     self._connections = set()
     self._handshake_done_connections = set()
     self._notifications_enabled = True
+    self._authenticated = True
 
   def handle_accept(self):
     (sock, addr) = self.accept()
-    xmpp_connection = XmppConnection(sock, self._socket_map, self, addr)
+    xmpp_connection = XmppConnection(
+      sock, self._socket_map, self, addr, self._authenticated)
     self._connections.add(xmpp_connection)
     # Return the new XmppConnection for testing.
     return xmpp_connection
@@ -553,6 +571,12 @@ class XmppServer(asyncore.dispatcher):
     self.ForwardNotification(None, notification_stanza)
     notification_stanza.unlink()
 
+  def SetAuthenticated(self, auth_valid):
+    self._authenticated = auth_valid
+
+  def GetAuthenticated(self):
+    return self._authenticated
+
   # XmppConnection delegate methods.
   def OnXmppHandshakeDone(self, xmpp_connection):
     self._handshake_done_connections.add(xmpp_connection)
diff --git a/net/tools/testserver/xmppserver_test.py b/net/tools/testserver/xmppserver_test.py
index b00885a..7431c84 100755
--- a/net/tools/testserver/xmppserver_test.py
+++ b/net/tools/testserver/xmppserver_test.py
@@ -96,7 +96,12 @@ class IdGeneratorTest(unittest.TestCase):
 class HandshakeTaskTest(unittest.TestCase):
 
   def setUp(self):
+    self.Reset()
+
+  def Reset(self):
     self.data_received = 0
+    self.handshake_done = False
+    self.jid = None
 
   def SendData(self, _):
     self.data_received += 1
@@ -105,13 +110,14 @@ class HandshakeTaskTest(unittest.TestCase):
     self.data_received += 1
 
   def HandshakeDone(self, jid):
+    self.handshake_done = True
     self.jid = jid
 
   def DoHandshake(self, resource_prefix, resource, username,
                   initial_stream_domain, auth_domain, auth_stream_domain):
-    self.data_received = 0
+    self.Reset()
     handshake_task = (
-      xmppserver.HandshakeTask(self, resource_prefix))
+      xmppserver.HandshakeTask(self, resource_prefix, True))
     stream_xml = xmppserver.ParseXml('<stream:stream xmlns:stream="foo"/>')
     stream_xml.setAttribute('to', initial_stream_domain)
     self.assertEqual(self.data_received, 0)
@@ -137,11 +143,15 @@ class HandshakeTaskTest(unittest.TestCase):
     handshake_task.FeedStanza(bind_xml)
     self.assertEqual(self.data_received, 6)
 
+    self.assertFalse(self.handshake_done)
+
     session_xml = xmppserver.ParseXml(
       '<iq type="set"><session></session></iq>')
     handshake_task.FeedStanza(session_xml)
     self.assertEqual(self.data_received, 7)
 
+    self.assertTrue(self.handshake_done)
+
     self.assertEqual(self.jid.username, username)
     self.assertEqual(self.jid.domain,
                      auth_stream_domain or auth_domain or
@@ -149,6 +159,34 @@ class HandshakeTaskTest(unittest.TestCase):
     self.assertEqual(self.jid.resource,
                      '%s.%s' % (resource_prefix, resource))
 
+    handshake_task.FeedStanza('<ignored/>')
+    self.assertEqual(self.data_received, 7)
+
+  def DoHandshakeUnauthenticated(self, resource_prefix, resource, username,
+                                 initial_stream_domain):
+    self.Reset()
+    handshake_task = (
+      xmppserver.HandshakeTask(self, resource_prefix, False))
+    stream_xml = xmppserver.ParseXml('<stream:stream xmlns:stream="foo"/>')
+    stream_xml.setAttribute('to', initial_stream_domain)
+    self.assertEqual(self.data_received, 0)
+    handshake_task.FeedStanza(stream_xml)
+    self.assertEqual(self.data_received, 2)
+
+    self.assertFalse(self.handshake_done)
+
+    auth_string = base64.b64encode('\0%s\0bar' % username)
+    auth_xml = xmppserver.ParseXml('<auth>%s</auth>'% auth_string)
+    handshake_task.FeedStanza(auth_xml)
+    self.assertEqual(self.data_received, 3)
+
+    self.assertTrue(self.handshake_done)
+
+    self.assertEqual(self.jid, None)
+
+    handshake_task.FeedStanza('<ignored/>')
+    self.assertEqual(self.data_received, 3)
+
   def testBasic(self):
     self.DoHandshake('resource_prefix', 'resource',
                      'foo', 'bar.com', 'baz.com', 'quux.com')
@@ -163,6 +201,10 @@ class HandshakeTaskTest(unittest.TestCase):
     self.DoHandshake('resource_prefix', 'resource',
                      'foo', '', '', '')
 
+  def testBasicUnauthenticated(self):
+    self.DoHandshakeUnauthenticated('resource_prefix', 'resource',
+                                    'foo', 'bar.com')
+
 
 class FakeSocket(object):
   """A fake socket object used for testing.
@@ -212,7 +254,7 @@ class XmppConnectionTest(unittest.TestCase):
   def testBasic(self):
     socket_map = {}
     xmpp_connection = xmppserver.XmppConnection(
-      self.fake_socket, socket_map, self, ('', 0))
+      self.fake_socket, socket_map, self, ('', 0), True)
     self.assertEqual(len(socket_map), 1)
     self.assertEqual(len(self.connections), 0)
     xmpp_connection.HandshakeDone(xmppserver.Jid('foo', 'bar'))
@@ -248,7 +290,27 @@ class XmppConnectionTest(unittest.TestCase):
     self.assertRaises(xmppserver.UnexpectedXml,
                       SendUnexpectedNotifierCommand)
 
-    # Test close
+    # Test close.
+    xmpp_connection.close()
+    self.assertEqual(len(socket_map), 0)
+    self.assertEqual(len(self.connections), 0)
+
+  def testBasicUnauthenticated(self):
+    socket_map = {}
+    xmpp_connection = xmppserver.XmppConnection(
+      self.fake_socket, socket_map, self, ('', 0), False)
+    self.assertEqual(len(socket_map), 1)
+    self.assertEqual(len(self.connections), 0)
+    xmpp_connection.HandshakeDone(None)
+    self.assertEqual(len(socket_map), 0)
+    self.assertEqual(len(self.connections), 0)
+
+    # Test unexpected stanza.
+    def SendUnexpectedStanza():
+      xmpp_connection.collect_incoming_data('<foo/>')
+    self.assertRaises(xmppserver.UnexpectedXml, SendUnexpectedStanza)
+
+    # Test redundant close.
     xmpp_connection.close()
     self.assertEqual(len(socket_map), 0)
     self.assertEqual(len(self.connections), 0)
diff --git a/sync/internal_api/sync_manager_impl.cc b/sync/internal_api/sync_manager_impl.cc
index dbb8839..40c8b16 100644
--- a/sync/internal_api/sync_manager_impl.cc
+++ b/sync/internal_api/sync_manager_impl.cc
@@ -529,6 +529,10 @@ const SyncScheduler* SyncManagerImpl::scheduler() const {
   return scheduler_.get();
 }
 
+bool SyncManagerImpl::GetHasInvalidAuthTokenForTest() const {
+  return connection_manager_->HasInvalidAuthToken();
+}
+
 bool SyncManagerImpl::OpenDirectory() {
   DCHECK(!initialized_) << "Should only happen once";
 
@@ -1217,8 +1221,13 @@ void SyncManagerImpl::OnInvalidatorStateChange(InvalidatorState state) {
   allstatus_.SetNotificationsEnabled(notifications_enabled);
   scheduler_->SetNotificationsEnabled(notifications_enabled);
 
-  // TODO(akalin): Treat a CREDENTIALS_REJECTED state as an auth
-  // error.
+  if (invalidator_state_ == syncer::INVALIDATION_CREDENTIALS_REJECTED) {
+    // If the invalidator's credentials were rejected, that means that
+    // our sync credentials are also bad, so invalidate those.
+    connection_manager_->InvalidateAndClearAuthToken();
+    FOR_EACH_OBSERVER(SyncManager::Observer, observers_,
+                      OnConnectionStatusChange(CONNECTION_AUTH_ERROR));
+  }
 
   if (js_event_handler_.IsInitialized()) {
     DictionaryValue details;
diff --git a/sync/internal_api/sync_manager_impl.h b/sync/internal_api/sync_manager_impl.h
index de242be..d36ca2e 100644
--- a/sync/internal_api/sync_manager_impl.h
+++ b/sync/internal_api/sync_manager_impl.h
@@ -187,6 +187,8 @@ class SyncManagerImpl :
 
   const SyncScheduler* scheduler() const;
 
+  bool GetHasInvalidAuthTokenForTest() const;
+
  private:
   friend class SyncManagerTest;
   FRIEND_TEST_ALL_PREFIXES(SyncManagerTest, NudgeDelayTest);
diff --git a/sync/internal_api/sync_manager_impl_unittest.cc b/sync/internal_api/sync_manager_impl_unittest.cc
index dd4b85c..e77eb71 100644
--- a/sync/internal_api/sync_manager_impl_unittest.cc
+++ b/sync/internal_api/sync_manager_impl_unittest.cc
@@ -1353,37 +1353,76 @@ TEST_F(SyncManagerTest, GetAllNodesTest) {
   EXPECT_TRUE(first_result->HasKey("NON_UNIQUE_NAME"));
 }
 
-TEST_F(SyncManagerTest, OnNotificationStateChange) {
-  InSequence dummy;
+// Simulate various invalidator state changes.  Those should propagate
+// JS events.
+TEST_F(SyncManagerTest, OnInvalidatorStateChangeJsEvents) {
   StrictMock<MockJsEventHandler> event_handler;
 
   DictionaryValue enabled_details;
   enabled_details.SetString("state", "INVALIDATIONS_ENABLED");
-  DictionaryValue disabled_details;
-  disabled_details.SetString("state", "TRANSIENT_INVALIDATION_ERROR");
+  DictionaryValue credentials_rejected_details;
+  credentials_rejected_details.SetString(
+      "state", "INVALIDATION_CREDENTIALS_REJECTED");
+  DictionaryValue transient_error_details;
+  transient_error_details.SetString("state", "TRANSIENT_INVALIDATION_ERROR");
+  DictionaryValue auth_error_details;
+  auth_error_details.SetString("status", "CONNECTION_AUTH_ERROR");
+
+  EXPECT_CALL(manager_observer_,
+              OnConnectionStatusChange(CONNECTION_AUTH_ERROR)).Times(3);
+
+  EXPECT_CALL(
+      event_handler,
+      HandleJsEvent("onConnectionStatusChange",
+                    HasDetailsAsDictionary(auth_error_details)));
 
   EXPECT_CALL(event_handler,
               HandleJsEvent("onNotificationStateChange",
                             HasDetailsAsDictionary(enabled_details)));
+
+  EXPECT_CALL(
+      event_handler,
+      HandleJsEvent("onNotificationStateChange",
+                    HasDetailsAsDictionary(credentials_rejected_details)));
+
   EXPECT_CALL(event_handler,
               HandleJsEvent("onNotificationStateChange",
-                            HasDetailsAsDictionary(disabled_details)));
+                            HasDetailsAsDictionary(transient_error_details)));
 
   SimulateInvalidatorStateChangeForTest(INVALIDATIONS_ENABLED);
+  SimulateInvalidatorStateChangeForTest(INVALIDATION_CREDENTIALS_REJECTED);
   SimulateInvalidatorStateChangeForTest(TRANSIENT_INVALIDATION_ERROR);
 
   SetJsEventHandler(event_handler.AsWeakHandle());
   SimulateInvalidatorStateChangeForTest(INVALIDATIONS_ENABLED);
+  SimulateInvalidatorStateChangeForTest(INVALIDATION_CREDENTIALS_REJECTED);
   SimulateInvalidatorStateChangeForTest(TRANSIENT_INVALIDATION_ERROR);
   SetJsEventHandler(WeakHandle<JsEventHandler>());
 
   SimulateInvalidatorStateChangeForTest(INVALIDATIONS_ENABLED);
+  SimulateInvalidatorStateChangeForTest(INVALIDATION_CREDENTIALS_REJECTED);
   SimulateInvalidatorStateChangeForTest(TRANSIENT_INVALIDATION_ERROR);
 
   // Should trigger the replies.
   PumpLoop();
 }
 
+// Simulate the invalidator's credentials being rejected.  That should
+// also clear the sync token.
+TEST_F(SyncManagerTest, OnInvalidatorStateChangeCredentialsRejected) {
+  EXPECT_CALL(manager_observer_,
+              OnConnectionStatusChange(CONNECTION_AUTH_ERROR));
+
+  EXPECT_FALSE(sync_manager_.GetHasInvalidAuthTokenForTest());
+
+  SimulateInvalidatorStateChangeForTest(INVALIDATION_CREDENTIALS_REJECTED);
+
+  EXPECT_TRUE(sync_manager_.GetHasInvalidAuthTokenForTest());
+
+  // Should trigger the replies.
+  PumpLoop();
+}
+
 TEST_F(SyncManagerTest, OnIncomingNotification) {
   StrictMock<MockJsEventHandler> event_handler;
 
@@ -2971,7 +3010,7 @@ TEST_F(SyncManagerTest, PurgePartiallySyncedTypes) {
   EXPECT_FALSE(partial_types.Has(PREFERENCES));
 }
 
-// Test CleanipDisabledTypes properly purges all disabled types as specified
+// Test CleanupDisabledTypes properly purges all disabled types as specified
 // by the previous and current enabled params. Enabled partial types should not
 // be purged.
 // Fails on Windows: crbug.com/139726
diff --git a/sync/notifier/invalidator_registrar.cc b/sync/notifier/invalidator_registrar.cc
index 0c8d0f7..7243e9d 100644
--- a/sync/notifier/invalidator_registrar.cc
+++ b/sync/notifier/invalidator_registrar.cc
@@ -120,6 +120,7 @@ void InvalidatorRegistrar::DispatchInvalidationsToHandlers(
 
 void InvalidatorRegistrar::UpdateInvalidatorState(InvalidatorState state) {
   DCHECK(thread_checker_.CalledOnValidThread());
+  DVLOG(1) << "New invalidator state: " << InvalidatorStateToString(state_);
   state_ = state;
   FOR_EACH_OBSERVER(InvalidationHandler, handlers_,
                     OnInvalidatorStateChange(state));