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

Като разработчик PyGOST на библиотеката (ГОСТ криптографски примитиви на чист Python), често получавам въпроси как на коляно да реализирам прост безопасен обмен на съобщения. Много хора считат приложната криптография за достатъчно проста работа, и извикването на .encrypt() на блочен шифър ще е достатъчно за безопасно изпращане по комуникационен канал. Други пък смятат, че приложната криптография е предназначена само за малцина и е приемливо, че богатите компании като Telegram с математически олимпиадисти не могат да реализират безопасен протокол.

Всичко това ме подтикна да напиша тази статия, за да покажа, че реализирането на криптографски протоколи и безопасен IM не е толкова сложна задача. Обаче, не е разумно да се изобретяват собствени протоколи за удостоверяване и споразумяване на ключове.

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

Дизайн на IM

Първо, трябва да разберем как ще изглежда нашият IM. За простота, нека бъде peer-to-peer мрежа, без откриване на участниците. Ръчно ще посочваме на какъв адрес: порт да се свързваме за комуникация с разговора.

Разбирам, че в момента предположението за наличието на директна връзка между два произволни компютъра е съществително ограничение за практическия IM. Но колкото повече разработчици реализират всякакви NAT-траверс капани, толкова по-дълго ще останем в 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 не осигуряват (готовият резултат може да се различава).

XDR е подходящ за тази задача, но аз избирам ASN.1 с DER-кодиране и PyDERASN библиотеката, тъй като ще разполагаме с високостепенни обекти, с които често е по-приятно и удобно да работим. В отличие от безсхемни 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()     │
        │<─────────────────│
        │                  │

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

Както вече казах, за всички операции със сокети ще се използва библиотеката asyncio. Да обявим какво очакваме в момента на стартиране:

parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
    "--our-name",
    required=True,
    help="Нашето име на връстника",
)
parser.add_argument(
    "--their-names",
    required=True,
    help="Нами на връстниците, разделени с запетая",
)
parser.add_argument(
    "--bind",
    default="::1",
    help="Адрес за слушане",
)
parser.add_argument(
    "--port",
    type=int,
    default=6666,
    help="Порт за слушане",
)
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))

Нека разгледаме инициатора. Първо, той открива свързване към посочения хост/порт и изпраща съобщение за ръкостискане с името си:

 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     # Изчакване на Handshake съобщение {{{
 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     # Стартиране на изпращача на текстови съобщения, инициализиране на декодера на транспорта {{{
 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 приема изходящите съобщения (поставени в опашката от входящия сокет), сериализира ги в съобщение 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 опашката, откъдето ще бъдат изпратени в изходния сокет на съответния събеседник. Защо не можем просто да направим .read() и да декодираме съобщението? Защото не е изключена ситуацията, в която няколко съобщения от потребителя са агрегирани в буфера на операционната система и изпратени в един TCP сегмент. Ще успеем да декодираме първото, а в буфера може да остане част от следващото. При всяка извънредна ситуация затваряме TCP връзката и спираме msg_sender корутината (изпращайки None в OUT_QUEUES опашката).

 174     buf = b""
 175     # Изчакайте тестови съобщения {{{
 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: превишена максимална големина на буфера", _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: неочаквано %s съобщение", _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: прекъсва връзката: %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: получено съобщение с %d знака", 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("Слушане на: %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) — фактът за загуба или повторение на съобщения трябва да бъде открит (и решаваме да прекъснем връзката);
• идентификация и автентикация на участниците по предварително зададени публични ключове — вече решихме, че създаваме peer-to-peer мрежа. Само след автентикация ще разберем с кого общуваме;
• наличие перфектна напреднина тайна свойства (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 атаки при повторно използване на еферните публични ключове. Поради свързаността между cookie и еферния публичен ключ, публичният ключ на противоположната страна може да бъде премахнат от подписа без необходимост.

┌─────┐                                                                 ┌─────┐
│PeerA│                                                                 │PeerB│
└──┬──┘                                                                 └──┬──┘
   │                          IdA, PubA, CookieA                           │ ╔═══════════════════════════╗
   │──────────────────────────────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║
   │                                                                       │ ║PrvA, PubA = DHgen()       ║
   │                                                                       │ ╚═══════════════════════════╝
   │IdB, PubB, CookieB, sign(SignPrvB, (CookieA, CookieB, PubB)), MAC(IdB) │ ╔═══════════════════════════╗
   ││ ║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 (функция за извеждане на ключ). Отново, няма да измисляме нищо ново: 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 данните:

  от pygost import gost3410
  от 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="Генериране на JSON с нова двойка ключове",
  )
  parser.add_argument(
      "--keys",
      default="keys.json",
      required=False,
      help="JSON с нашите и техните ключове",
  )
  parser.add_argument(
      "--bind",
      default="::1",
      help="Адрес за слушане",
  )
  parser.add_argument(
      "--port",
      type=int,
      default=6666,
      help="Порт за слушане",
  )
  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)
  
  # Парсване и разуплътняване на нашите и техните ключове {{{
  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 за хеш функции.

  Как се промени корутината на инициатора? Всичко е по схемата на ръкостискането: генерираме бисквитка (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     # Генерираме нашето епhemeral публично ключ и бисквитка, изпращаме съобщение Handshake 0 {{{
   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;
  • декодираме епhemeral публичния ключ на противоположната страна и изчисляваме сесионния ключ;
  • създаваме симетрични ключове, необходими за обработка на 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     # Валидираме съобщението за Handshake {{{
   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 — ключ за криптиране на ключ).

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

  След това се проверяват TBE и TBS частите на полученото съобщение:

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

   441     опит:
   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     освен ако ValueError като грешка:
   452         logging.warning("%s: %s, прекъсване на връзката", _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("невалиден 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     освен ако ASN1Error:
   150         raise ValueError("не може да се декодира 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("неизвестна идентичност")
   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("невалиден идентичност 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("невалиден подпис")
   174     return peer["name"]
  

  Както вече споменах, 34.13-2015 описва различни режими на работа на блочните шифри от 34.12-2015. Сред тях има режим за генериране на имитовставки, изчисляване на MAC. В PyGOST това е gost3413.mac(). Този режим изисква предаване на функция за шифриране (приемаща и връщаща един блок данни), размера на шифроблока и самите данни. Защо не може да се зададе размерът на шифроблока на твърдо? 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     # Подготовьте и отправьте сообщение Handshake 2 {{{
   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: сессия установлена: %s", _id, peer_name)
   

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

   499     # Запустите отправитель текстовых сообщений, инициализируйте декодер транспорта {{{
   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     # Ожидание тестовых сообщений {{{
   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, который также является вектором инициализации при шифровании в режиме счётчика. У каждого сообщения и блока сообщения гарантированно будут отличающиеся значения счётчика.

  асинхронная функция 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, занимающейся аутентификацией и дешифрацией:

  асинхронная функция 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("неожиданное значение nonce")
      mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode())
      if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])):
          raise ValueError("недопустимый 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