import json import os import socket import struct import subprocess import threading from pathlib import Path from .database import get_db SOCKET_PATH = Path("/run/parental-control.sock") BUFFER_SIZE = 65536 POLKIT_ACTION = "org.sarlink.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 class IPCOperations: @staticmethod def get_users(): with get_db() as db: rows = db.execute( """ SELECT * FROM users ORDER BY username COLLATE NOCASE """ ).fetchall() return [dict(row) for row in rows] @staticmethod def get_user(user_id): with get_db() as db: row = db.execute( """ SELECT * FROM users WHERE id = ? """, (user_id,), ).fetchone() return dict(row) if row else None @staticmethod def get_remaining_time(user_id): with get_db() as db: allowance_row = db.execute( """ SELECT COALESCE( allowance_seconds, 0 ) AS total FROM daily_allowances WHERE user_id = ? AND weekday = CAST( strftime('%w', 'now', 'localtime') AS INTEGER ) """, (user_id,), ).fetchone() allowance = ( allowance_row["total"] if allowance_row else 0 ) usage_row = db.execute( """ SELECT COALESCE( used_seconds, 0 ) AS total FROM usage WHERE user_id = ? AND date = date('now', 'localtime') """, (user_id,), ).fetchone() used = ( usage_row["total"] if usage_row else 0 ) remaining = max( allowance - used, 0, ) return remaining, allowance @staticmethod def get_allowances(user_id): with get_db() as db: rows = db.execute( """ SELECT weekday, allowance_seconds FROM daily_allowances WHERE user_id = ? """, (user_id,), ).fetchall() return { row["weekday"]: row["allowance_seconds"] for row in rows } @staticmethod def get_access_windows(user_id): with get_db() as db: rows = db.execute( """ SELECT weekday, start_minute, end_minute FROM access_windows WHERE user_id = ? ORDER BY weekday, start_minute """, (user_id,), ).fetchall() result = {} for row in rows: result.setdefault( row["weekday"], [], ).append( ( row["start_minute"], row["end_minute"], ) ) return result @staticmethod def create_user( username, enabled, allowances, access_windows, ): with get_db() as db: cursor = db.execute( """ INSERT INTO users ( username, enabled ) VALUES (?, ?) """, ( username, 1 if enabled else 0, ), ) user_id = cursor.lastrowid for weekday, seconds in allowances.items(): db.execute( """ INSERT INTO daily_allowances ( user_id, weekday, allowance_seconds ) VALUES (?, ?, ?) """, ( user_id, weekday, seconds, ), ) for weekday, windows in access_windows.items(): for start_minute, end_minute in windows: db.execute( """ INSERT INTO access_windows ( user_id, weekday, start_minute, end_minute ) VALUES (?, ?, ?, ?) """, ( user_id, weekday, start_minute, end_minute, ), ) return user_id @staticmethod def update_user( user_id, enabled, allowances, access_windows, ): with get_db() as db: db.execute( """ UPDATE users SET enabled = ? WHERE id = ? """, ( 1 if enabled else 0, user_id, ), ) for weekday, seconds in allowances.items(): db.execute( """ INSERT INTO daily_allowances ( user_id, weekday, allowance_seconds ) VALUES (?, ?, ?) ON CONFLICT(user_id, weekday) DO UPDATE SET allowance_seconds = excluded.allowance_seconds """, ( user_id, weekday, seconds, ), ) db.execute( """ DELETE FROM access_windows WHERE user_id = ? """, (user_id,), ) for weekday, windows in access_windows.items(): for start_minute, end_minute in windows: db.execute( """ INSERT INTO access_windows ( user_id, weekday, start_minute, end_minute ) VALUES (?, ?, ?, ?) """, ( user_id, weekday, start_minute, end_minute, ), ) @staticmethod def delete_user(user_id): with get_db() as db: db.execute( """ DELETE FROM users WHERE id = ? """, (user_id,), ) @staticmethod def set_enabled(user_id, enabled): with get_db() as db: db.execute( """ UPDATE users SET enabled = ? WHERE id = ? """, ( 1 if enabled else 0, user_id, ), ) @staticmethod def add_temporary_time( user_id, seconds, ): with get_db() as db: db.execute( """ INSERT INTO temporary_grants ( user_id, seconds, remaining_seconds ) VALUES (?, ?, ?) """, ( user_id, seconds, seconds, ), ) 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 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,"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()