GOSTIM: P2P F2F E2EE IM an einem Abend mit GOST-Kryptographie

Als Entwickler PyGOST einer Bibliothek (GOST-verschlüsselnde primitive auf reinem Python) bekomme ich häufig Fragen dazu, wie man auf die Schnelle einen einfachen sicheren Nachrichtenaustausch implementieren kann. Viele halten angewandte Kryptografie für eine relativ einfache Angelegenheit, und der Aufruf .encrypt() bei einem Blockcipher ist ausreichend für den sicheren Versand über einen Kommunikationskanal. Andere glauben, dass angewandte Kryptografie nur wenigen vorbehalten ist und es akzeptabel ist, dass reiche Unternehmen wie Telegram mit Mathematikern aus der Olympiade kein sicheres Protokoll implementieren können.

All dies hat mich dazu gebracht, diesen Artikel zu schreiben, um zu zeigen, dass die Implementierung von kryptografischen Protokollen und sicherem IM keine allzu komplexe Aufgabe ist. Allerdings ist es nicht ratsam, eigene Authentifizierungs- und Schlüsselvereinbarungsprotokolle zu erfinden.

In diesem Artikel wird peer-to-peer, friend-to-friend, end-to-end verschlüsselter Instant Messenger mit SIGMA-I Authentifizierungs- und Schlüsselvereinbarungsprotokoll (auf dessen Grundlage realisiert wurde IPsec IKE), und verwendet ausschließlich GOST-kryptografische Algorithmen aus der PyGOST-Bibliothek und ASN.1 Kodierung von Nachrichten mit der Bibliothek PyDERASN (über die ich bereits früher geschrieben habe). Eine Voraussetzung: Es muss so einfach sein, dass man es von Grund auf an einem Abend (oder Arbeitstag) schreiben kann, sonst ist es schon kein einfaches Programm mehr. Es sind mit Sicherheit Fehler, unnötige Komplexitäten und Mängel vorhanden, zudem ist es mein erstes Programm mit der asyncio-Bibliothek.

Design des IM

Um zu beginnen, müssen wir verstehen, wie unser IM aussehen wird. Zur Vereinfachung nehmen wir an, dass es sich um ein Peer-to-Peer-Netzwerk handelt, ohne irgendeine Teilnehmendenentdeckung. Wir geben manuell an, zu welcher Adresse und zu welchem Port wir für die Kommunikation mit dem Gesprächspartner verbinden wollen.

Ich verstehe, dass die Annahme von derzeitiger Verfügbarkeit einer direkten Verbindung zwischen zwei beliebigen Computern eine wesentliche Einschränkung für die praktische Anwendbarkeit von IM darstellt. Aber je mehr Entwickler verschiedene NAT-Traversal-Lösungen implementieren, desto länger werden wir im IPv4-Internet feststecken, mit ernüchternder Wahrscheinlichkeit einer Verbindung 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 oder findet nicht den Namen/Schlüssel, trennt die Verbindung oder fährt fort, im Wissen um den Gesprächspartner. Zweitens ist es im Allgemeinen sicher und schließt zahlreiche Angriffe aus.

Die IM-Oberfläche wird klassischen Lösungen ähnlich sein suckless-Projekten, die mir wegen ihres Minimalismus und ihrer 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 vom Gesprächspartner empfangenen Nachrichten gelesen;
• state — indem wir daraus lesen, erfahren wir, ob der Gesprächspartner gerade verbunden ist, Adresse/Port der Verbindung.

Darüber hinaus wird ein conn-Socket erstellt; indem wir den Host-Port dort hineinschreiben, initiieren wir die Verbindung zu einem 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 nicht streiten, es ist unmöglich, es jedem recht zu machen. Mit tmux und/oder multitail, kann man eine mehrfenstrige Oberfläche mit Syntaxhervorhebung erhalten. Und mithilfe von rlwrap kann man eine GNU Readline-kompatible Eingabezeile für Nachrichten erhalten.

