Move some OTA signing functions to a separte file

This makes it easier for other otatools to re-use these logic without
having to pull in lots of dependencies.

Test: th
Bug: 227848550

Change-Id: I81ed01c5cea4b934a074650731b6f89752221de9

tools/releasetools/Android.bp

@@ -151,6 +151,7 @@ python_defaults {
         "non_ab_ota.py",
         "ota_from_target_files.py",
         "ota_utils.py",
+        "payload_signer.py",
         "target_files_diff.py",
     ],
     libs: [

tools/releasetools/ota_from_target_files.py

@@ -269,6 +269,7 @@ from common import IsSparseImage
 import target_files_diff
 from check_target_files_vintf import CheckVintfIfTrebleEnabled
 from non_ab_ota import GenerateNonAbOtaPackage
+from payload_signer import PayloadSigner
 
 if sys.hexversion < 0x02070000:
   print("Python 2.7 or newer is required.", file=sys.stderr)
@@ -335,70 +336,6 @@ SECONDARY_PAYLOAD_SKIPPED_IMAGES = [
     'vendor', 'vendor_boot']
 
 
-class PayloadSigner(object):
-  """A class that wraps the payload signing works.
-
-  When generating a Payload, hashes of the payload and metadata files will be
-  signed with the device key, either by calling an external payload signer or
-  by calling openssl with the package key. This class provides a unified
-  interface, so that callers can just call PayloadSigner.Sign().
-
-  If an external payload signer has been specified (OPTIONS.payload_signer), it
-  calls the signer with the provided args (OPTIONS.payload_signer_args). Note
-  that the signing key should be provided as part of the payload_signer_args.
-  Otherwise without an external signer, it uses the package key
-  (OPTIONS.package_key) and calls openssl for the signing works.
-  """
-
-  def __init__(self):
-    if OPTIONS.payload_signer is None:
-      # Prepare the payload signing key.
-      private_key = OPTIONS.package_key + OPTIONS.private_key_suffix
-      pw = OPTIONS.key_passwords[OPTIONS.package_key]
-
-      cmd = ["openssl", "pkcs8", "-in", private_key, "-inform", "DER"]
-      cmd.extend(["-passin", "pass:" + pw] if pw else ["-nocrypt"])
-      signing_key = common.MakeTempFile(prefix="key-", suffix=".key")
-      cmd.extend(["-out", signing_key])
-      common.RunAndCheckOutput(cmd, verbose=False)
-
-      self.signer = "openssl"
-      self.signer_args = ["pkeyutl", "-sign", "-inkey", signing_key,
-                          "-pkeyopt", "digest:sha256"]
-      self.maximum_signature_size = self._GetMaximumSignatureSizeInBytes(
-          signing_key)
-    else:
-      self.signer = OPTIONS.payload_signer
-      self.signer_args = OPTIONS.payload_signer_args
-      if OPTIONS.payload_signer_maximum_signature_size:
-        self.maximum_signature_size = int(
-            OPTIONS.payload_signer_maximum_signature_size)
-      else:
-        # The legacy config uses RSA2048 keys.
-        logger.warning("The maximum signature size for payload signer is not"
-                       " set, default to 256 bytes.")
-        self.maximum_signature_size = 256
-
-  @staticmethod
-  def _GetMaximumSignatureSizeInBytes(signing_key):
-    out_signature_size_file = common.MakeTempFile("signature_size")
-    cmd = ["delta_generator", "--out_maximum_signature_size_file={}".format(
-        out_signature_size_file), "--private_key={}".format(signing_key)]
-    common.RunAndCheckOutput(cmd)
-    with open(out_signature_size_file) as f:
-      signature_size = f.read().rstrip()
-    logger.info("%s outputs the maximum signature size: %s", cmd[0],
-                signature_size)
-    return int(signature_size)
-
-  def Sign(self, in_file):
-    """Signs the given input file. Returns the output filename."""
-    out_file = common.MakeTempFile(prefix="signed-", suffix=".bin")
-    cmd = [self.signer] + self.signer_args + ['-in', in_file, '-out', out_file]
-    common.RunAndCheckOutput(cmd)
-    return out_file
-
-
 class Payload(object):
   """Manages the creation and the signing of an A/B OTA Payload."""
 
@@ -1266,7 +1203,8 @@ def GenerateAbOtaPackage(target_file, output_file, source_file=None):
   )
 
   # Sign the payload.
-  payload_signer = PayloadSigner()
+  payload_signer = PayloadSigner(
+      OPTIONS.package_key, OPTIONS.private_key_suffix)
   payload.Sign(payload_signer)
 
   # Write the payload into output zip.

