fix: made the return be self, given that the instance of the class itself is what is treated as the context manager by default

.gitignore

@@ -18,3 +18,4 @@ banner
 .galaxy_cache/
 galaxy_token
 uv.lock
+__pycache__/

anodes.py

@@ -1,207 +1,75 @@
-"""
+from custtypes import AnsibleRoles, AnsibleScopes, ExecutedPath
-Library of classes modeling Ansible nodes or their types.
+from whereami import USER_PATH, ANSIBLE_CONFIG, ANSIBLE_ROOTS
-"""
+from configparser import ConfigParser as cfg
-
+from typing import Sequence
-from pathlib import Path, PurePath
+from re import Pattern as RegEx
-from typing import TypeAlias as Neotype
+from pathlib import PurePath
-from typing import TypedDict as Dict
+from parse import Parser
-from typing import Never, Union, Literal
-from collections.abc import Callable
-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 
 
+# @TODO below class should mostly work as a context manager
 class ControlNode:
-    __user_path: ExecutedPath = USER_PATH
+    __parser = Parser()
-    __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):
+    def __init__(self):
-        if ansible_root is not None:
+        self.__user_path = USER_PATH()
-            self.__proj_root = ansible_proj_root
+        self.__config = cfg()
 
-    @property
+        if ANSIBLE_CONFIG.exists():
-    def user_path(self) -> ExecutedPath:
+            self.__config.read(str(ANSIBLE_CONFIG))
-        return self.__user_path
+        else:
+            raise Exception
 
-    @property
+        self.__data = None
-    def proj_root(self) -> ExecutedPath:
+        self.__filepath: ExecutedPath | None = None
-        return self.__proj_root
+        self.__file = None
 
-    @property
+    def __enter__(self):
-    def conf_paths(self) -> ExecutedPath:
+        self.__file = open(str(self.__filepath), "r+")
-        return self.__conf_paths
+        self.__data = self.__parser.load(self.__filepath)
+        return self.__data
 
-    @property
+    def __exit__(self, exc_type, exc_value, exc_traceback):
-    def conf_paths(self) -> ExecutedPath:
+        result = self.__parser.dump(self.__data)
-        return self.__data_paths
 
-class Softs(Enum):
+        if isinstance(result, str):
-    OpenSSH = "ssh"
+            self.__file.write(result)
+        else:
+            result.write(self.__file)
 
-userSSHParams = Dict("userSSHParams", {
+        self.__file.close()
-    "username": Required[str],
-    "paths": Required[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]
-    },
-    "fate": Literal["disposal", "retention"]
-}, total=False)
-__user_ssh_input = {
-    "username": "",
-    "paths": None,
-    "keys": {
-        "available": tuple(),
-        "selected": [0, 1],
-        "authorized": 1,
-        "used": 0
-    },
-    "fate": "disposal"
-}
-userSSH = type("userSSH", (), **__user_ssh_input)
 
-# @TODO continue to write below function
+    def __call__(self, scope: AnsibleScopes = AnsibleScopes.INVENTORY, pick: int | str | RegEx | AnsibleRoles = 0, filepath: ExecutedPath | str | None = None):
-def get_head_user():
+        if scope is None:
-    if shell_type == "nt":
             raise NotImplementedError
         else:
-        raise NotImplementedError
+            paths = ANSIBLE_ROOTS[scope.name.lower()]
 
-class RemoteNode:
+            if isinstance(paths, Sequence):
-    # __user_path =
+                path = None
-    _fqdn: str | None = None
+                if isinstance(pick, int):
-
+                    path = paths[pick]
-    def __init__(self, cnode: ControlNode, name: str | None, keywords: list[str] | None = None):
+                elif isinstance(pick, str):
-        self.root: dict = dict()
+                    path = tuple(filter(lambda p: str(p) == pick or pick in str(p), paths))
-        self.root["username"]: str = "root"
+                    if len(path) > 0:
-        self.root["software"]: Software = Software()
+                        path = path[0]
-
+                elif isinstance(pick, RegEx):
-        app_input = {
+                    path = tuple(filter(lambda p: pick.search(str(p)), paths))
-            SoftPathGroup.CONFIG.name: { 
+                    if len(path) > 0:
-                SoftScope.PERSONAL.name: PurePath(str(cnode.user_path), ("." + Softs.OpenSSH))
+                        path = path[0]
-            },
-            SoftPathGroup.DATA.name: {
-                SoftScope.GLOBAL.name: PurePath(str(cnode.conf_paths[0]), "update-motd.d")
-            }
-        }
-        self.root["software"].declare(Softs.OpenSSH, **app_input)
-
-        if name is not None:
-            self._fqdn = name
-            self.root["software"]._fqdn = name
-
-        self.apps: list = self.root["software"].list(contents = True)
-        self.keywords = keywords
-
-    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()
-        if isinstance(key_basenames, tuple):
-            for basename in key_basenames:
-                keyfiles += Path(str(self.root["software"].ssh.CONFIG[SoftScope.PERSONAL.name])).glob(basename)
-        elif isinstance(key_basenames, str):
-            keyfiles = Path(str(self.root["software"].ssh.CONFIG[SoftScope.PERSONAL.name])).glob(key_basenames)
                 else:
-            raise ValueError
+                    path = tuple(filter(lambda p: str(p) == pick.name.lower() or pick.name.lower() in str(p), paths))
+                    if len(path) > 0:
+                        path = path[0]
 
