add custom modules
continuous-integration/drone/push Build is passing Details

This commit is contained in:
Robert Kaussow 2020-08-18 23:44:49 +02:00
parent b84f50ed71
commit 660afb5392
Signed by: xoxys
GPG Key ID: 65362AE74AF98B61
10 changed files with 3314 additions and 4 deletions

View File

@ -1,5 +1,5 @@
[flake8]
ignore = D101, D102, D103, D107, D202, E402, W503
ignore = D101, D102, D103, D105, D107, D202, E402, W503
max-line-length = 99
inline-quotes = double
exclude =

View File

@ -1,9 +1,9 @@
"""Dynamic inventory plugin for Proxmox VE."""
# -*- coding: utf-8 -*-
# Copyright (c) 2014, Mathieu GAUTHIER-LAFAYE <gauthierl@lapth.cnrs.fr>
# Copyright (c) 2016, Matt Harris <matthaeus.harris@gmail.com>
# Copyright (c) 2020, Robert Kaussow <mail@thegeeklab.de>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
"""Dynamic inventory plugin for Proxmox VE."""
from __future__ import absolute_import, division, print_function

View File

@ -0,0 +1,256 @@
# -*- coding: utf-8 -*-
"""Module to control corenetworks DNS API."""
from __future__ import absolute_import, division, print_function
__metaclass__ = type
ANSIBLE_METADATA = {"metadata_version": "1.1", "status": ["preview"], "supported_by": "community"}
DOCUMENTATION = r"""
---
module: corenetworks_dns
short_description: Interface with the DNS API of core-networks.de
description:
- "Manages DNS zones and records via the core networks API, see the docs: U(https://beta.api.core-networks.de/doc/)."
options:
api_user:
description:
- Account API username. If omitted, the environment variables C(CN_API_USER) and C(CN_API_PASSWORD) will be looked for.
- You should prefere to use `api_token` or the `corenetworks_token` module to create one to prevent running into rate limits.
type: str
api_password:
description:
- Account API password.
type: str
api_token:
description:
- Account API token.
type: str
zone:
description:
- The name of the Zone to work with (e.g. "example.com").
- The Zone must already exist.
zone:
type: str
required: true
aliases: [ domain ]
record:
description:
- Used record relative to the given zone.
- Default is C(@) (e.g. the zone name).
type: str
default: "@"
aliases: [ name ]
type:
description:
- The type of DNS record to create.
choices: [ "A", "ALIAS", "CNAME", "MX", "SPF", "URL", "TXT", "NS", "SRV", "NAPTR", "PTR", "AAAA", "SSHFP", "HINFO", "POOL" ]
type: str
ttl:
description:
- The TTL to give the new record in seconds.
default: 3600
type: int
value:
description:
- Record value.
- Must be specified when trying to ensure a record exists.
type: str
solo:
description:
- Whether the record should be the only one for that record type and record name.
- Only use with C(state=present).
- This will delete all other records with the same record name and type.
type: bool
state:
description:
- whether the record should exist or not
choices: [ "present", "absent" ]
default: present
type: str
requirements:
- "corenetworks >= 0.1.4"
author: "Robert Kaussow (@xoxys)"
""" # noqa
EXAMPLES = """
- name: Create a test.my.com A record to point to 127.0.0.1
corenetworks_dns:
zone: my.com
record: test
type: A
value: 127.0.0.1
delegate_to: localhost
register: record
- name: Create a my.com CNAME record to example.com
corenetworks_dns:
zone: my.com
type: CNAME
value: example.com
state: present
delegate_to: localhost
- name: Change TTL value for a record
corenetworks_dns:
zone: my.com
type: CNAME
value: example.com
ttl: 600
state: present
delegate_to: localhost
- name: Delete the record
corenetworks_dns:
zone: my.com
type: CNAME
value: example.com
state: absent
delegate_to: localhost
"""
RETURN = r"""# """
import copy
import traceback
CORENETWORKS_IMP_ERR = None
try:
from corenetworks import CoreNetworks
from corenetworks.exceptions import CoreNetworksException
HAS_CORENETWORKS = True
except ImportError:
CORENETWORKS_IMP_ERR = traceback.format_exc()
HAS_CORENETWORKS = False
from ansible.module_utils.basic import AnsibleModule, missing_required_lib
def delete_records(client, module, zone, params, is_solo=False):
changed = False
search = copy.deepcopy(params)
if is_solo:
search.pop("data", None)
search.pop("ttl", None)
records = client.records(zone, params=search)
for r in records:
r["ttl"] = int(r["ttl"])
if is_solo:
if not (r["data"] == params["data"] and r["ttl"] == params["ttl"]):
changed = True
if not module.check_mode:
client.delete_record(zone, r)
else:
changed = True
if not module.check_mode:
client.delete_record(zone, r)
return changed
def add_record(client, module, zone, params):
changed = False
result = []
records = client.records(zone, params=params)
if len(records) > 1:
module.fail_json(
msg="More than one record already exists for the given attributes. "
"That should be impossible, please open an issue!"
)
if len(records) == 0:
changed = True
if not module.check_mode:
result = client.add_record(zone, params=params)
return result, changed
def main():
module = AnsibleModule(
argument_spec=dict(
api_user=dict(type="str"),
api_password=dict(type="str", no_log=True),
api_token=dict(type="str", no_log=True),
zone=dict(type="str", required=True, aliases=["domain"]),
record=dict(type="str", default="@", aliases=["name"]),
type=dict(
type="str",
choices=[
"A", "ALIAS", "CNAME", "MX", "SPF", "URL", "TXT", "NS", "SRV", "NAPTR", "PTR",
"AAAA", "SSHFP", "HINFO", "POOL"
]
),
ttl=dict(type="int", default=3600),
value=dict(type="str"),
solo=dict(type="bool", default=False),
state=dict(type="str", choices=["present", "absent"], default="present"),
),
required_together=[["record", "value"]],
supports_check_mode=True,
)
if not HAS_CORENETWORKS:
module.fail_json(msg=missing_required_lib("corenetworks"), exception=CORENETWORKS_IMP_ERR)
api_user = module.params.get("api_user")
api_password = module.params.get("api_password")
api_token = module.params.get("api_token")
zone = module.params.get("zone")
record = module.params.get("record")
record_type = module.params.get("type")
ttl = module.params.get("ttl")
value = module.params.get("value")
state = module.params.get("state")
is_solo = module.params.get("solo")
params = {"name": record, "ttl": ttl}
# sanity checks
if not record_type:
if state == "present":
module.fail_json(msg="Missing the record type")
else:
params["type"] = record_type
if not value:
if state == "present":
module.fail_json(msg="Missing the record value")
else:
params["data"] = value
if is_solo and state == "absent":
module.fail_json(msg="solo=true can only be used with state=present")
# perform actions
try:
# request throtteling to workaround the current rate limit
changed = False
if api_token:
client = CoreNetworks(api_token=api_token, auto_commit=True)
else:
client = CoreNetworks(user=api_user, password=api_password, auto_commit=True)
if state == "present":
changed_solo = False
if is_solo:
changed_solo = delete_records(client, module, zone, params, is_solo=True)
result, changed = add_record(client, module, zone, params)
module.exit_json(changed=changed_solo + changed, result=result)
# state is absent
else:
changed = delete_records(client, module, zone, params)
module.exit_json(changed=changed)
except CoreNetworksException as e:
module.fail_json(msg="Failure in core networks API communication: {}".format(str(e)))
module.fail_json(msg="Unknown what you wanted me to do")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,112 @@
# -*- coding: utf-8 -*-
"""Module to control corenetworks DNS API."""
from __future__ import absolute_import, division, print_function
__metaclass__ = type
ANSIBLE_METADATA = {"metadata_version": "1.1", "status": ["preview"], "supported_by": "community"}
DOCUMENTATION = """
---
module: corenetworks_dns
short_description: Interface with the DNS API of core-networks.de
description:
- "Manages DNS zones and records via the core networks API, see the docs: U(https://beta.api.core-networks.de/doc/)."
options:
api_user:
description:
- Account API username. If omitted, the environment variables C(CN_API_USER) and C(CN_API_PASSWORD) will be looked for.
type: str
api_password:
description:
- Account API password.
type: str
state:
description:
- whether the record should exist or not
choices: [ "present" ]
default: present
type: str
requirements:
- "corenetworks >= 0.1.3"
author: "Robert Kaussow (@xoxys)"
""" # noqa
EXAMPLES = """
- name: Obtain an API token using env variables
corenetworks_token:
delegate_to: localhost
register: my_token
- name: Obtain an API token using username and password attribute
corenetworks_token:
api_user: testuser
api_password: secure
delegate_to: localhost
register: my_token
- debug:
msg: "{{ my_token }}"
- name: Use the token
corenetworks_dns:
api_token: "{{ my_token.session.token }}"
zone: my.com
type: CNAME
value: example.com
state: present
delegate_to: localhost
"""
RETURN = r"""# """
import traceback
CORENETWORKS_IMP_ERR = None
try:
from corenetworks import CoreNetworks
from corenetworks.exceptions import CoreNetworksException
HAS_CORENETWORKS = True
except ImportError:
CORENETWORKS_IMP_ERR = traceback.format_exc()
HAS_CORENETWORKS = False
from ansible.module_utils.basic import AnsibleModule, missing_required_lib
def main():
module = AnsibleModule(
argument_spec=dict(
api_user=dict(type="str"),
api_password=dict(type="str", no_log=True),
state=dict(type="str", choices=["present"], default="present"),
),
supports_check_mode=True,
)
if not HAS_CORENETWORKS:
module.fail_json(msg=missing_required_lib("corenetworks"), exception=CORENETWORKS_IMP_ERR)
api_user = module.params.get("api_user")
api_password = module.params.get("api_password")
# perform actions
try:
# request throtteling to workaround the current rate limit
changed = False
client = CoreNetworks(user=api_user, password=api_password, auto_commit=True)
session = {"token": client._auth.token}
if hasattr(client._auth, "expires"):
session["expires"] = client._auth.expires.strftime("%Y-%m-%d, %H:%M:%S")
module.exit_json(changed=changed, session=session)
except CoreNetworksException as e:
module.fail_json(msg="Failure in core networks API communication: {}".format(str(e)))
module.fail_json(msg="Unknown what you wanted me to do")
if __name__ == "__main__":
main()

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,415 @@
# -*- coding: utf-8 -*-
"""OpenSSL PKCS12 module."""
ANSIBLE_METADATA = {"metadata_version": "1.0", "status": ["preview"], "supported_by": "community"}
DOCUMENTATION = """
---
module: openssl_pkcs12
author: "Guillaume Delpierre (@gdelpierre)"
version_added: "2.4"
short_description: Generate OpenSSL pkcs12 archive.
description:
- "This module allows one to (re-)generate PKCS#12."
requirements:
- "python-pyOpenSSL"
options:
ca_certificates:
required: False
description:
- List of CA certificate to include.
cert_path:
required: False
description:
- The path to read certificates and private keys from.
Must be in PEM format.
action:
required: False
default: "export"
choices: ["parse", "export"]
description:
- Create (export) or parse a PKCS#12.
src:
required: False
description:
- PKCS#12 file path to parse.
path:
required: True
default: null
description:
- Filename to write the PKCS#12 file to.
force:
required: False
default: False
description:
- Should the file be regenerated even it it already exists.
friendly_name:
required: False
default: null
aliases: "name"
description:
- Specifies the friendly name for the certificate and private key.
iter_size:
required: False
default: 2048
description:
- Number of times to repeat the encryption step.
maciter_size:
required: False
default: 1
description:
- Number of times to repeat the MAC step.
mode:
required: False
default: 0400
description:
- Default mode for the generated PKCS#12 file.
passphrase:
required: False
default: null
description:
- The PKCS#12 password.
privatekey_path:
required: False
description:
- File to read private key from.
privatekey_passphrase:
required: False
default: null
description:
- Passphrase source to decrypt any input private keys with.
state:
required: False
default: "present"
choices: ["present", "absent"]
description:
- Whether the file should exist or not.
"""
EXAMPLES = """
- name: "Generate PKCS#12 file"
openssl_pkcs12:
path: "/opt/certs/ansible.p12"
friendly_name: "raclette"
privatekey_path: "/opt/certs/keys/key.pem"
cert_path: "/opt/certs/cert.pem"
ca_certificates: "/opt/certs/ca.pem"
state: present
- name: "Change PKCS#12 file permission"
openssl_pkcs12:
path: "/opt/certs/ansible.p12"
friendly_name: "raclette"
privatekey_path: "/opt/certs/keys/key.pem"
cert_path: "/opt/certs/cert.pem"
ca_certificates: "/opt/certs/ca.pem"
state: present
mode: 0600
- name: "Regen PKCS#12 file"
openssl_pkcs12:
path: "/opt/certs/ansible.p12"
friendly_name: "raclette"
privatekey_path: "/opt/certs/keys/key.pem"
cert_path: "/opt/certs/cert.pem"
ca_certificates: "/opt/certs/ca.pem"
state: present
mode: 0600
force: True
- name: "Dump/Parse PKCS#12 file"
openssl_pkcs12:
src: "/opt/certs/ansible.p12"
path: "/opt/certs/ansible.pem"
state: present
- name: "Remove PKCS#12 file"
openssl_pkcs12:
path: "/opt/certs/ansible.p12"
state: absent
"""
RETURN = """
filename:
description: Path to the generate PKCS#12 file.
returned: changed or success
type: string
sample: /opt/certs/ansible.p12
"""
import errno
import os
from ansible.module_utils._text import to_native
from ansible.module_utils.basic import AnsibleModule
try:
from OpenSSL import crypto
except ImportError:
pyopenssl_found = False
else:
pyopenssl_found = True
class PkcsError(Exception):
pass
class Pkcs(object):
def __init__(self, module):
self.path = module.params["path"]
self.force = module.params["force"]
self.state = module.params["state"]
self.action = module.params["action"]
self.check_mode = module.check_mode
self.iter_size = module.params["iter_size"]
self.maciter_size = module.params["maciter_size"]
self.pkcs12 = None
self.src = module.params["src"]
self.privatekey_path = module.params["privatekey_path"]
self.privatekey_passphrase = module.params["privatekey_passphrase"]
self.cert_path = module.params["cert_path"]
self.ca_certificates = module.params["ca_certificates"]
self.friendly_name = module.params["friendly_name"]
self.passphrase = module.params["passphrase"]
self.mode = module.params["mode"]
self.changed = False
if not self.mode:
self.mode = int("0400", 8)
def load_privatekey(self, path, passphrase=None):
"""Load the specified OpenSSL private key."""
try:
if passphrase:
privatekey = crypto.load_privatekey(
crypto.FILETYPE_PEM,
open(path, "rb").read(), passphrase
)
else:
privatekey = crypto.load_privatekey(crypto.FILETYPE_PEM, open(path, "rb").read())
return privatekey
except (IOError, OSError) as exc:
raise PkcsError(exc)
def load_certificate(self, path):
"""Load the specified certificate."""
try:
cert_content = open(path, "rb").read()
cert = crypto.load_certificate(crypto.FILETYPE_PEM, cert_content)
return cert
except (IOError, OSError) as exc:
raise PkcsError(exc)
def load_pkcs12(self, path, passphrase=None):
"""Load pkcs12 file."""
try:
if passphrase:
return crypto.load_pkcs12(open(path, "rb").read(), passphrase)
else:
return crypto.load_pkcs12(open(path, "rb").read())
except (IOError, OSError) as exc:
raise PkcsError(exc)
def dump_privatekey(self, path):
"""Dump the specified OpenSSL private key."""
try:
return crypto.dump_privatekey(
crypto.FILETYPE_PEM,
self.load_pkcs12(path).get_privatekey()
)
except (IOError, OSError) as exc:
raise PkcsError(exc)
def dump_certificate(self, path):
"""Dump the specified certificate."""
try:
return crypto.dump_certificate(
crypto.FILETYPE_PEM,
self.load_pkcs12(path).get_certificate()
)
except (IOError, OSError) as exc:
raise PkcsError(exc)
def generate(self, module):
"""Generate PKCS#12 file archive."""
if not os.path.exists(self.path) or self.force:
self.pkcs12 = crypto.PKCS12()
try:
self.remove()
except PkcsError as exc:
module.fail_json(msg=to_native(exc))
if self.ca_certificates:
ca_certs = [self.load_certificate(ca_cert) for ca_cert in self.ca_certificates]
self.pkcs12.set_ca_certificates(ca_certs)
if self.cert_path:
self.pkcs12.set_certificate(self.load_certificate(self.cert_path))
if self.friendly_name:
self.pkcs12.set_friendlyname(self.friendly_name)
if self.privatekey_path:
self.pkcs12.set_privatekey(
self.load_privatekey(self.privatekey_path, self.privatekey_passphrase)
)
try:
with open(self.path, "wb", self.mode) as archive:
archive.write(
self.pkcs12.export(self.passphrase, self.iter_size, self.maciter_size)
)
module.set_mode_if_different(self.path, self.mode, False)
self.changed = True
except (IOError, OSError) as exc:
self.remove()
raise PkcsError(exc)
file_args = module.load_file_common_arguments(module.params)
if module.set_fs_attributes_if_different(file_args, False):
module.set_mode_if_different(self.path, self.mode, False)
self.changed = True
def parse(self, module):
"""Read PKCS#12 file."""
if not os.path.exists(self.path) or self.force:
try:
self.remove()
with open(self.path, "wb") as content:
content.write(
"{0}{1}".format(
self.dump_privatekey(self.src), self.dump_certificate(self.src)
)
)
module.set_mode_if_different(self.path, self.mode, False)
self.changed = True
except IOError as exc:
raise PkcsError(exc)
file_args = module.load_file_common_arguments(module.params)
if module.set_fs_attributes_if_different(file_args, False):
module.set_mode_if_different(self.path, self.mode, False)
self.changed = True
def remove(self):
"""Remove the PKCS#12 file archive from the filesystem."""
try:
os.remove(self.path)
self.changed = True
except OSError as exc:
if exc.errno != errno.ENOENT:
raise PkcsError(exc)
else:
pass
def check(self, module, perms_required=True):
def _check_pkey_passphrase():
if self.privatekey_passphrase:
try:
self.load_privatekey(self.path, self.privatekey_passphrase)
return True
except crypto.Error:
return False
return True
if not os.path.exists(self.path):
return os.path.exists(self.path)
return _check_pkey_passphrase
def dump(self):
"""Serialize the object into a dictionary."""
result = {
"changed": self.changed,
"filename": self.path,
}
if self.privatekey_path:
result["privatekey_path"] = self.privatekey_path
return result
def main():
argument_spec = dict(
action=dict(default="export", choices=["parse", "export"], type="str"),
ca_certificates=dict(type="list"),
cert_path=dict(type="path"),
force=dict(default=False, type="bool"),
friendly_name=dict(type="str", aliases=["name"]),
iter_size=dict(default=2048, type="int"),
maciter_size=dict(default=1, type="int"),
passphrase=dict(type="str", no_log=True),
path=dict(required=True, type="path"),
privatekey_path=dict(type="path"),
privatekey_passphrase=dict(type="str", no_log=True),
state=dict(default="present", choices=["present", "absent"], type="str"),
src=dict(type="path"),
)
required_if = [
["action", "export", ["friendly_name"]],
["action", "parse", ["src"]],
]
required_together = [
["privatekey_path", "friendly_name"],
]
module = AnsibleModule(
argument_spec=argument_spec,
add_file_common_args=True,
required_if=required_if,
required_together=required_together,
supports_check_mode=True,
)
if not pyopenssl_found:
module.fail_json(msg="The python pyOpenSSL library is required")
base_dir = os.path.dirname(module.params["path"])
if not os.path.isdir(base_dir):
module.fail_json(
name=base_dir,
msg="The directory {0} does not exist or "
"the file is not a directory".format(base_dir)
)
pkcs12 = Pkcs(module)
if module.params["state"] == "present":
if module.check_mode:
result = pkcs12.dump()
result["changed"] = module.params["force"] or not pkcs12.check(module)
module.exit_json(**result)
try:
if module.params["action"] == "export":
pkcs12.generate(module)
else:
pkcs12.parse(module)
except PkcsError as exc:
module.fail_json(msg=to_native(exc))
else:
if module.check_mode:
result = pkcs12.dump()
result["changed"] = os.path.exists(module.params["path"])
module.exit_json(**result)
try:
pkcs12.remove()
except PkcsError as exc:
module.fail_json(msg=to_native(exc))
result = pkcs12.dump()
module.exit_json(**result)
if __name__ == "__main__":
main()

