admin/skato-ansible
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects 3 Releases Wiki Activity

fix, feature: type-casted a variable or literal rather than type variable, added methods for key selection/removal, added method for outputting formetted instance data

Browse Source
This commit is contained in:
xenobyte
2026-01-02 17:00:59 -05:00
parent fa1b1540d9
commit 9e4f0c6881
1 changed files with 167 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
anodes.py
View File

@@ -11,9 +11,8 @@ from custtypes import ExecutedPath, IdlePath
from enum import Enum
from softman import Software, SoftPathGroup, SoftScope
from whereami import USER_PATH, PROJ_ROOT
# @TODO use below 2 imports in 'get_head_user' function to grab username of highest-privileged login user
from os import name as shell_type
# from os import environ as env_vars
from ansible_vault import Vault
from cerberus import Validator as constrain_by
class ControlNode:
__user_path: ExecutedPath = USER_PATH
@@ -29,8 +28,20 @@ class ControlNode:
)
def __init__(self, ansible_proj_root: ExecutedPath | None):
if ansible_root is not None:
if ansible_proj_root is not None:
self.__proj_root = ansible_proj_root
self.proj_roles: tuple[IdlePath] | list[IdlePath] = (
PurePath(str(self.__proj_root), "roles"),
PurePath(str(self.__proj_root), ".ansible/roles")
)
self.role_data: tuple[IdlePath] | list[IdlePath] = (
PurePath(str(self.proj_roles[0]), "files"),
PurePath(str(self.proj_roles[0]), "templates")
)
self.invvar_data: tuple[IdlePath] | list[IdlePath] = (
PurePath(str(self.__proj_root), "group_vars"),
PurePath(str(self.__proj_root), "host_vars")
)
@property
def user_path(self) -> ExecutedPath:
@@ -53,13 +64,14 @@ class Softs(Enum):
userSSHParams = Dict("userSSHParams", {
"username": Required[str],
"paths": Required[Software._Apps],
"paths": Software._Apps,
"keys": {
"available": Required[tuple[ExecutedPath]],
"selected": Required[ExecutedPath | list[ExecutedPath] | int | list[int]],
"authorized": ExecutedPath | list[ExecutedPath] | int | list[int],
"used": ExecutedPath | list[ExecutedPath] | int | list[int]
},
"password": Required[str],
"fate": Literal["disposal", "retention"]
}, total=False)
__user_ssh_input = {
@@ -71,24 +83,50 @@ __user_ssh_input = {
"authorized": 1,
"used": 0
},
"password": "password123",
"fate": "disposal"
}
userSSH = type("userSSH", (), **__user_ssh_input)
# @TODO continue to write below function
def get_head_user():
if shell_type == "nt":
raise NotImplementedError
else:
raise NotImplementedError
vps_schema = {
"fqdn": {"type": "string"},
"vps_service": {
"type": "dict",
"schema": {
"exists": {"type": "boolean"},
"password": {"type": "string"},
"api_key": {"type": "string"},
"type": {
"type": "string",
"anyof": ["linode"]
},
"region": {
"type": "string",
"anyof": ["us-east"]
},
"ssh_authorized_keys": {"type": "list"},
"root_fate": {
"type": "string",
"anyof": ["disposal", "retention"]
}
}
},
"keywords": {"type": "list"}
}
vps_schema = constrain_by(vps_schema)
class VirtualPrivateServers(Enum):
Linode = 0
class RemoteNode:
# __user_path =
_fqdn: str | None = None
def __init__(self, cnode: ControlNode, name: str | None, keywords: list[str] | None = None):
def __init__(self, cnode: ControlNode, api_key: str, password: str, name: str, service: VirtualPrivateServers = VirtualPrivateServers.Linode.name, keywords: list[str] | None = None):
self.root: dict = dict()
self.root["username"]: str = "root"
vault = Vault(password)
self.root["password"]: str = vault.dump(password)
self.root["software"]: Software = Software()
app_input = {
@@ -101,12 +139,44 @@ class RemoteNode:
}
self.root["software"].declare(Softs.OpenSSH, **app_input)
if name is not None:
self._fqdn = name
self.root["software"]._fqdn = name
self._fqdn = name
self.root["software"]._fqdn = name
root_ssh_input: userSSHParams = {
"username": self.root["username"],
"password": self.root["password"]
}
self.ssh: userSSH = userSSH(**root_ssh_input)
self.apps: list = self.root["software"].list(contents = True)
self.keywords = keywords
self._api_key: str | None = api_key
self.service = service
def set_region(self, name: Literal["us-east"] = "us-east") -> None:
self.region = name
def get_region(self) -> str:
return self.region
def set_password(self, password: str) -> None:
vault = Vault(value)
self.root["password"] = vault.dump(value)
self.ssh.password: str = self.root["password"]
def get_password(self) -> Vault | str:
return self.root["password"]
def set_api(self, key: str) -> None:
self._api_key = key
def get_api(self) -> Vault | str:
vault = Vault(self._api_key)
return vault.dump(self._api_key)
def add_tags(self, *name):
self.keywords: list = []
self.keywords += list(name)
def import_keys(self, key_basenames: str | tuple[str], match_sort: tuple[Callable, bool] = (lambda e: e, False)):
keyfiles: list[ExecutedPath] | list[None] | None = list()
@@ -124,14 +194,8 @@ class RemoteNode:
updated_keyfiles.append(new_keyfile.resolve())
keyfiles = updated_keyfiles
root_ssh_input: userSSHParams = {
"username": "root",
"paths": self.root["software"].ssh,
"keys": {
"available": tuple(keyfiles.sort(reverse = match_sort[1], key = match_sort[0]))
}
}
self.ssh: userSSH = userSSH(**root_ssh_input)
self.ssh.paths = self.root["software"].ssh
self.ssh.keys["available"] = tuple(keyfiles.sort(reverse = match_sort[1], key = match_sort[0]))
def show_keys(self, which: Literal["authorized", "used", "available", "selected"] = "available") -> tuple[tuple, str]:
delimiters: str | tuple = "{}"
@@ -174,17 +238,16 @@ class RemoteNode:
return (tuple(keyfiles), stringified_keyfiles)
# @TODO continue to work on two below methods
def pick_keys(self, source: Literal["authorized", "used", "available", "selected"] = "available", *selections: int | ExecutedPath | str) -> list[ExecutedPath] | Never:
keyfiles = self.ssh.keys[source]
authlist = []
if source == "available":
if source == "available" or source == "selected":
for s in selections:
if isinstance(s, int):
authlist.append(keyfiles[s])
elif isinstance(s, ExecutedPath):
path_set = set([ExecutedPath])
path_set = set([s])
kf_set = set(keyfiles)
overlap = kf_set & path_set
@@ -197,11 +260,84 @@ class RemoteNode:
if s in kf_strs:
authlist.append(Path(s))
self.ssh.keys["selected"] = authlist
elif source == "selected":
raise NotImplementedError
if source == "available":
self.ssh.keys["selected"] = authlist
return self.ssh.keys["selected"]
elif source == "selected":
self.ssh.keys["authorized"] = authlist
return self.ssh.keys["authorized"]
elif source == "authorized":
for s in selections:
if isinstance(s, int):
authlist.append(keyfiles[s])
elif isinstance(s, ExecutedPath):
for p in keyfiles:
if str(s) in str(p):
authlist.append(p)
else:
continue
elif isinstance(s, str):
for p in keyfiles:
if s in str(p):
authlist.append(p)
else:
continue
self.ssh.keys["used"] = authlist
return self.ssh.keys["used"]
def remove_keys(self, source: Literal["authorized", "used", "available", "selected"] = "available", *selections: int | ExecutedPath | str) -> list[ExecutedPath] | Never:
raise NotImplementedError
keyfiles = self.ssh.keys[source]
for s in selections:
if isinstance(s, int):
removed_elem = keyfiles.pop(s)
elif isinstance(s, ExecutedPath):
path_set = set([s])
kf_set = set(keyfiles)
overlap = kf_set & path_set
if overlap is not None and len(overlap) > 0:
item = list(overlap)[0]
keyfiles = [p for p in keyfiles if p != item]
self.ssh.keys[source] = keyfiles
else:
continue
elif isinstance(s, str):
kf_strs = list(map(lambda p: str(p), keyfiles))
if s in kf_strs:
keyfiles = [p for p in keyfiles if str(p) != S]
self.ssh.keys[source] = keyfiles
self.ssh.keys[source] = keyfiles
if source == "available":
self.remove_keys("selected", *selections)
elif source == "selected":
self.remove_keys("authorized", *selections)
elif source == "authorized":
self.remove_keys("used", *selections)
return self.ssh.keys
def itemize(self):
model = dict()
vault_api = Vault(self._api_key)
vault_pass = Vault(self.ssh.password)
model["fqdn"] = self._fqdn
model["vps_service"] = {
"exists": True,
"password": vault_pass.dump(self.ssh.password),
"api_key": vault_api.dump(self._api_key),
"type": self.service.lower(),
"region": self.region,
"ssh_authorized_keys": self.ssh.keys["authorized"],
"root_fate": self.ssh.fate
}
model["keywords"] = self.keywords
if vps_schema.validate(model):
return model