Compare commits
6 Commits
d9d81a43e0
...
c698bc831a
| Author | SHA1 | Date | |
|---|---|---|---|
|
c698bc831a
|
|||
|
d46f3a1f83
|
|||
|
ad8ea5d8b8
|
|||
|
0137fcdad9
|
|||
|
36e553092f
|
|||
|
8ccdf4547e
|
75
anodes.py
Normal file
75
anodes.py
Normal file
@@ -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__
|
||||
@@ -56,8 +56,8 @@ class GroupName(StrEnum):
|
||||
sudo = auto()
|
||||
|
||||
class Software(StrEnum):
|
||||
ssh = auto()
|
||||
sshd = auto()
|
||||
openssh_client = auto()
|
||||
openssh_server = auto()
|
||||
|
||||
class SoftwareRoles(StrEnum):
|
||||
client = auto()
|
||||
|
||||
247
entities.py
Normal file
247
entities.py
Normal file
@@ -0,0 +1,247 @@
|
||||
from typing import Self, Literal, Never, Callable, Sequence
|
||||
from custtypes import Roles, Scopes, PathCollection, PathRoles, Software, SoftwareRoles, ExecutedPath, PacMans, UserName, GroupName, IdlePath, ExecutedPath
|
||||
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
|
||||
|
||||
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 | str = UserName.root.name.lower(), password: str = "test", services: list = [Software.openssh_server.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
|
||||
self.__public_keys = self.__ssh_keys.publish(SSHKeyType.pubkey, datatype=list)
|
||||
self.__private_keys = self.__ssh_keys.publish(SSHKeyType.privkey, datatype=list)[0]
|
||||
# print("here")
|
||||
pubkeys = ssh_keys.publish(SSHKeyType.pubkey, datatype=list)
|
||||
self.__auth_keys: SSHKeyCollection = SSHKeyCollection()
|
||||
for p in pubkeys:
|
||||
self.__auth_keys.append(p)
|
||||
privkeys = ssh_keys.publish(SSHKeyType.privkey, datatype=list)
|
||||
self.__priv_keys: SSHKeyCollection = SSHKeyCollection()
|
||||
for p in privkeys[0]:
|
||||
self.__priv_keys.append(p)
|
||||
self.__priv_key_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]["auth_keys"] = self.__auth_keys
|
||||
users[self.username]["priv_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 add_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.__priv_keys.append(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.__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 available_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 available_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
|
||||
|
||||
82
parse.py
Normal file
82
parse.py
Normal file
@@ -0,0 +1,82 @@
|
||||
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
|
||||
|
||||
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"] | None = None
|
||||
self.__file: ExecutedPath | None = None
|
||||
|
||||
def load(self, filepath: ExecutedPath | str, **kwargs):
|
||||
if isinstance(filepath, ExecutedPath):
|
||||
self.__is_path = True
|
||||
self.__file = filepath
|
||||
filepath = str(filepath)
|
||||
else:
|
||||
self.__is_path = False
|
||||
if Path(filepath).exists():
|
||||
self.__file = Path(filepath)
|
||||
else:
|
||||
self.__content = filepath
|
||||
|
||||
if self.__is_yaml(filepath):
|
||||
self.__method = "yaml"
|
||||
if len(kwargs) > 0:
|
||||
self.__data = yams.load(filepath, Loader=yams.Loader, **kwargs)
|
||||
else:
|
||||
self.__data = yams.load(filepath, Loader=yams.Loader)
|
||||
elif self.__is_config(filepath):
|
||||
self.__method = "config"
|
||||
self.__data = cfg()
|
||||
if self.__is_path:
|
||||
if len(kwargs) > 0:
|
||||
self.__data.read(filepath, **kwargs)
|
||||
else:
|
||||
self.__data.read(filepath)
|
||||
else:
|
||||
if len(kwargs) > 0:
|
||||
self.__data.read_string(filepath, **kwargs)
|
||||
else:
|
||||
self.__data.read_string(filepath, **kwargs)
|
||||
else:
|
||||
raise TypeError
|
||||
|
||||
return self.__data
|
||||
|
||||
def dump(self, obj = None, **kwargs):
|
||||
if self.__method == "yaml":
|
||||
if obj is None:
|
||||
obj = self.__data
|
||||
|
||||
if len(kwargs) > 0:
|
||||
self.__content = yams.dump(obj, Dumper=yams.Dumper, **kwargs)
|
||||
else:
|
||||
self.__content = yams.dump(obj, Dumper=yams.Dumper)
|
||||
elif self.__method == "config":
|
||||
if obj is None:
|
||||
if self.__is_path:
|
||||
return self.__file
|
||||
else:
|
||||
if self.__file is None:
|
||||
return self.__content
|
||||
else:
|
||||
return self.__file
|
||||
else:
|
||||
if isinstance(obj, ConfigParser):
|
||||
return obj
|
||||
else:
|
||||
raise TypeError
|
||||
else:
|
||||
raise TypeError
|
||||
|
||||
return self.__content
|
||||
@@ -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 __enter__
|
||||
|
||||
# @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)
|
||||
159
sshkey.py
159
sshkey.py
@@ -37,8 +37,35 @@ 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:
|
||||
@@ -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()
|
||||
@@ -265,7 +303,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):
|
||||
@@ -502,6 +540,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 +562,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
|
||||
@@ -562,7 +603,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 +614,17 @@ 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 = SSHKeyType.pubkey, pref: int | None = None, datatype = dict):
|
||||
privkey = list()
|
||||
pubkey = list()
|
||||
self.__current = self.__first
|
||||
|
||||
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,16 +632,18 @@ 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)
|
||||
@@ -604,6 +651,7 @@ class SSHKeyCollection(Sequence):
|
||||
result = dict()
|
||||
|
||||
while self.__current is not None:
|
||||
# print(self.__current)
|
||||
if self.__current.category == SSHKeyType.privkey.name.lower():
|
||||
privkey.append(str(self.__current._SSHKey__value))
|
||||
elif self.__current.category == SSHKeyType.pubkey.name.lower():
|
||||
@@ -611,16 +659,17 @@ class SSHKeyCollection(Sequence):
|
||||
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)
|
||||
self.__current = next(self.__current)
|
||||
# print("publish running...")
|
||||
|
||||
if category == SSHKeyType.pubkey.name.lower():
|
||||
if category.name.lower() == SSHKeyType.pubkey.name.lower():
|
||||
result["ssh_authorized_keys"]: list[str] = pubkey
|
||||
|
||||
if category == SSHKeyType.privkey.name.lower():
|
||||
if category.name.lower() == 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():
|
||||
if category.name.lower() == 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)))
|
||||
|
||||
50
whereami.py
50
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)
|
||||
}
|
||||
Reference in New Issue
Block a user