Tatsächlich verwenden suckless-Projekte FIFO-Dateien. Persönlich konnte ich nicht verstehen, wie man mit asyncio konkurrierend mit Dateien arbeitet, ohne selbsthandwerkliche Unterlagen in Form von dedizierten Threads (für solche Dinge benutze ich seit langem die Sprache Go). Daher entschied ich mich, mit Unix-Domain-Sockets auszukommen. Leider beraubt das die Möglichkeit, echo 2001:470:dead::babe 6666 > conn zu machen. 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

TCP wird als Transport verwendet: es garantiert die Lieferung und deren Reihenfolge. UDP garantiert weder das eine noch das andere (was hilfreich wäre, wenn Kryptographie zum Einsatz käme), und Unterstützung für SCTP gibt es in Python nicht direkt.

Leider gibt es im TCP kein Konzept für Nachrichten, sondern nur für einen Byte-Stream. Daher ist es notwendig, ein Format für Nachrichten zu entwickeln, damit sie in diesem Stream voneinander getrennt werden können. Wir können uns darauf einigen, das Zeilenumbruchzeichen zu verwenden. Das ist für den Anfang in Ordnung, jedoch, wenn wir anfangen, unsere Nachrichten zu verschlüsseln, könnte dieses Zeichen irgendwo im Chiffretext auftauchen. In Netzwerken sind daher Protokolle beliebt, die zuerst die Länge der Nachricht in Bytes senden. Zum Beispiel gibt es in Python out-of-the-box die xdrlib, die es ermöglicht, mit einem solchen Format zu arbeiten. XDR.

Wir werden nicht korrekt und effizient mit dem Lesen von TCP arbeiten — wir vereinfachen den Code. Wir lesen in einer Endlosschleife die Daten aus dem Socket, bis wir die vollständige Nachricht dekodiert haben. Als Format für diesen Ansatz kann auch JSON mit XML verwendet werden. Aber wenn Kryptografie hinzukommt, müssen die Daten signiert und authentifiziert werden — und das erfordert eine byte-genaue identische Darstellung von Objekten, was JSON/XML nicht gewährleistet (das Ergebnis der Dumps kann variieren).

XDR eignet sich für diese Aufgabe, jedoch wähle ich ASN.1 mit DER-Codierung und PyDERASN eine Bibliothek, da wir es mit hochgradigen Objekten zu tun haben, mit denen oft angenehmer und bequemer zu arbeiten ist. Im Gegensatz zu schemalo bencode, MessagePack oder CBOR, wird ASN.1 die Daten automatisch gegen ein fest definiertes Schema ü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 empfohlene Nachricht wird Msg sein: entweder eine Textnachricht MsgText (bis jetzt mit einem einzigen Textfeld) oder eine Handshake-Nachricht MsgHandshake (in der der Name des Gesprächspartners übermittelt wird). Im Moment sieht es überkompliziert aus, aber das ist eine Grundlage für die Zukunft.

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

IM ohne Kryptografie

Wie ich bereits erwähnt habe, wird für alle Operationen mit Sockets die asyncio-Bibliothek verwendet. Lassen Sie uns festlegen, was wir zum Zeitpunkt des Starts 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 zum Lauschen",
)
parser.add_argument(
    "--port",
    type=int,
    default=6666,
    help="Port, auf dem gehört werden soll",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))

Es wird ein eigener Name festgelegt (—our-name alice). Alle erwarteten Gesprächspartner werden durch Kommas aufgelistet (—their-names bob,eve). Für jeden der Gesprächspartner wird ein Verzeichnis mit Unix-Sockets sowie eine Coroutine für jeden in, out, state erstellt:

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 werden in die IN_QUEUES-Warteschlangen vom in-Socket 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"))

