Sending auth error from python server.

net/tools/testserver/testserver.py

 #!/usr/bin/python2.4
 # Copyright (c) 2011 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.
 
 """This is a simple HTTP/FTP/SYNC/TCP/UDP/ server used for testing Chrome.
 
 It supports several test URLs, as specified by the handlers in TestPageHandler.
 By default, it listens on an ephemeral port and sends the port number back to
 the originating process over a pipe. The originating process can specify an
@@ skipping 114 matching lines @@
     # We import here to avoid pulling in chromiumsync's dependencies
     # unless strictly necessary.
     import chromiumsync
     import xmppserver
     StoppableHTTPServer.__init__(self, server_address, request_handler_class)
     self._sync_handler = chromiumsync.TestServer()
     self._xmpp_socket_map = {}
     self._xmpp_server = xmppserver.XmppServer(
       self._xmpp_socket_map, ('localhost', 0))
     self.xmpp_port = self._xmpp_server.getsockname()[1]
+    self.authenticated = True
 
   def GetXmppServer(self):
     return self._xmpp_server
 
   def HandleCommand(self, query, raw_request):
     return self._sync_handler.HandleCommand(query, raw_request)
 
   def HandleRequestNoBlock(self):
     """Handles a single request.
 
     Copied from SocketServer._handle_request_noblock().
     """
     try:
       request, client_address = self.get_request()
     except socket.error:
       return
     if self.verify_request(request, client_address):
       try:
         self.process_request(request, client_address)
       except:
         self.handle_error(request, client_address)
         self.close_request(request)
 
+  def SetAuthenticated(self, auth_valid):
+    self.authenticated = auth_valid
+
+  def GetAuthenticated(self):
+    return self.authenticated
+
   def serve_forever(self):
     """This is a merge of asyncore.loop() and SocketServer.serve_forever().
     """
 
     def HandleXmppSocket(fd, socket_map, handler):
       """Runs the handler for the xmpp connection for fd.
 
       Adapted from asyncore.read() et al.
       """
       xmpp_connection = socket_map.get(fd)
@@ skipping 1239 matching lines @@
   """Handler for the main HTTP sync server."""
 
   def __init__(self, request, client_address, sync_http_server):
     get_handlers = [self.ChromiumSyncMigrationOpHandler,
                     self.ChromiumSyncTimeHandler,
                     self.ChromiumSyncDisableNotificationsOpHandler,
                     self.ChromiumSyncEnableNotificationsOpHandler,
                     self.ChromiumSyncSendNotificationOpHandler,
                     self.ChromiumSyncBirthdayErrorOpHandler,
                     self.ChromiumSyncTransientErrorOpHandler,
-                    self.ChromiumSyncSyncTabsOpHandler]
+                    self.ChromiumSyncSyncTabsOpHandler,
+                    self.ChromiumSyncCredHandler]
 
     post_handlers = [self.ChromiumSyncCommandHandler,
                      self.ChromiumSyncTimeHandler]
     BasePageHandler.__init__(self, request, client_address,
                              sync_http_server, [], get_handlers,
                              post_handlers, [])
 
+
   def ChromiumSyncTimeHandler(self):
     """Handle Chromium sync .../time requests.
 
     The syncer sometimes checks server reachability by examining /time.
     """
     test_name = "/chromiumsync/time"
     if not self._ShouldHandleRequest(test_name):
       return False
 
     # Chrome hates it if we send a response before reading the request.
@@ skipping 12 matching lines @@
 
     This covers all sync protocol commands: authentication, getupdates, and
     commit.
     """
     test_name = "/chromiumsync/command"
     if not self._ShouldHandleRequest(test_name):
       return False
 
     length = int(self.headers.getheader('content-length'))
     raw_request = self.rfile.read(length)
+    http_response = 200
+    raw_reply = None
+    if not self.server.GetAuthenticated():
+      http_response = 401
+      challenge = 'GoogleLogin realm="http://127.0.0.1", service="chromiumsync"'
+    else:
+      http_response, raw_reply = self.server.HandleCommand(
+          self.path, raw_request)
 
-    http_response, raw_reply = self.server.HandleCommand(
-        self.path, raw_request)
+    ### Now send the response to the client. ###
     self.send_response(http_response)
+    if http_response == 401:
+      self.send_header('www-Authenticate', challenge)
     self.end_headers()
     self.wfile.write(raw_reply)
     return True
 
   def ChromiumSyncMigrationOpHandler(self):
     test_name = "/chromiumsync/migrate"
     if not self._ShouldHandleRequest(test_name):
       return False
 
     http_response, raw_reply = self.server._sync_handler.HandleMigrate(
         self.path)
     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 ChromiumSyncCredHandler(self):
+    test_name = "/chromiumsync/cred"
+    if not self._ShouldHandleRequest(test_name):
+      return False
+    try:
+      query = urlparse.urlparse(self.path)[4]
+      cred_valid = urlparse.parse_qs(query)['valid']
+      if cred_valid[0] == 'True':
+        self.server.SetAuthenticated(True)
+      else:
+        self.server.SetAuthenticated(False)
+    except:
+      self.server.SetAuthenticated(False)
+
+    http_response = 200
+    raw_reply = 'Authenicated: %s ' % self.server.GetAuthenticated()

[ncarter (slow), 2011/09/22 20:22:19]

"Authenicated" -> misspelled.

+    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):
       return False
     self.server.GetXmppServer().DisableNotifications()
     result = 200
     raw_reply = ('<html><title>Notifications disabled</title>'
                  '<H1>Notifications disabled</H1></html>')
     self.send_response(result)
     self.send_header('Content-Type', 'text/html')
@@ skipping 336 matching lines @@
                            'random key if none is specified on the command '
                            'line.')
   option_parser.add_option('', '--policy-user', default='user@example.com',
                            dest='policy_user',
                            help='Specify the user name the server should '
                            'report back to the client as the user owning the '
                            'token used for making the policy request.')
   options, args = option_parser.parse_args()
 
   sys.exit(main(options, args))