GOSTIM: P2P F2F E2EE IM za jeden večer s kryptografií GOST

Být vývojářem PyGOST knihovny (GOST kryptografická primitiva v čistém Pythonu), často dostávám otázky, jak implementovat nejjednodušší bezpečné zasílání zpráv na koleni. Mnoho lidí považuje aplikovanou kryptografii za docela jednoduchou a volání .encrypt() na blokovou šifru bude stačit k jejímu bezpečnému odeslání přes komunikační kanál. Jiní věří, že aplikovaná kryptografie je údělem několika málo lidí, a je přijatelné, aby bohaté společnosti jako Telegram s olympijskými matematiky nelze realizovat zabezpečený protokol.

To vše mě přimělo napsat tento článek, abych ukázal, že implementace kryptografických protokolů a bezpečných IM není tak obtížný úkol. Nevyplatí se však vymýšlet vlastní protokoly pro ověřování a dohody o klíčích.

Článek napíše peer-to-peer, přítel-ka-přítel, end-to-end šifrované instant messenger s SIGMA-I autentizační a klíčový protokol (na jehož základě je implementován IPsec IKE), používající výhradně kryptografické algoritmy GOST knihovnu PyGOST a knihovnu kódování zpráv ASN.1 PyDERASN (o kterém už jsem napsal dříve). Předpoklad: musí být tak jednoduchý, aby se dal napsat od nuly za jeden večer (nebo pracovní den), jinak už to není jednoduchý program. Asi má chyby, zbytečné komplikace, nedostatky, navíc je to můj první program využívající knihovnu asyncio.

IM design

Nejprve musíme pochopit, jak bude naše IM vypadat. Pro zjednodušení nechť je to síť peer-to-peer, bez jakéhokoli objevování účastníků. Osobně uvedeme, na kterou adresu: port se připojit pro komunikaci s účastníkem rozhovoru.

Chápu, že v současné době je předpoklad dostupnosti přímé komunikace mezi dvěma libovolnými počítači významným omezením použitelnosti IM v praxi. Čím více však vývojáři implementují nejrůznější berličky pro přechod přes NAT, tím déle zůstaneme na internetu IPv4 s deprimující pravděpodobností komunikace mezi libovolnými počítači. Jak dlouho dokážete tolerovat nedostatek IPv6 doma a v práci?

Budeme mít síť přátel: všichni možní partneři musí být známi předem. Za prvé, toto vše výrazně zjednodušuje: představili jsme se, našli nebo nenašli jméno/klíč, odpojili jsme se nebo pokračovali v práci, když jsme znali partnera. Za druhé, obecně je bezpečný a eliminuje mnoho útoků.

Rozhraní IM se bude blížit klasickým řešením nešťastné projekty, které se mi velmi líbí pro jejich minimalismus a filozofii Unix-way. Program IM vytvoří adresář se třemi sokety domény Unix pro každého partnera:

• in—zprávy odeslané partnerovi jsou v něm zaznamenány;
• out - zprávy přijaté od partnera jsou z něj čteny;
• stav - čtením z něj zjistíme, zda je účastník právě připojen, adresu spojení/port.

Kromě toho se vytvoří conn soket zapsáním hostitelského portu, do kterého iniciujeme spojení se vzdáleným účastníkem.

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

Tento přístup vám umožňuje provádět nezávislé implementace přenosu IM a uživatelského rozhraní, protože neexistuje žádný přítel, nemůžete se zavděčit všem. Použitím tmux и / nebo vícenásobné, můžete získat rozhraní pro více oken se zvýrazněním syntaxe. A s pomocí rlwrap můžete získat vstupní řádek zpráv kompatibilní s GNU Readline.

Ve skutečnosti, bezstarostné projekty používají soubory FIFO. Osobně jsem nedokázal pochopit, jak se soubory v asyncio kompetitivně pracovat bez ručně psaného pozadí z vyhrazených vláken (jazyk na takové věci používám už dlouho Go). Proto jsem se rozhodl vystačit s unixovými doménovými sockety. Bohužel to znemožňuje provést echo 2001:470:dead::babe 6666 > conn. Tento problém jsem vyřešil pomocí socat: echo 2001:470:dead::babe 6666 | socat - UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.

