Run black on the code

Reformat all the code with black to make the static analysis pass. Signed-off-by: Patrick Uiterwijk <patrick@puiterwijk.org>
puiterwijk · Jul 22, 2021 · c1da639 · c1da639
1 parent 23baf16
commit c1da639
Show file tree

Hide file tree

Showing 8 changed files with 246 additions and 195 deletions.
diff --git a/rpm_head_signing/__init__.py b/rpm_head_signing/__init__.py
@@ -1,4 +1,7 @@
 from .insert_signature import insert_signature
 from .extract_header import extract_header, NonHeaderSignablePackage
 from .extract_rpm_with_filesigs import extract_rpm_with_filesigs
-from .extract_signature_and_ima_info import parse_ima_signature, get_rpm_ima_signature_info
+from .extract_signature_and_ima_info import (
+    parse_ima_signature,
+    get_rpm_ima_signature_info,
+)
diff --git a/rpm_head_signing/extract_header.py b/rpm_head_signing/extract_header.py
@@ -1,6 +1,13 @@
 #!/usr/bin/env python
 import rpm
-from koji import (RawHeader, get_rpm_header, rpm_hdr_size, find_rpm_sighdr, RPM_TAG_FILEDIGESTALGO, RPM_FILEDIGESTALGO_IDS)
+from koji import (
+    RawHeader,
+    get_rpm_header,
+    rpm_hdr_size,
+    find_rpm_sighdr,
+    RPM_TAG_FILEDIGESTALGO,
+    RPM_FILEDIGESTALGO_IDS,
+)
 
 
 RPMTAG_PAYLOADDIGEST = 5092
@@ -49,10 +56,12 @@ def extract_header(input_path, header_out_path, digest_out_path):
                 df.write("%s %s\n" % (file_digestalgo, digest))
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import sys
 
     if len(sys.argv) != 4:
-        raise Exception("Call: %s <input-rpm> <output-hdr> <output-digests>" % sys.argv[0])
+        raise Exception(
+            "Call: %s <input-rpm> <output-hdr> <output-digests>" % sys.argv[0]
+        )
 
     extract_header(sys.argv[1], sys.argv[2], sys.argv[3])
diff --git a/rpm_head_signing/extract_rpm_with_filesigs.py b/rpm_head_signing/extract_rpm_with_filesigs.py
@@ -9,20 +9,20 @@
 import xattr
 
 
-rpm_version = subprocess.check_output(['rpm', '--version'])
-rpm_version = tuple(map(int, rpm_version.strip().split(b' ')[2].split(b'.')))
+rpm_version = subprocess.check_output(["rpm", "--version"])
+rpm_version = tuple(map(int, rpm_version.strip().split(b" ")[2].split(b".")))
 if rpm_version[0] != 4:
-    raise Exception('RPM version %s is not major version 4' % rpm_version)
+    raise Exception("RPM version %s is not major version 4" % rpm_version)
 
 
 def _extract_rpm(rpm_path, output_path):
     # To deal with zstd on RPM 4.11
     if rpm_version[1] == 11:
-        rpm2cpio = './test_assets/rpm2cpio.sh'
+        rpm2cpio = "./test_assets/rpm2cpio.sh"
     else:
-        rpm2cpio = 'rpm2cpio'
+        rpm2cpio = "rpm2cpio"
 
