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

anodes.py

@@ -1,152 +1,75 @@
-"""
-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
+from custtypes import AnsibleRoles, AnsibleScopes, ExecutedPath
+from whereami import USER_PATH, ANSIBLE_CONFIG, ANSIBLE_ROOTS
+from configparser import ConfigParser as cfg
+from typing import Sequence
+from re import Pattern as RegEx
+from pathlib import PurePath
+from parse import Parser
 
+# @TODO below class should mostly work as a context manager
 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"),)
-        })
+    __parser = Parser()
 
-    def get_scope(self, scope: AnsibleScopes = AnsibleScopes.INVENTORY.name, index = 0):
-        return getattr(self, scope)[index]
+    def __init__(self):
+        self.__user_path = USER_PATH()
+        self.__config = cfg()
 
-    @property
-    def home(self) -> ExecutedPath:
-        return self.__user_path
+        if ANSIBLE_CONFIG.exists():
+            self.__config.read(str(ANSIBLE_CONFIG))
+        else:
+            raise Exception
 
-    @property
-    def root(self) -> ExecutedPath:
-        return self.__proj_root
+        self.__data = None
+        self.__filepath: ExecutedPath | None = None
+        self.__file = None
 
-    @property
-    def sys_confs(self) -> ExecutedPath:
-        return self.__conf_paths
+    def __enter__(self):
+        self.__file = open(str(self.__filepath), "r+")
+        self.__data = self.__parser.load(self.__filepath)
+        return self.__data
 
-    @property
-    def sys_data(self) -> ExecutedPath:
-        return self.__data_paths
+    def __exit__(self, exc_type, exc_value, exc_traceback):
+        result = self.__parser.dump(self.__data)
 
-# userSSHParams = Dict("userSSHParams", {
-#     "username": Required[str],
-#     "paths": Apps,
-#     "keys": dict,
-#     "password": Required[str],
-#     "fate": Literal["disposal", "retention"]
-# }, total=False)
+        if isinstance(result, str):
+            self.__file.write(result)
+        else:
+            result.write(self.__file)
 
-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)
+        self.__file.close()
 
-class RemoteNode:
-    # __user_path =
-    _fqdn: str | None = None
+    def __call__(self, scope: AnsibleScopes = AnsibleScopes.INVENTORY, pick: int | str | RegEx | AnsibleRoles = 0, filepath: ExecutedPath | str | None = None):
+        if scope is None:
+            raise NotImplementedError
+        else:
+            paths = ANSIBLE_ROOTS[scope.name.lower()]
 
-    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
+            if isinstance(paths, Sequence):
+                path = None
+                if isinstance(pick, int):
+                    path = paths[pick]
+                elif isinstance(pick, str):
+                    path = tuple(filter(lambda p: str(p) == pick or pick in str(p), paths))
+                    if len(path) > 0:
+                        path = path[0]
+                elif isinstance(pick, RegEx):
+                    path = tuple(filter(lambda p: pick.search(str(p)), paths))
+                    if len(path) > 0:
+                        path = path[0]
+                else:
+                    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]
 
-        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)
+        if path is None:
+            raise KeyError
 
-        self._fqdn = name
-        self.root["software"]._fqdn = name
+        if isinstance(filepath, ExecutedPath):
+            filepath = str(filepath)
 
-        # 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)
+        if path.is_dir():
+            self.__filepath = path / filepath
+        elif path.is_file():
+            self.__filepath = path
 
+        return __enter__

confctx.py

@@ -1,103 +0,0 @@
-from pathlib import Path, PurePath
-from whereami import PROJ_ROOT
-from custtypes import ExecutedPath, IdlePath, VirtualPrivateServers, AnsibleScopes
-from anodes import RemoteNode, ControlNode, _userSSHSubParams
-# from ansible_vault import Vault
-import yaml as yams
-import re
-from typing import Literal
-
-# @TODO for 'Config' class, write methods to pull and push from Ansible YAML files
-# @NOTE https://docs.python.org/3/library/configparser.html#quick-start
-class Config:
-    path: ExecutedPath = PROJ_ROOT / "config.ini"
-    __controller = ControlNode()
-    setattr(__controller, AnsibleScopes.INTERNAL.name, {
-        "vars": (PurePath(str(path)),)
-    })
-
-    def __init__(self, filepath: str = "config.ini", scope: AnsibleScopes = AnsibleScopes.INTERNAL.name, index: int = 0, mode: str = "r+"):
-        self.scope = scope
-
-        parent_dir = self.__controller.get_scope(scope, index)
-        filepath = Path(parent_dir) / filepath
-        filename: str = filepath.name
-
-        self.parse_method = "YML"
-        if "." in filename:
-            filename_arr = filename.split(".")
-            basename: str = filename_arr[0]
-            ext: str = filename_arr[len(filename_arr) - 1]
-            
-        if re.match(r"toml|ini", ext):
-            self.parse_method: str = ext.upper()
-
-        self.filepath: ExecutedPath = filepath
-        self.mode = mode
-        self.model = None
-        self.file = None
-        self.__remote: RemoteNode | None = None
-
-    def __enter__(self):
-        if not self.filepath.exists():
-            self.__remote = RemoteNode(self.__controller)
-            return self.__remote
-
-        self.file = open(str(self.filepath), self.mode)
-        model = self.file.read()
-
-        if self.parse_method == "INI":
-            raise NotImplementedError
-        elif self.parse_method == "TOML":
-            raise NotImplementedError
-        else:
-            self.model = yams.load(model)
-
-        if self.scope == AnsibleScopes.GROUPVARS.name or self.scope == AnsibleScopes.HOSTVARS.name:
-            api_key: str = model["vps_service"]["api_key"]
-            vault = Vault(model["vps_service"]["password"])
-            password: str = vault.load(model["vps_service"]["password"])
-            fqdn: str = model["fqdn"]
-            service: VirtualPrivateServers = model["vps_service"]["type"]
-            region: Literal["us-east"] = model["vps_service"]["region"]
-            keywords: list[str] = model["keywords"]
-            self.__remote = RemoteNode(self.__controller, api_key, password, fqdn, service, region, keywords)
-            # @TODO add portion wherein SSH keys from 'model' are made to be present in 'self.__remote'
-            self.__remote.import_keys("*")
-            keyfiles = self.__remote.show_keys("available", list)
-            keyfiles_contents = set(map(lambda p: p.read_text(), keyfiles))
-            keyfiles_pub = list(set(model["vps_service"]["ssh_authorized_keys"]) - keyfiles_contents)
-            self.__remote.ssh.keys["available"] = tuple(keyfiles_pub)
-            self.__remote.ssh.keys["selected"] = keyfiles_pub
-            self.__remote.ssh.keys["authorized"] += keyfiles_pub
-            keyfiles_paths = set(map(lambda p: str(p), keyfiles))
-            keyfiles_priv = list(set(model["vps_service"]["ssh_private_key_paths"]) - keyfiles_paths)
-            self.__remote.ssh.keys["available"] = (*self.__remote.ssh.keys["available"],*keyfiles_priv)
-            self.__remote.ssh.keys["selected"] += keyfiles_priv
-            self.__remote.ssh.keys["used"] += keyfiles_priv
-            self.__remote.ssh.fate: Literal["disposal", "retention"] = model["vps_service"]["root_fate"]
-
-            return self.__remote
-        else:
-            raise ValueError
-
-    def __exit__(self):        
-        if self.scope == AnsibleScopes.GROUPVARS.name or self.scope == AnsibleScopes.HOSTVARS.name:
-            ansible_chunk = self.__remote.itemize()
-            ansible_chunk["vps_service"]["ssh_motd_script_basenames"] = self.model["ssh_motd_script_basenames"]
-            ansible_chunk["vps_service"]["ssh_private_key_path_pref"] = self.model["ssh_private_key_path_pref"]
-            del self.model["fqdn"]
-            del self.model["vps_service"]
-            del self.model["keywords"]
-        else:
-            raise ValueError
-
-        if self.parse_method == "INI":
-            raise NotImplementedError
-        elif self.parse_method == "TOML":
-            raise NotImplementedError
-        else:
-            file_model = yams.dump(self.model | ansible_chunk)
-        
-        self.file.write(file_model)
-        self.file.close()

custtypes.py

@@ -2,19 +2,81 @@
 Library of custom type hints.
 """
 