Původní nezabezpečený protokol

TCP se používá jako transport: garantuje doručení a jeho objednávku. UDP nezaručuje ani jedno (což by bylo užitečné při použití kryptografie), ale podporu SCTP Python nevychází z krabice.

Bohužel v TCP neexistuje žádný koncept zprávy, pouze proud bajtů. Proto je potřeba vymyslet formát zpráv, aby je bylo možné sdílet mezi sebou v tomto vláknu. Můžeme se dohodnout na použití znaku pro odřádkování. Pro začátek je to v pořádku, ale jakmile začneme šifrovat naše zprávy, může se tento znak objevit kdekoli v šifrovaném textu. V sítích jsou proto oblíbené protokoly, které nejprve odesílají délku zprávy v bajtech. Například Python má z krabice xdrlib, který umožňuje pracovat s podobným formátem XDR.

Se čtením TCP nebudeme pracovat správně a efektivně – zjednodušíme kód. Čteme data ze zásuvky v nekonečné smyčce, dokud nedekódujeme kompletní zprávu. JSON s XML lze také použít jako formát pro tento přístup. Ale když se přidá kryptografie, data budou muset být podepsána a ověřena - a to bude vyžadovat bajt po bajtu identickou reprezentaci objektů, což JSON/XML neposkytuje (výsledky výpisů se mohou lišit).

XDR je pro tento úkol vhodný, nicméně volím ASN.1 s kódováním DER a PyDERASN knihovny, protože budeme mít po ruce předměty na vysoké úrovni, se kterými je často příjemnější a pohodlnější pracovat. Na rozdíl od bez schématu bencode, MessagePack nebo CBOR, ASN.1 automaticky porovná data s pevně zakódovaným schématem.

# 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))),
    ))

Přijatá zpráva bude Msg: buď textová MsgText (zatím s jedním textovým polem) nebo MsgHandshake handshake zpráva (která obsahuje jméno partnera). Teď to vypadá překomplikovaně, ale tohle je základ do budoucna.

     ┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┬──┘ └──┘┘──s IdA) │ │───────── - │ │ MsgText() │ │──── MsgText() │ │ │

IM bez kryptografie

Jak jsem již řekl, pro všechny operace soketů bude použita knihovna asyncio. Pojďme oznámit, co očekáváme při spuštění:

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(","))

Nastavte si vlastní jméno (--naše-jméno Alice). Všichni očekávaní účastníci rozhovoru jsou uvedeni odděleni čárkami (—jejich jména bob, eve). Pro každého z účastníků dialogu je vytvořen adresář s unixovými sokety a také korutina pro každý vstupní a výstupní stav:

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"))

Zprávy přicházející od uživatele ze vstupního soketu jsou odesílány do fronty 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"))

Zprávy přicházející od účastníků rozhovoru jsou odesílány do front OUT_QUEUES, ze kterých jsou data zapisována do výstupního soketu:

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()

Při čtení ze stavového soketu program hledá adresu účastníka ve slovníku PEER_ALIVE. Pokud ještě není spojení s účastníkem komunikace, zapíše se prázdný řádek.

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()

Při zápisu adresy do conn soketu se spustí funkce „iniciátor“ připojení:

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))

Podívejme se na iniciátora. Nejprve zjevně otevře připojení k určenému hostiteli/portu a odešle zprávu handshake s jeho názvem:

 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()

Poté čeká na odpověď od vzdálené strany. Pokusí se dekódovat příchozí odpověď pomocí schématu Msg ASN.1. Předpokládáme, že celá zpráva bude odeslána v jednom TCP segmentu a při volání .read() ji přijmeme atomicky. Zkontrolujeme, že jsme obdrželi zprávu o podání ruky.

 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     # }}}

Zkontrolujeme, zda je nám známé přijaté jméno partnera. Pokud ne, přerušíme spojení. Zkontrolujeme, zda jsme s ním již navázali spojení (účastník opět dal příkaz ke spojení s námi) a uzavřeme jej. Fronta IN_QUEUES obsahuje pythonovské řetězce s textem zprávy, ale má speciální hodnotu None, která signalizuje korutině msg_sender, aby přestala pracovat, takže zapomene na svůj zapisovač spojený se starším připojením 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 přijímá odchozí zprávy (ve frontě z in socketu), serializuje je do zprávy MsgText a odesílá je přes TCP spojení. Může se kdykoli zlomit - jasně to zachytíme.

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))

