unit_file.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-

# (c) 2020-2023, Bodo Schulz <bodo@boone-schulz.de>
# Apache-2.0 (see LICENSE or https://opensource.org/license/apache-2-0)
# SPDX-License-Identifier: Apache-2.0

from __future__ import annotations

import os
import shutil

from ansible.module_utils.basic import AnsibleModule
from ansible_collections.bodsch.core.plugins.module_utils.checksum import Checksum
from ansible_collections.bodsch.core.plugins.module_utils.directory import (
    create_directory,
)

# ---------------------------------------------------------------------------------------

DOCUMENTATION = """
module: unit_file
author:
  - Bodo 'bodsch' Schulz (@bodsch)
short_description: Creates a systemd unit file.
version_added: 1.2.0

description:
  - Creates a systemd unit file.
  - The `service`, `timer` and `socket` types are supported.

options:
  unit_type:
    description:
      - The unit type to be created.
      - The (C(service), C(timer) and C(socket)) types are supported.
      - Default is C(service).
    type: str
    choices: [ service, timer, socket ]
  name:
    description:
      - The name of the unit file.
    type: str
  state:
    description:
      - Whether to install (C(present) or remove (C(absent) a unit file.
      - Default is C(present).
    type: str
    choices: [ absent, present ]
  drop_ins:
    description:
      - A list of possible systemd drop-ins.
      - The structure corresponds exactly to that of a normal unit file.
    type: list
  unit_file:
    description: The content of a systemd unit file.
    type: dict

"""

EXAMPLES = """
- name: create getty drop-ins
  bodsch.systemd.unit_file:
    name: "getty@tty1"
    state: "present"
    unit_type: "service"
    drop_ins:
      - name: autologin
        state: present
        service:
          ExecStart:
            - ""
            - "{% raw %}-/sbin/agetty -o '-p -f -- \\\\u' --noclear --autologin username %I $TERM{% endraw %}"
          Type: simple

      - name: noclear
        state: absent
        service:
          TTYVTDisallocate: false
  when:
    - ansible_service_mgr == 'systemd'

- name: create systemd unit file
  bodsch.systemd.unit_file:
    name: "vaultwarden"
    state: "present"
    unit_type: "service"
    unit_file:
      description: |
        #
        #
        #
        #
        #
        #

      Unit:
        Description: Vaultwarden API server
        Documentation: https://github.com/dani-garcia/vaultwarden
        After: network.target
      Service:
        Type: simple
        User: vaultwarden
        Group: vaultwarden
        LimitNOFILE: 1048576
        UMask: "0077"

        ExecStart: /usr/bin/vaultwarden
        EnvironmentFile: /etc/vaultwarden/config.env
      Install:
        WantedBy: multi-user.target
  when:
    - ansible_service_mgr == 'systemd'

- name: create systemd socket
  bodsch.systemd.unit_file:
    name: "systemd-initctl"
    state: "present"
    unit_type: "socket"
    unit_file:
      description: |
        #  SPDX-License-Identifier: LGPL-2.1-or-later
        #
        #  This file is part of systemd.
        #
        #  systemd is free software; you can redistribute it and/or modify it
        #  under the terms of the GNU Lesser General Public License as published by
        #  the Free Software Foundation; either version 2.1 of the License, or
        #  (at your option) any later version.
      unit:
        Description: initctl Compatibility Named Pipe
        Documentation: man:systemd-initctl.socket(8)
        DefaultDependencies: no
        Before: sockets.target

      Socket:
        ListenFIFO: /run/initctl
        Symlinks: /dev/initctl
        SocketMode: "0600"
  when:
    - ansible_service_mgr == 'systemd'


- name: create systemd timer
  bodsch.systemd.unit_file:
    name: "systemd-tmpfiles-clean"
    state: "present"
    unit_type: "timer"
    unit_file:
      unit:
        Description: Daily Cleanup of Temporary Directories
        Documentation: man:tmpfiles.d(5) man:systemd-tmpfiles(8)
        ConditionPathExists: "!/etc/initrd-release"
      timer:
        OnBootSec: 15min
        OnUnitActiveSec: 1d
  when:
    - ansible_service_mgr == 'systemd'
"""

RETURN = """
changed:
  type: bool
  description: Status of the action
msg:
  type: str
  description: Readable edition of what has been done.
"""

# --------------------------------------------------------------------------------------------------

UNIT_TPL = """# generated by ansible

{% if item.description is defined %}
{{ item.description }}
{% set _ = item.pop('description') %}
{% endif %}
{% for section, options in item.items() %}
[{{ section | capitalize }}]
  {% for option, values in options.items() %}
    {% if values is string or values is number %}
{{ option.ljust(34) }} = {{ values }}
    {% else %}
      {% for value in values %}
{{ option.ljust(34) }} = {{ value }}
      {% endfor %}
    {% endif %}
  {% endfor %}

{% endfor %}
{#
#}
"""


