153 lines
5.3 KiB
Python
153 lines
5.3 KiB
Python
"""
|
|
Library of classes modeling Ansible nodes or their types.
|
|
"""
|
|
|
|
from enum import Enum
|
|
from pathlib import Path, PurePath
|
|
from typing import TypedDict as Dict
|
|
from typing import Union, Literal, Required, Self
|
|
from custtypes import ExecutedPath, IdlePath, VirtualPrivateServers, AnsibleScopes
|
|
from softman import Software, SoftPathGroup, SoftScope, Apps, Softs
|
|
from whereami import USER_PATH, PROJ_ROOT
|
|
from ansible_vault import Vault
|
|
import secrets
|
|
|
|
class ControlNode:
|
|
__user_path: ExecutedPath = USER_PATH
|
|
__proj_root: ExecutedPath = PROJ_ROOT
|
|
__conf_paths: tuple[IdlePath] | list[IdlePath] = (
|
|
PurePath("/etc"),
|
|
PurePath("/usr", "local", "etc"),
|
|
PurePath(str(__user_path), ".config")
|
|
)
|
|
__data_paths: tuple[IdlePath] | list[IdlePath] = (
|
|
PurePath("/usr", "local", "share"),
|
|
PurePath(str(__user_path), ".local", "share")
|
|
)
|
|
|
|
def __init__(self, ansible_proj_root: ExecutedPath | None = None):
|
|
if ansible_proj_root is not None:
|
|
self.__proj_root = ansible_proj_root
|
|
role_paths: tuple[IdlePath] | list[IdlePath] = (
|
|
PurePath(str(self.__proj_root), "roles"),
|
|
PurePath(str(self.__proj_root), ".ansible/roles")
|
|
)
|
|
roledata_paths: tuple[IdlePath] | list[IdlePath] = (
|
|
PurePath(str(self.proj_roles[0]), "files"),
|
|
PurePath(str(self.proj_roles[0]), "templates"),
|
|
PurePath(str(self.proj_roles[1]), "files"),
|
|
PurePath(str(self.proj_roles[1]), "templates")
|
|
)
|
|
setattr(self, AnsibleScopes.ROLE.name, {
|
|
"data": roledata_paths,
|
|
"vars": role_paths
|
|
})
|
|
setattr(self, AnsibleScopes.GROUPVARS.name, {
|
|
"vars": (PurePath(str(self.__proj_root), "group_vars"),)
|
|
})
|
|
setattr(self, AnsibleScopes.HOSTVARS.name, {
|
|
"vars": (PurePath(str(self.__proj_root), "host_vars"),)
|
|
})
|
|
setattr(self, AnsibleScopes.INVENTORY.name, {
|
|
"vars": (PurePath(str(self.__proj_root), "hosts.yml"),)
|
|
})
|
|
|
|
def get_scope(self, scope: AnsibleScopes = AnsibleScopes.INVENTORY.name, index = 0):
|
|
return getattr(self, scope)[index]
|
|
|
|
@property
|
|
def home(self) -> ExecutedPath:
|
|
return self.__user_path
|
|
|
|
@property
|
|
def root(self) -> ExecutedPath:
|
|
return self.__proj_root
|
|
|
|
@property
|
|
def sys_confs(self) -> ExecutedPath:
|
|
return self.__conf_paths
|
|
|
|
@property
|
|
def sys_data(self) -> ExecutedPath:
|
|
return self.__data_paths
|
|
|
|
# userSSHParams = Dict("userSSHParams", {
|
|
# "username": Required[str],
|
|
# "paths": Apps,
|
|
# "keys": dict,
|
|
# "password": Required[str],
|
|
# "fate": Literal["disposal", "retention"]
|
|
# }, total=False)
|
|
|
|
vpsSchema = Dict("vpsSchema", {
|
|
"fqdn": Required[str],
|
|
"vps_service": {
|
|
"exists": Required[bool],
|
|
"password": Required[str],
|
|
"api_key": Required[str],
|
|
"type": Required[Literal["linode"]],
|
|
"region": Literal["us-east"],
|
|
"ssh_authorized_keys": list[IdlePath | ExecutedPath | str],
|
|
"ssh_private_key_paths": list[IdlePath | ExecutedPath | str],
|
|
"ssh_private_key_path_pref": int,
|
|
"root_fate": Required[Literal["disposal","retention"]],
|
|
"ssh_motd_script_basenames": list[str]
|
|
},
|
|
"keywords": list[str]
|
|
}, total=False)
|
|
|
|
class RemoteNode:
|
|
# __user_path =
|
|
_fqdn: str | None = None
|
|
|
|
def __init__(self, cnode: ControlNode, api_key: str = secrets.token_urlsafe(32), password: str = "password123", name: str = ".test", service: VirtualPrivateServers = VirtualPrivateServers.Linode.name, region: Literal["us-east"] | None = "us-east", keywords: list[str] | None = ["server"]):
|
|
self.root: dict = dict()
|
|
self.root["username"]: str = "root"
|
|
vault = Vault(password)
|
|
self.root["password"]: str = vault.dump(password)
|
|
self.root["software"]: Software = Software()
|
|
self.owner = cnode
|
|
|
|
app_input = {
|
|
SoftPathGroup.CONFIG.name: {
|
|
SoftScope.PERSONAL.name: PurePath(str(cnode.home), ("." + Softs.ssh.name))
|
|
},
|
|
SoftPathGroup.DATA.name: {
|
|
SoftScope.GLOBAL.name: PurePath(str(cnode.sys_confs[0]), "update-motd.d")
|
|
}
|
|
}
|
|
self.root["software"].append(Softs.ssh.name, **app_input)
|
|
|
|
self._fqdn = name
|
|
self.root["software"]._fqdn = name
|
|
|
|
# root_ssh_input: userSSHParams = {
|
|
root_ssh_input = {
|
|
"username": self.root["username"],
|
|
"password": self.root["password"]
|
|
}
|
|
self.ssh: UserSSH = UserSSH(**root_ssh_input)
|
|
|
|
self.apps: list = self.root["software"].show(contents = True)
|
|
self.keywords = keywords
|
|
self._api_key: str | None = api_key
|
|
self.service = service
|
|
self.region = region
|
|
self.model: dict | None = None
|
|
|
|
def set_region(self, name: Literal["us-east"] = "us-east") -> None:
|
|
self.region = name
|
|
|
|
def set_password(self, password: str) -> None:
|
|
vault = Vault(self.root["password"])
|
|
self.root["password"] = vault.dump(self.root["password"])
|
|
|
|
def set_api(self, key: str) -> None:
|
|
vault = Vault(self._api_key)
|
|
self._api_key = vault.dump(self._api_key)
|
|
|
|
def add_tags(self, *name):
|
|
self.keywords: list = []
|
|
self.keywords += list(name)
|
|
|