# 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
of managed users.
The format of the file is JSON. The root dictionary contains a list under the
key "managed_users". It contains auth tokens for which the server will claim
that the user is managed. The token string "*" indicates that all users are
claimed to be managed. Other keys in the root dictionary identify request
scopes. The user-request scope is described by a dictionary that holds two
sub-dictionaries: "mandatory" and "recommended". Both these hold the policy
definitions as key/value stores, their format is identical to what the Linux
implementation reads from /etc.
The device-scope holds the policy-definition directly as key/value stores in the
protobuf-format.
Example:
{
"google/chromeos/device" : {
"guest_mode_enabled" : false
},
"google/chromeos/user" : {
"mandatory" : {
"HomepageLocation" : "http://www.chromium.org",
"IncognitoEnabled" : false
},
"recommended" : {
"JavascriptEnabled": false
}
},
"google/chromeos/publicaccount/user@example.com" : {
"mandatory" : {
"HomepageLocation" : "http://www.chromium.org"
},
"recommended" : {
}
},
"managed_users" : [
"secret123456"
],
"current_key_index": 0,
"robot_api_auth_code": "fake_auth_code",
"invalidation_source": 1025,
"invalidation_name": "UENUPOL"
}
"""
import base64
import BaseHTTPServer
import cgi
import glob
import google.protobuf.text_format
import hashlib
import logging
import os
import random
import re
import sys
import time
import tlslite
import tlslite.api
import tlslite.utils
import tlslite.utils.cryptomath
import urlparse
# 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 testserver_base
import device_management_backend_pb2 as dm
import cloud_policy_pb2 as cp
import chrome_extension_policy_pb2 as ep
# Device policy is only available on Chrome OS builds.
try:
import chrome_device_policy_pb2 as dp
except ImportError:
dp = None
# ASN.1 object identifier for PKCS#1/RSA.
PKCS1_RSA_OID = '\x2a\x86\x48\x86\xf7\x0d\x01\x01\x01'
# List of bad machine identifiers that trigger the |valid_serial_number_missing|
# flag to be set set in the policy fetch response.
BAD_MACHINE_IDS = [ '123490EN400015' ]
# List of machines that trigger the server to send kiosk enrollment response
# for the register request.
KIOSK_MACHINE_IDS = [ 'KIOSK' ]
# Dictionary containing base64-encoded policy signing keys plus per-domain
# signatures. Format is:
# {
# 'key': <base64-encoded PKCS8-format private key>,
# 'signatures': {
# <domain1>: <base64-encdoded SHA256 signature for key + domain1>
# <domain2>: <base64-encdoded SHA256 signature for key + domain2>
# ...
# }
# }
SIGNING_KEYS = [
# Key1
{'key':
'MIIBVQIBADANBgkqhkiG9w0BAQEFAASCAT8wggE7AgEAAkEA2c3KzcPqvnJ5HCk3OZkf1'
'LMO8Ht4dw4FO2U0EmKvpo0zznj4RwUdmKobH1AFWzwZP4CDY2M67MsukE/1Jnbx1QIDAQ'
'ABAkBkKcLZa/75hHVz4PR3tZaw34PATlfxEG6RiRIwXlf/FFlfGIZOSxdW/I1A3XRl0/9'
'nZMuctBSKBrcTRZQWfT/hAiEA9g8xbQbMO6BEH/XCRSsQbPlvj4c9wDtVEzeAzZ/ht9kC'
'IQDiml+/lXS1emqml711jJcYJNYJzdy1lL/ieKogR59oXQIhAK+Pl4xa1U2VxAWpq7r+R'
'vH55wdZT03hB4p2h4gvEzXBAiAkw9kvE0eZPiBZoRrrHIFTOH7FnnHlwBmV2+/2RsiVPQ'
'IhAKqx/4qisivvmoM/xbzUagfoxwsu1A/4mGjhBKiS0BCq',
'signatures':
{'example.com':
'l+sT5mziei/GbmiP7VtRCCfwpZcg7uKbW2OlnK5B/TTELutjEIAMdHduNBwbO44qOn'
'/5c7YrtkXbBehaaDYFPGI6bGTbDmG9KRxhS+DaB7opgfCQWLi79Gn/jytKLZhRN/VS'
'y+PEbezqMi3d1/xDxlThwWZDNwnhv9ER/Nu/32ZTjzgtqonSn2CQtwXCIILm4FdV/1'
'/BdmZG+Ge4i4FTqYtInir5YFe611KXU/AveGhQGBIAXo4qYg1IqbVrvKBSU9dlI6Sl'
'9TJJLbJ3LGaXuljgFhyMAl3gcy7ftC9MohEmwa+sc7y2mOAgYQ5SSmyAtQwQgAkX9J'
'3+tfxjmoA/dg==',
'chromepolicytest.com':
'TzBiigZKwBdr6lyP6tUDsw+Q9wYO1Yepyxm0O4JZ4RID32L27sWzC1/hwC51fRcCvP'
'luEVIW6mH+BFODXMrteUFWfbbG7jgV+Wg+QdzMqgJjxhNKFXPTsZ7/286LAd1vBY/A'
'nGd8Wog6AhzfrgMbLNsH794GD0xIUwRvXUWFNP8pClj5VPgQnJrIA9aZwW8FNGbteA'
'HacFB0T/oqP5s7XT4Qvkj14RLmCgTwEM8Vcpqy5teJaF8yN17wniveddoOQGH6s0HC'
'ocprEccrH5fP/WVAPxCfx4vVYQY5q4CZ4K3f6dTC2FV4IDelM6dugEkvSS02YCzDaO'
'N+Z7IwElzTKg==',
'managedchrome.com':
'T0wXC5w3GXyovA09pyOLX7ui/NI603UfbZXYyTbHI7xtzCIaHVPH35Nx4zdqVrdsej'
'ErQ12yVLDDIJokY4Yl+/fj/zrkAPxThI+TNQ+jo0i+al05PuopfpzvCzIXiZBbkbyW'
'3XfedxXP3IPN2XU2/3vX+ZXUNG6pxeETem64kGezkjkUraqnHw3JVzwJYHhpMcwdLP'
'PYK6V23BbEHEVBtQZd/ledXacz7gOzm1zGni4e+vxA2roAdJWyhbjU0dTKNNUsZmMv'
'ryQH9Af1Jw+dqs0RAbhcJXm2i8EUWIgNv6aMn1Z2DzZwKKjXsKgcYSRo8pdYa8RZAo'
'UExd9roA9a5w==',
}
},
# Key2
{'key':
'MIIBVAIBADANBgkqhkiG9w0BAQEFAASCAT4wggE6AgEAAkEAmZhreV04M3knCi6wibr49'
'oDesHny1G33PKOX9ko8pcxAiu9ZqsKCj7wNW2PGqnLi81fddACwQtYn5xdhCtzB9wIDAQ'
'ABAkA0z8m0cy8N08xundspoFZWO71WJLgv/peSDBYGI0RzJR1l9Np355EukQUQwRs5XrL'
'3vRQZy2vDqeiR96epkAhRAiEAzJ4DVI8k3pAl7CGv5icqFkJ02viExIwehhIEXBcB6p0C'
'IQDAKmzpoRpBEZRQ9xrTvPOi+Ea8Jnd478BU7CI/LFfgowIgMfLIoVWoDGRnvXKju60Hy'
'xNB70oHLut9cADp64j6QMkCIDrgxN4QbmrhaAAmtiGKE1wrlgCwCIsVamiasSOKAqLhAi'
'EAo/ItVcFtQPod97qG71CY/O4JzOciuU6AMhprs181vfM=',
'signatures':
# Key2 signatures
{'example.com':
'cO0nQjRptkeefKDw5QpJSQDavHABxUvbR9Wvoa235OG9Whw1RFqq2ye6pKnI3ezW6/'
'7b4ANcpi5a7HV5uF8K7gWyYdxY8NHLeyrbwXxg5j6HAmHmkP1UZcf/dAnWqo7cW8g4'
'DIQOhC43KkveMYJ2HnelwdXt/7zqkbe8/3Yj4nhjAUeARx86Sb8Nzydwkrvqs5Jw/x'
'5LG+BODExrXXcGu/ubDlW4ivJFqfNUPQysqBXSMY2XCHPJDx3eECLGVVN/fFAWWgjM'
'HFObAriAt0b18cc9Nr0mAt4Qq1oDzWcAHCPHE+5dr8Uf46BUrMLJRNRKCY7rrsoIin'
'9Be9gs3W+Aww==',
'chromepolicytest.com':
'mr+9CCYvR0cTvPwlzkxqlpGYy55gY7cPiIkPAPoql51yHK1tkMTOSFru8Dy/nMt+0o'
'4z7WO60F1wnIBGkQxnTj/DsO6QpCYi7oHqtLmZ2jsLQFlMyvPGUtpJEFvRwjr/TNbh'
'6RqUtz1LQFuJQ848kBrx7nkte1L8SuPDExgx+Q3LtbNj4SuTdvMUBMvEERXiLuwfFL'
'BefGjtsqfWETQVlJTCW7xcqOLedIX8UYgEDBpDOZ23A3GzCShuBsIut5m87R5mODht'
'EUmKNDK1+OMc6SyDpf+r48Wph4Db1bVaKy8fcpSNJOwEgsrmH7/+owKPGcN7I5jYAF'
'Z2PGxHTQ9JNA==',
'managedchrome.com':
'o5MVSo4bRwIJ/aooGyXpRXsEsWPG8fNA2UTG8hgwnLYhNeJCCnLs/vW2vdp0URE8jn'
'qiG4N8KjbuiGw0rJtO1EygdLfpnMEtqYlFjrOie38sy92l/AwohXj6luYzMWL+FqDu'
'WQeXasjgyY4s9BOLQVDEnEj3pvqhrk/mXvMwUeXGpbxTNbWAd0C8BTZrGOwU/kIXxo'
'vAMGg8L+rQaDwBTEnMsMZcvlrIyqSg5v4BxCWuL3Yd2xvUqZEUWRp1aKetsHRnz5hw'
'H7WK7DzvKepDn06XjPG9lchi448U3HB3PRKtCzfO3nD9YXMKTuqRpKPF8PeK11CWh1'
'DBvBYwi20vbQ==',
},
},
]
class PolicyRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
"""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, request, client_address, server):
"""Initialize the handler.
Args:
request: The request data received from the client as a string.
client_address: The client address.
server: The TestServer object to use for (un)registering clients.
"""
BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, request,
client_address, server)
def GetUniqueParam(self, name):
"""Extracts a unique query parameter from the request.
Args:
name: Names the parameter to fetch.
Returns:
The parameter value or None if the parameter doesn't exist or is not
unique.
"""
if not hasattr(self, '_params'):
self._params = cgi.parse_qs(self.path[self.path.find('?') + 1:])
param_list = self._params.get(name, [])
if len(param_list) == 1:
return param_list[0]
return None
def do_GET(self):
"""Handles GET requests.
Currently this is only used to serve external policy data."""
sep = self.path.find('?')
path = self.path if sep == -1 else self.path[:sep]
if path == '/externalpolicydata':
http_response, raw_reply = self.HandleExternalPolicyDataRequest()
else:
http_response = 404
raw_reply = 'Invalid path'
self.send_response(http_response)
self.end_headers()
self.wfile.write(raw_reply)
def do_POST(self):
http_response, raw_reply = self.HandleRequest()
self.send_response(http_response)
if (http_response == 200):
self.send_header('Content-Type', 'application/x-protobuffer')
self.end_headers()
self.wfile.write(raw_reply)
def HandleExternalPolicyDataRequest(self):
"""Handles a request to download policy data for a component."""
policy_key = self.GetUniqueParam('key')
if not policy_key:
return (400, 'Missing key parameter')
data = self.server.ReadPolicyDataFromDataDir(policy_key)
if data is None:
return (404, 'Policy not found for ' + policy_key)
return (200, data)
def HandleRequest(self):
"""Handles a request.
Parses the data supplied at construction time and returns a pair indicating
http status code and response data to be sent back to the client.
Returns:
A tuple of HTTP status code and response data to send to the client.
"""
rmsg = dm.DeviceManagementRequest()
length = int(self.headers.getheader('content-length'))
rmsg.ParseFromString(self.rfile.read(length))
logging.debug('gaia auth token -> ' +
self.headers.getheader('Authorization', ''))
logging.debug('oauth token -> ' + str(self.GetUniqueParam('oauth_token')))
logging.debug('deviceid -> ' + str(self.GetUniqueParam('deviceid')))
self.DumpMessage('Request', rmsg)
request_type = self.GetUniqueParam('request')
# Check server side requirements, as defined in
# device_management_backend.proto.
if (self.GetUniqueParam('devicetype') != '2' or
self.GetUniqueParam('apptype') != 'Chrome' or
len(self.GetUniqueParam('deviceid')) >= 64 or
len(self.GetUniqueParam('agent')) >= 64):
return (400, 'Invalid request parameter')
if request_type == 'register':
response = self.ProcessRegister(rmsg.register_request)
elif request_type == 'api_authorization':
response = self.ProcessApiAuthorization(rmsg.service_api_access_request)
elif request_type == 'unregister':
response = self.ProcessUnregister(rmsg.unregister_request)
elif request_type == 'policy':
response = self.ProcessPolicy(rmsg, request_type)
elif request_type == 'enterprise_check':
response = self.ProcessAutoEnrollment(rmsg.auto_enrollment_request)
elif request_type == 'device_state_retrieval':
response = self.ProcessDeviceStateRetrievalRequest(
rmsg.device_state_retrieval_request)
else:
return (400, 'Invalid request parameter')
self.DumpMessage('Response', response[1])
return (response[0], response[1].SerializeToString())
def CreatePolicyForExternalPolicyData(self, policy_key):
"""Returns an ExternalPolicyData protobuf for policy_key.
If there is policy data for policy_key then the download url will be
set so that it points to that data, and the appropriate hash is also set.
Otherwise, the protobuf will be empty.
Args:
policy_key: The policy type and settings entity id, joined by '/'.
Returns:
A serialized ExternalPolicyData.
"""
settings = ep.ExternalPolicyData()
data = self.server.ReadPolicyDataFromDataDir(policy_key)
if data:
settings.download_url = urlparse.urljoin(
self.server.GetBaseURL(), 'externalpolicydata?key=%s' % policy_key)
settings.secure_hash = hashlib.sha256(data).digest()
return settings.SerializeToString()
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')
if oauth_token:
return oauth_token
match = re.match('GoogleLogin auth=(\\w+)',
self.headers.getheader('Authorization', ''))
if match:
return match.group(1)
return None
def ProcessRegister(self, msg):
"""Handles a register request.
Checks the query for authorization and device identifier, registers the
device with the server and constructs a response.
Args:
msg: The DeviceRegisterRequest message received from the client.
Returns:
A tuple of HTTP status code and response data to send to the client.
"""
# Check the auth token and device ID.
auth = self.CheckGoogleLogin()
if not auth:
return (403, 'No authorization')
policy = self.server.GetPolicies()
if ('*' not in policy['managed_users'] and
auth not in policy['managed_users']):
return (403, 'Unmanaged')
device_id = self.GetUniqueParam('deviceid')
if not device_id:
return (400, 'Missing device identifier')
token_info = self.server.RegisterDevice(device_id,
msg.machine_id,
msg.type)
# Send back the reply.
response = dm.DeviceManagementResponse()
response.register_response.device_management_token = (
token_info['device_token'])
response.register_response.machine_name = token_info['machine_name']
response.register_response.enrollment_type = token_info['enrollment_mode']
return (200, response)
def ProcessApiAuthorization(self, msg):
"""Handles an API authorization request.
Args:
msg: The DeviceServiceApiAccessRequest message received from the client.
Returns:
A tuple of HTTP status code and response data to send to the client.
"""
policy = self.server.GetPolicies()
# Return the auth code from the config file if it's defined,
# else return a descriptive default value.
response = dm.DeviceManagementResponse()
response.service_api_access_response.auth_code = policy.get(
'robot_api_auth_code', 'policy_testserver.py-auth_code')
return (200, response)
def ProcessUnregister(self, msg):
"""Handles a register request.
Checks for authorization, unregisters the device and constructs the
response.
Args:
msg: The DeviceUnregisterRequest message received from the client.
Returns:
A tuple of HTTP status code and response data to send to the client.
"""
# Check the management token.
token, response = self.CheckToken()
if not token:
return response
# Unregister the device.
self.server.UnregisterDevice(token['device_token'])
# Prepare and send the response.
response = dm.DeviceManagementResponse()
response.unregister_response.CopyFrom(dm.DeviceUnregisterResponse())
return (200, response)
def ProcessPolicy(self, msg, request_type):
"""Handles a policy request.
Checks for authorization, encodes the policy into protobuf representation
and constructs the response.
Args:
msg: The DeviceManagementRequest message received from the client.
Returns:
A tuple of HTTP status code and response data to send to the client.
"""
token_info, error = self.CheckToken()
if not token_info:
return error
key_update_request = msg.device_state_key_update_request
if len(key_update_request.server_backed_state_key) > 0:
self.server.UpdateStateKeys(token_info['device_token'],
key_update_request.server_backed_state_key)
# If this is a |publicaccount| request, get the |username| now and use
# it in every PolicyFetchResponse produced. This is required to validate
# policy for extensions in device-local accounts.
# Unfortunately, the |username| can't be obtained from |msg| because that
# requires interacting with GAIA.
username = None
for request in msg.policy_request.request:
if request.policy_type == 'google/chromeos/publicaccount':
username = request.settings_entity_id
response = dm.DeviceManagementResponse()
for request in msg.policy_request.request:
if (request.policy_type in
('google/android/user',
'google/chromeos/device',
'google/chromeos/publicaccount',
'google/chromeos/user',
'google/chrome/user',
'google/ios/user')):
fetch_response = response.policy_response.response.add()
self.ProcessCloudPolicy(request, token_info, fetch_response, username)
elif request.policy_type == 'google/chrome/extension':
self.ProcessCloudPolicyForExtensions(
request, response.policy_response, token_info, username)
else:
fetch_response.error_code = 400
fetch_response.error_message = 'Invalid policy_type'
return (200, response)
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.
16: replies with a new modulus, 16.
32: replies with a new modulus, 1.
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.hash.extend(
self.server.GetMatchingStateKeyHashes(msg.modulus, msg.remainder))
elif msg.modulus == 2:
auto_enrollment_response.expected_modulus = 4
elif msg.modulus == 4:
auto_enrollment_response.expected_modulus = 2
elif msg.modulus == 8:
return (400, 'Server error')
elif msg.modulus == 16:
auto_enrollment_response.expected_modulus = 16
elif msg.modulus == 32:
auto_enrollment_response.expected_modulus = 1
response = dm.DeviceManagementResponse()
response.auto_enrollment_response.CopyFrom(auto_enrollment_response)
return (200, response)
def ProcessDeviceStateRetrievalRequest(self, msg):
"""Handles a device state retrieval request.
Response data is taken from server configuration.
Returns:
A tuple of HTTP status code and response data to send to the client.
"""
device_state_retrieval_response = dm.DeviceStateRetrievalResponse()
client = self.server.LookupByStateKey(msg.server_backed_state_key)
if client is not None:
state = self.server.GetPolicies().get('device_state', {})
FIELDS = [
'management_domain',
'restore_mode',
]
for field in FIELDS:
if field in state:
setattr(device_state_retrieval_response, field, state[field])
response = dm.DeviceManagementResponse()
response.device_state_retrieval_response.CopyFrom(
device_state_retrieval_response)
return (200, response)
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:
assert type(field_value) == list
entries = group_message.__getattribute__(field.name)
if field.message_type is None:
for list_item in field_value:
entries.append(list_item)
else:
# This field is itself a protobuf.
sub_type = field.message_type
for sub_value in field_value:
assert type(sub_value) == dict
# Add a new sub-protobuf per list entry.
sub_message = entries.add()
# Now iterate over its fields and recursively add them.
for sub_field in sub_message.DESCRIPTOR.fields:
if sub_field.name in sub_value:
value = sub_value[sub_field.name]
self.SetProtobufMessageField(sub_message, sub_field, value)
return
elif field.type == field.TYPE_BOOL:
assert type(field_value) == bool
elif field.type == field.TYPE_STRING:
assert type(field_value) == str or type(field_value) == unicode
elif field.type == field.TYPE_INT64:
assert type(field_value) == int
elif (field.type == field.TYPE_MESSAGE and
field.message_type.name == 'StringList'):
assert type(field_value) == list
entries = group_message.__getattribute__(field.name).entries
for list_item in field_value:
entries.append(list_item)
return
else:
raise Exception('Unknown field type %s' % field.type)
group_message.__setattr__(field.name, field_value)
def GatherDevicePolicySettings(self, settings, policies):
"""Copies all the policies from a dictionary into a protobuf of type
CloudDeviceSettingsProto.
Args:
settings: The destination ChromeDeviceSettingsProto protobuf.
policies: The source dictionary containing policies in JSON format.
"""
for group in settings.DESCRIPTOR.fields:
# Create protobuf message for group.
group_message = eval('dp.' + group.message_type.name + '()')
# Indicates if at least one field was set in |group_message|.
got_fields = False
# Iterate over fields of the message and feed them from the
# policy config file.
for field in group_message.DESCRIPTOR.fields:
field_value = None
if field.name in policies:
got_fields = True
field_value = policies[field.name]
self.SetProtobufMessageField(group_message, field, field_value)
if got_fields:
settings.__getattribute__(group.name).CopyFrom(group_message)
def GatherUserPolicySettings(self, settings, policies):
"""Copies all the policies from a dictionary into a protobuf of type
CloudPolicySettings.
Args:
settings: The destination: a CloudPolicySettings protobuf.
policies: The source: a dictionary containing policies under keys
'recommended' and 'mandatory'.
"""
for field in settings.DESCRIPTOR.fields:
# |field| is the entry for a specific policy in the top-level
# CloudPolicySettings proto.
# Look for this policy's value in the mandatory or recommended dicts.
if field.name in policies.get('mandatory', {}):
mode = cp.PolicyOptions.MANDATORY
value = policies['mandatory'][field.name]
elif field.name in policies.get('recommended', {}):
mode = cp.PolicyOptions.RECOMMENDED
value = policies['recommended'][field.name]
else:
continue
# Create protobuf message for this policy.
policy_message = eval('cp.' + field.message_type.name + '()')
policy_message.policy_options.mode = mode
field_descriptor = policy_message.DESCRIPTOR.fields_by_name['value']
self.SetProtobufMessageField(policy_message, field_descriptor, value)
settings.__getattribute__(field.name).CopyFrom(policy_message)
def ProcessCloudPolicyForExtensions(self, request, response, token_info,
username=None):
"""Handles a request for policy for extensions.
A request for policy for extensions is slightly different from the other
cloud policy requests, because it can trigger 0, one or many
PolicyFetchResponse messages in the response.
Args:
request: The PolicyFetchRequest that triggered this handler.
response: The DevicePolicyResponse message for the response. Multiple
PolicyFetchResponses will be appended to this message.
token_info: The token extracted from the request.
username: The username for the response. May be None.
"""
# Send one PolicyFetchResponse for each extension that has
# configuration data at the server.
ids = self.server.ListMatchingComponents('google/chrome/extension')
for settings_entity_id in ids:
# Reuse the extension policy request, to trigger the same signature
# type in the response.
request.settings_entity_id = settings_entity_id
fetch_response = response.response.add()
self.ProcessCloudPolicy(request, token_info, fetch_response, username)
# Don't do key rotations for these messages.
fetch_response.ClearField('new_public_key')
fetch_response.ClearField('new_public_key_signature')
fetch_response.ClearField('new_public_key_verification_signature')
def ProcessCloudPolicy(self, msg, token_info, response, username=None):
"""Handles a cloud policy request. (New protocol for policy requests.)
Encodes the policy into protobuf representation, signs it and constructs
the response.
Args:
msg: The CloudPolicyRequest message received from the client.
token_info: The token extracted from the request.
response: A PolicyFetchResponse message that should be filled with the
response data.
username: The username for the response. May be None.
"""
if msg.machine_id:
self.server.UpdateMachineId(token_info['device_token'], msg.machine_id)
# Response is only given if the scope is specified in the config file.
# Normally 'google/chromeos/device', 'google/chromeos/user' and
# 'google/chromeos/publicaccount' should be accepted.
policy = self.server.GetPolicies()
policy_value = ''
policy_key = msg.policy_type
if msg.settings_entity_id:
policy_key += '/' + msg.settings_entity_id
if msg.policy_type in token_info['allowed_policy_types']:
if msg.policy_type in ('google/android/user',
'google/chromeos/publicaccount',
'google/chromeos/user',
'google/chrome/user',
'google/ios/user'):
settings = cp.CloudPolicySettings()
payload = self.server.ReadPolicyFromDataDir(policy_key, settings)
if payload is None:
self.GatherUserPolicySettings(settings, policy.get(policy_key, {}))
payload = settings.SerializeToString()
elif dp is not None and msg.policy_type == 'google/chromeos/device':
settings = dp.ChromeDeviceSettingsProto()
payload = self.server.ReadPolicyFromDataDir(policy_key, settings)
if payload is None:
self.GatherDevicePolicySettings(settings, policy.get(policy_key, {}))
payload = settings.SerializeToString()
elif msg.policy_type == 'google/chrome/extension':
settings = ep.ExternalPolicyData()
payload = self.server.ReadPolicyFromDataDir(policy_key, settings)
if payload is None:
payload = self.CreatePolicyForExternalPolicyData(policy_key)
else:
response.error_code = 400
response.error_message = 'Invalid policy type'
return
else:
response.error_code = 400
response.error_message = 'Request not allowed for the token used'
return
# Sign with 'current_key_index', defaulting to key 0.
signing_key = None
req_key = None
current_key_index = policy.get('current_key_index', 0)
nkeys = len(self.server.keys)
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']
# 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 = payload
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
policy_data.service_account_identity = policy.get(
'service_account_identity',
'policy_testserver.py-service_account_identity')
invalidation_source = policy.get('invalidation_source')
if invalidation_source is not None:
policy_data.invalidation_source = invalidation_source
# Since invalidation_name is type bytes in the proto, the Unicode name
# provided needs to be encoded as ASCII to set the correct byte pattern.
invalidation_name = policy.get('invalidation_name')
if invalidation_name is not None:
policy_data.invalidation_name = invalidation_name.encode('ascii')
if signing_key:
policy_data.public_key_version = current_key_index + 1
if username:
policy_data.username = username
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.policy_data = signed_data
if signing_key:
response.policy_data_signature = (
bytes(signing_key['private_key'].hashAndSign(signed_data)))
if msg.public_key_version != current_key_index + 1:
response.new_public_key = signing_key['public_key']
# Set the verification signature appropriate for the policy domain.
# TODO(atwilson): Use the enrollment domain for public accounts when
# we add key validation for ChromeOS (http://crbug.com/328038).
if 'signatures' in signing_key:
verification_sig = self.GetSignatureForDomain(
signing_key['signatures'], policy_data.username)
if verification_sig:
assert len(verification_sig) == 256, \
'bad signature size: %d' % len(verification_sig)
response.new_public_key_verification_signature = verification_sig
if req_key:
response.new_public_key_signature = (
bytes(req_key.hashAndSign(response.new_public_key)))
return (200, response.SerializeToString())
def GetSignatureForDomain(self, signatures, username):
parsed_username = username.split("@", 1)
if len(parsed_username) != 2:
logging.error('Could not extract domain from username: %s' % username)
return None
domain = parsed_username[1]
# Lookup the domain's signature in the passed dictionary. If none is found,
# fallback to a wildcard signature.
if domain in signatures:
return signatures[domain]
if '*' in signatures:
return signatures['*']
# No key matching this domain.
logging.error('No verification signature matching domain: %s' % domain)
return None
def CheckToken(self):
"""Helper for checking whether the client supplied a valid DM token.
Extracts the token from the request and passed to the server in order to
look up the client.
Returns:
A pair of token information record and error response. If the first
element is None, then the second contains an error code to send back to
the client. Otherwise the first element is the same structure that is
returned by LookupToken().
"""
error = 500
dmtoken = None
request_device_id = self.GetUniqueParam('deviceid')
match = re.match('GoogleDMToken token=(\\w+)',
self.headers.getheader('Authorization', ''))
if match:
dmtoken = match.group(1)
if not dmtoken:
error = 401
else:
token_info = self.server.LookupToken(dmtoken)
if (not token_info or
not request_device_id or
token_info['device_id'] != request_device_id):
error = 410
else:
return (token_info, None)
logging.debug('Token check failed with error %d' % error)
return (None, (error, 'Server error %d' % error))
def DumpMessage(self, label, msg):
"""Helper for logging an ASCII dump of a protobuf message."""
logging.debug('%s\n%s' % (label, str(msg)))
class PolicyTestServer(testserver_base.BrokenPipeHandlerMixIn,
testserver_base.StoppableHTTPServer):
"""Handles requests and keeps global service state."""
def __init__(self, server_address, data_dir, policy_path, client_state_file,
private_key_paths, server_base_url):
"""Initializes the server.
Args:
server_address: Server host and port.
policy_path: Names the file to read JSON-formatted policy from.
private_key_paths: List of paths to read private keys from.
"""
testserver_base.StoppableHTTPServer.__init__(self, server_address,
PolicyRequestHandler)
self._registered_tokens = {}
self.data_dir = data_dir
self.policy_path = policy_path
self.client_state_file = client_state_file
self.server_base_url = server_base_url
self.keys = []
if private_key_paths:
# Load specified keys from the filesystem.
for key_path in private_key_paths:
try:
key_str = open(key_path).read()
except IOError:
print 'Failed to load private key from %s' % key_path
continue
try:
key = tlslite.api.parsePEMKey(key_str, private=True)
except SyntaxError:
key = tlslite.utils.python_rsakey.Python_RSAKey._parsePKCS8(
bytearray(key_str))
assert key is not None
key_info = { 'private_key' : key }
# Now try to read in a signature, if one exists.
try:
key_sig = open(key_path + '.sig').read()
# Create a dictionary with the wildcard domain + signature
key_info['signatures'] = {'*': key_sig}
except IOError:
print 'Failed to read validation signature from %s.sig' % key_path
self.keys.append(key_info)
else:
# Use the canned private keys if none were passed from the command line.
for signing_key in SIGNING_KEYS:
decoded_key = base64.b64decode(signing_key['key']);
key = tlslite.utils.python_rsakey.Python_RSAKey._parsePKCS8(
bytearray(decoded_key))
assert key is not None
# Grab the signature dictionary for this key and decode all of the
# signatures.
signature_dict = signing_key['signatures']
decoded_signatures = {}
for domain in signature_dict:
decoded_signatures[domain] = base64.b64decode(signature_dict[domain])
self.keys.append({'private_key': key,
'signatures': decoded_signatures})
# 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
# Load client state.
if self.client_state_file is not None:
try:
file_contents = open(self.client_state_file).read()
self._registered_tokens = json.loads(file_contents, strict=False)
except IOError:
pass
def GetPolicies(self):
"""Returns the policies to be used, reloaded form the backend file every
time this is called.
"""
policy = {}
if json is None:
print 'No JSON module, cannot parse policy information'
else :
try:
policy = json.loads(open(self.policy_path).read(), strict=False)
except IOError:
print 'Failed to load policy from %s' % self.policy_path
return policy
def RegisterDevice(self, device_id, machine_id, type):
"""Registers a device or user and generates a DM token for it.
Args:
device_id: The device identifier provided by the client.
Returns:
The newly generated device token for the device.
"""
dmtoken_chars = []
while len(dmtoken_chars) < 32:
dmtoken_chars.append(random.choice('0123456789abcdef'))
dmtoken = ''.join(dmtoken_chars)
allowed_policy_types = {
dm.DeviceRegisterRequest.BROWSER: [
'google/chrome/user',
'google/chrome/extension'
],
dm.DeviceRegisterRequest.USER: [
'google/chromeos/user',
'google/chrome/extension'
],
dm.DeviceRegisterRequest.DEVICE: [
'google/chromeos/device',
'google/chromeos/publicaccount',
'google/chrome/extension'
],
dm.DeviceRegisterRequest.ANDROID_BROWSER: [
'google/android/user'
],
dm.DeviceRegisterRequest.IOS_BROWSER: [
'google/ios/user'
],
dm.DeviceRegisterRequest.TT: ['google/chromeos/user',
'google/chrome/user'],
}
if machine_id in KIOSK_MACHINE_IDS:
enrollment_mode = dm.DeviceRegisterResponse.RETAIL
else:
enrollment_mode = dm.DeviceRegisterResponse.ENTERPRISE
self._registered_tokens[dmtoken] = {
'device_id': device_id,
'device_token': dmtoken,
'allowed_policy_types': allowed_policy_types[type],
'machine_name': 'chromeos-' + machine_id,
'machine_id': machine_id,
'enrollment_mode': enrollment_mode,
}
self.WriteClientState()
return self._registered_tokens[dmtoken]
def UpdateMachineId(self, dmtoken, machine_id):
"""Updates the machine identifier for a registered device.
Args:
dmtoken: The device management token provided by the client.
machine_id: Updated hardware identifier value.
"""
if dmtoken in self._registered_tokens:
self._registered_tokens[dmtoken]['machine_id'] = machine_id
self.WriteClientState()
def UpdateStateKeys(self, dmtoken, state_keys):
"""Updates the state keys for a given client.
Args:
dmtoken: The device management token provided by the client.
state_keys: The state keys to set.
"""
if dmtoken in self._registered_tokens:
self._registered_tokens[dmtoken]['state_keys'] = map(
lambda key : key.encode('hex'), state_keys)
self.WriteClientState()
def LookupToken(self, dmtoken):
"""Looks up a device or a user by DM token.
Args:
dmtoken: The device management token provided by the client.
Returns:
A dictionary with information about a device or user that is registered by
dmtoken, or None if the token is not found.
"""
return self._registered_tokens.get(dmtoken, None)
def LookupByStateKey(self, state_key):
"""Looks up a device or a user by a state key.
Args:
state_key: The state key provided by the client.
Returns:
A dictionary with information about a device or user or None if there is
no matching record.
"""
for client in self._registered_tokens.values():
if state_key.encode('hex') in client.get('state_keys', []):
return client
return None
def GetMatchingStateKeyHashes(self, modulus, remainder):
"""Returns all clients registered with the server.
Returns:
The list of registered clients.
"""
state_keys = sum([ c.get('state_keys', [])
for c in self._registered_tokens.values() ], [])
hashed_keys = map(lambda key: hashlib.sha256(key.decode('hex')).digest(),
set(state_keys))
return filter(
lambda hash : int(hash.encode('hex'), 16) % modulus == remainder,
hashed_keys)
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]
self.WriteClientState()
def WriteClientState(self):
"""Writes the client state back to the file."""
if self.client_state_file is not None:
json_data = json.dumps(self._registered_tokens)
open(self.client_state_file, 'w').write(json_data)
def GetBaseFilename(self, policy_selector):
"""Returns the base filename for the given policy_selector.
Args:
policy_selector: The policy type and settings entity id, joined by '/'.
Returns:
The filename corresponding to the policy_selector, without a file
extension.
"""
sanitized_policy_selector = re.sub('[^A-Za-z0-9.@-]', '_', policy_selector)
return os.path.join(self.data_dir or '',
'policy_%s' % sanitized_policy_selector)
def ListMatchingComponents(self, policy_type):
"""Returns a list of settings entity IDs that have a configuration file.
Args:
policy_type: The policy type to look for. Only settings entity IDs for
file selectors That match this policy_type will be returned.
Returns:
A list of settings entity IDs for the given |policy_type| that have a
configuration file in this server (either as a .bin, .txt or .data file).
"""
base_name = self.GetBaseFilename(policy_type)
files = glob.glob('%s_*.*' % base_name)
len_base_name = len(base_name) + 1
return [ file[len_base_name:file.rfind('.')] for file in files ]
def ReadPolicyFromDataDir(self, policy_selector, proto_message):
"""Tries to read policy payload from a file in the data directory.
First checks for a binary rendition of the policy protobuf in
<data_dir>/policy_<sanitized_policy_selector>.bin. If that exists, returns
it. If that file doesn't exist, tries
<data_dir>/policy_<sanitized_policy_selector>.txt and decodes that as a
protobuf using proto_message. If that fails as well, returns None.
Args:
policy_selector: Selects which policy to read.
proto_message: Optional protobuf message object used for decoding the
proto text format.
Returns:
The binary payload message, or None if not found.
"""
base_filename = self.GetBaseFilename(policy_selector)
# Try the binary payload file first.
try:
return open(base_filename + '.bin').read()
except IOError:
pass
# If that fails, try the text version instead.
if proto_message is None:
return None
try:
text = open(base_filename + '.txt').read()
google.protobuf.text_format.Merge(text, proto_message)
return proto_message.SerializeToString()
except IOError:
return None
except google.protobuf.text_format.ParseError:
return None
def ReadPolicyDataFromDataDir(self, policy_selector):
"""Returns the external policy data for |policy_selector| if found.
Args:
policy_selector: Selects which policy to read.
Returns:
The data for the corresponding policy type and entity id, if found.
"""
base_filename = self.GetBaseFilename(policy_selector)
try:
return open(base_filename + '.data').read()
except IOError:
return None
def GetBaseURL(self):
"""Returns the server base URL.
Respects the |server_base_url| configuration parameter, if present. Falls
back to construct the URL from the server hostname and port otherwise.
Returns:
The URL to use for constructing URLs that get returned to clients.
"""
base_url = self.server_base_url
if base_url is None:
base_url = 'http://%s:%s' % self.server_address[:2]
return base_url
class PolicyServerRunner(testserver_base.TestServerRunner):
def __init__(self):
super(PolicyServerRunner, self).__init__()
def create_server(self, server_data):
data_dir = self.options.data_dir or ''
config_file = (self.options.config_file or
os.path.join(data_dir, 'device_management'))
server = PolicyTestServer((self.options.host, self.options.port),
data_dir, config_file,
self.options.client_state_file,
self.options.policy_keys,
self.options.server_base_url)
server_data['port'] = server.server_port
return server
def add_options(self):
testserver_base.TestServerRunner.add_options(self)
self.option_parser.add_option('--client-state', dest='client_state_file',
help='File that client state should be '
'persisted to. This allows the server to be '
'seeded by a list of pre-registered clients '
'and restarts without abandoning registered '
'clients.')
self.option_parser.add_option('--policy-key', action='append',
dest='policy_keys',
help='Specify a path to a PEM-encoded '
'private key to use for policy signing. May '
'be specified multiple times in order to '
'load multiple keys into the server. If the '
'server has multiple keys, it will rotate '
'through them in at each request in a '
'round-robin fashion. The server will '
'use a canned key if none is specified '
'on the command line. The test server will '
'also look for a verification signature file '
'in the same location: <filename>.sig and if '
'present will add the signature to the '
'policy blob as appropriate via the '
'new_public_key_verification_signature '
'field.')
self.option_parser.add_option('--log-level', dest='log_level',
default='WARN',
help='Log level threshold to use.')
self.option_parser.add_option('--config-file', dest='config_file',
help='Specify a configuration file to use '
'instead of the default '
'<data_dir>/device_management')
self.option_parser.add_option('--server-base-url', dest='server_base_url',
help='The server base URL to use when '
'constructing URLs to return to the client.')
def run_server(self):
logger = logging.getLogger()
logger.setLevel(getattr(logging, str(self.options.log_level).upper()))
if (self.options.log_to_console):
logger.addHandler(logging.StreamHandler())
if (self.options.log_file):
logger.addHandler(logging.FileHandler(self.options.log_file))
testserver_base.TestServerRunner.run_server(self)
if __name__ == '__main__':
sys.exit(PolicyServerRunner().main())