-from typing import TypeAlias as Neotype
-from pathlib import PurePosixPath, PureWindowsPath, PosixPath, WindowsPath
-from enum import Enum
+from typing import TypeAlias as Neotype, TypedDict as Dict
+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 = PosixPath | WindowsPath
 IdlePath: Neotype = PurePosixPath | PureWindowsPath
+File: Neotype = TextIOBase | BufferedIOBase | RawIOBase
 
-class VirtualPrivateServers(Enum):
-    Linode = 0
+class RootFate(StrEnum):
+    disposal = auto()
+    retention = auto()
 
-class AnsibleScopes(Enum):
-    INTERNAL = 0
-    INVENTORY = 1
-    GROUPVARS = 2
-    HOSTVARS = 3
-    ROLE = 4
+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

@@ -3,17 +3,63 @@ Library for the CLI commands and the related classes and functions
 """
 
 import click as cli
-
-domain_pattern = r'^(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'
-# @TODO create regex pattern for matching IP addresses
-# ip_pattern = r''
+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
 
 @cli.group()
-@cli.option("-d", "--debug", type=bool, is_flag=True, default=True, help="Use debugging mode")
+@cli.option("-d", "--debug", type=bool, is_flag=True, default=False, help="Use debugging mode")
 @cli.pass_context
 def skansible(ctx, debug):
     ctx.ensure_object(dict)
-    ctx.obj["DEBUG"] = True
+    ctx.obj["DEBUG"] = debug
+
+@skansible.command()
+@cli.argument("api_key")
+@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]
+
+    inv_vars = []
+    for h in host:
+        inv_vars += list(ANSIBLE_ROOTS[AnsibleScopes.HOSTVARS.name.lower()].glob(h)) + list(ANSIBLE_ROOTS[AnsibleScopes.GROUPVARS.name.lower()].glob(h))
+
+    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(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

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

softman.py

@@ -1,182 +1,157 @@
-"""
-Library of classes modeling software and software-related 
-data as represented in or used by Ansible.
-"""
+from typing import Self, Never, Callable, Sequence
+from custtypes import Roles, Scopes, PathCollection, PathRoles
+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
-from typing import TypedDict as Dict
-from typing import Never, Union, Callable
-from custtypes import ExecutedPath, IdlePath
-from enum import Enum
-from pathlib import Path, PurePath
-from whereami import USER_PATH, PROJ_ROOT
-from collections.abc import Sequence
+class App:
+    def __init__(self, name: Software, role: SoftwareRoles = SoftwareRoles.client, paths: PathRoles | None = None):
+        self.__name = name.name.lower().replace("_", "-")
+        # @TODO create dict type hint for below data struct
+        self.alt_names = dict()
+        self.alt_names[PacMans.APT.name.lower()] = self.__name
+        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):
-    PERSONAL = 0
-    LOCAL = 1
-    GLOBAL = 2
+    @property
+    def data_paths(self):
+        if hasattr(self, "_" + Roles.DATA.name.lower()):
+            return self._data
+        else:
+            raise Exception
 
-class SoftPathGroup(Enum):
-    CONFIG = 0
-    DATA = 1
-    MEM = 2
-    EXE = 3
+    @property
+    def exec_paths(self):
+        if hasattr(self, "_" + Roles.EXE.name.lower()):
+            return self._exe
+        else:
+            raise Exception
 
-_SubAppParams = Dict("_SubAppParams", {
-    SoftScope.PERSONAL.name: IdlePath | list[IdlePath],
-    SoftScope.LOCAL.name: IdlePath | list[IdlePath],
-    SoftScope.GLOBAL.name: IdlePath | list[IdlePath]
-}, total=False)
-AppParams = Dict("AppParams", {
-    SoftPathGroup.CONFIG.name: _SubAppParams,
-    SoftPathGroup.DATA.name: _SubAppParams,
-    SoftPathGroup.MEM.name: _SubAppParams,
-    SoftPathGroup.EXE.name: _SubAppParams
-}, total=False)
+    @property
+    def mem_paths(self):
+        if hasattr(self, "_" + Roles.MEM.name.lower()):
+            return self._mem
+        else:
+            raise Exception
 
-def __AppsInit(self, CONFIG = None, DATA = None, MEM = None, EXE = None):
-    self.CONFIG = CONFIG
-    self.DATA = DATA
-    self.MEM = MEM
-    self.EXE = EXE
-__app_input = {
-    SoftPathGroup.CONFIG.name: {
-        SoftScope.PERSONAL.name: [],
-        SoftScope.LOCAL.name: [],
-        SoftScope.GLOBAL.name: []
-    },
-    SoftPathGroup.DATA.name: {
-        SoftScope.PERSONAL.name: [],
-        SoftScope.LOCAL.name: [],
-        SoftScope.GLOBAL.name: []
-    },
-    SoftPathGroup.MEM.name: {
-        SoftScope.PERSONAL.name: [],
-        SoftScope.LOCAL.name: [],
-        SoftScope.GLOBAL.name: []
-    },
-    SoftPathGroup.EXE.name: {
-        SoftScope.PERSONAL.name: [],
-        SoftScope.LOCAL.name: [],
-        SoftScope.GLOBAL.name: []
-    },
-    "__init__": __AppsInit
-}
-Apps = type("Apps", (), __app_input)
+    def get_paths(self, role: Roles = Roles.CONF):
+        if hasattr(self, "_" + role.name.lower()):
+            return getattr(self, "_" + role.name.lower())
+        else:
+            raise Exception
 
-# @TODO continue adding magic methods to below class
-# @NOTE https://rszalski.github.io/magicmethods/#sequence
-class Software (Sequence):
-    __user_path: ExecutedPath = USER_PATH
+    def append(self, datatype: Roles = Roles.CONF, scope: Scopes | None = None, path: IdlePath | ExecutedPath | str | PathCollection | None = None):
+        if path is None:
+            raise TypeError
 
-    def __init__(self):
-        self._fqdn: str | None = None
-
-    # @TODO fix NameError for 'Software' in parameter type check
-    def append(self, name: str, **kwpaths: _SubAppParams) -> AppParams:
-        keyword_args: AppParams = kwpaths
+        datatype = datatype.name.lower()
         
-        app = Apps(**keyword_args)
-        setattr(self, name, app)
-        return app
-
-    def __getitem__(self, key: str) -> AppParams | Never:
-        if hasattr(self, key):
-            app: Apps = getattr(self, key)
+        if hasattr(self, "_" + datatype):
+            paths = getattr(self, "_" + datatype)
         else:
-            raise KeyError
-        return app
+            setattr(self, "_" + datatype, dict())
+            paths = getattr(self, "_" + datatype)
 
-    def __setitem__(self, key: tuple[str, SoftPathGroup], **value: IdlePath | list[IdlePath]) -> None | Never:
-        if len(value) < 1 or len(value) > 3:
-            raise ValueError
+        if scope is not None:
+            if isinstance(path, str):
+                path: ExecutedPath = Path(path)
 
-        app_params: _SubAppParams = value
+            if scope.name.lower() not in paths:
+                paths[scope.name.lower()] = []
 
-        if hasattr(self, key[0]):
-            app: Apps = getattr(self, key[0])
-
-            if hasattr(app, key[1]):
-                app_child: _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)
+            paths[scope.name.lower()].append(path)
         else:
-            raise KeyError
+            paths: PathCollection = path
 
-    def __delitem__(self, key: tuple[str | SoftPathGroup]) -> None | Never:
-        if len(key) < 1 or len(key) > 3:
-            raise KeyError
+        setattr(self, "_" + datatype, paths)
 
-        if not hasattr(self, key[0]):
-            raise KeyError
+    def inherit(self, other):
+        if other._App__children is None:
+            other._App__children = tuple()
+        
+        if self not in other._App__children:
+            other.adopt(self)
 
-        if len(key) == 1:
-            delattr(self, key[0])
-        elif len(key) > 1:
-            app: Apps = getattr(self, key[0])
-            delattr(app, key[1])
-            setattr(self, key[0], app)
+        self.__parents = (*self.__parents, other)
 
-    def show(self, contents: bool = False) -> tuple[str]:
-        apps: tuple[str] | tuple[Apps] = tuple(
-            filter(
-                lambda a: isinstance(getattr(self, a), Apps),
-                dir(self)
-            )
-        )
+    def adopt(self, other: Self):
+        if other._App__parents is None:
+            other._App__parents = tuple()
 
-        if contents:
-            apps = tuple(
-                map(
-                    lambda a: getattr(self, a),
-                    apps
-                )
-            )
+        if self not in other._App__parent:
+            other.inherit(self)
 
-        return apps
+        self.__children = (*self.__children, other)
 
-    def __len__(self) -> int:
-        apps: tuple[str] = tuple(
-            filter(
-                lambda a: isinstance(getattr(self, a), Apps),
-                dir(self)
-            )
-        )
-        return len(apps)
+    def __enter__(self) -> dict | Sequence:
+        self.__content = self.__current_filepath.read_text()
+        return self.__content
 
-    def pop(self) -> Never:
+    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 remove(self) -> Never:
-        raise NotImplementedError
+    def declare(self, **kwargs):
+        for k, v in kwargs.items():
+            setattr(self, k, v)
 
-    def __contains__(self) -> Never:
-        raise NotImplementedError
-
-    def count(self) -> Never:
-        return NotImplementedError
-
-    def __missing__(self) -> Never:
-        raise NotImplementedError
-
-    def __iter__(self) -> Never:
-        raise NotImplementedError
-
-    def reverse(self) -> Never:
-        raise NotImplementedError
-
-    def sort(self, key: Callable = (lambda e: e), reverse: bool = False) -> Never:
-        raise NotImplementedError
-
-class Softs(Enum):
-    ssh = 0
+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
+    }
+}
+sshd = App(Software.openssh_server, SoftwareRoles.server, sshd_paths)

sshkey.py

@@ -1,27 +1,19 @@
 from re import Pattern as RegEx
 from re import fullmatch as Match
 from pathlib import Path, PurePath
-from custtypes import ExecutedPath, IdlePath, VirtualPrivateServers, AnsibleScopes
-from enum import Enum
-from softman import Apps
+from custtypes import ExecutedPath, IdlePath
+from enum import StrEnum, auto
 from random import choice as gamble
-from collections.abc import Sequence
-from typing import Never, Union, Self, Callable, Required, Literal
-from typing import TypedDict as Dict
-from glob import glob as globbify
+# from collections.abc import Sequence, Iterable
+from typing import Never, Self, Callable, Iterable, Sequence
 from whereami import USER_PATH
-from softman import Softs
+from itertools import chain
 # import os
 
-class RootFate(Enum):
-    disposal = 0
-    retention = 1
-
-
-class SSHKeyType(Enum):
-    pubkey = 0
-    privkey = 1
-    dual = 2
+class SSHKeyType(StrEnum):
+    pubkey = auto()
+    privkey = auto()
+    dual = auto()
 
 
 # @TODO create unit tests for below class
@@ -40,18 +32,44 @@ class SSHKey:
         if len(path) < 2:
             self.__value: ExecutedPath | tuple[ExecutedPath] = path[0]
         else:
-            self.category = SSHKeyType.dual.name
+            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:
-        key_basename = Path(str(self.__value)).name
+        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 "SSHKey(" + str(self.__value) + ")"
+        return "%s(%r)" % (self.__class__.__name__, self.__value)
 
     def __nonzero__(self) -> bool:
         return True
@@ -116,41 +134,75 @@ class SSHKey:
 
         return self
 
-    def __add__(self, other: Self | ExecutedPath | str):
-        if isinstance(self.__value, tuple):
-            raise ValueError
-
+    def __add__(self, other: Self | ExecutedPath | str) -> Self:
         if isinstance(other, (str, ExecutedPath)):
-            result = self.update(self.__value, other)
+            if isinstance(self.__value, tuple):
+                self.update(*self.__value, other)
+            else:
+                self.update(*self.__value, other)
         else:
-            if isinstance(other.__SSHKey__value, tuple):
-                raise ValueError
+            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)
 
-            result = self.update(self.__value, other._SSHKey__value)
+        self.update_status()
 
-        return result
+        return self
 
-    def __radd__(self, other: Self | ExecutedPath | str):
+    def __radd__(self, other: Self | ExecutedPath | str) -> Self:
         if isinstance(self.__value, tuple):
-            raise ValueError
-
-        if isinstance(other, (str, ExecutedPath)):
-            result = self.update(other, self.__value)
+            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.__SSHKey__value, tuple):
-                raise ValueError
+            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)
 
-            result = self.update(other._SSHKey__value, self.__value)
+        other.update_status()
 
-        return result
+        return self
 
     # @TODO write following 2 subtraction algorithms using 'set' data type conversion and methods
-    def __sub__(self, other: Self | ExecutedPath | str):
+    def __sub__(self, other: Self | ExecutedPath | str) -> Never:
         raise NotImplementedError
 
-    def __rsub__(self, other: Self | ExecutedPath | str):
+    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)
@@ -215,29 +267,6 @@ class SSHKey:
 
         return result
 
-    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
-            elif pubkey_present or privkey_present:
-                if pubkey_present:
-                    self.category = SSHKeyType.pubkey.name
-                if privkey_present:
-                    self.category = SSHKeyType.privkey.name
-        elif isinstance(self.__value, ExecutedPath):
-            if "-----BEGIN OPENSSH PRIVATE KEY-----" in self.__value.read_text():
-                self.category = SSHKeyType.privkey.name
-            else:
-                self.category = SSHKeyType.pubkey.name
-
     @property
     def status(self) -> str:
         self.update_status()
@@ -253,6 +282,8 @@ class SSHKey:
             result = self
         return result
 
+
+
 def prev(arg):
     if isinstance(arg, SSHKey):
         return arg._SSHKey__prev__()
@@ -274,7 +305,7 @@ def stream_unequal(s1, s2) -> bool | Never:
 
 # @TODO create unit tests for below class
 class SSHKeyCollection(Sequence):
-    __user_path: ExecutedPath = USER_PATH
+    __user_path: ExecutedPath = USER_PATH()
     __ssh_path: ExecutedPath = __user_path / ".ssh"
 
     def __init__(self):
@@ -283,7 +314,8 @@ class SSHKeyCollection(Sequence):
         self.__last: SSHKey | None = None
         self.__indices: range | None = None
 
-    # @TODO have other item magic methods mimic this one for slicing purposes
+        # @TODO allow initialization with unpacked parameter or sequence/iterable argument
+
     def __getitem__(self, key: int | slice) -> SSHKey | Never:
         self.__current = self.__first
 
@@ -511,6 +543,8 @@ class SSHKeyCollection(Sequence):
         return self.__last
 
     def __contains__(self, value:  ExecutedPath | str) -> bool:
+        self.__current = self.__first
+
         if isinstance(value, ExecutedPath):
             value = str(value)
 
@@ -531,7 +565,8 @@ class SSHKeyCollection(Sequence):
         self.__current = next(self.__current)
         if self.__current is not None:
             return self.__current
-        raise StopIteration
+        else:
+            raise StopIteration
 
     def __iter__(self) -> Self | Never:
         self.__current = self.__first
@@ -541,7 +576,6 @@ class SSHKeyCollection(Sequence):
     def count(self, query: RegEx | str) -> int | Never:
         raise NotImplementedError
 
-    # @TODO make sure to implement below method
     def pull(self, query: RegEx | str = "*") -> None:
         if isinstance(query, RegEx):
             keypaths = self.__ssh_path.glob("*")
@@ -571,7 +605,7 @@ class SSHKeyCollection(Sequence):
 
         self.__current = self.__first
 
-        concat = lambda s: str(s)[1:] + ", "
+        concat = lambda s: str(s) + ", "
         content = str()
         count = 0
         while self.__current is not None:
@@ -582,48 +616,45 @@ class SSHKeyCollection(Sequence):
 
         return prefix + content + postfix
 
-    def publish(self, pref: int | None = None, datatype = dict):
+    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:
-                if self.__current.category == SSHKeyType.privkey.name:
-                    privkey.append(str(self.__current._SSHKey__value))
-                elif self.__current.category == SSHKeyType.pubkey.name:
-                    pubkey.append(self.__current._SSHKey__value.read_text())
-                elif self.__current.category == SSHKeyType.dual.name:
-                    privkey.append(str(self.__current._SSHKey__value[0]))
-                    pubkey.append(self.__current._SSHKey__value[1].read_text())
-                self.__current = next(self.__first)
+                # 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...")
 
-            return (privkey, pubkey, gamble(range(len(privkey))))
-        elif datatype == dict:
-            result = dict()
+            if pref is None:
+                preference = gamble(range(len(privkey)))
+            else:
+                preference = pref
             
-            while self.__current is not None:
-                if self.__current.category == SSHKeyType.privkey.name:
-                    privkey.append(str(self.__current._SSHKey__value))
-                elif self.__current.category == SSHKeyType.pubkey.name:
-                    pubkey.append(self.__current._SSHKey__value.read_text())
-                elif self.__current.category == SSHKeyType.dual.name:
-                    privkey.append(str(self.__current._SSHKey__value[0]))
-                    pubkey.append(self.__current._SSHKey__value[1].read_text())
-                self.__current = next(self.__first)
-
-            result["ssh_authorized_keys"]: list[str] = pubkey
-            result["ssh_private_key_paths"]: list[str] = privkey
-            result["ssh_private_key_path_pref"]: int = pref if pref is not None else gamble(range(len(privkey)))
+            # 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
 
-# @TODO maybe move to separate module for classes for handling users and groups
-class UserSSH:
-    def __init__(self, username: str = "root", paths: Apps | None = None, keys: dict = dict(), password: str = "password123", fate: RootFate = RootFate.disposal.name):
-        self.username = username
-        self.paths = paths
-        self.keys = keys
-        self.password = password
-        self.fate = fate
-
+    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()
-PROJ_ROOT: ExecutedPath = Path(__file__).parent.resolve()
+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)
+}