ansible-role-luks/library/luks_manage.py

#!/usr/bin/python3

from ansible.module_utils.basic import *
import subprocess

class LuksManage:
  def __init__(self, device):
    self.device = device

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

  def create(self, cipher, size, key):
    p = subprocess.Popen(
      [
        'cryptsetup', '-q', 'luksFormat', '-c', cipher,
        '-s', str(size), self.device, '-d', '-'
      ], stdin=subprocess.PIPE
    )
    p.stdin.write(key.encode())
    p.communicate()[0]
    p.stdin.close()

    if p.returncode != 0:
      raise ValueError('Error to create the luks device {}'.format(self.device))

def main():
  fields = {
    'device': { 'type': 'str', 'required': True },
    'cipher': { 'type': 'str', 'default': 'aes-xts-plain' },
    'size':   { 'type': 'int', 'default': 256 },
    'key':    { 'type': 'str', 'required': True, 'no_log': True }
  }
  module = AnsibleModule(argument_spec=fields)
  changed = False

  luks = LuksManage(module.params['device'])
  if not luks.is_luks():
    luks.create(
      module.params['cipher'],
      module.params['size'],
      module.params['key']
    )
    changed = True

  module.exit_json(changed=changed)

if __name__ == '__main__':
  main()