feat: Greetd Service

ignis/__init__.py

@@ -9,7 +9,11 @@
 CACHE_DIR = None
 
 if "sphinx" not in sys.modules:
-    CACHE_DIR = f"{GLib.get_user_cache_dir()}/ignis"
+    if not os.getenv("GREETD_SOCK"):
+        CACHE_DIR = f"{GLib.get_user_cache_dir()}/ignis"
+    else:
+        CACHE_DIR = "/tmp/ignis_greetd"
+
     os.makedirs(CACHE_DIR, exist_ok=True)
 
     try:

ignis/exceptions.py

@@ -491,3 +491,14 @@ def __init__(self, *args: object) -> None:
             "GnomeBluetooth-3.0 is not found! To use the Bluetooth Service, install GnomeBluetooth-3.0",
             *args,
         )
+
+class GreetdSockNotFoundError(Exception):
+    """
+    Raised when greetd socket is not found.
+    """
+
+    def __init__(self, *args: object) -> None:
+        super().__init__(
+            "Greetd socket is not found! To use the Greetd Service, ensure you are running Ignis in greetd",
+            *args,
+        )

ignis/logging.py

@@ -2,9 +2,11 @@
 import sys
 from loguru import logger
 from gi.repository import GLib  # type: ignore
+from . import CACHE_DIR
 
 LOG_DIR = os.path.expanduser("~/.ignis")
-LOG_FILE = f"{LOG_DIR}/ignis.log"
+DEFAULT_LOG_FILE = f"{LOG_DIR}/ignis.log"
+GREETD_LOG_FILE = f"{CACHE_DIR}/ignis_greetd.log"
 LOG_FORMAT = "{time:YYYY-MM-DD HH:mm:ss} [<level>{level}</level>] {message}"
 
 
@@ -51,7 +53,13 @@ def configure_logger(debug: bool) -> None:
         LEVEL = "INFO"
 
     logger.add(sys.stderr, colorize=True, format=LOG_FORMAT, level=LEVEL)
-    logger.add(LOG_FILE, format=LOG_FORMAT, level=LEVEL, rotation="1 day")
+
+    if not os.getenv("GREETD_SOCK"):
+        log_file = DEFAULT_LOG_FILE
+    else:
+        log_file = GREETD_LOG_FILE
+
+    logger.add(log_file, format=LOG_FORMAT, level=LEVEL, rotation="1 day")
 
     logger.level("INFO", color="<green>")
 

ignis/options.py

@@ -61,7 +61,7 @@ class Options(OptionsManager):
     def __init__(self):
         try:
             super().__init__(file=f"{CACHE_DIR}/ignis_options.json")
-        except FileNotFoundError:
+        except (FileNotFoundError, PermissionError):
             pass
 
     class Notifications(OptionsGroup):

ignis/services/greetd/__init__.py

@@ -0,0 +1,3 @@
+from .service import GreetdService
+
+__all__ = ["GreetdService"]

ignis/services/greetd/response.py

@@ -0,0 +1,49 @@
+from typing import Literal
+
+
+class GreetdBaseResponse:
+    def __init__(self, resp: dict[str, str], type_: str):
+        self._resp = resp
+        if resp["type"] != type_:
+            raise ValueError(
+                f"Incorrect response type: expected {type_}, got {resp['type']}"
+            )
+
+    @property
+    def resp(self) -> dict[str, str]:
+        return self._resp
+
+    @property
+    def resp_type(self) -> str:
+        return self._resp["type"]
+
+
+class GreetdSuccessResponse(GreetdBaseResponse):
+    def __init__(self, resp: dict[str, str]):
+        super().__init__(resp, "success")
+
+
+class GreetdErrorResponse(GreetdBaseResponse):
+    def __init__(self, resp: dict[str, str]):
+        super().__init__(resp, "error")
+
+    @property
+    def error_type(self) -> Literal["auth_error", "error"]:
+        return self._resp["error_type"]
+
+    @property
+    def description(self) -> str:
+        return self._resp["description"]
+
+
+class GreetdAuthMessage(GreetdBaseResponse):
+    def __init__(self, resp: dict[str, str]):
+        super().__init__(resp, "auth_message")
+
+    @property
+    def auth_message_type(self) -> Literal["visible", "secret", "info", "error"]:
+        return self._resp["auth_message_type"]
+
+    @property
+    def auth_message(self) -> str:
+        return self._resp["auth_message"]

ignis/services/greetd/service.py

@@ -0,0 +1,73 @@
+import os
+import socket
+import json
+from gi.repository import GObject  # type: ignore
+from ignis.base_service import BaseService
+from ignis.exceptions import GreetdSockNotFoundError
+from .response import (
+    GreetdSuccessResponse,
+    GreetdErrorResponse,
+    GreetdAuthMessage,
+    GreetdBaseResponse,
+)
+
+GREETD_SOCK = os.getenv("GREETD_SOCK", "")
+
+GREETD_RESPONSES: dict[str, GreetdBaseResponse] = {
+    "success": GreetdSuccessResponse,
+    "error": GreetdErrorResponse,
+    "auth_message": GreetdAuthMessage,
+}
+
+GreetdResponse = GreetdSuccessResponse | GreetdErrorResponse | GreetdAuthMessage
+
+
+class GreetdService(BaseService):
+    def __init__(self):
+        super().__init__()
+        self._sock: socket.socket | None = None
+
+    @GObject.Property
+    def is_available(self) -> bool:
+        return os.path.exists(GREETD_SOCK)
+
+    def __send_request(self, request: dict) -> str:
+        if not self.is_available:
+            raise GreetdSockNotFoundError()
+
+        if not self._sock:
+            self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            self._sock.connect(GREETD_SOCK)
+
+        json_request = json.dumps(request)
+        self._sock.send(
+            len(json_request).to_bytes(4, "little") + json_request.encode("utf-8")
+        )
+        resp_raw = self._sock.recv(128)
+        resp_len = int.from_bytes(resp_raw[0:4], "little")
+        resp_trimmed = resp_raw[4 : resp_len + 4].decode()
+
+        response = json.loads(resp_trimmed)
+
+        return GREETD_RESPONSES[response["type"]](response)
+
+    def create_session(self, username: str) -> GreetdResponse:
+        return self.__send_request(
+            request={"type": "create_session", "username": username}
+        )
+
+    def post_auth_message_response(self, response: str | None = None) -> GreetdResponse:
+        request = {"type": "post_auth_message_response"}
+
+        if response is not None:
+            request["response"] = response
+
+        return self.__send_request(request=request)
+
+    def start_session(self, cmd: list[str], env: list[str]) -> GreetdResponse:
+        return self.__send_request(
+            request={"type": "start_session", "cmd": cmd, "env": env}
+        )
+
+    def cancel_session(self, username: str) -> GreetdResponse:
+        return self.__send_request(request={"type": "cancel_session"})