Eingehende Nachrichten von Gesprächspartnern 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 aus dem state-Socket sucht das Programm im PEER_ALIVE-Wörterbuch die Adresse des Gesprächspartners. Wenn keine Verbindung zum Gesprächspartner besteht, wird eine leere Zeile 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 Funktion des "Initiators" der 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. Zunächst ö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 eine Antwort von der Remote-Seite. Es versucht, die eingegangene 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 dem Aufruf .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, trennen", _id)
 145         writer.close()
 146         return
 147     try:
 148         msg, _ = Msg().decode(data)
 149     except ASN1Error:
 150         logging.warning("%s: unverarbeitbare Antwort, trennen", _id)
 151         writer.close()
 152         return
 153     logging.info("%s: erhaltene %s Nachricht", _id, msg.choice)
 154     if msg.choice != "handshake":
 155         logging.warning("%s: unerwartete Nachricht, trennen", _id)
 156         writer.close()
 157         return
 158     # }}}

Wir überprüfen, ob der empfangene Name des Gesprächspartners uns bekannt ist. Wenn nicht, trennen wir die Verbindung. Wir prüfen, ob bereits eine Verbindung zu ihm hergestellt war (der Gesprächspartner hat erneut den Befehl zur Verbindung zu uns gegeben) und schließen diese. In IN_QUEUES werden Python-Strings mit dem Text der Nachricht abgelegt, jedoch gibt es einen speziellen Wert None, der dem msg_sender-Koroutine signalisiert, die Arbeit zu beenden, damit sie ihren writer, der mit der veralteten TCP-Verbindung verbunden ist, vergisst.

 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     # Starte Textnachrichtensender, initiiere Transportdekoder {{{
 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-Queue gelegt werden), serialisiert sie in die MsgText-Nachricht und sendet sie über die TCP-Verbindung. Diese kann jederzeit unterbrochen werden – das fangen wir explizit ab.

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: gesendete Nachricht mit %d Zeichen", peer_name, len(text))

Am Ende tritt der Initiator in einen endlosen Zyklus des Lesens von Nachrichten aus dem Socket ein. Er überprüft, ob es sich um Textnachrichten handelt und fügt sie in die OUT_QUEUES-Warteschlange ein, aus der sie an den entsprechenden Gegenüber-Socket gesendet werden. Warum kann man nicht einfach .read() verwenden und die Nachricht dekodieren? Weil nicht ausgeschlossen werden kann, dass mehrere Nachrichten vom Benutzer im Puffer des Betriebssystems aggregiert und in einem TCP-Segment gesendet werden. Wir können die erste dekodieren, aber im Puffer kann noch ein Teil der nachfolgenden Nachrichten bleiben. Bei jeder außergewöhnlichen 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 sich: %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: empfangen %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 zu Beginn des Programms 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ört auf: %s", server.sockets[0].getsockname())
loop.run_forever()

Der Responder ähnelt dem Initiator und führt alle gleichen Aktionen spiegelbildlich aus, aber die endlose Schleife zum Lesen von Nachrichten wird sofort gestartet, um es einfacher zu machen. Momentan sendet das Handshake-Protokoll jeweils eine Nachricht von jeder Seite, wird aber in Zukunft zwei vom Initiator der Verbindung senden, nach denen sofort das Senden von Textnachrichten möglich ist.

  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         # Lies, bis wir eine 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 Puffergröß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         # Verarbeite die Handshake-Nachricht {{{
 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 Zeit, unsere Kommunikation abzusichern. Was verstehen wir unter Sicherheit und was wollen wir:

