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

@@ -0,0 +1,75 @@
+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:
+    __parser = Parser()
+
+    def __init__(self):
+        self.__user_path = USER_PATH()
+        self.__config = cfg()
+
+        if ANSIBLE_CONFIG.exists():
+            self.__config.read(str(ANSIBLE_CONFIG))
+        else:
+            raise Exception
+
+        self.__data = None
+        self.__filepath: ExecutedPath | None = None
+        self.__file = None
+
+    def __enter__(self):
+        self.__file = open(str(self.__filepath), "r+")
+        self.__data = self.__parser.load(self.__filepath)
+        return self.__data
+
+    def __exit__(self, exc_type, exc_value, exc_traceback):
+        result = self.__parser.dump(self.__data)
+
+        if isinstance(result, str):
+            self.__file.write(result)
+        else:
+            result.write(self.__file)
+
+        self.__file.close()
+
+    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()]
+
+            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]
+
+        if path is None:
+            raise KeyError
+
+        if isinstance(filepath, ExecutedPath):
+            filepath = str(filepath)
+
+        if path.is_dir():
+            self.__filepath = path / filepath
+        elif path.is_file():
+            self.__filepath = path
+
+        return __enter__

custtypes.py

@@ -7,9 +7,11 @@ 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 RootFate(StrEnum):
     disposal = auto()
@@ -56,8 +58,8 @@ class GroupName(StrEnum):
     sudo = auto()
 
 class Software(StrEnum):
-    ssh = auto()
-    sshd = auto()
+    openssh_client = auto()
+    openssh_server = auto()
 
 class SoftwareRoles(StrEnum):
     client = auto()

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

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,18 +1,18 @@
-from typing import TypedDict as Dict
-from typing import Self, Literal, Never
-from collections.abc import Sequence
-from custtypes import Roles, Scopes, PathCollection, PathRoles, Software, SoftwareRoles, ExecutedPath, PacMans, UserName, GroupName, IdlePath, ExecutedPath
-from pathlib import Path, PurePath
+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
 
 class App:
     def __init__(self, name: Software, role: SoftwareRoles = SoftwareRoles.client, paths: PathRoles | None = None):
-        self.__name = name
+        self.__name = name.name.lower().replace("_", "-")
         # @TODO create dict type hint for below data struct
         self.alt_names = dict()
-        self.alt_names[PacMans.APT] = self.__name
-        self.role = role
+        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()])
@@ -20,6 +20,8 @@ class App:
                 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
@@ -47,6 +49,19 @@ class App:
         else:
             raise Exception
 
+    @property
+    def mem_paths(self):
+        if hasattr(self, "_" + Roles.MEM.name.lower()):
+            return self._mem
+        else:
+            raise Exception
+
+    def get_paths(self, role: Roles = Roles.CONF):
+        if hasattr(self, "_" + role.name.lower()):
+            return getattr(self, "_" + role.name.lower())
+        else:
+            raise Exception
+
     def append(self, datatype: Roles = Roles.CONF, scope: Scopes | None = None, path: IdlePath | ExecutedPath | str | PathCollection | None = None):
         if path is None:
             raise TypeError
@@ -57,6 +72,7 @@ class App:
             paths = getattr(self, "_" + datatype)
         else:
             setattr(self, "_" + datatype, dict())
+            paths = getattr(self, "_" + datatype)
 
         if scope is not None:
             if isinstance(path, str):
@@ -99,11 +115,13 @@ class App:
         # del txt
         self.__file.close()
 
-    def __call__(self, path = str, mode = "r+", scope: Scopes = Scopes.PROJ, index: int = 0) -> None:
-        if not hasattr(self, "_" + Roles.CONF.name.lower()):
+    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 = self._conf[scope.name.lower()]
+        conf_coll = config[scope.name.lower()]
 
         if isinstance(conf_coll, Sequence):
             conf_coll = conf_coll[index]
@@ -116,58 +134,24 @@ class App:
         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(): [
-            PROJ_ROLES / "bootstrap" / "files" / "sshd_config.d",
-            PROJ_ROLES / "bootstrap" / "templates" / "sshd_config.d"
-        ],
+        Scopes.PROJ.name.lower(): sshd_proj_files + sshd_proj_templates
     }
 }
