Надёжные сетевые приложения с помощью asyncio, socket и TLS

Сетевые сценарии для повторного использования

Приведённые примеры расширяют базовую работу с сокетами и демонстрируют совмещение разных модулей Python 3.

Как организовать локальный обмен данными через Unix domain socket?

Unix socket привязан к файлу в файловой системе. Это удобно для связи контейнеров и процессов на одном хосте.

import os
import socket

path = '/tmp/py_net_service.sock'
if os.path.exists(path):
    os.remove(path)

server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server.bind(path)
server.listen(1)
print('Ожидание клиента')
connection, _ = server.accept()
with connection:
    request = connection.recv(1024)
    print('Запрос:', request.decode())
    connection.sendall(b'ответ локального сервера')
server.close()

Клиент подключается к пути без указания port.

import socket

client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
client.connect('/tmp/py_net_service.sock')
client.sendall(b'hello from client')
response = client.recv(1024)
print('Ответ:', response.decode())
client.close()

Сервер выводит запрос и отправляет ответ. Дополнительный пример вывода для сервера:

Ожидание клиента
Запрос: hello from client

Для клиента результат будет таким.

Ответ: ответ локального сервера

Как проверить порты с таймаутом с помощью asyncio?

Функция open_connection пытается установить TCP-соединение. Обёртка wait_for ограничивает время ожидания.

import asyncio

async def check_port(host, port, timeout=1.5):
    try:
        reader, writer = await asyncio.wait_for(
            asyncio.open_connection(host, port), timeout)
    except (asyncio.TimeoutError, OSError):
        return False
    writer.close()
    await writer.wait_closed()
    return True

async def scan(host, ports):
    tasks = [asyncio.create_task(check_port(host, p)) for p in ports]
    results = await asyncio.gather(*tasks)
    return [port for port, is_open in zip(ports, results) if is_open]

if __name__ == '__main__':
    open_ports = asyncio.run(scan('127.0.0.1', [22, 80, 443, 8000, 8888]))
    print('Открытые порты:', open_ports)

Когда хост недоступен, возникает OSError, и функция возвращает False. Вывод может выглядеть так.

Открытые порты: [8000, 8888]

Как отправить защищённый HTTP-запрос с проверкой сертификата?

Модуль ssl создаёт контекст с сертификатами доверенных центров. Класс HTTPSConnection применяет этот контекст при установке TLS-сессии.

import http.client
import ssl

context = ssl.create_default_context()
connection = http.client.HTTPSConnection('example.com', timeout=5, context=context)
connection.request('GET', '/')
response = connection.getresponse()
body = response.read(100)
print(response.status, response.reason)
print(body.decode('utf-8', errors='replace').strip())
connection.close()

Программа выполнит TLS-руккопожатие и проверит имя хоста в сертификате. Возможный фрагмент вывода.

200 OK



    

Как реализовать UDP-приёмник на основе асинхронного протокола?

DatagramProtocol подходит для неблокирующего UDP в приложениях asyncio.

import asyncio

class UdpEchoProtocol(asyncio.DatagramProtocol):
    def connection_made(self, transport):
        self.transport = transport
        print('UDP транспорт запущен')

    def datagram_received(self, data, addr):
        message = data.decode()
        print(f'Пакет от {addr[0]}:{addr[1]}: {message}')
        self.transport.sendto(f'эхо: {message}'.encode(), addr)

    def error_received(self, exception):
        print('Ошибка сокета:', exception)

async def run_udp():
    loop = asyncio.get_running_loop()
    transport, _ = await loop.create_datagram_endpoint(
        UdpEchoProtocol,
        local_addr=('127.0.0.1', 8889)
    )
    try:
        await asyncio.Event().wait()
    finally:
        transport.close()

if __name__ == '__main__':
    try:
        asyncio.run(run_udp())
    except KeyboardInterrupt:
        pass

Синхронный клиент отправляет дейтаграмму на порт 8889.

import socket

with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
    client.sendto(b'hello async udp', ('127.0.0.1', 8889))
    client.settimeout(2)
    try:
        response, _ = client.recvfrom(1024)
        print(response.decode())
    except socket.timeout:
        print('Таймаут')

Вывод сервера.

UDP транспорт запущен
Пакет от 127.0.0.1:60001: hello async udp

Вывод клиента.

эхо: hello async udp