import abc
import logging
import string
from fortrace.core.virsh_domain import VirshDomain
from fortrace.utility.distribution_constants import ShellType
[docs]
class ServerSetup(abc.ABC):
"""Interface to handle setup of servers participating on the scenario."""
def __init__(self, server: VirshDomain, config: dict):
"""Construct a ServerSetup object to be called from Simulation Monitor.
To be called at the beginning of the scenario, before the Generator takes over and executes the actual
simulation.
Args:
server: handle to domain to be configured
config: the whole configuration of the domain
"""
self._server = server
self._config = config
self._pty = server.open_pty(
config["domain"]["username"],
config["domain"]["password"],
ShellType[config["domain"]["shell_type"].split(".")[1]],
20,
)
[docs]
def setup(self):
"""Method to be called from a SimulationMonitor.
Registers all users, sets up all services and stars them.
"""
self._create_users()
for service_name, config in (
self._config["domain"]["server"].get("services", {}).items()
):
# get callable method defined in service"s config, if service must be configured
if import_statement := config.get("installation", {}).get("import"):
exec(import_statement)
template = string.Template(
config["installation"]["construct"]["template"]
)
substitutions = config["installation"]["construct"].get(
"substitutions", {}
)
substitutions["pty"] = "self._pty"
substitutions["configuration"] = "self._config"
exec(template.substitute(substitutions))
if import_statement := config.get("operation", {}).get("import"):
exec(import_statement)
template = string.Template(config["operation"]["construct"]["template"])
substitutions = config["operation"]["construct"].get(
"substitutions", {}
)
substitutions["pty"] = "self._pty"
substitutions["configuration"] = "self._config"
exec(template.substitute(substitutions))
@abc.abstractmethod
def _create_users(self):
"""Register users listed in config file."""
pass
[docs]
class LinuxServerSetup(ServerSetup):
"""Handle Linux server setup."""
def _create_users(self):
for user_config in self._config["domain"]["server"].get("users", []):
# as prompt will change, we use virsh instead
self._pty.run_command(f"useradd -m {user_config['username']}", True)
self._pty.run_command(
f"passwd {user_config['username']}", True, new_prompt=": "
)
self._pty.run_command(user_config["password"], new_prompt=": ")
self._pty.restore_prompt()
out = self._pty.run_command(user_config["password"])
if out != "passwd: password updated successfully":
raise RuntimeError("Cannot change password at user creation")
logging.info("Successfully added user %s", user_config["username"])
[docs]
class WindowsServerSetup(ServerSetup):
"""Class to handle Windows Server setup."""
def _create_users(self):
raise NotImplementedError("Currently not implemented for Windows Servers")