🥇GOSTIM: P2P F2F E2EE IM in einer Nacht mit GOST-Kryptographie

Als Entwickler PyGOST einer Bibliothek (GOST-kryptografische Primitive in reinem Python) bekomme ich häufig Fragen, wie man unkompliziert eine einfache sichere Nachrichtenübermittlung implementiert. Viele glauben, dass angewandte Kryptografie ziemlich einfach ist und ein Aufruf von .encrypt() bei einem Blockcipher ausreicht, um eine sichere Übertragung über einen Kommunikationskanal zu gewährleisten. Andere denken, dass angewandte Kryptografie nur etwas für wenige ist und es akzeptabel ist, dass wohlhabende Unternehmen wie Telegram Mitarbeiter mit mathematischem Hintergrund nicht in der Lage sind, ein sicheres Protokoll zu implementieren.

Um das zu entkräften, habe ich diesen Artikel geschrieben, um zu zeigen, dass die Implementierung kryptografischer Protokolle und sicherer IMs keine allzu komplizierte Aufgabe ist. Dennoch sollte man keine eigenen Authentifizierungs- und Schlüsselvereinbarungsprotokolle erfinden.

Im Artikel wird ein Peer-to-Peer, Friend-to-Friend, End-to-End-verschlüsselter Instant Messenger mit SIGMA-I Authentifizierungs- und Schlüsselvereinigungsprotokoll (auf dessen Basis IPsec IKE) realisiert wird, wobei ausschließlich GOST-kryptografische Algorithmen der PyGOST-Bibliothek und ASN.1-Codierung von Nachrichten über die Bibliothek verwendet werden PyDERASN (über die ich bereits früher geschrieben habe.). Eine notwendige Bedingung: Es muss so einfach sein, dass man es an einem Abend (oder Arbeitstag) von Grund auf schreiben kann, sonst ist es kein einfaches Programm mehr. Es wird sicherlich Fehler, überflüssige Komplexität und Mängel enthalten, zudem ist es mein erstes Programm mit der asyncio-Bibliothek.

IM-Design

Zunächst müssen wir verstehen, wie unser IM aussehen wird. Zur Vereinfachung lassen Sie uns ein Peer-to-Peer-Netzwerk wählen, ohne Teilnehmererkennung. Wir werden manuell die Adresse und den Port angeben, um mit dem Gesprächspartner zu kommunizieren.

Mir ist bewusst, dass die Annahme, dass eine direkte Verbindung zwischen zwei beliebigen Computern möglich ist, derzeit eine wesentliche Einschränkung für die praktische Anwendbarkeit von IM darstellt. Doch je mehr Entwickler verschiedene NAT-Traversal-Lösungen umsetzen, desto länger werden wir im IPv4-Internet bleiben, mit einer frustrierenden Wahrscheinlichkeit für Verbindungen zwischen beliebigen Computern. Wie lange können wir das Fehlen von IPv6 zu Hause und bei der Arbeit noch ertragen?

Wir werden ein Freund-zu-Freund-Netzwerk haben: Alle möglichen Gesprächspartner müssen im Voraus bekannt sein. Erstens vereinfacht das alles erheblich: Man stellt sich vor, findet das Name/Schlüssel oder findet es nicht, trennt die Verbindung oder setzt die Arbeit fort, im Wissen um den Gesprächspartner. Zweitens ist es im Allgemeinen sicher und schließt viele Angriffe aus.

Die IM-Oberfläche wird klassischen Lösungen ähneln. Suckless-Projekten, die mir wegen ihres Minimalismus und der Unix-Way-Philosophie sehr gefallen. Das IM-Programm erstellt für jeden Gesprächspartner ein Verzeichnis mit drei Unix-Domain-Sockets:

in — hier werden die an den Gesprächspartner gesendeten Nachrichten geschrieben;
out — hier werden die von dem Gesprächspartner empfangenen Nachrichten gelesen;
state — durch Lesen davon erfahren wir, ob der Gesprächspartner momentan verbunden ist, sowie die Adresse/Port der Verbindung.

Außerdem wird ein conn-Socket erstellt; durch das Schreiben des Host-Ports initiieren wir eine Verbindung zu dem entfernten Gesprächspartner.

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

Dieser Ansatz ermöglicht unabhängige Implementierungen des IM-Transports und der Benutzeroberfläche, denn über Geschmack lässt sich bekanntlich streiten, es ist schwer, es jedem recht zu machen. Mit tmux und/oder multitail, man kann eine mehrteilige Schnittstelle mit Syntax-Highlighting erhalten. Und mit Hilfe von rlwrap kann man eine GNU Readline-kompatible Eingabezeile für Nachrichten erstellen.

Tatsächlich verwenden die Suckless-Projekte FIFO-Dateien. Ich persönlich konnte nicht verstehen, wie man in asyncio mit Dateien konkurrieren kann, ohne eine eigene Abdeckung aus dedizierten Threads (für solche Dinge benutze ich schon lange die Sprache Go). Daher habe ich mich entschieden, Unix-Domain-Sockets zu verwenden. Leider macht das es unmöglich, echo 2001:470:dead::babe 6666 > conn auszuführen. Ich habe dieses Problem gelöst, indem ich socat: echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.

