Compare commits
5 Commits
e3683fe955
...
ffcf6b2596
| Author | SHA1 | Date | |
|---|---|---|---|
|
ffcf6b2596
|
|||
|
0d44ec029f
|
|||
|
6af0d4b48c
|
|||
|
f37c2d5998
|
|||
|
a2d921a158
|
@@ -7,9 +7,11 @@ from typing import Required
|
|||||||
from collections.abc import Sequence
|
from collections.abc import Sequence
|
||||||
from pathlib import Path, PurePath, PurePosixPath, PureWindowsPath, PosixPath, WindowsPath
|
from pathlib import Path, PurePath, PurePosixPath, PureWindowsPath, PosixPath, WindowsPath
|
||||||
from enum import StrEnum, auto
|
from enum import StrEnum, auto
|
||||||
|
from io import TextIOBase, BufferedIOBase, RawIOBase
|
||||||
|
|
||||||
ExecutedPath: Neotype = PosixPath | WindowsPath
|
ExecutedPath: Neotype = PosixPath | WindowsPath
|
||||||
IdlePath: Neotype = PurePosixPath | PureWindowsPath
|
IdlePath: Neotype = PurePosixPath | PureWindowsPath
|
||||||
|
File: Neotype = TextIOBase | BufferedIOBase | RawIOBase
|
||||||
|
|
||||||
class RootFate(StrEnum):
|
class RootFate(StrEnum):
|
||||||
disposal = auto()
|
disposal = auto()
|
||||||
|
|||||||
67
entities.py
67
entities.py
@@ -1,21 +1,22 @@
|
|||||||
from typing import Self, Literal, Never, Callable, Sequence
|
from typing import Self, Literal, Never, Callable, Sequence
|
||||||
from custtypes import Roles, Scopes, PathCollection, PathRoles
|
from custtypes import Roles, Scopes, PathCollection, PathRoles
|
||||||
from custtypes import Software, SoftwareRoles, PacMans
|
from custtypes import Software, SoftwareRoles, PacMans
|
||||||
from custtypes import ExecutedPath, IdlePath
|
from custtypes import ExecutedPath, IdlePath, File
|
||||||
from custtypes import UserName, GroupName
|
from custtypes import UserName, GroupName, VPS, VPSRegion, RootFate
|
||||||
from pathlib import Path, PurePath
|
from pathlib import Path, PurePath
|
||||||
from sshkey import SSHKeyCollection, SSHKeyType, SSHKey
|
from sshkey import SSHKeyCollection, SSHKeyType, SSHKey
|
||||||
from random import choice as gamble
|
from random import choice as gamble
|
||||||
from re import Pattern as RegEx
|
from re import Pattern as RegEx
|
||||||
from softman import sshd
|
from softman import sshd
|
||||||
from yaml import YAMLObject
|
from yaml import YAMLObject
|
||||||
|
from ansible_vault import Vault
|
||||||
|
|
||||||
|
|
||||||
class Group(YAMLObject):
|
class Group(YAMLObject):
|
||||||
yaml_tag = u"!Group"
|
yaml_tag = u"!Group"
|
||||||
|
|
||||||
# @TODO create Enum class child for category parameter type hinting in below method
|
# @TODO create Enum class child for category parameter type hinting in below method
|
||||||
def __init__(self, group_name: GroupName = GroupName.sudo, category: Literal["system", "regular"] = "system", gid: int | str = 27):
|
def __init__(self, group_name: GroupName | str = GroupName.sudo, category: Literal["system", "regular"] = "system", gid: int | str | None = 27):
|
||||||
if isinstance(group_name, GroupName):
|
if isinstance(group_name, GroupName):
|
||||||
self.group_name = group_name.name.lower()
|
self.group_name = group_name.name.lower()
|
||||||
else:
|
else:
|
||||||
@@ -36,7 +37,7 @@ class Group(YAMLObject):
|
|||||||
class User(YAMLObject):
|
class User(YAMLObject):
|
||||||
yaml_tag = u"!User"
|
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 = 0):
|
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
|
self.exists = True
|
||||||
|
|
||||||
if isinstance(username, UserName):
|
if isinstance(username, UserName):
|
||||||
@@ -128,7 +129,7 @@ class User(YAMLObject):
|
|||||||
else:
|
else:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
def add_keypair(self, private_key: SSHKey | ExecutedPath | str | int | RegEx, public_key: SSHKey | ExecutedPath | str | int | RegEx, from_host = True):
|
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:
|
if not self.__ssh_keypair_chosen:
|
||||||
self.__priv_keys = SSHKeyCollection()
|
self.__priv_keys = SSHKeyCollection()
|
||||||
self.__auth_keys = SSHKeyCollection()
|
self.__auth_keys = SSHKeyCollection()
|
||||||
@@ -297,3 +298,59 @@ class User(YAMLObject):
|
|||||||
self.id
|
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
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
31
parse.py
31
parse.py
@@ -25,22 +25,26 @@ class Parser:
|
|||||||
self.__file = filepath
|
self.__file = filepath
|
||||||
filepath = str(filepath)
|
filepath = str(filepath)
|
||||||
else:
|
else:
|
||||||
self.__is_path = False
|
if isinstance(filepath, str) and Path(filepath).exists():
|
||||||
if Path(filepath).exists():
|
|
||||||
self.__file = Path(filepath)
|
self.__file = Path(filepath)
|
||||||
|
self.__is_path = True
|
||||||
else:
|
else:
|
||||||
self.__content = filepath
|
self.__is_path = False
|
||||||
|
|
||||||
|
if isinstance(filepath, str):
|
||||||
|
self.__content = filepath
|
||||||
|
|
||||||
if self.__is_yaml(filepath) or method == "yaml":
|
if self.__is_yaml(filepath) or method == "yaml":
|
||||||
self.__method = "yaml"
|
self.__method = "yaml"
|
||||||
|
|
||||||
if self.__is_path:
|
if self.__is_path:
|
||||||
filepath = open(str(filepath), "r")
|
filepath = open(str(filepath), "r+")
|
||||||
|
|
||||||
if len(kwargs) > 0:
|
if len(kwargs) > 0:
|
||||||
self.__data = yams.load(filepath, Loader=yams.Loader, **kwargs)
|
self.__data = yams.load_all(filepath, Loader=yams.Loader, **kwargs)
|
||||||
else:
|
else:
|
||||||
self.__data = yams.load(filepath, Loader=yams.Loader)
|
self.__data = yams.load_all(filepath, Loader=yams.Loader)
|
||||||
|
filepath.close()
|
||||||
elif self.__is_config(filepath) or method == "config":
|
elif self.__is_config(filepath) or method == "config":
|
||||||
self.__method = "config"
|
self.__method = "config"
|
||||||
self.__data = cfg()
|
self.__data = cfg()
|
||||||
@@ -60,15 +64,15 @@ class Parser:
|
|||||||
return self.__data
|
return self.__data
|
||||||
|
|
||||||
def dump(self, obj = None, method: Literal["yaml", "config", "generic"] | None = "generic", **kwargs):
|
def dump(self, obj = None, method: Literal["yaml", "config", "generic"] | None = "generic", **kwargs):
|
||||||
if self.__method == "yaml" or method == "yaml":
|
if isinstance(obj, yams.YAMLObject) or self.__method == "yaml" or method == "yaml":
|
||||||
if obj is None:
|
if obj is None:
|
||||||
obj = self.__data
|
obj = self.__data
|
||||||
|
|
||||||
if len(kwargs) > 0:
|
if len(kwargs) > 0:
|
||||||
self.__content = yams.dump(obj, Dumper=yams.Dumper, **kwargs)
|
self.__content = "---\n" + yams.dump(obj, Dumper=yams.Dumper, **kwargs)
|
||||||
else:
|
else:
|
||||||
self.__content = yams.dump(obj, Dumper=yams.Dumper)
|
self.__content = "---\n" + yams.dump(obj, Dumper=yams.Dumper)
|
||||||
elif self.__method == "config" or method == "config":
|
elif isinstance(obj, ConfigParser) or self.__method == "config" or method == "config":
|
||||||
if obj is None:
|
if obj is None:
|
||||||
if self.__is_path:
|
if self.__is_path:
|
||||||
return self.__file.read_text()
|
return self.__file.read_text()
|
||||||
@@ -77,11 +81,8 @@ class Parser:
|
|||||||
return self.__content
|
return self.__content
|
||||||
else:
|
else:
|
||||||
return self.__file.read_text()
|
return self.__file.read_text()
|
||||||
else:
|
|
||||||
if isinstance(obj, ConfigParser):
|
raise NotImplementedError
|
||||||
return obj
|
|
||||||
else:
|
|
||||||
raise TypeError
|
|
||||||
else:
|
else:
|
||||||
raise TypeError
|
raise TypeError
|
||||||
|
|
||||||
|
|||||||
@@ -9,6 +9,6 @@ dependencies = [
|
|||||||
"ansible-lint>=25.12.1",
|
"ansible-lint>=25.12.1",
|
||||||
"ansible-navigator>=25.12.0",
|
"ansible-navigator>=25.12.0",
|
||||||
"ansible-vault>=4.1.0",
|
"ansible-vault>=4.1.0",
|
||||||
"cerberus>=1.3.8",
|
|
||||||
"click>=8.3.1",
|
"click>=8.3.1",
|
||||||
|
"validators>=0.35.0",
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -134,7 +134,7 @@ class App:
|
|||||||
self.__current_filepath = filepath
|
self.__current_filepath = filepath
|
||||||
self.__file = open(str(filepath), mode)
|
self.__file = open(str(filepath), mode)
|
||||||
|
|
||||||
return __enter__
|
return self
|
||||||
|
|
||||||
# @TODO write below method to duplicate file or template in local to project role file/template
|
# @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:
|
def clone(self, source_scope: Scopes = Scopes.SYS, target_scope: Scopes = Scopes.PROJ, index: int = 0) -> Never:
|
||||||
|
|||||||
Reference in New Issue
Block a user