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

.bin/ansible_aliases

@@ -0,0 +1,6 @@
+#!/bin/bash
+set -euo pipefail
+
+alias ansible-galaxy="/usr/bin/ansible-galaxy"
+alias ansible-vault="/usr/bin/ansible-vault"
+alias ansible-playbook="/usr/bin/ansible-playbook"

.bin/skansible.sh

@@ -0,0 +1,187 @@
+#!/bin/bash
+set -euo pipefail
+
+SKATO_ANSIBLE_ROOT=$(dirname "$0")
+SKATO_ANSIBLE_ROOT=$(dirname "$SKATO_ANSIBLE_ROOT")
+export SKATO_ANSIBLE_ROOT
+printf "root=%s\n" "$SKATO_ANSIBLE_ROOT" > "./config" # INI format
+export SKATO_BOOTSTRAP_ROLE="${SKATO_ANSIBLE_ROOT}/roles/bootstrap"
+export SKANSIBLE_SECRETS="${SKATO_ANSIBLE_ROOT}/.secrets"
+
+if [[ -f "./ansible_aliases" ]]; then
+    source ./ansible_aliases
+fi
+
+# Relative directory paths for role templates/files
+export SKANSIBLE_ARIA="aria2"
+export SKANSIBLE_PROFTPD="proftpd"
+export SKANSIBLE_PROFTPD_CONFS="${SKANSIBLE_PROFTPD}/conf.d"
+# @NOTE below 4 filepaths have filenames that must correspond to 
+# the filenames in role ProFTPd templates'/files' Display settings 
+export SKANSIBLE_PROFTPD_CONFS_WELCOME="${SKANSIBLE_PROFTPD}/conf.d/WELCOME.txt"
+export SKANSIBLE_PROFTPD_CONFS_BANNER="${SKANSIBLE_PROFTPD}/conf.d/BANNER.txt"
+export SKANSIBLE_PROFTPD_CONFS_SUCCESS="${SKANSIBLE_PROFTPD}/conf.d/SUCCESS.txt"
+export SKANSIBLE_PROFTPD_CONFS_EXIT="${SKANSIBLE_PROFTPD}/conf.d/BYE.txt"
+export SKANSIBLE_SSHD_CONFS="sshd_config.d"
+export SKANSIBLE_SYSTEMD="systemd"
+export SKANSIBLE_SYSTEMD_USER_UNITS="${SKANSIBLE_SYSTEMD}/user"
+export SKANSIBLE_FAIL2BAN="fail2ban"
+export SKANSIBLE_FAIL2BAN_JAILS="${SKANSIBLE_FAIL2BAN}/jail.d"
+export SKANSIBLE_FAIL2BAN_FILTERS="${SKANSIBLE_FAIL2BAN}/filter.d"
+export SKANSIBLE_GITCONFIG_CONFS="gitconfig.d"
+# @NOTE files in here must have extension "key" with IDs in  
+# "gpg_keys" inventory variable list as basenames.
+export SKANSIBLE_GPG="gnupg"
+# @NOTE files in path below must have extensions "key" (private), 
+# "crt" (signed), or "pem" (public) with inventory host FQDN as basename
+export SKANSIBLE_SSL="ca-certificates"
+
+set-root () {
+    if [[ $# -eq 0 ]]; then
+        SKATO_ANSIBLE_ROOT=$(awk -F "=" '/root/ {print $2}' "./config")
+        export SKATO_ANSIBLE_ROOT
+    elif [[ -z "$1" ]]; then
+        SKATO_ANSIBLE_ROOT="$1"
+        export SKATO_ANSIBLE_ROOT
+        sed -i 's|^(root=).*||g' "./config"
+        sed -i "1 i\root=${SKATO_ANSIBLE_ROOT}" "./config"
+    fi
+}
+
+gxy () {
+    ansible-galaxy "$@"
+}
+
+vult () {
+    ansible-vault "$@"
+}
+
+play () {
+    ansible-playbook "$@"
+}
+
+import-gpg () {
+    for id in "$@";
+    do
+        gpg --export-secret-keys "$id" > "${SKATO_BOOTSTRAP_ROLE}/files/${SKANSIBLE_GPG}/${id}.key"
+        printf "Please manually add GPG key with 'id' of '%s' in 'users.\$username.gpg_keys' list of inventory file." "$id"
+    done
+    printf "Please manually change ID attribute of GPG keys in 'users.\$username.gpg_keys' list of inventory file."
+}
+
+import-ssl () {
+    for domain in "$@";
+    do
+        cp "/usr/local/share/ca-certificates/${domain}.key" "${SKATO_BOOTSTRAP_ROLE}/files/${SKANSIBLE_SSL}/${domain}.key"
+        cp "/usr/local/share/ca-certificates/${domain}.pem" "${SKATO_BOOTSTRAP_ROLE}/files/${SKANSIBLE_SSL}/${domain}.pem"
+        cp "/usr/local/share/ca-certificates/${domain}.crt" "${SKATO_BOOTSTRAP_ROLE}/files/${SKANSIBLE_SSL}/${domain}.crt"
+        printf "Please manually change 'fqdn' attribute in inventory group or host variable file to '%s'." "$domain"
+    done
+}
+
+import () {
+    case "$1" in
+        ssl) shift; import-ssl "$@";;
+        gpg) shift; import-gpg "$@";;
+        *) exit 1;;
+    esac
+}
+
+decrypt () {
+    while getopts "mv:i:d:" flag; do
+        case "$flag" in
+            m) METHOD=$OPTARG;;
+            v) VAULT_ID=$OPTARG;;
+            i) INPUT_FILE=$OPTARG;;
+            d) OUTPUT_PATH=$OPTARG;;
+            *) exit 1;;
+        esac
+    done
+
+    if ! [[ "$VAULT_ID" == *"@"* ]]; then
+        ID_TAG="$VAULT_ID"
+
+        if [[ "$METHOD" == "prompt" ]]; then
+            VAULT_ID="${VAULT_ID}@prompt"
+        elif [[ "$METHOD" == "file" ]]; then
+            if [[ -z "$INPUT_FILE" ]]; then
+                exit 1
+            else
+                VAULT_ID="${VAULT_ID}@${INPUT_FILE}"
+            fi
+        else
+            exit 1
+        fi
+    fi
+
+    if [[ -z "$OUTPUT_PATH" ]]; then
+        OUTPUT_FILE="${SKANSIBLE_SECRETS}/${ID_TAG}.txt"
+    else
+        mkdir -p "${SKANSIBLE_SECRETS}/${OUTPUT_PATH}"
+        OUTPUT_FILE="${SKANSIBLE_SECRETS}/${OUTPUT_PATH}/${ID_TAG}.txt"
+    fi
+
+    ansible-vault decrypt --vault-id "$VAULT_ID" --output "$OUTPUT_FILE" "$INPUT_FILE"
+}
+
+encrypt () {
+    while getopts "mv:d:pn:" flag; do
+        case "$flag" in
+            m) METHOD="$OPTARG";;
+            v) VAULT_ID="$OPTARG";;
+            d) PASS_PATH="$OPTARG";;
+            p) read -rp "Provide intended password: " PASSWORD;;
+            n) VAR_NAME="$OPTARG";;
+            *) exit 1;;
+        esac
+    done
+
+    while [[ -z "$PASSWORD" ]]; do
+        printf "Password missing. \nPlease specify a password. \n"
+        read -rp "Provide intended password: " PASSWORD
+    done
+
+    if ! [[ "$VAULT_ID" == *"@"* ]]; then
+        ID_TAG="${VAULT_ID}"
+
+        if [[ "$METHOD" ==  "prompt" ]]; then
+            VAULT_ID="${VAULT_ID}@prompt"
+        elif [[ "$METHOD" == "file" ]]; then
+            if [[ -z "$PASS_PATH" ]]; then
+                PASS_FILE="${SKANSIBLE_SECRETS}/${VAULT_ID}.txt"
+            else
+                mkdir -p "${SKANSIBLE_SECRETS}/${PASS_PATH}"
+                PASS_FILE="${SKANSIBLE_SECRETS}/${PASS_PATH}/${VAULT_ID}.txt"
+            fi
+            printf "%s\n" "$PASSWORD" > "$PASS_FILE"
+            VAULT_ID="${VAULT_ID}@${PASS_FILE}"
+        fi
+    fi
+
+    printf "Make sure to copy following to appropriate location in appropriate YAML file under %s: \n" "$SKATO_ANSIBLE_ROOT"
+    if [[ -z "$VAR_NAME" ]]; then
+        ansible-vault encrypt_string --name "$VAR_NAME" --stdin-name "$VAR_NAME" --vault-id "$VAULT_ID" --output - "$PASSWORD"
+    else
+        ansible-vault encrypt_string --stdin-name "$ID_TAG" --vault-id "$VAULT_ID" --output - "$PASSWORD"
+    fi
+    YAMLS_WITH_PASSWORDS=("${SKATO_BOOTSTRAP_ROLE}/vars/main/software.yml" "${SKATO_BOOTSTRAP_ROLE}/defaults/main/software.yml")
+    printf "Examples of common YAML files passwords may be in: \n"
+    printf "    1. any YAML file in %s \n" "${SKATO_ANSIBLE_ROOT}/hostvars"
+    printf "    2. any YAML file in %s \n" "${SKATO_ANSIBLE_ROOT}/groupvars"
+    for i in "${!YAMLS_WITH_PASSWORDS[@]}"; do
+        printf "    %u. %s \n" "$(( i + 3 ))" "${YAMLS_WITH_PASSWORDS[$i]}"
+    done
+}
+
+# source ./extensions.d/edit.sh
+
+case "$1" in
+    set-root) shift; set-root "$1";;
+    gxy) shift; gxy "$@";;
+    vult) shift; vult "$@";;
+    play) shift; play "$@";;
+    import) shift; import "$@";;
+    decrypt) shift; decrypt "$@";;
+    encrypt) shift; encrypt "$@";;
+    *) exit 1;;
+esac