-sshd = App(Software.sshd, SoftwareRoles.server, sshd_paths)
-
-class Group:
-    def __init__(self, group_name: GroupName = GroupName.sudo, gid: int = 27):
-        self.group_name = group_name
-        self.id = gid
-        self.category: Literal["system", "regular"] = "system"
-
-class User:
-    def __init__(self, username: UserName = UserName.root.name, password: str = "test", services: list = [Software.sshd.name.lower()], uid: int = 0):
-        self.exists = True
-        self.username = username
-        self.id = uid
-        self.password = password
-        self.services: tuple = tuple(services)
-        self.shell = "/bin/bash"
-        self.home = "/"
-        self.category: Literal["system", "regular"] = "regular"
-        group = Group(username, self.id)
-        self.primary_group = group
-        self.supp_groups = None
-
-        if self.supp_groups is None:
-            self.is_admin = True
-        elif isinstance(self.supp_groups, Sequence) and GroupName.sudo in self.supp_groups:
-            self.is_admin = True
-        else:
-            self.is_admin = False
-
-        ssh_keys = SSHKeyCollection()
-        ssh_keys.pull()
-        self.ssh_keys = ssh_keys
-        pubkeys = ssh_keys.publish(SSHKeyType.pubkey.name.lower(), datatype=list)
-        self.__auth_keys: list[str] = list(map(lambda k: k.read_text(), pubkeys))
-        privkeys = ssh_keys.publish(SSHKeyType.privkey.name.lower(), datatype=list)
-        self.__priv_keys: list[str] = list(map(lambda k: str(k), privkeys[0]))
-        self.__priv_key_pref: int = privkeys[1]
-        self.__apps = (sshd,)
-
-    def add_keypair(self, private_key, public_key):
-        raise NotImplementedError
-
+sshd = App(Software.openssh_server, SoftwareRoles.server, sshd_paths)

sshkey.py

@@ -4,9 +4,10 @@ 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
-from typing import Never, Self, Callable
+# 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):
@@ -37,12 +38,38 @@ class SSHKey:
     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
@@ -107,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)
@@ -206,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.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()
-
     @property
     def status(self) -> str:
         self.update_status()
@@ -244,6 +282,8 @@ class SSHKey:
             result = self
         return result
 
+
+
 def prev(arg):
     if isinstance(arg, SSHKey):
         return arg._SSHKey__prev__()
@@ -265,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):
@@ -274,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
 
@@ -502,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)
 
@@ -522,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
@@ -532,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("*")
@@ -562,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:
@@ -573,13 +616,18 @@ class SSHKeyCollection(Sequence):
 
         return prefix + content + postfix
 
-    def publish(self, category: SSHKeyType = SSHKeyType.pubkey.name.lower(), 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:
+                # 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():
@@ -587,42 +635,26 @@ class SSHKeyCollection(Sequence):
                 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.__first)
+                self.__current = next(self.__current)
+            # print("publish running...")
 
             if pref is None:
                 preference = gamble(range(len(privkey)))
             else:
                 preference = pref
             
-            if category == SSHKeyType.pubkey.name.lower():
+            # print(category)
+            if category.name.lower() == SSHKeyType.pubkey.name.lower():
                 return pubkey
-            elif category == SSHKeyType.privkey.name.lower():
+            elif category.name.lower() == SSHKeyType.privkey.name.lower():
                 return (privkey, preference)
             else:
                 return (privkey, pubkey, preference)
         elif datatype == dict:
-            result = dict()
-            
-            while self.__current is not None:
-                if self.__current.category == SSHKeyType.privkey.name.lower():
-                    privkey.append(str(self.__current._SSHKey__value))
-                elif self.__current.category == SSHKeyType.pubkey.name.lower():
-                    pubkey.append(self.__current._SSHKey__value.read_text())
-                elif self.__current.category == SSHKeyType.dual.name.lower():
-                    privkey.append(str(self.__current._SSHKey__value[0]))
-                    pubkey.append(self.__current._SSHKey__value[1].read_text())
-                self.__current = next(self.__first)
-
-            if category == SSHKeyType.pubkey.name.lower():
-                result["ssh_authorized_keys"]: list[str] = pubkey
-
-            if category == SSHKeyType.privkey.name.lower():
-                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)))
-
-            if category == SSHKeyType.dual.name.lower():
-                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)))
+            # @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,25 +2,61 @@
 Library of path constants to be used or referenced elsewhere.
 """
 
-from custtypes import ExecutedPath, Roles, Scopes, AnsibleScopes
+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()
-PROJ_ROLES: ExecutedPath = PROJ_ROOT / "roles"
 
+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 / ".config",
+        Scopes.USER.name.lower(): USER_PATH(NodeType.remote) / ".config",
         Scopes.SHARED.name.lower(): Path("/usr/share"),
-        Scopes.PROJ.name.lower(): tuple(list(PROJ_ROLES.glob("**/files")) + list(PROJ_ROLES.glob("**/templates"))) 
+        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(): None,
-    AnsibleScopes.ROLE.name.lower(): None
+    AnsibleScopes.INVENTORY.name.lower(): tuple(PROJ_INVENTORIES),
+    AnsibleScopes.ROLE.name.lower(): tuple(PROJ_ROLES)
 }