linux-user-timer-native-app/desktop/client.py

import json
import socket
from pathlib import Path


SOCKET_PATH = Path("/run/parental-control.sock")
SOCKET_TIMEOUT = 10


class IPCError(Exception):
    """Base exception for IPC errors."""

class DatabaseError(IPCError):
    """Raised when a database operation fails."""

class AuthorizationError(IPCError):
    """Raised when an administrative operation is denied."""


class IPCClient:
    @staticmethod
    def call(method, **arguments):
        request = {
            "method": method,
            "args": arguments,
        }

        payload = (
            json.dumps(
                request,
                separators=(",", ":"),
            ).encode("utf-8")
            + b"\n"
        )

        data = bytearray()

        try:
            with socket.socket(
                socket.AF_UNIX,
                socket.SOCK_STREAM,
            ) as connection:

                connection.settimeout(
                    SOCKET_TIMEOUT
                )

                connection.connect(
                    str(SOCKET_PATH)
                )

                connection.sendall(
                    payload
                )

                while True:
                    chunk = connection.recv(
                        65536
                    )

                    if not chunk:
                        break

                    data.extend(chunk)

                    if b"\n" in chunk:
                        break

        except FileNotFoundError:
            raise IPCError(
                "Parental control service is not running."
            )

        except ConnectionRefusedError:
            raise IPCError(
                "Could not connect to the parental control service."
            )

        except TimeoutError:
            raise IPCError(
                "The parental control service did not respond."
            )

        except OSError as exc:
            raise IPCError(
                f"Could not connect to the parental control service: {exc}"
            )

        if not data:
            raise IPCError(
                "The parental control service returned no response."
            )

        line = bytes(data).split(
            b"\n",
            1,
        )[0]

        try:
            response = json.loads(
                line.decode("utf-8")
            )

        except (
            UnicodeDecodeError,
            json.JSONDecodeError,
        ):
            raise IPCError(
                "The parental control service returned an invalid response."
            )

        if response.get("ok"):
            return response.get("result")

        error = response.get(
            "error",
            "Unknown error",
        )

        if response.get("error_type") == "authorization":
            raise AuthorizationError(error)

        if response.get("error_type") == "database":
            raise DatabaseError(error)

        raise IPCError(error)


class Database:
    """
    Compatibility interface for the GUI.

    The GUI keeps using Database.* methods, but all operations
    now go through the root parental-control daemon instead of
    accessing SQLite directly.
    """

    @staticmethod
    def get_users():
        return IPCClient.call(
            "get_users"
        )

    @staticmethod
    def get_user(user_id):
        return IPCClient.call(
            "get_user",
            user_id=user_id,
        )

    @staticmethod
    def get_remaining_time(user_id):
        result = IPCClient.call(
            "get_remaining_time",
            user_id=user_id,
        )

        return (
            result[0],
            result[1],
        )

    @staticmethod
    def get_allowances(user_id):
        result = IPCClient.call(
            "get_allowances",
            user_id=user_id,
        )

        return {
            int(key): value
            for key, value in result.items()
        }

    @staticmethod
    def get_access_windows(user_id):
        result = IPCClient.call(
            "get_access_windows",
            user_id=user_id,
        )

        return {
            int(key): [
                tuple(window)
                for window in windows
            ]
            for key, windows in result.items()
        }

    @staticmethod
    def create_user(
        username,
        enabled,
        allowances,
        access_windows,
    ):
        return IPCClient.call(
            "create_user",
            username=username,
            enabled=enabled,
            allowances={
                str(key): value
                for key, value in allowances.items()
            },
            access_windows={
                str(key): [
                    list(window)
                    for window in windows
                ]
                for key, windows in access_windows.items()
            },
        )

    @staticmethod
    def update_user(
        user_id,
        enabled,
        allowances,
        access_windows,
    ):
        return IPCClient.call(
            "update_user",
            user_id=user_id,
            enabled=enabled,
            allowances={
                str(key): value
                for key, value in allowances.items()
            },
            access_windows={
                str(key): [
                    list(window)
                    for window in windows
                ]
                for key, windows in access_windows.items()
            },
        )

    @staticmethod
    def delete_user(user_id):
        return IPCClient.call(
            "delete_user",
            user_id=user_id,
        )

    @staticmethod
    def set_enabled(
        user_id,
        enabled,
    ):
        return IPCClient.call(
            "set_enabled",
            user_id=user_id,
            enabled=enabled,
        )

    @staticmethod
    def add_temporary_time(
        user_id,
        seconds,
    ):
        return IPCClient.call(
            "add_temporary_time",
            user_id=user_id,
            seconds=seconds,
        )