| # Copyright 2019 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """The payload API is the entry point for payload functionality.""" |
| |
| from copy import deepcopy |
| import os |
| import re |
| from typing import Optional, Tuple, Union |
| |
| from chromite.api.gen.chromite.api import payload_pb2 |
| from chromite.api.gen.chromiumos import common_pb2 |
| from chromite.lib import chroot_util |
| from chromite.lib.paygen import gspaths |
| from chromite.lib.paygen import paygen_build_lib |
| from chromite.lib.paygen import paygen_payload_lib |
| |
| |
| class Error(Exception): |
| """Base module error.""" |
| |
| |
| class ImageTypeUnknownError(Error): |
| """An error raised when image type is unknown.""" |
| |
| |
| class ImageMismatchError(Error): |
| """An error raised when src and tgt aren't compatible.""" |
| |
| |
| class PayloadConfig(object): |
| """Value object to hold the GeneratePayload configuration options.""" |
| |
| def __init__(self, |
| tgt_image: Optional[Union[payload_pb2.UnsignedImage, |
| payload_pb2.SignedImage, |
| payload_pb2.DLCImage]] = None, |
| src_image: Optional[Union[payload_pb2.UnsignedImage, |
| payload_pb2.SignedImage, |
| payload_pb2.DLCImage]] = None, |
| dest_bucket: Optional[str] = None, |
| minios: bool = False, |
| verify: bool = True, |
| upload: bool = True, |
| cache_dir: Optional[str] = None): |
| """Init method, sets up all the paths and configuration. |
| |
| Args: |
| tgt_image: Proto for destination image. |
| src_image: Proto for source image. |
| dest_bucket: Destination bucket to place the final artifacts in. |
| minios: Whether the payload is for the image's miniOS partition. |
| verify: If delta is made, verify the integrity of the payload. |
| upload: Whether the payload generation results should be uploaded. |
| cache_dir: The cache dir for paygen to use or None for default. |
| """ |
| |
| # Set when we call GeneratePayload on this object. |
| self.paygen = None |
| self.tgt_image = tgt_image |
| self.src_image = src_image |
| self.dest_bucket = dest_bucket |
| self.minios = minios |
| self.verify = verify |
| self.upload = upload |
| self.delta_type = 'delta' if self.src_image else 'full' |
| self.image_type = _ImageTypeToStr(tgt_image.image_type) |
| self.cache_dir = cache_dir |
| |
| # This block ensures that we have paths to the correct perm of images. |
| src_image_path = None |
| if isinstance(self.tgt_image, payload_pb2.UnsignedImage): |
| tgt_image_path = _GenUnsignedGSPath(self.tgt_image, self.image_type, |
| self.minios) |
| elif isinstance(self.tgt_image, payload_pb2.SignedImage): |
| tgt_image_path = _GenSignedGSPath(self.tgt_image, self.image_type, |
| self.minios) |
| elif isinstance(self.tgt_image, payload_pb2.DLCImage): |
| tgt_image_path = _GenDLCImageGSPath(self.tgt_image) |
| if self.delta_type == 'delta': |
| if isinstance(self.tgt_image, payload_pb2.UnsignedImage): |
| src_image_path = _GenUnsignedGSPath(self.src_image, self.image_type, |
| self.minios) |
| if isinstance(self.tgt_image, payload_pb2.SignedImage): |
| src_image_path = _GenSignedGSPath(self.src_image, self.image_type, |
| self.minios) |
| elif isinstance(self.tgt_image, payload_pb2.DLCImage): |
| src_image_path = _GenDLCImageGSPath(self.src_image) |
| |
| payload_build = deepcopy(tgt_image_path.build) |
| payload_build.bucket = dest_bucket |
| |
| self.payload = gspaths.Payload( |
| build=payload_build, |
| tgt_image=tgt_image_path, |
| src_image=src_image_path, |
| minios=self.minios, |
| uri=None) |
| |
| if self.upload: |
| self.payload.uri = paygen_build_lib.DefaultPayloadUri(self.payload) |
| |
| def GeneratePayload(self) -> Tuple[str, str]: |
| """Do payload generation (& maybe sign) on Google Storage CrOS images. |
| |
| Returns: |
| A tuple of containing: |
| The location of the local generated artifact. |
| (e.g. /tmp/wdjaio/delta.bin) |
| The remote location that the payload was uploaded or None. |
| (e.g. 'gs://cr/beta-channel/coral/12345.0.1/payloads/...') |
| |
| Raises: |
| paygen_payload_lib.PayloadGenerationSkippedException: If paygen was |
| skipped for any reason. |
| """ |
| # Leave the generated artifact local. This is ok because if we're testing |
| # it's likely we want the artifact anyway, and in production this is ran on |
| # single shot bots in the context of an overlayfs and will get cleaned up |
| # anyway. |
| with chroot_util.TempDirInChroot(delete=False) as temp_dir: |
| signer = paygen_payload_lib.PaygenSigner( |
| work_dir=temp_dir, payload_build=self.payload.build) |
| self.paygen = paygen_payload_lib.PaygenPayload( |
| self.payload, |
| temp_dir, |
| signer=signer, |
| verify=self.verify, |
| upload=self.upload, |
| cache_dir=self.cache_dir) |
| |
| # We run and expect failures to raise, so if we get passed |
| # self.paygen.Run() then it's safe to assume the bin is in place. |
| local_path = os.path.join(temp_dir, 'delta.bin') |
| remote_uri = self.paygen.Run() |
| return (local_path, remote_uri) |
| |
| |
| def _ImageTypeToStr(image_type_n: int) -> str: |
| """The numeral image type enum in proto to lowercase string.""" |
| ret = common_pb2.ImageType.Name(image_type_n).lower() |
| return re.sub('^image_type_', '', ret) |
| |
| |
| def _GenSignedGSPath(image: payload_pb2.SignedImage, |
| image_type: str, |
| minios: bool) -> gspaths.Image: |
| """Take a SignedImage_pb2 and return a gspaths.Image. |
| |
| Args: |
| image: The build to create the gspath from. |
| image_type: The image type, either "recovery" or "base". |
| minios: Whether or not it's a miniOS image. |
| |
| Returns: |
| A gspaths.Image instance. |
| """ |
| build = gspaths.Build(board=image.build.build_target.name, |
| version=image.build.version, |
| channel=image.build.channel, |
| bucket=image.build.bucket) |
| |
| build_uri = gspaths.ChromeosReleases.ImageUri( |
| build, image.key, image_type) |
| |
| build.uri = build_uri |
| |
| if minios: |
| return gspaths.MiniOSImage(build=build, |
| image_type=image_type, |
| key=image.key, |
| uri=build_uri) |
| else: |
| return gspaths.Image(build=build, |
| image_type=image_type, |
| key=image.key, |
| uri=build_uri) |
| |
| |
| def _GenUnsignedGSPath(image: payload_pb2.UnsignedImage, |
| image_type: str, |
| minios: bool) -> gspaths.UnsignedImageArchive: |
| """Take an UnsignedImage_pb2 and return a gspaths.UnsignedImageArchive. |
| |
| Args: |
| image: The build to create the gspath from. |
| image_type: The image type, either "recovery" or "test". |
| minios: Whether or not it's a miniOS image. |
| |
| Returns: |
| A gspaths.UnsignedImageArchive instance. |
| """ |
| build = gspaths.Build(board=image.build.build_target.name, |
| version=image.build.version, |
| channel=image.build.channel, |
| bucket=image.build.bucket) |
| |
| build_uri = gspaths.ChromeosReleases.UnsignedImageUri( |
| build, image.milestone, image_type) |
| |
| build.uri = build_uri |
| |
| if minios: |
| return gspaths.UnsignedMiniOSImageArchive(build=build, |
| milestone=image.milestone, |
| image_type=image_type, |
| uri=build_uri) |
| else: |
| return gspaths.UnsignedImageArchive(build=build, |
| milestone=image.milestone, |
| image_type=image_type, |
| uri=build_uri) |
| |
| |
| def _GenDLCImageGSPath(image: payload_pb2.DLCImage) -> gspaths.DLCImage: |
| """Take a DLCImage_pb2 and return a gspaths.DLCImage. |
| |
| Args: |
| image: The dlc image to create the gspath from. |
| |
| Returns: |
| A gspaths.DLCImage instance. |
| """ |
| build = gspaths.Build( |
| board=image.build.build_target.name, |
| version=image.build.version, |
| channel=image.build.channel, |
| bucket=image.build.bucket) |
| |
| dlc_image_uri = gspaths.ChromeosReleases.DLCImageUri(build, image.dlc_id, |
| image.dlc_package, |
| image.dlc_image) |
| |
| return gspaths.DLCImage( |
| build=build, |
| image_type=image.image_type, |
| key='', |
| uri=dlc_image_uri, |
| dlc_id=image.dlc_id, |
| dlc_package=image.dlc_package, |
| dlc_image=image.dlc_image) |