Na konci vstoupí iniciátor do nekonečné smyčky čtení zpráv ze zásuvky. Zkontroluje, zda se jedná o textové zprávy, a zařadí je do fronty OUT_QUEUES, ze které budou odeslány do výstupního soketu odpovídajícího partnera. Proč prostě nemůžete udělat .read() a dekódovat zprávu? Protože je možné, že několik zpráv od uživatele bude agregováno ve vyrovnávací paměti operačního systému a odesláno v jednom segmentu TCP. Můžeme dekódovat první a část následujícího pak může zůstat ve vyrovnávací paměti. V případě jakékoli abnormální situace uzavřeme TCP spojení a zastavíme korutinu msg_sender (zasláním None do fronty 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)

Vraťme se k hlavnímu kódu. Po vytvoření všech korutin v době spuštění programu spustíme TCP server. Pro každé navázané spojení vytvoří korutinu respondéru.

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()

respondér je podobný iniciátoru a zrcadlí všechny stejné akce, ale nekonečná smyčka čtení zpráv se pro jednoduchost spustí okamžitě. V současné době protokol handshake odesílá jednu zprávu z každé strany, ale v budoucnu budou dvě od iniciátora připojení, po kterých lze okamžitě odesílat textové zprávy.

  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()

Zabezpečený protokol

Je čas zabezpečit naši komunikaci. Co myslíme bezpečností a co chceme:

• důvěrnost přenášených zpráv;
• autentičnost a integrita přenášených zpráv – jejich změny musí být detekovány;
• ochrana proti replay útokům - musí být detekována skutečnost chybějících nebo opakovaných zpráv (a my se rozhodneme ukončit spojení);
• identifikace a autentizace účastníků komunikace pomocí předem zadaných veřejných klíčů – již dříve jsme se rozhodli, že vytvoříme síť přátel. Teprve po autentizaci pochopíme, s kým komunikujeme;
• dostupnost dokonalé dopředné utajení vlastnosti (PFS) – kompromitace našeho dlouhodobého podpisového klíče by neměla vést ke schopnosti číst veškerou předchozí korespondenci. Nahrávání zachyceného provozu se stává zbytečným;
• platnost/platnost zpráv (transport a handshake) pouze v rámci jedné TCP relace. Vkládání správně podepsaných/ověřených zpráv z jiné relace (ani se stejným partnerem) by nemělo být možné;
• pasivní pozorovatel by neměl vidět ani uživatelské identifikátory, přenášené dlouhodobé veřejné klíče ani hashe z nich. Jistá anonymita od pasivního pozorovatele.

Překvapivě téměř každý chce mít toto minimum v jakémkoli protokolu handshake a jen velmi málo z výše uvedeného je nakonec splněno pro „domácí“ protokoly. Teď nebudeme vymýšlet nic nového. Určitě bych doporučil používat Noise framework pro stavební protokoly, ale zvolme něco jednoduššího.

Dva nejoblíbenější protokoly jsou:

• TLS - velmi složitý protokol s dlouhou historií chyb, záseků, zranitelností, špatného myšlení, složitosti a nedostatků (to však nemá mnoho společného s TLS 1.3). Ale neuvažujeme o tom, protože je to příliš komplikované.
• IPsec с IKE — nemají vážné kryptografické problémy, i když také nejsou jednoduché. Pokud čtete o IKEv1 a IKEv2, tak jejich zdroj je STS, ISO/IEC IS 9798-3 a protokoly SIGMA (SIGN-and-MAc) – dostatečně jednoduché na implementaci za jeden večer.

Co je dobrého na SIGMA, jako nejnovější článek ve vývoji STS/ISO protokolů? Splňuje všechny naše požadavky (včetně „skrytí“ identifikátorů partnera) a nemá žádné známé kryptografické problémy. Je minimalistická – odstranění alespoň jednoho prvku ze zprávy protokolu povede k její nejistotě.

