GOSTIM: P2P F2F E2EE IM за один вечер с ГОСТ-криптографией

Будучи разработчиком PyGOST библиотеки (ГОСТовые криптографические примитивы на чистом Python), я нередко получаю вопросы о том как на коленке реализовать простейший безопасный обмен сообщениями. Многие считают прикладную криптографию достаточно простой штукой, и .encrypt() вызова у блочного шифра будет достаточно для безопасной отсылки по каналу связи. Другие же считают, что прикладная криптография — удел немногих, и приемлемо, что богатые компании типа Telegram с олимпиадниками-математиками не могут реализовать безопасный протокол.

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

В статье будет написан peer-to-peer, friend-to-friend, end-to-end зашифрованный instant messenger с SIGMA-I протоколом аутентификации и согласования ключей (на базе которого реализован IPsec IKE), используя исключительно ГОСТовые криптографические алгоритмы PyGOST библиотеки и ASN.1 кодирование сообщений библиотекой PyDERASN (про которую я уже писал раньше). Необходимое условие: он должен быть настолько прост, чтобы его можно было написать с нуля за один вечер (или рабочий день), иначе это уже не простая программа. В ней наверняка есть ошибки, излишние сложности, недочёты, плюс это моя первая программа с использованием asyncio библиотеки.

Дизайн IM

Для начала, надо понять как будет выглядеть наш IM. Для простоты, пускай это будет peer-to-peer сеть, без какого-либо обнаружения участников. Собственноручно будем указывать к какому адресу: порту подключаться для общения с собеседником.

Я понимаю, что, на данный момент, предположение о доступности прямой связи между двумя произвольными компьютерами — существенное ограничение применимости IM на практике. Но чем больше разработчиков будут реализовывать всякие NAT-traversal костыли, тем дольше мы так и будем оставаться в IPv4 Интернете, с удручающей вероятностью связи между произвольными компьютерами. Ну сколько можно терпеть отсутствие IPv6 дома и на работе?

У нас будет friend-to-friend сеть: все возможные собеседники заранее должны быть известны. Во-первых, это сильно всё упрощает: представились, нашли или не нашли имя/ключ, отключились или продолжаем работу, зная собеседника. Во-вторых, в общем случае, это безопасно и исключает множество атак.

Интерфейс IM-а будет близок к классическим решениям suckless-проектов, которые мне очень нравятся своим минимализмом и Unix-way философией. IM программа для каждого собеседника создаёт директорию с тремя Unix domain socket:

• in — в него записываются отправляемые собеседнику сообщения;
• out — из него читаются принимаемые от собеседника сообщения;
• state — читая из него, мы узнаём подключён ли сейчас собеседник, адрес/порт подключения.

Кроме того, создаётся conn сокет, записав в который хост порт, мы инициируем подключение к удалённому собеседнику.

