from __future__ import annotations import json import os import socket import struct import subprocess import threading from pathlib import Path from typing import Any from .config import ( DAY_NAMES, find_user, load_users_config, next_user_id, normalize_access_windows, normalize_allowances, save_users_config, ) from .enforcement import ( get_remaining_grant_seconds, ) from .users import user_exists SOCKET_PATH = Path( "/run/parental-control.sock" ) BUFFER_SIZE = 65536 POLKIT_ACTION = ( "org.parentalcontrol.modify" ) def get_peer_credentials( connection, ): credentials = connection.getsockopt( socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i"), ) return struct.unpack( "3i", credentials, ) def authorize_write( pid: int, ) -> bool: try: os.kill( pid, 0, ) except OSError: return False try: result = subprocess.run( [ "pkcheck", "--action-id", POLKIT_ACTION, "--process", str(pid), "--allow-user-interaction", ], capture_output=True, text=True, check=False, ) return result.returncode == 0 except FileNotFoundError: return False def _user_to_api( user: dict[str, Any], ) -> dict[str, Any]: user_id = int( user.get( "id", 0, ) ) username = str( user.get( "username", "", ) ) enabled = bool( user.get( "enabled", True, ) ) allowances = normalize_allowances( user.get( "daily_allowance", {}, ) ) access_windows = normalize_access_windows( user.get( "access_windows", {}, ) ) return { "id": user_id, "username": username, "enabled": enabled, "daily_allowance": allowances, "allowances": allowances, "access_windows": access_windows, } def _find_user_or_raise( user_id: int, ) -> tuple[dict, dict, int]: config = load_users_config() try: requested_id = int( user_id ) except ( TypeError, ValueError, ): raise ValueError( f"Invalid user id: {user_id}" ) for index, user in enumerate( config["users"] ): try: current_id = int( user.get("id") ) except ( TypeError, ValueError, ): continue if current_id == requested_id: return ( config, user, index, ) raise ValueError( f"User not found: {user_id}" ) def _parse_access_windows( access_windows, ) -> dict: return normalize_access_windows( access_windows ) def _parse_allowances( allowances, ) -> dict: """ Normalize allowance data coming from the GUI. The original GUI uses numeric weekday indexes: 0 = sunday 1 = monday 2 = tuesday 3 = wednesday 4 = thursday 5 = friday 6 = saturday The configuration format uses weekday names. Accept both formats so the IPC layer remains compatible with the original GUI and the config file format. """ if not isinstance( allowances, dict, ): return normalize_allowances( {} ) converted = {} for key, value in allowances.items(): # Numeric weekday index from the original GUI. try: numeric_key = int( key ) except ( TypeError, ValueError, ): numeric_key = None if ( numeric_key is not None and 0 <= numeric_key < len(DAY_NAMES) ): weekday = DAY_NAMES[ numeric_key ] else: weekday = str( key ).lower().strip() converted[ weekday ] = value return normalize_allowances( converted ) class IPCOperations: @staticmethod def get_users(): config = load_users_config() users = [] for user in config.get( "users", [], ): try: users.append( _user_to_api( user ) ) except ( TypeError, ValueError, ): continue users.sort( key=lambda item: item["username"].lower() ) return users @staticmethod def get_user( user_id, ): user = find_user( int(user_id) ) if user is None: return None return _user_to_api( user ) @staticmethod def get_remaining_time( user_id, ): user = find_user( int(user_id) ) if user is None: return 0, 0 from datetime import datetime from .config import ( python_weekday_to_name, load_state, ) weekday = python_weekday_to_name( datetime.now().weekday() ) allowances = normalize_allowances( user.get( "daily_allowance", {}, ) ) allowance = allowances.get( weekday, 0, ) username = str( user.get( "username", "", ) ) today = ( datetime.now() .date() .isoformat() ) state = load_state() usage = state.get( "usage", {}, ) used = 0 if isinstance( usage, dict, ): user_usage = usage.get( username, {}, ) if isinstance( user_usage, dict, ): try: used = int( user_usage.get( today, 0, ) ) except ( TypeError, ValueError, ): used = 0 remaining = max( 0, allowance - used, ) grant = get_remaining_grant_seconds( int(user_id) ) return ( remaining + grant, allowance, ) @staticmethod def get_allowances( user_id, ): user = find_user( int(user_id) ) if user is None: return {} return normalize_allowances( user.get( "daily_allowance", {}, ) ) @staticmethod def get_access_windows( user_id, ): user = find_user( int(user_id) ) if user is None: return {} return normalize_access_windows( user.get( "access_windows", {}, ) ) @staticmethod def create_user( username, enabled, allowances, access_windows, ): username = str( username ).strip() if not username: raise ValueError( "Username cannot be empty" ) if not user_exists( username ): raise ValueError( f"Linux user does not exist: " f"{username}" ) config = load_users_config() for existing in config[ "users" ]: if str( existing.get( "username", "", ) ) == username: raise ValueError( f"User already configured: " f"{username}" ) users = config[ "users" ] user_id = next_user_id( users ) user = { "id": user_id, "username": username, "enabled": bool( enabled ), "daily_allowance": _parse_allowances( allowances ), "access_windows": _parse_access_windows( access_windows ), } users.append( user ) save_users_config( config ) return user_id @staticmethod def update_user( user_id, enabled, allowances, access_windows, ): config, user, _ = ( _find_user_or_raise( user_id ) ) user[ "enabled" ] = bool( enabled ) user[ "daily_allowance" ] = _parse_allowances( allowances ) user[ "access_windows" ] = _parse_access_windows( access_windows ) save_users_config( config ) return True @staticmethod def delete_user( user_id, ): config, user, index = ( _find_user_or_raise( user_id ) ) username = str( user.get( "username", "", ) ) from .config import ( load_state, save_state, ) config["users"].pop( index ) save_users_config( config ) state = load_state() usage = state.get( "usage", {}, ) if isinstance( usage, dict, ): usage.pop( username, None, ) grants = state.get( "temporary_grants", [], ) if isinstance( grants, list, ): state[ "temporary_grants" ] = [ grant for grant in grants if str( grant.get( "username", "", ) ) != username ] save_state( state ) return True @staticmethod def set_enabled( user_id, enabled, ): config, user, _ = ( _find_user_or_raise( user_id ) ) user[ "enabled" ] = bool( enabled ) save_users_config( config ) return True @staticmethod def add_temporary_time( user_id, seconds, ): from datetime import ( datetime, timedelta, ) user = find_user( int(user_id) ) if user is None: raise ValueError( f"User not found: {user_id}" ) try: seconds = int( seconds ) except ( TypeError, ValueError, ): raise ValueError( "Invalid temporary time" ) if seconds <= 0: raise ValueError( "Temporary time must " "be greater than zero" ) username = str( user.get( "username", "", ) ) from .config import ( load_state, save_state, ) state = load_state() grants = state.setdefault( "temporary_grants", [], ) grant = { "username": username, "seconds": seconds, "remaining_seconds": seconds, "created_at": datetime.now().isoformat(), "expires_at": ( datetime.now() + timedelta( seconds=seconds ) ).isoformat(), } grants.append( grant ) save_state( state ) return True READ_METHODS = { "get_users", "get_user", "get_remaining_time", "get_allowances", "get_access_windows", } WRITE_METHODS = { "create_user", "update_user", "delete_user", "set_enabled", "add_temporary_time", } def handle_request( request, pid, ): method = request.get( "method" ) arguments = request.get( "args", {}, ) if method in READ_METHODS: operation = getattr( IPCOperations, method, None, ) if operation is None: raise RuntimeError( f"Unknown method: {method}" ) return operation( **arguments ) if method in WRITE_METHODS: if not authorize_write( pid ): raise PermissionError( "Administrative authorization required" ) operation = getattr( IPCOperations, method, None, ) if operation is None: raise RuntimeError( f"Unknown method: {method}" ) return operation( **arguments ) raise RuntimeError( f"Unknown IPC method: {method}" ) def handle_client( connection, ): try: pid, uid, gid = ( get_peer_credentials( connection ) ) data = bytearray() while True: chunk = connection.recv( BUFFER_SIZE ) if not chunk: break data.extend( chunk ) if b"\n" in chunk: break if not data: return line = bytes( data ).split( b"\n", 1, )[0] request = json.loads( line.decode( "utf-8" ) ) try: result = handle_request( request, pid, ) response = { "ok": True, "result": result, } except PermissionError as exc: response = { "ok": False, "error": str(exc), "error_type": "authorization", } except ValueError as exc: response = { "ok": False, "error": str(exc), "error_type": "validation", } except Exception as exc: response = { "ok": False, "error": str(exc), "error_type": "server", } payload = ( json.dumps( response, separators=( ",", ":", ), ).encode( "utf-8" ) + b"\n" ) connection.sendall( payload ) except Exception: try: connection.sendall( b'{"ok":false,' b'"error":"Invalid request",' b'"error_type":"server"}\n' ) except Exception: pass finally: connection.close() class IPCServer: def __init__( self, socket_path=SOCKET_PATH, ): self.socket_path = Path( socket_path ) self.server_socket = None self._thread = None self._stop_event = ( threading.Event() ) def start(self): if self._thread is not None: return self.socket_path.parent.mkdir( parents=True, exist_ok=True, ) try: self.socket_path.unlink() except FileNotFoundError: pass self.server_socket = socket.socket( socket.AF_UNIX, socket.SOCK_STREAM, ) self.server_socket.bind( str( self.socket_path ) ) os.chmod( self.socket_path, 0o666, ) self.server_socket.listen( 16 ) self._stop_event.clear() self._thread = threading.Thread( target=self._run, name="parental-control-ipc", daemon=True, ) self._thread.start() def stop(self): self._stop_event.set() if self.server_socket is not None: try: self.server_socket.close() except Exception: pass if self._thread is not None: self._thread.join( timeout=2 ) self._thread = None self.server_socket = None try: self.socket_path.unlink() except FileNotFoundError: pass def _run(self): while not self._stop_event.is_set(): try: connection, _ = ( self.server_socket.accept() ) except OSError: if self._stop_event.is_set(): break continue thread = threading.Thread( target=handle_client, args=(connection,), daemon=True, ) thread.start() ipc_server = IPCServer() def start_ipc_server(): ipc_server.start() def stop_ipc_server(): ipc_server.stop()