Pojďme od nejjednoduššího domácího protokolu k SIGMA. Nejzákladnější operace, která nás zajímá, je klíčová dohoda: Funkce, která dává oběma účastníkům stejnou hodnotu, kterou lze použít jako symetrický klíč. Aniž bychom zacházeli do podrobností: každá ze stran vygeneruje pomíjivý (použitý pouze v rámci jedné relace) pár klíčů (veřejný a soukromý klíč), vymění si veřejné klíče, zavolá funkci dohody, na jejímž vstupu předá svůj soukromý klíč a veřejný klíč. klíč partnera.

┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └,─┘ ┘── I┘ ┘── I┘ ┘─ I─ Ad │ ╔══════════ ══════════╗ │───────────────>─ ──>│ ╚DDHA, Hospoda A ════════ ═══════════╝ │ IdB, PubB │ ╔══════════════╕════╕═══════ │<───────── ──────│ ║PrvB, PubB = DHgen()║ │ │ ╚════════════════‐‐ ───┐ ╔════ ═══╧════════════╗ │ ║Klíč = DH(PrvA, PubB)║ <──────┕╕╤ ═══════ ════╝ │ │ │ │

Kdokoli může skočit doprostřed a nahradit veřejné klíče svými vlastními - v tomto protokolu neexistuje žádná autentizace účastníků rozhovoru. Přidejme podpis s dlouhověkými klíči.

┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┬└,─┘ └,─┘ ┘──┘ ┘── A znak(SignPrvA, (PubA)) │ ╔═ Podepište se = load()║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚═══════ ═══════ ══════════╕╂══╕╂ ═══╂ ═══════════╕╂══╕╂ , podepsat (SignPrvB, (PubB)) - ─────────── ───────────── ──│ ║SignPrvB, SignPubB = load( )║ │ │ ║PrvB, PubB = DHgen() ║─ ════════ – ═════╗ │ │ ║ověřit( SignPubB, ...)║ │ <───┘ ║Klíč = DH(Pr vA, PubB) ║ │ │ ╚═══════╕════╕════╕══════ ═╝ │ │ │

Takový podpis nebude fungovat, protože není vázán na konkrétní relaci. Takové zprávy jsou také „vhodné“ pro sezení s ostatními účastníky. Celý kontext se musí přihlásit. To nás nutí přidat také další zprávu od A.

Kromě toho je důležité přidat svůj vlastní identifikátor pod podpis, protože jinak můžeme nahradit IdXXX a znovu podepsat zprávu klíčem jiného známého partnera. Aby se zabránilo reflexní útoky, je nutné, aby prvky pod podpisem byly na jasně definovaných místech podle jejich významu: pokud A značí (PubA, PubB), musí se podepisovat B (PubB, PubA). To také vypovídá o důležitosti výběru struktury a formátu serializovaných dat. Například sady v kódování ASN.1 DER jsou seřazeny: SET OF(PubA, PubB) bude totožné s SET OF(PubB, PubA).

┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └,─┘ ┘── I┘ ┘── I┘ ┘─ I─ Ad │ ╔══════════ ═════════════════╗ │───────────└└└———└—— — ────────── ─────────────>│ ║SignPrvA, SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ──│ ┕║─│ ═══════ ═══════════════╝ │IdB, PubB, sign(SignPrvB, (IdB, PubA, PubB)) ││␕╕╕═␕╕╕═␕╕╕══╕╕═══╕ ═════ ════════════╗ │<─────────────────└———└—— ────────── ─────────│ ║SignPrvB, SignPubB = load()║ │ │ ║PrvB, PubB = DHgen() ║ │ │ │ ═╕╕═ ═╕╕══║╕═╕-║ ═══════ ══════════╝ │ znak (SignPrvA, (IdA, PubB, PubA)) │ ╔═════╕═══╕════╕════╕══ ═══╗ │─ - ───>│ ║ověřit (SignPubB, ...) ║ │ │ ║klíč = dh (prva, PUBB) ║ │ │ │

Stále jsme však „neprokázali“, že jsme pro tuto relaci vygenerovali stejný sdílený klíč. V zásadě se bez tohoto kroku obejdeme - hned první dopravní spojení bude neplatný, ale chceme, abychom po dokončení handshaku měli jistotu, že je opravdu vše domluveno. V tuto chvíli máme k dispozici protokol ISO/IEC IS 9798-3.

