Skip to content

Instantly share code, notes, and snippets.

@yann2192

yann2192/devault.py

Last active July 20, 2022 10:58
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save yann2192/b59310264e0728a5c6c5592f1a27863a to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save yann2192/b59310264e0728a5c6c5592f1a27863a to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. yann2192 revised this gist Jun 25, 2020. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion devault.py
    View file
    Original file line number Diff line number Diff line change
    @@ -30,7 +30,7 @@ def encrypt_string(obj):
    elif type(obj) is str:
    t = subprocess.check_output(['ansible-vault', 'encrypt_string', obj])
    return yaml.load(t.decode(), Loader=yaml.Loader)
    elif type(VaultTag):
    elif type(obj) is VaultTag:
    t = subprocess.check_output(['ansible-vault', 'decrypt'], input=obj.env_var.encode())
    return t.decode()
    else:
  2. yann2192 created this gist Jun 25, 2020.
    62 changes: 62 additions & 0 deletions devault.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,62 @@
    import subprocess
    import sys

    import yaml


    class VaultTag(yaml.YAMLObject):
    yaml_tag = "!vault"

    def __init__(self, env_var):
    self.env_var = env_var

    def __repr__(self):
    return "VaultTag({})".format(self.env_var)

    @classmethod
    def from_yaml(cls, loader, node):
    return VaultTag(node.value)

    @classmethod
    def to_yaml(cls, dumper, data):
    return dumper.represent_scalar(cls.yaml_tag, data.env_var, style='|')


    def encrypt_string(obj):
    if type(obj) is dict:
    for k in obj.keys():
    obj[k] = encrypt_string(obj[k])
    return obj
    elif type(obj) is str:
    t = subprocess.check_output(['ansible-vault', 'encrypt_string', obj])
    return yaml.load(t.decode(), Loader=yaml.Loader)
    elif type(VaultTag):
    t = subprocess.check_output(['ansible-vault', 'decrypt'], input=obj.env_var.encode())
    return t.decode()
    else:
    raise RuntimeError("unknown type {}".format(type(obj)))



    def replace(path):
    try:
    tmp = subprocess.check_output(['ansible-vault', 'view', path])
    except:
    with open(path, 'r') as f:
    tmp = f.read()

    data = yaml.load(tmp, Loader=yaml.Loader)
    data = encrypt_string(data)

    data = yaml.dump(data, Dumper=yaml.Dumper)

    with open(path, 'w') as f:
    f.write(data)


    if __name__ == "__main__":
    yaml.Loader.add_constructor('!vault', VaultTag.from_yaml)
    yaml.Dumper.add_representer(VaultTag, VaultTag.to_yaml)
    for i in sys.argv[1:]:
    print(i)
    replace(i)