almalinux
/
lorax
14
0
Fork 0
Code Issues Pull Requests Releases Activity

lorax-composer: Add services support to blueprints 

Add support for enabling and disabling systemd services in the
blueprint. It works like this:

    [customizations.services]
    enabled = ["sshd", "cockpit.socket", "httpd"]
    disabled = ["postfix", "telnetd"]

They are *added* to any existing settings in the kickstart templates.
This commit is contained in:
Brian C. Lane 2019-04-16 16:40:40 -07:00
parent 6ea1c45734
commit 1111aee92d
3 changed files with 209 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
src/pylorax/api/compose.py
View File

@ -328,6 +328,76 @@ def get_firewall_settings(recipe):
return settings
def services_cmd(line, settings):
""" Update the services line with additional services to enable/disable
:param line: The services ... line
:type line: str
:param settings: A dict with the list of services to enable and disable
:type settings: dict
Using pykickstart to process the line is the best way to make sure it
is parsed correctly, and re-assembled for inclusion into the final kickstart
"""
# Empty services and no additional settings, return an empty string
if not line and not settings["enabled"] and not settings["disabled"]:
return ""
ks_version = makeVersion()
ks = KickstartParser(ks_version, errorsAreFatal=False, missingIncludeIsFatal=False)
# Allow passing in a 'default' so that the enable/disable may be applied to it, without
# parsing it and emitting a kickstart error message
if line != "services":
ks.readKickstartFromString(line)
# Add to any existing services, removing any duplicates
ks.handler.services.enabled = sorted(set(settings["enabled"] + ks.handler.services.enabled))
ks.handler.services.disabled = sorted(set(settings["disabled"] + ks.handler.services.disabled))
# Converting back to a string includes a comment, return just the keyboard line
return str(ks.handler.services).splitlines()[-1]
def get_services(recipe):
"""Return the customizations.services settings
:param recipe: The recipe
:type recipe: Recipe object
:returns: A dict of settings
:rtype: dict
"""
settings = {"enabled": [], "disabled": []}
if "customizations" not in recipe or \
"services" not in recipe["customizations"]:
return settings
settings["enabled"] = sorted(recipe["customizations"]["services"].get("enabled", []))
settings["disabled"] = sorted(recipe["customizations"]["services"].get("disabled", []))
return settings
def get_default_services(recipe):
"""Get the default string for services, based on recipe
:param recipe: The recipe
:type recipe: Recipe object
:returns: string with "services" or ""
:rtype: str
When no services have been selected we don't need to add anything to the kickstart
so return an empty string. Otherwise return "services" which will be updated with
the settings.
"""
services = get_services(recipe)
if services["enabled"] or services["disabled"]:
return "services"
else:
return ""
def customize_ks_template(ks_template, recipe):
""" Customize the kickstart template and return it
@ -364,6 +434,9 @@ def customize_ks_template(ks_template, recipe):
"firewall": [firewall_cmd,
get_firewall_settings(recipe),
'firewall --enabled', True],
"services": [services_cmd,
get_services(recipe),
get_default_services(recipe), True]
}
found = {}

123
tests/pylorax/test_compose.py
View File

@ -25,6 +25,7 @@ from pylorax.api.compose import add_customizations, get_extra_pkgs, compose_type
from pylorax.api.compose import timezone_cmd, get_timezone_settings
from pylorax.api.compose import lang_cmd, get_languages, keyboard_cmd, get_keyboard_layout
from pylorax.api.compose import firewall_cmd, get_firewall_settings
from pylorax.api.compose import services_cmd, get_services, get_default_services
from pylorax.api.compose import get_kernel_append, bootloader_append, customize_ks_template
from pylorax.api.config import configure, make_dnf_dirs
from pylorax.api.dnfbase import get_base_object
@ -449,6 +450,89 @@ disabled = ["telnet"]
"enabled": ["ftp", "ntp", "dhcp"], "disabled": ["telnet"]}),
"firewall --disabled")
def test_get_services(self):
"""Test get_services function"""
blueprint_data = """name = "test-services"
description = "test recipe"
version = "0.0.1"
[customizations.services]
"""
enable_services = """
enabled = ["sshd", "cockpit.socket", "httpd"]
"""
disable_services = """
disabled = ["postfix", "telnetd"]
"""
blueprint2_data = blueprint_data + enable_services
blueprint3_data = blueprint_data + disable_services
blueprint4_data = blueprint_data + enable_services + disable_services
recipe = recipe_from_toml(blueprint_data)
self.assertEqual(get_services(recipe), {'enabled': [], 'disabled': []})
recipe = recipe_from_toml(blueprint2_data)
self.assertEqual(get_services(recipe),
{"enabled": ["cockpit.socket", "httpd", "sshd"], "disabled": []})
recipe = recipe_from_toml(blueprint3_data)
self.assertEqual(get_services(recipe),
{"enabled": [], "disabled": ["postfix", "telnetd"]})
recipe = recipe_from_toml(blueprint4_data)
self.assertEqual(get_services(recipe),
{"enabled": ["cockpit.socket", "httpd", "sshd"], "disabled": ["postfix", "telnetd"]})
def test_services_cmd(self):
"""Test services_cmd function"""
self.assertEqual(services_cmd("", {"enabled": [], "disabled": []}), "")
self.assertEqual(services_cmd("", {"enabled": ["cockpit.socket", "httpd", "sshd"], "disabled": []}),
'services --enabled="cockpit.socket,httpd,sshd"')
self.assertEqual(services_cmd("", {"enabled": [], "disabled": ["postfix", "telnetd"]}),
'services --disabled="postfix,telnetd"')
self.assertEqual(services_cmd("", {"enabled": ["cockpit.socket", "httpd", "sshd"],
"disabled": ["postfix", "telnetd"]}),
'services --disabled="postfix,telnetd" --enabled="cockpit.socket,httpd,sshd"')
self.assertEqual(services_cmd("services --enabled=pop3", {"enabled": ["cockpit.socket", "httpd", "sshd"],
"disabled": ["postfix", "telnetd"]}),
'services --disabled="postfix,telnetd" --enabled="cockpit.socket,httpd,pop3,sshd"')
self.assertEqual(services_cmd("services --disabled=imapd", {"enabled": ["cockpit.socket", "httpd", "sshd"],
"disabled": ["postfix", "telnetd"]}),
'services --disabled="imapd,postfix,telnetd" --enabled="cockpit.socket,httpd,sshd"')
self.assertEqual(services_cmd("services --enabled=pop3 --disabled=imapd", {"enabled": ["cockpit.socket", "httpd", "sshd"],
"disabled": ["postfix", "telnetd"]}),
'services --disabled="imapd,postfix,telnetd" --enabled="cockpit.socket,httpd,pop3,sshd"')
def test_get_default_services(self):
"""Test get_default_services function"""
blueprint_data = """name = "test-services"
description = "test recipe"
version = "0.0.1"
[customizations.services]
"""
enable_services = """
enabled = ["sshd", "cockpit.socket", "httpd"]
"""
disable_services = """
disabled = ["postfix", "telnetd"]
"""
blueprint2_data = blueprint_data + enable_services
blueprint3_data = blueprint_data + disable_services
blueprint4_data = blueprint_data + enable_services + disable_services
recipe = recipe_from_toml(blueprint_data)
self.assertEqual(get_default_services(recipe), "")
recipe = recipe_from_toml(blueprint2_data)
self.assertEqual(get_default_services(recipe), "services")
recipe = recipe_from_toml(blueprint3_data)
self.assertEqual(get_default_services(recipe), "services")
recipe = recipe_from_toml(blueprint4_data)
self.assertEqual(get_default_services(recipe), "services")
def _checkBootloader(self, result, append_str, line_limit=0):
"""Find the bootloader line and make sure append_str is in it"""
# Optionally check to make sure the change is at the top of the template
@ -536,6 +620,27 @@ disabled = ["telnet"]
line_num += 1
return False
def _checkServices(self, result, settings, line_limit=0):
"""Find the services line and make sure it is as expected"""
# Optionally check to make sure the change is at the top of the template
line_num = 0
for line in result.splitlines():
if line.startswith("services"):
# First layout is used twice, so total count should be n+1
enabled = all([bool(e in line) for e in settings["enabled"]])
disabled = all([bool(d in line) for d in settings["disabled"]])
if enabled and disabled:
if line_limit == 0 or line_num < line_limit:
return True
else:
print("FAILED: services not in the first %d lines of the output" % line_limit)
return False
else:
print("FAILED: %s not matching %s" % (settings, line))
line_num += 1
return False
def test_template_defaults(self):
"""Test that customize_ks_template includes defaults correctly"""
blueprint_data = """name = "test-kernel"
@ -555,6 +660,7 @@ version = "*"
self.assertTrue(self._checkBootloader(result, "none", line_limit=2))
self.assertEqual(sum([1 for l in result.splitlines() if l.startswith("timezone")]), 1)
self.assertTrue(self._checkTimezone(result, {"timezone": "UTC", "ntpservers": []}, line_limit=2))
self.assertTrue("services" not in result)
# Make sure that a kickstart with a bootloader, and no timezone has timezone added to the top
result = customize_ks_template("firewall --enabled\nbootloader --location=mbr\n", recipe)
@ -563,6 +669,7 @@ version = "*"
self.assertTrue(self._checkBootloader(result, "mbr"))
self.assertEqual(sum([1 for l in result.splitlines() if l.startswith("timezone")]), 1)
self.assertTrue(self._checkTimezone(result, {"timezone": "UTC", "ntpservers": []}, line_limit=1))
self.assertTrue("services" not in result)
# Make sure that a kickstart with a bootloader and timezone has neither added
result = customize_ks_template("firewall --enabled\nbootloader --location=mbr\ntimezone US/Samoa\n", recipe)
@ -571,6 +678,7 @@ version = "*"
self.assertTrue(self._checkBootloader(result, "mbr"))
self.assertEqual(sum([1 for l in result.splitlines() if l.startswith("timezone")]), 1)
self.assertTrue(self._checkTimezone(result, {"timezone": "US/Samoa", "ntpservers": []}))
self.assertTrue("services" not in result)
def test_customize_ks_template(self):
"""Test that customize_ks_template works correctly"""
@ -595,6 +703,10 @@ ports = ["22:tcp", "80:tcp", "imap:tcp", "53:tcp", "53:udp"]
[customizations.firewall.services]
enabled = ["ftp", "ntp", "dhcp"]
disabled = ["telnet"]
[customizations.services]
enabled = ["sshd", "cockpit.socket", "httpd"]
disabled = ["postfix", "telnetd"]
"""
tz_dict = {"timezone": "US/Samoa", "ntpservers": ["0.north-america.pool.ntp.org", "1.north-america.pool.ntp.org"]}
recipe = recipe_from_toml(blueprint_data)
@ -613,6 +725,10 @@ disabled = ["telnet"]
{"ports": ["22:tcp", "80:tcp", "imap:tcp", "53:tcp", "53:udp"],
"enabled": ["ftp", "ntp", "dhcp"], "disabled": ["telnet"]}, line_limit=6))
self.assertEqual(sum([1 for l in result.splitlines() if l.startswith("firewall")]), 1)
self.assertTrue(self._checkServices(result,
{"enabled": ["cockpit.socket", "httpd", "sshd"], "disabled": ["postfix", "telnetd"]},
line_limit=8))
self.assertEqual(sum([1 for l in result.splitlines() if l.startswith("services")]), 1)
# Test against a kickstart with a bootloader line
result = customize_ks_template("firewall --enabled\nbootloader --location=mbr\n", recipe)
@ -665,6 +781,13 @@ disabled = ["telnet"]
if sum([1 for l in result.splitlines() if l.startswith("firewall")]) != 1:
errors.append(("firewall for compose_type %s failed: More than 1 entry" % compose_type, result))
if not self._checkServices(result,
{"enabled": ["cockpit.socket", "httpd", "sshd"],
"disabled": ["postfix", "telnetd"]}):
errors.append(("services for compose_type %s failed" % compose_type, result))
if sum([1 for l in result.splitlines() if l.startswith("services")]) != 1:
errors.append(("services for compose_type %s failed: More than 1 entry" % compose_type, result))
# Print the bad results
for e, r in errors:
print("%s:\n%s\n\n" % (e, r))

13
tests/pylorax/test_recipes.py
View File

@ -573,6 +573,19 @@ disabled = ["telnet"]
self.assertEqual(ks.handler.firewall.services, ["ftp", "ntp", "dhcp"])
self.assertEqual(ks.handler.firewall.remove_services, ["telnet"])
def test_services(self):
blueprint_data = """name = "test-services"
description = "test recipe"
version = "0.0.1"
[customizations.services]
enabled = ["sshd", "cockpit.socket", "httpd"]
disabled = ["postfix", "telnetd"]
"""
ks = self._blueprint_to_ks(blueprint_data)
self.assertEqual(sorted(ks.handler.services.enabled), ["cockpit.socket", "httpd", "sshd"])
self.assertEqual(sorted(ks.handler.services.disabled), ["postfix", "telnetd"])
def test_user(self):
blueprint_data = """name = "test-user"
description = "test recipe"