Add sign_sepolicy_path for a binary to sign sepolicy.apex.

Bug: b/217570541
Test: sign_apex --container_key=testdata/testkey
--payload_key=testdata/testkey_RSA4096.key
--codename_to_api_level_map=S:31,T:32,UpsideDownCake:33
-e SEPolicy-33.zip=testdata/testkey com.android.sepolicy.apex
signed_sepolicy.apex

Change-Id: Ifdd1c6e7db07cbc4674450ab0e2226494c98605a

tools/releasetools/apex_utils.py

@@ -66,7 +66,7 @@ class ApexApkSigner(object):
     self.avbtool = avbtool if avbtool else "avbtool"
     self.sign_tool = sign_tool
 
-  def ProcessApexFile(self, apk_keys, payload_key, signing_args=None):
+  def ProcessApexFile(self, apk_keys, payload_key, signing_args=None, is_sepolicy=False):
     """Scans and signs the payload files and repack the apex
 
     Args:
@@ -84,9 +84,13 @@ class ApexApkSigner(object):
                 self.debugfs_path, 'list', self.apex_path]
     entries_names = common.RunAndCheckOutput(list_cmd).split()
     apk_entries = [name for name in entries_names if name.endswith('.apk')]
+    sepolicy_entries = []
+    if is_sepolicy:
+      sepolicy_entries = [name for name in entries_names if
+          name.startswith('./etc/SEPolicy') and name.endswith('.zip')]
 
     # No need to sign and repack, return the original apex path.
-    if not apk_entries and self.sign_tool is None:
+    if not apk_entries and not sepolicy_entries and self.sign_tool is None:
       logger.info('No apk file to sign in %s', self.apex_path)
       return self.apex_path
 
@@ -102,14 +106,14 @@ class ApexApkSigner(object):
                        ' %s', entry)
 
     payload_dir, has_signed_content = self.ExtractApexPayloadAndSignContents(
-        apk_entries, apk_keys, payload_key, signing_args)
+        apk_entries, sepolicy_entries, apk_keys, payload_key, signing_args)
     if not has_signed_content:
-      logger.info('No contents has been signed in %s', self.apex_path)
+      logger.info('No contents have been signed in %s', self.apex_path)
       return self.apex_path
 
     return self.RepackApexPayload(payload_dir, payload_key, signing_args)
 
-  def ExtractApexPayloadAndSignContents(self, apk_entries, apk_keys, payload_key, signing_args):
+  def ExtractApexPayloadAndSignContents(self, apk_entries, sepolicy_entries, apk_keys, payload_key, signing_args):
     """Extracts the payload image and signs the containing apk files."""
     if not os.path.exists(self.debugfs_path):
       raise ApexSigningError(
@@ -120,11 +124,11 @@ class ApexApkSigner(object):
     extract_cmd = ['deapexer', '--debugfs_path',
                    self.debugfs_path, 'extract', self.apex_path, payload_dir]
     common.RunAndCheckOutput(extract_cmd)
+    assert os.path.exists(self.apex_path)
 
     has_signed_content = False
     for entry in apk_entries:
       apk_path = os.path.join(payload_dir, entry)
-      assert os.path.exists(self.apex_path)
 
       key_name = apk_keys.get(os.path.basename(entry))
       if key_name in common.SPECIAL_CERT_STRINGS:
@@ -141,6 +145,37 @@ class ApexApkSigner(object):
           codename_to_api_level_map=self.codename_to_api_level_map)
       has_signed_content = True
 
+    for entry in sepolicy_entries:
+      sepolicy_path = os.path.join(payload_dir, entry)
+
+      if not 'etc' in entry:
+        logger.warning('Sepolicy path does not contain the intended directory name etc:'
+                       ' %s', entry)
+
+      key_name = apk_keys.get(os.path.basename(entry))
+      if key_name is None:
+        logger.warning('Failed to find signing keys for {} in'
+                       ' apex {}, payload key will be used instead.'
+                       ' Use "-e <name>=" to specify a key'
+                       .format(entry, self.apex_path))
+        key_name = payload_key
+
+      if key_name in common.SPECIAL_CERT_STRINGS:
+        logger.info('Not signing: %s due to special cert string', sepolicy_path)
+        continue
+
+      if OPTIONS.sign_sepolicy_path is not None:
+        sig_path = os.path.join(payload_dir, sepolicy_path + '.sig')
+        fsv_sig_path = os.path.join(payload_dir, sepolicy_path + '.fsv_sig')
+        old_sig = common.MakeTempFile()
+        old_fsv_sig = common.MakeTempFile()
+        os.rename(sig_path, old_sig)
+        os.rename(fsv_sig_path, old_fsv_sig)
+
+      logger.info('Signing sepolicy file %s in apex %s', sepolicy_path, self.apex_path)
+      if common.SignSePolicy(sepolicy_path, key_name, self.key_passwords.get(key_name)):
+        has_signed_content = True
+
     if self.sign_tool:
       logger.info('Signing payload contents in apex %s with %s', self.apex_path, self.sign_tool)
       # Pass avbtool to the custom signing tool
@@ -324,7 +359,8 @@ def ParseApexPayloadInfo(avbtool, payload_path):
 
 def SignUncompressedApex(avbtool, apex_file, payload_key, container_key,
                          container_pw, apk_keys, codename_to_api_level_map,
-                         no_hashtree, signing_args=None, sign_tool=None):
+                         no_hashtree, signing_args=None, sign_tool=None,
+                         is_sepolicy=False):
   """Signs the current uncompressed APEX with the given payload/container keys.
 
   Args:
@@ -337,6 +373,7 @@ def SignUncompressedApex(avbtool, apex_file, payload_key, container_key,
     no_hashtree: Don't include hashtree in the signed APEX.
     signing_args: Additional args to be passed to the payload signer.
     sign_tool: A tool to sign the contents of the APEX.
+    is_sepolicy: Indicates if the apex is a sepolicy.apex
 
   Returns:
     The path to the signed APEX file.
@@ -346,7 +383,8 @@ def SignUncompressedApex(avbtool, apex_file, payload_key, container_key,
   apk_signer = ApexApkSigner(apex_file, container_pw,
                              codename_to_api_level_map,
                              avbtool, sign_tool)
-  apex_file = apk_signer.ProcessApexFile(apk_keys, payload_key, signing_args)
+  apex_file = apk_signer.ProcessApexFile(
+      apk_keys, payload_key, signing_args, is_sepolicy)
 
   # 2a. Extract and sign the APEX_PAYLOAD_IMAGE entry with the given
   # payload_key.
@@ -400,7 +438,8 @@ def SignUncompressedApex(avbtool, apex_file, payload_key, container_key,
 
 def SignCompressedApex(avbtool, apex_file, payload_key, container_key,
                        container_pw, apk_keys, codename_to_api_level_map,
-                       no_hashtree, signing_args=None, sign_tool=None):
+                       no_hashtree, signing_args=None, sign_tool=None,
+                       is_sepolicy=False):
   """Signs the current compressed APEX with the given payload/container keys.
 
   Args:
@@ -412,6 +451,7 @@ def SignCompressedApex(avbtool, apex_file, payload_key, container_key,
     codename_to_api_level_map: A dict that maps from codename to API level.
     no_hashtree: Don't include hashtree in the signed APEX.
     signing_args: Additional args to be passed to the payload signer.
+    is_sepolicy: Indicates if the apex is a sepolicy.apex
 
   Returns:
     The path to the signed APEX file.
@@ -438,7 +478,8 @@ def SignCompressedApex(avbtool, apex_file, payload_key, container_key,
       codename_to_api_level_map,
       no_hashtree,
       signing_args,
-      sign_tool)
+      sign_tool,
+      is_sepolicy)
 
   # 3. Compress signed original apex.
   compressed_apex_file = common.MakeTempFile(prefix='apex-container-',
@@ -465,8 +506,8 @@ def SignCompressedApex(avbtool, apex_file, payload_key, container_key,
 
 
 def SignApex(avbtool, apex_data, payload_key, container_key, container_pw,
-             apk_keys, codename_to_api_level_map,
+             apk_keys, codename_to_api_level_map, no_hashtree,
-             no_hashtree, signing_args=None, sign_tool=None):
+             signing_args=None, sign_tool=None, is_sepolicy=False):
   """Signs the current APEX with the given payload/container keys.
 
   Args:
@@ -478,6 +519,7 @@ def SignApex(avbtool, apex_data, payload_key, container_key, container_pw,
     codename_to_api_level_map: A dict that maps from codename to API level.
     no_hashtree: Don't include hashtree in the signed APEX.
     signing_args: Additional args to be passed to the payload signer.
+    is_sepolicy: Indicates if the apex is a sepolicy.apex
 
   Returns:
     The path to the signed APEX file.
@@ -503,7 +545,8 @@ def SignApex(avbtool, apex_data, payload_key, container_key, container_pw,
           no_hashtree=no_hashtree,
           apk_keys=apk_keys,
           signing_args=signing_args,
-          sign_tool=sign_tool)
+          sign_tool=sign_tool,
+          is_sepolicy=is_sepolicy)
     elif apex_type == 'COMPRESSED':
       return SignCompressedApex(
           avbtool,
@@ -515,7 +558,8 @@ def SignApex(avbtool, apex_data, payload_key, container_key, container_pw,
           no_hashtree=no_hashtree,
           apk_keys=apk_keys,
           signing_args=signing_args,
-          sign_tool=sign_tool)
+          sign_tool=sign_tool,
+          is_sepolicy=is_sepolicy)
     else:
       # TODO(b/172912232): support signing compressed apex
       raise ApexInfoError('Unsupported apex type {}'.format(apex_type))

tools/releasetools/common.py

@@ -72,7 +72,9 @@ class Options(object):
       if "ANDROID_HOST_OUT" in os.environ:
         self.search_path = os.environ["ANDROID_HOST_OUT"]
     self.signapk_shared_library_path = "lib64"   # Relative to search_path
+    self.sign_sepolicy_path = None
     self.extra_signapk_args = []
+    self.extra_sign_sepolicy_args = []
     self.aapt2_path = "aapt2"
     self.java_path = "java"  # Use the one on the path by default.
     self.java_args = ["-Xmx2048m"]  # The default JVM args.
@@ -97,6 +99,7 @@ class Options(object):
     self.stash_threshold = 0.8
     self.logfile = None
     self.host_tools = {}
+    self.sepolicy_name = 'sepolicy.apex'
 
 
 OPTIONS = Options()
@@ -2379,6 +2382,35 @@ def SignFile(input_name, output_name, key, password, min_api_level=None,
         "Failed to run signapk.jar: return code {}:\n{}".format(
             proc.returncode, stdoutdata))
 
+def SignSePolicy(sepolicy, key, password):
+  """Sign the sepolicy zip, producing an fsverity .fsv_sig and
+  an RSA .sig signature files.
+  """
+
+  if OPTIONS.sign_sepolicy_path is None:
+    return False
+
+  java_library_path = os.path.join(
+      OPTIONS.search_path, OPTIONS.signapk_shared_library_path)
+
+  cmd = ([OPTIONS.java_path] + OPTIONS.java_args +
+          ["-Djava.library.path=" + java_library_path,
+          "-jar", os.path.join(OPTIONS.search_path, OPTIONS.sign_sepolicy_path)] +
+          OPTIONS.extra_sign_sepolicy_args)
+
+  cmd.extend([key + OPTIONS.public_key_suffix,
+              key + OPTIONS.private_key_suffix,
+              sepolicy])
+
+  proc = Run(cmd, stdin=subprocess.PIPE)
+  if password is not None:
+    password += "\n"
+  stdoutdata, _ = proc.communicate(password)
+  if proc.returncode != 0:
+    raise ExternalError(
+        "Failed to run sign sepolicy: return code {}:\n{}".format(
+            proc.returncode, stdoutdata))
+  return True
 
 def CheckSize(data, target, info_dict):
   """Checks the data string passed against the max size limit.
@@ -2555,7 +2587,8 @@ def ParseOptions(argv,
     opts, args = getopt.getopt(
         argv, "hvp:s:x:" + extra_opts,
         ["help", "verbose", "path=", "signapk_path=",
-         "signapk_shared_library_path=", "extra_signapk_args=", "aapt2_path=",
+         "signapk_shared_library_path=", "extra_signapk_args=",
+         "sign_sepolicy_path=", "extra_sign_sepolicy_args=", "aapt2_path=",
          "java_path=", "java_args=", "android_jar_path=", "public_key_suffix=",
          "private_key_suffix=", "boot_signer_path=", "boot_signer_args=",
          "verity_signer_path=", "verity_signer_args=", "device_specific=",
@@ -2579,6 +2612,10 @@ def ParseOptions(argv,
       OPTIONS.signapk_shared_library_path = a
     elif o in ("--extra_signapk_args",):
       OPTIONS.extra_signapk_args = shlex.split(a)
+    elif o in ("--sign_sepolicy_path",):
+      OPTIONS.sign_sepolicy_path = a
+    elif o in ("--extra_sign_sepolicy_args",):
+      OPTIONS.extra_sign_sepolicy_args = shlex.split(a)
     elif o in ("--aapt2_path",):
       OPTIONS.aapt2_path = a
     elif o in ("--java_path",):

tools/releasetools/sign_apex.py

@@ -52,6 +52,7 @@ import apex_utils
 import common
 
 logger = logging.getLogger(__name__)
+OPTIONS = common.OPTIONS
 
 
 def SignApexFile(avbtool, apex_file, payload_key, container_key, no_hashtree,
@@ -70,7 +71,8 @@ def SignApexFile(avbtool, apex_file, payload_key, container_key, no_hashtree,
       no_hashtree=no_hashtree,
       apk_keys=apk_keys,
       signing_args=signing_args,
-      sign_tool=sign_tool)
+      sign_tool=sign_tool,
+      is_sepolicy=apex_file.endswith(OPTIONS.sepolicy_name))
 
 
 def main(argv):

tools/releasetools/test_sign_apex.py

@@ -58,6 +58,21 @@ class SignApexTest(test_utils.ReleaseToolsTestCase):
         apk_keys)
     self.assertTrue(os.path.exists(signed_test_apex))
 
+  @test_utils.SkipIfExternalToolsUnavailable()
+  def test_SignSepolicyApex(self):
+    test_apex = os.path.join(self.testdata_dir, 'sepolicy.apex')
+    payload_key = os.path.join(self.testdata_dir, 'testkey_RSA4096.key')
+    container_key = os.path.join(self.testdata_dir, 'testkey')
+    apk_keys = {'SEPolicy-33.zip': os.path.join(self.testdata_dir, 'testkey')}
+    signed_test_apex = sign_apex.SignApexFile(
+        'avbtool',
+        test_apex,
+        payload_key,
+        container_key,
+        False,
+        None)
+    self.assertTrue(os.path.exists(signed_test_apex))
+
   @test_utils.SkipIfExternalToolsUnavailable()
   def test_SignCompressedApexFile(self):
     apex = os.path.join(test_utils.get_current_dir(), 'com.android.apex.compressed.v1.capex')