|-- alice
|   |-- in
|   |-- out
|   `-- state
|-- bob
|   |-- in
|   |-- out
|   `-- state
`- conn

Такой подход позволяет делать независимые реализации IM транспорта и пользовательского интерфейса, ведь на вкус и цвет товарища нет, каждому не угодишь. Используя tmux и/или multitail, можно получить многооконный интерфейс с синтаксической подсветкой. А с помощью rlwrap можно получить GNU Readline-совместимую строку для ввода сообщений.

На самом деле, suckless проекты используют FIFO-файлы. Лично я не смог понять как в asyncio работать с файлами конкурентно без собственноручной подложки из выделенных тредов (для таких вещей давно использую язык Go). Поэтому решил обойтись Unix domain сокетами. К сожалению, это лишает возможности сделать echo 2001:470:dead::babe 6666 > conn. Я решил эту проблему, используя socat: echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.

Первоначальный небезопасный протокол

В качестве транспорта используется TCP: он гарантирует доставку и её порядок. UDP не гарантирует ни того, ни другого (что было бы полезным, когда применится криптография), а поддержки SCTP в Python из коробки нет.

К сожалению, в TCP нет понятия сообщения, а только потока байт. Поэтому необходимо придумать формат для сообщений, чтобы их можно было разделять между собой в этом потоке. Можем условиться использовать символ перевода строки. Для начала подойдёт, однако, когда мы начнём шифровать наши сообщения, этот символ может появиться где угодно в шифротексте. В сетях поэтому популярны протоколы отправляющие сначала длину сообщения в байтах. Например, в Python из коробки есть xdrlib позволяющая работать с подобным форматом XDR.

Мы не будем правильно и эффективно работать с TCP чтением — упростим код. Читаем в бесконечном цикле данные из сокета, пока не декодируем полное сообщение. В качестве формата для такого подхода можно использовать и JSON с XML. Но, когда добавится криптография, то данные придётся подписывать и аутентифицировать — а это потребует байт-в-байт идентичного представления объектов, чего не обеспечивают JSON/XML (dumps результат может отличаться).

XDR подходит для такой задачи, однако я выбираю ASN.1 с DER-кодированием и PyDERASN библиотеку, так как на руках у нас будут высокоуровневые объекты с которыми часто приятнее и удобнее работать. В отличии от schemaless bencode, MessagePack или CBOR, ASN.1 автоматически проверит данные напротив жёстко заданной схемы.

# Msg ::= CHOICE {
#       text      MsgText,
#       handshake [0] EXPLICIT MsgHandshake }
class Msg(Choice):
    schema = ((
        ("text", MsgText()),
        ("handshake", MsgHandshake(expl=tag_ctxc(0))),
    ))

# MsgText ::= SEQUENCE {
#       text UTF8String (SIZE(1..MaxTextLen))}
class MsgText(Sequence):
    schema = ((
        ("text", UTF8String(bounds=(1, MaxTextLen))),
    ))

# MsgHandshake ::= SEQUENCE {
#       peerName UTF8String (SIZE(1..256)) }
class MsgHandshake(Sequence):
    schema = ((
        ("peerName", UTF8String(bounds=(1, 256))),
    ))

Принимаемым сообщением будет Msg: либо текстовое MsgText (пока с одним текстовым полем), либо сообщение рукопожатия MsgHandshake (в котором передаётся имя собеседника). Сейчас выглядит переусложнённым, но это задел на будущее.

     ┌─────┐            ┌─────┐
     │PeerA│            │PeerB│
     └──┬──┘            └──┬──┘
        │MsgHandshake(IdA) │
        │─────────────────>│
        │                  │
        │MsgHandshake(IdB) │
        │<─────────────────│
        │                  │
        │    MsgText()     │
        │─────────────────>│
        │                  │
        │    MsgText()     │
        │<─────────────────│
        │                  │

IM без криптографии

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

parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
    "--our-name",
    required=True,
    help="Our peer name",
)
parser.add_argument(
    "--their-names",
    required=True,
    help="Their peer names, comma-separated",
)
parser.add_argument(
    "--bind",
    default="::1",
    help="Address to listen on",
)
parser.add_argument(
    "--port",
    type=int,
    default=6666,
    help="Port to listen on",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))

Задаётся собственное имя (—our-name alice). Через запятую перечисляются все ожидаемые собеседники (—their-names bob,eve). Для каждого из собеседников, создаётся директория с Unix сокетами, а также по корутине на каждый in, out, state:

for peer_name in THEIR_NAMES:
    makedirs(peer_name, mode=0o700, exist_ok=True)
    out_queue = asyncio.Queue()
    OUT_QUEUES[peer_name] = out_queue
    asyncio.ensure_future(asyncio.start_unix_server(
        partial(unixsock_out_processor, out_queue=out_queue),
        path.join(peer_name, "out"),
    ))
    in_queue = asyncio.Queue()
    IN_QUEUES[peer_name] = in_queue
    asyncio.ensure_future(asyncio.start_unix_server(
        partial(unixsock_in_processor, in_queue=in_queue),
        path.join(peer_name, "in"),
    ))
    asyncio.ensure_future(asyncio.start_unix_server(
        partial(unixsock_state_processor, peer_name=peer_name),
        path.join(peer_name, "state"),
    ))
asyncio.ensure_future(asyncio.start_unix_server(unixsock_conn_processor, "conn"))

Приходящие от пользователя сообщения из in сокета отправляются в IN_QUEUES очереди:

async def unixsock_in_processor(reader, writer, in_queue: asyncio.Queue) -> None:
    while True:
        text = await reader.read(MaxTextLen)
        if text == b"":
            break
        await in_queue.put(text.decode("utf-8"))

Приходящие от собеседников сообщения отправляются в OUT_QUEUES очереди, из которых данные записываются в out сокет:

async def unixsock_out_processor(reader, writer, out_queue: asyncio.Queue) -> None:
    while True:
        text = await out_queue.get()
        writer.write(("[%s] %s" % (datetime.now(), text)).encode("utf-8"))
        await writer.drain()

При чтении из state сокета, программа ищет в PEER_ALIVE словаре адрес собеседника. Если подключения к собеседнику ещё нет, то записывается пустая строка.

async def unixsock_state_processor(reader, writer, peer_name: str) -> None:
    peer_writer = PEER_ALIVES.get(peer_name)
    writer.write(
        b"" if peer_writer is None else (" ".join([
            str(i) for i in peer_writer.get_extra_info("peername")[:2]
        ]).encode("utf-8") + b"n")
    )
    await writer.drain()
    writer.close()

При записи адреса в conn сокет, запускается функция «инициатора» соединения:

async def unixsock_conn_processor(reader, writer) -> None:
    data = await reader.read(256)
    writer.close()
    host, port = data.decode("utf-8").split(" ")
    await initiator(host=host, port=int(port))

Рассмотрим инициатора. Сначала он, очевидно, открывает соединение до указанного хоста/порта и отправляет handshake сообщение со своим именем:

 130 async def initiator(host, port):
 131     _id = repr((host, port))
 132     logging.info("%s: dialing", _id)
 133     reader, writer = await asyncio.open_connection(host, port)
 134     # Handshake message {{{
 135     writer.write(Msg(("handshake", MsgHandshake((
 136         ("peerName", OUR_NAME),
 137     )))).encode())
 138     # }}}
 139     await writer.drain()

Затем, ждёт ответа от удалённой стороны. Пытается декодировать пришедший ответ по Msg ASN.1 схеме. Предполагаем что всё сообщение будет отправлено одним TCP-сегментом и мы атомарно его получим при вызове .read(). Проверяем что мы получили именно handshake сообщение.

 141     # Wait for Handshake message {{{
 142     data = await reader.read(256)
 143     if data == b"":
 144         logging.warning("%s: no answer, disconnecting", _id)
 145         writer.close()
 146         return
 147     try:
 148         msg, _ = Msg().decode(data)
 149     except ASN1Error:
 150         logging.warning("%s: undecodable answer, disconnecting", _id)
 151         writer.close()
 152         return
 153     logging.info("%s: got %s message", _id, msg.choice)
 154     if msg.choice != "handshake":
 155         logging.warning("%s: unexpected message, disconnecting", _id)
 156         writer.close()
 157         return
 158     # }}}

Проверяем что пришедшее имя собеседника нам известно. Если нет, то рвём соединение. Проверяем не было ли у нас уже установлено с ним соединение (собеседник вновь дал команду на подключение к нам) и закрываем его. В IN_QUEUES очередь помещаются Python-строки с текстом сообщения, но имеется особое значение None, сигнализирующее msg_sender корутину прекратить работу, чтобы она забыла о своём writer, связанным с устаревшим TCP-соединением.

 159     msg_handshake = msg.value
 160     peer_name = str(msg_handshake["peerName"])
 161     if peer_name not in THEIR_NAMES:
 162         logging.warning("unknown peer name: %s", peer_name)
 163         writer.close()
 164         return
 165     logging.info("%s: session established: %s", _id, peer_name)
 166     # Run text message sender, initialize transport decoder {{{
 167     peer_alive = PEER_ALIVES.pop(peer_name, None)
 168     if peer_alive is not None:
 169         peer_alive.close()
 170         await IN_QUEUES[peer_name].put(None)
 171     PEER_ALIVES[peer_name] = writer
 172     asyncio.ensure_future(msg_sender(peer_name, writer))
 173     # }}}

msg_sender принимает исходящие сообщения (подкладываемые в очередь из in сокета), сериализует их в MsgText сообщение и отправляет по TCP-соединению. Оно может оборваться в любой момент — это мы явно перехватываем.

async def msg_sender(peer_name: str, writer) -> None:
    in_queue = IN_QUEUES[peer_name]
    while True:
        text = await in_queue.get()
        if text is None:
            break
        writer.write(Msg(("text", MsgText((
            ("text", UTF8String(text)),
        )))).encode())
        try:
            await writer.drain()
        except ConnectionResetError:
            del PEER_ALIVES[peer_name]
            return
        logging.info("%s: sent %d characters message", peer_name, len(text))

В конце инициатор входит в бесконечный цикл чтения сообщений из сокета. Проверяет текстовые ли это сообщения и помещает в OUT_QUEUES очередь, из которой они будут отправлены в out сокет соответствующего собеседника. Почему нельзя просто делать .read() и декодировать сообщение? Потому что не исключена ситуация, когда несколько сообщений от пользователя будут агрегированы в буфере операционной системы и отправлены одним TCP-сегментом. Декодировать то мы сможем первое, а дальше в буфере может остаться часть от последующего. При любой нештатной ситуации мы закрываем TCP-соединение и останавливаем msg_sender корутину (посылкой None в OUT_QUEUES очередь).

 174     buf = b""
 175     # Wait for test messages {{{
 176     while True:
 177         data = await reader.read(MaxMsgLen)
 178         if data == b"":
 179             break
 180         buf += data
 181         if len(buf) > MaxMsgLen:
 182             logging.warning("%s: max buffer size exceeded", _id)
 183             break
 184         try:
 185             msg, tail = Msg().decode(buf)
 186         except ASN1Error:
 187             continue
 188         buf = tail
 189         if msg.choice != "text":
 190             logging.warning("%s: unexpected %s message", _id, msg.choice)
 191             break
 192         try:
 193             await msg_receiver(msg.value, peer_name)
 194         except ValueError as err:
 195             logging.warning("%s: %s", err)
 196             break
 197     # }}}
 198     logging.info("%s: disconnecting: %s", _id, peer_name)
 199     IN_QUEUES[peer_name].put(None)
 200     writer.close()

  66 async def msg_receiver(msg_text: MsgText, peer_name: str) -> None:
  67     text = str(msg_text["text"])
  68     logging.info("%s: received %d characters message", peer_name, len(text))
  69     await OUT_QUEUES[peer_name].put(text)

Вернёмся к основному коду. После создания всех корутин в момент запуска программы, мы стартуем TCP-сервер. На каждое установленное соединение он создаёт responder (ответчик) корутину.

logging.basicConfig(
    level=logging.INFO,
    format="%(levelname)s %(asctime)s: %(funcName)s: %(message)s",
)
loop = asyncio.get_event_loop()
server = loop.run_until_complete(asyncio.start_server(responder, args.bind, args.port))
logging.info("Listening on: %s", server.sockets[0].getsockname())
loop.run_forever()

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

  72 async def responder(reader, writer):
  73     _id = writer.get_extra_info("peername")
  74     logging.info("%s: connected", _id)
  75     buf = b""
  76     msg_expected = "handshake"
  77     peer_name = None
  78     while True:
  79         # Read until we get Msg message {{{
  80         data = await reader.read(MaxMsgLen)
  81         if data == b"":
  82             logging.info("%s: closed connection", _id)
  83             break
  84         buf += data
  85         if len(buf) > MaxMsgLen:
  86             logging.warning("%s: max buffer size exceeded", _id)
  87             break
  88         try:
  89             msg, tail = Msg().decode(buf)
  90         except ASN1Error:
  91             continue
  92         buf = tail
  93         # }}}
  94         if msg.choice != msg_expected:
  95             logging.warning("%s: unexpected %s message", _id, msg.choice)
  96             break
  97         if msg_expected == "text":
  98             try:
  99                 await msg_receiver(msg.value, peer_name)
 100             except ValueError as err:
 101                 logging.warning("%s: %s", err)
 102                 break
 103         # Process Handshake message {{{
 104         elif msg_expected == "handshake":
 105             logging.info("%s: got %s message", _id, msg_expected)
 106             msg_handshake = msg.value
 107             peer_name = str(msg_handshake["peerName"])
 108             if peer_name not in THEIR_NAMES:
 109                 logging.warning("unknown peer name: %s", peer_name)
 110                 break
 111             writer.write(Msg(("handshake", MsgHandshake((
 112                 ("peerName", OUR_NAME),
 113             )))).encode())
 114             await writer.drain()
 115             logging.info("%s: session established: %s", _id, peer_name)
 116             peer_alive = PEER_ALIVES.pop(peer_name, None)
 117             if peer_alive is not None:
 118                 peer_alive.close()
 119                 await IN_QUEUES[peer_name].put(None)
 120             PEER_ALIVES[peer_name] = writer
 121             asyncio.ensure_future(msg_sender(peer_name, writer))
 122             msg_expected = "text"
 123         # }}}
 124     logging.info("%s: disconnecting", _id)
 125     if msg_expected == "text":
 126         IN_QUEUES[peer_name].put(None)
 127     writer.close()

Безопасный протокол

Пришло время обезопасить наше общение. Что же мы подразумеваем под безопасностью и что хотим:

• конфиденциальность передаваемых сообщений;
• аутентичность и целостность передаваемых сообщений — их изменение должно быть обнаружено;
• защита от атак перепроигрывания (replay attack) — факт пропажи или повтора сообщений должен быть обнаружен (и мы решаем обрывать соединение);
• идентификация и аутентификация собеседников по заранее вбитым публичным ключам — мы уже решили ранее, что делаем friend-to-friend сеть. Только после аутентификации мы поймём с кем общаемся;
• наличие perfect forward secrecy свойства (PFS) — компрометация нашего долгоживущего ключа подписи не должна приводить к возможности чтения всей предыдущей переписки. Запись перехваченного трафика становится бесполезной;
• действительность/валидность сообщений (транспортных и рукопожатия) только в пределах одной TCP-сессии. Вставка корректно подписанных/аутентифицированных сообщений из другой сессии (даже с этим же собеседником) не должна быть возможной;
• пассивный наблюдатель не должен видеть ни идентификаторов пользователей, ни передаваемых долгоживущих публичных ключей, ни хэшей от них. Некая анонимность от пассивного наблюдателя.

Удивительно, но этот минимум практически все хотят иметь в любом протоколе рукопожатия, и крайне мало из перечисленного в итоге выполняется для «доморощенных» протоколов. Вот и сейчас не будем изобретать нового. Я бы однозначно рекомендовал использовать Noise framework для построения протоколов, но выберем что-то попроще.

Наиболее популярны два протокола:

• TLS — сложнейший протокол с длинной историей багов, косяков, уязвимостей, плохой продуманности, сложности и недочётов (впрочем, к TLS 1.3 это мало относится). Но не рассматриваем его из-за переусложнённости.
• IPsec с IKE — не имеют серьёзных криптографических проблем, хотя тоже не просты. Если почитать про IKEv1 и IKEv2, то их истоком являются STS, ISO/IEC IS 9798-3 и SIGMA (SIGn-and-MAc) протоколы — достаточно простыми для реализации за один вечер.

Чем SIGMA, как последнее звено развития STS/ISO протоколов, хорош? Он удовлетворяет всем нашим требованиям (в том числе «скрытия» идентификаторов собеседников), не имеет известных криптографических проблем. Он минималистичен — удаление хотя бы одного элемента из сообщения протокола приведёт к его небезопасности.

Давайте пройдёмся от простейшего доморощенного протокола до SIGMA. Самая базовая интересующая нас операция это согласование ключей: функция, на выходе которой оба участника получат одно и то же значение, которое можно будет использовать в качестве симметричного ключа. Не вдаваясь в подробности: каждая из сторон генерирует эфемерную (использующуюся только в пределах одной сессии) ключевую пару (публичный и приватный ключи), обмениваются публичными ключами, вызывают функцию согласования, на вход которой передают свой приватный ключ и публичный ключ собеседника.

┌─────┐          ┌─────┐
│PeerA│          │PeerB│
└──┬──┘          └──┬──┘
   │   IdA, PubA    │ ╔════════════════════╗
   │───────────────>│ ║PrvA, PubA = DHgen()║
   │                │ ╚════════════════════╝
   │   IdB, PubB    │ ╔════════════════════╗
   │<───────────────│ ║PrvB, PubB = DHgen()║
   │                │ ╚════════════════════╝
   ────┐    ╔═══════╧════════════╗
       │    ║Key = DH(PrvA, PubB)║
   <───┘    ╚═══════╤════════════╝
   │                │
   │                │

Любой может встрять-по-середине и заменить публичные ключи своими собственными — в данном протоколе нет аутентификации собеседников. Добавим подпись долгоживущими ключами.

┌─────┐                            ┌─────┐
│PeerA│                            │PeerB│
└──┬──┘                            └──┬──┘
   │IdA, PubA, sign(SignPrvA, (PubA)) │ ╔═══════════════════════════╗
   │─────────────────────────────────>│ ║SignPrvA, SignPubA = load()║
   │                                  │ ║PrvA, PubA = DHgen()       ║
   │                                  │ ╚═══════════════════════════╝
   │IdB, PubB, sign(SignPrvB, (PubB)) │ ╔═══════════════════════════╗
   │<─────────────────────────────────│ ║SignPrvB, SignPubB = load()║
   │                                  │ ║PrvB, PubB = DHgen()       ║
   │                                  │ ╚═══════════════════════════╝
   ────┐    ╔═════════════════════╗   │
       │    ║verify(SignPubB, ...)║   │
   <───┘    ║Key = DH(PrvA, PubB) ║   │
   │        ╚═════════════════════╝   │
   │                                  │

Такая подпись не подойдёт, так как она не привязана к конкретной сессии. Такие сообщения «подойдут» и для сессий с другими участниками. Подписываться должен весь контекст. Это вынуждает также добавить посылку ещё одного сообщения от A.

Кроме того, критично добавить под подпись и собственный идентификатор, так как, в противном случае, мы можем подменить IdXXX и переподписать сообщение ключом другого известным собеседника. Для предотвращения reflection атак, необходимо чтобы элементы под подписью находились в чётко заданных местах по своему смыслу: если A подписывает (PubA, PubB), то B должен подписывать (PubB, PubA). Это ещё и говорит о важности выбора структуры и формата сериализованных данных. Например, множества в ASN.1 DER кодировании сортируются: SET OF(PubA, PubB) будет идентичен SET OF(PubB, PubA).

┌─────┐                                       ┌─────┐
│PeerA│                                       │PeerB│
└──┬──┘                                       └──┬──┘
   │                 IdA, PubA                   │ ╔═══════════════════════════╗
   │────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║
   │                                             │ ║PrvA, PubA = DHgen()       ║
   │                                             │ ╚═══════════════════════════╝
   │IdB, PubB, sign(SignPrvB, (IdB, PubA, PubB)) │ ╔═══════════════════════════╗
   │<────────────────────────────────────────────│ ║SignPrvB, SignPubB = load()║
   │                                             │ ║PrvB, PubB = DHgen()       ║
   │                                             │ ╚═══════════════════════════╝
   │     sign(SignPrvA, (IdA, PubB, PubA))       │ ╔═════════════════════╗
   │────────────────────────────────────────────>│ ║verify(SignPubB, ...)║
   │                                             │ ║Key = DH(PrvA, PubB) ║
   │                                             │ ╚═════════════════════╝
   │                                             │

Однако мы всё ещё не «доказали» что выработали одинаковый общий ключ для этой сессии. В принципе, можно обойтись и без этого шага — первое же транспортное сообщение будет невалидным, но мы хотим, чтобы когда рукопожатие завершилось, то были бы уверены что всё действительно согласовано. На данный момент у нас на руках ISO/IEC IS 9798-3 протокол.

Мы могли бы подписывать и сам выработанный ключ. Это опасно, так как не исключено, что в используемом алгоритме подписи могут быть утечки (пускай биты-на-подпись, но всё же утечки). Можно подписывать хэш от выработанного ключа, но утечка даже хэша от выработанного ключа может иметь ценность при brute-force атаке на функцию выработки. SIGMA использует MAC функцию, аутентифицирующую идентификатор отправителя.

┌─────┐                                            ┌─────┐
│PeerA│                                            │PeerB│
└──┬──┘                                            └──┬──┘
   │                    IdA, PubA                     │ ╔═══════════════════════════╗
   │─────────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║
   │                                                  │ ║PrvA, PubA = DHgen()       ║
   │                                                  │ ╚═══════════════════════════╝
   │IdB, PubB, sign(SignPrvB, (PubA, PubB)), MAC(IdB) │ ╔═══════════════════════════╗
   │<─────────────────────────────────────────────────│ ║SignPrvB, SignPubB = load()║
   │                                                  │ ║PrvB, PubB = DHgen()       ║
   │                                                  │ ╚═══════════════════════════╝
   │                                                  │ ╔═════════════════════╗
   │     sign(SignPrvA, (PubB, PubA)), MAC(IdA)       │ ║Key = DH(PrvA, PubB) ║
   │─────────────────────────────────────────────────>│ ║verify(Key, IdB)     ║
   │                                                  │ ║verify(SignPubB, ...)║
   │                                                  │ ╚═════════════════════╝
   │                                                  │

В качестве оптимизации, некоторые могут захотеть переиспользовать свои эфемерные ключи (что, конечно, плачевно для PFS). Например, мы сгенерировали ключевую пару, попытались подключиться, но TCP не был доступен или оборвался где-то на середине протокола. Жалко тратить потраченную энтропию и ресурсы процессора на новую пару. Поэтому введём, так называемый, cookie — псевдослучайное значение, которое защитит от возможных случайных replay атак при повторном использовании эфемерных публичных ключей. Из-за binding-а между cookie и эфемерным публичным ключом, публичный ключ противоположного участника можно убрать из подписи за ненадобностью.

┌─────┐                                                                 ┌─────┐
│PeerA│                                                                 │PeerB│
└──┬──┘                                                                 └──┬──┘
   │                          IdA, PubA, CookieA                           │ ╔═══════════════════════════╗
   │──────────────────────────────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║
   │                                                                       │ ║PrvA, PubA = DHgen()       ║
   │                                                                       │ ╚═══════════════════════════╝
   │IdB, PubB, CookieB, sign(SignPrvB, (CookieA, CookieB, PubB)), MAC(IdB) │ ╔═══════════════════════════╗
   │<──────────────────────────────────────────────────────────────────────│ ║SignPrvB, SignPubB = load()║
   │                                                                       │ ║PrvB, PubB = DHgen()       ║
   │                                                                       │ ╚═══════════════════════════╝
   │                                                                       │ ╔═════════════════════╗
   │          sign(SignPrvA, (CookieB, CookieA, PubA)), MAC(IdA)           │ ║Key = DH(PrvA, PubB) ║
   │──────────────────────────────────────────────────────────────────────>│ ║verify(Key, IdB)     ║
   │                                                                       │ ║verify(SignPubB, ...)║
   │                                                                       │ ╚═════════════════════╝
   │                                                                       │

Наконец, мы хотим получить приватность наших идентификаторов собеседников от пассивного наблюдателя. Для этого SIGMA предлагает сначала обменяться эфемерными ключами, выработать общий ключ, на котором зашифровать аутентифицирующие и идентифицирующие сообщения. SIGMA описывает два варианта:

• SIGMA-I — защищает инициатора от активных атак, ответчика от пассивных: инициатор аутентифицирует ответчика и если что-то не сошлось, то свою идентификацию он не выдаёт. Ответчик же выдаёт свою идентификацию если с ним начать активный протокол. Пассивный наблюдатель ничего не узнает;
  SIGMA-R — защищает ответчика от активных атак, инициатора от пассивных. Всё с точностью до наоборот, но в этом протоколе уже четыре сообщения рукопожатия передаётся.

  Выбираем SIGMA-I как более похожий на то, что мы ожидаем от клиент-серверных привычных вещей: клиента узнает только аутентифицированный сервер, а сервер и так знают все. Плюс он проще в реализации из-за меньшего количества сообщений рукопожатия. Всё что мы вносим в протокол, так это шифрование части сообщения и перенос идентификатора A в шифрованную часть последнего сообщения:

  ┌─────┐                                                                        ┌─────┐
  │PeerA│                                                                        │PeerB│
  └──┬──┘                                                                        └──┬──┘
     │                                PubA, CookieA                                 │ ╔═══════════════════════════╗
     │─────────────────────────────────────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║
     │                                                                              │ ║PrvA, PubA = DHgen()       ║
     │                                                                              │ ╚═══════════════════════════╝
     │PubB, CookieB, Enc((IdB, sign(SignPrvB, (CookieA, CookieB, PubB)), MAC(IdB))) │ ╔═══════════════════════════╗
     │<─────────────────────────────────────────────────────────────────────────────│ ║SignPrvB, SignPubB = load()║
     │                                                                              │ ║PrvB, PubB = DHgen()       ║
     │                                                                              │ ╚═══════════════════════════╝
     │                                                                              │ ╔═════════════════════╗
     │       Enc((IdA, sign(SignPrvA, (CookieB, CookieA, PubA)), MAC(IdA)))         │ ║Key = DH(PrvA, PubB) ║
     │─────────────────────────────────────────────────────────────────────────────>│ ║verify(Key, IdB)     ║
     │                                                                              │ ║verify(SignPubB, ...)║
     │                                                                              │ ╚═════════════════════╝
     │                                                                              │
  

  • Для подписи используется ГОСТ Р 34.10-2012 алгоритм с 256-бит ключами.
  • Для выработки общего ключа используется 34.10-2012 VKO.
  • В качестве MAC используется CMAC. Технически это особый режим работы блочного шифра, описанный в ГОСТ Р 34.13-2015. В качестве функции шифрования для этого режима — Кузнечик (34.12-2015).
  • В качестве идентификатора собеседника используется хэш от его публичного ключа. В качестве хэша применяется Стрибог-256 (34.11-2012 256 бит).

  После рукопожатия у нас будет согласован общий ключ. Его мы можем использовать для аутентифицированного шифрования транспортных сообщений. Эта часть совсем простая и в ней сложно ошибиться: инкрементируем счётчик сообщений, шифруем сообщение, аутентифицируем (MAC) счётчик и шифротекст, отправляем. При приёме сообщения проверяем что счётчик имеет ожидаемое значение, аутентифицируем шифротекст с счётчиком, дешифруем. Каким ключом шифровать сообщения рукопожатия, транспортные, каким аутентифицировать? Использовать один ключ для всех этих задач опасно и неразумно. Необходимо вырабатывать ключи, используя специализированные функции KDF (key derivation function). Опять же, не будем мудрить и что-то изобретать: HKDF давно известна, хорошо исследована и не имеет известных проблем. К сожалению, в родной библиотеке Python нет этой функции, поэтому используем hkdf пакет. HKDF внутри использует HMAC, который, в свою очередь, использует хэш-функцию. Пример реализации на Python на странице Wikipedia занимает считанные строки кода. Как и в случае с 34.10-2012, в качестве хэш-функции будем использовать Стрибог-256. Выход нашей функции согласования ключей будет называться сессионным ключом, из которого будут вырабатываться недостающие симметричные:

  kdf = Hkdf(None, key_session, hash=GOST34112012256)
  kdf.expand(b"handshake1-mac-identity")
  kdf.expand(b"handshake1-enc")
  kdf.expand(b"handshake1-mac")
  kdf.expand(b"handshake2-mac-identity")
  kdf.expand(b"handshake2-enc")
  kdf.expand(b"handshake2-mac")
  kdf.expand(b"transport-initiator-enc")
  kdf.expand(b"transport-initiator-mac")
  kdf.expand(b"transport-responder-enc")
  kdf.expand(b"transport-responder-mac")
  

  Структуры/схемы

  Рассмотрим какие же теперь ASN.1 структуры у нас получились для передачи всех этих данных:

  class Msg(Choice):
      schema = ((
          ("text", MsgText()),
          ("handshake0", MsgHandshake0(expl=tag_ctxc(0))),
          ("handshake1", MsgHandshake1(expl=tag_ctxc(1))),
          ("handshake2", MsgHandshake2(expl=tag_ctxc(2))),
      ))
  
  class MsgText(Sequence):
      schema = ((
          ("payload", MsgTextPayload()),
          ("payloadMac", MAC()),
      ))
  
  class MsgTextPayload(Sequence):
      schema = ((
          ("nonce", Integer(bounds=(0, float("+inf")))),
          ("ciphertext", OctetString(bounds=(1, MaxTextLen))),
      ))
  
  class MsgHandshake0(Sequence):
      schema = ((
          ("cookieInitiator", Cookie()),
          ("pubKeyInitiator", PubKey()),
      ))
  
  class MsgHandshake1(Sequence):
      schema = ((
          ("cookieResponder", Cookie()),
          ("pubKeyResponder", PubKey()),
          ("ukm", OctetString(bounds=(8, 8))),
          ("ciphertext", OctetString()),
          ("ciphertextMac", MAC()),
      ))
  
  class MsgHandshake2(Sequence):
      schema = ((
          ("ciphertext", OctetString()),
          ("ciphertextMac", MAC()),
      ))
  
  class HandshakeTBE(Sequence):
      schema = ((
          ("identity", OctetString(bounds=(32, 32))),
          ("signature", OctetString(bounds=(64, 64))),
          ("identityMac", MAC()),
      ))
  
  class HandshakeTBS(Sequence):
      schema = ((
          ("cookieTheir", Cookie()),
          ("cookieOur", Cookie()),
          ("pubKeyOur", PubKey()),
      ))
  
  class Cookie(OctetString): bounds = (16, 16)
  class PubKey(OctetString): bounds = (64, 64)
  class MAC(OctetString): bounds = (16, 16)
  

  HandshakeTBS — то, что будет подписываться (to be signed). HandshakeTBE — то, что будет зашифровано (to be encrypted). Обращаю внимание на поле ukm в MsgHandshake1. 34.10 VKO, для ещё большей рандомизации вырабатываемых ключей, включает параметр UKM (user keying material) — просто дополнительная энтропия.

  Добавление криптографии в код

  Рассмотрим лишь только внесённые к оригинальному коду изменения, так как каркас остался прежним (на самом деле, сначала была написана окончательная реализация, а потом из неё выпиливалась вся криптография).

  Так как аутентификация и идентификация собеседников будет проводится по публичным ключам, то теперь их надо где-то долговременно хранить. Для простоты используем JSON такого вида:

  {
      "our": {
          "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98",
          "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1"
      },
      "their": {
          "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce",
          "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a"
      }
  }
  

  our — наша ключевая пара, шестнадцатеричные приватный и публичные ключи. their — имена собеседников и их публичные ключи. Изменим аргументы командной строки и добавим постобработку JSON данных:

  from pygost import gost3410
  from pygost.gost34112012256 import GOST34112012256
  
  CURVE = gost3410.GOST3410Curve(
      *gost3410.CURVE_PARAMS["GostR3410_2001_CryptoPro_A_ParamSet"]
  )
  
  parser = argparse.ArgumentParser(description="GOSTIM")
  parser.add_argument(
      "--keys-gen",
      action="store_true",
      help="Generate JSON with our new keypair",
  )
  parser.add_argument(
      "--keys",
      default="keys.json",
      required=False,
      help="JSON with our and their keys",
  )
  parser.add_argument(
      "--bind",
      default="::1",
      help="Address to listen on",
  )
  parser.add_argument(
      "--port",
      type=int,
      default=6666,
      help="Port to listen on",
  )
  args = parser.parse_args()
  
  if args.keys_gen:
      prv_raw = urandom(32)
      pub = gost3410.public_key(CURVE, gost3410.prv_unmarshal(prv_raw))
      pub_raw = gost3410.pub_marshal(pub)
      print(json.dumps({
          "our": {"prv": hexenc(prv_raw), "pub": hexenc(pub_raw)},
          "their": {},
      }))
      exit(0)
  
  # Parse and unmarshal our and their keys {{{
  with open(args.keys, "rb") as fd:
      _keys = json.loads(fd.read().decode("utf-8"))
  KEY_OUR_SIGN_PRV = gost3410.prv_unmarshal(hexdec(_keys["our"]["prv"]))
  _pub = hexdec(_keys["our"]["pub"])
  KEY_OUR_SIGN_PUB = gost3410.pub_unmarshal(_pub)
  KEY_OUR_SIGN_PUB_HASH = OctetString(GOST34112012256(_pub).digest())
  for peer_name, pub_raw in _keys["their"].items():
      _pub = hexdec(pub_raw)
      KEYS[GOST34112012256(_pub).digest()] = {
          "name": peer_name,
          "pub": gost3410.pub_unmarshal(_pub),
      }
  # }}}
  

  Приватный ключ 34.10 алгоритма — случайное число. Размером 256-бит для 256-бит эллиптических кривых. PyGOST работает не с набором байт, а с большими числами, поэтому наш приватный ключ (urandom(32)) необходимо преобразовать в число, используя gost3410.prv_unmarshal(). Публичный ключ детерминировано вычисляется из приватного, используя gost3410.public_key(). Публичный ключ 34.10 — два больших числа, которые тоже нужно преобразовать в байтовую последовательность для удобства хранения и передачи, используя gost3410.pub_marshal().

  После чтения JSON файла, публичные ключи, соответственно, нужно преобразовать назад, используя gost3410.pub_unmarshal(). Так как нам будут приходить идентификаторы собеседников в виде хэша от публичного ключа, то их можно сразу же заранее вычислить и поместить в словарь для быстрого поиска. Стрибог-256 хэш это gost34112012256.GOST34112012256(), полностью удовлетворяющий hashlib интерфейсу хэш-функций.

  Как изменилась корутина инициатора? Всё, как по схеме рукопожатия: генерируем cookie (128-бит вполне предостаточно), эфемерную ключевую пару 34.10, которая будет использоваться для VKO функции согласования ключей.

   395 async def initiator(host, port):
   396     _id = repr((host, port))
   397     logging.info("%s: dialing", _id)
   398     reader, writer = await asyncio.open_connection(host, port)
   399     # Generate our ephemeral public key and cookie, send Handshake 0 message {{{
   400     cookie_our = Cookie(urandom(16))
   401     prv = gost3410.prv_unmarshal(urandom(32))
   402     pub_our = gost3410.public_key(CURVE, prv)
   403     pub_our_raw = PubKey(gost3410.pub_marshal(pub_our))
   404     writer.write(Msg(("handshake0", MsgHandshake0((
   405         ("cookieInitiator", cookie_our),
   406         ("pubKeyInitiator", pub_our_raw),
   407     )))).encode())
   408     # }}}
   409     await writer.drain()
  

  • ждём ответа и декодируем пришедшее Msg сообщение;
  • убеждаемся что получили handshake1;
  • декодируем эфемерный публичный ключ противоположной стороны и вычисляем сессионный ключ;
  • вырабатываем симметричные ключи необходимые для обработки TBE части сообщения.

   423     logging.info("%s: got %s message", _id, msg.choice)
   424     if msg.choice != "handshake1":
   425         logging.warning("%s: unexpected message, disconnecting", _id)
   426         writer.close()
   427         return
   428     # }}}
   429     msg_handshake1 = msg.value
   430     # Validate Handshake message {{{
   431     cookie_their = msg_handshake1["cookieResponder"]
   432     pub_their_raw = msg_handshake1["pubKeyResponder"]
   433     pub_their = gost3410.pub_unmarshal(bytes(pub_their_raw))
   434     ukm_raw = bytes(msg_handshake1["ukm"])
   435     ukm = ukm_unmarshal(ukm_raw)
   436     key_session = kek_34102012256(CURVE, prv, pub_their, ukm, mode=2001)
   437     kdf = Hkdf(None, key_session, hash=GOST34112012256)
   438     key_handshake1_mac_identity = kdf.expand(b"handshake1-mac-identity")
   439     key_handshake1_enc = kdf.expand(b"handshake1-enc")
   440     key_handshake1_mac = kdf.expand(b"handshake1-mac")
  

  UKM это 64-бит число (urandom(8)), которое тоже требует десериализации из байтового представления, используя gost3410_vko.ukm_unmarshal(). VKO функция для 34.10-2012 256-бит это gost3410_vko.kek_34102012256() (KEK — key encryption key).

  Выработанный сессионный ключ уже является 256-бит байтовой псевдослучайной последовательностью. Поэтому его сразу же можно использовать в HKDF функции. Так как GOST34112012256 удовлетворяет hashlib интерфейсу, то его можно сразу же использовать в Hkdf классе. Соль (первый аргумент Hkdf) мы не указываем, так как выработанный ключ из-за эфемерности участвующих ключевых пар будет разным для каждой сессии и в нём уже достаточно энтропии. kdf.expand() по умолчанию уже выдаёт ключи длиной 256-бит, требуемые для Кузнечика в дальнейшем.

  Далее проверяются TBE и TBS части пришедшего сообщения:

  • вычисляется и проверяется MAC над пришедшим шифротекстом;
  • дешифруется шифротекст;
  • декодируется TBE структура;
  • из неё берётся идентификатор собеседника и проверяется известен ли он нам вообще;
  • вычисляется и проверятся MAC над этим идентификатором;
  • проверяется подпись над TBS структурой, в которую входят cookie обеих сторон и публичный эфемерный ключ противоположной стороны. Подпись проверяется долгоживущим ключом подписи собеседника.

   441     try:
   442         peer_name = validate_tbe(
   443             msg_handshake1,
   444             key_handshake1_mac_identity,
   445             key_handshake1_enc,
   446             key_handshake1_mac,
   447             cookie_our,
   448             cookie_their,
   449             pub_their_raw,
   450         )
   451     except ValueError as err:
   452         logging.warning("%s: %s, disconnecting", _id, err)
   453         writer.close()
   454         return
   455     # }}}
  
   128 def validate_tbe(
   129         msg_handshake: Union[MsgHandshake1, MsgHandshake2],
   130         key_mac_identity: bytes,
   131         key_enc: bytes,
   132         key_mac: bytes,
   133         cookie_their: Cookie,
   134         cookie_our: Cookie,
   135         pub_key_our: PubKey,
   136 ) -> str:
   137     ciphertext = bytes(msg_handshake["ciphertext"])
   138     mac_tag = mac(GOST3412Kuznechik(key_mac).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext)
   139     if not compare_digest(mac_tag, bytes(msg_handshake["ciphertextMac"])):
   140         raise ValueError("invalid MAC")
   141     plaintext = ctr(
   142         GOST3412Kuznechik(key_enc).encrypt,
   143         KUZNECHIK_BLOCKSIZE,
   144         ciphertext,
   145         8 * b"x00",
   146     )
   147     try:
   148         tbe, _ = HandshakeTBE().decode(plaintext)
   149     except ASN1Error:
   150         raise ValueError("can not decode TBE")
   151     key_sign_pub_hash = bytes(tbe["identity"])
   152     peer = KEYS.get(key_sign_pub_hash)
   153     if peer is None:
   154         raise ValueError("unknown identity")
   155     mac_tag = mac(
   156         GOST3412Kuznechik(key_mac_identity).encrypt,
   157         KUZNECHIK_BLOCKSIZE,
   158         key_sign_pub_hash,
   159     )
   160     if not compare_digest(mac_tag, bytes(tbe["identityMac"])):
   161         raise ValueError("invalid identity MAC")
   162     tbs = HandshakeTBS((
   163         ("cookieTheir", cookie_their),
   164         ("cookieOur", cookie_our),
   165         ("pubKeyOur", pub_key_our),
   166     ))
   167     if not gost3410.verify(
   168         CURVE,
   169         peer["pub"],
   170         GOST34112012256(tbs.encode()).digest(),
   171         bytes(tbe["signature"]),
   172     ):
   173         raise ValueError("invalid signature")
   174     return peer["name"]
  

  Как уже писал выше, 34.13-2015 описывает различные режимы работы блочных шифров из 34.12-2015. Среди них есть режим выработки имитовставки, вычисления MAC-а. В PyGOST это gost3413.mac(). Этот режим требует передачи функции шифрования (принимающая и возвращающая один блок данных), размера шифроблока и, собственно, самих данных. Почему нельзя hardcode-ить размер шифроблока? 34.12-2015 описывает не только 128-битный шифр Кузнечик, но ещё и 64-битную Магму — немного изменённый ГОСТ 28147-89, созданный ещё в КГБ и до сих пор имеющий один из самых высоких порогов безопасности.

  Кузнечик инициализируется gost.3412.GOST3412Kuznechik(key) вызовом и возвращает объект с .encrypt()/.decrypt() методами, пригодными для передачи в 34.13 функции. MAC вычисляется следующим образом: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Для сравнения вычисленного и пришедшего MAC-а нельзя использовать обычное сравнение (==) байтовых строк, так как это операция даёт утечки времени сравнения, что, в общем случае, может приводить к фатальным уязвимостям типа BEAST атаки на TLS. В Python имеется специальная hmac.compare_digest функция для этого.

  Функция блочного шифра может зашифровать только один блок данных. Для большего количества, да ещё и не кратной длины, необходимо использовать режим шифрования. В 34.13-2015 описаны следующие: ECB, CTR, OFB, CBC, CFB. У каждого свои допустимые сферы применения и характеристики. К огромному сожалению, у нас до сих пор нет стандартизованных аутентифицированных режимов шифрования (типа CCM, OCB, GCM и подобных) — мы вынуждены самостоятельно хотя бы добавлять MAC. Я выбираю режим счётчика (CTR): он не требует дополнения до размера блока, может распараллеливаться, использует только функцию шифрования, может быть безопасно использован для шифрования большого количества сообщений (в отличии от CBC, у которого относительно быстро начинаются коллизии).

  Как и .mac(), .ctr() принимает похожие данные на входе: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). Требуется задание вектора инициализации, длиной ровно в половину шифроблока. Если наш ключ шифрования используется только для шифрования одного сообщения (пускай и из нескольких блоков), то безопасно задать нулевой вектор инициализации. Для шифрования handshake сообщений у нас используется каждый раз отдельный ключ.

  Проверка подписи gost3410.verify() тривиальна: передаём эллиптическую кривую в пределах которой работаем (её мы просто фиксируем в нашем GOSTIM протоколе), публичный ключ подписанта (не забываем, что это должен быть кортеж из двух больших чисел, а не байтовая строка), 34.11-2012 хэш и сама пришедшая подпись.

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

   456     # Prepare and send Handshake 2 message {{{
   457     tbs = HandshakeTBS((
   458         ("cookieTheir", cookie_their),
   459         ("cookieOur", cookie_our),
   460         ("pubKeyOur", pub_our_raw),
   461     ))
   462     signature = gost3410.sign(
   463         CURVE,
   464         KEY_OUR_SIGN_PRV,
   465         GOST34112012256(tbs.encode()).digest(),
   466     )
   467     key_handshake2_mac_identity = kdf.expand(b"handshake2-mac-identity")
   468     mac_tag = mac(
   469         GOST3412Kuznechik(key_handshake2_mac_identity).encrypt,
   470         KUZNECHIK_BLOCKSIZE,
   471         bytes(KEY_OUR_SIGN_PUB_HASH),
   472     )
   473     tbe = HandshakeTBE((
   474         ("identity", KEY_OUR_SIGN_PUB_HASH),
   475         ("signature", OctetString(signature)),
   476         ("identityMac", MAC(mac_tag)),
   477     ))
   478     tbe_raw = tbe.encode()
   479     key_handshake2_enc = kdf.expand(b"handshake2-enc")
   480     key_handshake2_mac = kdf.expand(b"handshake2-mac")
   481     ciphertext = ctr(
   482         GOST3412Kuznechik(key_handshake2_enc).encrypt,
   483         KUZNECHIK_BLOCKSIZE,
   484         tbe_raw,
   485         8 * b"x00",
   486     )
   487     mac_tag = mac(
   488         GOST3412Kuznechik(key_handshake2_mac).encrypt,
   489         KUZNECHIK_BLOCKSIZE,
   490         ciphertext,
   491     )
   492     writer.write(Msg(("handshake2", MsgHandshake2((
   493         ("ciphertext", OctetString(ciphertext)),
   494         ("ciphertextMac", MAC(mac_tag)),
   495     )))).encode())
   496     # }}}
   497     await writer.drain()
   498     logging.info("%s: session established: %s", _id, peer_name)
   

  Когда сессия установлена, то вырабатываются транспортные ключи (отдельный ключ для шифрования, для аутентификации, для каждой из сторон), инициализируется Кузнечик для дешифрования и проверки MAC-а:

   499     # Run text message sender, initialize transport decoder {{{
   500     key_initiator_enc = kdf.expand(b"transport-initiator-enc")
   501     key_initiator_mac = kdf.expand(b"transport-initiator-mac")
   502     key_responder_enc = kdf.expand(b"transport-responder-enc")
   503     key_responder_mac = kdf.expand(b"transport-responder-mac")
   ...
   509     asyncio.ensure_future(msg_sender(
   510         peer_name,
   511         key_initiator_enc,
   512         key_initiator_mac,
   513         writer,
   514     ))
   515     encrypter = GOST3412Kuznechik(key_responder_enc).encrypt
   516     macer = GOST3412Kuznechik(key_responder_mac).encrypt
   517     # }}}
   519     nonce_expected = 0
  
   520     # Wait for test messages {{{
   521     while True:
   522         data = await reader.read(MaxMsgLen)
   ...
   530             msg, tail = Msg().decode(buf)
   ...
   537         try:
   538             await msg_receiver(
   539                 msg.value,
   540                 nonce_expected,
   541                 macer,
   542                 encrypter,
   543                 peer_name,
   544             )
   545         except ValueError as err:
   546             logging.warning("%s: %s", err)
   547             break
   548         nonce_expected += 1
   549     # }}}
  

  msg_sender корутина теперь шифрует сообщения, перед отправкой в TCP-соединение. У каждого сообщения монотонно возрастающий nonce, также являющийся и вектором инициализации при шифровании в режиме счётчика. У каждого сообщения и блока сообщения гарантированно будут отличающиеся значения счётчика.

  async def msg_sender(peer_name: str, key_enc: bytes, key_mac: bytes, writer) -> None:
      nonce = 0
      encrypter = GOST3412Kuznechik(key_enc).encrypt
      macer = GOST3412Kuznechik(key_mac).encrypt
      in_queue = IN_QUEUES[peer_name]
      while True:
          text = await in_queue.get()
          if text is None:
              break
          ciphertext = ctr(
              encrypter,
              KUZNECHIK_BLOCKSIZE,
              text.encode("utf-8"),
              long2bytes(nonce, 8),
          )
          payload = MsgTextPayload((
              ("nonce", Integer(nonce)),
              ("ciphertext", OctetString(ciphertext)),
          ))
          mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode())
          writer.write(Msg(("text", MsgText((
              ("payload", payload),
              ("payloadMac", MAC(mac_tag)),
          )))).encode())
          nonce += 1
  

  Приходящие сообщения обрабатываются корутиной msg_receiver, занимающейся аутентификацией и дешифрацией:

  async def msg_receiver(
          msg_text: MsgText,
          nonce_expected: int,
          macer,
          encrypter,
          peer_name: str,
  ) -> None:
      payload = msg_text["payload"]
      if int(payload["nonce"]) != nonce_expected:
          raise ValueError("unexpected nonce value")
      mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode())
      if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])):
          raise ValueError("invalid MAC")
      plaintext = ctr(
          encrypter,
          KUZNECHIK_BLOCKSIZE,
          bytes(payload["ciphertext"]),
          long2bytes(nonce_expected, 8),
      )
      text = plaintext.decode("utf-8")
      await OUT_QUEUES[peer_name].put(text)
  

  Заключение

  GOSTIM предполагается использовать исключительно в учебных целях (так как не покрыт тестами, как минимум)! Исходный код программы можно скачать тут (Стрибог-256 хэш: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Как и все мои проекты, типа GoGOST, PyDERASN, NNCP, GoVPN, GOSTIM является полностью свободным ПО, распространяемым на условиях GPLv3+.

  Сергей Матвеев, шифропанк, член Фонда СПО, Python/Go-разработчик, главный специалист ФГУП «НТЦ „Атлас“.

Источник: habr.com