Compare commits
2 Commits
f46d397e3c
...
86cd3584b3
| Author | SHA1 | Date | |
|---|---|---|---|
|
86cd3584b3
|
|||
|
e412e3d5ab
|
16
anodes.py
16
anodes.py
@@ -4,10 +4,10 @@ Library of classes modeling Ansible nodes or their types.
|
|||||||
|
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from pathlib import Path, PurePath
|
from pathlib import Path, PurePath
|
||||||
from typing import TypeAlias as Neotype, TypedDict as Dict
|
from typing import TypedDict as Dict
|
||||||
from typing import Never, Union, Literal, Required, Self
|
from typing import Union, Literal, Required, Self
|
||||||
from custtypes import ExecutedPath, IdlePath, VirtualPrivateServers, AnsibleScopes
|
from custtypes import ExecutedPath, IdlePath, VirtualPrivateServers, AnsibleScopes
|
||||||
from softman import Software, SoftPathGroup, SoftScope, Apps
|
from softman import Software, SoftPathGroup, SoftScope, Apps, Softs
|
||||||
from whereami import USER_PATH, PROJ_ROOT
|
from whereami import USER_PATH, PROJ_ROOT
|
||||||
from ansible_vault import Vault
|
from ansible_vault import Vault
|
||||||
import secrets
|
import secrets
|
||||||
@@ -71,9 +71,6 @@ class ControlNode:
|
|||||||
def sys_data(self) -> ExecutedPath:
|
def sys_data(self) -> ExecutedPath:
|
||||||
return self.__data_paths
|
return self.__data_paths
|
||||||
|
|
||||||
class Softs(Enum):
|
|
||||||
ssh = 0
|
|
||||||
|
|
||||||
# userSSHParams = Dict("userSSHParams", {
|
# userSSHParams = Dict("userSSHParams", {
|
||||||
# "username": Required[str],
|
# "username": Required[str],
|
||||||
# "paths": Apps,
|
# "paths": Apps,
|
||||||
@@ -119,7 +116,7 @@ class RemoteNode:
|
|||||||
SoftScope.GLOBAL.name: PurePath(str(cnode.sys_confs[0]), "update-motd.d")
|
SoftScope.GLOBAL.name: PurePath(str(cnode.sys_confs[0]), "update-motd.d")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
self.root["software"].declare(Softs.ssh.name, **app_input)
|
self.root["software"].append(Softs.ssh.name, **app_input)
|
||||||
|
|
||||||
self._fqdn = name
|
self._fqdn = name
|
||||||
self.root["software"]._fqdn = name
|
self.root["software"]._fqdn = name
|
||||||
@@ -136,12 +133,7 @@ class RemoteNode:
|
|||||||
self._api_key: str | None = api_key
|
self._api_key: str | None = api_key
|
||||||
self.service = service
|
self.service = service
|
||||||
self.region = region
|
self.region = region
|
||||||
self.__authkeys_selected = False
|
|
||||||
self.__usedkeys_selected = False
|
|
||||||
self.__keys_selected = False
|
|
||||||
self.__finalized_keys = False
|
|
||||||
self.model: dict | None = None
|
self.model: dict | None = None
|
||||||
self.__key_accumulator: tuple | list | None = []
|
|
||||||
|
|
||||||
def set_region(self, name: Literal["us-east"] = "us-east") -> None:
|
def set_region(self, name: Literal["us-east"] = "us-east") -> None:
|
||||||
self.region = name
|
self.region = name
|
||||||
|
|||||||
29
softman.py
29
softman.py
@@ -10,6 +10,7 @@ from custtypes import ExecutedPath, IdlePath
|
|||||||
from enum import Enum
|
from enum import Enum
|
||||||
from pathlib import Path, PurePath
|
from pathlib import Path, PurePath
|
||||||
from whereami import USER_PATH, PROJ_ROOT
|
from whereami import USER_PATH, PROJ_ROOT
|
||||||
|
from collections.abc import Sequence
|
||||||
|
|
||||||
AppPath: Neotype = Union[ExecutedPath, IdlePath]
|
AppPath: Neotype = Union[ExecutedPath, IdlePath]
|
||||||
|
|
||||||
@@ -68,14 +69,14 @@ Apps = type("Apps", (), __app_input)
|
|||||||
|
|
||||||
# @TODO continue adding magic methods to below class
|
# @TODO continue adding magic methods to below class
|
||||||
# @NOTE https://rszalski.github.io/magicmethods/#sequence
|
# @NOTE https://rszalski.github.io/magicmethods/#sequence
|
||||||
class Software:
|
class Software (Sequence):
|
||||||
__user_path: ExecutedPath = USER_PATH
|
__user_path: ExecutedPath = USER_PATH
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self._fqdn: str | None = None
|
self._fqdn: str | None = None
|
||||||
|
|
||||||
# @TODO fix NameError for 'Software' in parameter type check
|
# @TODO fix NameError for 'Software' in parameter type check
|
||||||
def declare(self, name: str, **kwpaths: _SubAppParams) -> AppParams:
|
def append(self, name: str, **kwpaths: _SubAppParams) -> AppParams:
|
||||||
keyword_args: AppParams = kwpaths
|
keyword_args: AppParams = kwpaths
|
||||||
|
|
||||||
app = Apps(**keyword_args)
|
app = Apps(**keyword_args)
|
||||||
@@ -153,11 +154,29 @@ class Software:
|
|||||||
)
|
)
|
||||||
return len(apps)
|
return len(apps)
|
||||||
|
|
||||||
def __contains__(self):
|
def pop(self) -> Never:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def __missing__(self):
|
def remove(self) -> Never:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def __iter__(self):
|
def __contains__(self) -> Never:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def count(self) -> Never:
|
||||||
|
return NotImplementedError
|
||||||
|
|
||||||
|
def __missing__(self) -> Never:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def __iter__(self) -> Never:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def reverse(self) -> Never:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def sort(self, key: Callable = (lambda e: e), reverse: bool = False) -> Never:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
class Softs(Enum):
|
||||||
|
ssh = 0
|
||||||
195
sshkey_man.py
195
sshkey_man.py
@@ -4,6 +4,12 @@ from custtypes import ExecutedPath, IdlePath, VirtualPrivateServers, AnsibleScop
|
|||||||
from enum import Enum
|
from enum import Enum
|
||||||
from softman import Apps
|
from softman import Apps
|
||||||
from random import gamble
|
from random import gamble
|
||||||
|
from collections.abc import Sequence
|
||||||
|
from typing import Never, Union, Self, Callable
|
||||||
|
from glob import glob as globbify
|
||||||
|
from whereami import USER_PATH
|
||||||
|
from softman import Softs
|
||||||
|
import os
|
||||||
|
|
||||||
class RootFate(Enum):
|
class RootFate(Enum):
|
||||||
disposal = 0
|
disposal = 0
|
||||||
@@ -38,7 +44,7 @@ class SSHKey:
|
|||||||
return str(self.__value)
|
return str(self.__value)
|
||||||
|
|
||||||
def __repr__(self) -> ExecutedPath | tuple[ExecutedPath]:
|
def __repr__(self) -> ExecutedPath | tuple[ExecutedPath]:
|
||||||
return self.__value
|
return "SSHKey(" + str(self.__value) + ")"
|
||||||
|
|
||||||
def __nonzero__(self) -> bool:
|
def __nonzero__(self) -> bool:
|
||||||
return True
|
return True
|
||||||
@@ -58,9 +64,26 @@ class SSHKey:
|
|||||||
def __prev__(self) -> ExecutedPath | tuple[ExecutedPath]:
|
def __prev__(self) -> ExecutedPath | tuple[ExecutedPath]:
|
||||||
return self.__prev
|
return self.__prev
|
||||||
|
|
||||||
def __call__(self) -> ExecutedPath | tuple[ExecutedPath]:
|
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
|
return self.__value
|
||||||
|
|
||||||
|
def __eq__(self, other: Self) -> bool:
|
||||||
|
return self.__value == other._SSHKey__value
|
||||||
|
|
||||||
|
def __eqcontent__(self, other: Self) -> bool:
|
||||||
|
return self.__value.read_text() == other._SSHKey__value.read_text()
|
||||||
|
|
||||||
def update(self, *path: ExecutedPath | str) -> None | Never:
|
def update(self, *path: ExecutedPath | str) -> None | Never:
|
||||||
if len(path) > 2 or len(path) < 1:
|
if len(path) > 2 or len(path) < 1:
|
||||||
raise ValueError
|
raise ValueError
|
||||||
@@ -118,10 +141,13 @@ class SSHKey:
|
|||||||
else:
|
else:
|
||||||
raise ValueError
|
raise ValueError
|
||||||
|
|
||||||
def publish(self, idx: int | None = None) -> str | tuple[str]:
|
def read(self, idx: int | None = None) -> str | tuple[str]:
|
||||||
if idx is not None:
|
if idx is not None and isinstance(self.__value, tuple):
|
||||||
result = self.__value[idx]
|
result = self.__value[idx]
|
||||||
else:
|
else:
|
||||||
|
if idx is not None:
|
||||||
|
raise KeyError
|
||||||
|
|
||||||
result = self.__value
|
result = self.__value
|
||||||
|
|
||||||
if isinstance(result, tuple):
|
if isinstance(result, tuple):
|
||||||
@@ -136,49 +162,54 @@ class SSHKey:
|
|||||||
# @TODO this method should return string or Enum value after analyzing whether this key is public or private
|
# @TODO this method should return string or Enum value after analyzing whether this key is public or private
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
class SSHKeyCollection:
|
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
|
||||||
|
|
||||||
|
class SSHKeyCollection(Sequence):
|
||||||
|
__user_path: ExecutedPath = USER_PATH
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.__current: SSHKey | None = None
|
self.__current: SSHKey | None = None
|
||||||
self.__first: SSHKey | None = None
|
self.__first: SSHKey | None = None
|
||||||
self.__last: SSHKey | None = None
|
self.__last: SSHKey | None = None
|
||||||
self.__indices: range | None = None
|
self.__indices: range | None = None
|
||||||
|
|
||||||
def __setitem__(self, key: int, *value: ExecutedPath | str) -> None | Never:
|
def __getitem__(self, key: int) -> SSHKey | 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))
|
|
||||||
|
|
||||||
if self.__current is None:
|
|
||||||
self.__current = SSHKey(*value)
|
|
||||||
self.__current._SSHKey__idx = key
|
|
||||||
elif int(self.__current) == key:
|
|
||||||
if self.__current() is None or len(self.__current()) < 1:
|
|
||||||
self.__current.update(*value)
|
|
||||||
else:
|
|
||||||
while int(self.__current) != key:
|
|
||||||
if next(self.__current) is not None:
|
|
||||||
self.__current = next(self.__current)
|
|
||||||
else:
|
|
||||||
break
|
|
||||||
|
|
||||||
self.__current.update(*value)
|
|
||||||
|
|
||||||
def __getitem__(self, key: int) -> SSHKey:
|
|
||||||
if self.__current is None:
|
if self.__current is None:
|
||||||
raise KeyError
|
raise KeyError
|
||||||
elif int(self.__current) == key:
|
elif int(self.__current) == key:
|
||||||
return self.__current
|
return self.__current
|
||||||
else:
|
else:
|
||||||
|
self.__current = self.__first
|
||||||
|
|
||||||
while int(self.__current) != key:
|
while int(self.__current) != key:
|
||||||
if next(self.__current) is not None:
|
if next(self.__current) is not None:
|
||||||
self.__current = next(self.__current)
|
self.__current = next(self.__current)
|
||||||
else:
|
else:
|
||||||
|
# raise StopIteration
|
||||||
break
|
break
|
||||||
|
|
||||||
return self.__current
|
return self.__current
|
||||||
|
|
||||||
def __delitem__(self, key: int) -> Never:
|
def __len__(self) -> int:
|
||||||
|
if self.__indices is None:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
return len(self.__indices)
|
||||||
|
|
||||||
|
def pop(self) -> Never:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def remove(self) -> Never:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def append(self, *value: ExecutedPath | str) -> None | Never:
|
def append(self, *value: ExecutedPath | str) -> None | Never:
|
||||||
@@ -194,44 +225,114 @@ class SSHKeyCollection:
|
|||||||
self.__indices = range(ssh_key._SSHKey__idx + 1)
|
self.__indices = range(ssh_key._SSHKey__idx + 1)
|
||||||
self.__first = ssh_key
|
self.__first = ssh_key
|
||||||
|
|
||||||
if self.__last is None:
|
|
||||||
self.__last = ssh_key
|
|
||||||
|
|
||||||
self.__first._SSHKey__next = self.__last
|
|
||||||
self.__last._SSHKey__prev = self.__first
|
|
||||||
|
|
||||||
self.__current = self.__first
|
self.__current = self.__first
|
||||||
else:
|
else:
|
||||||
ssh_key._SSHKey__idx = len(self.__indices)
|
ssh_key._SSHKey__idx = len(self.__indices)
|
||||||
|
|
||||||
ssh_key._SSHKey__prev = self.__last
|
if self.__last is not None:
|
||||||
self.__last = ssh_key
|
self.__last._SSHKey__next = ssh_key
|
||||||
|
self.__last._SSHKey__next._SSHKey__prev = self.__last
|
||||||
|
self.__last = next(self.__last)
|
||||||
|
else:
|
||||||
|
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.__indices = range(ssh_key._SSHKey__idx + 1)
|
||||||
self.__current = self.__last
|
self.__current = self.__last
|
||||||
|
|
||||||
def pop(self) -> Never:
|
def __setitem__(self, key: int, *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))
|
||||||
|
|
||||||
|
if self.__current is None:
|
||||||
|
raise KeyError
|
||||||
|
elif int(self.__current) == key:
|
||||||
|
if self.__current() is None or len(self.__current()) < 1:
|
||||||
|
self.__current(*value)
|
||||||
|
else:
|
||||||
|
self.__current = self.__first
|
||||||
|
|
||||||
|
while int(self.__current) != key:
|
||||||
|
if next(self.__current) is not None:
|
||||||
|
self.__current = next(self.__current)
|
||||||
|
else:
|
||||||
|
# raise StopIteration
|
||||||
|
break
|
||||||
|
|
||||||
|
self.__current(*value)
|
||||||
|
|
||||||
|
def __delitem__(self, key: int) -> None | Never:
|
||||||
|
if self.__current is None:
|
||||||
|
raise KeyError
|
||||||
|
elif int(self.__current) == key:
|
||||||
|
prior = self.__current._SSHKey__prev
|
||||||
|
posterior = self.__current._SSHKey__next
|
||||||
|
posterior._SSHKey__idx = self.__current._SSHKey__idx
|
||||||
|
prior._SSHKey__next = posterior
|
||||||
|
prior._SSHKey__next._SSHKey__prev = prior
|
||||||
|
self.__current = prior._SSHKey__next
|
||||||
|
|
||||||
|
while next(self.__current) is not None:
|
||||||
|
self.__current = next(self.__current)
|
||||||
|
self.__current._SSHKey__idx += 1
|
||||||
|
else:
|
||||||
|
self.__current = self.__first
|
||||||
|
|
||||||
|
while int(self.__current) != key:
|
||||||
|
if next(self.__current) is not None:
|
||||||
|
self.__current = next(self.__current)
|
||||||
|
else:
|
||||||
|
# raise StopIteration
|
||||||
|
break
|
||||||
|
|
||||||
|
prior = self.__current._SSHKey__prev
|
||||||
|
posterior = self.__current._SSHKey__next
|
||||||
|
posterior._SSHKey__idx = self.__current._SSHKey__idx
|
||||||
|
prior._SSHKey__next = posterior
|
||||||
|
prior._SSHKey__next._SSHKey__prev = prior
|
||||||
|
self.__current = prior._SSHKey__next
|
||||||
|
|
||||||
|
while next(self.__current) is not None:
|
||||||
|
self.__current = next(self.__current)
|
||||||
|
self.__current._SSHKey__idx += 1
|
||||||
|
|
||||||
|
@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
|
||||||
|
|
||||||
|
# @TODO make sure to implement below method
|
||||||
|
def __contains__(self) -> bool | Never:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def remove(self) -> Never:
|
def __missing__(self) -> Never:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
# @TODO make sure to implement below method
|
||||||
|
def __iter__(self) -> Self | Never:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def count(self, query: RegEx | str) -> int | Never:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def pull(self, query: RegEx | str = "*") -> Never:
|
def pull(self, query: RegEx | str = "*") -> Never:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def __len__(self):
|
def reverse(self) -> None | Never:
|
||||||
return len(self.__indices)
|
|
||||||
|
|
||||||
def __contains__(self):
|
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def __missing__(self):
|
def sort(self, key: Callable = (lambda e: e), reverse: bool = False) -> None | Never:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
def __iter__(self):
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
|
|
||||||
class UserSSH:
|
class UserSSH:
|
||||||
def __init__(self, username: str = "root", paths: Apps | None = None, keys: _userSSHSubParams = __user_ssh_keys, password: str = "password123", fate: RootFate = RootFate.disposal.name):
|
def __init__(self, username: str = "root", paths: Apps | None = None, keys: _userSSHSubParams = __user_ssh_keys, password: str = "password123", fate: RootFate = RootFate.disposal.name):
|
||||||
self.username = username
|
self.username = username
|
||||||
|
|||||||
Reference in New Issue
Block a user