ansible-role-luks/library/luks_decrypt.py

#!/usr/bin/python3

from ansible.module_utils.basic import *
import subprocess
import os

class LuksDecrypt:
  def __init__(self, name, device):
    self.name   = name
    self.device = device

  def is_luks(self):
    if subprocess.call(['cryptsetup', 'isLuks', self.device]) == 0:
      return True
    else:
      return False

  def is_decrypted(self):
    return os.path.exists('/dev/mapper/{}'.format(self.name)) 

  def decrypt(self, key):
    p = subprocess.Popen(
      [
        'cryptsetup', '-q', 'open', '-d', '-', self.device, self.name
      ], stdin=subprocess.PIPE
    )
    p.stdin.write(key.encode())
    p.communicate()[0]
    p.stdin.close()

    if p.returncode != 0:
      raise ValueError('Error during the decrypt of device {}'.format(self.device))

def main():
  fields = {
    'name':   { 'type': 'str', 'required': True },
    'device': { 'type': 'str', 'required': True },
    'key':    { 'type': 'str', 'required': True }
  }
  module = AnsibleModule(argument_spec=fields)
  changed = False

  luks = LuksDecrypt(module.params['name'], module.params['device'])
  if not luks.is_luks():
    raise ValueError('Error the device {} is not a LUKS device'.format(module.params['device']))
  elif not luks.is_decrypted():
    luks.decrypt(module.params['key'])
    changed = True

  module.exit_json(changed=changed)

if __name__ == '__main__':
  main()