1#!/usr/bin/env python3
2#
3#   Copyright 2017 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import time
18from zipfile import ZipFile
19
20"""The setup time in seconds."""
21SL4A_SERVICE_SETUP_TIME = 5
22
23
24"""The path to the metadata found within the OTA package."""
25OTA_PACKAGE_METADATA_PATH = 'META-INF/com/android/metadata'
26
27
28class OtaError(Exception):
29    """Raised when an error in the OTA Update process occurs."""
30
31
32class InvalidOtaUpdateError(OtaError):
33    """Raised when the update from one version to another is not valid."""
34
35
36class OtaRunner(object):
37    """The base class for all OTA Update Runners."""
38
39    def __init__(self, ota_tool, android_device):
40        self.ota_tool = ota_tool
41        self.android_device = android_device
42        self.serial = self.android_device.serial
43
44    def _update(self):
45        post_build_id = self.get_post_build_id()
46        log = self.android_device.log
47        old_info = self.android_device.adb.getprop('ro.build.fingerprint')
48        log.info('Starting Update. Beginning build info: %s', old_info)
49        log.info('Stopping services.')
50        self.android_device.stop_services()
51        log.info('Beginning tool.')
52        self.ota_tool.update(self)
53        log.info('Tool finished. Waiting for boot completion.')
54        self.android_device.wait_for_boot_completion()
55        new_info = self.android_device.adb.getprop('ro.build.fingerprint')
56        if not old_info or old_info == new_info:
57            raise OtaError('The device was not updated to a new build. '
58                           'Previous build: %s. Current build: %s. '
59                           'Expected build: %s' % (old_info, new_info,
60                                                   post_build_id))
61        log.info('Boot completed. Rooting adb.')
62        self.android_device.root_adb()
63        log.info('Root complete.')
64        if self.android_device.skip_sl4a:
65            self.android_device.log.info('Skipping SL4A install.')
66        else:
67            for _ in range(3):
68                self.android_device.log.info('Re-installing SL4A from "%s".',
69                                             self.get_sl4a_apk())
70                self.android_device.adb.install(
71                    '-r -g %s' % self.get_sl4a_apk(), ignore_status=True)
72                time.sleep(SL4A_SERVICE_SETUP_TIME)
73                if self.android_device.is_sl4a_installed():
74                    break
75        log.info('Starting services.')
76        self.android_device.start_services()
77        self.android_device.update_sdk_api_level()
78        log.info('Services started. Running ota tool cleanup.')
79        self.ota_tool.cleanup(self)
80        log.info('Cleanup complete.')
81
82    def get_ota_package_metadata(self, requested_field):
83        """Returns a variable found within the OTA package's metadata.
84
85        Args:
86            requested_field: the name of the metadata field
87
88        Will return None if the variable cannot be found.
89        """
90        ota_zip = ZipFile(self.get_ota_package(), 'r')
91        if OTA_PACKAGE_METADATA_PATH in ota_zip.namelist():
92            with ota_zip.open(OTA_PACKAGE_METADATA_PATH) as metadata:
93                timestamp_line = requested_field.encode('utf-8')
94                timestamp_offset = len(timestamp_line) + 1
95
96                for line in metadata.readlines():
97                    if line.startswith(timestamp_line):
98                        return line[timestamp_offset:].decode('utf-8').strip()
99        return None
100
101    def validate_update(self):
102        """Raises an error if updating to the next build is not valid.
103
104        Raises:
105            InvalidOtaUpdateError if the ota version is not valid, or cannot be
106                validated.
107        """
108        # The timestamp the current device build was created at.
109        cur_img_timestamp = self.android_device.adb.getprop('ro.build.date.utc')
110        ota_img_timestamp = self.get_ota_package_metadata('post-timestamp')
111
112        if ota_img_timestamp is None:
113            raise InvalidOtaUpdateError('Unable to find the timestamp '
114                                        'for the OTA build.')
115
116        try:
117            if int(ota_img_timestamp) <= int(cur_img_timestamp):
118                cur_fingerprint = self.android_device.adb.getprop(
119                    'ro.bootimage.build.fingerprint')
120                ota_fingerprint = self.get_post_build_id()
121                raise InvalidOtaUpdateError(
122                    'The OTA image comes from an earlier build than the '
123                    'source build. Current build: Time: %s -- %s, '
124                    'OTA build: Time: %s -- %s' %
125                    (cur_img_timestamp, cur_fingerprint,
126                     ota_img_timestamp, ota_fingerprint))
127        except ValueError:
128            raise InvalidOtaUpdateError(
129                'Unable to parse timestamps. Current timestamp: %s, OTA '
130                'timestamp: %s' % (ota_img_timestamp, cur_img_timestamp))
131
132    def get_post_build_id(self):
133        """Returns the post-build ID found within the OTA package metadata.
134
135        Raises:
136            InvalidOtaUpdateError if the post-build ID cannot be found.
137        """
138        return self.get_ota_package_metadata('post-build')
139
140    def can_update(self):
141        """Whether or not an update package is available for the device."""
142        return NotImplementedError()
143
144    def get_ota_package(self):
145        raise NotImplementedError()
146
147    def get_sl4a_apk(self):
148        raise NotImplementedError()
149
150
151class SingleUseOtaRunner(OtaRunner):
152    """A single use OtaRunner.
153
154    SingleUseOtaRunners can only be ran once. If a user attempts to run it more
155    than once, an error will be thrown. Users can avoid the error by checking
156    can_update() before calling update().
157    """
158
159    def __init__(self, ota_tool, android_device, ota_package, sl4a_apk):
160        super(SingleUseOtaRunner, self).__init__(ota_tool, android_device)
161        self._ota_package = ota_package
162        self._sl4a_apk = sl4a_apk
163        self._called = False
164
165    def can_update(self):
166        return not self._called
167
168    def update(self):
169        """Starts the update process."""
170        if not self.can_update():
171            raise OtaError('A SingleUseOtaTool instance cannot update a device '
172                           'multiple times.')
173        self._called = True
174        self._update()
175
176    def get_ota_package(self):
177        return self._ota_package
178
179    def get_sl4a_apk(self):
180        return self._sl4a_apk
181
182
183class MultiUseOtaRunner(OtaRunner):
184    """A multiple use OtaRunner.
185
186    MultiUseOtaRunner can only be ran for as many times as there have been
187    packages provided to them. If a user attempts to run it more than the number
188    of provided packages, an error will be thrown. Users can avoid the error by
189    checking can_update() before calling update().
190    """
191
192    def __init__(self, ota_tool, android_device, ota_packages, sl4a_apks):
193        super(MultiUseOtaRunner, self).__init__(ota_tool, android_device)
194        self._ota_packages = ota_packages
195        self._sl4a_apks = sl4a_apks
196        self.current_update_number = 0
197
198    def can_update(self):
199        return not self.current_update_number == len(self._ota_packages)
200
201    def update(self):
202        """Starts the update process."""
203        if not self.can_update():
204            raise OtaError('This MultiUseOtaRunner has already updated all '
205                           'given packages onto the phone.')
206        self._update()
207        self.current_update_number += 1
208
209    def get_ota_package(self):
210        return self._ota_packages[self.current_update_number]
211
212    def get_sl4a_apk(self):
213        return self._sl4a_apks[self.current_update_number]
214