-        updated_keyfiles: list[ExecutedPath] = []
+        if path is None:
-        for filename in keyfiles:
+            raise KeyError
-            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 = {
+        if isinstance(filepath, ExecutedPath):
-            "username": "root",
+            filepath = str(filepath)
-            "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)
 
-    def show_keys(self, which: Literal["authorized", "used", "available", "selected"] = "available") -> tuple[tuple, str]:
+        if path.is_dir():
-        delimiters: str | tuple = "{}"
+            self.__filepath = path / filepath
-        gap: Callable[[bool], str | None] = lambda b: " " if b else ""
+        elif path.is_file():
-        sep_format: str = "{0}"
+            self.__filepath = path
-        sep: Callable[[str, bool], str] = lambda s, b: sep_format.format(s) + gap(b)
-        label_format: str = "{0}:"
-        label: Callable[[str, bool], str] = lambda s, b: label_format.format(s) + gap(b)
-
-        def render(kfs, source: str):
-            if isinstance(kfs, list):
-                member_ints: list = list()
-                for f in kfs:
-                    if isinstance(f, int):
-                        members_ints.append(f)
-                
-                if len(member_ints) > 0:
-                    kfs = [self.ssh.keys[source][i] for i in kfs]
-            elif isinstance(kfs, int):
-                kfs = [self.ssh.keys[source][kfs]]
-            elif isinstance(kfs, ExecutedPath):
-                kfs = [self.ssh.keys[source]]
-            
-            return kfs
-
-        keyfiles = self.ssh.keys[which]
-        
-        if which == "selected":
-            keyfiles = render(keyfiles, "available")
-
-        if which == "authorized":
-            keyfiles = render(keyfiles, "selected")
-
-        if which == "used":
-            keyfiles = render(keyfiles, "selected")
-
-        stringified_keyfiles = list(map(lambda t: label(t[0], True) + str(t[1]), enumerate(keyfiles)))
-        stringified_keyfiles = sep(",", True).join(stringified_keyfiles)
-        stringified_keyfiles = delimiters[0:len(delimiters)] + stringified_keyfiles[:(len(stringified_keyfiles) - 2)] + delimiters[1:]
-
-        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":
-            for s in selections:
-                if isinstance(s, int):
-                    authlist.append(keyfiles[s])
-                elif isinstance(s, ExecutedPath):
-                    path_set = set([ExecutedPath])
-                    kf_set = set(keyfiles)
-                    overlap = kf_set & path_set
-
-                    if overlap is not None and len(overlap) > 0:
-                        authlist.append(list(overlap)[0])
-                    else:
-                        continue
-                elif isinstance(s, str):
-                    kf_strs = list(map(lambda p: str(p), keyfiles))
-
-                    if s in kf_strs:
-                        authlist.append(Path(s))
-            self.ssh.keys["selected"] = authlist
-        elif source == "selected":
-            raise NotImplementedError
-
-
-    def remove_keys(self, source: Literal["authorized", "used", "available", "selected"] = "available", *selections: int | ExecutedPath | str) -> list[ExecutedPath] | Never:
-        raise NotImplementedError
 
+        return __enter__

custtypes.py

@@ -2,8 +2,81 @@
 Library of custom type hints.
 """
 
-from typing import TypeAlias as Neotype
+from typing import TypeAlias as Neotype, TypedDict as Dict
-from pathlib import PurePosixPath, PureWindowsPath, PosixPath, WindowsPath
+from typing import Required
+from collections.abc import Sequence
+from pathlib import Path, PurePath, PurePosixPath, PureWindowsPath, PosixPath, WindowsPath
+from enum import StrEnum, auto
+from io import TextIOBase, BufferedIOBase, RawIOBase
 
-ExecutedPath: Neotype = Union[PosixPath, WindowsPath]
+ExecutedPath: Neotype = PosixPath | WindowsPath
-IdlePath: Neotype = Union[PurePosixPath, PureWindowsPath]
+IdlePath: Neotype = PurePosixPath | PureWindowsPath
+File: Neotype = TextIOBase | BufferedIOBase | RawIOBase
+
+class RootFate(StrEnum):
+    disposal = auto()
+    retention = auto()
+
+class NodeType(StrEnum):
+    remote = auto()
+    control = auto()
+
+class VPS(StrEnum):
+    Linode = auto()
+
+class VPSRegion(StrEnum):
+    us_east = auto()
+
+class AnsibleScopes(StrEnum):
+    INTERNAL = auto()
+    INVENTORY = auto()
+    GROUPVARS = auto()
+    HOSTVARS = auto()
+    ROLE = auto()
+
+class AnsibleRoles(StrEnum):
+    bootstrap = auto()
+
+class Scopes(StrEnum):
+    SYS = auto()
+    USER = auto()
+    LOCAL = auto()
+    PROJ = auto()
+    SHARED = auto()
+
+class Roles(StrEnum):
+    CONF = auto()
+    DATA = auto()
+    MEM = auto()
+    EXE = auto()
+
+class UserName(StrEnum):
+    root = auto()
+
+class GroupName(StrEnum):
+    remote = auto()
+    sudo = auto()
+
+class Software(StrEnum):
+    openssh_client = auto()
+    openssh_server = auto()
+
+class SoftwareRoles(StrEnum):
+    client = auto()
+    server = auto()
+
+class PacMans(StrEnum):
+    APT = auto()
+
+class PathCollection(Dict, total=False):
+    sys: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+    user: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+    local: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+    proj: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+    shared: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+
+class PathRoles(Dict, total=False):
+    conf: PathCollection
+    data: PathCollection
+    mem: PathCollection
+    exe: PathCollection

entities.py

@@ -0,0 +1,356 @@
+from typing import Self, Literal, Never, Callable, Sequence
+from custtypes import Roles, Scopes, PathCollection, PathRoles
+from custtypes import Software, SoftwareRoles, PacMans
+from custtypes import ExecutedPath, IdlePath, File
+from custtypes import UserName, GroupName, VPS, VPSRegion, RootFate
+from pathlib import Path, PurePath
+from sshkey import SSHKeyCollection, SSHKeyType, SSHKey
+from random import choice as gamble
+from re import Pattern as RegEx
+from softman import sshd
+from yaml import YAMLObject
+from ansible_vault import Vault
+
+
+class Group(YAMLObject):
+    yaml_tag = u"!Group"
+
+    # @TODO create Enum class child for category parameter type hinting in below method
+    def __init__(self, group_name: GroupName | str = GroupName.sudo, category: Literal["system", "regular"] = "system", gid: int | str | None = 27):
+        if isinstance(group_name, GroupName):
+            self.group_name = group_name.name.lower()
+        else:
+            self.group_name = group_name
+
+        self.id = str(gid)
+        self.type = category
+
+    def __repr__(self):
+        return "%s(group_name=%r,category=%r,gid=%r)" % (
+            self.__class__.__name__,
+            self.group_name,
+            self.category,
+            self.id
+        )
+
+
+class User(YAMLObject):
+    yaml_tag = u"!User"
+
+    def __init__(self, username: UserName | str = UserName.root, password: str = "test", services: list[str | Software] = [Software.openssh_server], uid: int | str | None = 0):
+        self.exists = True
+
+        if isinstance(username, UserName):
+            self.username = username.name.lower()
+        else:
+            self.username = username
+
+        self.id = str(uid)
+        self.password = password
+
+        new_services = []
+        for s in services:
+            if isinstance(s, Software):
+                new_services.append(s.name.lower())
+            else:
+                new_services.append(s)
+        self.services: tuple = tuple(new_services)
+
+        self.shell = "/bin/bash"
+        self.home = "/"
+        self.category: Literal["system", "regular"] = "regular"
+        group = Group(username, self.id)
+        self.group = group
+        self.groups: list[str | GroupName] | None = None
+
+        if self.groups is None:
+            self.admin = True
+        elif isinstance(self.groups, Sequence) and GroupName.sudo in self.groups:
+            self.admin = True
+        else:
+            self.admin = False
+
+        ssh_keys = SSHKeyCollection()
+        ssh_keys.pull()
+        self.__ssh_keys = ssh_keys
+        # print("here")
+
+        self.__public_keys = self.__ssh_keys.publish(SSHKeyType.pubkey, datatype=list)
+        pubkeys = ssh_keys.publish(SSHKeyType.pubkey, datatype=list)
+        self.__auth_keys: SSHKeyCollection = SSHKeyCollection()
+        for p in pubkeys:
+            self.__auth_keys.append(p)
+        self.ssh_authorized_keys: list[str | None] = []
+
+        self.__private_keys = self.__ssh_keys.publish(SSHKeyType.privkey, datatype=list)[0]
+        privkeys = ssh_keys.publish(SSHKeyType.privkey, datatype=list)
+        self.__priv_keys: SSHKeyCollection = SSHKeyCollection()
+        for p in privkeys[0]:
+            self.__priv_keys.append(p)
+        self.ssh_private_key_paths: list[str | None] = []
+        self.__priv_key_pref: int = privkeys[1]
+        self.ssh_private_key_path_pref: int = privkeys[1]
+
+        self.__apps = (sshd,)
+        self.__ssh_keypairs: tuple[SSHKey | tuple[SSHKey]] | SSHKey = tuple()
+        self.__ssh_keypair_chosen = False
+    
+    def get_app(self, name: Software) -> Never:
+        raise NotImplementedError
+
+    def update_app(self, name: Software, attr: str, method = None) -> Never:
+        raise NotImplementedError
+
+    def __update_sshd(self, app: Software = Software.openssh_server):
+        for a in self.__apps:
+            if a.alt_names[PacMans.APT.name.lower()] == app.name.lower():
+                if hasattr(a, "users"):
+                    users = getattr(a, "users")
+
+                    if self.username not in users:
+                        users[self.username] = dict()
+
+                    users[self.username]["authorized_keys"] = self.__auth_keys
+                    users[self.username]["credential_keys"] = self.__priv_keys
+                    users[self.username]["keys"] = self.__ssh_keys
+                    users[self.username]["keypairs"] = self.__ssh_keypairs
+                    users[self.username]["preferred_priv_key"] = self.__priv_key_pref
+                    setattr(self, "users", users)
+                else:
+                    users = {
+                        self.username: {
+                            "auth_keys": self.__auth_keys,
+                            "priv_keys": self.__priv_keys,
+                            "keypairs": self.__ssh_keypairs,
+                            "keys": self.__ssh_keys
+                        }
+                    }
+                    a.declare(users = users)
+            else:
+                continue
+
+    def choose_keypair(self, private_key: SSHKey | ExecutedPath | str | int | RegEx, public_key: SSHKey | ExecutedPath | str | int | RegEx, from_host = True):
+        if not self.__ssh_keypair_chosen:
+            self.__priv_keys = SSHKeyCollection()
+            self.__auth_keys = SSHKeyCollection()
+
+        if from_host:
+            pubkeys = self.__ssh_keys.publish(SSHKeyType.pubkey, datatype=list)
+
+            # print(pubkeys)
+            if isinstance(public_key, int):
+                public_key = pubkeys[public_key]
+            elif isinstance(public_key, SSHKey):
+                public_key = tuple(filter(lambda k: str(k) == str(public_key()), pubkeys))
+                if len(public_key) > 0:
+                    public_key = public_key[0]
+                else:
+                    public_key = None
+            elif isinstance(public_key, str):
+                public_key = tuple(filter(lambda k: str(k) == public_key or public_key in str(k), pubkeys))
+                if len(public_key) > 0:
+                    public_key = public_key[0]
+                else:
+                    public_key = None
+            elif isinstance(public_key, RegEx):
+                public_key = tuple(filter(lambda k: public_key.search(str(k)), pubkeys))
+                if len(public_key) > 0:
+                    public_key = public_key[0]
+                else:
+                    public_key = None
+            else:
+                public_key = tuple(filter(lambda k: str(k) == str(public_key), pubkeys))
+                if len(public_key) > 0:
+                    public_key = public_key[0]
+                else:
+                    public_key = None
+
+            privkeys = self.__ssh_keys.publish(SSHKeyType.privkey, datatype=list)[0]
+
+            if isinstance(private_key, int):
+                private_key = privkeys[private_key]
+            elif isinstance(private_key, SSHKey):
+                private_key = tuple(filter(lambda k: str(k) == str(private_key()), privkeys))
+                if len(private_key) > 0:
+                    private_key = private_key[0]
+                else:
+                    private_key = None
+            elif isinstance(private_key, str):
+                private_key = tuple(filter(lambda k: str(k) == private_key or private_key in str(k), privkeys))
+                if len(private_key) > 0:
+                    private_key = private_key[0]
+                else:
+                    private_key = None
+            elif isinstance(private_key, RegEx):
+                private_key = tuple(filter(lambda k: private_key.search(str(k)), privkeys))
+                if len(private_key) > 0:
+                    private_key = private_key[0]
+                else:
+                    private_key = None
+            else:
+                private_key = tuple(filter(lambda k: str(k) == str(private_key), privkeys))
+                if len(private_key) > 0:
+                    private_key = private_key[0]
+                else:
+                    private_key = None
+        else:
+            if isinstance(public_key, SSHKey):
+                public_key = public_key()
+            elif isinstance(public_key, str):
+                public_key = Path(public_key)
+
+            self.__ssh_keys.append(public_key)
+
+            if isinstance(private_key, SSHKey):
+                private_key = private_key()
+            elif isinstance(private_key, str):
+                private_key = Path(private_key)
+
+            self.__ssh_keys.append(private_key)
+
+        if private_key is None or public_key is None:
+            raise KeyError
+
+        self.__auth_keys.append(public_key)
+        self.ssh_authorized_keys.append(public_key.read_text())
+        self.__priv_keys.append(private_key)
+        self.ssh_private_key_paths.append(str(private_key))
+        self.__ssh_keypairs = (*self.__ssh_keypairs, (self.__priv_keys.tail, self.__auth_keys.tail),)
+        self.__priv_key_pref = len(self.__ssh_keypairs) - 1
+        self.ssh_private_key_path_pref = len(self.__ssh_keypairs) - 1
+        self.__update_sshd()
+
+        self.__ssh_keypair_chosen = True
+
+    def get_keypair(self, preference: int):
+        if not self.__ssh_keypair_chosen:
+            raise Exception
+
+        if isinstance(self.__ssh_keypairs, SSHKey) and not isinstance(self.__ssh_keypairs(), tuple):
+            raise ValueError
+
+        if isinstance(self.__ssh_keypairs, SSHKey):
+            if isinstance(self.__ssh_keypairs(), tuple):
+                return self.__ssh_keypairs[preference]
+            else:
+                return self.__ssh_keypairs
+        else:
+            return self.__ssh_keypairs[preference]
+
+    @property
+    def keys(self) -> SSHKeyCollection:
+        return self.__ssh_keys
+
+    @property
+    def public_keys(self) -> SSHKeyCollection:
+        return self.__public_keys
+
+    @property
+    def private_keys(self) -> SSHKeyCollection:
+        return self.__private_keys
+
+    @property
+    def keypair_preference(self) -> int:
+        return self.__priv_key_pref
+
+    def prefer_keypair(self, preference: int | RegEx | str):
+        if isinstance(preference, int):
+            if preference < len(self.__ssh_keypairs):
+                self.__priv_key_pref = preference
+            else:
+                raise KeyError
+        elif isinstance(preference, RegEx):
+            count = 0
+            for keypair in self.__ssh_keypairs:
+                if preference.search(keypair[0]) or preference.search(keypair[1]):
+                    self.__priv_key_pref = count
+                count += 1
+        else:
+            count = 0
+            for keypair in self.__ssh_keypairs:
+                if preference in str(keypair[0]()) or preference in str(keypair[1]()):
+                    self.__priv_key_pref = count
+                count += 1
+
+    @property
+    def keypairs(self) -> tuple[SSHKey] | SSHKey | None:
+        return self.__ssh_keypairs
+
+    @property
+    def keypair(self):
+        kp = self.__ssh_keypairs[self.__priv_key_pref]
+        return kp[0] + kp[1]
+
+    @property
+    def authorized_keys(self) -> SSHKeyCollection:
+        return self.__auth_keys
+
+    @property
+    def credential_keys(self) -> SSHKeyCollection:
+        return self.__priv_keys
+
+    def __repr__(self) -> str:
+        return "%s(username=%r,password=%r,services=%r,uid=%r)" % (
+            self.__class__.__name__,
+            self.username,
+            self.password,
+            self.services,
+            self.id
+        )
+
+
+class AnsibleCrypt:
+    def __init__(self, string: str, source: File | None = None):
+        self.__args = (string, source)
+        self.__lock = Vault(string).dump
+        self.__stream = None
+
+        if source is not None:
+            self.__stream = source.read()
+            self.__data = self.__lock(string, self.__stream)
+        else:
+            self.__data = self.__lock(string)
+
+    def unlock(self, string: str):
+        unlock = Vault(string).load
+
+        if self.__stream is not None:
+            result = unlock(self.__stream)
+        else:
+            result = unlock(string)
+
+        return result
+
+    def __str__(self):
+        return self.__data
+
+    def __repr__(self):
+        return "%s(%r, source=%r)" % (
+            self.__class__.__name__,
+            *self.__args
+        )
+
+
+class VirtualPrivateServer(YAMLObject):
+    yaml_tag = u"!VirtualPrivateServer"
+
+    def __init__(self, root: User, api: str, vps: VPS | str = VPS.Linode):
+        self.region: VPSRegion | None = None
+
+        if vps == VPS.Linode:
+            self.region = VPSRegion.us_east
+
+        api_key = AnsibleCrypt(api)
+        self.__api_key: AnsibleCrypt = api_key
+        self.api_key: str = str(api_key)
+        self.password: str = root.password
+        self.exists: bool = True
+        self.type: str = vps.name.lower()
+        self.__default_fate: RootFate = RootFate.disposal
+        self.root_fate: str = self.__default_fate.name.lower()
+        self.ssh_authorized_keys: list[str] = root.ssh_authorized_keys
+        self.ssh_private_key_paths: list[str] = root.ssh_private_key_paths
+        self.ssh_private_key_path_pref: int = root.ssh_private_key_path_pref
+        # @TODO add SSH MOTD attribute
+
+

gpgkey.py

@@ -0,0 +1 @@
+# @TODO create classes similar to those in sshkey module, for GPG keys

main.py

@@ -1,28 +1,65 @@
-import click
+"""
-from pathlib import Path, PurePath, PurePosixPath, PureWindowsPath, PosixPath, WindowsPath
+Library for the CLI commands and the related classes and functions
-from typing import TypeAlias as Neotype
+"""
-from typing import Union
-# import configparser as ini
-# from cerberus import Validator as constrain_by
-# import skansible_types as skato
 
-ExecutedPath: Neotype = Union[PosixPath, WindowsPath]
+import click as cli
-IdlePath: Neotype = Union[PurePosixPath, PureWindowsPath]
+from custtypes import AnsibleScopes, VPS, VPSRegion, RootFate, UserName
+from whereami import PROJ_ROOT, ANSIBLE_ROOTS
+from servs import User
+from pathlib import PurePath, Path
+from sshkey import SSHKeyType
+from ansible_vault import Vault
+import yaml as yams
 
-# @NOTE https://docs.python.org/3/library/configparser.html#quick-start
+@cli.group()
-class Config:
+@cli.option("-d", "--debug", type=bool, is_flag=True, default=False, help="Use debugging mode")
-    path: IdlePath = PurePath(str(Path(__file__).parent.resolve())) / "config.ini"
+@cli.pass_context
+def skansible(ctx, debug):
+    ctx.ensure_object(dict)
+    ctx.obj["DEBUG"] = debug
 
-@click.group()
+@skansible.command()
-def skansible():
+@cli.argument("api_key")
-    raise NotImplementedError
+@cli.option("-s", "--vps", type=cli.Choice(VPS, case_sensitive=False), default="Linode", help="Set the type of VPS")
+@cli.option("-r", "--region", type=cli.Choice(VPSRegion, case_sensitive=False), default="us_east", help="Set the VPS region")
+@cli.option("-0", "--root", type=bool, is_flag=True, default=True, help="Declare root SSH login credentials")
+@cli.option("-f", "--fate", type=cli.Choice(RootFate, case_sensitive=False), default="disposal", help="Choose the eventual fate of the root account")
+@cli.option("-h", "--host", multiple=True, type=str, default="all", help="Specify what inventory host or group this is being set")
+@cli.pass_context
+def init(ctx, vps, region, root, fate, host, api_key):
+    if root:
+        password = cli.prompt("Please enter a password: ", type=str, hide_input=True, confirmation_prompt=True)
+        root = User(UserName.root.name.lower(), password)
+        pubkeys = root.ssh_keys.publish(SSHKeyType.pubkey.name.lower(), datatype=list)
+        pubkey_opts = map(lambda k: str(k), pubkeys)
+        chosen_pubkey = cli.prompt("Authorize one of the following SSH public keys: ", type=cli.Choice(pubkey_opts, case_sensitive=True), show_choices=True)
+        chosen_pubkey = Path(chosen_pubkey)
+        privkeys = root.ssh_keys.publish(SSHKeyType.privkey.name.lower(), datatype=list)[0]
+        chosen_privkey = tuple(filter(lambda k: k.stem == chosen_pubkey.stem, privkeys))[0]
 
-@click.command()
+    inv_vars = []
-def init():
+    for h in host:
-    if Path(str(Config.path)).exists():
+        inv_vars += list(ANSIBLE_ROOTS[AnsibleScopes.HOSTVARS.name.lower()].glob(h)) + list(ANSIBLE_ROOTS[AnsibleScopes.GROUPVARS.name.lower()].glob(h))
-        click.echo("")
+
+    if len(inv_vars) > 0:
+        for p in inv_vars:
+            with open(str(p), "r+") as file:
+                content = yams.load(file.read(), Loader=yams.Loader)
+                
+                if "vps_service" in content:
+                    content["vps_service"]["exists"] = True
+                    crypt_key = Vault(api_key)
+                    content["vps_service"]["api_key"] = crypt_key.dump(api_key)
+                    content["vps_service"]["type"] = vps.lower()
+                    content["vps_service"]["region"] = region.replace("_", "-")
+                    content["vps_service"]["root_fate"] = fate
+                    crypt_key = Vault(root.password)
+                    content["vps_service"]["password"] = crypt_key.dump(root.password)
     else:
+        for h in host:
+            path = ANSIBLE_ROOTS[AnsibleScopes.GROUPVARS.name.lower()] / h
+            with open(str(path), "w") as file:
                 pass
 
 if __name__ == "__main__":
-    skansible()
+    skansible(obj={})

parse.py

@@ -0,0 +1,89 @@
+from pathlib import Path
+import yaml as yams
+from configparser import ConfigParser as cfg
+from custtypes import ExecutedPath
+from re import compile as rgx, IGNORECASE
+from whereami import PROJ_ROOT
+from typing import Literal
+
+class Parser:
+    def __init__(self):
+        self.__is_yaml = rgx(r".*\.ya?ml$", flags=IGNORECASE).match
+        self.__is_ini = rgx(r".*\.ini$", flags=IGNORECASE).match
+        self.__is_config = rgx(r".*\.cfg$", flags=IGNORECASE).match
+        self.__is_json = rgx(r".*\.[jb]son$", flags=IGNORECASE).match
+        self.__data = None
+        self.__content: str | None = None
+        self.__is_path: bool = False
+        # @TODO use Enum child class for below type hint instead
+        self.__method: Literal["yaml", "config", "generic"] = "generic"
+        self.__file: ExecutedPath | None = None
+
+    def load(self, filepath: ExecutedPath | str, method: Literal["yaml", "config", "generic"] = "generic", **kwargs):
+        if isinstance(filepath, ExecutedPath):
+            self.__is_path = True
+            self.__file = filepath
+            filepath = str(filepath)
+        else:
+            if isinstance(filepath, str) and Path(filepath).exists():
+                self.__file = Path(filepath)
+                self.__is_path = True
+            else:
+                self.__is_path = False
+
+                if isinstance(filepath, str):
+                    self.__content = filepath
+
+        if self.__is_yaml(filepath) or method == "yaml":
+            self.__method = "yaml"
+
+            if self.__is_path:
+                filepath = open(str(filepath), "r+")
+            
+            if len(kwargs) > 0:
+                self.__data = yams.load_all(filepath, Loader=yams.Loader, **kwargs)
+            else:
+                self.__data = yams.load_all(filepath, Loader=yams.Loader)
+            filepath.close()
+        elif self.__is_config(filepath) or method == "config":
+            self.__method = "config"
+            self.__data = cfg()
+
+            if self.__is_path:
+                read = self.__data.read
+            else:
+                read = self.__data.read_string
+            
+            if len(kwargs) > 0:
+                self.__data.read(filepath, **kwargs)
+            else:
+                self.__data.read(filepath)
+        else:
+            raise TypeError
+        
+        return self.__data
+
+    def dump(self, obj = None, method: Literal["yaml", "config", "generic"] | None = "generic", **kwargs):
+        if isinstance(obj, yams.YAMLObject) or self.__method == "yaml" or method == "yaml":
+            if obj is None:
+                obj = self.__data
+
+            if len(kwargs) > 0:
+                self.__content = "---\n" + yams.dump(obj, Dumper=yams.Dumper, **kwargs)
+            else:
+                self.__content = "---\n" + yams.dump(obj, Dumper=yams.Dumper)
+        elif isinstance(obj, ConfigParser) or self.__method == "config" or method == "config":
+            if obj is None:
+                if self.__is_path:
+                    return self.__file.read_text()
+                else:
+                    if self.__file is None:
+                        return self.__content
+                    else:
+                        return self.__file.read_text()
+
+            raise NotImplementedError
+        else:
+            raise TypeError
+
+        return self.__content

pyproject.toml

@@ -8,6 +8,7 @@ dependencies = [
     "ansible>=13.1.0",
     "ansible-lint>=25.12.1",
     "ansible-navigator>=25.12.0",
-    "cerberus>=1.3.8",
+    "ansible-vault>=4.1.0",
     "click>=8.3.1",
+    "validators>=0.35.0",
 ]

softman.py

@@ -1,153 +1,157 @@
-"""
+from typing import Self, Never, Callable, Sequence
-Library of classes modeling software and software-related 
+from custtypes import Roles, Scopes, PathCollection, PathRoles
-data as represented in or used by Ansible.
+from custtypes import Software, SoftwareRoles, ExecutedPath
-"""
+from custtypes import PacMans, UserName, GroupName, IdlePath
+from custtypes import ExecutedPath, AnsibleRoles
+from sshkey import SSHKeyCollection, SSHKeyType, SSHKey
+from whereami import PROJ_ROLES
 
-from typing import TypeAlias as Neotype
+class App:
-from typing import TypedDict as Dict
+    def __init__(self, name: Software, role: SoftwareRoles = SoftwareRoles.client, paths: PathRoles | None = None):
-from typing import Never, Union
+        self.__name = name.name.lower().replace("_", "-")
-from custtypes import ExecutedPath, IdlePath
+        # @TODO create dict type hint for below data struct
-from enum import Enum
+        self.alt_names = dict()
-from pathlib import Path, PurePath
+        self.alt_names[PacMans.APT.name.lower()] = self.__name
-from whereami import USER_PATH, PROJ_ROOT
+        self.role = role.name.lower()
+        if paths is not None:
+            if Roles.EXE in paths:
+                setattr(self, "_" + Roles.EXE.name.lower(), paths[Roles.EXE.name.lower()])
+            if Roles.CONF in paths:
+                setattr(self, "_" + Roles.CONF.name.lower(), paths[Roles.CONF.name.lower()])
+            if Roles.DATA in paths:
+                setattr(self, "_" + Roles.DATA.name.lower(), paths[Roles.DATA.name.lower()])
+            if Roles.MEM in paths:
+                setattr(self, "_" + Roles.MEM.name.lower(), paths[Roles.MEM.name.lower()])
+        self.__parents: tuple[Self] | None = None
+        self.__children: tuple[Self| None] = []
+        self.__api: str | None = None
+        self.__current_filepath: IdlePath | ExecutedPath | None = None
+        self.__content: str | None = None
 
-AppPath: Neotype = Union[ExecutedPath, IdlePath]
+    @property
+    def conf_paths(self):
+        if hasattr(self, "_" + Roles.CONF.name.lower()):
+            return self._conf
+        else:
+            raise Exception
 
-class SoftScope(Enum):
+    @property
-    PERSONAL = 0
+    def data_paths(self):
-    LOCAL = 1
+        if hasattr(self, "_" + Roles.DATA.name.lower()):
-    GLOBAL = 2
+            return self._data
+        else:
+            raise Exception
 
-class SoftPathGroup(Enum):
+    @property
-    CONFIG = 0
+    def exec_paths(self):
-    DATA = 1
+        if hasattr(self, "_" + Roles.EXE.name.lower()):
-    MEM = 2
+            return self._exe
+        else:
+            raise Exception
 
-# @TODO continue adding magic methods to below class
+    @property
-# @NOTE https://rszalski.github.io/magicmethods/#sequence
+    def mem_paths(self):
-class Software:
+        if hasattr(self, "_" + Roles.MEM.name.lower()):
-    # @TODO 2 options: add additional requirements to type definition below, 
+            return self._mem
-    # or ensure attribute which stores data of this type also stores other data
+        else:
-    _AppParams = Dict("_AppParams", {
+            raise Exception
-        SoftPathGroup.CONFIG.name: {
-            SoftScope.PERSONAL.name: IdlePath | list[IdlePath],
-            SoftScope.LOCAL.name: IdlePath | list[IdlePath],
-            SoftScope.GLOBAL.name: IdlePath | list[IdlePath]
-        },
-        SoftPathGroup.DATA.name: {
-            SoftScope.PERSONAL.name: IdlePath | list[IdlePath],
-            SoftScope.LOCAL.name: IdlePath | list[IdlePath],
-            SoftScope.GLOBAL.name: IdlePath | list[IdlePath]
-        },
-        SoftPathGroup.MEM.name: {
-            SoftScope.PERSONAL.name: IdlePath | list[IdlePath],
-            SoftScope.LOCAL.name: IdlePath | list[IdlePath],
-            SoftScope.GLOBAL.name: IdlePath | list[IdlePath]
-        }
-    }, total=False)
-    _SubAppParams = Dict("_SubAppParams", {
-        SoftScope.PERSONAL.name: IdlePath | list[IdlePath],
-        SoftScope.LOCAL.name: IdlePath | list[IdlePath],
-        SoftScope.GLOBAL.name: IdlePath | list[IdlePath]
-    }, total=False)
 
-    __app_input = {
+    def get_paths(self, role: Roles = Roles.CONF):
-        SoftPathGroup.CONFIG.name: {
+        if hasattr(self, "_" + role.name.lower()):
-            SoftScope.PERSONAL.name: [],
+            return getattr(self, "_" + role.name.lower())
-            SoftScope.LOCAL.name: [],
+        else:
-            SoftScope.GLOBAL.name: []
+            raise Exception
-        },
+
-        SoftPathGroup.DATA.name: {
+    def append(self, datatype: Roles = Roles.CONF, scope: Scopes | None = None, path: IdlePath | ExecutedPath | str | PathCollection | None = None):
-            SoftScope.PERSONAL.name: [],
+        if path is None:
-            SoftScope.LOCAL.name: [],
+            raise TypeError
-            SoftScope.GLOBAL.name: []
+
-        },
+        datatype = datatype.name.lower()
-        SoftPathGroup.MEM.name: {
+        
-            SoftScope.PERSONAL.name: [],
+        if hasattr(self, "_" + datatype):
-            SoftScope.LOCAL.name: [],
+            paths = getattr(self, "_" + datatype)
-            SoftScope.GLOBAL.name: []
+        else:
+            setattr(self, "_" + datatype, dict())
+            paths = getattr(self, "_" + datatype)
+
+        if scope is not None:
+            if isinstance(path, str):
+                path: ExecutedPath = Path(path)
+
+            if scope.name.lower() not in paths:
+                paths[scope.name.lower()] = []
+
+            paths[scope.name.lower()].append(path)
+        else:
+            paths: PathCollection = path
+
+        setattr(self, "_" + datatype, paths)
+
+    def inherit(self, other):
+        if other._App__children is None:
+            other._App__children = tuple()
+        
+        if self not in other._App__children:
+            other.adopt(self)
+
+        self.__parents = (*self.__parents, other)
+
+    def adopt(self, other: Self):
+        if other._App__parents is None:
+            other._App__parents = tuple()
+
+        if self not in other._App__parent:
+            other.inherit(self)
+
+        self.__children = (*self.__children, other)
+
+    def __enter__(self) -> dict | Sequence:
+        self.__content = self.__current_filepath.read_text()
+        return self.__content
+
+    def __exit__(self, exc_type, exc_value, exc_traceback) -> None:
+        # txt = yams.dump(self.__content)
+        # self.__file.write(txt)
+        # del txt
+        self.__file.close()
+
+    def __call__(self, path = str, mode = "r+", scope: Scopes = Scopes.PROJ, role: Roles = Roles.CONF, index: int = 0) -> Callable:
+        if not hasattr(self, "_" + role.name.lower()):
+            raise Exception
+        else:
+            config = getattr(self, "_" + role.name.lower())
+
+        conf_coll = config[scope.name.lower()]
+
+        if isinstance(conf_coll, Sequence):
+            conf_coll = conf_coll[index]
+        
+        if isinstance(conf_coll, str):
+            conf_coll = Path(conf_coll)
+        
+        filepath = conf_coll / path
+
+        self.__current_filepath = filepath
+        self.__file = open(str(filepath), mode)
+
+        return self
+
+    # @TODO write below method to duplicate file or template in local to project role file/template
+    def clone(self, source_scope: Scopes = Scopes.SYS, target_scope: Scopes = Scopes.PROJ, index: int = 0) -> Never:
+        raise NotImplementedError
+
+    def declare(self, **kwargs):
+        for k, v in kwargs.items():
+            setattr(self, k, v)
+
+sshd_proj_files = map(lambda r: r / AnsibleRoles.bootstrap.name.lower() / "files" / "sshd_config.d", PROJ_ROLES)
+sshd_proj_files = list(filter(lambda p: p.exists(), sshd_proj_files))
+sshd_proj_templates = map(lambda r: r / AnsibleRoles.bootstrap.name.lower() / "templates" / "sshd_config.d", PROJ_ROLES)
+sshd_proj_templates = list(filter(lambda p: p.exists(), sshd_proj_templates))
+# @TODO rewrite below using DIR_ROOTS var from whereami module
+sshd_paths: PathRoles = {
+    Roles.CONF.name.lower(): {
+        Scopes.PROJ.name.lower(): sshd_proj_files + sshd_proj_templates
     }
 }
-    _Apps = type("_Apps", (), **__app_input)
+sshd = App(Software.openssh_server, SoftwareRoles.server, sshd_paths)
-    __user_path: ExecutedPath = USER_PATH
-
-    def __init__(self):
-        self._fqdn: str | None = None
-
-    def declare(self, name: str, **kwpaths: Software._SubAppParams) -> Software._AppParams:
-        keyword_args: Software._AppParams = kwpaths
-        
-        app = Software._Apps(**keyword_args)
-        setattr(self, name, app)
-        return app
-
-    def __getitem__(self, key: str) -> Software._AppParams | Never:
-        if hasattr(self, key):
-            app: Software._Apps = getattr(self, key)
-        else:
-            raise KeyError
-        return app
-
-    def __setitem__(self, key: tuple[str, SoftPathGroup], **value: IdlePath | list[IdlePath]) -> None | Never:
-        if len(value) < 1 or len(value) > 3:
-            raise ValueError
-
-        app_params: Software._SubAppParams = value
-
-        if hasattr(self, key[0]):
-            app: Software._Apps = getattr(self, key[0])
-
-            if hasattr(app, key[1]):
-                app_child: Software._SubAppParams = getattr(app, key[1])
-
-                for k, v in app_params.items():
-                    v = [v] if not isinstance(v, list) else v
-                    app_child[k]: IdlePath | list[IdlePath] = v
-                    setattr(app, key[1], app_child)
-            else:
-                raise KeyError
-
-            setattr(self, key[0], app)
-        else:
-            raise KeyError
-
-    def __delitem__(self, key: tuple[str | SoftPathGroup]) -> None | Never:
-        if len(key) < 1 or len(key) > 3:
-            raise KeyError
-
-        if not hasattr(self, key[0]):
-            raise KeyError
-
-        if len(key) == 1:
-            delattr(self, key[0])
-        elif len(key) > 1:
-            app: Software._Apps = getattr(self, key[0])
-            delattr(app, key[1])
-            setattr(self, key[0], app)
-
-    def list(self, contents: bool = False) -> tuple[str]:
-        apps: tuple[str] | tuple[Software._Apps] = tuple(
-            filter(
-                lambda a: isinstance(getattr(self, a), Software._Apps),
-                dir(self)
-            )
-        )
-
-        if contents:
-            apps = tuple(
-                map(
-                    lambda a: getattr(self, a),
-                    apps
-                )
-            )
-
-        return apps
-
-    def __len__(self) -> int:
-        apps: tuple[str] = tuple(
-            filter(
-                lambda a: isinstance(getattr(self, a), Software._Apps),
-                dir(self)
-            )
-        )
-        return len(apps)

sshkey.py

@@ -0,0 +1,660 @@
+from re import Pattern as RegEx
+from re import fullmatch as Match
+from pathlib import Path, PurePath
+from custtypes import ExecutedPath, IdlePath
+from enum import StrEnum, auto
+from random import choice as gamble
+# from collections.abc import Sequence, Iterable
+from typing import Never, Self, Callable, Iterable, Sequence
+from whereami import USER_PATH
+from itertools import chain
+# import os
+
+class SSHKeyType(StrEnum):
+    pubkey = auto()
+    privkey = auto()
+    dual = auto()
+
+
+# @TODO create unit tests for below class
+class SSHKey:
+    def __init__(self, *path: ExecutedPath | str):
+        if len(path) > 2 or len(path) < 1:
+            raise ValueError
+
+        path = tuple(map(lambda s: Path(s) if isinstance(s, str) else s, path))
+
+        self.__idx: int = 0
+        self.__prev: Self | None = None
+        self.__next: Self | None = None
+
+        self.category: SSHKeyType | None = None
+        if len(path) < 2:
+            self.__value: ExecutedPath | tuple[ExecutedPath] = path[0]
+        else:
+            self.category = SSHKeyType.dual.name.lower()
+            self.__value: ExecutedPath  | tuple[ExecutedPath] = path
+
+    def __int__(self) -> int:
+        return self.__idx
+
+    def update_status(self) -> None:
+        if isinstance(self.__value, tuple):
+            privkey_present = False
+            pubkey_present = False
+            for p in self.__value:
+                if "-----BEGIN OPENSSH PRIVATE KEY-----" in p.read_text():
+                    privkey_present = True
+                else:
+                    pubkey_present = True
+
+            if pubkey_present and privkey_present:
+                self.category = SSHKeyType.dual.name.lower()
+            elif pubkey_present or privkey_present:
+                if pubkey_present:
+                    self.category = SSHKeyType.pubkey.name.lower()
+                if privkey_present:
+                    self.category = SSHKeyType.privkey.name.lower()
+        elif isinstance(self.__value, ExecutedPath):
+            if "-----BEGIN OPENSSH PRIVATE KEY-----" in self.__value.read_text():
+                self.category = SSHKeyType.privkey.name.lower()
+            else:
+                self.category = SSHKeyType.pubkey.name.lower()
+
+    def __str__(self) -> str:
+        if isinstance(self.__value, tuple):
+            key_basename = Path(str(self.__value[0])).stem + "." + self.category
+        else:
+            key_basename = Path(str(self.__value)).name
+        return "🔑" + key_basename
+
+    def __repr__(self) -> str:
+        return "%s(%r)" % (self.__class__.__name__, self.__value)
+
+    def __nonzero__(self) -> bool:
+        return True
+
+    def __format__(self, formatstr) -> str:
+        match formatstr:
+            case "item":
+                return str(self.__idx) + ": " + str(self.__value)
+            case "int":
+                return str(self.__idx)
+            case _:
+                return str(self)
+
+    def __next__(self) -> ExecutedPath | tuple[ExecutedPath]:
+        return self.__next
+
+    def __prev__(self) -> ExecutedPath | tuple[ExecutedPath]:
+        return self.__prev
+
+    def __call__(self, *path: ExecutedPath | str | None) -> ExecutedPath | tuple[ExecutedPath]:
+        if path is not None or len(path) > 0:
+            if len(path) > 2:
+                raise ValueError
+
+            path = tuple(map(lambda s: Path(s) if isinstance(s, str) else s, path))
+
+            if len(path) < 2:
+                self.__value = path[0]
+            else:
+                self.__value = path
+
+        return self.__value
+
+    def __eq__(self, other: Self) -> bool:
+        return self.__value == other._SSHKey__value
+
+    def __ne__(self, other: Self) -> bool | Never:
+        return self.__value != other._SSHKey__value
+
+    def __eqcontent__(self, other: Self) -> bool:
+        return self.__value.read_text() == other._SSHKey__value.read_text()
+
+    def __neqcontent__(self, other: Self) -> bool:
+        return self.__value.read_text() != other._SSHKey__value.read_text()
+
+    def __len__(self):
+        if isinstance(self.__value, tuple):
+            return len(self.__value)
+        else:
+            return 1
+
+    def update(self, *path: ExecutedPath | str) -> Self | Never:
+        if len(path) > 2 or len(path) < 1:
+            raise ValueError
+
+        path = tuple(map(lambda s: Path(s) if isinstance(s, str) else s, path))
+
+        if len(path) < 2:
+            self.__value = path[0]
+        else:
+            self.__value = path
+
+        return self
+
+    def __add__(self, other: Self | ExecutedPath | str) -> Self:
+        if isinstance(other, (str, ExecutedPath)):
+            if isinstance(self.__value, tuple):
+                self.update(*self.__value, other)
+            else:
+                self.update(*self.__value, other)
+        else:
+            if isinstance(other._SSHKey__value, tuple):
+                if isinstance(self.__value, tuple):
+                    self.update(*self.__value, *other._SSHKey__value)
+                else:
+                    self.update(self.__value, *other._SSHKey__value)
+            else:
+                if isinstance(self.__value, tuple):
+                    self.update(*self.__value, other._SSHKey__value)
+                else:
+                    self.update(self.__value, other._SSHKey__value)
+
+        self.update_status()
+
+        return self
+
+    def __radd__(self, other: Self | ExecutedPath | str) -> Self:
+        if isinstance(self.__value, tuple):
+            if isinstance(other, (ExecutedPath, str)):
+                other.update(other, *self.__value)
+            else:
+                if isinstance(other._SSHKey__value, tuple):
+                    other.update(*other._SSHKey__value, *self.__value)
+                else:
+                    other.update(other._SSHKey__value, *self.__value)
+        else:
+            if isinstance(other, (ExecutedPath, str)):
+                other.update(other, self.__value)
+            else:
+                if isinstance(other._SSHKey__value, tuple):
+                    other.update(*other._SSHKey__value, self.__value)
+                else:
+                    other.update(other._SSHKey__value, self.__value)
+
+        other.update_status()
+
+        return self
+
+    # @TODO write following 2 subtraction algorithms using 'set' data type conversion and methods
+    def __sub__(self, other: Self | ExecutedPath | str) -> Never:
+        raise NotImplementedError
+
+    def __rsub__(self, other: Self | ExecutedPath | str) -> Never:
+        raise NotImplementedError
+
+    def __getitem__(self, key: int) -> ExecutedPath | str | Never:
+        if isinstance(self.__value, tuple):
+            return self.__value[key]
+        else:
+            raise KeyError
+
+    def __setitem__(self, key: int, value: ExecutedPath | str) -> None:
+        if isinstance(self.__value, tuple):
+            new_entry = list(self.__value)
+
+            if isinstance(value, str):
+                value = Path(value)
+
+            new_entry[key] = value
+            self.__value = tuple(new_entry)
+        else:
+            raise KeyError
+
+    def replace(self, old: ExecutedPath | str | tuple[ExecutedPath | str] | list[ExecutedPath | str], new: ExecutedPath  | str | tuple[ExecutedPath | str] | list[ExecutedPath | str]) -> Self | Never:
+        if isinstance(old, str):
+            old = Path(old)
+        if isinstance(new, str):
+            new = Path(new)
+
+        if isinstance(old, (list, tuple)):
+            if len(old) > 2 or len(old) < 1:
+                raise ValueError
+
+            old = tuple(map(lambda p: Path(p) if isinstance(p, str) else p, old))
+        if isinstance(new, (list, tuple)):
+            if len(new) > 2 or len(new) < 1:
+                raise ValueError
+
+            new = tuple(map(lambda p: Path(p) if isinstance(p, str) else p, new))
+
+        if isinstance(self.__value, (tuple, list)):
+            if isinstance(old, tuple):
+                remaining_value = list(filter(lambda p: p not in old, self.__value))
+
+                if isinstance(new, tuple):
+                    self.__value = (*remaining_value, *new)
+                else:
+                    self.__value = (*remaining_value, new)
+            else:
+                remaining_value = list(filter(lambda p: p != old, self.__value))
+
+                if isinstance(new, tuple):
+                    self.__value = (*remaining_value, *new)
+                else:
+                    self.__value = (*remaining_value, new)
+
+            if len(self.__value) > 2:
+                self.__value = self.__value[0]
+        elif isinstance(self.__value, ExecutedPath):
+            if isinstance(old, tuple):
+                remaining_value = None if self.__value in old else self.__value
+            else:
+                remaining_value = None if self.__value == old else self.__value
+
+            if remaining_value is None:
+                self.__value = new
+            else:
+                raise ValueError
+
+        return self
+
+    def read(self, idx: int | None = None) -> str | tuple[str]:
+        if idx is not None and isinstance(self.__value, tuple):
+            result = self.__value[idx]
+        else:
+            if idx is not None:
+                raise KeyError
+
+            result = self.__value
+        
+        if isinstance(result, tuple):
+            result = tuple(map(lambda p: p.read_text(), result))
+        else:
+            result = result.read_text()
+
+        return result
+
+    @property
+    def status(self) -> str:
+        self.update_status()
+
+        return self.category
+
+    def reverse(self) -> Never:
+        if isinstance(self.__value, tuple):
+            v1 = self.__value[0]
+            v2 = self.__value[1]
+            result = self.update(v2, v1)
+        else:
+            result = self
+        return result
+
+
+
+def prev(arg):
+    if isinstance(arg, SSHKey):
+        return arg._SSHKey__prev__()
+    else:
+        raise TypeError
+
+def stream_equal(s1, s2) -> bool | Never:
+    if isinstance(s1, SSHKey) and isinstance(s2, SSHKey):
+        return s1._SSHKey__eqcontent__(s2)
+    else:
+        raise TypeError
+
+def stream_unequal(s1, s2) -> bool | Never:
+    if isinstance(s1, SSHKey) and isinstance(s2, SSHKey):
+        return s1._SSHKey__neqcontent__(s2)
+    else:
+        raise TypeError
+
+
+# @TODO create unit tests for below class
+class SSHKeyCollection(Sequence):
+    __user_path: ExecutedPath = USER_PATH()
+    __ssh_path: ExecutedPath = __user_path / ".ssh"
+
+    def __init__(self):
+        self.__current: SSHKey | None = None
+        self.__first: SSHKey | None = None
+        self.__last: SSHKey | None = None
+        self.__indices: range | None = None
+
+        # @TODO allow initialization with unpacked parameter or sequence/iterable argument
+
+    def __getitem__(self, key: int | slice) -> SSHKey | Never:
+        self.__current = self.__first
+
+        if self.__current is None:
+            raise KeyError
+
+        if isinstance(key, int):    
+            if int(self.__current) == key:
+                return self.__current
+            else:
+                while int(self.__current) != key:
+                    if self.__current is None:
+                        raise KeyError
+
+                    self.__current = next(self.__current)
+
+            result = self.__current
+        elif isinstance(key, slice):
+            step = key.step
+            sshkcoll = SSHKeyCollection()
+            if hasattr(key, "start"):
+                if getattr(key, "start") is None:
+                    start = 0
+                else:
+                    start = key.start
+            if hasattr(key, "stop"):
+                if getattr(key, "stop") is None:
+                    stop = len(self.__indices)
+                else:
+                    stop = key.stop
+            if hasattr(key, "step"):
+                if getattr(key, "step") is None:
+                    step = 1
+                else:
+                    step = key.step
+            indices = range(start, stop, step)
+            # test_coll = []
+            while int(self.__current) < stop:
+                if int(self.__current) < start:
+                    continue
+                elif int(self.__current) >= start:
+                    if int(self.__current) in indices:
+                        sshkcoll.append(self.__current)
+                        # test_coll.append(self.__current)
+                    else:
+                        continue
+
+                self.__current = next(self.__current)
+
+                if self.__current is None:
+                    break
+
+            # print(test_coll)
+            result = sshkcoll
+
+        return result
+
+    def __len__(self) -> int:
+        if self.__indices is None:
+            return 0
+
+        return len(self.__indices)
+
+    def pop(self, key: int = -1) -> Never:
+        self.__current = self.__first
+
+        if self.__current is None:
+            raise KeyError
+
+        if key == -1:
+            if self.__last is not None:
+                past = self.__last
+                self.__last._SSHKey__prev._SSHKey__next = None
+                self.__last = self.__last._SSHKey__prev
+
+                self.__current = self.__last
+            else:
+                past = self.__first
+                self.__first = None
+            return past
+        elif key <= -2:
+            raise NotImplementedError
+        else:
+            while int(self.__current) != key:
+                self.__current = next(self.__current)
+
+                if self.__current is None:
+                    raise KeyError
+            past = self.__current
+
+            count = self.__current._SSHKey__idx
+            prior = self.__current._SSHKey__prev
+            posterior = self.__current._SSHKey__next
+            posterior._SSHKey__prev = prior
+            posterior._SSHKey__prev._SSHKey__next = posterior
+            self.__current = posterior
+            while self.__current is not None:
+                self.__current._SSHKey__idx = count
+                self.__current = next(self.__current)
+                count += 1
+
+        return past
+
+    def remove(self) -> Never:
+        raise NotImplementedError
+
+    def append(self, *value: ExecutedPath | str) -> SSHKey:
+        if len(value) < 1 or len(value) > 2:
+            raise ValueError
+
+        value = tuple(map(lambda s: Path(s) if isinstance(s, str) else s, value))
+
+        ssh_key = SSHKey(*value)
+
+        if self.__first is None:
+            # print("branch1")
+            ssh_key._SSHKey__idx = 0
+            # print(ssh_key._SSHKey__idx)
+            ssh_key.update_status()
+            self.__indices = range(ssh_key._SSHKey__idx + 1)
+            self.__first = ssh_key
+
+            self.__current = self.__first
+        else:
+            # print("branch2")
+            if self.__last is not None:
+                # print("branch2.1")
+                ssh_key._SSHKey__idx = self.__last._SSHKey__idx + 1
+                # print(ssh_key._SSHKey__idx)
+                ssh_key.update_status()
+                self.__last._SSHKey__next = ssh_key
+                self.__last._SSHKey__next._SSHKey__prev = self.__last
+                self.__last = next(self.__last)
+            else:
+                # print("branch2.2")
+                ssh_key._SSHKey__idx = self.__first._SSHKey__idx + 1
+                # print(ssh_key._SSHKey__idx)
+                ssh_key.update_status()
+                self.__first._SSHKey__next = ssh_key
+                self.__first._SSHKey__next._SSHKey__prev = self.__first
+                self.__last = self.__first._SSHKey__next
+
+            self.__indices = range(ssh_key._SSHKey__idx + 1)
+            self.__current = self.__last
+
+        #print(self.__current)
+        return self.__current
+
+    def __setitem__(self, key: int | slice, *value: ExecutedPath | str) -> None | Never:
+        if len(value) < 1 or len(value) > 2:
+            raise ValueError
+
+        value = tuple(map(lambda s: Path(s) if isinstance(s, str) else s, value))
+
+        self.__current = self.__first
+
+        if self.__current is None:
+            raise KeyError
+        
+        if isinstance(key, int):
+            if int(self.__current) == key:
+                if self.__current() is None or len(self.__current()) < 1:
+                    self.__current(*value)
+            else:
+                if int(self.__current) == key:
+                    return self.__current(*value)
+
+                while int(self.__current) != key:
+                    if self.__current is None:
+                        raise KeyError
+
+                    self.__current = next(self.__current)
+
+                self.__current(*value)
+        elif isinstance(key, slice):
+            raise NotImplementedError
+
+    def __delitem__(self, key: int | slice) -> None | Never:
+        self.__current = self.__first
+
+        if self.__current is None:
+            raise KeyError
+
+        if isinstance(key, int):
+            if key == -1:
+                if self.__last is not None:
+                    self.__last._SSHKey__prev._SSHKey__next = None
+                    self.__last = self.__last._SSHKey__prev
+
+                    self.__current = self.__last
+                else:
+                    self.__first = None
+                return past
+            elif key <= -2:
+                raise NotImplementedError
+            else:
+                while int(self.__current) != key:
+                    self.__current = next(self.__current)
+
+                    if self.__current is None:
+                        raise KeyError
+
+                count = self.__current._SSHKey__idx
+                prior = self.__current._SSHKey__prev
+                posterior = self.__current._SSHKey__next
+                posterior._SSHKey__prev = prior
+                posterior._SSHKey__prev._SSHKey__next = posterior
+                self.__current = posterior
+                while self.__current is not None:
+                    self.__current._SSHKey__idx = count
+                    self.__current = next(self.__current)
+                    count += 1
+        elif isinstance(key, slice):
+            raise NotImplementedError
+
+    @property
+    def head(self) -> SSHKey | None:
+        return self.__first
+
+    @property
+    def tail(self) -> SSHKey | None:
+        if self.__last is None:
+            return self.__first
+
+        return self.__last
+
+    def __contains__(self, value:  ExecutedPath | str) -> bool:
+        self.__current = self.__first
+
+        if isinstance(value, ExecutedPath):
+            value = str(value)
+
+        is_contained = False
+        while self.__current is not None:
+            if str(self.__current._SSHKey__value) == value:
+                is_contained = True
+                break
+
+            self.__current = next(self.__current)
+
+        return is_contained
+
+    def __missing__(self, value: ExecutedPath | str) -> Never:
+        raise NotImplementedError
+
+    def __next__(self):
+        self.__current = next(self.__current)
+        if self.__current is not None:
+            return self.__current
+        else:
+            raise StopIteration
+
+    def __iter__(self) -> Self | Never:
+        self.__current = self.__first
+        # return self.__current
+        return self
+
+    def count(self, query: RegEx | str) -> int | Never:
+        raise NotImplementedError
+
+    def pull(self, query: RegEx | str = "*") -> None:
+        if isinstance(query, RegEx):
+            keypaths = self.__ssh_path.glob("*")
+
+            for p in keypaths:
+                if query.fullmatch(p.name):
+                    if not Match("(known_hosts|authorized_keys|config).*", p.name):
+                        self.append(p)
+                else:
+                    continue
+        else:
+            keypaths = self.__ssh_path.glob(query)
+
+            for p in keypaths:
+                if not Match("(known_hosts|authorized_keys|config).*", p.name):
+                    self.append(p)
+
+    def reverse(self) -> None | Never:
+        raise NotImplementedError
+
+    def sort(self, key: Callable = (lambda e: e), reverse: bool = False) -> None | Never:
+        raise NotImplementedError
+
+    def __str__(self) -> str:
+        prefix = "[("
+        postfix = "|]"
+
+        self.__current = self.__first
+
+        concat = lambda s: str(s) + ", "
+        content = str()
+        count = 0
+        while self.__current is not None:
+            content += str(count) + ">" + concat(self.__current)
+            self.__current = next(self.__current)
+            count += 1
+        content = content[0:len(content)-2]
+
+        return prefix + content + postfix
+
+    def index(self, item: str | ExecutedPath) -> Never:
+        raise NotImplementedError
+
+    def publish(self, category: SSHKeyType | str | None = SSHKeyType.pubkey, pref: int | None = None, datatype = list):
+        privkey = list()
+        pubkey = list()
+        self.__current = self.__first
+
+        # @TODO create conditional case that publishes all keys
+        if datatype == list:
+            while self.__current is not None:
+                # print(self.__current)
+                if self.__current.category == SSHKeyType.privkey.name.lower():
+                    privkey.append(self.__current._SSHKey__value)
+                elif self.__current.category == SSHKeyType.pubkey.name.lower():
+                    pubkey.append(self.__current._SSHKey__value)
+                elif self.__current.category == SSHKeyType.dual.name.lower():
+                    privkey.append(self.__current._SSHKey__value[0])
+                    pubkey.append(self.__current._SSHKey__value[1])
+                self.__current = next(self.__current)
+            # print("publish running...")
+
+            if pref is None:
+                preference = gamble(range(len(privkey)))
+            else:
+                preference = pref
+            
+            # print(category)
+            if category.name.lower() == SSHKeyType.pubkey.name.lower():
+                return pubkey
+            elif category.name.lower() == SSHKeyType.privkey.name.lower():
+                return (privkey, preference)
+            else:
+                return (privkey, pubkey, preference)
+        elif datatype == dict:
+            # @TODO have result var equal to instance of a yaml.YAMLObject class from parse module
+            raise NotImplementedError
+
+            return result
+
+    def __repr__(self) -> str:
+        return "%s()" % (self.__class__.__name__)

whereami.py

@@ -2,8 +2,61 @@
 Library of path constants to be used or referenced elsewhere.
 """
 
-from custtypes import ExecutedPath
+from custtypes import ExecutedPath, Roles, Scopes, AnsibleScopes, NodeType, UserName
 from pathlib import Path
+from configparser import ConfigParser as cfg
+from itertools import chain
+from typing import Callable
 
-USER_PATH: ExecutedPath = Path.home()
+def get_home(node: NodeType = NodeType.control, home: UserName | str | None = None) -> ExecutedPath:
+    if node == NodeType.control:
+        return Path.home()
+    else:
+        if home is None:
+            return Path("~")
+        else:
+            if isinstance(home, UserName):
+                return Path("/home") / home.name
+            else:
+                return Path(home)
+
+USER_PATH: Callable = get_home
 PROJ_ROOT: ExecutedPath = Path(__file__).parent.resolve()
+
+config = cfg()
+ANSIBLE_CONFIG = PROJ_ROOT / "ansible.cfg"
+
+if ANSIBLE_CONFIG.exists():
+    config.read(str(ANSIBLE_CONFIG))
+    role_candidates = config["defaults"]["roles_path"].split(":")
+    role_candidates = map(lambda s: PROJ_ROOT / s, role_candidates)
+    role_paths = list(filter(lambda p: p.exists(), role_candidates))
+
+    inv_candidates = config["defaults"]["inventory"].split(",")
+    inv_candidates = map(lambda s: PROJ_ROOT / s, inv_candidates)
+    inv_paths = list(filter(lambda p: p.exists(), inv_candidates))
+else:
+    role_paths = [PROJ_ROOT / "roles"]
+    inv_paths = PROJ_ROOT.glob("hosts*.*")
+PROJ_ROLES: list[ExecutedPath] = role_paths
+PROJ_INVENTORIES: list[ExecutedPath] = inv_paths
+
+proj_paths = map(lambda r: r.glob("**/files"), PROJ_ROLES)
+proj_paths = list(chain.from_iterable(proj_paths))
+template_paths = map(lambda r: r.glob("**/templates"), PROJ_ROLES)
+template_paths = list(chain.from_iterable(template_paths))
+APP_ROOTS = {
+    Roles.CONF.name.lower(): {
+        Scopes.SYS.name.lower(): Path("/etc"),
+        Scopes.USER.name.lower(): USER_PATH(NodeType.remote) / ".config",
+        Scopes.SHARED.name.lower(): Path("/usr/share"),
+        Scopes.PROJ.name.lower(): tuple(proj_paths + template_paths)
+    }
+}
+
+ANSIBLE_ROOTS = {
+    AnsibleScopes.GROUPVARS.name.lower(): PROJ_ROOT / "group_vars",
+    AnsibleScopes.HOSTVARS.name.lower(): PROJ_ROOT / "host_vars",
+    AnsibleScopes.INVENTORY.name.lower(): tuple(PROJ_INVENTORIES),
+    AnsibleScopes.ROLE.name.lower(): tuple(PROJ_ROLES)
+}