File diff suppressed because it is too large Load Diff

116
plugins/modules/ucr.py Normal file
View File

@ -0,0 +1,116 @@
# -*- coding: utf-8 -*-
"""Module to control Univention Corporate Registry."""
ANSIBLE_METADATA = {"metadata_version": "1.1", "status": ["preview"], "supported_by": "community"}
DOCUMENTATION = """
---
module: ucr
short_description: Manage variables in univention configuration registry.
version_added: "2.6"
description:
- "This module allows to manage variables inside the univention configuration registry
on a univention corporate server (UCS)."
options:
path:
description:
- Path for the variable
required: True
default: null
value:
description:
- New value of the variable
required: False
state:
required: False
default: "present"
choices: ["present", "absent"]
description:
- Whether the variable should be exist or not.
author:
- Robert Kaussow (@xoxys)
"""
EXAMPLES = """
# Set variable to force https in ucs frontend
- name: Force https
ucr:
path: apache2/force_https
value: yes
# Allow another user as root to login as ssh
- name: Add ssh user
ucr:
path: auth/sshd/user/myuser
value: yes
"""
RETURN = """
original_message:
description: The original name param that was passed in
type: str
message:
description: The output message that the sample module generates
"""
from ansible.module_utils.basic import AnsibleModule
from univention.config_registry import ConfigRegistry # noqa
from univention.config_registry.frontend import ucr_update # noqa
def get_variable(ucr, path):
ucr.load()
if path in ucr:
value = ucr.get(path)
else:
value = None
return value
def set_variable(ucr, path, value, result):
org_value = get_variable(ucr, path)
ucr_update(ucr, {path: value})
new_value = get_variable(ucr, path)
return not org_value == new_value
def dry_variable(ucr, path, value, result):
org_value = get_variable(ucr, path)
return not org_value == value
def main():
ucr = ConfigRegistry()
module_args = dict(
path=dict(type="str", required=True, aliases=["name"]),
value=dict(type="str", required=False, default=""),
state=dict(default="present", choices=["present", "absent"], type="str")
)
required_if = [["state", "present", ["value"]]]
module = AnsibleModule(
argument_spec=module_args, supports_check_mode=True, required_if=required_if
)
result = dict(changed=False, original_message="", message="")
path = module.params["path"]
value = module.params["value"]
if module.params["state"] == "present":
if value is None or value == "None":
value = ""
elif module.params["state"] == "absent":
value = None
if not module.check_mode:
result["changed"] = set_variable(ucr, path, value, result)
else:
result["changed"] = dry_variable(ucr, path, value, result)
module.exit_json(**result)
if __name__ == "__main__":
main()

View File

@ -4,7 +4,7 @@ known_first_party = ansiblelater
sections = FUTURE,STDLIB,THIRDPARTY,FIRSTPARTY,LOCALFOLDER
force_single_line = true
line_length = 99
skip_glob = **/.env*,**/env/*,**/docs/*,**/inventory/*
skip_glob = **/.env*,**/env/*,**/docs/*,**/inventory/*,**/modules/*
[yapf]
based_on_style = google

View File

@ -1,5 +1,11 @@
ansible
# requirement for the proxmox module
# requirement for the proxmox modules
proxmoxer
requests
# requirement for the corenetworks modules
corenetworks
# requirement for the openssl_pkcs12 module
pyOpenSSL