tools/releasetools/payload_signer.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2022 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import common
+import logging
+from common import OPTIONS
+
+logger = logging.getLogger(__name__)
+
+
+class PayloadSigner(object):
+  """A class that wraps the payload signing works.
+
+  When generating a Payload, hashes of the payload and metadata files will be
+  signed with the device key, either by calling an external payload signer or
+  by calling openssl with the package key. This class provides a unified
+  interface, so that callers can just call PayloadSigner.Sign().
+
+  If an external payload signer has been specified (OPTIONS.payload_signer), it
+  calls the signer with the provided args (OPTIONS.payload_signer_args). Note
+  that the signing key should be provided as part of the payload_signer_args.
+  Otherwise without an external signer, it uses the package key
+  (OPTIONS.package_key) and calls openssl for the signing works.
+  """
+
+  def __init__(self, package_key=None, private_key_suffix=None, pw=None, payload_signer=None):
+    if package_key is None:
+      package_key = OPTIONS.package_key
+    if private_key_suffix is None:
+      private_key_suffix = OPTIONS.private_key_suffix
+
+    if payload_signer is None:
+      # Prepare the payload signing key.
+      private_key = package_key + private_key_suffix
+
+      cmd = ["openssl", "pkcs8", "-in", private_key, "-inform", "DER"]
+      cmd.extend(["-passin", "pass:" + pw] if pw else ["-nocrypt"])
+      signing_key = common.MakeTempFile(prefix="key-", suffix=".key")
+      cmd.extend(["-out", signing_key])
+      common.RunAndCheckOutput(cmd, verbose=True)
+
+      self.signer = "openssl"
+      self.signer_args = ["pkeyutl", "-sign", "-inkey", signing_key,
+                          "-pkeyopt", "digest:sha256"]
+      self.maximum_signature_size = self._GetMaximumSignatureSizeInBytes(
+          signing_key)
+    else:
+      self.signer = payload_signer
+      self.signer_args = OPTIONS.payload_signer_args
+      if OPTIONS.payload_signer_maximum_signature_size:
+        self.maximum_signature_size = int(
+            OPTIONS.payload_signer_maximum_signature_size)
+      else:
+        # The legacy config uses RSA2048 keys.
+        logger.warning("The maximum signature size for payload signer is not"
+                       " set, default to 256 bytes.")
+        self.maximum_signature_size = 256
+
+  @staticmethod
+  def _GetMaximumSignatureSizeInBytes(signing_key):
+    out_signature_size_file = common.MakeTempFile("signature_size")
+    cmd = ["delta_generator", "--out_maximum_signature_size_file={}".format(
+        out_signature_size_file), "--private_key={}".format(signing_key)]
+    common.RunAndCheckOutput(cmd, verbose=True)
+    with open(out_signature_size_file) as f:
+      signature_size = f.read().rstrip()
+    logger.info("%s outputs the maximum signature size: %s", cmd[0],
+                signature_size)
+    return int(signature_size)
+
+  def Sign(self, in_file):
+    """Signs the given input file. Returns the output filename."""
+    out_file = common.MakeTempFile(prefix="signed-", suffix=".bin")
+    cmd = [self.signer] + self.signer_args + ['-in', in_file, '-out', out_file]
+    common.RunAndCheckOutput(cmd)
+    return out_file

tools/releasetools/test_ota_from_target_files.py

@@ -31,10 +31,12 @@ from ota_from_target_files import (
     GetTargetFilesZipForPartialUpdates,
     GetTargetFilesZipForSecondaryImages,
     GetTargetFilesZipWithoutPostinstallConfig,
-    Payload, PayloadSigner, POSTINSTALL_CONFIG,
+    Payload, POSTINSTALL_CONFIG,
     StreamingPropertyFiles, AB_PARTITIONS)
 from apex_utils import GetApexInfoFromTargetFiles
 from test_utils import PropertyFilesTestCase
+from common import OPTIONS
+from payload_signer import PayloadSigner
 
 
 def construct_target_files(secondary=False, compressedApex=False):
@@ -1142,10 +1144,10 @@ class PayloadSignerTest(test_utils.ReleaseToolsTestCase):
     self.assertEqual('openssl', payload_signer.signer)
 
   def test_init_withExternalSigner(self):
-    common.OPTIONS.payload_signer = 'abc'
     common.OPTIONS.payload_signer_args = ['arg1', 'arg2']
     common.OPTIONS.payload_signer_maximum_signature_size = '512'
-    payload_signer = PayloadSigner()
+    payload_signer = PayloadSigner(
+        OPTIONS.package_key, OPTIONS.private_key_suffix, payload_signer='abc')
     self.assertEqual('abc', payload_signer.signer)
     self.assertEqual(['arg1', 'arg2'], payload_signer.signer_args)
     self.assertEqual(512, payload_signer.maximum_signature_size)
@@ -1175,12 +1177,12 @@ class PayloadSignerTest(test_utils.ReleaseToolsTestCase):
 
   def test_Sign_withExternalSigner_openssl(self):
     """Uses openssl as the external payload signer."""
-    common.OPTIONS.payload_signer = 'openssl'
     common.OPTIONS.payload_signer_args = [
         'pkeyutl', '-sign', '-keyform', 'DER', '-inkey',
         os.path.join(self.testdata_dir, 'testkey.pk8'),
         '-pkeyopt', 'digest:sha256']
-    payload_signer = PayloadSigner()
+    payload_signer = PayloadSigner(
+        OPTIONS.package_key, OPTIONS.private_key_suffix, payload_signer="openssl")
     input_file = os.path.join(self.testdata_dir, self.SIGFILE)
     signed_file = payload_signer.Sign(input_file)
 
@@ -1189,12 +1191,13 @@ class PayloadSignerTest(test_utils.ReleaseToolsTestCase):
 
   def test_Sign_withExternalSigner_script(self):
     """Uses testdata/payload_signer.sh as the external payload signer."""
-    common.OPTIONS.payload_signer = os.path.join(
+    external_signer = os.path.join(
         self.testdata_dir, 'payload_signer.sh')
-    os.chmod(common.OPTIONS.payload_signer, 0o700)
+    os.chmod(external_signer, 0o700)
     common.OPTIONS.payload_signer_args = [
         os.path.join(self.testdata_dir, 'testkey.pk8')]
-    payload_signer = PayloadSigner()
+    payload_signer = PayloadSigner(
+        OPTIONS.package_key, OPTIONS.private_key_suffix, payload_signer=external_signer)
     input_file = os.path.join(self.testdata_dir, self.SIGFILE)
     signed_file = payload_signer.Sign(input_file)