Das ursprüngliche unsichere Protokoll

Verwendet wird TCP als Transport: es garantiert die Zustellung und deren Reihenfolge. UDP garantiert weder das eine noch das andere (was nützlich wäre, wenn Kryptografie zur Anwendung kommt), und Unterstützung für SCTP gibt es in Python nicht out of the box.

Leider gibt es im TCP kein Konzept von Nachrichten, sondern nur von Byte-Strömen. Deshalb müssen wir ein Format für Nachrichten erfinden, um sie innerhalb dieses Stroms unterscheiden zu können. Wir können uns darauf einigen, das Zeilenumbruchzeichen zu verwenden. Das wäre für den Anfang in Ordnung, aber wenn wir damit beginnen, unsere Nachrichten zu verschlüsseln, kann dieses Zeichen überall im verschlüsselten Text auftreten. Deshalb sind in Netzwerken Protokolle populär, die anfänglich die Länge der Nachricht in Bytes senden. Zum Beispiel gibt es in Python die Bibliothek xdrlib, die das Arbeiten mit einem solchen Format ermöglicht. XDR.

Wir werden nicht richtig und effizient mit dem Lesen von TCP arbeiten — wir vereinfachen den Code. Wir lesen in einer Endlosschleife Daten aus dem Socket, bis wir die vollständige Nachricht dekodiert haben. Für einen solchen Ansatz kann man auch JSON oder XML als Format verwenden. Aber wenn die Kryptografie hinzukommt, müssen die Daten signiert und authentifiziert werden — das erfordert eine bytegenaue identische Darstellung der Objekte, was JSON/XML nicht gewährleisten (die Dumps können unterschiedlich sein).

XDR eignet sich für diese Aufgabe, jedoch wähle ich ASN.1 mit DER-Codierung und PyDERASN eine Bibliothek, da wir mit hochgradigen Objekten arbeiten werden, die häufig angenehmer und bequemer zu handhaben sind. Im Gegensatz zu schemalosen Ansätzen. bencode, MessagePack oder CBOR, wird ASN.1 die Daten automatisch anhand des festgelegten Schemas überprüfen.

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

Die akzeptierte Nachricht wird Msg sein: entweder eine Textnachricht MsgText (zunächst mit einem einzigen Textfeld) oder eine Handshake-Nachricht MsgHandshake (in der der Name des Gesprächspartners übertragen wird). Das wirkt momentan kompliziert, aber es ist eine Grundlage für die Zukunft.

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

IM ohne Kryptografie

Wie bereits erwähnt, wird für alle Socket-Operationen die asyncio-Bibliothek verwendet. Lassen Sie uns festlegen, was wir beim Start erwarten:

parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
    "--our-name",
    required=True,
    help="Unser Peer-Name",
)
parser.add_argument(
    "--their-names",
    required=True,
    help="Ihre Peer-Namen, durch Kommas getrennt",
)
parser.add_argument(
    "--bind",
    default="::1",
    help="Adresse, auf die gehört werden soll",
)
parser.add_argument(
    "--port",
    type=int,
    default=6666,
    help="Port, auf den gehört werden soll",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))

Setzen Sie den eigenen Namen fest (—our-name alice). Die erwarteten Gesprächspartner werden durch Kommas getrennt aufgelistet (—their-names bob,eve). Für jeden Gesprächspartner wird ein Verzeichnis mit Unix-Sockets erstellt, sowie eine Coroutine für jeden 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"))

Eingehende Nachrichten vom Benutzer aus dem in Socket werden in die IN_QUEUES-Warteschlangen gesendet:

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

Die von den Gesprächspartnern gesendeten Nachrichten werden in die OUT_QUEUES-Warteschlangen gesendet, aus denen die Daten in den Out-Socket geschrieben werden:

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

Beim Lesen vom State-Socket sucht das Programm im PEER_ALIVE-Dictionary nach der Adresse des Gesprächspartners. Wenn noch keine Verbindung zum Gesprächspartner besteht, wird eine leere Zeichenfolge geschrieben.

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

Beim Schreiben der Adresse in den Conn-Socket wird die «Initiator»-Funktion für die Verbindung gestartet:

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

Betrachten wir den Initiator. Zuerst öffnet er offensichtlich eine Verbindung zum angegebenen Host/Port und sendet eine Handshake-Nachricht mit seinem Namen:

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

Dann wartet es auf die Antwort der entfernten Seite. Es versucht, die empfangene Antwort gemäß dem Msg ASN.1-Schema zu dekodieren. Wir gehen davon aus, dass die gesamte Nachricht in einem TCP-Segment gesendet wird und wir sie atomar bei Aufruf von .read() erhalten. Wir überprüfen, ob wir tatsächlich die Handshake-Nachricht erhalten haben.

 141     # Warte auf Handshake-Nachricht {{{
 142     data = await reader.read(256)
 143     if data == b"":
 144         logging.warning("%s: keine Antwort, trenne Verbindung", _id)
 145         writer.close()
 146         return
 147     try:
 148         msg, _ = Msg().decode(data)
 149     except ASN1Error:
 150         logging.warning("%s: nicht dekodierbare Antwort, trenne Verbindung", _id)
 151         writer.close()
 152         return
 153     logging.info("%s: erhielt %s Nachricht", _id, msg.choice)
 154     if msg.choice != "handshake":
 155         logging.warning("%s: unerwartete Nachricht, trenne Verbindung", _id)
 156         writer.close()
 157         return
 158     # }}}