• Vertraulichkeit der übermittelten Nachrichten;
• Authentizität und Integrität der übermittelten Nachrichten – Veränderungen sollten festgestellt werden;
• Schutz vor Wiederholungsangriffen – das Fehlen oder die Wiederholung von Nachrichten sollte erkannt werden (und wir entscheiden uns, die Verbindung zu trennen);
• Identifikation und Authentifizierung der Gesprächspartner anhand vordefinierter öffentlicher Schlüssel – wir haben bereits entschieden, dass wir ein friend-to-friend Netzwerk aufbauen. Nur nach der Authentifizierung verstehen wir, mit wem wir kommunizieren;
• die Existenz Perfect Forward Secrecy Eigenschaften (PFS) – die Kompromittierung unseres langlebigen Signaturschlüssels sollte nicht zur Möglichkeit führen, die gesamte vorherige Korrespondenz zu lesen. Aufgezeichnete abgefangene Daten werden wertlos;
• Die Gültigkeit der Nachrichten (Transport- und Handshake-Nachrichten) nur innerhalb einer TCP-Sitzung. Das Einfügen korrekt signierter/authentifizierter Nachrichten aus einer anderen Sitzung (selbst mit demselben Gesprächspartner) sollte nicht möglich sein;
• Ein passiver Beobachter darf weder die Benutzeridentifikatoren noch die übermittelten langlebigen öffentlichen Schlüssel oder deren Hashes sehen. Eine gewisse Anonymität gegenüber einem passiven Beobachter.

Überraschenderweise möchten fast alle dieses Minimum in jedem Handshake-Protokoll haben, und nur sehr wenig davon wird letztendlich für 'hausgemachte' Protokolle umgesetzt. Lassen Sie uns daher auch jetzt nichts Neues erfinden. Ich würde eindeutig empfehlen, das Noise-Framework für den Aufbau von Protokollen zu verwenden, aber etwas Einfacheres wählen.

Die beiden beliebtesten Protokolle sind:

• TLS — ein äußerst komplexes Protokoll mit einer langen Geschichte von Bugs, Fehlern, Schwachstellen, mangelnder Überlegung, Komplexität und Mängeln (wenngleich dies wenig mit TLS 1.3 zu tun hat). Wir betrachten es jedoch wegen seiner Überkomplexität nicht.
• IPsec c IKE — weisen keine ernsthaften kryptografischen Probleme auf, sind jedoch 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 recht einfach in einer Nacht implementiert werden können.

Was macht SIGMA, als letzte Entwicklungsstufe der STS/ISO-Protokolle, gut? Es erfüllt all unsere Anforderungen (einschließlich des 'Versteckens' der Identifikatoren der Gesprächspartner) und hat keine bekannten kryptografischen Probleme. Es ist minimalistischer – das Entfernen auch nur eines Elements aus der Nachrichtenprotokoll führt zu seiner Unsicherheit.

Lassen Sie uns vom einfachsten hausgemachten Protokoll bis hin zu SIGMA gehen. Die grundlegendste Operation, die uns interessiert, ist die Schlüsselaushandlung: eine Funktion, bei der beide Teilnehmer dasselbe Ergebnis erhalten, das als symmetrischer Schlüssel verwendet werden kann. Ohne ins Detail zu gehen: Jede der Parteien generiert ein ephemeres (nur innerhalb einer Sitzung verwendetes) Schlüsselpaar (öffentliche und private Schlüssel), tauscht die öffentlichen Schlüssel aus und ruft die Aushandlungsfunktion auf, der sie ihren privaten Schlüssel und den öffentlichen Schlüssel des Gesprächspartners übergeben.

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

Jeder kann dazwischenfunken und die öffentlichen Schlüssel durch eigene ersetzen – in diesem Protokoll gibt es keine Authentifizierung der Gesprächspartner. Wir fügen eine Signatur mit langlebigen Schlüsseln hinzu.

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

Eine solche Signatur eignet sich nicht, da sie nicht an eine bestimmte Sitzung gebunden ist. Solche Nachrichten „passen“ auch für Sitzungen mit anderen Teilnehmern. Der gesamte Kontext muss signiert werden. Das zwingt uns auch dazu, eine weitere Nachricht von A zu senden.

