diff --git a/ignis/__init__.py b/ignis/__init__.py index c37f89cb..433c38d4 100644 --- a/ignis/__init__.py +++ b/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: diff --git a/ignis/exceptions.py b/ignis/exceptions.py index 56ae2755..ad3f5a34 100644 --- a/ignis/exceptions.py +++ b/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, + ) diff --git a/ignis/logging.py b/ignis/logging.py index 9f602509..edb4c130 100644 --- a/ignis/logging.py +++ b/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}] {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="") diff --git a/ignis/options.py b/ignis/options.py index a72dc926..75d5c192 100644 --- a/ignis/options.py +++ b/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): diff --git a/ignis/services/greetd/__init__.py b/ignis/services/greetd/__init__.py new file mode 100644 index 00000000..65003ecf --- /dev/null +++ b/ignis/services/greetd/__init__.py @@ -0,0 +1,3 @@ +from .service import GreetdService + +__all__ = ["GreetdService"] diff --git a/ignis/services/greetd/response.py b/ignis/services/greetd/response.py new file mode 100644 index 00000000..d78508e6 --- /dev/null +++ b/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"] diff --git a/ignis/services/greetd/service.py b/ignis/services/greetd/service.py new file mode 100644 index 00000000..60b1db08 --- /dev/null +++ b/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"})