Wir überprüfen, ob uns der Name des Gesprächspartners bekannt ist. Wenn nicht, wird die Verbindung getrennt. Außerdem prüfen wir, ob bereits eine Verbindung zu ihm bestand (der Gesprächspartner hat erneut einen Verbindungsauftrag an uns gesendet) und schließen diese. In IN_QUEUES werden Python-Strings mit dem Text der Nachricht platziert, wobei ein spezieller Wert None signalisiert, dass die msg_sender-Koroutine ihre Arbeit einstellen soll, um sich von ihrem Writer, der mit der abgelaufenen TCP-Verbindung verbunden ist, zu trennen.

 159     msg_handshake = msg.value
 160     peer_name = str(msg_handshake["peerName"])
 161     if peer_name not in THEIR_NAMES:
 162         logging.warning("unbekannter Peer-Name: %s", peer_name)
 163         writer.close()
 164         return
 165     logging.info("%s: Sitzung hergestellt: %s", _id, peer_name)
 166     # Textnachrichtensender starten, Transportdecoder initialisieren {{{
 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 empfängt ausgehende Nachrichten (die aus der in Socket in die Warteschlange eingelegt werden), serialisiert sie in eine MsgText-Nachricht und sendet sie über die TCP-Verbindung. Diese kann jederzeit abbrechen — das erfassen wir aktiv.

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

Am Ende geht der Initiator in eine Endlosschleife, um Nachrichten vom Socket zu lesen. Er prüft, ob es sich um Textnachrichten handelt, und fügt sie der OUT_QUEUES-Warteschlange hinzu, von der sie an den entsprechenden Kommunikationspartner gesendet werden. Warum kann man nicht einfach .read() durchführen und die Nachricht dekodieren? Weil es möglich ist, dass mehrere Nachrichten vom Benutzer im Puffer des Betriebssystems aggregiert und in einem einzigen TCP-Segment gesendet werden. Wir könnten die erste dekodieren, aber danach könnte ein Teil der nächsten im Puffer bleiben. Bei jeder abnormalen Situation schließen wir die TCP-Verbindung und stoppen die msg_sender-Koroutine (indem wir None in die OUT_QUEUES-Warteschlange senden).

 174     buf = b""
 175     # Auf Testnachrichten warten {{{
 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: Maximale Puffergröße überschritten", _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: Unerwartete %s-Nachricht", _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: trennt die Verbindung: %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: Empfangene %d Zeichen-Nachricht", peer_name, len(text))
  69     await OUT_QUEUES[peer_name].put(text)

Kehren wir zum Hauptcode zurück. Nach der Erstellung aller Koroutinen starten wir beim Programmstart den TCP-Server. Für jede hergestellte Verbindung wird eine Responder-Koroutine erstellt.

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("Höre auf: %s", server.sockets[0].getsockname())
loop.run_forever()