Darüber hinaus ist es entscheidend, die eigene Identifikationsnummer unter die Signatur zu setzen, da wir sonst IdXXX ersetzen und die Nachricht mit dem Schlüssel eines anderen bekannten Gesprächspartners neu signieren können. Zur Verhinderung von Reflexion-Angriffen, es ist notwendig, dass die Elemente unter der Beschriftung an klar definierten Orten entsprechend ihrem Sinn platziert sind: wenn A (PubA, PubB) signiert, dann muss B (PubB, PubA) signieren. Das weist auch auf die Wichtigkeit der Auswahl der Struktur und des Formats der serialisierten Daten hin. Zum Beispiel werden Mengen im ASN.1 DER-Codierung 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 für diese Sitzung denselben gemeinsamen Schlüssel entwickelt haben. Im Grunde genommen könnte man diesen Schritt auch auslassen — die erste Transportnachricht wäre ungültig, aber wir möchten sicher sein, dass nach Abschluss des Handshakes alles tatsächlich abgestimmt ist. Derzeit haben wir das ISO/IEC IS 9798-3 Protokoll zur Verfügung.

Wir könnten auch den generierten Schlüssel selbst signieren. Das ist riskant, da es nicht ausgeschlossen ist, dass es im verwendeten Signaturalgorithmus Lecks geben kann (auch wenn es sich um Bits zur Signatur handelt, sind es dennoch Lecks). Man könnte den Hash des generierten Schlüssels signieren, aber selbst ein Leak des Hashes des generierten Schlüssels könnte bei Brute-Force-Angriffen auf die Schlüsselgenerierungsfunktion wertvoll sein. SIGMA verwendet eine MAC-Funktion, die den Absender-Identifikator authentifiziert.

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

Als Optimierung möchten einige vielleicht ihre ephemeral Schlüssel wiederverwenden (was natürlich für PFS nachteilig ist). Zum Beispiel haben wir ein Schlüsselpaar generiert, versucht uns zu verbinden, aber TCP war nicht verfügbar oder brach irgendwo mitten im Protokoll ab. Es wäre schade, die aufgebrachte Entropie und die Prozessorressourcen für ein neues Paar zu verschwenden. Daher führen wir sozusagen ein Cookie ein — einen pseudorandomisierten Wert, der vor möglichen Replay-Angriffen beim Wiederverwenden ephemeral öffentlicher Schlüssel schützt. Aufgrund der Bindung zwischen dem Cookie und dem ephemeral öffentlichen Schlüssel kann der öffentliche Schlüssel des anderen Teilnehmers 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, ...)║
   │                                                                       │ ╚═════════════════════╝
   │                                                                       │

Letztendlich wollen wir die Privatsphäre unserer Identifikatoren gegenüber einem passiven Beobachter wahren. Zu diesem Zweck schlägt SIGMA vor, zunächst ephemeral Schlüssel auszutauschen, um einen gemeinsamen Schlüssel zu erarbeiten, mit dem authentifizierende und identifizierende Nachrichten verschlüsselt werden. SIGMA beschreibt zwei Varianten:

• SIGMA-I — schützt den Initiator vor aktiven Angriffen und den Antwortenden vor passiven: Der Initiator authentifiziert den Antwortenden und gibt seine Identifizierung nicht preis, wenn etwas nicht stimmt. Der Antwortende hingegen gibt seine Identifizierung preis, wenn er in ein aktives Protokoll eingebunden wird. Ein passiver Beobachter erfährt nichts;
  SIGMA-R — schützt den Antwortenden vor aktiven Angriffen und den Initiator vor passiven. Alles ist genau umgekehrt, aber in diesem Protokoll werden bereits vier Nachrichten im Handshake übertragen.

  Wir wählen SIGMA-I, da es mehr mit dem übereinstimmt, was wir von gewohnten Client-Server-Szenarien erwarten: Nur der authentifizierte Server erkennt den Client, während der Server bereits alles weiß. Außerdem ist es aufgrund der geringeren Anzahl an Handshake-Nachrichten einfacher in der Implementierung. Alles, was wir in das Protokoll einfügen, ist die Verschlüsselung eines Teils der Nachricht und der Transfer 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, ...)║
     │                                                                              │ ╚═════════════════════╝
     │                                                                              │
  

  • Für die Signatur wird GOST R verwendet. 34.10-2012 Der Algorithmus verwendet 256-Bit-Schlüssel.
  • Für die Ableitung des gemeinsamen Schlüssels wird 34.10-2012 VKO verwendet.
  • Als MAC wird CMAC verwendet. Technisch handelt es sich um einen speziellen Betriebsmodus eines Blockchiffriers, der in GOST R 34.13-2015 beschrieben ist. Für diesen Modus wird — Grashüpfer (34.12-2015).
  • Als Identifikator des Gesprächspartners wird der Hash seines öffentlichen Schlüssels verwendet. Für den Hash wird Streebog-256 (34.11-2012 256 Bit) verwendet.

  Nach dem Handshake wird unser gemeinsamer Schlüssel vereinbart. Diesen können wir für die authentifizierte Verschlüsselung von Transportnachrichten verwenden. Dieser Teil ist ganz einfach und dabei kaum fehleranfällig: Wir inkrementieren den Nachrichtenzähler, verschlüsseln die Nachricht, authentifizieren (MAC) den Zähler und den Chiffretext und senden dann. Beim 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 wir die Handshake-, Transportnachrichten verschlüsseln und authentifizieren? Einen Schlüssel für all diese Aufgaben zu verwenden, ist riskant und unklug. Es ist notwendig, Schlüssel zu erzeugen, indem spezialisierte Funktionen verwendet werden KDF (Schlüsselerzeugungsfunktion). Noch einmal, wir werden es uns nicht schwer machen und etwas erfinden: HKDF ist seit langem bekannt, gut erforscht und hat keine bekannten Probleme. Leider gibt es in der Standardbibliothek von Python keine solche Funktion, daher verwenden wir hkdf Paket. HKDF verwendet dabei Dieses System, das seinerseits eine Hash-Funktion verwendet. Ein Beispiel für eine Implementierung in Python auf der Wikipedia-Seite umfasst nur wenige Zeilen Code. Wie bei 34.10-2012 werden wir als Hash-Funktion Streebog-256 verwenden. Der Ausgang unserer Schlüsselvereinbarungsfunktion wird als Sitzungsschlüssel bezeichnet, aus dem die fehlenden symmetrischen Schlüssel erzeugt 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/Schemata

  Betrachten wir, welche ASN.1-Strukturen wir jetzt 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 unterschrieben 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, zur weiteren Randomisierung der generierten Schlüssel, enthält den Parameter UKM (user keying material) — einfach zusätzliche Entropie.

  Hinzufügen von Kryptographie zum Code

  Betrachten wir nur die Änderungen, die am ursprünglichen Code vorgenommen wurden, da das Gerüst unverändert geblieben ist (tatsächlich wurde zuerst die endgültige Implementierung geschrieben und dann die gesamte Kryptographie daraus entfernt).

  Da die Authentifizierung und Identifizierung der Gesprächspartner über öffentliche Schlüssel durchgeführt wird, müssen diese nun langfristig irgendwo 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 deren öffentliche Schlüssel. Ändern wir 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="JSON mit unserem neuen Schlüsselpaar generieren",
  )
  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, um zuzuhören",
  )
  parser.add_argument(
      "--port",
      type=int,
      default=6666,
      help="Port, um zuzuhören",
  )
  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)
  
  # Schlüsseldaten parsen und unmarshallen {{{
  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 Algorithmus 34.10 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() verwenden. Der öffentliche Schlüssel 34.10 besteht aus zwei großen Zahlen, die ebenfalls zur Vereinfachung der Speicherung und Übertragung in eine Bytefolge umgewandelt werden müssen, unter Verwendung von gost3410.pub_marshal().

  Nach dem Lesen der JSON-Datei müssen die öffentlichen Schlüssel entsprechend mit gost3410.pub_unmarshal() zurückgewandelt werden. Da wir die Identifikatoren der Gesprächspartner in Form eines Hashs des öffentlichen Schlüssels erhalten, können diese sofort vorausberechnet und in ein Dictionary für eine schnelle Suche eingefügt werden. Der Streibog-256-Hash ist gost34112012256.GOST34112012256(), der den Anforderungen des hashlib-Schnittstelle für Hashfunktionen vollständig entspricht.

  Wie hat sich die Coroutine des Initiators geändert? Alles läuft wie im Handshake-Schema: Wir generieren ein Cookie (128 Bit sind vollkommen ausreichend), 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: wählt", _id)
   398     reader, writer = await asyncio.open_connection(host, port)
   399     # Generiere unseren temporären ö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 temporären öffentlichen Schlüssel der Gegenpartei 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: erhielt %s Nachricht", _id, msg.choice)
   424     if msg.choice != "handshake1":
   425         logging.warning("%s: unerwartete Nachricht, trennen", _id)
   426         writer.close()
   427         return
   428     # }}}
   429     msg_handshake1 = msg.value
   430     # Validieren der Handshake-Nachricht {{{
   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 ihrer byte-repräsentation 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 erzeugte Sitzungsschlüssel ist bereits eine 256-Bit-byte pseudorandom sequenz. Daher kann er sofort in der HKDF-Funktion verwendet werden. Da GOST34112012256 das hashlib-Interface erfüllt, kann er sofort in der Hkdf-Klasse verwendet werden. Das Salt (erster Argument von Hkdf) geben wir nicht an, da der erzeugte Schlüssel aufgrund der Temporärheit der beteiligten Schlüsselpaare 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-Schlüssel für den Kuznechik aus.

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

  • Es wird ein MAC über den empfangenen Chiffretext berechnet und überprüft;
  • Der Chiffretext wird entschlüsselt;
  • Die TBE-Struktur wird dekodiert;
  • Daraus wird die Identifikation des Gesprächspartners entnommen und überprüft, ob wir ihn überhaupt kennen;
  • Es wird ein MAC über diese Identifikation berechnet und überprüft;
  • Die Signatur über der TBS-Struktur wird überprüft, in die die Cookies beider Parteien und der öffentliche temporäre Schlüssel der Gegenseite eingehen. Die Signatur wird mit dem langlebigen Schlüssel des Gesprächspartners überprüft.

   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("ungültiges 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("kann TBE nicht decodieren")
   151     key_sign_pub_hash = bytes(tbe["identity"])
   152     peer = KEYS.get(key_sign_pub_hash)
   153     if peer is None:
   154         raise ValueError("unbekannte Identität")
   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("ungültiges Identitäts-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("ungültige Signatur")
   174     return peer["name"]
  

  Wie bereits oben erwähnt, beschreibt 34.13-2015 verschiedene Betriebsmodi von Blockchiffren aus 34.12-2015. Darunter gibt es den Modus zur Erzeugung von MACs. In PyGOST wird dies durch gost3413.mac() realisiert. Dieser Modus erfordert die Übergabe einer Verschlüsselungsfunktion (die einen Block Daten entgegennimmt und zurückgibt), die Größe des Geheimblocks und schließlich der Daten selbst. Warum kann die Größe des Geheimblocks nicht hardcodiert werden? 34.12-2015 beschreibt nicht nur die 128-Bit-Chiffre Kuznechik, sondern auch die 64-Bit- Magmа — ein leicht modifiziertes GOST 28147-89, das noch zu Zeiten des KGB geschaffen wurde und immer noch einen der höchsten Sicherheitsstandards hat.

  Der Katalog wird mit gost.3412.GOST3412Kuznechik(key) initialisiert und gibt ein Objekt mit den Methoden .encrypt() und .decrypt() zurück, die für die Übergabe an die 34.13-Funktion geeignet sind. Der MAC wird wie folgt berechnet: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Zum Vergleich des berechneten und empfangenen MACs dürfen übliche Vergleiche (==) von Byte-Strings nicht verwendet werden, da diese Operation Zeitvergleiche leckt, was im Allgemeinen zu schwerwiegenden Sicherheitsanfälligkeiten führen kann. BEAST Angriffe auf TLS. In Python gibt es eine spezielle Funktion hmac.compare_digest dafür.

  Die Blockverschlüsselungsfunktion kann nur einen Datenblock verschlüsseln. Für größere Mengen, insbesondere solche, die keine Vielfachen der Blockgröße sind, muss der Verschlüsselungsmodus verwendet werden. In 34.13-2015 sind die folgenden beschrieben: ECB, CTR, OFB, CBC, CFB. Jeder hat seine eigenen zulässigen Anwendungsbereiche und Eigenschaften. Leider verfügen wir nach wie vor nicht über standardisierte authentifizierte Verschlüsselungsmodi (wie CCM, OCB, GCM und ähnliche) – wir sind gezwungen, zumindest MAC selbst hinzuzufügen. Ich wähle den Zählermodus (CTR): Er erfordert keine Auffüllung auf die Blockgröße, kann parallelisiert werden, verwendet nur die Verschlüsselungsfunktion und kann sicher zum Verschlüsseln großer Nachrichtenmengen verwendet werden (im Gegensatz zu CBC, bei dem relativ schnell Kollisionen auftreten).

  Wie .mac() nimmt .ctr() ähnliche Daten als Eingabe: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). Ein Initialisierungsvektor muss angegeben werden, der genau die Hälfte der Blockgröße umfasst. Wenn unser Verschlüsselungsschlüssel nur für die Verschlüsselung einer Nachricht verwendet wird (auch wenn diese aus mehreren Blöcken besteht), kann ein Nullvektor für die Initialisierung sicher angegeben werden. Für die Verschlüsselung von Handshake-Nachrichten verwenden wir jedes Mal einen separaten Schlüssel.

  Die Signaturüberprüfung gost3410.verify() ist trivial: Wir übergeben die elliptische Kurve, innerhalb derer wir arbeiten (diese fixieren wir einfach in unserem GOSTIM-Protokoll), den öffentlichen Schlüssel des Unterzeichners (wir dürfen nicht vergessen, dass dies ein Tupel aus zwei großen Zahlen sein muss, nicht ein Byte-String), den 34.11-2012 Hash und die tatsächlich empfangene Signatur.

  Anschließend bereiten wir im Initiator die handshake2-Nachricht vor und senden sie ab, wobei wir die gleichen Schritte wie bei der Überprüfung durchführen, jedoch symmetrisch: eine Signatur auf unseren Schlüsseln anstelle der Überprüfung usw.

   456     # Bereite die Handshake-2-Nachricht vor und sende 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 hergestellt: %s", _id, peer_name)
   

  Wenn die Sitzung hergestellt ist, werden die Transportkey (ein separater Schlüssel für die Verschlüsselung, für die Authentifizierung, für jede der Parteien) generiert und Kuznechik für die Entschlüsselung und MAC-Prüfung initialisiert:

   499     # Starte den Textnachrichtensender, initialisiere den Transportdecoder {{{
   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     # Warte 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 nun Nachrichten, bevor sie über die TCP-Verbindung gesendet werden. Jede Nachricht hat einen monoton wachsenden nonce, der auch als Initialisierungsvektor beim Verschlüsseln im Zählermodus dient. Für jede Nachricht und Nachrichtenblock sind unterschiedliche Zählerwerte garantiert.

  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 der Authentifizierung und Dechiffrierung beschäftigt:

  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 ist ausschließlich zu Bildungszwecken gedacht (da es zumindest nicht getestet ist)! Der Quellcode des Programms kann heruntergeladen werden hier (Stribog-256 Hash: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Wie alle meine Projekte, ist GoGOST, PyDERASN, NNCP, GoVPN, GOSTIM ist vollständig freie Software, die unter den Bedingungen von GPLv3+.

  Sergej Matwejew, Cryptopunk, Mitglied der Open-Source-Stiftung, Python/Go-Entwickler, Hauptspezialist FGUP „NTZ Atlas“.

Quelle: habr.com