tools/telemetry/third_party/webpagereplay/adb_install_cert.py

   1 # Copyright 2014 Google Inc. All Rights Reserved.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 """Installs certificate on phone with KitKat."""
  16
  17 import argparse
  18 import logging
  19 import os
  20 import subprocess
  21 import sys
  22
  23 KEYCODE_ENTER = '66'
  24 KEYCODE_TAB = '61'
  25
  26 class CertInstallError(Exception):
  27   pass
  28
  29 class CertRemovalError(Exception):
  30   pass
  31
  32
  33 class AndroidCertInstaller(object):
  34   """Certificate installer for phones with KitKat."""
  35
  36   def __init__(self, device_id, cert_name, cert_path):
  37     if not os.path.exists(cert_path):
  38       raise ValueError('Not a valid certificate path')
  39     self.device_id = device_id
  40     self.cert_name = cert_name
  41     self.cert_path = cert_path
  42     self.file_name = os.path.basename(self.cert_path)
  43     self.reformatted_cert_fname = None
  44     self.reformatted_cert_path = None
  45     self.android_cacerts_path = None
  46
  47   @staticmethod
  48   def _run_cmd(cmd, dirname=None):
  49     return subprocess.check_output(cmd, cwd=dirname)
  50
  51   def _adb(self, *args):
  52     """Runs the adb command."""
  53     cmd = ['adb']
  54     if self.device_id:
  55       cmd.extend(['-s', self.device_id])
  56     cmd.extend(args)
  57     return self._run_cmd(cmd)
  58
  59   def _adb_su_shell(self, *args):
  60     """Runs command as root."""
  61     cmd = ['shell', 'su', '-c']
  62     cmd.extend(args)
  63     return self._adb(*cmd)
  64
  65   def _get_property(self, prop):
  66     return self._adb('shell', 'getprop', prop).strip()
  67
  68   def check_device(self):
  69     install_warning = False
  70     if self._get_property('ro.product.device') != 'hammerhead':
  71       logging.warning('Device is not hammerhead')
  72       install_warning = True
  73     if self._get_property('ro.build.version.release') != '4.4.2':
  74       logging.warning('Version is not 4.4.2')
  75       install_warning = True
  76     if install_warning:
  77       logging.warning('Certificate may not install properly')
  78
  79   def _input_key(self, key):
  80     """Inputs a keyevent."""
  81     self._adb('shell', 'input', 'keyevent', key)
  82
  83   def _input_text(self, text):
  84     """Inputs text."""
  85     self._adb('shell', 'input', 'text', text)
  86
  87   @staticmethod
  88   def _remove(file_name):
  89     """Deletes file."""
  90     if os.path.exists(file_name):
  91       os.remove(file_name)
  92
  93   def _format_hashed_cert(self):
  94     """Makes a certificate file that follows the format of files in cacerts."""
  95     self._remove(self.reformatted_cert_path)
  96     contents = self._run_cmd(['openssl', 'x509', '-inform', 'PEM', '-text',
  97                               '-in', self.cert_path])
  98     description, begin_cert, cert_body = contents.rpartition('-----BEGIN '
  99                                                              'CERTIFICATE')
 100     contents = ''.join([begin_cert, cert_body, description])
 101     with open(self.reformatted_cert_path, 'w') as cert_file:
 102       cert_file.write(contents)
 103
 104   def _remove_cert_from_cacerts(self):
 105     self._adb_su_shell('mount', '-o', 'remount,rw', '/system')
 106     self._adb_su_shell('rm', self.android_cacerts_path)
 107
 108   def _is_cert_installed(self):
 109     return (self._adb_su_shell('ls', self.android_cacerts_path).strip() ==
 110             self.android_cacerts_path)
 111
 112   def _generate_reformatted_cert_path(self):
 113     # Determine OpenSSL version, string is of the form
 114     # 'OpenSSL 0.9.8za 5 Jun 2014' .
 115     openssl_version = self._run_cmd(['openssl', 'version']).split()
 116
 117     if len(openssl_version) < 2:
 118       raise ValueError('Unexpected OpenSSL version string: ', openssl_version)
 119
 120     # subject_hash flag name changed as of OpenSSL version 1.0.0 .
 121     is_old_openssl_version = openssl_version[1].startswith('0')
 122     subject_hash_flag = (
 123         '-subject_hash' if is_old_openssl_version else '-subject_hash_old')
 124
 125     output = self._run_cmd(['openssl', 'x509', '-inform', 'PEM',
 126                             subject_hash_flag, '-in', self.cert_path],
 127                            os.path.dirname(self.cert_path))
 128     self.reformatted_cert_fname = output.partition('\n')[0].strip() + '.0'
 129     self.reformatted_cert_path = os.path.join(os.path.dirname(self.cert_path),
 130                                               self.reformatted_cert_fname)
 131     self.android_cacerts_path = ('/system/etc/security/cacerts/%s' %
 132                                  self.reformatted_cert_fname)
 133
 134   def remove_cert(self):
 135     self._generate_reformatted_cert_path()
 136
 137     if self._is_cert_installed():
 138       self._remove_cert_from_cacerts()
 139
 140     if self._is_cert_installed():
 141       raise CertRemovalError('Cert Removal Failed')
 142
 143   def install_cert(self, overwrite_cert=False):
 144     """Installs a certificate putting it in /system/etc/security/cacerts."""
 145     self._generate_reformatted_cert_path()
 146
 147     if self._is_cert_installed():
 148       if overwrite_cert:
 149         self._remove_cert_from_cacerts()
 150       else:
 151         logging.info('cert is already installed')
 152         return
 153
 154     self._format_hashed_cert()
 155     self._adb('push', self.reformatted_cert_path, '/sdcard/')
 156     self._remove(self.reformatted_cert_path)
 157     self._adb_su_shell('mount', '-o', 'remount,rw', '/system')
 158     self._adb_su_shell(
 159         'cp', '/sdcard/%s' % self.reformatted_cert_fname,
 160         '/system/etc/security/cacerts/%s' % self.reformatted_cert_fname)
 161     self._adb_su_shell('chmod', '644', self.android_cacerts_path)
 162     if not self._is_cert_installed():
 163       raise CertInstallError('Cert Install Failed')
 164
 165   def install_cert_using_gui(self):
 166     """Installs certificate on the device using adb commands."""
 167     self.check_device()
 168     # TODO(mruthven): Add a check to see if the certificate is already installed
 169     # Install the certificate.
 170     logging.info('Installing %s on %s', self.cert_path, self.device_id)
 171     self._adb('push', self.cert_path, '/sdcard/')
 172
 173     # Start credential install intent.
 174     self._adb('shell', 'am', 'start', '-W', '-a', 'android.credentials.INSTALL')
 175
 176     # Move to and click search button.
 177     self._input_key(KEYCODE_TAB)
 178     self._input_key(KEYCODE_TAB)
 179     self._input_key(KEYCODE_ENTER)
 180
 181     # Search for certificate and click it.
 182     # Search only works with lower case letters
 183     self._input_text(self.file_name.lower())
 184     self._input_key(KEYCODE_ENTER)
 185
 186     # These coordinates work for hammerhead devices.
 187     self._adb('shell', 'input', 'tap', '300', '300')
 188
 189     # Name the certificate and click enter.
 190     self._input_text(self.cert_name)
 191     self._input_key(KEYCODE_TAB)
 192     self._input_key(KEYCODE_TAB)
 193     self._input_key(KEYCODE_TAB)
 194     self._input_key(KEYCODE_ENTER)
 195
 196     # Remove the file.
 197     self._adb('shell', 'rm', '/sdcard/' + self.file_name)
 198
 199
 200 def parse_args():
 201   """Parses command line arguments."""
 202   parser = argparse.ArgumentParser(description='Install cert on device.')
 203   parser.add_argument(
 204       '-n', '--cert-name', default='dummycert', help='certificate name')
 205   parser.add_argument(
 206       '--overwrite', default=False, action='store_true',
 207       help='Overwrite certificate file if it is already installed')
 208   parser.add_argument(
 209       '--remove', default=False, action='store_true',
 210       help='Remove certificate file if it is installed')
 211   parser.add_argument(
 212       '--device-id', help='device serial number')
 213   parser.add_argument(
 214       'cert_path', help='Certificate file path')
 215   return parser.parse_args()
 216
 217
 218 def main():
 219   args = parse_args()
 220   cert_installer = AndroidCertInstaller(args.device_id, args.cert_name,
 221                                         args.cert_path)
 222   if args.remove:
 223     cert_installer.remove_cert()
 224   else:
 225     cert_installer.install_cert(args.overwrite)
 226
 227
 228 if __name__ == '__main__':
 229   sys.exit(main())