diff --git a/keyringctl b/keyringctl index 9228229..c7ae1bd 100755 --- a/keyringctl +++ b/keyringctl @@ -125,7 +125,7 @@ def convert_certificate(working_dir: Path, certificate: Path, name_override: Opt debug(f'Processing certificate {certificate}') - for packet in packet_split(working_dir, certificate): + for packet in packet_split(working_dir=working_dir, certificate=certificate): debug(f'Processing packet {packet.name}') if packet.name.endswith('--PublicKey'): pubkey = packet @@ -432,10 +432,67 @@ def packet_dump_field(packet: Path, field: str) -> str: return lines[0].split(maxsplit=1)[1] -def packet_split(working_dir: Path, key: Path) -> Iterable[Path]: +def sanitize_certificate_file(working_dir: Path, certificate: Path) -> List[Path]: + """Sanitize a certificate file, potentially containing several certificates + + If the input file holds several certificates, they are split into respective files below working_dir and their + paths are returned. Else the path to the input certificate file is returned. + + Parameters + ---------- + working_dir: Path + The path of the working directory below which to create split certificates + certificate: Path + The absolute path of a file containing one or several PGP certificates + + Returns + ------- + List[Path] + A list of paths that either contain the input certificate or several split certificates + """ + + begin_cert = "-----BEGIN PGP PUBLIC KEY BLOCK-----" + end_cert = "-----END PGP PUBLIC KEY BLOCK-----" + certificate_list: List[Path] = [] + + with open(file=certificate, mode="r") as certificate_file: + file = certificate_file.read() + if file.count(begin_cert) > 1 and file.count(end_cert) > 1: + debug(f"Several public keys detected in file: {certificate}") + for split_cert in [f"{cert}{end_cert}\n" for cert in list(filter(None, file.split(f"{end_cert}\n")))]: + split_cert_dir = Path(mkdtemp(dir=working_dir)).absolute() + split_cert_path = split_cert_dir / certificate.name + with open(file=split_cert_path, mode="w") as split_cert_file: + split_cert_file.write(split_cert) + debug(f"Writing split cert to file: {split_cert_path}") + certificate_list.append(split_cert_path) + return certificate_list + else: + return [certificate] + + +def packet_split(working_dir: Path, certificate: Path) -> Iterable[Path]: + """Split a file containing a PGP certificate into separate packet files + + The files are split using sq + + Parameters + ---------- + working_dir: Path + The path of the working directory below which to create the output files + certificate: Path + The absolute path of a file containing one PGP certificate + + Returns + ------- + Iterable[Path] + An iterable over the naturally sorted list of packet files derived from certificate + """ + packet_dir = Path(mkdtemp(dir=working_dir)).absolute() + with cwd(packet_dir): - system(['sq', 'packet', 'split', '--prefix', '', str(key)]) + system(['sq', 'packet', 'split', '--prefix', '', str(certificate)]) return natural_sort_path(packet_dir.iterdir()) @@ -463,13 +520,15 @@ def convert( directories: List[Path] = [] if source.is_dir(): for key in source.iterdir(): - directories.append( - convert_certificate(working_dir=working_dir, certificate=key, name_override=name_override) - ) + for sane_cert in sanitize_certificate_file(working_dir=working_dir, certificate=key): + directories.append( + convert_certificate(working_dir=working_dir, certificate=sane_cert, name_override=name_override) + ) else: - directories.append( - convert_certificate(working_dir=working_dir, certificate=source, name_override=name_override) - ) + for sane_cert in sanitize_certificate_file(working_dir=working_dir, certificate=source): + directories.append( + convert_certificate(working_dir=working_dir, certificate=sane_cert, name_override=name_override) + ) for path in directories: (target_dir / path.name).mkdir(exist_ok=True)