add rpm-sign as method of signature (#170)

Author: shokakucarrier

README.md

@@ -14,9 +14,10 @@ future. And Ronda service will be hosted in AWS S3.
 
 See [AWS CLi V2 installation](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2-linux.html#cliv2-linux-install)
 
-### [Optional] GnuPG CLI tool
+### [Optional] GnuPG CLI tool or rpm-sign
 
 For artifact signing using keys already in GPG keystore, not required when using exported secret key file.
+Can be configured to use rpm-sign for release.
 
 ## Installation
 

charon/cmd/command.py

@@ -109,12 +109,21 @@
     "-K",
     help="""
     GPG key file path to sign artifacts, --passphrase is needed.
+    Will be ignored when using sign_method rpm-sign
+    """,
+)
+@option(
+    "--sign_method",
+    help="""
+    Choose 'gpg' or 'rpm-sign' to warp the actuall command, default to use gpg.
+    key_id will be used to set key for signature.
     """,
 )
 @option(
     "--passphrase",
     help="""
-    The passphrase for GPG key, will require input if this parameter is not set.
+    The passphrase for GPG key, Necessary when using gpg as sign method.
+    Will require input when it's not set
     """,
 )
 @option(
@@ -143,6 +152,7 @@ def upload(
     work_dir: str = None,
     sign_keyid: str = None,
     sign_keyfile: str = None,
+    sign_method: str = "gpg",
     passphrase: str = None,
     debug=False,
     quiet=False,
@@ -169,6 +179,10 @@ def upload(
             logger.error("No AWS profile specified!")
             sys.exit(1)
 
+        if sign_method != 'gpg' and sign_method != 'rpm-sign':
+            logger.error("Signature method must be gpg or rpm-sign")
+            sys.exit(1)
+
         archive_path = __get_local_repo(repo)
         npm_archive_type = detect_npm_archive(archive_path)
         product_key = f"{product}-{version}"
@@ -184,6 +198,7 @@ def upload(
                 dir_=work_dir,
                 key_id=sign_keyid,
                 key_file=sign_keyfile,
+                sign_method=sign_method,
                 passphrase=passphrase,
                 dry_run=dryrun,
                 manifest_bucket_name=manifest_bucket_name
@@ -207,6 +222,7 @@ def upload(
                 dir_=work_dir,
                 key_id=sign_keyid,
                 key_file=sign_keyfile,
+                sign_method=sign_method,
                 passphrase=passphrase,
                 dry_run=dryrun,
                 manifest_bucket_name=manifest_bucket_name

charon/pkgs/maven.py

@@ -263,6 +263,7 @@ def handle_maven_uploading(
     do_index=True,
     key_id=None,
     key_file=None,
+    sign_method=None,
     passphrase=None,
     dry_run=False,
     manifest_bucket_name=None
@@ -389,7 +390,7 @@ def handle_maven_uploading(
                 logger.info("archetype-catalog.xml updating done in bucket %s\n", bucket_name)
 
         # 10. Generate signature file if gpg ket is provided
-        if (key_id is not None or key_file is not None) and passphrase is not None:
+        if key_id is not None or key_file is not None:
             conf = get_config()
             if not conf:
                 sys.exit(1)
@@ -400,7 +401,7 @@ def handle_maven_uploading(
                 PACKAGE_TYPE_MAVEN, artifacts,
                 top_level, prefix,
                 s3_client, bucket_name,
-                key_id, key_file, passphrase
+                key_id, key_file, sign_method, passphrase
             )
             failed_metas.extend(_failed_metas)
             generated_signs.extend(_generated_signs)

charon/pkgs/npm.py

@@ -74,6 +74,7 @@ def handle_npm_uploading(
         do_index=True,
         key_id=None,
         key_file=None,
+        sign_method=None,
         passphrase=None,
         dry_run=False,
         manifest_bucket_name=None
@@ -166,7 +167,7 @@ def handle_npm_uploading(
             failed_metas.extend(_failed_metas)
             logger.info("package.json uploading done")
 
-        if (key_id is not None or key_file is not None) and passphrase is not None:
+        if key_id is not None or key_file is not None:
             conf = get_config()
             if not conf:
                 sys.exit(1)
@@ -179,7 +180,7 @@ def handle_npm_uploading(
                 PACKAGE_TYPE_NPM, artifacts,
                 target_dir, prefix,
                 client, bucket_name,
-                key_id, key_file, passphrase
+                key_id, key_file, sign_method, passphrase
             )
             failed_metas.extend(_failed_metas)
             generated_signs.extend(_generated_signs)

charon/pkgs/signature.py

@@ -21,6 +21,7 @@
 import gnupg
 from typing import Awaitable, Callable, List, Tuple
 from charon.storage import S3Client
+from getpass import getpass
 
 logger = logging.getLogger(__name__)
 
@@ -34,6 +35,7 @@ def generate_sign(
     bucket: str,
     key_id: str = None,
     key_file: str = None,
+    sign_method: str = None,
     passphrase: str = None
 ) -> Tuple[List[str], List[str]]:
     """ This Python function generates a digital signature for a list of metadata files using
@@ -50,11 +52,15 @@ def generate_sign(
     and another with the failed to generate files due to exceptions.
     """
 
+    gpg = None
     if key_file is not None:
         gnupg_home_path = os.path.join(os.getenv("HOME"), ".charon", ".gnupg")
         gpg = gnupg.GPG(gnupghome=gnupg_home_path)
         gpg.import_keys_file(key_file)
 
+    if sign_method == 'gpg' and passphrase is None:
+        passphrase = getpass('Passphrase for your gpg key:')
+
     async def sign_file(
         filename: str, failed_paths: List[str], generated_signs: List[str]
     ):
@@ -87,45 +93,20 @@ async def sign_file(
                 logger.debug(".asc file %s existed, skipping", remote)
                 return
 
-            command = [
-                'gpg',
-                '--batch',
-                '--armor',
-                '-u', key_id,
-                '--passphrase', passphrase,
-                '--sign', artifact
-            ]
-
-            if key_file is None:
-                # use GPG command line tool to sign artifact if key_id is passed
-                try:
-                    # result = await __run_cmd_async(command)
-                    result = subprocess.run(command, capture_output=True, text=True, check=True)
-                except subprocess.CalledProcessError as e:
-                    logger.error(
-                            "Error: signature generation failed due to error: %s", e
-                        )
-                    failed_paths.append(local)
-                    return
-                if result.returncode == 0:
-                    generated_signs.append(local)
-                else:
-                    logger.info(
-                        "signature failed with exit code %s, message %s",
-                        result.returncode, result.stderr.decode()
-                    )
+            if sign_method == "rpm-sign":
+                result = detach_rpm_sign_files(key_id, artifact)
+            elif sign_method == "gpg":
+                result = gpg_sign_files(
+                    gpg=gpg,
+                    artifact=artifact,
+                    key_id=key_id, key_file=key_file,
+                    passphrase=passphrase
+                )
+
+            if result == 0:
+                generated_signs.append(local)
             else:
-                try:
-                    with open(artifact, "rb") as f:
-                        gpg.sign_file(f, passphrase=passphrase, output=local, detach=True)
-                        generated_signs.append(local)
-                except ValueError as e:
-                    logger.error(
-                            "Error: signature generation failed due to error: %s", e
-                        )
-                    failed_paths.append(local)
-
-            return
+                failed_paths.append(local)
 
     return __do_path_cut_and(
             file_paths=artifact_path,
@@ -134,6 +115,82 @@ async def sign_file(
         )
 
 
+def detach_rpm_sign_files(key: str, artifact: str) -> int:
+    # Usage: detach_sign_files KEYNAME FILE ...
+
+    # let's make sure we can actually sign with the given key
+    command = [
+        'rpm-sign',
+        '--list-keys',
+        '|',
+        'grep',
+        key
+    ]
+    result = subprocess.run(command, capture_output=True, text=True, check=True)
+    if result.returncode != 0:
+        logger.error("Key %s is not in list of allowed keys", key)
+        return result.returncode
+
+    # okay, now let's actually sign this thing
+    command = [
+        'rpm-sign',
+        '--detachsign',
+        '--key',
+        key,
+        artifact
+    ]
+    try:
+        # result = await __run_cmd_async(command)
+        result = subprocess.run(command, capture_output=True, text=True, check=True)
+    except subprocess.CalledProcessError as e:
+        logger.error(
+                "Error: signature generation failed due to error: %s", e
+            )
+        return result.returncode
+
+    return 1
+
+
+def gpg_sign_files(
+    artifact: str,
+    gpg=None,
+    key_id: str = None,
+    key_file: str = None,
+    passphrase: str = None
+) -> int:
+    command = [
+        'gpg',
+        '--batch',
+        '--armor',
+        '-u', key_id,
+        '--passphrase', passphrase,
+        '--sign', artifact
+    ]
+
+    if key_file is None:
+        # use GPG command line tool to sign artifact if key_id is passed
+        try:
+            # result = await __run_cmd_async(command)
+            result = subprocess.run(command, capture_output=True, text=True, check=True)
+        except subprocess.CalledProcessError as e:
+            logger.error(
+                    "Error: signature generation failed due to error: %s", e
+                )
+        return result.returncode
+    else:
+        try:
+            with open(artifact, "rb") as f:
+                local = artifact + '.asc'
+                gpg.sign_file(f, passphrase=passphrase, output=local, detach=True)
+                return 0
+        except ValueError as e:
+            logger.error(
+                    "Error: signature generation failed due to error: %s", e
+                )
+            return 1
+    return 1
+
+
 def __do_path_cut_and(
     file_paths: List[str],
     path_handler: Callable[[str, List[str], List[str], asyncio.Semaphore], Awaitable[bool]],