python-script that imitates helm-secret (or sops) in minimal environment

opensslcrypt.py

#!/usr/bin/env python3
"""
The goal of this script:
- take yaml file
- parse objects inside
- encrypt/decrypt (de/base64) objects if it not dict or dict 

```sh
cat some.yaml
secret_value: some value

./opensslcrypt.py -e -f some.yaml -k 123

cat some.yaml
secret_value: U2FsdGVkX19OYhOhVkjZcgS4Ie43sXhf+MaellABZ/M=

./opensslcrypt.py -d -f some.yaml -k 123

cat some.yaml
secret_value: some value
```

"""
import argparse
import os
import yaml
import subprocess
import base64


def str_presenter(dumper, data):  # resolving multiline problem on dump to yaml
    if '\n' in data:  # check for multiline string
        return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')
    return dumper.represent_scalar('tag:yaml.org,2002:str', data)


yaml.add_representer(str, str_presenter)
# to use with safe_dump:
yaml.representer.SafeRepresenter.add_representer(str, str_presenter)


script_dir = os.path.dirname(os.path.realpath(__file__))

parser = argparse.ArgumentParser()
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--encrypt", "-e", action='store_true',
                   help="encrypt the given values file")
group.add_argument("--decrypt", "-d", action='store_true',
                   help="decrypt the given values file")
group.add_argument('--base64', '-b', action='store_true',
                   help="convert values in file to base64")
group.add_argument('--unbase64', '-B', action='store_true',
                   help="de-base64 values in file")
parser.add_argument("--key", "-k", type=str,
                    help="the base64 encoded aes key used to decrypt or encrypt the secret")
parser.add_argument("--file", "-f", type=str, nargs='+',
                    help="the path(s) to the target file(s)")


def string2b64(value: str) -> str:
    b = value.encode('utf-8')
    bb = base64.b64encode(b)
    return bb.decode('ascii')


def b64_to_string(value: str) -> str:
    bb = value.encode('ascii')
    b = base64.b64decode(bb)
    return b.decode('utf-8')


def string2aes(value: str) -> str:
    ossl_proc = subprocess.run(['openssl', 'aes-256-cbc', '-k', key,
                                '-pbkdf2', '-a'], input=value, capture_output=True, text=True)
    if ossl_proc.returncode != 0:
        print('error while decrypt:', ossl_proc.stderr, 'with value: \n', value)
        exit(1)
    return ossl_proc.stdout.strip().replace('\n', '')


def aes2string(value: str) -> str:
    try:
        ossl_proc = subprocess.run(['openssl', 'aes-256-cbc', '-d', '-k', key,'-pbkdf2', '-a', '-A'], input=value, capture_output=True, text=True)
    except UnicodeDecodeError as e:
        print(value, e)
    if ossl_proc.returncode != 0:
        print('error while decrypt:', ossl_proc.stderr, 'with value: \n', value)
        exit(1)
    ret = ossl_proc.stdout.strip()
    return ret


def perform_strings_in_file(tf: str, action):
    return action(yaml.safe_load(open(tf, 'r', encoding='utf-8')))


def encr(obj):
    if isinstance(obj, dict):
        for k, v in obj.items():
            if isinstance(v, dict):
                encr(v)
            elif isinstance(v, list):
                obj[k] = [encr(e) for e in v]
            else:
                obj[k] = string2aes(str(v))
        return obj
    elif isinstance(obj, str):
        return string2aes(obj)


def decr(obj):
    if isinstance(obj, dict):
        for k, v in obj.items():
            if isinstance(v, str):
                obj[k] = aes2string(v)
            else:
                obj[k] = decr(v)
        return obj
    elif isinstance(obj, list):
        return [decr(e) for e in obj]
    else:
        return aes2string(obj)


args = parser.parse_args()
key = args.key or os.environ.get('KEY') or None
if not key:
    print("# key arg must be supplied or \nexport KEY=$KEY")
    exit(1)
files = [*args.file] or None
if not files:
    print('files with secrets must be supplied via -f')
    exit(1)

to_, from_ = (decr, encr) if args.decrypt else (encr, decr)
for f in files:
    if os.path.isfile(f):
        result = perform_strings_in_file(f, to_)
        result_f, result_ext = os.path.splitext(os.path.abspath(f))
        result_f = os.path.join(result_f.replace(
            from_.__name__, to_.__name__) + result_ext)
        yaml.dump(result, open(result_f, 'w', encoding='utf-8'),
                  allow_unicode=True, sort_keys=False)
    else:
        print('error with opening file')
        exit(1)