Added auto-enrollment request support to the device_management_backend.

net/tools/testserver/device_management.py

 # 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.
 
 """A bare-bones test server for testing cloud policy support.
 
 This implements a simple cloud policy test server that can be used to test
 chrome's device management service client. The policy information is read from
 the file named device_management in the server's data directory. It contains
 enforced and recommended policies for the device and user scope, and a list
@@ skipping 26 matching lines @@
     }
   },
   "managed_users" : [
     "secret123456"
   ]
 }
 
 """
 
 import cgi
+import hashlib
 import logging
 import os
 import random
 import re
 import sys
 import time
 import tlslite
 import tlslite.api
 import tlslite.utils
 
 # The name and availability of the json module varies in python versions.
 try:
   import simplejson as json
 except ImportError:
   try:
     import json
   except ImportError:
     json = None
 
 import asn1der
 import device_management_backend_pb2 as dm
 import cloud_policy_pb2 as cp
 import chrome_device_policy_pb2 as dp
 
 # ASN.1 object identifier for PKCS#1/RSA.
 PKCS1_RSA_OID = '\x2a\x86\x48\x86\xf7\x0d\x01\x01\x01'
 
+# SHA256 sum of "0".
+SHA256_0 = hashlib.sha256('0').digest()

[pastarmovj, 2011/12/01 09:30:04]

If you only use haslib for this call why don't you just put the string
representation instead (possibly better escaped than what i pasted here :)
"_\xec\xebf\xff\xc8o8\xd9Rxlmily\xc2\xdb\xc29\xddN\x91\xb4g)\xd7:'\xfbW\xe9"

[Joao da Silva, 2011/12/01 09:42:07]

I prefer this format since it's less magical, and it's easier to change the
serial number and add more, if the need arises. wdyt?

[pastarmovj, 2011/12/01 09:49:03]

Ok I am fine with that.

+
 class RequestHandler(object):
   """Decodes and handles device management requests from clients.
 
   The handler implements all the request parsing and protobuf message decoding
   and encoding. It calls back into the server to lookup, register, and
   unregister clients.
   """
 
   def __init__(self, server, path, headers, request):
     """Initialize the handler.
@@ skipping 53 matching lines @@
         (request_type != 'ping' and
          len(self.GetUniqueParam('deviceid')) >= 64) or
         len(self.GetUniqueParam('agent')) >= 64):
       return (400, 'Invalid request parameter')
     if request_type == 'register':
       return self.ProcessRegister(rmsg.register_request)
     elif request_type == 'unregister':
       return self.ProcessUnregister(rmsg.unregister_request)
     elif request_type == 'policy' or request_type == 'ping':
       return self.ProcessPolicy(rmsg.policy_request, request_type)
+    elif request_type == 'enterprise_check':
+      return self.ProcessAutoEnrollment(rmsg.auto_enrollment_request)
     else:
       return (400, 'Invalid request parameter')
 
   def CheckGoogleLogin(self):
     """Extracts the auth token from the request and returns it. The token may
     either be a GoogleLogin token from an Authorization header, or an OAuth V2
     token from the oauth_token query parameter. Returns None if no token is
     present.
     """
     oauth_token = self.GetUniqueParam('oauth_token')
@@ skipping 90 matching lines @@
     for request in msg.request:
       if (request.policy_type in
              ('google/chromeos/user', 'google/chromeos/device')):
         if request_type != 'policy':
           return (400, 'Invalid request type')
         else:
           return self.ProcessCloudPolicy(request)
       else:
         return (400, 'Invalid policy_type')
 
+  def ProcessAutoEnrollment(self, msg):
+    """Handles an auto-enrollment check request.
+
+    The reply depends on the value of the modulus:
+      1: replies with no new modulus and the sha256 hash of "0"
+      2: replies with a new modulus, 4.
+      4: replies with a new modulus, 2.
+      8: fails with error 400.
+      anything else: replies with no new modulus and an empty list of hashes
+
+    These allow the client to pick the testing scenario its wants to simulate.
+
+    Args:
+      msg: The DeviceAutoEnrollmentRequest message received from the client.
+
+    Returns:
+      A tuple of HTTP status code and response data to send to the client.
+    """
+    auto_enrollment_response = dm.DeviceAutoEnrollmentResponse()
+
+    if msg.modulus == 1:
+      auto_enrollment_response.hashes.append(SHA256_0)
+    elif msg.modulus == 2:
+      auto_enrollment_response.modulus = 4
+    elif msg.modulus == 4:
+      auto_enrollment_response.modulus = 2
+    elif msg.modulus == 8:
+      return (400, 'Server error')
+
+    response = dm.DeviceManagementResponse()
+    response.auto_enrollment_response.CopyFrom(auto_enrollment_response)
+    return (200, response.SerializeToString())
+
   def SetProtobufMessageField(self, group_message, field, field_value):
     '''Sets a field in a protobuf message.
 
     Args:
       group_message: The protobuf message.
       field: The field of the message to set, it should be a member of
           group_message.DESCRIPTOR.fields.
       field_value: The value to set.
     '''
     if field.label == field.LABEL_REPEATED:
@@ skipping 305 matching lines @@
     return self._registered_tokens.get(dmtoken, None)
 
   def UnregisterDevice(self, dmtoken):
     """Unregisters a device identified by the given DM token.
 
     Args:
       dmtoken: The device management token provided by the client.
     """
     if dmtoken in self._registered_tokens.keys():
       del self._registered_tokens[dmtoken]