Merge branch 'develop' into ci

app/config/settings.py

@@ -1,18 +1,33 @@
 from functools import lru_cache
-
-from pydantic_settings import BaseSettings, SettingsConfigDict
-from pydantic import BaseModel
 from pathlib import Path
+
 import yaml
+from pydantic import BaseModel
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
 
 class AppConfig(BaseModel):
   host: str = "127.0.0.1"
   port: int = 8000
   reload: bool = True
 
+
+class GitConfig(BaseModel):
+  directory: str = "/opt/repo/sample"
+  branch: str = "master"
+  remote: str = "origin"
+
+
+class KeePassConfig(BaseModel):
+  file: str = "database.kdbx"
+  secret: Path | str = "/run/secrets/kp_secret"
+
+
 class Settings(BaseSettings):
   model_config = SettingsConfigDict(env_prefix="KARL_", env_nested_delimiter="__")
   app: AppConfig = AppConfig()
+  git: GitConfig = GitConfig()
+  kp: KeePassConfig = KeePassConfig()
 
   @classmethod
   def from_yaml(cls, path: Path | str = "config/config.yaml") -> "Settings":
@@ -23,6 +38,7 @@ class Settings(BaseSettings):
         data = yaml.safe_load(fh) or {}
     return cls(**data)
 
+
 @lru_cache
 def get_settings() -> Settings:
   return Settings.from_yaml()

app/model/containers.py

@@ -0,0 +1,40 @@
+from dataclasses import dataclass, field
+from datetime import datetime
+
+from docker.models.containers import Container
+
+
+@dataclass
+class SimpleContainer:
+  name: str
+  image: str
+  status: str
+  health: str
+  created: datetime
+
+  @staticmethod
+  def from_container(container: Container):
+    created = datetime.strptime(container.attrs['Created'].split('.')[0], '%Y-%m-%dT%H:%M:%S')
+    return SimpleContainer(
+      name=container.name,
+      image=container.image.tags[0],
+      status=container.status,
+      health=container.health,
+      created=created
+    )
+
+
+@dataclass
+class Compose:
+  directory: str
+  containers: list[SimpleContainer] = field(default_factory=list)
+
+  @property
+  def last_modified(self):
+    return max(self.containers, key=lambda c: c.created).created
+
+
+@dataclass
+class Tree:
+  composes: dict[str, Compose] = field(default_factory=dict)
+  containers: list[SimpleContainer] = field(default_factory=list)

app/model/passwords.py

@@ -0,0 +1,49 @@
+from dataclasses import dataclass, field
+from typing import Type
+
+# TODO: unnecessary?
+
+@dataclass
+class PathItem:
+  name: str
+  t: Type
+
+@dataclass
+class Path:
+  path: list[PathItem] = field(default_factory=list)
+
+  def append(self, name, t):
+    self.path.append(PathItem(name, t))
+
+  def __str__(self):
+    return "/".join([i.name for i in self.path])
+
+
+@dataclass
+class Group:
+  name: str
+  passwords: list["Password"] = field(default_factory=list)
+  parent: "Group|None" = None
+
+  @property
+  def path(self):
+    if self.parent is None:
+      new_path = Path()
+      new_path.append(self.name, type(self))
+      return new_path
+    return self.parent.path.append(self.name, type(self))
+
+
+@dataclass
+class Password:
+  name: str
+  group: Group
+
+  @property
+  def path(self):
+    return self.group.path.append(self.name, type(self))
+
+class UnencryptedPassword(Password):
+  def __init__(self, name: str,  value: str, group: Group):
+    super().__init__(name, group)
+    self.value = value

app/services/containers.py

@@ -0,0 +1,28 @@
+import docker
+from docker.models.containers import Container
+
+from app.model.containers import Tree, Compose, SimpleContainer
+
+
+class DockerService:
+  def __init__(self):
+    self._client = docker.from_env()
+    self._tree = self._init_tree()
+
+  def _init_tree(self) -> Tree:
+    tree = Tree()
+    container: Container
+    for container in self._client.containers.list():
+      labels = container.labels
+      working_dir = labels.get("com.docker.compose.project.working_dir")
+      if working_dir:
+        if tree.composes.get(working_dir) is None:
+          tree.composes[working_dir] = Compose(working_dir)
+        tree.composes[working_dir].containers.append(SimpleContainer.from_container(container))
+      else:
+        tree.containers.append(SimpleContainer.from_container(container))
+    return tree
+
+  @property
+  def tree(self) -> Tree:
+    return self._tree

app/services/passwords.py

@@ -0,0 +1,38 @@
+import os.path
+
+from pykeepass import PyKeePass, create_database, Group
+
+
+class Passwords:
+  def __init__(self):
+    from app.config import get_settings
+    settings = get_settings()
+
+    with open(settings.kp.secret, "r") as fh:
+      secret = fh.read()
+
+    self._kp_org = self.__get_or_create_store(settings.kp.file, secret)
+    self._kp = self.__get_lock(settings.kp.file, secret)
+
+  @staticmethod
+  def __get_or_create_store(path, passwd) -> PyKeePass:
+    if os.path.exists(path):
+      return PyKeePass(
+        path,
+        password=passwd,
+      )
+    return create_database(path, passwd)
+
+  @staticmethod
+  def __get_lock(path, passwd) -> PyKeePass:
+    lock_path = path + ".lock"
+    import shutil
+    shutil.copyfile(path, lock_path)
+    return Passwords.__get_or_create_store(lock_path, passwd)
+
+  @property
+  def store(self):
+    return self._kp.root_group
+
+  def save(self, group: Group):
+    pass

app/services/system.py

@@ -0,0 +1 @@
+

app/services/vcs.py

@@ -0,0 +1,28 @@
+from git import Repo, Remote
+
+from app.config import get_settings
+
+
+class GitService:
+  def __init__(self):
+    self._settings = get_settings()
+    self._repo = Repo(self._settings.git.directory)
+    self._origin: Remote = self._repo.remotes.origin
+
+  def get_modified_compose(self) -> str | None:
+    self._update()
+    return self._diff()
+
+  def _update(self):
+    self._origin.pull()
+
+  def _diff(self) -> str | None:
+    diff = self._repo.head.commit.diff("HEAD~1")
+    composes = [f for f in diff if f.a_path.endswith("docker-compose.yml")]
+    match len(composes):
+      case 0:
+        return None
+      case 1:
+        return composes[0].a_path
+      case _:
+        raise Exception("Multiple compose files modified")

app/util/dicts.py

@@ -0,0 +1,11 @@
+from types import SimpleNamespace
+
+
+class NestedNamespace(SimpleNamespace):
+  def __init__(self, dictionary, **kwargs):
+    super().__init__(**kwargs)
+    for key, value in dictionary.items():
+      if isinstance(value, dict):
+        self.__setattr__(key, NestedNamespace(value))
+      else:
+        self.__setattr__(key, value)