| #!/usr/bin/env python3 |
| |
| """ |
| A little tool to encrypt/decrypt git secrets. Kinda like password-store, but |
| more purpose specific and portable. |
| |
| It generally expects to work with directory structures as follows: |
| |
| foo/bar/secrets/plain: plaintext files |
| /cipher: ciphertext files, with names corresponding to |
| plaintext files |
| |
| Note: currently all plaintext/cipher files are at a single level, ie.: there |
| cannot be any subdirectory within a /plain or /cipher directory. |
| |
| There are multiple secret 'roots' like this in hscloud, notably: |
| |
| - cluster/secrets |
| - hswaw/kube/secrets |
| |
| In the future, some repository-based configuration might exist to specify these |
| roots in a nicer way, possibly with different target keys per root. |
| |
| This tool a bit of a swiss army knife, and can be used in the following ways: |
| |
| - as a CLI tool to encrypt/decrypt files directly |
| - as a library for its encryption/decryption methods, and for a SecretStore |
| API, which allows for basic programmatic access to secrets, decrypting |
| things if necessary |
| - as a CLI tool to 'synchronize' a directory containing plain/cipher files, |
| which means encrypting every new plaintext file (or new ciphertext file), |
| and re-encrypting all files whose keys are different from the keys list |
| defined in this file. |
| |
| """ |
| |
| import argparse |
| import logging |
| import os |
| import sys |
| import subprocess |
| import tempfile |
| |
| # Keys that are to be used to encrypt all secret roots. |
| keys = [ |
| "63DFE737F078657CC8A51C00C29ADD73B3563D82", # q3k |
| "482FF104C29294AD1CAF827BA43890A3DE74ECC7", # inf |
| "F07205946C07EEB2041A72FBC60C64879534F768", # cz2 |
| "0879F9FCA1C836677BB808C870FD60197E195C26", # implr |
| ] |
| |
| # Currently, Patryk's GPG key is expired. This hacks around that by pretending |
| # it's January 2021. |
| # TODO(q3k/patryk): remove this once Patryk updates his key. |
| systime = '20210101T000000' |
| |
| _logger_name = __name__ |
| if _logger_name == '__main__': |
| _logger_name = 'secretstore' |
| logger = logging.getLogger(_logger_name) |
| |
| |
| class CLIException(Exception): |
| pass |
| |
| |
| def encrypt(src, dst): |
| cmd = [ |
| 'gpg' , |
| '--encrypt', |
| '--faked-system-time', systime, |
| '--trust-model', 'always', |
| '--armor', |
| '--batch', '--yes', |
| '--output', dst, |
| ] |
| for k in keys: |
| cmd.append('--recipient') |
| cmd.append(k) |
| cmd.append(src) |
| subprocess.check_call(cmd) |
| |
| |
| def decrypt(src, dst): |
| cmd = ['gpg', '--decrypt', '--batch', '--yes', '--output', dst, src] |
| # catch stdout to make this code less chatty. |
| subprocess.check_output(cmd, stderr=subprocess.STDOUT) |
| |
| |
| def _encryption_key_for_fingerprint(fp): |
| """ |
| Returns the encryption key ID for a given GPG fingerprint (eg. one from the |
| 'keys' list. |
| """ |
| cmd = ['gpg', '-k', '--faked-system-time', systime, '--keyid-format', 'long', fp] |
| res = subprocess.check_output(cmd).decode() |
| |
| # Sample output: |
| # pub rsa4096/70FD60197E195C26 2014-02-22 [SC] [expires: 2021-02-05] |
| # 0879F9FCA1C836677BB808C870FD60197E195C26 |
| # uid [ultimate] Bartosz Stebel <bartoszstebel@gmail.com> |
| # uid [ultimate] Bartosz Stebel <implr@hackerspace.pl> |
| # sub rsa4096/E203C94E5CEBB3EF 2014-02-22 [E] [expires: 2021-02-05] |
| # |
| # We want to extract the 'sub' key with the [E] tag. |
| for line in res.split('\n'): |
| line = line.strip() |
| if not line: |
| continue |
| parts = line.split() |
| if len(parts) < 4: |
| continue |
| if parts[0] != 'sub': |
| continue |
| |
| if not parts[3].startswith('[') or not parts[3].endswith(']'): |
| continue |
| usages = parts[3].strip('[]') |
| if 'E' not in usages: |
| continue |
| |
| # Okay, we found the encryption key. |
| return parts[1].split('/')[1] |
| |
| raise Exception("Could not find encryption key ID for fingerprint {}".format(fp)) |
| |
| |
| _encryption_keys_cache = None |
| def encryption_keys(): |
| """ |
| Return all encryption keys associated with the keys array. |
| """ |
| global _encryption_keys_cache |
| if _encryption_keys_cache is None: |
| _encryption_keys_cache = [_encryption_key_for_fingerprint(fp) for fp in keys] |
| |
| return _encryption_keys_cache |
| |
| |
| def encrypted_for(path): |
| """ |
| Return for which encryption keys is a given GPG ciphertext file encrypted. |
| """ |
| cmd = ['gpg', '--pinentry-mode', 'cancel', '--list-packets', path] |
| res = subprocess.check_output(cmd, stderr=subprocess.STDOUT).decode() |
| |
| # Sample output: |
| # gpg: encrypted with 4096-bit RSA key, ID E203C94E5CEBB3EF, created 2014-02-22 |
| # "Bartosz Stebel <bartoszstebel@gmail.com>" |
| # gpg: encrypted with 2048-bit RSA key, ID 5C1B6B69E9F5EABE, created 2013-01-29 |
| # "Piotr Dobrowolski <piotr.tytus.dobrowolski@gmail.com>" |
| # gpg: encrypted with 2048-bit RSA key, ID 386E893E110BC55B, created 2012-01-10 |
| # "Sergiusz Bazanski (Low Latency Consulting) <serge@lowlatency.ie>" |
| # gpg: public key decryption failed: Operation cancelled |
| # gpg: decryption failed: No secret key |
| # # off=0 ctb=85 tag=1 hlen=3 plen=268 |
| # :pubkey enc packet: version 3, algo 1, keyid 386E893E110BC55B |
| # data: [2047 bits] |
| # # off=271 ctb=85 tag=1 hlen=3 plen=268 |
| # :pubkey enc packet: version 3, algo 1, keyid 5C1B6B69E9F5EABE |
| # data: [2048 bits] |
| # # off=542 ctb=85 tag=1 hlen=3 plen=524 |
| # :pubkey enc packet: version 3, algo 1, keyid E203C94E5CEBB3EF |
| # data: [4095 bits] |
| # # off=1069 ctb=d2 tag=18 hlen=2 plen=121 new-ctb |
| # :encrypted data packet: |
| # length: 121 |
| # mdc_method: 2 |
| |
| keys = [] |
| for line in res.split('\n'): |
| line = line.strip() |
| if not line: |
| continue |
| |
| parts = line.split() |
| if len(parts) < 9: |
| continue |
| |
| |
| if parts[:4] != [':pubkey', 'enc', 'packet:', 'version']: |
| continue |
| |
| if parts[7] != 'keyid': |
| continue |
| |
| keys.append(parts[8]) |
| |
| # Make unique. |
| return list(set(keys)) |
| |
| |
| class SyncAction: |
| """ |
| SyncAction is a possible action taken to synchronize some secrets. |
| |
| An action is some sort of side-effect bearing OS action (ie execution of |
| script or file move, or...) that can also 'describe' that it's acting - ie, |
| just return a human readable string of what it would be doing. These |
| describe descriptions are used for dry-runs of the secretstore sync |
| functionality. |
| """ |
| def describe(self): |
| return "" |
| |
| def act(self): |
| pass |
| |
| class SyncActionEncrypt: |
| def __init__(self, src, dst, reason): |
| self.src = src |
| self.dst = dst |
| self.reason = reason |
| |
| def describe(self): |
| return f'Encrypting {os.path.split(self.src)[-1]} ({self.reason})' |
| |
| def act(self): |
| return encrypt(self.src, self.dst) |
| |
| |
| class SyncActionDecrypt: |
| def __init__(self, src, dst, reason): |
| self.src = src |
| self.dst = dst |
| self.reason = reason |
| |
| def describe(self): |
| return f'Decrypting {os.path.split(self.src)[-1]} ({self.reason})' |
| |
| def act(self): |
| return decrypt(self.src, self.dst) |
| |
| |
| def sync(path: str, dry: bool): |
| """Synchronize (decrypt and encrypt what's needed) a given secrets directory.""" |
| |
| # Turn the path into an absolute path just to make things safer. |
| path = os.path.abspath(path) |
| # Trim all trailing slashes to canonicalize. |
| path = path.rstrip('/') |
| |
| plain_path = os.path.join(path, "plain") |
| cipher_path = os.path.join(path, "cipher") |
| |
| # Ensure that at least one of the plain/cipher paths exist. |
| plain_exists = os.path.exists(plain_path) |
| cipher_exists = os.path.exists(cipher_path) |
| if not plain_exists and not cipher_exists: |
| raise CLIException('Given directory must contain a plain/ or cipher/ subdirectory.') |
| |
| # Make missing directories. |
| if not plain_exists: |
| os.mkdir(plain_path) |
| if not cipher_exists: |
| os.mkdir(cipher_path) |
| |
| # List files on both sides: |
| plain_files = [f for f in os.listdir(plain_path) if f != '.gitignore' and os.path.isfile(os.path.join(plain_path, f))] |
| cipher_files = [f for f in os.listdir(cipher_path) if os.path.isfile(os.path.join(cipher_path, f))] |
| |
| # Helper function to turn a short filename within a directory to a pair |
| # of plain/cipher full paths. |
| def pc(p): |
| return os.path.join(plain_path, p), os.path.join(cipher_path, p) |
| |
| # Make a set of all file names - no matter if only available as plain, as |
| # cipher, or as both. |
| all_files = set(plain_files + cipher_files) |
| |
| # We'll be making a list of actions to perform to bring up given directory |
| # pair to a stable state. |
| actions = [] # type: List[SyncAction] |
| |
| # First, for every possible file (either encrypted or decrypted), figure |
| # out which side is fresher based on file presence and mtime. |
| fresher = {} # type: Dict[str, str] |
| for p in all_files: |
| # Handle the easy case when the file only exists on one side. |
| if p not in cipher_files: |
| fresher[p] = 'plain' |
| continue |
| if p not in plain_files: |
| fresher[p] = 'cipher' |
| continue |
| |
| plain, cipher = pc(p) |
| |
| # Otherwise, we have both the cipher and plain version. |
| # Check if the decrypted version matches the plaintext version. If so, |
| # they're both equal. |
| |
| f = tempfile.NamedTemporaryFile(delete=False) |
| f.close() |
| decrypt(cipher, f.name) |
| |
| with open(f.name, 'rb') as fd: |
| decrypted_data = fd.read() |
| with open(plain, 'rb') as fc: |
| current_data = fc.read() |
| |
| if decrypted_data == current_data: |
| fresher[p] = 'equal' |
| os.unlink(f.name) |
| continue |
| |
| os.unlink(f.name) |
| |
| # The plain and cipher versions differ. Let's choose based on mtime. |
| mtime_plain = os.path.getmtime(plain) |
| mtime_cipher = os.path.getmtime(cipher) |
| |
| if mtime_plain > mtime_cipher: |
| fresher[p] = 'plain' |
| elif mtime_cipher > mtime_plain: |
| fresher[p] = 'cipher' |
| else: |
| raise CLIException(f'cipher/plain stalemate on {p}: contents differ, but files have same mtime') |
| |
| # Find all files that need to be re-encrypted for changed keys. |
| reencrypt = set() |
| for p in cipher_files: |
| _, cipher = pc(p) |
| current = set(encrypted_for(cipher)) |
| want = set(encryption_keys()) |
| |
| if current != want: |
| reencrypt.add(p) |
| |
| # Okay, now actually construct a list of actions. |
| # First, all fresh==cipher keys need to be decrypted. |
| for p, v in fresher.items(): |
| if v != 'cipher': |
| continue |
| |
| plain, cipher = pc(p) |
| actions.append(SyncActionDecrypt(cipher, plain, "cipher version is newer")) |
| |
| encrypted = set() |
| # Then, encrypt all fresh==plain files, and make note of what those |
| # are. |
| for p, v in fresher.items(): |
| if v != 'plain': |
| continue |
| |
| plain, cipher = pc(p) |
| actions.append(SyncActionEncrypt(plain, cipher, "plain version is newer")) |
| encrypted.add(p) |
| |
| # Finally, re-encrypt all the files that aren't already being encrypted. |
| for p in reencrypt.difference(encrypted): |
| plain, cipher = pc(p) |
| actions.append(SyncActionEncrypt(plain, cipher, "needs to be re-encrypted for different keys")) |
| |
| if len(actions) == 0: |
| logger.info('Nothing to do!') |
| else: |
| if dry: |
| logger.info('Would perform the following:') |
| else: |
| logger.info('Running actions...') |
| for a in actions: |
| logger.info(a.describe()) |
| if not dry: |
| a.act() |
| |
| |
| class SecretStoreMissing(Exception): |
| pass |
| |
| |
| class SecretStore(object): |
| def __init__(self, plain_root, cipher_root): |
| self.proot = plain_root |
| self.croot = cipher_root |
| |
| def exists(self, suffix): |
| p = os.path.join(self.proot, suffix) |
| c = os.path.join(self.croot, suffix) |
| return os.path.exists(c) or os.path.exists(p) |
| |
| def plaintext(self, suffix): |
| p = os.path.join(self.proot, suffix) |
| c = os.path.join(self.croot, suffix) |
| |
| has_p = os.path.exists(p) |
| has_c = os.path.exists(c) |
| |
| if has_c and has_p and os.path.getctime(p) < os.path.getctime(c): |
| logger.info("Decrypting {} ({})...".format(suffix, c)) |
| decrypt(c, p) |
| |
| return p |
| |
| def open(self, suffix, mode, *a, **kw): |
| p = os.path.join(self.proot, suffix) |
| c = os.path.join(self.croot, suffix) |
| if 'w' in mode: |
| return open(p, mode, *a, **kw) |
| |
| if not self.exists(suffix): |
| raise SecretStoreMissing("Secret {} does not exist".format(suffix)) |
| |
| if not os.path.exists(p) or os.path.getctime(p) < os.path.getctime(c): |
| logger.info("Decrypting {} ({})...".format(suffix, c)) |
| decrypt(c, p) |
| |
| return open(p, mode, *a, **kw) |
| |
| |
| def main(): |
| parser = argparse.ArgumentParser(description='Manage hscloud git-based secrets.') |
| subparsers = parser.add_subparsers(dest='mode') |
| |
| parser_decrypt = subparsers.add_parser('decrypt', help='decrypt a single secret file') |
| parser_decrypt.add_argument('input', type=str, help='encrypted file path') |
| parser_decrypt.add_argument('output', type=str, default='-', help='decrypted file path file path (or - for stdout)') |
| |
| parser_encrypt = subparsers.add_parser('encrypt', help='encrypt a single secret file') |
| parser_encrypt.add_argument('input', type=str, help='plaintext file path') |
| parser_encrypt.add_argument('output', type=str, default='-', help='encrypted file path file path (or - for stdout)') |
| |
| parser_sync = subparsers.add_parser('sync', help='Synchronize a canonically formatted secrets/{plain,cipher} directory') |
| parser_sync.add_argument('dir', type=str, help='Path to secrets directory to synchronize') |
| parser_sync.add_argument('--dry', dest='dry', action='store_true') |
| parser_sync.set_defaults(dry=False) |
| |
| logging.basicConfig(level='INFO') |
| |
| args = parser.parse_args() |
| |
| if args.mode == None: |
| parser.print_help() |
| sys.exit(1) |
| |
| try: |
| if args.mode == 'encrypt': |
| encrypt(args.input, args.output) |
| elif args.mode == 'decrypt': |
| decrypt(args.input, args.output) |
| elif args.mode == 'sync': |
| sync(args.dir, dry=args.dry) |
| else: |
| # ??? |
| raise Exception('invalid mode {}'.format(args.mode)) |
| except CLIException as e: |
| logger.error(e) |
| sys.exit(1) |
| |
| if __name__ == '__main__': |
| sys.exit(main() or 0) |