Der Responder ähnelt dem Initiator und führt dieselben Aktionen spiegelbildlich aus. Um die Handhabung zu erleichtern, wird jedoch sofort eine unendliche Schleife zum Lesen der Nachrichten gestartet. Aktuell sendet das Handshake-Protokoll jeweils eine Nachricht von beiden Seiten. Zukünftig wird der Initiator jedoch zwei Nachrichten senden, bevor Textnachrichten sofort verschickt werden können.

  72 async def responder(reader, writer):
  73     _id = writer.get_extra_info("peername")
  74     logging.info("%s: verbunden", _id)
  75     buf = b""
  76     msg_expected = "handshake"
  77     peer_name = None
  78     while True:
  79         # Lesen, bis wir die Msg-Nachricht erhalten {{{
  80         data = await reader.read(MaxMsgLen)
  81         if data == b"":
  82             logging.info("%s: Verbindung geschlossen", _id)
  83             break
  84         buf += data
  85         if len(buf) > MaxMsgLen:
  86             logging.warning("%s: Maximale Pufferspeichergröße überschritten", _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: unerwartete %s-Nachricht", _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         # Handshake-Nachricht verarbeiten {{{
 104         elif msg_expected == "handshake":
 105             logging.info("%s: %s-Nachricht erhalten", _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("unbekannter 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: Sitzung etabliert: %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: trennt sich", _id)
 125     if msg_expected == "text":
 126         IN_QUEUES[peer_name].put(None)
 127     writer.close()

Sicheres Protokoll

Es ist an der Zeit, unsere Kommunikation zu sichern. Was meinen wir mit Sicherheit und was möchten wir:

Vertraulichkeit der übermittelten Nachrichten;
Authentizität und Integrität der übermittelten Nachrichten – ihre Veränderung muss erkannt werden;
Schutz vor Replay-Angriffen – das Verschwinden oder Wiederholen von Nachrichten muss erkannt werden (und wir entscheiden, die Verbindung zu trennen);
Identifizierung und Authentifizierung der Gesprächspartner anhand vorher festgelegter öffentlicher Schlüssel – wir haben bereits entschieden, dass wir ein Friend-to-Friend-Netzwerk aufbauen. Nur nach der Authentifizierung wissen wir, mit wem wir kommunizieren;
die Existenz Perfekte Vorwärtsgeheimhaltung Eigenschaften (PFS) – die Kompromittierung unseres langlebigen Signaturschlüssels darf nicht dazu führen, dass frühere Kommunikationen lesbar werden. Aufgezeichnetem Datenverkehr wird damit wertlos;
Gültigkeit/Validität der Nachrichten (Transport- und Handshake-Nachrichten) nur innerhalb einer TCP-Session. Das Einfügen korrekt signierter/authentifizierter Nachrichten aus einer anderen Session (selbst mit dem gleichen Gesprächspartner) sollte nicht möglich sein;
Ein passiver Beobachter sollte weder Benutzer-IDs noch übertragene langlebige öffentliche Schlüssel oder deren Hashes sehen. Eine gewisse Anonymität gegenüber dem passiven Beobachter ist wichtig.

Überraschenderweise möchten dies fast alle in jedem Handshake-Protokoll haben, aber nur sehr wenig davon wird für 'hausgemachte' Protokolle letztendlich umgesetzt. Lassen Sie uns auch diesmal nichts Neues erfinden. Ich würde definitiv empfehlen, das Noise-Framework zur Erstellung von Protokollen zu verwenden, aber wählen wir etwas Einfacheres.

Zwei Protokolle sind am beliebtesten:

TLS — ein äußerst komplexes Protokoll mit einer langen Geschichte von Bugs, Fehlern, Sicherheitsanfälligkeiten, mangelnder Überlegung, Komplexität und Mängeln (obwohl dies für TLS 1.3 nur wenig zutrifft). Aber wir ziehen es vor, es aufgrund seiner Überkomplizierung nicht zu betrachten.
IPsec mit IKE — haben keine ernsthaften kryptografischen Probleme, sind aber ebenfalls nicht einfach. Wenn man über IKEv1 und IKEv2 liest, stammen sie von STS, ISO/IEC IS 9798-3 und SIGMA (SIGn-and-MAc) Protokollen, die ausreichend einfach sind, um sie an einem Abend zu implementieren.

Was macht SIGMA, als letzte Entwicklung von STS/ISO-Protokollen, so gut? Es erfüllt all unsere Anforderungen (einschließlich der "Verschleierung" der Identitäten der Gesprächspartner) und hat keine bekannten kryptografischen Probleme. Es ist minimalistisch – das Entfernen auch nur eines Elements aus der Protokollnachricht führt zu seiner Unsicherheit.

Lassen Sie uns vom einfachsten selbstgemachten Protokoll zu SIGMA übergehen. Die grundlegendste für uns interessante Operation ist Schlüsselvereinbarung: eine Funktion, bei der beide Teilnehmer denselben Wert erhalten, der als symmetrischer Schlüssel verwendet werden kann. Ohne ins Detail zu gehen: Jede Seite generiert ein ephemeres (nur für eine Sitzung verwendetes) Schlüsselpaar (öffentlicher und privater Schlüssel), tauscht öffentliche Schlüssel aus und ruft die Vereinbarungsfunktion auf, in die sie ihren privaten Schlüssel und den öffentlichen Schlüssel des Gesprächspartners eingeben.

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

Jeder kann sich einmischen und die öffentlichen Schlüssel durch eigene ersetzen – in diesem Protokoll gibt es keine Authentifizierung der Gesprächspartner. Lassen Sie uns eine Signatur mit langlebigen Schlüsseln hinzufügen.

┌─────┐                            ┌─────┐
│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) ║   │
   │        ╚═════════════════════╝   │
   │                                  │

Diese Signatur ist nicht geeignet, da sie nicht an eine spezifische Sitzung gebunden ist. Solche Nachrichten eignen sich auch für Sitzungen mit anderen Teilnehmern. Der gesamte Kontext muss signiert werden. Dies zwingt dazu, zusätzlich eine Nachricht von A zu senden.

Zudem ist es entscheidend, dass die Unterschrift und die eigene Kennung hinzugefügt werden, da wir ansonsten IdXXX ändern und die Nachricht mit dem Schlüssel eines anderen bekannten Gesprächspartners neu signieren könnten. Um dies zu vermeiden, sind Reflection-Angriffe, ist es notwendig, dass die Elemente der Unterschrift logisch in festgelegten Positionen angeordnet sind: Wenn A (PubA, PubB) signiert, sollte B (PubB, PubA) signieren. Dies zeigt auch die Bedeutung der Wahl der Struktur und des Formats der serialisierten Daten. Zum Beispiel werden Mengen in ASN.1 DER Kodierung sortiert: SET OF(PubA, PubB) ist identisch mit SET OF(PubB, PubA).

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

Wir haben jedoch immer noch nicht "bewiesen", dass wir einen gemeinsamen Schlüssel für diese Sitzung erzeugt haben. Im Grunde könnte man auf diesen Schritt verzichten – die erste Transportnachricht wäre ungültig, aber wir möchten sicherstellen, dass alles wirklich abgestimmt ist, wenn das Handshake abgeschlossen ist. Momentan haben wir das ISO/IEC IS 9798-3 Protokoll zur Verfügung.

Wir könnten auch den erzeugten Schlüssel signieren. Das ist riskant, da es möglich ist, dass im verwendeten Signaturalgorithmus Lecks auftreten können (auch wenn es sich nur um Bit-Flecken handelt, bleibt es dennoch ein Leak). Man kann den Hash des erzeugten Schlüssels signieren, aber auch ein Leak dieses Hashs könnte wertvoll sein bei Brute-Force-Angriffen auf die Erzeugungsfunktion. SIGMA verwendet eine MAC-Funktion, die die Identität des Absenders authentifiziert.

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

Zur Optimierung möchten einige möglicherweise ihre ephemeral keys wiederverwenden (was natürlich nachteilig für PFS ist). Zum Beispiel haben wir ein Schlüsselpaar erzeugt, versucht uns zu verbinden, aber TCP war nicht verfügbar oder brach mitten im Protokoll ab. Es ist schade, die verwendete Entropie und die CPU-Ressourcen für ein neues Paar aufzuwenden. Daher führen wir sozusagen ein Cookie ein — einen pseudo-zufälligen Wert, der vor möglichen zufälligen Replay-Angriffen beim Wiederverwenden der ephemeral public keys schützt. Aufgrund der Bindung zwischen dem Cookie und dem ephemeral public key kann der öffentliche Schlüssel des Gegenübers aus der Signatur entfernt werden, da er nicht mehr benötigt wird.

┌─────┐                                                                 ┌─────┐
│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, ...)║
   │                                                                       │ ╚═════════════════════╝
   │                                                                       │