-    with TemporaryFile(prefix='rpm-', suffix='.cpio') as cpiof:
+    with TemporaryFile(prefix="rpm-", suffix=".cpio") as cpiof:
         subprocess.check_call(
             [
                 rpm2cpio,
@@ -33,11 +33,11 @@ def _extract_rpm(rpm_path, output_path):
         cpiof.seek(0, 0)
         subprocess.check_call(
             [
-                'cpio',
-                '--extract',
-                '--make-directories',
-                '--no-preserve-owner',
-                '--no-absolute-filenames',
+                "cpio",
+                "--extract",
+                "--make-directories",
+                "--no-preserve-owner",
+                "--no-absolute-filenames",
             ],
             stdin=cpiof,
             cwd=output_path,
@@ -64,13 +64,13 @@ def _get_header_type_8(raw_hdr, tag):
     filesigs = []
     for i in range(count):
         if sys.version_info.major == 2:
-            end = raw_hdr.header.find('\0', pos)
+            end = raw_hdr.header.find("\0", pos)
         elif sys.version_info.major == 3:
-            end = raw_hdr.header.find(b'\0', pos)
+            end = raw_hdr.header.find(b"\0", pos)
         else:
             raise Exception("Unsupported Python")
         filesig = raw_hdr.header[pos:end]
-        filesig = filesig.decode('utf8')
+        filesig = filesig.decode("utf8")
         filesig = bytearray.fromhex(filesig)
         filesigs.append(filesig)
         pos = end + 1
@@ -91,9 +91,15 @@ def _extract_filesigs(rpm_path):
     basenames = rpm_hdr[rpm.RPMTAG_BASENAMES]
 
     if len(basenames) != len(filesigs):
-        raise Exception("Invalid number of file signatures (%d) for basenames (%d)" % (len(filesigs), len(basenames)))
+        raise Exception(
+            "Invalid number of file signatures (%d) for basenames (%d)"
+            % (len(filesigs), len(basenames))
+        )
     if len(diridxs) != len(basenames):
-        raise Exception("Invalid number of diridxs (%d) for basenames (%d)" % (len(diridxs), len(basenames)))
+        raise Exception(
+            "Invalid number of diridxs (%d) for basenames (%d)"
+            % (len(diridxs), len(basenames))
+        )
 
     signatures = {}
 
@@ -110,8 +116,8 @@ def _extract_filesigs(rpm_path):
 
 def _install_filesigs(signatures, output_path):
     for path in signatures:
-        full_path = os.path.join(output_path, path.lstrip('/'))
-        xattr.setxattr(full_path, 'user.ima', signatures[path])
+        full_path = os.path.join(output_path, path.lstrip("/"))
+        xattr.setxattr(full_path, "user.ima", signatures[path])
 
 
 def extract_rpm_with_filesigs(rpm_path, output_path):
@@ -120,10 +126,10 @@ def extract_rpm_with_filesigs(rpm_path, output_path):
     _install_filesigs(filesigs, output_path)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import sys
 
     if len(sys.argv) != 3:
-        raise Exception('Call: %s <rpm-path> <output-path>' % sys.argv[0])
+        raise Exception("Call: %s <rpm-path> <output-path>" % sys.argv[0])
 
     extract_rpm_with_filesigs(sys.argv[1], sys.argv[2])
diff --git a/rpm_head_signing/extract_signature_and_ima_info.py b/rpm_head_signing/extract_signature_and_ima_info.py
@@ -17,52 +17,54 @@ def parse_ima_signature(sig):
         return None
 
     info = {
-        'type': sig[0],
-        'version': sig[1],
-        'alg_id': sig[2],
-        'key_id': bytes(sig[3:7]),
-        'sig_size': struct.unpack('>H', sig[7:9])[0],
-
-        'error': 'Did not finish parsing',
+        "type": sig[0],
+        "version": sig[1],
+        "alg_id": sig[2],
+        "key_id": bytes(sig[3:7]),
+        "sig_size": struct.unpack(">H", sig[7:9])[0],
+        "error": "Did not finish parsing",
     }
 
     readable_key_id = binascii.hexlify(sig[3:7])
     if not isinstance(readable_key_id, str):
-        readable_key_id = readable_key_id.decode('utf8')
-    info['user_readable_key_id'] = readable_key_id
+        readable_key_id = readable_key_id.decode("utf8")
+    info["user_readable_key_id"] = readable_key_id
 
-    if info['type'] != 3:
-        info['error'] = 'Unsupported type'
+    if info["type"] != 3:
+        info["error"] = "Unsupported type"
         return info
-    if info['version'] != 2:
-        info['error'] = 'Unsupported version'
+    if info["version"] != 2:
+        info["error"] = "Unsupported version"
         return info
 
-    if info['alg_id'] == 7:  # SHA224
-        info['alg_name'] = 'SHA224'
+    if info["alg_id"] == 7:  # SHA224
+        info["alg_name"] = "SHA224"
         crypto_algo = crypto_hashes.SHA224()
-    elif info['alg_id'] == 4:  # SHA256
-        info['alg_name'] = 'SHA256'
+    elif info["alg_id"] == 4:  # SHA256
+        info["alg_name"] = "SHA256"
         crypto_algo = crypto_hashes.SHA256()
-    elif info['alg_id'] == 5:  # SHA384
-        info['alg_name'] = 'SHA384'
+    elif info["alg_id"] == 5:  # SHA384
+        info["alg_name"] = "SHA384"
         crypto_algo = crypto_hashes.SHA384()
-    elif info['alg_id'] == 6:  # SHA512
-        info['alg_name'] = 'SHA512'
+    elif info["alg_id"] == 6:  # SHA512
+        info["alg_name"] = "SHA512"
         crypto_algo = crypto_hashes.SHA512()
     else:
-        info['error'] = 'Unsupported algorithm %d' % info['alg_id']
+        info["error"] = "Unsupported algorithm %d" % info["alg_id"]
         return info
-    info['hashing_algorithm'] = crypto_algo
+    info["hashing_algorithm"] = crypto_algo
     crypto_algo = Prehashed(crypto_algo)
-    info['algorithm'] = crypto_ec.ECDSA(crypto_algo)
+    info["algorithm"] = crypto_ec.ECDSA(crypto_algo)
 
-    if (len(sig) - 9) != info['sig_size']:
-        info['error'] = 'Signature length mismatch: %d (actual) != %d (expected)' % (len(sig) - 9, info['sig_size'])
+    if (len(sig) - 9) != info["sig_size"]:
+        info["error"] = "Signature length mismatch: %d (actual) != %d (expected)" % (
+            len(sig) - 9,
+            info["sig_size"],
+        )
         return info
 
-    info['signature'] = bytes(sig[9:])
-    info['error'] = None
+    info["signature"] = bytes(sig[9:])
+    info["error"] = None
 
     return info
 

diff --git a/rpm_head_signing/insert_signature.py b/rpm_head_signing/insert_signature.py
@@ -23,7 +23,7 @@ def insert_signature(rpm_path, sig_path, ima_presigned_path=None, return_header=
         return_header = 0
 
     # Add RSA Header record
-    with open(sig_path, 'rb') as sigfile:
+    with open(sig_path, "rb") as sigfile:
         rpm_signature = bytearray(sigfile.read())
 
     # Add IMA signature record
@@ -35,10 +35,10 @@ def insert_signature(rpm_path, sig_path, ima_presigned_path=None, return_header=
         )
     else:
         ima_signature_lookup = {}
-        with open(ima_presigned_path, 'r') as sigpath:
+        with open(ima_presigned_path, "r") as sigpath:
             for line in sigpath.readlines():
-                algo, digest, signature = line.strip().split(' ')
-                signature = binascii.hexlify(b'\x03' + base64.b64decode(signature))
+                algo, digest, signature = line.strip().split(" ")
+                signature = binascii.hexlify(b"\x03" + base64.b64decode(signature))
                 if algo not in ima_signature_lookup:
                     ima_signature_lookup[algo] = {}
                 ima_signature_lookup[algo][digest.lower()] = signature
@@ -51,7 +51,7 @@ def insert_signature(rpm_path, sig_path, ima_presigned_path=None, return_header=
         )
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import sys
 
     if len(sys.argv) == 3:

diff --git a/setup.py b/setup.py
@@ -1,38 +1,38 @@
 from setuptools import setup, Extension
 import subprocess
 
-rpm_version = subprocess.check_output(['rpm', '--version'])
-rpm_version = tuple(map(int, rpm_version.strip().split(b' ')[2].split(b'.')))
+rpm_version = subprocess.check_output(["rpm", "--version"])
+rpm_version = tuple(map(int, rpm_version.strip().split(b" ")[2].split(b".")))
 if rpm_version[0] != 4:
-    raise Exception('RPM version %s is not major version 4' % rpm_version)
+    raise Exception("RPM version %s is not major version 4" % rpm_version)
 ext_defines = []
 if rpm_version[1] >= 15:
-    ext_defines.append(('RPM_415', None))
+    ext_defines.append(("RPM_415", None))
 elif rpm_version[1] == 14:
-    ext_defines.append(('RPM_414', None))
+    ext_defines.append(("RPM_414", None))
 elif rpm_version[1] == 11:
-    ext_defines.append(('RPM_411', None))
+    ext_defines.append(("RPM_411", None))
 else:
-    raise Exception('Unsupported RPM version %s' % rpm_version)
+    raise Exception("Unsupported RPM version %s" % rpm_version)
 
 insertlib = Extension(
-    'insertlib',
-    libraries = ['rpm', 'rpmio'],
-    sources = ['rpm_head_signing/insertlib.c'],
-    extra_compile_args = ['-Wall', '-Werror'],
-    define_macros = ext_defines,
+    "insertlib",
+    libraries=["rpm", "rpmio"],
+    sources=["rpm_head_signing/insertlib.c"],
+    extra_compile_args=["-Wall", "-Werror"],
+    define_macros=ext_defines,
 )
 
 setup(
-    name='rpm_head_signing',
-    version='1.0',
-    packages=['rpm_head_signing'],
-    ext_package='rpm_head_signing',
+    name="rpm_head_signing",
+    version="1.0",
+    packages=["rpm_head_signing"],
+    ext_package="rpm_head_signing",
     ext_modules=[insertlib],
     install_requires=[
-        'requests',
-        'koji',
-        'rpm',
-        'pyxattr',
-    ]
+        "requests",
+        "koji",
+        "rpm",
+        "pyxattr",
+    ],
 )