class SystemdUnitFile(object):
    """ """

    module = None

    def __init__(self, module):
        """ """
        self.module = module

        self.unit_type = module.params.get("unit_type")
        self.name = module.params.get("name")
        self.state = module.params.get("state")
        self.overwrite = module.params.get("overwrite")
        self.drop_ins = module.params.get("drop_ins")
        self.unit_file = module.params.get("unit_file")

        self.tmp_directory = os.path.join(
            "/run/.ansible", f"systemd_unit.{str(os.getpid())}"
        )

    def run(self):
        """ """
        result = dict(
            rc=1,
            failed=True,
            changed=False,
        )

        self.checksum = Checksum(self.module)

        if self.state == "absent":
            result = self.clean_unit_files()

        else:
            result = self.create_unit_files()

        if os.path.exists(self.tmp_directory):
            shutil.rmtree(self.tmp_directory)

        return result

    def create_unit_files(self):
        """ """
        if isinstance(self.drop_ins, list) and len(self.drop_ins) > 0:

            create_directory(directory=self.tmp_directory, mode="0750")
            result = self.create_drop_in(self.drop_ins)

        if isinstance(self.unit_file, dict) and len(self.unit_file) > 0:

            create_directory(directory=self.tmp_directory, mode="0750")
            result = self.create_unit_file(self.unit_file)

        return result

    def clean_unit_files(self):
        """ """
        if isinstance(self.drop_ins, list) and len(self.drop_ins) > 0:

            service_name = f"/etc/systemd/system/{self.name}.d"

            for drop_in in self.drop_ins:
                name = drop_in.get("name")
                state = drop_in.get("state", "present")

                if state == "absent":
                    unit_file = os.path.join(service_name, f"{name}.conf")
                    result = self.__remove_unit(unit_file)

        if isinstance(self.unit_file, dict) and len(self.unit_file) > 0:
            unit_file = os.path.join(
                "/lib/systemd/system", f"{self.name}.{self.unit_type}"
            )
            result = self.__remove_unit(unit_file)

        return result

    def create_drop_in(self, data):
        """ """
        service_name = f"/etc/systemd/system/{self.name}.d"

        if not os.path.exists(service_name):
            create_directory(service_name)

        for drop_in in data:
            name = drop_in.get("name")
            state = drop_in.get("state", "present")

            unit_file = os.path.join(service_name, f"{name}.conf")
            file_temporary = os.path.join(self.tmp_directory, f"{name}.conf")

            if state == "present":
                file_temporary = os.path.join(self.tmp_directory, f"{name}.conf")

                data = self.__template(drop_in)

                with open(file_temporary, "w") as f:
                    f.write(data)

                result = self.__changed(file_temporary, unit_file)

            else:
                result = self.__remove_unit(unit_file)

        return result

    def create_unit_file(self, data):
        """ """
        unit_file = os.path.join("/lib/systemd/system", f"{self.name}.{self.unit_type}")
        file_temporary = os.path.join(
            self.tmp_directory, f"{self.name}.{self.unit_type}"
        )

        if self.state == "present":
            data = self.__template(data)

            with open(file_temporary, "w") as f:
                f.write(data)

            result = self.__changed(file_temporary, unit_file)
        else:
            pass

            result = dict(changed=False, failed=False)

        return result

    def __changed(self, file_temporary, unit_file):
        """ """

        old_checksum = self.checksum.checksum_from_file(unit_file)
        new_checksum = self.checksum.checksum_from_file(file_temporary)

        changed = not (new_checksum == old_checksum)
        new_file = False
        msg = "The unit-file has not been changed"

        # self.module.log(msg=f"   file_name      {unit_file}")
        # self.module.log(msg=f"   file_temporary {file_temporary}")
        # self.module.log(msg=f"   old_checksum   {old_checksum}")
        # self.module.log(msg=f"   new_checksum   {new_checksum}")
        # self.module.log(msg=f"   changed        {changed}")

        if changed:
            new_file = old_checksum is None
            shutil.move(file_temporary, unit_file)
            msg = "The unit-file was successfully changed"

        if new_file:
            msg = "The unit-file was successfully created"

        result = dict(changed=changed, msg=msg)

        return result

    def __remove_unit(self, unit_file):
        """ """
        if os.path.exists(unit_file):
            os.remove(unit_file)

            result = dict(
                changed=True,
                failed=False,
                msg="The unit-file was successfully removed.",
            )
        else:
            result = dict(
                changed=False,
                failed=False,
                msg="The unit-file has already been removed.",
            )

        return result

    def __template(self, data):
        """ """
        # self.module.log(msg=f"__template({data} ({type(data)}))")

        if isinstance(data, dict):
            from jinja2 import Template

            if data.get("name"):
                _ = data.pop("name")
            if data.get("state"):
                _ = data.pop("state")

            # self.module.log(msg=f"{data} ({type(data)})")

            tm = Template(UNIT_TPL, trim_blocks=True, lstrip_blocks=True)
            d = tm.render(item=data)
        else:
            d = None

        return d


def main():
    """ """
    args = dict(
        unit_type=dict(
            choose=["service", "socket", "timer"], default="service", type="str"
        ),
        name=dict(required=True, type="str"),
        state=dict(
            choose=[
                "absent",
                "present",
            ],
            default="present",
            type="str",
        ),
        overwrite=dict(required=False, default=False, type="bool"),
        drop_ins=dict(required=False, default=[], type=list),
        unit_file=dict(required=False, default={}, type=dict),
    )

    module = AnsibleModule(
        argument_spec=args,
        supports_check_mode=False,
    )

    k = SystemdUnitFile(module)
    result = k.run()

    # module.log(msg=f"= result: {result}")

    module.exit_json(**result)


# import module snippets
if __name__ == "__main__":
    main()