Zuletzt möchten wir die Privatsphäre unserer Kommunikationspartner vor passiven Beobachtern schützen. Zu diesem Zweck bietet SIGMA zunächst den Austausch von ephemeral Keys an, um einen gemeinsamen Schlüssel zu generieren, mit dem authentifizierende und identifizierende Nachrichten verschlüsselt werden. SIGMA beschreibt hierzu zwei Varianten:

SIGMA-I – schützt den Initiator vor aktiven Angriffen und den Antwortenden vor passiven. Der Initiator authentifiziert den Antwortenden, und falls es Unstimmigkeiten gibt, gibt er seine Identität nicht preis. Der Antwortende hingegen gibt seine Identität preis, wenn ein aktives Protokoll gestartet wird. Ein passiver Beobachter erfährt nichts;
SIGMA-R – schützt den Antwortenden vor aktiven Angriffen und den Initiator vor passiven. Hier ist alles genau umgekehrt, jedoch werden in diesem Protokoll bereits vier Handshake-Nachrichten übermittelt.

Wir wählen SIGMA-I, da es den Erwartungen an traditionelle Client-Server-Anwendungen näher kommt: Der Client wird nur vom authentifizierten Server erkannt, während der Server bereits alles kennt. Zudem ist es aufgrund der geringeren Anzahl an Handshake-Nachrichten einfacher umzusetzen. Alles, was wir in das Protokoll einbringen, ist die Verschlüsselung eines Teils der Nachricht und die Übertragung der ID A in den verschlüsselten Teil der letzten Nachricht:

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

Zum Signieren wird GOST R verwendet 34.10-2012 Algorithmus mit 256-Bit-Schlüsseln.
Zur Ableitung des gemeinsamen Schlüssels wird 34.10-2012 VKO verwendet.
Es wird CMAC als MAC verwendet. Technisch handelt es sich um einen speziellen Betriebsmodus eines Blockcipher, der in GOST R 34.13-2015 beschrieben ist. Als Verschlüsselungsfunktion für diesen Modus dient Grashüpfer (34.12-2015).
Als Identifikator des Gesprächspartners wird der Hash seines öffentlichen Schlüssels verwendet. Als Hash wird angewendet Stribog-256 (34.11-2012 256 Bit).

Nach dem Handshake haben wir einen vereinbarten gemeinsamen Schlüssel. Diesen können wir zur authentifizierten Verschlüsselung von Transportnachrichten verwenden. Dieser Teil ist sehr einfach und kaum fehleranfällig: Zählen der Nachrichten, Verschlüsseln der Nachricht, Authentifizieren (MAC) des Zählers und des Chiffretextes, Senden. Bei Empfang der Nachricht überprüfen wir, ob der Zähler den erwarteten Wert hat, authentifizieren den Chiffretext mit dem Zähler und entschlüsseln. Mit welchem Schlüssel sollen die Handshake-Nachrichten und Transportnachrichten verschlüsselt werden, und welcher ist zur Authentifizierung zu verwenden? Einen Schlüssel für all diese Aufgaben zu verwenden, ist gefährlich und unvernünftig. Es müssen Schlüssel unter Verwendung spezialisierter Funktionen erzeugt werden KDF (Schlüsselleitungsfunktion). Lassen Sie uns nicht den Verstand verlieren und etwas erfinden: HKDF ist seit langem bekannt, gut erforscht und hat keine bekannten Probleme. Leider fehlt diese Funktion in der Standardbibliothek von Python, daher verwenden wir hkdf Paket. HKDF verwendet intern HMAC, das wiederum eine Hash-Funktion nutzt. Ein Beispiel für die Implementierung in Python auf der Wikipedia-Seite umfasst nur wenige Codezeilen. Wie im Fall von 34.10-2012 verwenden wir als Hash-Funktion Streebog-256. Der Ausgang unserer Schlüsselvereinbarungsfunktion wird als Sitzungsschlüssel bezeichnet, aus dem die fehlenden symmetrischen Schlüssel abgeleitet werden:

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