Mohli bychom podepsat samotný vygenerovaný klíč. To je nebezpečné, protože je možné, že v použitém podpisovém algoritmu může docházet k únikům (sice bitů na podpis, ale stále dochází k únikům). Je možné podepsat hash derivačního klíče, ale únik dokonce i hashe odvozeného klíče může být cenný při útoku hrubou silou na derivační funkci. SIGMA používá funkci MAC, která ověřuje ID odesílatele.

┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └,─┘ ┘── I┘ ┘── I┘ ┘─ I─ Ad │ ╔══════════ ═════════════════╗ │───────────└└└———└—— — ────────── ──────────────────>│ ║SignPrvA, SignPubA = load()║ │ │ =║DHA│ ║DHA, PubA ═══════ ════════════════════╝ │IdB, PubB, podepsat (SignPrvB, (PubA, PubB)╕ₕₕ ═══ │<───────────────── ─────────────└──└─ —└─ — ────────── ─│ ║SignPrvB, SignPubB = načíst()║ │ │ ║PrvB, PubB = DHgen() ║ │ │ ╚════ ═╕═══╕═══════ ═══════ ══╝ │ │ ╔════════════ ═════════, Publikum A) │ (Ib(Pub)A)Prv. ║Klíč = DH( PrvA, PubB) ║ │───────────────────── ─────—└─——└─— ────────── ─────>│ ║ověřit (klíč, IdB) ║ │ │ ║ověřit (SignPubB, ...)║ │ │ ╚══╕════╕═␕␕═╕═␕════ ═════ ═╝ │ │

Jako optimalizaci mohou někteří chtít znovu použít své pomíjivé klíče (což je samozřejmě pro PFS nešťastné). Například jsme vygenerovali pár klíčů, pokusili se připojit, ale TCP nebyl dostupný nebo byl někde uprostřed protokolu přerušen. Je škoda plýtvat plýtváním entropií a zdroji procesoru na nový pár. Proto zavedeme tzv. cookie – pseudonáhodnou hodnotu, která bude chránit před případnými útoky náhodného přehrání při opětovném použití efemérních veřejných klíčů. Vzhledem k vazbě mezi cookie a efemérním veřejným klíčem může být veřejný klíč protější strany z podpisu odstraněn jako nepotřebný.

┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┬└,─┘ └,─┘ └,── A┘ ┘── A CookieA │ ╔════════ ═══════════════════╗ │──────└└└———└—— — ────────── - ─>│ ║SignPrvA, SignPubA = load( )║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚══════════════════════════════════════════════════ ══╝ │IdB, PubB, CookieB , sign(SignPrvB, (CookieA, CookieB, PubB)), MAC(IdB) │ ╔═══════════════════╕════╕════╕ ╗ │< - ────────── ────────────────────│ ║SignPrvB, SignPubB = načíst()║ │ │ ┕, ║ │ │ DH, ║ │ Prv BDH, │B) B) ╚══════ - ═══════╗ │ podepsat( SignPrvA, (CookieB, CookieA, PubA)), MAC(IdA) │ ║Klíč = DH(PrvA, PubB) ║ │───────────└——└└—— ─ - ───────>│ ║ ověřit (Klíč, IdB) ║ │ │ ║ověřit (SignPubB, ...)║ │ │ ╚══════════╕═══╕════╕═ₕ═╕═ₕ═╕═ │

Nakonec chceme získat soukromí našich konverzačních partnerů od pasivního pozorovatele. Za tímto účelem SIGMA navrhuje nejprve vyměňovat pomíjivé klíče a vyvinout společný klíč, na kterém se budou šifrovat autentizační a identifikační zprávy. SIGMA popisuje dvě možnosti:

