Клиент-серверный IPC на Python multiprocessing

Статья отражает личный опыт разработки CLI приложения для Linux.

В ней рассмотрен способ выполнения привилегированных системных вызовов процессом суперпользователя по запросам управляющей программы через строго описанный API.

Исходный код написан на Python для реального коммерческого приложения, но для публикации абстрагирован от конкретных задач.

Введение

«Межпроцессное взаимодействие (англ. inter-process communication, IPC) — обмен данными между потоками одного или разных процессов. Реализуется посредством механизмов, предоставляемых ядром ОС или процессом, использующим механизмы ОС и реализующим новые возможности IPC». — Википедия

У процессов могут быть разные причины для обмена информацией. На мой взгляд, все они являются следствием политики безопасности ядра Unix.

Как известно, ядро Unix — это автономная система, которая функционирует без вмешательства человека. Собственно говоря, пользователь — это объект операционной системы, который появился чтобы обезопасить ядро от несанкционированного вмешательства.

Обеспечение безопасности ядра заключается в разделении адресного пространства операционной системы на пространство ядра и пространство пользователя. Отсюда два режима работы системы: режим пользователя и режим ядра. Причем, смена режимов — это переключение между двумя пространствами.

В режиме пользователя недоступны области памяти, зарезервированные ядром, как и системные вызовы, которые изменяют состояние системы.

Тем не менее таким доступом обладает суперпользователь.

Предпосылки параллелизма

Если ваша программа не использует привилегированные системные вызовы, вам не нужен суперпользователь, а значит можно писать монолит без параллелизма.

В противном случае вам придётся запускать свою программу под рутом.

На этом можно было бы закончить, если бы не проблема доступа к эксклюзивным ресурсам пользовательского окружения из окружения суперпользователя.

Представьте, что вы обращаетесь к ресурсу или к переменной окружения, которая присутствует только в пользовательском пространстве.

Например, служба, которая вам нужна, зависит от окружения рабочего стола, которое доступно только когда пользовательская сессия активна. В этом случае объект, который вам нужен просто не существует в окружении суперпользователя. Для того, чтобы получить к нему доступ, вам нужен процесс в пользовательском окружении, а для системных вызовов — процесс в руте.

При этом вы можете запросить у процесса в руте исполнение системного вызова из пользовательского процесса при помощи одного из методов IPC.

Для своего приложения я выбрал сокеты и написал API для коммуникации между процессами.

Таким образом в распоряжении конечного пользователя оказывается программа, которая не требует прав суперпользователя, но запрашивает исполнение системных вызовов у процесса в руте через соответствующий сокет.

При этом процесс в руте запускается при загрузке системы и остаётся активным всегда, прослушивая сокет на наличие входящих дейтаграмм.

Историческая справка

Традиционно процессы, которые запускаются при загрузке системы и остаются активными в фоне, классифицируются как daemon. Имена исполняемых файлов таких программ по соглашению заканчиваются на «d». Пример: systemd.

Программы пользовательского пространства, взаимодействующие с daemon можно назвать управляющими, что также по соглашению отражено в их названиях. Пример: systemctl.

Известны и другие примеры: ssh и sshd.

В более современной интерпретации их называют клиентами и серверами с тем отличием, что последние взаимодействуют по сетевым протоколам. Однако, это не мешает нам использовать сокеты исключительно для локальных процессов когда сокет принимает дейтаграммы через конвейер.

Структура проекта

Для сервера и клиента я использую одинаковую структуру.

.
├── core
│   ├── api.py
│   └── __init__.py
├── main.py

core — это пакет, в который можно положить модули с любой логикой. В модуле api реализованы методы обращения процессов друг к другу.

Реализация API клиента

from multiprocessing.connection import Client
from multiprocessing.connection import Listener

# адрес сервера (процесса в руте) для исходящих
# запросов
daemon = ('localhost', 6000)
# адрес клиента (этого процесса) для входящих
# ответов от сервера
cli = ('localhost', 6001)

def send(request: dict) -> bool or dict:
    """
    Принимает словарь аргументов удалённого метода.
    Отправляет запрос, после чего открывет сокет
    и ждет на нем ответ от сервера.
    """
    with Client(daemon) as conn:
        conn.send(request)
    with Listener(cli) as listener:
        with listener.accept() as conn:
            try:
                return conn.recv()
            except EOFError:
                return False

def hello(name: str) -> send:
    """
    Формирует уникальный запрос и вызывает функцию
    send для его отправки.
    """
    return send({
        "method": "hello",
        "name": name
    })

В модуле connection пакета multiprocessing есть два класса, реализующих API высокого уровня над низкоуровнивым аналогом стандартной библиотеки — socket.

Client — класс, который содержит методы отправки дейтаграмм.

Listener принимает дейтаграммы.

Отправляемые запросы содержат название целевого метода сервера.

Причем запросы не требуют никаких преобразований на сервере, ведь он тоже написан на Python, который интерпретирует поступающие данные также, как и клиент. Всё это происходит «под капотом» и не может не радовать.

Использование API

В main.py я импортирую модуль api для дальнейшего использования.

from core import api

response = api.hello("World!")
print(response)

Этот код представлен для демонстрации. В работе я использовал Сlick Framework для создания СLI приложения с опциями, которые вызывают методы API.

Реализация API сервера

По идее метод должен выполнять системный вызов, который требует прав суперпользователя. Иначе всё это теряет смысл.

def hello(request: dict) -> str:
    """
    Привилегированный системный вызов.
    """
    return " ".join(["Hello", request["name"])

Использование API

from core import api
import logger
from multiprocessing.connection import Listener
from multiprocessing.connection import Client

# адрес сервера (этого процесса) для входящих запросов
daemon = ('localhost', 6000)
# адрес клиента для исходящих ответов
cli = ('localhost', 6001)
with Listener(daemon) as listener:
    response = None
    while not response:
        request = None
        while not request:
            with listener.accept() as conn:
                request = conn.recv()
        if request["method"] == "connect":
            response = api.connect(request)
        elif request["method"] == "disconnect":
            response = api.disconnect(request)
        else:
            response = None
        if response:
          try:
            with Client(cli) as conn:
                conn.send(response)
          except ConnectionRefusedError as e:
              logger.warning(e)
          else:
            response = None

Сервер должен быть активен всегда, поэтому я запускаю его в бесконечном цикле.

Таким образом он всегда слушает порт 6000 и, при поступлении дейтаграммы, анализирует запрос. Затем он вызывает указанный в запросе метод и возвращает результат исполнения клиенту.

Дополнительно

Советую снабдить свой сервер пакетом systemd, который позволяет программам на Python писать лог в journald.

Для сборки вы можете использовать pyinstaller — он запакует ваш код в бинарный файл со всеми зависимостями. Не забудьте про соглашение о наименовании исполняемых файлов, упомянутое ранее.

Спасибо за внимание!