Strukturen/Schemas

Lassen Sie uns nun ansehen, welche ASN.1-Strukturen wir für die Übertragung all dieser Daten erhalten haben:

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 — das, was signiert werden soll (to be signed). HandshakeTBE — das, was verschlüsselt werden soll (to be encrypted). Ich weise auf das Feld ukm in MsgHandshake1 hin. 34.10 VKO, um die Randomisierung der erzeugten Schlüssel weiter zu erhöhen, enthält den Parameter UKM (user keying material) — einfach zusätzliche Entropie.

Hinzufügen von Kryptografie zum Code

Wir betrachten nur die Änderungen, die am ursprünglichen Code vorgenommen wurden, da der Rahmen unverändert blieb (tatsächlich wurde zuerst die endgültige Implementierung geschrieben und dann die gesamte Kryptografie herausgefiltert).

Da die Authentifizierung und Identifizierung der Kommunikationspartner über öffentliche Schlüssel erfolgt, müssen diese nun irgendwo langfristig gespeichert werden. Zur Vereinfachung verwenden wir JSON in folgender Form:

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

our – unser Schlüsselpaar, hexadezimale private und öffentliche Schlüssel. their – die Namen der Gesprächspartner und ihre öffentlichen Schlüssel. Wir ändern die Argumente der Befehlszeile und fügen eine Nachbearbeitung der JSON-Daten hinzu:

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="Generieren Sie JSON mit unserem neuen Schlüssel-Paar",
)
parser.add_argument(
    "--keys",
    default="keys.json",
    required=False,
    help="JSON mit unseren und ihren Schlüsseln",
)
parser.add_argument(
    "--bind",
    default="::1",
    help="Adresse, auf der gehört werden soll",
)
parser.add_argument(
    "--port",
    type=int,
    default=6666,
    help="Port, auf dem gehört werden soll",
)
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 und unmarshale unsere und ihre Schlüssel {{{
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),
    }
# }}}

Der private Schlüssel des GOST 34.10-Algorithmus ist eine Zufallszahl. Er hat eine Größe von 256 Bit für 256-Bit-Elliptische Kurven. PyGOST arbeitet nicht mit einem Byte-Array, sondern mit großen Zahlen, daher muss unser privater Schlüssel (urandom(32)) in eine Zahl umgewandelt werden, indem wir gost3410.prv_unmarshal() verwenden. Der öffentliche Schlüssel wird deterministisch aus dem privaten Schlüssel berechnet, indem wir gost3410.public_key() nutzen. Der öffentliche Schlüssel 34.10 besteht aus zwei großen Zahlen, die ebenfalls in eine Bytefolge umgewandelt werden müssen, um die Speicherung und Übertragung zu erleichtern, wobei wir gost3410.pub_marshal() verwenden.

Nach dem Einlesen der JSON-Datei müssen die öffentlichen Schlüssel entsprechend zurückgewandelt werden, indem wir gost3410.pub_unmarshal() verwenden. Da die Identifikatoren der Gesprächspartner in Form eines Hashes des öffentlichen Schlüssels ankommen, können diese im Voraus berechnet und in ein Wörterbuch für einen schnellen Zugriff eingefügt werden. Der Streibog-256-Hash ist gost34112012256.GOST34112012256(), der die hashlib-Schnittstelle für Hash-Funktionen vollständig erfüllt.

Wie hat sich die Coroutine des Initiators verändert? Alles erfolgt gemäß dem Handshake-Schema: Wir erzeugen ein Cookie (128-Bit ist völlig ausreichend) und ein ephemeral key pair 34.10, das für die VKO-Funktion zur Schlüsselvereinbarung verwendet wird.

 395 async def initiator(host, port):
 396     _id = repr((host, port))
 397     logging.info("%s: Wahl", _id)
 398     reader, writer = await asyncio.open_connection(host, port)
 399     # Erzeuge unseren flüchtigen öffentlichen Schlüssel und Cookie, sende Handshake 0 Nachricht {{{
 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()

Wir warten auf die Antwort und dekodieren die empfangene Msg-Nachricht;
Wir stellen sicher, dass wir handshake1 erhalten haben;
Wir dekodieren den flüchtigen öffentlichen Schlüssel der Gegenstelle und berechnen den Sitzungsschlüssel;
Wir erzeugen die symmetrischen Schlüssel, die zur Verarbeitung des TBE-Teils der Nachricht erforderlich sind.

 423     logging.info("%s: Nachricht %s empfangen", _id, msg.choice)
 424     if msg.choice != "handshake1":
 425         logging.warning("%s: unerwartete Nachricht, Verbindung wird getrennt", _id)
 426         writer.close()
 427         return
 428     # }}}
 429     msg_handshake1 = msg.value
 430     # Handshake-Nachricht validieren {{{
 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 ist eine 64-Bit-Zahl (urandom(8)), die ebenfalls aus der byte-basierten Darstellung deserialisiert werden muss, unter Verwendung von gost3410_vko.ukm_unmarshal(). Die VKO-Funktion für 34.10-2012 256-Bit ist gost3410_vko.kek_34102012256() (KEK – Schlüsselverschlüsselungsschlüssel).

Der generierte Sitzungsschlüssel ist bereits eine 256-Bit lange pseudorandomisierte Bytefolge. Daher kann er sofort in der HKDF-Funktion verwendet werden. Da GOST34112012256 den hashlib-Schnittstellen entspricht, kann es direkt in der Hkdf-Klasse verwendet werden. Ein Salt (erster Parameter von Hkdf) ist nicht erforderlich, da der generierte Schlüssel aufgrund der Ephemerität der beteiligten Schlüsselpärchen für jede Sitzung unterschiedlich sein wird und bereits genügend Entropie enthält. kdf.expand() gibt standardmäßig bereits die benötigten 256-Bit langen Schlüssel für Kuznechik aus.

Anschließend werden die TBE- und TBS-Teile der eingegangenen Nachricht überprüft:

der MAC über den empfangenen Chiffretext wird berechnet und überprüft;
der Chiffretext wird entschlüsselt;
die TBE-Struktur wird dekodiert;
aus dieser wird der Identifikator des Gesprächspartners entnommen und überprüft, ob er uns bekannt ist;
der MAC über diesen Identifikator wird berechnet und überprüft;
die Signatur über die TBS-Struktur, die die Cookies beider Parteien und den öffentlichen ephemeral Schlüssel der Gegenpartei enthält, wird überprüft. Die Signatur wird mit dem langlebigen Signaturschlüssel des Gesprächspartners überprüft.

 try:
    peer_name = validate_tbe(
        msg_handshake1,
        key_handshake1_mac_identity,
        key_handshake1_enc,
        key_handshake1_mac,
        cookie_our,
        cookie_their,
        pub_their_raw,
    )
except ValueError as err:
    logging.warning("%s: %s, trennen der Verbindung", _id, err)
    writer.close()
    return
# }}}

def validate_tbe(
        msg_handshake: Union[MsgHandshake1, MsgHandshake2],
        key_mac_identity: bytes,
        key_enc: bytes,
        key_mac: bytes,
        cookie_their: Cookie,
        cookie_our: Cookie,
        pub_key_our: PubKey,
) -> str:
    ciphertext = bytes(msg_handshake["ciphertext"])
    mac_tag = mac(GOST3412Kuznechik(key_mac).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext)
    if not compare_digest(mac_tag, bytes(msg_handshake["ciphertextMac"])):
        raise ValueError("ungültige MAC")
    plaintext = ctr(
        GOST3412Kuznechik(key_enc).encrypt,
        KUZNECHIK_BLOCKSIZE,
        ciphertext,
        8 * b"x00",
    )
    try:
        tbe, _ = HandshakeTBE().decode(plaintext)
    except ASN1Error:
        raise ValueError("kann TBE nicht dekodieren")
    key_sign_pub_hash = bytes(tbe["identity"])
    peer = KEYS.get(key_sign_pub_hash)
    if peer is None:
        raise ValueError("unbekannte Identität")
    mac_tag = mac(
        GOST3412Kuznechik(key_mac_identity).encrypt,
        KUZNECHIK_BLOCKSIZE,
        key_sign_pub_hash,
    )
    if not compare_digest(mac_tag, bytes(tbe["identityMac"])):
        raise ValueError("ungültige Identitäts-MAC")
    tbs = HandshakeTBS((
        ("cookieTheir", cookie_their),
        ("cookieOur", cookie_our),
        ("pubKeyOur", pub_key_our),
    ))
    if not gost3410.verify(
        CURVE,
        peer["pub"],
        GOST34112012256(tbs.encode()).digest(),
        bytes(tbe["signature"]),
    ):
        raise ValueError("ungültige Signatur")
    return peer["name"]

Wie bereits oben erwähnt, beschreibt 34.13-2015 verschiedene Betriebsmodi für Blockchiffren aus 34.12-2015. Darunter gibt es den Modus zur Erzeugung von MACs. In PyGOST entspricht dies gost3413.mac(). Dieser Modus erfordert die Übermittlung der Verschlüsselungsfunktion (eine Funktion, die einen Block von Daten annimmt und zurückgibt), der Blockgröße sowie der eigentlichen Daten. Warum kann die Blockgröße nicht hardcodiert werden? 34.12-2015 beschreibt nicht nur den 128-Bit-Chiffre Kuznechik, sondern auch die 64-Bit-Version Magmа — eine leicht modifizierte Version des GOST 28147-89, der noch zu Zeiten des KGB entwickelt wurde und bis heute einen der höchsten Sicherheitsstandards aufweist.

Kuznechik wird durch den Aufruf gost.3412.GOST3412Kuznechik(key) initialisiert und gibt ein Objekt mit den Methoden .encrypt()/.decrypt() zurück, das für die Übergabe an die Funktion 34.13 geeignet ist. Der MAC wird wie folgt berechnet: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Ein gewöhnlicher Vergleich (==) von Byte-Strings kann nicht für den abgeleiteten und den empfangenen MAC verwendet werden, da dieser Betrieb Zeitvergleiche offenbart, was im Allgemeinen zu fatalen Sicherheitsanfälligkeiten führen kann, wie BEAST z.B. Angriffe auf TLS. In Python gibt es dafür eine spezielle Funktion, hmac.compare_digest.

Die Blockverschlüsselungsfunktion kann nur einen Datenblock verschlüsseln. Für eine größere Anzahl, insbesondere bei nicht ganzzahliger Länge, muss ein Verschlüsselungsmodus verwendet werden. In der Norm 34.13-2015 sind folgende Modi beschrieben: ECB, CTR, OFB, CBC, CFB. Jeder hat seine eigenen zulässigen Anwendungsbereiche und Eigenschaften. Leider gibt es bisher noch keine standardisierten authentifizierten Verschlüsselungsmodi wie CCM, OCB, GCM und ähnliche – wir sind gezwungen, zumindest einen MAC selbst hinzuzufügen. Ich wähle den Zählermodus (CTR): Er erfordert keine Auffüllung auf Blockgröße, kann parallelisiert werden, nutzt nur die Verschlüsselungsfunktion und kann sicher für die Verschlüsselung einer großen Anzahl von Nachrichten verwendet werden (im Gegensatz zu CBC, bei dem relativ schnell Kollisionen auftreten).

Wie .mac(), so akzeptiert auch .ctr() ähnliche Eingabedaten: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). Es ist erforderlich, einen Initialisierungsvektor festzulegen, der genau die Hälfte der Blockgröße des Algorithmus beträgt. Wenn unser Verschlüsselungsschlüssel nur zum Verschlüsseln einer einzigen Nachricht (auch wenn sie aus mehreren Blöcken besteht) verwendet wird, kann ein Nullinitialisierungsvektor sicher festgelegt werden. Für die Verschlüsselung der Handshake-Nachrichten verwenden wir jedes Mal einen anderen Schlüssel.

