1
0
Fork 0

Commits vergleichen

..

Keine gemeinsamen Commits. "main" und "v0.1.0" haben vollständig unterschiedliche Historien.
main ... v0.1.0

29 geänderte Dateien mit 939 neuen und 2225 gelöschten Zeilen

Datei anzeigen

@ -1,18 +0,0 @@
# EditorConfig is awesome: https://EditorConfig.org
# top-most EditorConfig file
root = true
[*]
indent_style = space
indent_size = 4
tab_width = 4
end_of_line = lf
charset = utf-8
trim_trailing_whitespace = true
insert_final_newline = true
[*.{yaml,yml}]
indent_size = 2
[Makefile]
indent_style = tab

2
.gitignore vendored
Datei anzeigen

@ -1,8 +1,6 @@
# ---> Ansible
*.retry
tests/output
*.tar.gz
changelogs/.plugin-cache.yaml
# ---> Python
# Byte-compiled / optimized / DLL files

Datei anzeigen

@ -1,154 +0,0 @@
=====================================
sebastian.systemd 0.4.3 Release Notes
=====================================
.. contents:: Topics
v0.4.3
======
Changelog
---------
added new options to system_service
v0.4.2
======
Changelog
---------
upgraded to the new method used by sebastian.base
v0.4.0
======
Changelog
---------
added timer module
v0.3.8
======
Changelog
---------
fixed the import again.
v0.3.7
======
Changelog
---------
fixed the import path
v0.3.6
======
Changelog
---------
fixed the import of the module utils
v0.3.5
======
Changelog
---------
added some ignores and changed my name
v0.3.4
======
Changelog
---------
moved the module_utils to an indipendent repository
v0.3.3
======
Changelog
---------
added netdev module
v0.3.2
======
Changelog
---------
added virtualization and negative matches to link and network
v0.3.1
======
Changelog
---------
fixed runtime.yml
v0.3.0
======
Changelog
---------
added socket module and some small fixes
v0.2.0
======
Changelog
---------
added an rudimentary system_service module
v0.1.3
======
Changelog
---------
renamed all modules to names without systemd_ prefix
v0.1.2
======
Changelog
---------
Pfad der auflösung für Tests und autodoc geändert, weil Mitogen sonst blockiert
v0.1.1
======
Changelog
---------
kleines feature update
kleine Änderungen
-----------------
- update_doc hinzugefügt. um Dokumentation automatisch zu aktualisieren
v0.1.0
======
Changelog
---------
Erstes Release
Große Änderungen
----------------
- added systemd_link module
- added systemd_mount module
- added systemd_network module

Datei anzeigen

@ -1,15 +0,0 @@
format:
black .
isort .
changelog:
antsibull-changelog generate
docs: format
update-doc
release: changelog docs
ansible-galaxy collection build --output-path dist
upload: release
./upload.sh

Datei anzeigen

@ -1,76 +0,0 @@
ancestor: null
releases:
0.1.0:
changes:
major_changes:
- added systemd_link module
- added systemd_mount module
- added systemd_network module
release_summary: Erstes Release
release_date: "2023-04-15"
0.1.1:
changes:
minor_changes:
- "update_doc hinzugef\xFCgt. um Dokumentation automatisch zu aktualisieren"
release_summary: kleines feature update
release_date: "2023-04-15"
0.1.2:
changes:
release_summary: "Pfad der aufl\xF6sung f\xFCr Tests und autodoc ge\xE4ndert, weil Mitogen sonst blockiert"
release_date: "2023-04-15"
0.1.3:
changes:
release_summary: renamed all modules to names without systemd_ prefix
release_date: "2023-04-20"
0.2.0:
changes:
release_summary: added an rudimentary system_service module
release_date: "2023-04-21"
0.3.0:
changes:
release_summary: added socket module and some small fixes
release_date: "2023-04-23"
0.3.1:
changes:
release_summary: fixed runtime.yml
release_date: "2023-04-23"
0.3.2:
changes:
release_summary: added virtualization and negative matches to link and network
release_date: "2023-07-15"
0.3.3:
changes:
release_summary: added netdev module
release_date: "2023-07-16"
0.3.4:
changes:
release_summary: moved the module_utils to an indipendent repository
release_date: "2023-11-26"
0.3.5:
changes:
release_summary: added some ignores and changed my name
release_date: "2023-12-30"
0.3.6:
changes:
release_summary: fixed the import of the module utils
release_date: "2024-02-11"
0.3.7:
changes:
release_summary: fixed the import path
release_date: "2024-02-11"
0.3.8:
changes:
release_summary: fixed the import again.
release_date: "2024-02-11"
0.4.0:
release_date: "2024-02-24"
changes:
release_summary: added timer module
0.4.2:
release_date: "2024-03-09"
changes:
release_summary: upgraded to the new method used by sebastian.base
0.4.3:
release_date: "2024-03-09"
changes:
release_summary: added new options to system_service

Datei anzeigen

@ -1,32 +0,0 @@
changelog_filename_template: ../CHANGELOG.rst
changelog_filename_version_depth: 3
changes_file: changelog.yaml
changes_format: combined
ignore_other_fragment_extensions: true
keep_fragments: false
mention_ancestor: true
new_plugins_after_name: removed_features
notesdir: fragments
prelude_section_name: release_summary
prelude_section_title: Changelog
sanitize_changelog: true
sections:
- - major_changes
- Große Änderungen
- - minor_changes
- kleine Änderungen
- - breaking_changes
- API Änderungen
- - deprecated_features
- Veraltete Features
- - removed_features
- Entfernte Features
- - security_fixes
- Sicherheitsfixes
- - bugfixes
- Bugfixes
- - known_issues
- Bekannte ungefixte Fehler
title: sebastian.systemd
trivial_section_name: trivial
use_fqcn: true

Datei anzeigen

@ -1,30 +1,24 @@
---
namespace: sebastian
name: systemd
version: 0.4.3
version: 0.1.0
# The path to the Markdown (.md) readme file. This path is relative to the root of the collection
readme: README.md
# A list of the collection's content authors. Can be just the name or in the format 'Full Name <email> (url)
# @nicks:irc/im.site#channel'
authors:
- Sebastian Tobie
description: An simple for generating systemd units with ansible
license_file: "LICENSE"
license_file: 'LICENSE'
tags:
- systemd
- linux
dependencies:
sebastian.base: ">=0.4.3"
repository: https://gitea.sebastian-tobie.de/ansible/ansible-systemd.git
dependencies: {}
repository: https://gitea.sebastian-tobie.de/sebastian/ansible-systemd
# documentation: http://docs.example.com
homepage: https://gitea.sebastian-tobie.de/ansible/ansible-systemd
issues: https://gitea.sebastian-tobie.de/ansible/ansible-systemd/issues
build_ignore:
- "*.gz"
- ".*"
- Makefile
- pyproject.toml
- upload.sh
- htmlcov
- changelogs
- docs
# homepage:
issues: https://gitea.sebastian-tobie.de/sebastian/ansible-systemd/issues
build_ignore: []
# manifest: null

Datei anzeigen

@ -1,4 +1,52 @@
---
# Collections must specify a minimum required ansible version to upload
# to galaxy
requires_ansible: '>=2.9.10'
# requires_ansible: '>=2.9.10'
# Content that Ansible needs to load from another location or that has
# been deprecated/removed
# plugin_routing:
# action:
# redirected_plugin_name:
# redirect: ns.col.new_location
# deprecated_plugin_name:
# deprecation:
# removal_version: "4.0.0"
# warning_text: |
# See the porting guide on how to update your playbook to
# use ns.col.another_plugin instead.
# removed_plugin_name:
# tombstone:
# removal_version: "2.0.0"
# warning_text: |
# See the porting guide on how to update your playbook to
# use ns.col.another_plugin instead.
# become:
# cache:
# callback:
# cliconf:
# connection:
# doc_fragments:
# filter:
# httpapi:
# inventory:
# lookup:
# module_utils:
# modules:
# netconf:
# shell:
# strategy:
# terminal:
# test:
# vars:
# Python import statements that Ansible needs to load from another location
# import_redirection:
# ansible_collections.ns.col.plugins.module_utils.old_location:
# redirect: ansible_collections.ns.col.plugins.module_utils.new_location
# Groups of actions/modules that take a common set of options
# action_groups:
# group_name:
# - module1
# - module2

0
plugins/__init__.py Normale Datei
Datei anzeigen

Datei anzeigen

Datei anzeigen

