Verify the signature on user cloud policy downloads.

net/tools/testserver/device_management.py

 # Copyright (c) 2012 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 27 matching lines @@
   },
   "google/chromeos/publicaccount/user@example.com" : {
     "mandatory" : {
       "HomepageLocation" : "http://www.chromium.org"
     },
      "recommended" : {
     }
   },
   "managed_users" : [
     "secret123456"
-  ]
+  ],
+  "current_key_index": 0
 }
 
 """
 
 import cgi
 import hashlib
 import logging
 import os
 import random
 import re
@@ skipping 389 matching lines @@
     if msg.policy_type in token_info['allowed_policy_types']:
       if (msg.policy_type == 'google/chromeos/user' or
           msg.policy_type == 'google/chrome/user' or
           msg.policy_type == 'google/chromeos/publicaccount'):
         settings = cp.CloudPolicySettings()
         self.GatherUserPolicySettings(settings, policy.get(policy_key, {}))
       elif msg.policy_type == 'google/chromeos/device':
         settings = dp.ChromeDeviceSettingsProto()
         self.GatherDevicePolicySettings(settings, policy.get(policy_key, {}))
 
-    # Figure out the key we want to use. If multiple keys are configured, the
-    # server will rotate through them in a round-robin fashion.
+    # Sign with 'current_key_index', defaulting to key 0.
     signing_key = None
     req_key = None
-    key_version = 1
+    current_key_index = policy.get('current_key_index', 0)
     nkeys = len(self._server.keys)
-    if msg.signature_type == dm.PolicyFetchRequest.SHA1_RSA and nkeys > 0:
+    if (msg.signature_type == dm.PolicyFetchRequest.SHA1_RSA and
+        current_key_index in range(nkeys)):
+      signing_key = self._server.keys[current_key_index]
       if msg.public_key_version in range(1, nkeys + 1):
         # requested key exists, use for signing and rotate.
         req_key = self._server.keys[msg.public_key_version - 1]['private_key']
-        key_version = (msg.public_key_version % nkeys) + 1
-      signing_key = self._server.keys[key_version - 1]
 
     # Fill the policy data protobuf.
     policy_data = dm.PolicyData()
     policy_data.policy_type = msg.policy_type
     policy_data.timestamp = int(time.time() * 1000)
     policy_data.request_token = token_info['device_token']
     policy_data.policy_value = settings.SerializeToString()
     policy_data.machine_name = token_info['machine_name']
     policy_data.valid_serial_number_missing = (
         token_info['machine_id'] in BAD_MACHINE_IDS)
     policy_data.settings_entity_id = msg.settings_entity_id
 
     if signing_key:
-      policy_data.public_key_version = key_version
+      policy_data.public_key_version = current_key_index + 1
     if msg.policy_type == 'google/chromeos/publicaccount':
       policy_data.username = msg.settings_entity_id
     else:
       # For regular user/device policy, there is no way for the testserver to
       # know the user name belonging to the GAIA auth token we received (short
       # of actually talking to GAIA). To address this, we read the username from
       # the policy configuration dictionary, or use a default.
       policy_data.username = policy.get('policy_user', 'user@example.com')
     policy_data.device_id = token_info['device_id']
     signed_data = policy_data.SerializeToString()
 
     response = dm.DeviceManagementResponse()
     fetch_response = response.policy_response.response.add()
     fetch_response.policy_data = signed_data
     if signing_key:
       fetch_response.policy_data_signature = (
           signing_key['private_key'].hashAndSign(signed_data).tostring())
-      if msg.public_key_version != key_version:
+      if msg.public_key_version != current_key_index + 1:
         fetch_response.new_public_key = signing_key['public_key']
         if req_key:
           fetch_response.new_public_key_signature = (
               req_key.hashAndSign(fetch_response.new_public_key).tostring())
 
     self.DumpMessage('Response', response)
 
     return (200, response.SerializeToString())
 
   def CheckToken(self):
@@ skipping 53 matching lines @@
       for key_path in private_key_paths:
         try:
           key = tlslite.api.parsePEMKey(open(key_path).read(), private=True)
         except IOError:
           print 'Failed to load private key from %s' % key_path
           continue
 
         assert key is not None
         self.keys.append({ 'private_key' : key })
     else:
-      # Generate a key if none were specified.
-      key = tlslite.api.generateRSAKey(1024)
-      assert key is not None
-      self.keys.append({ 'private_key' : key })
+      # Generate 2 private keys if none were passed from the command line.
+      for i in range(2):
+        key = tlslite.api.generateRSAKey(512)
+        assert key is not None
+        self.keys.append({ 'private_key' : key })
 
-    # Derive the public keys from the loaded private keys.
+    # Derive the public keys from the private keys.
     for entry in self.keys:
       key = entry['private_key']
 
       algorithm = asn1der.Sequence(
           [ asn1der.Data(asn1der.OBJECT_IDENTIFIER, PKCS1_RSA_OID),
             asn1der.Data(asn1der.NULL, '') ])
       rsa_pubkey = asn1der.Sequence([ asn1der.Integer(key.n),
                                       asn1der.Integer(key.e) ])
       pubkey = asn1der.Sequence([ algorithm, asn1der.Bitstring(rsa_pubkey) ])
       entry['public_key'] = pubkey;
@@ skipping 85 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]