Die Überprüfung der Signatur mit gost3410.verify() ist trivial: Wir geben die elliptische Kurve an, innerhalb derer wir arbeiten (diese wird einfach in unserem GOSTIM-Protokoll fixiert), den öffentlichen Schlüssel des Unterzeichners (denken Sie daran, dass dies ein Tupel aus zwei großen Zahlen und keine Byte-Zeichenkette sein muss), den 34.11-2012 Hash und die ankommende Signatur selbst.

Anschließend bereiten wir im Initiator die Handshake2-Nachricht vor und senden sie, indem wir die gleichen Schritte wie bei der Überprüfung durchführen, nur symmetrisch: Signatur auf unseren Schlüsseln anstelle der Überprüfung, und so weiter…

 456     # Bereiten Sie die Handshake 2 Nachricht vor und senden Sie sie {{{
 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: Sitzung etabliert: %s", _id, peer_name)

Sobald die Sitzung hergestellt ist, werden die Transport-Schlüssel generiert (ein separater Schlüssel für Verschlüsselung, einer für Authentifizierung, für jede der Parteien), und der Kuznechik wird zur Entschlüsselung und Überprüfung des MAC initialisiert:

 499     # Führen Sie den Textnachrichtensender aus, initialisieren Sie den Transportdekoder {{{
 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     # Warten auf Testnachrichten {{{
 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     # }}}

Die msg_sender-Koroutine verschlüsselt jetzt Nachrichten, bevor sie über die TCP-Verbindung gesendet werden. Jede Nachricht hat einen monoton steigenden nonce, der auch als Initialisierungsvektor beim Verschlüsseln im Zählermodus dient. Jede Nachricht und jeder Nachrichtenblock wird garantiert unterschiedliche Zählerwerte haben.

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

Eingehende Nachrichten werden von der Coroutine msg_receiver verarbeitet, die sich mit Authentifizierung und Dekodierung befasst:

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("unerwarteter Nonce-Wert")
    mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode())
    if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])):
        raise ValueError("ungültiger 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)

Fazit

GOSTIM soll ausschließlich zu Ausbildungszwecken verwendet werden (da es mindestens nicht getestet ist)! Der Quellcode der Anwendung kann heruntergeladen werden. hier (Stribog-256 Hash: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Wie alle meine Projekte, ist GoGOST, PyDERASN, NNCP, GoVPNGOSTIM vollkommen freie Software, die unter den Bedingungen von GPLv3+.

Sergey Matveev, Kryptopunk, Mitglied des Open-Source-Fonds, Python/Go-Entwickler, Hauptspezialist FSUE "NTC 'Atlas'"..

Quelle: habr.com