.env

@@ -1,4 +1,7 @@
-# SKATO_ANSIBLE_ROOT=
+SKATO_ANSIBLE_ROOT=$(dirname "$0")
+SKATO_ANSIBLE_ROOT=$(dirname "$SKATO_ANSIBLE_ROOT")
+SKATO_BOOTSTRAP_ROLE="${SKATO_ANSIBLE_ROOT}/roles/bootstrap"
+SKANSIBLE_SECRETS="${SKATO_ANSIBLE_ROOT}/.secrets"
 
 # Relative directory paths for role templates/files
 SKANSIBLE_ARIA="aria2"
@@ -22,7 +25,4 @@ SKANSIBLE_GITCONFIG_CONFS="gitconfig.d"
 SKANSIBLE_GPG="gnupg"
 # @NOTE files in path below must have extensions "key" (private), 
 # "crt" (signed), or "pem" (public) with inventory host FQDN as basename
-SKANSIBLE_SSL="ca-certificates"
+SKANSIBLE_SSL="ca-certificates"
-
-# Other directories
-SKANSIBLE_SECRETS="${SKATO_ANSIBLE_ROOT}/.secrets"

.gitignore

@@ -1,4 +1,4 @@
-.env/
+.venv/
 *.bak
 hosts.yml
 .secrets/*
@@ -16,4 +16,6 @@ collections/
 motd
 banner
 .galaxy_cache/
-galaxy_token
+galaxy_token
+uv.lock
+__pycache__/

.python-version

@@ -0,0 +1 @@
+3.13

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

@@ -0,0 +1,82 @@
+"""
+Library of custom type hints.
+"""
+
+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 RootFate(StrEnum):
+    disposal = auto()
+    retention = auto()
+
+class NodeType(StrEnum):
+    remote = auto()
+    control = auto()
+
+class VPS(StrEnum):
+    Linode = auto()
+
+class VPSRegion(StrEnum):
+    us_east = auto()
+
+class AnsibleScopes(StrEnum):
+    INTERNAL = auto()
+    INVENTORY = auto()
+    GROUPVARS = auto()
+    HOSTVARS = auto()
+    ROLE = auto()
+
+class AnsibleRoles(StrEnum):
+    bootstrap = auto()
+
+class Scopes(StrEnum):
+    SYS = auto()
+    USER = auto()
+    LOCAL = auto()
+    PROJ = auto()
+    SHARED = auto()
+
+class Roles(StrEnum):
+    CONF = auto()
+    DATA = auto()
+    MEM = auto()
+    EXE = auto()
+
+class UserName(StrEnum):
+    root = auto()
+
+class GroupName(StrEnum):
+    remote = auto()
+    sudo = auto()
+
+class Software(StrEnum):
+    openssh_client = auto()
+    openssh_server = auto()
+
+class SoftwareRoles(StrEnum):
+    client = auto()
+    server = auto()
+
+class PacMans(StrEnum):
+    APT = auto()
+
+class PathCollection(Dict, total=False):
+    sys: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+    user: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+    local: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+    proj: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+    shared: ExecutedPath | IdlePath | str | Sequence[ExecutedPath | IdlePath | str]
+
+class PathRoles(Dict, total=False):
+    conf: PathCollection
+    data: PathCollection
+    mem: PathCollection
+    exe: PathCollection

entities.py

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

gpgkey.py

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

init_env

@@ -1,6 +0,0 @@
-#!/bin/bash
-
-SKATO_ANSIBLE_ROOT=$(dirname "$0")
-set -o allexport
-source "${SKATO_ANSIBLE_ROOT}/standard_paths"
-set +o allexport

main.py

@@ -0,0 +1,65 @@
+"""
+Library for the CLI commands and the related classes and functions
+"""
+
+import click as cli
+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=False, help="Use debugging mode")
+@cli.pass_context
+def skansible(ctx, debug):
+    ctx.ensure_object(dict)
+    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

@@ -0,0 +1,14 @@
+[project]
+name = "skansible"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "ansible>=13.1.0",
+    "ansible-lint>=25.12.1",
+    "ansible-navigator>=25.12.0",
+    "ansible-vault>=4.1.0",
+    "click>=8.3.1",
+    "validators>=0.35.0",
+]

roles/bootstrap/handlers/proftpd.yml

@@ -14,7 +14,7 @@
         group: "{{ item[0]['group'] | default(item[0]['username']) }}"
         path: "{{ item[0]['home'] | default('/home/' ~ item[0]['username']) }}/{{ item[1]['username'] }}"
         state: directory
-      loop: "{{ hostvars[inventory_hostname]['users'].values() | product(config['proftpd']['vusers'].values()) }}"
+      loop: "{{ hostvars[inventory_hostname]['users'].values() | product(config['proftpd']['users'].values()) }}"
     - name: Create ProFTPd FTP public directory for anonymous logins
       when: "'ftps' in item.value['services']"
       ansible.builtin.file:
@@ -65,7 +65,7 @@
             owner: root
             path: "{{ item.value }}"
             state: touch
-          loop: "{{ lookup('ansible.builtin.dict', config['proftpd']['auth_filepaths']) }}"
+          loop: "{{ lookup('ansible.builtin.dict', config['proftpd']['auth_paths']) }}"
         - name: Create the virtual users
           when: "not 'caddy' in item.value['services'] and not 'httpd' in item.value['services'] and not 'www-data' in item.value['services'] and not 'http' in item.value['services'] and not 'https' in item.value['services']"
           ansible.builtin.command:
@@ -73,14 +73,14 @@
               - ftpasswd
               - --passwd
               - "--name={{ item.value['username'] }}"
-              - "--uid=$(id -u {{ item.value['id_of'] }})"
+              - "--uid=$(id -u {{ item.value['id'] }})"
-              - "--gid=$(id -g {{ item.value['gid_of'] }})"
+              - "--gid=$(id -g {{ item.value['gid'] }})"
               - "--home={{ hostvars[inventory_hostname]['users']['ftp']['home'] | default('/srv/ftp') }}/{{ item.value['username'] }}"
               - --shell=/sbin/nologin
-              - --file={{ config['proftpd']['auth_filepaths']['users_path'] }}
+              - --file={{ config['proftpd']['auth_paths']['users'] }}
               - --stdin
             stdin: "{{ item.value['password'] }}"
-          loop: "{{ lookup('ansible.builtin.dict', config['proftpd']['vusers']) }}"
+          loop: "{{ lookup('ansible.builtin.dict', config['proftpd']['users']) }}"
         - name: Create the virtual groups of virtual users
           when: "not 'caddy' in item.value['services'] and not 'httpd' in item.value['services'] and not 'www-data' in item.value['services'] and not 'http' in item.value['services'] and not 'https' in item.value['services']"
           ansible.builtin.command:
@@ -88,10 +88,10 @@
               - ftpasswd
               - --group
               - "--name={{ item.value['username'] }}"
-              - "--gid=$(id -g {{ item.value['gid_of'] }})"
+              - "--gid=$(id -g {{ item.value['gid'] }})"
               - "--member={{ item.value['username'] }}"
-              - --file={{ config['proftpd']['auth_filepaths']['groups_path'] }}
+              - --file={{ config['proftpd']['auth_paths']['groups'] }}
-          loop: "{{ lookup('ansible.builtin.dict', config['proftpd']['vusers']) }}"
+          loop: "{{ lookup('ansible.builtin.dict', config['proftpd']['users']) }}"
     # @TODO create tasks in block integrating LDAP users to ProFTPd
     # - name: Integrate LDAP users into ProFTPd
     - name: Create ProFTPd FTPS virtual host
@@ -108,9 +108,9 @@
         validate: proftpd --configtest
       vars:
         ftp_server_name: "{{ config['proftpd']['name'].uppercase() }}'s Archive'"
-        allowed_users: "{{ ','.join(list(map(lambda u: u['username'], filter(lambda u: not 'http' in u['services'] and not 'https' in u['services'] and not 'httpd' in u['services'] and not 'caddy' in u['services'] and not 'www-data' in u['services'], config['proftpd']['vusers'].values())))) }}"
+        allowed_users: "{{ ','.join(list(map(lambda u: u['username'], filter(lambda u: not 'http' in u['services'] and not 'https' in u['services'] and not 'httpd' in u['services'] and not 'caddy' in u['services'] and not 'www-data' in u['services'], config['proftpd']['users'].values())))) }}"
         anon_root: "{{ map(lambda u: u['home'], filter(lambda u: 'ftps' in u['services'] or 'proftpd' in u['services'], hostvars[inventory_hostname]['users'].values())) | list | random }}/public"
-        anon_user: "{{ config['proftpd']['vusers']['smuggler']['username'] }}"
+        anon_user: "{{ config['proftpd']['users']['smuggler']['username'] }}"
     - name: Set ProFTPd jail in fail2ban
       block:
         - name: Create fail2ban system configuration directory

roles/bootstrap/templates/aria2/aria2.conf.j2

@@ -15,7 +15,7 @@ rpc-allow-origin-all=true
 rpc-max-request-size=10M
 rpc-listen-all=true
 rpc-listen-port=6800
-rpc-secret={{ config.aria.secret }}
+rpc-secret={{ config['aria']['api_key'] }}
 # rpc-certificate=
 # rpc-private-key=
 # rpc-secure=true

roles/bootstrap/templates/proftpd/conf.d/vhost@vps1-fq.dn.conf.example.j2

@@ -20,8 +20,8 @@
 
         # AuthOrder mod_auth_pam.c mod_auth_unix.c*
         AuthOrder mod_auth_file.c
-        AuthUserFile {{ config.proftpd.auth_filepaths.users_path }}
+        AuthUserFile {{ config.proftpd.auth_paths.users }}
-        AuthGroupFile {{ config.proftpd.auth_filepaths.groups_path }}
+        AuthGroupFile {{ config.proftpd.auth_paths.groups }}
         AuthFileOptions SyntaxCheck
 
         TLSEngine on

roles/bootstrap/vars/main/software.yml

@@ -249,16 +249,16 @@ config:
       editor: nvim
   proftpd:
     name: "{{ hostvars[inventory_hostname].fqdn.split('.')[0] }}"
-    auth_filepaths:
+    auth_paths:
-      users_path: /etc/proftpd/ftpd.passwd
+      users: /etc/proftpd/ftpd.passwd
-      groups_path: /etc/proftpd/ftpd.group
+      groups: /etc/proftpd/ftpd.group
     msg:
       welcome: "Our head librarians Furcas and Marbas welcome you!"  
-    vusers:
+    users:
       webmaster:
         username: webmaster
-        id_of: "{{ ['caddy', 'www-data'][0] }}"
+        id: "{{ ['caddy', 'www-data'][0] }}"
-        gid_of: "{{ ['caddy', 'www-data'][0] }}"
+        gid: "{{ ['caddy', 'www-data'][0] }}"
         # @TODO create vaulted password for this ProFTPd virtual user
         password: !vault |
           $ANSIBLE_VAULT;1.2;AES256;vps1-webmaster
@@ -270,8 +270,8 @@ config:
         services: [http,https]
       smuggler:
         username: smuggler
-        id_of: "{{ hostvars[inventory_hostname].users.ftp.username }}"
+        id: "{{ hostvars[inventory_hostname].users.ftp.username }}"
-        gid_of: "{{ hostvars[inventory_hostname].users.ftp.group | default(hostvars[inventory_hostname].users.ftp.username) }}"
+        gid: "{{ hostvars[inventory_hostname].users.ftp.group | default(hostvars[inventory_hostname].users.ftp.username) }}"
         # @TODO create vaulted password for this ProFTPd virtual user
         password: !vault |
           $ANSIBLE_VAULT;1.2;AES256;vps1-smuggler
@@ -293,5 +293,5 @@ config:
     phone_region: US
   aria:
     checksum: ~
-    secret: ~
+    api_key: ~
 

softman.py

@@ -0,0 +1,157 @@
+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.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
+
+    @property
+    def conf_paths(self):
+        if hasattr(self, "_" + Roles.CONF.name.lower()):
+            return self._conf
+        else:
+            raise Exception
+
+    @property
+    def data_paths(self):
+        if hasattr(self, "_" + Roles.DATA.name.lower()):
+            return self._data
+        else:
+            raise Exception
+
+    @property
+    def exec_paths(self):
+        if hasattr(self, "_" + Roles.EXE.name.lower()):
+            return self._exe
+        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
+
+        datatype = datatype.name.lower()
+        
+        if hasattr(self, "_" + datatype):
+            paths = getattr(self, "_" + datatype)
+        else:
+            setattr(self, "_" + datatype, dict())
+            paths = getattr(self, "_" + datatype)
+
+        if scope is not None:
+            if isinstance(path, str):
+                path: ExecutedPath = Path(path)
+
+            if scope.name.lower() not in paths:
+                paths[scope.name.lower()] = []
+
+            paths[scope.name.lower()].append(path)
+        else:
+            paths: PathCollection = path
+
+        setattr(self, "_" + datatype, paths)
+
+    def inherit(self, other):
+        if other._App__children is None:
+            other._App__children = tuple()
+        
+        if self not in other._App__children:
+            other.adopt(self)
+
+        self.__parents = (*self.__parents, other)
+
+    def adopt(self, other: Self):
+        if other._App__parents is None:
+            other._App__parents = tuple()
+
+        if self not in other._App__parent:
+            other.inherit(self)
+
+        self.__children = (*self.__children, other)
+
+    def __enter__(self) -> dict | Sequence:
+        self.__content = self.__current_filepath.read_text()
+        return self.__content
+
+    def __exit__(self, exc_type, exc_value, exc_traceback) -> None:
+        # txt = yams.dump(self.__content)
+        # self.__file.write(txt)
+        # del txt
+        self.__file.close()
+
+    def __call__(self, path = str, mode = "r+", scope: Scopes = Scopes.PROJ, role: Roles = Roles.CONF, index: int = 0) -> Callable:
+        if not hasattr(self, "_" + role.name.lower()):
+            raise Exception
+        else:
+            config = getattr(self, "_" + role.name.lower())
+
+        conf_coll = config[scope.name.lower()]
+
+        if isinstance(conf_coll, Sequence):
+            conf_coll = conf_coll[index]
+        
+        if isinstance(conf_coll, str):
+            conf_coll = Path(conf_coll)
+        
+        filepath = conf_coll / path
+
+        self.__current_filepath = filepath
+        self.__file = open(str(filepath), mode)
+
+        return self
+
+    # @TODO write below method to duplicate file or template in local to project role file/template
+    def clone(self, source_scope: Scopes = Scopes.SYS, target_scope: Scopes = Scopes.PROJ, index: int = 0) -> Never:
+        raise NotImplementedError
+
+    def declare(self, **kwargs):
+        for k, v in kwargs.items():
+            setattr(self, k, v)
+
+sshd_proj_files = map(lambda r: r / AnsibleRoles.bootstrap.name.lower() / "files" / "sshd_config.d", PROJ_ROLES)
+sshd_proj_files = list(filter(lambda p: p.exists(), sshd_proj_files))
+sshd_proj_templates = map(lambda r: r / AnsibleRoles.bootstrap.name.lower() / "templates" / "sshd_config.d", PROJ_ROLES)
+sshd_proj_templates = list(filter(lambda p: p.exists(), sshd_proj_templates))
+# @TODO rewrite below using DIR_ROOTS var from whereami module
+sshd_paths: PathRoles = {
+    Roles.CONF.name.lower(): {
+        Scopes.PROJ.name.lower(): sshd_proj_files + sshd_proj_templates
+    }
+}
+sshd = App(Software.openssh_server, SoftwareRoles.server, sshd_paths)

sshkey.py

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

tasks.org

@@ -3,7 +3,11 @@
 #+language: en
 
 * PLANNED
-** TODO [#A] Write documentation on the expected conventional names to be used in the inventory file
+** DONE [#A] Write documentation on the expected conventional names to be used in the inventory file
+** DONE [#A] Write documentation on the expected conventional paths to be used in the inventory file
+** TODO [#A] Create Python Click library/package- based CLI
+** TODO [#A] Soft-code relative paths for role files/templates in Ansible tasks/plays
+** TODO [#A] Soft-code project root and paths to passwords/secrets files for Ansible tasks/plays
 ** TODO [#A] Rewrite dot notation usage of keys for accessing values in custom dictionary variables to bracket notation usage of keys across whole project
 
 * IN PROGRESS

whereami.py

@@ -0,0 +1,62 @@
+"""
+Library of path constants to be used or referenced elsewhere.
+"""
+
+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
+
+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)
+}