@ -0,0 +1,121 @@
import pathlib
from functools import partial
from typing import Any, Callable, Dict, Optional, Sequence, Tuple, Type, Union
__all__ = (
"Types",
"SYSTEMD_SERVICE_CONFIG",
"SYSTEMD_NETWORK_CONFIG",
"SYSTEMD_CONFIG_ROOT",
)
SYSTEMD_CONFIG_ROOT = pathlib.Path("/etc/systemd")
SYSTEMD_NETWORK_CONFIG = SYSTEMD_CONFIG_ROOT / "network"
SYSTEMD_SERVICE_CONFIG = SYSTEMD_CONFIG_ROOT / "system"
class _sdict(dict):
_help: Optional[str]
__name__: str
class _Type(type):
def __new__(metacls, cls, bases, classdict, **kwds):
individual = dict()
virtfunc = None
virtual = ()
special = dict()
for key, value in classdict.items():
if key.startswith("_"):
if key == "__getattr__":
virtfunc = value
elif key == "__dir__":
virtual = tuple(value(None))
elif key in ("__doc__",):
special[key] = value
else:
individual[key] = value
if len(virtual) != 0 and virtfunc is None: # pragma: nocover
raise TypeError(
"Virtual funcs defined, but no func to generate them defined"
)
special["_attr"] = tuple(virtual + tuple(individual.keys()))
special["_vfunc"] = virtfunc
special["_virtual"] = virtual
special["_individual"] = individual
annotations = dict()
if len(virtual) != 0 and virtfunc is not None: # pragma: nocover
anno = virtfunc(None, virtual[0]).__annotations__
for virtualkey in virtual:
annotations[virtualkey] = Callable[[*anno.values()], Dict[str, Any]]
annotations["__dir__"] = Callable[[], Tuple[str]]
special["__annotations__"] = annotations
inst = super().__new__(metacls, cls, bases, special, **kwds)
return inst
def __getattribute__(self, __name: str) -> Any:
if __name in (
"__dict__",
"__doc__",
"_attr",
"__annotations__",
"_virtual",
"_vfunc",
"_individual",
):
return super().__getattribute__(__name)
if __name in self._virtual:
return self._vfunc(self, __name)
if __name in self._individual:
return partial(self._individual[__name], self)
raise AttributeError(f"Attribut {__name} not found.")
class Types(metaclass=_Type):
"""Provides helpers for the ansible types"""
def list(
self,
elements: Union[Type[object], str],
required: bool = False,
help: Optional[str] = None,
) -> dict:
if not isinstance(elements, str):
elements = elements.__name__
option = _sdict(type="list", elements=elements, required=required)
option._help = help
return option
def __dir__(self) -> tuple:
return (
"str",
"dict",
"bool",
"int",
"float",
"path",
"raw",
"jsonarg",
"json",
"bytes",
"bits",
)
def __getattr__(self, name: str):
def argument(
required: bool = False,
help: Optional[str] = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
):
output = _sdict(type=name, required=required)
if choices is not None:
output["choices"] = choices
if default is not None:
output["default"] = default
output._help = help
return output
argument.__name__ = name
return argument

240
plugins/module_utils/module.py Normale Datei
Datei anzeigen

@ -0,0 +1,240 @@
import pathlib
from typing import Any, Callable, ClassVar, Dict, Optional, TypeVar, NoReturn, overload
import ansible.module_utils.basic as basic
from ansible.module_utils.generic import _sdict
__all__ = (
"AnsibleModule",
"SystemdUnitModule",
)
T = TypeVar("T")
class AnsibleModule(object):
"""Simple wrapper for the basic.AnsibleModule"""
#: name of the module. This is required for the generation of the Ansible documentation
name: ClassVar[str]
#: The AnsibleModule for this Module
module: basic.AnsibleModule
#: The result of this Module call. It always contains the changed key, so in any case an Module can report if it changed anything
result: dict
#: the specification of the arguments. Subclasses that are usable Modules must set this value.
module_spec: ClassVar[dict]
#: This is set by classes that define common things for their subclasses, like behaviour of the run and check methods. This is used by `SystemdUnitModule`
_common_args = dict()
@property
def params(self) -> Dict[str, Any]:
"""params is an wrapper for the module.params"""
return self.module.params # type: ignore
def __init__(self):
self.result = dict(changed=False)
specs = dict()
specs.update(self._common_args)
specs.update(self.module_spec)
self.module = basic.AnsibleModule(**specs)
self.tmpdir = pathlib.Path(self.module.tmpdir)
def set(self, key: str, value):
"""sets an value for the result"""
self.result[key] = value
@overload
def diff(self, diff: Dict[str, str]):
pass
def diff(
self,diff: Optional[Dict[str, str]] = None,*,
before: Optional[str] = None,
after: Optional[str] = None,
before_header: Optional[str] = None,
after_header: Optional[str] = None,
):
"""adds the special return value "diff". This allows Modules to present the changes of files to the caller. it takes care of the special semantics of the return value"""
if "diff" not in self.result:
self.result["diff"] = list()
if diff is not None and not any((before is not None, after is not None)):
pass
elif all((before is not None, after is not None, diff is None)):
diff = dict(
before=before,
after=after,
)
if before_header is not None:
diff["before_header"] = before_header
if after_header is not None:
diff["after_header"] = after_header
else:
raise TypeError("only diff or before and after can be set, not both of them")
self.result["diff"].append(diff)
def get(self, key: str, default: T = None) -> T:
"""returns an Parameter of the Module."""
if key not in self.params.keys():
return default
if self.params[key] is None and default is not None:
return default
if self.params[key] is None or key not in self.params:
raise KeyError()
return self.params[key]
@property
def changed(self):
"""returns if changes were detected/made"""
return self.result["changed"]
@changed.setter
def changed_set(self, value):
"""sets the changed value. this is always converted to bool"""
self.result["changed"] = not not value
def prepare(self):
raise NotImplementedError()
def check(self):
raise NotImplementedError()
def run(self):
raise NotImplementedError()
def __call__(self) -> NoReturn:
"""This calls the module. first prepare is called and then check or run, depending on the check mode.
If an exception is raised this is catched and the module automatically fails with an traceback"""
self.prepare()
try:
if self.module.check_mode:
self.check()
else:
self.run()
except Exception as exc:
import traceback
self.module.fail_json(
"".join(traceback.format_exception(type(exc), exc, exc.__traceback__)),
**self.result,
)
self.module.exit_json(**self.result)
@classmethod
def doc(cls) -> str:
"""this returns the documentation string of the module. If the help arguments of an Types method was given, it adds this as an helptext of this parameter"""
try:
import yaml
except ImportError:
return "---\n"
doc = cls.__doc__
if doc is None:
doc = ""
options = dict()
help: _sdict
for option, help in cls.module_spec["argument_spec"].items():
options[option] = dict(
type=help["type"],
)
if hasattr(help, "_help") and help._help is not None:
options[option]["description"] = help._help.split("\n")
if "required" in help and help["required"]:
options[option]["required"] = True
else:
options[option]["required"] = False
if help["type"] == "list":
options[option]["elements"] = help["elements"]
if not options[option]["required"]:
options[option]["default"] = []
if "default" in help:
options[option]["default"] = help["default"]
if "choices" in help:
options[option]["choices"] = tuple(help["choices"])
docu = doc.split("\n")
return str(
yaml.safe_dump(
dict(
module=cls.name,
short_description=docu[0],
description=docu,
options=options,
),
stream=None,
explicit_start="---",
)
)
class SystemdUnitModule(AnsibleModule):
#: path of the unitfile managed by this module
unitfile: pathlib.Path
#: subclasses of this always support the file common args and the check mode
_common_args = dict(
supports_check_mode=True,
add_file_common_args=True,
)
#: if defined it will be called after run has changed the unitfile
post: Optional[Callable[[], None]]
def unit(self) -> str:
raise NotImplementedError()
def unitfile_gen(self):
path = self.tmpdir / "newunit"
with open(path, "w") as unit:
unit.write(self.unit())
self.module.set_owner_if_different(path.as_posix(), "root", False)
self.module.set_group_if_different(path.as_posix(), "root", False)
self.module.set_mode_if_different(path.as_posix(), "0644", False)
if self.unitfile.exists():
diff = dict()
self.changed = self.changed | self.module.set_owner_if_different(
self.unitfile.as_posix(),
"root",
self.result["changed"],
diff,
)
self.diff(diff)
diff = dict()
self.changed = self.changed | self.module.set_group_if_different(
self.unitfile.as_posix(),
"root",
self.result["changed"],
diff,
)
self.diff(diff)
diff = dict()
self.changed = self.changed | self.module.set_mode_if_different(
self.unitfile.as_posix(),
"0644",
self.result["changed"],
diff,
)
self.diff(diff)
def check(self):
self.unitfile_gen()
if not self.unitfile.exists():
self.diff(before="", after=self.unit(), before_header=self.unitfile.as_posix())
self.changed = True
else:
if self.module.sha256(self.unitfile.as_posix()) != self.module.sha256(
(self.tmpdir / "newunit").as_posix()
):
self.changed = True
self.diff(
before=self.unitfile.read_text(),
after=self.unit(),
before_header=self.unitfile.as_posix(),
)
def run(self):
self.check()
if not self.changed:
return
self.module.atomic_move(
src=(self.tmpdir / "newunit").as_posix(),
dest=self.unitfile.as_posix(),
)
if hasattr(self, "post") and self.post is not None:
self.post()

0
plugins/modules/__init__.py Normale Datei
Datei anzeigen

Datei anzeigen