• SIGMA-I - chrání iniciátora před aktivními útoky, respondéra před pasivními: iniciátor autentizuje respondéra a pokud něco neodpovídá, neprozradí svou identifikaci. Obžalovaný předá svou identifikaci, pokud je s ním zahájen aktivní protokol. Pasivní pozorovatel se nic nedozví;
  SIGMA-R - chrání respondéra před aktivními útoky, iniciátora před pasivními. Všechno je přesně naopak, ale v tomto protokolu jsou již přenášeny čtyři zprávy o podání ruky.

  Vybrali jsme SIGMA-I, protože se více podobá tomu, co očekáváme od známých věcí typu klient-server: klienta rozpozná pouze ověřený server a server už zná každý. Navíc se snáze implementuje díky menšímu počtu zpráv typu handshake. Vše, co do protokolu přidáváme, je zašifrovat část zprávy a přenést identifikátor A do zašifrované části poslední zprávy:

  PubA, CookieA │ ╔══════════ ═════════════――└―――└―――└――― ────────── ───── ────────── ─────────────—————— ─────────── – ═══════ ═════════ ════╝ │ PubB, CookieB, Enc((IdB, sign(SignPrvB, (CookieA, CookieB, PubB)), MAC(IdB)╕═␐)╕)═)╕) ═════════ ═══════════════ ═══════╗ │<───—└———└— — ────────── ───── ────────── ║SignP rvB, SignPubB = load()║ │ │ ═ PrvB, PubB = DHgen() ║ DHgen() ║ ═══════ - ══╗ │ Enc((IdA, sign( SignPrvA, (CookieB, CookieA, PubA)), MAC(IdA))) │ ║Klíč = DH(PrvA, PubB) ║ │──────────└─—─└─—— - ─────────── ──────>│ ║ověřit (klíč, IdB) ║ │ │ ║ověřit ( SignPubB, ...)║ │ │ ╚═╕═══╕════════════ ═════ ══╝ │ │
  

  • GOST R se používá pro podpis 34.10-2012 algoritmus s 256bitovými klíči.
  • Pro vygenerování veřejného klíče se používá 34.10 VKO.
  • CMAC se používá jako MAC. Technicky se jedná o speciální režim provozu blokové šifry, popsaný v GOST R 34.13-2015. Jako šifrovací funkce pro tento režim − Kobylka (34.12-2015).
  • Hash jeho veřejného klíče se používá jako identifikátor partnera. Používá se jako hash Stribog-256 (34.11. 2012. 256 XNUMX bitů).

  Po podání ruky se dohodneme na sdíleném klíči. Můžeme jej použít pro autentizované šifrování transportních zpráv. Tato část je velmi jednoduchá a těžko uděláme chybu: inkrementujeme počítadlo zpráv, zašifrujeme zprávu, ověříme (MAC) počítadlo a šifrovaný text, odešleme. Při příjmu zprávy zkontrolujeme, že počítadlo má očekávanou hodnotu, šifrovaný text ověříme počítadlem a dešifrujeme. Jaký klíč bych měl použít k šifrování zpráv handshake, přenosu zpráv a jak je ověřit? Používání jednoho klíče pro všechny tyto úkoly je nebezpečné a nerozumné. Klíče je nutné generovat pomocí specializovaných funkcí KDF (funkce odvození klíče). Opět si netřepeme vlasy a něco vymyslíme: HKDF je dlouho známá, dobře prozkoumaná a nemá žádné známé problémy. Bohužel nativní knihovna Pythonu tuto funkci nemá, takže používáme hkdf Igelitová taška. HKDF interně používá HMAC, který zase používá hashovací funkci. Příklad implementace v Pythonu na stránce Wikipedie zabere jen pár řádků kódu. Stejně jako v případě 34.10 použijeme jako hashovací funkci Stribog-2012. Výstup naší funkce shody klíčů se bude nazývat klíč relace, ze kterého se vygenerují chybějící symetrické:

  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")
  

  Struktury/schéma

  Podívejme se, jaké struktury ASN.1 nyní máme pro přenos všech těchto dat:

  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 je to, co bude podepsáno. HandshakeTBE - co bude zašifrováno. Upozorňuji na pole ukm v MsgHandshake1. 34.10 VKO, pro ještě větší randomizaci generovaných klíčů, obsahuje parametr UKM (uživatelský klíčovací materiál) - jen další entropie.

  Přidání kryptografie do kódu

  Uvažujme pouze změny provedené v původním kódu, protože framework zůstal stejný (ve skutečnosti byla nejprve napsána konečná implementace a poté z ní byla vyříznuta veškerá kryptografie).

  Vzhledem k tomu, že autentizace a identifikace účastníků budou probíhat pomocí veřejných klíčů, je nyní třeba je někde ukládat na dlouhou dobu. Pro jednoduchost používáme JSON takto:

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

  náš - náš pár klíčů, hexadecimální soukromý a veřejný klíč. jejich — jména účastníků rozhovoru a jejich veřejné klíče. Pojďme změnit argumenty příkazového řádku a přidat následné zpracování dat 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),
      }
  # }}}
  

  Soukromý klíč algoritmu 34.10 je náhodné číslo. 256bitová velikost pro 256bitové eliptické křivky. PyGOST nepracuje se sadou bajtů, ale s vysoká čísla, takže náš soukromý klíč (urandom(32)) je třeba převést na číslo pomocí gost3410.prv_unmarshal(). Veřejný klíč je určen deterministicky ze soukromého klíče pomocí gost3410.public_key(). Veřejný klíč 34.10 jsou dvě velká čísla, která je také potřeba převést na bajtovou sekvenci pro snadné ukládání a přenos pomocí gost3410.pub_marshal().

  Po přečtení souboru JSON je třeba veřejné klíče převést zpět pomocí gost3410.pub_unmarshal(). Vzhledem k tomu, že z veřejného klíče obdržíme identifikátory účastníků ve formě hashe, lze je okamžitě předem vypočítat a umístit do slovníku pro rychlé vyhledávání. Hash Stribog-256 je gost34112012256.GOST34112012256(), který plně vyhovuje rozhraní hashlib hashovacích funkcí.

  Jak se změnil korutin iniciátoru? Vše je podle schématu handshake: vygenerujeme soubor cookie (128 bitů je dostatek), pomíjivý pár klíčů 34.10, který bude použit pro funkci dohody klíčů 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()
  

  • čekáme na odpověď a dekódujeme příchozí zprávu Msg;
  • ujistěte se, že dostanete handshake1;
  • dekódovat pomíjivý veřejný klíč protější strany a vypočítat klíč relace;
  • Generujeme symetrické klíče potřebné pro zpracování TBE části zprávy.

   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 je 64bitové číslo (urandom(8)), které také vyžaduje deserializaci z jeho bytové reprezentace pomocí gost3410_vko.ukm_unmarshal(). Funkce VKO pro 34.10 2012-bit je gost256_vko.kek_3410() (KEK - šifrovací klíč).

  Vygenerovaný klíč relace je již 256bitová pseudonáhodná sekvence bajtů. Proto může být okamžitě použit ve funkcích HKDF. Protože GOST34112012256 vyhovuje rozhraní hashlib, lze jej okamžitě použít ve třídě Hkdf. Salt (první argument Hkdf) neuvádíme, protože vygenerovaný klíč bude kvůli pomíjivosti zúčastněných párů klíčů pro každou relaci jiný a již obsahuje dostatek entropie. kdf.expand() ve výchozím nastavení již vytváří 256bitové klíče požadované pro Grasshopper později.

  Dále se zkontrolují části TBE a TBS příchozí zprávy:

  • MAC přes příchozí šifrový text se vypočítá a zkontroluje;
  • šifrovaný text je dešifrován;
  • Struktura TBE je dekódována;
  • odebere se z něj identifikátor partnera a zkontroluje se, zda je nám vůbec znám;
  • MAC přes tento identifikátor se vypočítá a zkontroluje;
  • je ověřen podpis nad strukturou TBS, která zahrnuje cookie obou stran a veřejný efemérní klíč protější strany. Podpis je ověřen trvanlivým podpisovým klíčem účastníka jednání.

   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"]
  

  Jak jsem psal výše, 34.13 popisuje různé provozní režimy blokové šifry ze dne 34.12. Mezi nimi je režim pro generování imitací vložek a výpočty MAC. V PyGOST je to gost2015.mac(). Tento režim vyžaduje předání šifrovací funkce (příjem a vrácení jednoho bloku dat), velikost šifrovacího bloku a vlastně i data samotná. Proč nemůžete pevně zakódovat velikost šifrovacího bloku? 3413 popisuje nejen 34.12bitovou šifru Grasshopper, ale také 2015bitovou Magma - mírně upravený GOST 28147-89, vytvořený zpět v KGB a stále má jeden z nejvyšších bezpečnostních prahů.

  Kuznechik je inicializován voláním gost.3412.GOST3412Kuznechik(key) a vrací objekt s metodami .encrypt()/.decrypt() vhodnými pro předání funkcím 34.13. MAC se vypočítá následovně: gost3413.mac(GOST3412Kuznechik(klíč).šifrovat, KUZNECHIK_BLOCKSIZE, šifrový text). Pro porovnání vypočtené a přijaté MAC nemůžete použít obvyklé srovnání (==) bajtových řetězců, protože tato operace uniká porovnávací čas, což v obecném případě může vést k fatálním zranitelnostem, jako je BESTIE útoky na TLS. Python má k tomu speciální funkci hmac.compare_digest.

  Funkce blokové šifry dokáže zašifrovat pouze jeden blok dat. Pro větší počet, a to ani ne násobek délky, je nutné použít režim šifrování. 34.13-2015 popisuje následující: ECB, CTR, OFB, CBC, CFB. Každý má své vlastní přijatelné oblasti použití a vlastnosti. Bohužel stále nemáme standardizované ověřené režimy šifrování (jako CCM, OCB, GCM a podobně) - jsme nuceni si MAC alespoň přidat sami. vybírám si režim čítače (CTR): nevyžaduje výplň do velikosti bloku, lze paralelizovat, používá pouze funkci šifrování, lze jej bezpečně použít k šifrování velkého množství zpráv (na rozdíl od CBC, které má kolize poměrně rychle).

  Stejně jako .mac() přijímá .ctr() podobný vstup: šifrový text = gost3413.ctr(GOST3412Kuznechik(klíč).encrypt, KUZNECHIK_BLOCKSIZE, prostý text, iv). Je nutné zadat inicializační vektor, který je přesně poloviční než délka šifrovacího bloku. Pokud je náš šifrovací klíč použit pouze k zašifrování jedné zprávy (i když z více bloků), pak je bezpečné nastavit nulový inicializační vektor. K šifrování zpráv typu handshake používáme pokaždé samostatný klíč.

  Ověření podpisu gost3410.verify() je triviální: předáme eliptickou křivku, ve které pracujeme (prostě ji zaznamenáme v našem protokolu GOSTIM), veřejný klíč podepisujícího (nezapomeňte, že by to měla být n-tice dvou velká čísla, nikoli bajtový řetězec), hash 34.11. 2012. XNUMX a samotný podpis.

  Dále v iniciátoru připravíme a odešleme handshake zprávu do handshake2, přičemž provedeme stejné akce jako při ověřování, pouze symetricky: podepisování našich klíčů místo kontroly atd...

   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)
   

  Po navázání relace se vygenerují přenosové klíče (samostatný klíč pro šifrování, pro ověřování, pro každou ze stran) a inicializuje se Grasshopper, aby dešifroval a zkontroloval 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     # }}}
  

  Korutina msg_sender nyní šifruje zprávy před jejich odesláním na TCP spojení. Každá zpráva má monotónně rostoucí nonce, což je také inicializační vektor při zašifrování v režimu čítače. Je zaručeno, že každá zpráva a blok zpráv bude mít jinou hodnotu čítače.

  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
  

  Příchozí zprávy jsou zpracovávány rutinou msg_receiver, která zajišťuje ověřování a dešifrování:

  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)
  

  Závěr

  GOSTIM je určen výhradně pro vzdělávací účely (protože se na něj nevztahují alespoň testy)! Zdrojový kód programu lze stáhnout zde (Стрибог-256 хэш: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Как и все мои проекты, типа GoGOST, PyDERASN, NCCP, GoVPN, GOSTIM je zcela svobodný softwaredistribuován za podmínek GPLv3 +.

  Sergej Matvejev, cypherpunk, člen Nadace SPO, Python/Go-vývojář, hlavní specialista Federal State Unitary Enterprise "STC "Atlas".

Zdroj: www.habr.com