From 9e4f0c688105529c76fa2f747c6c00cea586d2dc Mon Sep 17 00:00:00 2001 From: Alex Tavarez Date: Fri, 2 Jan 2026 17:00:59 -0500 Subject: [PATCH] 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 --- anodes.py | 198 +++++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 167 insertions(+), 31 deletions(-) diff --git a/anodes.py b/anodes.py index 2ab7425..6655f55 100644 --- a/anodes.py +++ b/anodes.py @@ -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() @@ -123,15 +193,9 @@ class RemoteNode: new_keyfile = Path(str(self.root["software"].ssh.CONFIG[SoftScope.PERSONAL.name])) / str(filename) 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