@ -1,215 +0,0 @@
#!/usr/bin/python3
import pathlib
from typing import List, Optional
try:
from ansible_module.module_utils.generic import SYSTEMD_NETWORK_CONFIG, Types, modspec
from ansible_module.module_utils.module import SystemdUnitModule
except ImportError:
from ansible_collections.sebastian.base.plugins.module_utils.generic import SYSTEMD_NETWORK_CONFIG, Types, modspec
from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdUnitModule
class Module(SystemdUnitModule): # type: ignore
"""generates an systemd-networkd link"""
name = "link"
module_spec = modspec(
argument_spec=dict(
mac=Types.str(help="The Mac address of the device." "An ! before the value matches anything but this value."),
permanentmac=Types.str(
help="The Permanent Mac address advertised by the device. " "An ! before the value matches anything but this value."
),
path=Types.str(
help="A shell-style glob matching the persistent path, as exposed by the udev property ID_PATH. "
"An ! before the value matches anything but this value."
),
driver=Types.str(
help="A glob matching the driver currently bound to the device. " "An ! before the value matches anything but this value."
),
type=Types.str(
help="A glob matching the device type, as exposed by networkctl list. "
"An ! before the value matches anything but this value."
),
kind=Types.str(
help="a glob matching the device kind, as exposed by networkctl status INTERFACE or ip -d link show INTERFACE. "
"An ! before the value matches anything but this value."
),
virtualization=Types.str(help="The virtualization type. An ! before the value matches anything but this value."),
description=Types.str(help="The description for the link"),
name=Types.str(required=True, help="The new name of the device"),
mtu=Types.int(help="The maximum Transmission unit for the link"),
),
required_one_of=(
("mac", "permanentmac", "path", "driver", "type", "kind"),
("name", "mac", "permanentmac"),
),
)
def prepare(self):
self.__unit = None
newname = self.get("name", "") or self.get("mac", "") or self.get("permanentmac", "")
newname = newname.replace(":", "").replace("/", "-").lower()
self.unitfile = SYSTEMD_NETWORK_CONFIG.joinpath("50-" + newname).with_suffix(".link")
def unit(self) -> str:
if self.__unit is None:
self.__unit = self._unit(self.match(), self.link())
return self.__unit
def match(self) -> Optional[str]:
options = self.map_param(
mac="MACAddress",
permanentmac="PermanentAddress",
path="Path",
driver="Driver",
type="Type",
kind="Kind",
virtualization="Virtualization",
)
if len(options) == 0:
return None
return "[Match]\n" + "".join(options)
def link(self) -> Optional[str]:
options = []
if self.get("description", False):
options.append("Description={}\n".format(self.get("description", False)))
if self.get("name", False):
options.append("Name={}\n".format(self.get("name", False)))
if self.get("mtu", False):
options.append("MTUBytes={}\n".format(self.get("mtu", False)))
if len(options) == 0:
return None
return "[Link]\n" + "".join(options)
def post(self):
if not self.changed:
return
args = [
"/usr/bin/udevadm",
"trigger",
"-c",
"add",
]
if self.module.check_mode:
args.append("-n")
if self.get("mac", False):
args.append("--attr-match=address={}".format(self.get("mac")))
if self.get("path", False):
args.append(self.get("path"))
self.module.run_command(args, check_rc=True)
DOCUMENTATION = """---
description:
- generates an systemd-networkd link
module: link
options:
after:
default: []
description:
- list of units that this unit wants to be started after this unit
elements: str
required: false
type: list
before:
default: []
description:
- list of units that this unit needs to be started before this unit.
elements: str
required: false
type: list
description:
description:
- The description for the link
required: false
type: str
documentation:
default: []
description:
- Paths where documentation can be found
elements: str
required: false
type: list
driver:
description:
- A glob matching the driver currently bound to the device. An ! before the value
matches anything but this value.
required: false
type: str
kind:
description:
- a glob matching the device kind, as exposed by networkctl status INTERFACE or
ip -d link show INTERFACE. An ! before the value matches anything but this value.
required: false
type: str
mac:
description:
- The Mac address of the device.An ! before the value matches anything but this
value.
required: false
type: str
mtu:
description:
- The maximum Transmission unit for the link
required: false
type: int
name:
description:
- The new name of the device
required: true
type: str
partof:
default: []
description:
- list of units that this unit is part of.
- If the restart this unit does it too, but if this restarts it does not affect
the other units.
elements: str
required: false
type: list
path:
description:
- A shell-style glob matching the persistent path, as exposed by the udev property
ID_PATH. An ! before the value matches anything but this value.
required: false
type: str
permanentmac:
description:
- The Permanent Mac address advertised by the device. An ! before the value matches
anything but this value.
required: false
type: str
requires:
default: []
description:
- list of units that this unit requires. If it fails or can't be started this
unit fails. without before/after this is started at the same time
elements: str
required: false
type: list
type:
description:
- A glob matching the device type, as exposed by networkctl list. An ! before
the value matches anything but this value.
required: false
type: str
virtualization:
description:
- The virtualization type. An ! before the value matches anything but this value.
required: false
type: str
wants:
default: []
description:
- list of units that this unit wants. If it fails or can't be started it does
not affect this unit
elements: str
required: false
type: list
short_description: generates an systemd-networkd link
"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -1,175 +0,0 @@
#!/usr/bin/python3
import pathlib
from typing import List, Optional
try:
from ansible_module.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
from ansible_module.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
except ImportError:
from ansible_collections.sebastian.base.plugins.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
SYSTEMD_SERVICE_CONFIG = pathlib.Path("/etc/systemd/system")
OPTION_MAPPING = dict(
required_by="RequiredBy",
wanted_by="WantedBy",
)
@installable
class Module(SystemdUnitModule, SystemdReloadMixin): # type: ignore[misc]
"""Creates an systemd mount"""
name = "mount"
module_spec = modspec(
argument_spec=dict(
fs=Types.str(required=True, help="The filesystem that is used for the mount"),
where=Types.path(required=True, help="The Path where the filesystem is mounted to"),
what=Types.str(required=True, help="The device or an string that will be mounted"),
state=Types.str(
choices=("present", "absent"),
default="present",
help="the state the mount is",
),
options=Types.list(elements=str, help="The options for the mount"),
),
)
def prepare(self):
self.mountdir = pathlib.Path(self.params["where"])
self.unitfile = SYSTEMD_SERVICE_CONFIG.joinpath(self.mountdir.relative_to("/").as_posix().replace("/", "-")).with_suffix(".mount")
self.__unit = None
if self.get("description", False) is False:
self.params["description"] = "Mount for {}".format(self.mountdir.as_posix())
def unit(self) -> str:
if self.__unit is None:
self.__unit = self._unit(
self.header(),
self.mount(),
self.install(), # type: ignore[misc,call-arg]
)
return self.__unit
def header(self) -> str:
description = self.get("description", "Mount for {}".format(self.get("where")))
return f"[Unit]\nDescription={description}\n"
def mount(self) -> str:
options = []
options.append("Where={}\n".format(self.get("where")))
options.append("What={}\n".format(self.get("what")))
options.append("Type={}\n".format(self.get("fs")))
if self.get("options", False):
options.append("Options={}\n".format(",".join(self.get("options"))))
return "[Mount]\n" + "".join(options)
DOCUMENTATION = """---
description:
- Creates an systemd mount
module: mount
options:
after:
default: []
description:
- list of units that this unit wants to be started after this unit
elements: str
required: false
type: list
before:
default: []
description:
- list of units that this unit needs to be started before this unit.
elements: str
required: false
type: list
description:
description:
- An description for programs that access systemd
required: false
type: str
documentation:
default: []
description:
- Paths where documentation can be found
elements: str
required: false
type: list
fs:
description:
- The filesystem that is used for the mount
required: true
type: str
options:
default: []
description:
- The options for the mount
elements: str
required: false
type: list
partof:
default: []
description:
- list of units that this unit is part of.
- If the restart this unit does it too, but if this restarts it does not affect
the other units.
elements: str
required: false
type: list
required_by:
default: []
description:
- systemd units that require this mount
elements: str
required: false
type: list
requires:
default: []
description:
- list of units that this unit requires. If it fails or can't be started this
unit fails. without before/after this is started at the same time
elements: str
required: false
type: list
state:
choices:
- present
- absent
default: present
description:
- the state the mount is
required: false
type: str
wanted_by:
default: []
description:
- systemd units that want the mount, but not explicitly require it. Commonly used
for target if not service explicitly require it.
elements: str
required: false
type: list
wants:
default: []
description:
- list of units that this unit wants. If it fails or can't be started it does
not affect this unit
elements: str
required: false
type: list
what:
description:
- The device or an string that will be mounted
required: true
type: str
where:
description:
- The Path where the filesystem is mounted to
required: true
type: path
short_description: Creates an systemd mount
"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -1,210 +0,0 @@
#!/usr/bin/python3
import pathlib
from typing import List, Optional
try:
from ansible_module.module_utils.generic import SYSTEMD_NETWORK_CONFIG, Types, joindict, modspec, systemdbool
from ansible_module.module_utils.module import SystemdUnitModule
except ImportError:
from ansible_collections.sebastian.base.plugins.module_utils.generic import (
SYSTEMD_NETWORK_CONFIG,
Types,
joindict,
modspec,
systemdbool,
)
from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdUnitModule
kinds = (
"bond",
"bridge",
"dummy",
"wireguard",
"vlan",
)
matchspec = dict(
host=Types.str(help="hostname of the host that is matched against"),
virtualization=Types.str("Virtualization that is checked against"),
kernelcmd=Types.str(help="checks an kernel commandline argument"),
)
netdevspec = dict(
description=Types.str(help="Description of the device"),
name=Types.str(required=True, help="name of the device"),
kind=Types.str(required=True, help="type of the device", choices=kinds),
)
bridgespec = dict(
stp=Types.bool(help="enable the stp protocol"),
priority=Types.int(help="Priority of the bridge"),
)
bondspec = dict(
mode=Types.str(
help="bonding policy",
choices=("balance-rr", "active-backup", "balance-xor", "broadcast", "802.3ad", "balance-tlb", "balance-alb"),
),
minlinks=Types.int(help="Specifies the minimum number of links that must be active before asserting carrier."),
)
wireguardspec = dict(
privatekey=Types.str(help="private key of this side of the tunnel"),
privatekeyfile=Types.path(help="Path of the private key on the host."),
port=Types.raw(help="Port that wireguard uses to listen for packets. the value 'auto' means that the port is automatically decided."),
)
class Module(SystemdUnitModule):
"""Creates an netdev unit that creates an virtual devices"""
name = "netdev"
module_spec = modspec(argument_spec=joindict(bondspec, bridgespec, matchspec, netdevspec))
_common_args = dict(
supports_check_mode=True,
add_file_common_args=True,
)
def prepare(self):
self.__unit = None
self.unitfile = SYSTEMD_NETWORK_CONFIG.joinpath(self.get("name")).with_suffix(".netdev")
def unit(self):
if self.__unit is None:
kind = self.get("kind")
parts = [self.match(), self.netdev()]
if kind != "dummy":
parts.append(getattr(self, kind)())
self.__unit = self._unit(*parts)
return self.__unit
def match(self) -> Optional[str]:
options = self.map_param(
host="Host",
kernelcmd="KernelCommandLine",
virtualization="Virtualization",
)
if len(options) == 0:
return None
return "[Match]\n" + "".join(options)
def netdev(self) -> Optional[str]:
options = self.map_param(
description="Description",
name="Name",
kind="Kind",
)
if len(options) == 0:
return None
return "[NetDev]\n" + "".join(options)
def bond(self) -> Optional[str]:
options = self.map_param(
mode="Mode",
minlinks="MinLinks",
)
if len(options) == 0:
return None
return "[Bond]\n" + "".join(options)
def bridge(self) -> Optional[str]:
options = self.map_param(
stp="STP",
priority="Priority",
)
if len(options) == 0:
return None
return "[Bridge]\n" + "".join(options)
def wireguard(self) -> Optional[str]:
options = self.map_param(
privatekey="PrivateKey",
privatekeyfile="PrivateKeyFile",
port="ListenPort",
)
port = self.get("port", False)
if port not in (False, "auto") and (port < 1 or port > 65535):
raise ValueError("Port must be between 0 and 65536")
if len(options) == 0:
return None
return "[Wireguard]\n" + "".join(options)
def vlan(self) -> Optional[str]:
options: list[str] = []
if len(options) == 0:
return None
return "[Vlan]\n" + "".join(options)
DOCUMENTATION = """---
description:
- Creates an netdev unit that creates an virtual devices
module: netdev
options:
description:
description:
- Description of the device
required: false
type: str
host:
description:
- hostname of the host that is matched against
required: false
type: str
kernelcmd:
description:
- checks an kernel commandline argument
required: false
type: str
kind:
choices:
- bond
- bridge
- dummy
- wireguard
- vlan
description:
- type of the device
required: true
type: str
minlinks:
description:
- Specifies the minimum number of links that must be active before asserting carrier.
required: false
type: int
mode:
choices:
- balance-rr
- active-backup
- balance-xor
- broadcast
- 802.3ad
- balance-tlb
- balance-alb
description:
- bonding policy
required: false
type: str
name:
description:
- name of the device
required: true
type: str
priority:
description:
- Priority of the bridge
required: false
type: int
stp:
description:
- enable the stp protocol
required: false
type: bool
virtualization:
required: true
type: str
short_description: Creates an netdev unit that creates an virtual devices
"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -1,250 +0,0 @@
#!/usr/bin/python3
import pathlib
from typing import List, Optional, Union
try:
from ansible_module.module_utils.generic import SYSTEMD_NETWORK_CONFIG, Types, modspec, systemdbool
from ansible_module.module_utils.module import SystemdUnitModule
except ImportError:
from ansible_collections.sebastian.base.plugins.module_utils.generic import SYSTEMD_NETWORK_CONFIG, Types, modspec, systemdbool
from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdUnitModule
class Module(SystemdUnitModule): # type: ignore
"""Sets up the systemd network unit"""
name = "network"
module_spec = modspec(
argument_spec=dict(
mac=Types.str(help="The MAC-Address of the device. An ! before the value matches anything but this value."),
device=Types.str(help="The name of the network device. An ! before the value matches anything but this value."),
virtualization=Types.str(help="The virtualization type. An ! before the value matches anything but this value."),
name=Types.str(required=True, help="name of the unit"),
dot=Types.bool(help="if DNS-over-TLS should be required or disabled. If it is unset, it will used if the server supports it"),
dnssec=Types.bool(
help="if the Domainqueries should require DNSSEC or not.\nIf its missing, domains that have DNSSEC enabled will be validated, all others it will be assumed to be okay."
),
dns=Types.list(elements=str, help="List of DNS-Servers"),
domain=Types.list(elements=str, help="List of domains that are on this device"),
defaultdns=Types.bool(
help="If the DNS-Server(s) on this device should be used for all domains that are not set on other devices"
),
address=Types.list(elements=str, required=True, help="IP-Addresses of this networkdevice"),
route=Types.list(
elements=str,
help="Routes of networks that can be reached with this device",
),
masquerade=Types.str(
help="how the packets are modified to look like the come from the computer itself.",
choices=("true", "false", "both", "ipv4", "ipv6", "no"),
),
),
required_if=(("defaultdns", True, ("dns",), False),),
required_one_of=(("mac", "device", "virtualization"),),
)
def prepare(self):
self.unitfile = SYSTEMD_NETWORK_CONFIG.joinpath(self.get("name")).with_suffix(".network")
self.__unit = None
def unit(self) -> str:
if self.__unit is None:
self.__unit = self._unit(
self.match(),
self.network(),
self.addresses(),
self.routes(),
)
return self.__unit
def match(self) -> Optional[str]:
matches = self.map_param(
mac="MACAddress",
device="Name",
virtualization="Virtualization",
)
if len(matches) == 0:
return None
return "[Match]\n" + "".join(matches)
def network(self) -> Optional[str]:
options = []
if self.get("description", None) is None:
options.append("Description={}".format(self.get("description")))
server: str
for server in self.get("dns", []):
options.append(f"DNS={server}")
options.append("DNSDefaultRoute={}".format(self.get("defaultdns", False)))
if self.get("domain", False):
options.append("Domains={}".format(" ".join(self.get("domain"))))
options.append("DNSOverTLS={}".format(systemdbool(self.get("dot", "opportunistic"))))
options.append("DNSSEC={}".format(systemdbool(self.get("dnssec", "allow-downgrade"))))
if self.get("masquerade", None) is not None:
masquerade: str = self.get("masquerade")
if masquerade == "true":
masquerade = "both"
elif masquerade == "false":
masquerade = "no"
options.append(f"IPMasquerade={masquerade}")
if len(options) == 0:
return None
return "[Network]\n" + "".join(options)
def addresses(self) -> str:
output = []
for address in self.get("address"):
output.append(f"[Address]\nAddress={address}\n")
return "\n".join(output)
def routes(self) -> Optional[str]:
output = []
routes: list[str] = self.get("route", [])
for gw in routes:
output.append(f"[Route]\nGateway={gw}\nGatewayOnLink=yes\nQuickAck=yes\n")
if len(output) == 0:
return None
return "\n".join(output)
DOCUMENTATION = """---
description:
- Sets up the systemd network unit
module: network
options:
address:
description:
- IP-Addresses of this networkdevice
elements: str
required: true
type: list
after:
default: []
description:
- list of units that this unit wants to be started after this unit
elements: str
required: false
type: list
before:
default: []
description:
- list of units that this unit needs to be started before this unit.
elements: str
required: false
type: list
defaultdns:
description:
- If the DNS-Server(s) on this device should be used for all domains that are
not set on other devices
required: false
type: bool
description:
description:
- An description for programs that access systemd
required: false
type: str
device:
description:
- The name of the network device. An ! before the value matches anything but this
value.
required: false
type: str
dns:
default: []
description:
- List of DNS-Servers
elements: str
required: false
type: list
dnssec:
description:
- if the Domainqueries should require DNSSEC or not.
- If its missing, domains that have DNSSEC enabled will be validated, all others
it will be assumed to be okay.
required: false
type: bool
documentation:
default: []
description:
- Paths where documentation can be found
elements: str
required: false
type: list
domain:
default: []
description:
- List of domains that are on this device
elements: str
required: false
type: list
dot:
description:
- if DNS-over-TLS should be required or disabled. If it is unset, it will used
if the server supports it
required: false
type: bool
mac:
description:
- The MAC-Address of the device. An ! before the value matches anything but this
value.
required: false
type: str
masquerade:
choices:
- 'true'
- 'false'
- both
- ipv4
- ipv6
- 'no'
description:
- how the packets are modified to look like the come from the computer itself.
required: false
type: str
name:
description:
- name of the unit
required: true
type: str
partof:
default: []
description:
- list of units that this unit is part of.
- If the restart this unit does it too, but if this restarts it does not affect
the other units.
elements: str
required: false
type: list
requires:
default: []
description:
- list of units that this unit requires. If it fails or can't be started this
unit fails. without before/after this is started at the same time
elements: str
required: false
type: list
route:
default: []
description:
- Routes of networks that can be reached with this device
elements: str
required: false
type: list
virtualization:
description:
- The virtualization type. An ! before the value matches anything but this value.
required: false
type: str
wants:
default: []
description:
- list of units that this unit wants. If it fails or can't be started it does
not affect this unit
elements: str
required: false
type: list
short_description: Sets up the systemd network unit
"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -1,197 +0,0 @@
#!/usr/bin/python3
import pathlib
from typing import List, Optional, Union
try:
from ansible_module.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
from ansible_module.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
except ImportError:
from ansible_collections.sebastian.base.plugins.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
@installable
class Module(SystemdUnitModule, SystemdReloadMixin): # type: ignore
"""Creates socket units."""
name = "socket"
module_spec = modspec(
argument_spec=dict(
name=Types.str(required=True, help="Name of the socket"),
stream=Types.list(
Types.str(
help="Name of the stream socket. The name can be a path, an portnumber or an ip with an port. addresses in square brackets are always ipv6 addresses"
)
),
datagram=Types.list(
Types.str(
help="Name of the datagram socket. The name can be a path, an portnumber or an ip with an port. addresses in square brackets are always ipv6 addresses"
)
),
sequential=Types.list(
Types.str(
help="Name of the sequential socket. The name can be a path, an portnumber or an ip with an port. addresses in square brackets are always ipv6 addresses"
)
),
fifo=Types.list(Types.path(help="Name of the fifo. The name must be an absolute path")),
socketuser=Types.str(help="User that owns the socket/fifo"),
socketgroup=Types.str(help="Group that owns the socket/fifo"),
socketmode=Types.str(help="mode of the socket in octal notation", default="0666"),
service=Types.list(str, help="Name of the services that use this socket"),
),
required_one_of=(("stream", "datagram", "sequential", "fifo"),),
)
restartable = False
def prepare(self):
self.unitfile = (SYSTEMD_SERVICE_CONFIG / self.get("name")).with_suffix(".socket")
self.__unit = None
def socket(self) -> Optional[str]:
section = self.map_param(
stream="ListenStream",
datagram="ListenDatagram",
sequential="ListenSequential",
fifo="ListenFIFO",
socketuser="SocketUser",
socketgroup="SocketGroup",
socketmode="SocketMode",
)
if len(section) == 0:
return None
return "[Socket]\n" + "".join(section)
def unit(self) -> str:
if self.__unit is None:
self.__unit = self._unit(
self.header(),
self.socket(),
self.install(), # type: ignore[call-arg,misc]
)
return self.__unit
DOCUMENTATION = """---
description:
- Creates socket units.
module: socket
options:
after:
default: []
description:
- list of units that this unit wants to be started after this unit
elements: str
required: false
type: list
before:
default: []
description:
- list of units that this unit needs to be started before this unit.
elements: str
required: false
type: list
datagram:
default: []
elements: str
required: false
type: list
description:
description:
- An description for programs that access systemd
required: false
type: str
documentation:
default: []
description:
- Paths where documentation can be found
elements: str
required: false
type: list
fifo:
default: []
elements: path
required: false
type: list
name:
description:
- Name of the socket
required: true
type: str
partof:
default: []
description:
- list of units that this unit is part of.
- If the restart this unit does it too, but if this restarts it does not affect
the other units.
elements: str
required: false
type: list
required_by:
default: []
description:
- systemd units that require this mount
elements: str
required: false
type: list
requires:
default: []
description:
- list of units that this unit requires. If it fails or can't be started this
unit fails. without before/after this is started at the same time
elements: str
required: false
type: list
sequential:
default: []
elements: str
required: false
type: list
service:
default: []
description:
- Name of the services that use this socket
elements: str
required: false
type: list
socketgroup:
description:
- Group that owns the socket/fifo
required: false
type: str
socketmode:
default: '0666'
description:
- mode of the socket in octal notation
required: false
type: str
socketuser:
description:
- User that owns the socket/fifo
required: false
type: str
stream:
default: []
elements: str
required: false
type: list
wanted_by:
default: []
description:
- systemd units that want the mount, but not explicitly require it. Commonly used
for target if not service explicitly require it.
elements: str
required: false
type: list
wants:
default: []
description:
- list of units that this unit wants. If it fails or can't be started it does
not affect this unit
elements: str
required: false
type: list
short_description: Creates socket units.
"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -1,410 +0,0 @@
#!/usr/bin/python3
import pathlib
from typing import List, Union
try:
from ansible_module.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types
from ansible_module.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
except ImportError:
from ansible_collections.sebastian.base.plugins.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types
from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
@installable
class Module(SystemdUnitModule, SystemdReloadMixin): # type: ignore[misc]
"""Creates System Services units"""
name = "system_service"
module_spec = dict(
argument_spec=dict(
name=Types.str(required=True, help="Name of the service"),
serviceuser=Types.str(help="Username of under which the commands run at.", default="root"),
servicegroup=Types.str(help="Group of under which the commands run at.", default="root"),
type=Types.str(
choices=("simple", "exec", "forking", "oneshot", "dbus", "notify", "notify-reload", "idle"),
default="simple",
help="Type of the systemd service.\n"
"simple and exec start long running services that run in the same process over the whole time, exec is waiting until the process was started completly.\n"
"forking does some things in the foreground, starts an background process and then exits to leave the work to the background process.\n"
"oneshot processes are started by systemd, do their work and then exit, similar to cronjobs.\n"
"dbus services will be considered started up once they aquire the specified dbus bus"
"notify and notify-reload notify systemd about the start up via sd_notify. notify-reload needs also inform systemd on reloads and when it is ready again after an reload.\n"
"idle is similar to simple, but it can delay the start up by a few seconds.",
),
pre=Types.list(elements=str, help="command or list of commands that are started before the main command"),
start=Types.list(
elements=str,
required=True,
help="command or list of commands that are started as main programm. Multiple commands are only allowed in a oneshot command",
),
stop=Types.str(help="command that is started to stop the main program."),
remain=Types.bool(help="should the service remain as started after the command exited"),
post=Types.list(str, help="Command or list of commands that are started after the main command(s) stopped without problems."),
environmentfile=Types.list(
elements=str,
help="List of file that are containing environment variables. They are evaluated before each pre/start/post command",
),
environment=Types.list(
elements=Types.dict(
name=Types.str(help="name of the Environment variable", required=True),
value=Types.str(help="value of the Environment variable", required=True),
),
help="List of environment variables that are set to each command before they run",
),
workingdirectory=Types.str(
help="The Directory that is used for the processes as current working directory",
),
rwpath=Types.list(
elements=Types.path(),
help="Path(s) that are readable and writable (if permission allow)",
),
ropath=Types.list(
elements=Types.path(),
help="Path(s) that are read only",
),
notreadablepath=Types.list(
elements=Types.path(),
help="Path(s) that are not accessible by the applications",
),
execpath=Types.list(
elements=Types.path(),
help="Path(s) where executable files are",
),
noexecpath=Types.list(
elements=Types.path(),
help="Path(s) which are never executable (uploaded files, user accessible paths)",
),
protecthome=Types.str(
help="if true makes user specific directories (/home, /root, /run/user) inaccessible. read-only makes them read only and tmpfs is useful to create binds in it",
choices=("true", "false", "read-only", "tmpfs"),
),
protectsystem=Types.str(
help="makes the system read only. if true /usr, /boot and /efi are read only, if full additionally /etc and if strict all except /proc, /sys and /dev",
choices=("true", "false", "full", "strict"),
),
nonewprivileges=Types.bool(
help="disables the ability to get new capabilities for processes than already granted ones",
),
statedirectory=Types.str(
help="creates an unit specific state directory in /var/lib and sets the env var STATE_DIRECTORY with the path to it. Its cleaned up after the unit is stopped"
),
runtimedirectory=Types.str(
help="creates an unit specific runtime directory in /run and sets the env var RUNTIME_DIRECTORY with the path to it. Its cleaned up after the unit is stopped"
),
restart=Types.str(),
restartsec=Types.str(),
),
)
def prepare(self):
self.unitfile = (SYSTEMD_SERVICE_CONFIG / self.get("name")).with_suffix(".service")
self.__unit = None
if self.get("type", "simple") != "oneshot" and len(self.get("start")) > 1:
self.fail("only oneshot services are allowed to have multiple start commands")
def service(self) -> str:
params = []
if self.get("environment", False):
for env in self.get("environment"):
params.append(f"Environment={env['name']}={env['value']}\n")
params.extend(
self.map_param(
type="Type",
pre="ExecStartPre",
start="ExecStart",
stop="ExecStop",
post="ExecStartPost",
serviceuser="User",
servicegroup="Group",
workingdirectory="WorkingDirectory",
environmentfile="EnvironmentFile",
protecthome="ProtectHome",
protectsystem="ProtectSystem",
rwpath="ReadWritePaths",
ropath="ReadOnlyPaths",
notreadablepath="InaccessiblePaths",
execpath="ExecPaths",
noexecpath="NoExecPaths",
statedirectory="StateDirectory",
runtimedirectory="RuntimeDirectory",
nonewprivileges="NoNewPriviledges",
remain="RemainAfterExit",
restart="Restart",
restartsec="RestartSec",
)
)
return "[Service]\n" + "".join(params)
def unit(self) -> str:
if self.__unit is None:
self.__unit = self._unit(
self.header(),
self.service(),
self.install(), # type: ignore[call-arg,misc]
)
return self.__unit
DOCUMENTATION = """---
description:
- Creates System Services units
module: system_service
options:
after:
default: []
description:
- list of units that this unit wants to be started after this unit
elements: str
required: false
type: list
before:
default: []
description:
- list of units that this unit needs to be started before this unit.
elements: str
required: false
type: list
description:
description:
- An description for programs that access systemd
required: false
type: str
documentation:
default: []
description:
- Paths where documentation can be found
elements: str
required: false
type: list
environment:
default: []
description:
- List of environment variables that are set to each command before they run
elements: dict
options:
name:
description:
- name of the Environment variable
required: true
type: str
value:
description:
- value of the Environment variable
required: true
type: str
required: false
type: list
environmentfile:
default: []
description:
- List of file that are containing environment variables. They are evaluated before
each pre/start/post command
elements: str
required: false
type: list
execpath:
default: []
description:
- Path(s) where executable files are
elements: path
required: false
type: list
name:
description:
- Name of the service
required: true
type: str
noexecpath:
default: []
description:
- Path(s) which are never executable (uploaded files, user accessible paths)
elements: path
required: false
type: list
nonewprivileges:
description:
- disables the ability to get new capabilities for processes than already granted
ones
required: false
type: bool
notreadablepath:
default: []
description:
- Path(s) that are not accessible by the applications
elements: path
required: false
type: list
partof:
default: []
description:
- list of units that this unit is part of.
- If the restart this unit does it too, but if this restarts it does not affect
the other units.
elements: str
required: false
type: list
post:
default: []
description:
- Command or list of commands that are started after the main command(s) stopped
without problems.
elements: str
required: false
type: list
pre:
default: []
description:
- command or list of commands that are started before the main command
elements: str
required: false
type: list
protecthome:
choices:
- 'true'
- 'false'
- read-only
- tmpfs
description:
- if true makes user specific directories (/home, /root, /run/user) inaccessible.
read-only makes them read only and tmpfs is useful to create binds in it
required: false
type: str
protectsystem:
choices:
- 'true'
- 'false'
- full
- strict
description:
- makes the system read only. if true /usr, /boot and /efi are read only, if full
additionally /etc and if strict all except /proc, /sys and /dev
required: false
type: str
remain:
description:
- should the service remain as started after the command exited
required: false
type: bool
required_by:
default: []
description:
- systemd units that require this mount
elements: str
required: false
type: list
requires:
default: []
description:
- list of units that this unit requires. If it fails or can't be started this
unit fails. without before/after this is started at the same time
elements: str
required: false
type: list
restart:
required: false
type: str
restartsec:
required: false
type: str
ropath:
default: []
description:
- Path(s) that are read only
elements: path
required: false
type: list
runtimedirectory:
description:
- creates an unit specific runtime directory in /run and sets the env var RUNTIME_DIRECTORY
with the path to it. Its cleaned up after the unit is stopped
required: false
type: str
rwpath:
default: []
description:
- Path(s) that are readable and writable (if permission allow)
elements: path
required: false
type: list
servicegroup:
default: root
description:
- Group of under which the commands run at.
required: false
type: str
serviceuser:
default: root
description:
- Username of under which the commands run at.
required: false
type: str
start:
description:
- command or list of commands that are started as main programm. Multiple commands
are only allowed in a oneshot command
elements: str
required: true
type: list
statedirectory:
description:
- creates an unit specific state directory in /var/lib and sets the env var STATE_DIRECTORY
with the path to it. Its cleaned up after the unit is stopped
required: false
type: str
stop:
description:
- command that is started to stop the main program.
required: false
type: str
type:
choices:
- simple
- exec
- forking
- oneshot
- dbus
- notify
- notify-reload
- idle
default: simple
description:
- Type of the systemd service.
- simple and exec start long running services that run in the same process over
the whole time, exec is waiting until the process was started completly.
- forking does some things in the foreground, starts an background process and
then exits to leave the work to the background process.
- oneshot processes are started by systemd, do their work and then exit, similar
to cronjobs.
- dbus services will be considered started up once they aquire the specified dbus
busnotify and notify-reload notify systemd about the start up via sd_notify.
notify-reload needs also inform systemd on reloads and when it is ready again
after an reload.
- idle is similar to simple, but it can delay the start up by a few seconds.
required: false
type: str
wanted_by:
default: []
description:
- systemd units that want the mount, but not explicitly require it. Commonly used
for target if not service explicitly require it.
elements: str
required: false
type: list
wants:
default: []
description:
- list of units that this unit wants. If it fails or can't be started it does
not affect this unit
elements: str
required: false
type: list
workingdirectory:
description:
- The Directory that is used for the processes as current working directory
required: false
type: str
short_description: Creates System Services units
"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -0,0 +1,157 @@
#!/usr/bin/python3
import pathlib
from typing import List
from ansible.module_utils.generic import SYSTEMD_NETWORK_CONFIG as SYSTEMD_PATH
from ansible.module_utils.generic import Types
from ansible.module_utils.module import SystemdUnitModule
class Module(SystemdUnitModule):
"""generates an systemd-networkd link"""
name = "systemd_link"
module_spec = dict(
argument_spec=dict(
mac=Types.str(help="The Mac address of the device"),
permanentmac=Types.str(
help="The Permanent Mac address advertised by the device"
),
path=Types.str(
help="A shell-style glob matching the persistent path, as exposed by the udev property ID_PATH."
),
driver=Types.str(
help="A glob matching the driver currently bound to the device"
),
type=Types.str(
help="A glob matching the device type, as exposed by networkctl list"
),
kind=Types.str(
help="a glob matching the device kind, as exposed by networkctl status INTERFACE or ip -d link show INTERFACE."
),
description=Types.str("The description for the link"),
name=Types.str(required=True, help="The new name of the device"),
mtu=Types.int(help="The maximum Transmission unit for the link"),
),
required_one_of=(
("mac", "permanentmac", "path", "driver", "type", "kind"),
("name", "mac", "permanentmac"),
),
)
def prepare(self):
self.__unit = None
newname = (
self.get("name", "") or self.get("mac", "") or self.get("permanentmac", "")
)
newname = newname.replace(":", "").replace("/", "-").lower()
self.unitfile = SYSTEMD_PATH.joinpath("50-" + newname).with_suffix(".link")
def unit(self) -> str:
if self.__unit is None:
self.__unit = "\n".join((self.match(), self.link()))
return self.__unit
def match(self) -> str:
options = []
if self.get("mac", False):
options.append("MACAddress={}\n".format(self.get("mac", False)))
if self.get("permanentmac", False):
options.append(
"PermanentAddress={}\n".format(self.get("permanentmac", False))
)
if self.get("path", False):
options.append("Path={}\n".format(self.get("path", False)))
if self.get("driver", False):
options.append("Driver={}\n".format(self.get("driver", False)))
if self.get("type", False):
options.append("Type={}\n".format(self.get("type", False)))
if self.get("kind", False):
options.append("Kind={}\n".format(self.get("kind", False)))
return "[Match]\n" + "".join(options)
def link(self) -> str:
options = []
if self.get("description", False):
options.append("Description={}\n".format(self.get("description", False)))
if self.get("name", False):
options.append("Name={}\n".format(self.get("name", False)))
if self.get("mtu", False):
options.append("MTUBytes={}\n".format(self.get("mtu", False)))
return "[Link]\n" + "".join(options)
def post(self):
if not self.changed:
return
args = [
"/usr/bin/udevadm",
"trigger",
"-c",
"add",
]
if self.module.check_mode:
args.append("-n")
if self.get("mac", False):
args.append("--attr-match=address={}".format(self.get("mac")))
if self.get("path", False):
args.append(self.get("path"))
self.module.run_command(args, check_rc=True)
DOCUMENTATION = """---
description:
- generates an systemd-networkd link
module: systemd_link
options:
description:
description:
- The description for the link
required: false
type: str
driver:
description:
- A glob matching the driver currently bound to the device
required: false
type: str
kind:
description:
- a glob matching the device kind, as exposed by networkctl status INTERFACE or
ip -d link show INTERFACE.
required: false
type: str
mac:
description:
- The Mac address of the device
required: false
type: str
mtu:
description:
- The maximum Transmission unit for the link
required: false
type: int
name:
description:
- The new name of the device
required: false
type: str
path:
description:
- A shell-style glob matching the persistent path, as exposed by the udev property
ID_PATH.
required: false
type: str
permanentmac:
description:
- The Permanent Mac address advertised by the device
required: false
type: str
type:
description:
- A glob matching the device type, as exposed by networkctl list
required: false
type: str
short_description: generates an systemd-networkd link
"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -0,0 +1,165 @@
#!/usr/bin/python3
import pathlib
from typing import List, Optional
from ansible.module_utils.generic import SYSTEMD_SERVICE_CONFIG as SYSTEMD_PATH
from ansible.module_utils.generic import Types
from ansible.module_utils.module import SystemdUnitModule
SYSTEMD_PATH = pathlib.Path("/etc/systemd/system")
OPTION_MAPPING = dict(
required_by="RequiredBy",
wanted_by="WantedBy",
)
class Module(SystemdUnitModule):
"""Creates an systemd mount"""
name = "systemd_mount"
module_spec = dict(
argument_spec=dict(
fs=Types.str(
required=True, help="The filesystem that is used for the mount"
),
where=Types.path(
required=True, help="The Path where the filesystem is mounted to"
),
what=Types.str(
required=True, help="The device or an string that will be mounted"
),
state=Types.str(
choices=("present", "absent"),
default="present",
help="the state the mount is",
),
options=Types.list(elements=str, help="The options for the mount"),
description=Types.str(
help="An description for programs that access systemd"
),
required_by=Types.list(
elements=str, help="systemd units that require this mount"
),
wanted_by=Types.list(
elements=str,
help="systemd units that want the mount, but not explicitly require it. Commonly used for target if not service explicitly require it.",
),
),
)
def prepare(self):
self.mountdir = pathlib.Path(self.params["where"])
self.unitfile = SYSTEMD_PATH.joinpath(
self.mountdir.relative_to("/").as_posix().replace("/", "-")
).with_suffix(".mount")
self.__unit = None
def unit(self) -> str:
if self.__unit is None:
self.__unit = "\n".join(
(
self.header(),
self.mount(),
self.install(),
)
)
return self.__unit
def header(self) -> str:
return "[Unit]\nDescription={}\n".format(
self.get("description", "Mount for {}".format(self.get("where")))
)
def mount(self) -> str:
output = "[Mount]\n"
output += "Where={}\n".format(self.get("where"))
output += "What={}\n".format(self.get("what"))
output += "Type={}\n".format(self.get("fs"))
if self.get("options", False):
output += "Options={}\n".format(",".join(self.get("options")))
return output
def install(self) -> str:
output = "[Install]\n"
for argument, key in OPTION_MAPPING.items():
if self.get(argument, False):
for unit in self.get(argument):
output += "{}={}\n".format(key, unit)
return output
def post(self):
if not self.changed:
return
systemctl = self.module.get_bin_path("systemctl", required=True)
self.module.run_command([systemctl, "daemon-reload"], check_rc=True)
(rc, _, _) = self.module.run_command(
[systemctl, "is-enabled", self.unitfile.name], check_rc=False
)
if rc == 0:
self.module.run_command(
[systemctl, "restart", self.unitfile.name], check_rc=True
)
DOCUMENTATION = """---
description:
- Creates an systemd mount
module: systemd_mount
options:
description:
description:
- An description for programs that access systemd
required: false
type: str
fs:
description:
- The filesystem that is used for the mount
required: true
type: str
options:
default: []
description:
- The options for the mount
elements: str
required: false
type: list
required_by:
default: []
description:
- systemd units that require this mount
elements: str
required: false
type: list
state:
choices:
- present
- absent
default: present
description:
- the state the mount is
required: false
type: str
wanted_by:
default: []
description:
- systemd units that want the mount, but not explicitly require it. Commonly used
for target if not service explicitly require it.
elements: str
required: false
type: list
what:
description:
- The device or an string that will be mounted
required: true
type: str
where:
description:
- The Path where the filesystem is mounted to
required: true
type: path
short_description: Creates an systemd mount
"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -0,0 +1,157 @@
#!/usr/bin/python3
import pathlib
from typing import List, Union
from ansible.module_utils.generic import SYSTEMD_NETWORK_CONFIG as SYSTEMD_PATH
from ansible.module_utils.generic import Types
from ansible.module_utils.module import SystemdUnitModule
def boolconvert(b: Union[bool, str]) -> str:
if b is True:
return "yes"
elif b is False:
return "no"
return b
class Module(SystemdUnitModule):
"""Sets up the systemd network unit"""
name = "systemd_network"
module_spec = dict(
argument_spec=dict(
mac=Types.str(),
device=Types.str(),
name=Types.str(required=True),
description=Types.str(),
dot=Types.bool(),
dnssec=Types.bool(),
dns=Types.list(elements=str),
domain=Types.list(elements=str),
defaultdns=Types.bool(),
address=Types.list(elements=str, required=True),
route=Types.list(elements=str),
),
required_if=(("defaultdns", True, ("dns",), False),),
required_one_of=(("mac", "device"),),
)
def prepare(self):
self.unitfile = SYSTEMD_PATH.joinpath(self.get("name")).with_suffix(".network")
self.__unit = None
def unit(self) -> str:
if self.__unit is None:
self.__unit = "\n".join(
(
self.match(),
self.network(),
self.addresses(),
self.routes(),
)
)
return self.__unit
def match(self) -> str:
matches = []
if self.get("mac", False):
matches.append("MACAddress={}\n".format(self.get("mac")))
if self.get("device", False):
matches.append("Name={}\n".format(self.get("device")))
return "[Match]\n" + "".join(matches)
def network(self) -> str:
output = "[Network]\n"
options = []
try:
options.append("Description={}".format(self.get("description")))
except KeyError:
pass
try:
for server in self.get("dns", []):
options.append(f"DNS={server}")
options.append("DNSDefaultRoute={}".format(self.get("defaultdns", False)))
except KeyError:
pass
try:
domain = self.get("domain")
self.set("domainlog", str(domain))
options.append("Domains={}".format(" ".join(domain)))
options.append(
"DNSOverTLS={}".format(boolconvert(self.get("dot", "opportunistic")))
)
options.append(
"DNSSEC={}".format(boolconvert(self.get("dnssec", "allow-downgrade")))
)
except KeyError:
pass
output += "\n".join(options)
return output
def addresses(self) -> str:
output = []
for address in self.get("address"):
output.append(f"[Address]\nAddress={address}\n")
return "\n".join(output)
def routes(self) -> str:
output = []
routes = self.get("route", [])
self.set("routes", routes)
for gw in routes:
output.append(f"[Route]\nGateway={gw}\nGatewayOnLink=yes\nQuickAck=yes\n")
self.set("routes", output)
return "\n".join(output)
DOCUMENTATION = """---
description:
- Sets up the systemd network unit
module: systemd_network
options:
address:
elements: str
required: true
type: list
defaultdns:
required: false
type: bool
description:
required: false
type: str
dns:
default: []
elements: str
required: false
type: list
dnssec:
required: false
type: bool
domain:
default: []
elements: str
required: false
type: list
dot:
required: false
type: bool
name:
required: true
type: str
mac:
elements: str
required: true
type: list
route:
default: []
elements: str
required: false
type: list
short_description: Sets up the systemd network unit
"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -1,139 +0,0 @@
#!/usr/bin/python3
import pathlib
from typing import List, Union
try:
from ansible_module.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, systemdbool
from ansible_module.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
except ImportError:
from ansible_collections.sebastian.base.plugins.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, systemdbool
from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
__module_name__ = "TargetModule"
@installable
class TargetModule(SystemdUnitModule, SystemdReloadMixin): # type: ignore[misc]
"""Creates Target units"""
name = "target"
module_spec = dict(
argument_spec=dict(
description=Types.str(help="description of the target"),
name=Types.str(required=True, help="name of the target"),
allow_isolate=Types.bool(
default=False,
help="allows admins to restrict the system to only start units that are wanted by this unit and subsequent units",
),
),
)
restartable = False
def prepare(self):
self.unitfile = (SYSTEMD_SERVICE_CONFIG / self.get("name")).with_suffix(".target")
self.__unit = None
def header(self) -> str:
section = super().header()
if section is None:
section = "[Unit]\n"
section += "AllowIsolate={}\n".format(systemdbool(self.get("allow_isolate", False)))
return section
def unit(self) -> str:
if self.__unit is None:
self.__unit = self._unit(
self.header(),
self.install(), # type: ignore[call-arg,misc]
)
return self.__unit
DOCUMENTATION = """---
description:
- Creates Target units
module: target
options:
after:
default: []
description:
- list of units that this unit wants to be started after this unit
elements: str
required: false
type: list
allow_isolate:
default: false
description:
- allows admins to restrict the system to only start units that are wanted by
this unit and subsequent units
required: false
type: bool
before:
default: []
description:
- list of units that this unit needs to be started before this unit.
elements: str
required: false
type: list
description:
description:
- description of the target
required: false
type: str
documentation:
default: []
description:
- Paths where documentation can be found
elements: str
required: false
type: list
name:
description:
- name of the target
required: true
type: str
partof:
default: []
description:
- list of units that this unit is part of.
- If the restart this unit does it too, but if this restarts it does not affect
the other units.
elements: str
required: false
type: list
required_by:
default: []
description:
- systemd units that require this mount
elements: str
required: false
type: list
requires:
default: []
description:
- list of units that this unit requires. If it fails or can't be started this
unit fails. without before/after this is started at the same time
elements: str
required: false
type: list
wanted_by:
default: []
description:
- systemd units that want the mount, but not explicitly require it. Commonly used
for target if not service explicitly require it.
elements: str
required: false
type: list
wants:
default: []
description:
- list of units that this unit wants. If it fails or can't be started it does
not affect this unit
elements: str
required: false
type: list
short_description: Creates Target units
"""
if __name__ == "__main__":
TargetModule()()

Datei anzeigen

@ -1,252 +0,0 @@
#!/usr/bin/python3
import pathlib
from typing import List, Union
try:
from ansible_module.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
from ansible_module.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
except ImportError:
from ansible_collections.sebastian.base.plugins.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
__module_name__ = "TimerModule"
@installable
class TimerModule(SystemdUnitModule, SystemdReloadMixin): # type: ignore[misc]
"""Creates Timer units"""
name = "timer"
module_spec = modspec(
argument_spec=dict(
name=Types.str(required=True, help="Name of the unit"),
description=Types.str(help="Description of the timer"),
onactive=Types.list(
help="Starts the service x seconds after this timer was activated",
elements=Types.str(),
),
onboot=Types.list(
help="Starts the service x seconds after the device was booted or the container was started",
elements=Types.str(),
),
onstartup=Types.list(
help="Starts the service x seconds after the System's/User's systemd instance was started",
elements=Types.str(),
),
onunitactive=Types.list(
help="Starts the service x seconds after the unit this timer activates was last activated",
elements=Types.str(),
),
onunitinactive=Types.list(
help="Starts the service x seconds after the unit this timer activates was last deactivated",
elements=Types.str(),
),
oncalendar=Types.list(
help="Uses an Time string to start the unit.",
elements=Types.str(),
),
persistent=Types.bool(
help="If the system was down in the time the timer would have started the unit, start the unit as soon as possible."
),
randomdelay=Types.str(help="delays the activation by an random delay between 0 and the value"),
fixdelay=Types.bool(
help="set the random delay to an fixed value. It uses the timername, the user of the servicemanager and the machineid as the seed."
),
unit=Types.str(help="The name of the unit. only needed if its not {{name}}.service"),
),
required_one_of=[
(
"onactive",
"onboot",
"onstartup",
"onunitactive",
"onunitinactive",
"oncalendar",
),
],
)
def prepare(self):
self.unitfile = (SYSTEMD_SERVICE_CONFIG / self.get("name")).with_suffix(".timer")
self.__unit = None
def body(self):
section = "[Timer]\n"
params = []
params.extend(
self.map_param(
onactive="OnActiveSec",
onboot="OnBootSec",
onstartup="OnStartupSec",
onunitactive="OnUnitActiveSec",
onunitinactive="OnUnitInactiveSec",
oncalendar="OnCalendar",
persistent="Persistent",
randomdelay="RandomizedDelaySec",
fixdelay="FixedRandomDelay",
unit="Unit",
),
)
if len(params) == 0:
return None
section += "".join(params)
return section
def unit(self) -> str:
if self.__unit is None:
self.__unit = self._unit(
self.header(),
self.body(),
self.install(), # type: ignore[call-arg,misc]
)
return self.__unit
DOCUMENTATION = """---
description:
- Creates Timer units
module: timer
options:
after:
default: []
description:
- list of units that this unit wants to be started after this unit
elements: str
required: false
type: list
before:
default: []
description:
- list of units that this unit needs to be started before this unit.
elements: str
required: false
type: list
description:
description:
- Description of the timer
required: false
type: str
documentation:
default: []
description:
- Paths where documentation can be found
elements: str
required: false
type: list
fixdelay:
description:
- set the random delay to an fixed value. It uses the timername, the user of the
servicemanager and the machineid as the seed.
required: false
type: bool
name:
description:
- Name of the unit
required: true
type: str
onactive:
default: []
description:
- Starts the service x seconds after this timer was activated
elements: str
required: false
type: list
onboot:
default: []
description:
- Starts the service x seconds after the device was booted or the container was
started
elements: str
required: false
type: list
oncalendar:
default: []
description:
- Uses an Time string to start the unit.
elements: str
required: false
type: list
onstartup:
default: []
description:
- Starts the service x seconds after the System's/User's systemd instance was
started
elements: str
required: false
type: list
onunitactive:
default: []
description:
- Starts the service x seconds after the unit this timer activates was last activated
elements: str
required: false
type: list
onunitinactive:
default: []
description:
- Starts the service x seconds after the unit this timer activates was last deactivated
elements: str
required: false
type: list
partof:
default: []
description:
- list of units that this unit is part of.
- If the restart this unit does it too, but if this restarts it does not affect
the other units.
elements: str
required: false
type: list
persistent:
description:
- If the system was down in the time the timer would have started the unit, start
the unit as soon as possible.
required: false
type: bool
randomdelay:
description:
- delays the activation by an random delay between 0 and the value
required: false
type: str
required_by:
default: []
description:
- systemd units that require this mount
elements: str
required: false
type: list
requires:
default: []
description:
- list of units that this unit requires. If it fails or can't be started this
unit fails. without before/after this is started at the same time
elements: str
required: false
type: list
unit:
description:
- The name of the unit. only needed if its not {{name}}.service
required: false
type: str
wanted_by:
default: []
description:
- systemd units that want the mount, but not explicitly require it. Commonly used
for target if not service explicitly require it.
elements: str
required: false
type: list
wants:
default: []
description:
- list of units that this unit wants. If it fails or can't be started it does
not affect this unit
elements: str
required: false
type: list
short_description: Creates Timer units
"""
if __name__ == "__main__":
TimerModule()()

Datei anzeigen

@ -1,47 +0,0 @@
#!/usr/bin/python3
import pathlib
from typing import List, Union
try:
from ansible_module.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
from ansible_module.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
except ImportError:
from ansible_collections.sebastian.base.plugins.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
@installable
class Module(SystemdUnitModule, SystemdReloadMixin): # type: ignore
"""Creates units"""
name = "unit"
module_spec = modspec(
argument_spec=dict(
name=Types.str(required=True, help="Name of the unit"),
)
)
def prepare(self):
self.unitfile = (SYSTEMD_SERVICE_CONFIG / self.get("name")).with_suffix(".")
self.__unit = None
def body(self):
section = "[]\n"
return section
def unit(self) -> str:
if self.__unit is None:
self.__unit = "\n".join(
(
self.header(),
self.body(),
self.install(),
)
)
return self.__unit
DOCUMENTATION = """"""
if __name__ == "__main__":
Module()()

Datei anzeigen

@ -1,10 +0,0 @@
[tool.black]
line-length = 140
[tool.isort]
atomic = true
profile = "black"
line_length = 140
[tool.mypy]
disable_error_code = ["import-untyped", "no-redef", "attr-defined"]

Datei anzeigen

@ -0,0 +1,40 @@
import os
import unittest
try: # pragma: nocover
from ansible_collections.sebastian.systemd.plugins.module_utils.generic import \
Types
except ImportError: # pragma: nocover
import sys
sys.path.append("plugins/module_utils")
from generic import Types
class TestTypes(unittest.TestCase):
"""tests the Types class"""
def testsimpletype(self):
"""this tests if an simple type is correctly build"""
output = Types.str(required=True, help="test", choices=("a", "1"), default="1")
self.assertIn("type", output)
self.assertIn("required", output)
self.assertIn("default", output)
self.assertIn("choices", output)
self.assertEquals(output["type"], "str")
self.assertEquals(Types.str.__name__, "str")
self.assertEquals(output["required"], True)
self.assertEquals(output["default"], "1")
self.assertTupleEqual(output["choices"], ("a", "1"))
Types.str()
def testlisttype(self):
"""this tests if the special type list is correctly build"""
output = Types.list(str)
Types.list("str")
self.assertIn("type", output)
self.assertIn("elements", output)
self.assertIn("required", output)
self.assertEquals(output["type"], "list")
self.assertEquals(output["required"], False)
self.assertEquals(output["elements"], "str")

Datei anzeigen

@ -1,6 +0,0 @@
#!/bin/bash
user=$(yq -r .namespace galaxy.yml)
package=$(yq -r .name galaxy.yml)
version=$(yq -r .version galaxy.yml)
printf "Namespace: %s\nPackage: %s\nVersion: %s\nfile name: %s\n" $user $package $version "$user-$package-$version.tar.gz"
ansible-galaxy collection publish dist/*