GOSTIM : P2P F2F E2EE IM en une soirée avec la cryptographie GOST

En tant que développeur PyGOST de la bibliothèque (primitives cryptographiques conformes aux normes GOST en Python pur), je reçois souvent des questions sur la façon de mettre en œuvre un échange de messages sécurisé de manière rudimentaire. Beaucoup pensent que la cryptographie appliquée est une chose assez simple, et qu'un appel à .encrypt() d'un chiffre par bloc suffit pour un envoi sécurisé par un canal de communication. D'autres estiment que la cryptographie appliquée n'est la prérogative que de quelques-uns, et qu'il est acceptable que des entreprises riches comme Telegram, avec des mathématiciens olympiques, ne puissent pas implémenter un protocole sécurisé.

Tout cela m'a poussé à écrire cet article pour montrer que la mise en œuvre de protocoles cryptographiques et d'un IM sécurisé n'est pas une tâche si complexe. Cependant, il n'est pas recommandé d'inventer ses propres protocoles d'authentification et d'échange de clés.

L'article décrira un messager instantané, ami à ami, chiffré de bout en bout avec le protocole SIGMA-I d'authentification et d'échange de clés (sur la base duquel est réalisé IPsec IKE ), en utilisant exclusivement des algorithmes cryptographiques GOST de la bibliothèque PyGOST et le codage ASN.1 des messages avec la bibliothèquePyDERASN (que j'ai déjà mentionné précédemment ). La condition nécessaire : il doit être suffisamment simple pour pouvoir être écrit de A à Z en une seule soirée (ou journée de travail), sinon ce n'est plus un programme simple. Il y aura certainement des erreurs, des complexités inutiles, des défauts, de plus, c'est mon premier programme utilisant la bibliothèque asyncio.Design de l'IM

Pour commencer, il faut comprendre à quoi ressemblera notre IM. Pour simplifier, supposons qu'il s'agisse d'un réseau peer-to-peer, sans aucune détection des participants. Nous indiquerons manuellement à quel adresse : port nous connecter pour discuter avec notre correspondant.

Je comprends qu'à l'heure actuelle, l'hypothèse de la disponibilité d'une connexion directe entre deux ordinateurs arbitraires est une limitation majeure de l'applicabilité de l'IM en pratique. Mais plus de développeurs mettront en œuvre divers dispositifs de contournement de NAT, plus nous resterons longtemps sur Internet en IPv4, avec une probabilité déprimante de connexion entre des ordinateurs arbitraires. Combien de temps encore pourrons-nous supporter l'absence d'IPv6 à la maison et au travail ?

Je comprends qu'à l'heure actuelle, supposer qu'une connexion directe soit disponible entre deux ordinateurs arbitraires constitue une limitation significative de l'applicabilité de l'IM dans la pratique. Cependant, plus il y aura de développeurs mettant en œuvre divers bricolages pour le NAT traversal, plus nous resterons longtemps dans un Internet IPv4, avec une probabilité décourageante de connexion entre des ordinateurs quelconques. Jusqu'à quand devrons-nous supporter l'absence d'IPv6 chez nous et au travail ?

Nous aurons un réseau friend-to-friend : tous les interlocuteurs possibles doivent être connus à l'avance. Tout d'abord, cela simplifie énormément les choses : vous vous êtes présentés, vous avez trouvé ou non le nom/clés, vous vous êtes déconnectés ou nous continuons à travailler, sachant qui est notre interlocuteur. Deuxièmement, en général, c'est sécurisé et cela exclut de nombreuses attaques.

L'interface IM sera proche des solutions classiques. des projets suckless, qui me plaisent beaucoup pour leur minimalisme et leur philosophie Unix-way. Le programme IM crée un répertoire pour chaque interlocuteur avec trois sockets de domaine Unix :

• in — dans lequel sont enregistrés les messages envoyés à l'interlocuteur ;
• out — d'où sont lus les messages reçus de l'interlocuteur ;
• state — en le lisant, nous savons si l'interlocuteur est actuellement connecté, adresse/port de connexion.

De plus, un socket conn est créé, en y enregistrant l'hôte et le port, nous initions la connexion à l'interlocuteur distant.

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

Cette approche permet de réaliser des implémentations indépendantes du transport IM et de l'interface utilisateur, car les goûts et les couleurs ne se discutent pas, il est impossible de plaire à tout le monde. En utilisant tmux et/ou multitail, vous pouvez obtenir une interface multi-fenêtres avec coloration syntaxique. Et grâce à rlwrap , vous pouvez obtenir une ligne de saisie de messages compatible avec GNU Readline.

En réalité, les projets suckless utilisent des fichiers FIFO. Personnellement, je n'ai pas pu comprendre comment travailler avec des fichiers de manière concurrente avec asyncio sans fournir ma propre couche de threads dédiés (pour ce genre de choses, j'utilise depuis longtemps le langage Go). C'est pourquoi j'ai décidé de m'en tenir aux sockets de domaine Unix. Malheureusement, cela empêche de faire echo 2001:470:dead::babe 6666 > conn. J'ai résolu ce problème en utilisant socat: echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.

Le protocole initial non sécurisé

Utilise TCP comme transport : il garantit la livraison et son ordre. UDP ne garantit ni l'un ni l'autre (ce qui serait utile lors de l'application de la cryptographie), et le support SCTP n'est pas disponible en Python par défaut.

Malheureusement, il n'y a pas de notion de message dans TCP, seulement un flux de bytes. Il est donc nécessaire de concevoir un format pour les messages afin de pouvoir les séparer dans ce flux. Nous pouvons convenir d'utiliser le caractère de retour à la ligne. Cela conviendra pour commencer, cependant, lorsque nous commencerons à chiffrer nos messages, ce caractère pourrait apparaître n'importe où dans le texte chiffré. C'est pourquoi, dans les réseaux, les protocoles qui envoient d'abord la longueur du message en bytes sont populaires. Par exemple, en Python, il existe la bibliothèque xdrlib qui permet de travailler avec un tel format. XDR.

Nous ne travaillerons pas correctement et efficacement en lisant TCP — simplifions le code. Nous lisons en boucle sans fin les données du socket jusqu'à ce que nous ayons décodé le message complet. Comme format pour cette approche, nous pouvons utiliser JSON ou XML. Cependant, lorsque la cryptographie sera ajoutée, les données devront être signées et authentifiées — et cela nécessitera une représentation identique byte à byte des objets, ce que JSON/XML ne garantit pas (le résultat des dumps peut différer).

XDR convient pour cette tâche, mais je choisis ASN.1 avec codage DER et (que j'ai déjà la bibliothèque, car nous aurons à disposition des objets de haut niveau avec lesquels il est souvent plus agréable et plus pratique de travailler. Contrairement à schemaless bencode, MessagePack ou CBOR, ASN.1 vérifiera automatiquement les données par rapport à un schéma défini de manière rigide.

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

Le message accepté sera Msg : soit un MsgText textuel (pour l'instant avec un seul champ de texte), soit un message de poignée de main MsgHandshake (dans lequel le nom de l'interlocuteur est transmis). Cela semble pour l'instant trop complexe, mais c'est une base pour l'avenir.

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

IM sans cryptographie

Comme je l'ai déjà mentionné, pour toutes les opérations avec les sockets, nous utiliserons la bibliothèque asyncio. Déclarons ce que nous attendons au moment du lancement :

parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
    "--our-name",
    required=True,
    help="Notre nom de pair",
)
parser.add_argument(
    "--their-names",
    required=True,
    help="Leurs noms de pairs, séparés par des virgules",
)
parser.add_argument(
    "--bind",
    default="::1",
    help="Adresse à écouter",
)
parser.add_argument(
    "--port",
    type=int,
    default=6666,
    help="Port à écouter",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))

Un nom propre est défini (—our-name alice). Tous les pairs attendus sont énumérés séparés par des virgules (—their-names bob,eve). Pour chaque pair, un répertoire avec des sockets Unix est créé, ainsi qu'une coroutine pour chaque 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"))

Les messages entrants des utilisateurs depuis le socket in sont envoyés dans les queues IN_QUEUES :

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

Les messages entrants des pairs sont envoyés dans les queues OUT_QUEUES, d'où les données sont écrites dans le socket out :

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

En lisant depuis le socket state, le programme cherche dans le dictionnaire PEER_ALIVE l'adresse du pair. Si aucune connexion avec le pair n'existe, une chaîne vide est enregistrée.

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

Lors de l'écriture de l'adresse dans le socket conn, la fonction «initiateur» de la connexion est lancée :

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

Considérons l'initiateur. D'abord, il établit clairement une connexion avec l'hôte/port spécifié et envoie un message de handshake avec son nom :

 130 async def initiator(hôte, port):
 131     _id = repr((hôte, port))
 132     logging.info("%s : numérotation", _id)
 133     lecteur, écrivain = await asyncio.open_connection(hôte, port)
 134     # Message de poignée de main {{{
 135     écrivain.write(Msg(("poignée de main", MsgHandshake((
 136         ("nomPeer", NOTRE_NOM),
 137     )))).encode())
 138     # }}}
 139     await écrivain.drain()

Ensuite, il attend la réponse de la partie distante. Il tente de décoder la réponse reçue selon le schéma Msg ASN.1. Supposons que l'ensemble du message sera envoyé dans un seul segment TCP et que nous le recevrons de manière atomique lors de l'appel à .read(). Vérifions que nous avons bien reçu un message de poignée de main.

 141     # Attendre le message de poignée de main {{{
 142     données = await lecteur.read(256)
 143     if données == b"":
 144         logging.warning("%s : pas de réponse, déconnexion", _id)
 145         écrivain.close()
 146         return
 147     try:
 148         msg, _ = Msg().decode(données)
 149     except ASN1Error:
 150         logging.warning("%s : réponse indécodable, déconnexion", _id)
 151         écrivain.close()
 152         return
 153     logging.info("%s : reçu %s message", _id, msg.choice)
 154     if msg.choice != "poignée de main":
 155         logging.warning("%s : message inattendu, déconnexion", _id)
 156         écrivain.close()
 157         return
 158     # }}}

Vérifions que le nom de l'interlocuteur reçu nous est connu. Si ce n'est pas le cas, nous interrompons la connexion. Vérifions s'il y avait déjà une connexion établie avec lui (l'interlocuteur a de nouveau donné l'ordre de se connecter à nous) et fermons-la. Dans IN_QUEUES, des chaînes Python contenant le texte du message sont placées, mais une valeur spéciale None signale à la coroutine msg_sender d'arrêter de fonctionner, afin qu'elle oublie son écrivain, lié à l'ancienne connexion TCP.

 159     msg_handshake = msg.value
 160     nom_peer = str(msg_handshake["nomPeer"])
 161     if nom_peer not in LEURS_NOMS:
 162         logging.warning("nom peer inconnu : %s", nom_peer)
 163         écrivain.close()
 164         return
 165     logging.info("%s : session établie : %s", _id, nom_peer)
 166     # Exécuter l'expéditeur de messages texte, initialiser le décodeur de transport {{{
 167     peer_alive = PEER_ALIVES.pop(nom_peer, None)
 168     if peer_alive is not None:
 169         peer_alive.close()
 170         await IN_QUEUES[nom_peer].put(None)
 171     PEER_ALIVES[nom_peer] = écrivain
 172     asyncio.ensure_future(msg_sender(nom_peer, écrivain))
 173     # }}}

msg_sender gère les messages sortants (insérés dans la file d'attente depuis le socket in), les sérialise en message MsgText et les envoie via la connexion TCP. Celle-ci peut se rompre à tout moment — c'est ce que nous interceptons explicitement.

async def msg_sender(nom_peer: str, écrivain) -> None:
    in_queue = IN_QUEUES[nom_peer]
    while True:
        texte = await in_queue.get()
        if texte is None:
            break
        écrivain.write(Msg(("texte", MsgText((
            ("texte", UTF8String(texte)),
        )))).encode())
        try:
            await écrivain.drain()
        except ConnectionResetError:
            del PEER_ALIVES[nom_peer]
            return
        logging.info("%s : message envoyé de %d caractères", nom_peer, len(texte))

À la fin, l'initiateur entre dans une boucle infinie pour lire les messages du socket. Il vérifie s'il s'agit de messages texte et les place dans la file OUT_QUEUES, d'où ils seront envoyés au socket de l'interlocuteur correspondant. Pourquoi ne pas simplement faire .read() et décoder le message ? Parce qu'il est possible que plusieurs messages de l'utilisateur soient agrégés dans le tampon du système d'exploitation et envoyés en un seul segment TCP. Nous pourrons décoder le premier, mais il se peut qu'il reste une partie du suivant dans le tampon. En cas de toute situation anormale, nous fermons la connexion TCP et arrêtons la coroutine msg_sender (en envoyant None dans la file OUT_QUEUES).

 174     buf = b""
 175     # Attendre les messages de test {{{
 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: taille maximale du tampon dépassée", _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: message inattendu %s", _id, msg.choice)
 191             break
 192         try:
 193             await msg_receiver(msg.value, peer_name)
 194         except ValueError as err:
 195             logging.warning("%s: %s", err)
 196             break
 197     # }}}
 198     logging.info("%s: déconnexion : %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: message de %d caractères reçu", peer_name, len(text))
  69     await OUT_QUEUES[peer_name].put(text)

Revenons au code principal. Après la création de toutes les coroutines au moment du lancement du programme, nous démarrons le serveur TCP. Pour chaque connexion établie, il crée une coroutine de réponse (responder).

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

Le responder est similaire à l'initiateur et effectue toutes les mêmes actions de façon miroir, mais la boucle infinie de lecture des messages démarre immédiatement, pour la simplicité. Actuellement, le protocole de poignée de main envoie un message de chaque côté, mais à l'avenir, il y en aura deux de la part de l'initiateur de la connexion, après quoi l'envoi de texte sera immédiatement possible.

  72 async def responder(reader, writer):
  73     _id = writer.get_extra_info("peername")
  74     logging.info("%s: connecté", _id)
  75     buf = b""
  76     msg_expected = "handshake"
  77     peer_name = None
  78     while True:
  79         # Lire jusqu'à obtenir un message Msg {{{
  80         data = await reader.read(MaxMsgLen)
  81         if data == b"":
  82             logging.info("%s: connexion fermée", _id)
  83             break
  84         buf += data
  85         if len(buf) > MaxMsgLen:
  86             logging.warning("%s: taille maximale du tampon dépassée", _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: message %s inattendu", _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         # Traiter le message de Handshake {{{
 104         elif msg_expected == "handshake":
 105             logging.info("%s: reçu le message %s", _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("nom de pair inconnu : %s", peer_name)
 110                 break
 111             writer.write(Msg(("handshake", MsgHandshake((
 112                 ("peerName", OUR_NAME),
 113             )))).encode())
 114             await writer.drain()
 115             logging.info("%s: session établie : %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: déconnexion", _id)
 125     if msg_expected == "text":
 126         IN_QUEUES[peer_name].put(None)
 127     writer.close()

Protocole sécurisé

Il est temps de sécuriser notre communication. Que entendons-nous par sécurité et que voulons-nous :

• la confidentialité des messages transmis ;
• l'authenticité et l'intégrité des messages transmis — toute modification doit être détectée ;
• protection contre les attaques de répétition (replay attack) — la perte ou la répétition des messages doit être détectée (et nous décidons de couper la connexion) ;
• identification et authentification des interlocuteurs par des clés publiques préenregistrées — nous avons déjà décidé de faire un réseau ami-à-ami. Ce n'est qu'après authentification que nous saurons avec qui nous communiquons ;
• la présence perfect forward secrecy propriétés (PFS) — la compromission de notre clé de signature à longue durée de vie ne doit pas permettre de lire toute la correspondance précédente. L'enregistrement du trafic intercepté devient inutile ;
• La validité des messages (de transport et de poignée de main) n’est valable que dans le cadre d'une seule session TCP. L'insertion de messages correctement signés/authentifiés d'une autre session (même avec le même interlocuteur) ne devrait pas être possible;
• un observateur passif ne doit voir ni les identifiants des utilisateurs, ni les clés publiques à long terme transmises, ni les hachages correspondants. Une certaine anonymité vis-à-vis de l'observateur passif.

Étonnamment, presque tous souhaitent avoir ce minimum dans tout protocole de poignée de main, et très peu de ce qui est mentionné est finalement respecté pour les protocoles « maison ». Donc, cette fois, ne cherchons pas à inventer quelque chose de nouveau. Je recommanderais sans hésitation d'utiliser Noise framework pour construire des protocoles, mais choisissons quelque chose de plus simple.

Deux protocoles sont les plus populaires :

• TLS – un protocole très complexe avec une longue histoire de bogues, de défauts, de vulnérabilités, de mauvaises conceptions, de complexité et de ratés (cela dit, cela s'applique peu à TLS 1.3). Mais nous ne le considérons pas en raison d’une trop grande complexité.
• IPsec avec IKE – n'ont pas de problèmes cryptographiques sérieux, bien qu'ils ne soient pas non plus simples. Si on lit sur IKEv1 et IKEv2, leur origine est STS, ISO/IEC IS 9798-3 et les protocoles SIGMA (SIGn-and-MAc) – suffisamment simples à mettre en œuvre en une soirée.

Qu'est-ce qui rend SIGMA, en tant que dernière étape de l’évolution des protocoles STS/ISO, si bon ? Il satisfait à toutes nos exigences (y compris la « dissimulation » des identifiants des interlocuteurs), n’a pas de problèmes cryptographiques connus. Il est minimaliste : la suppression d'au moins un élément du message du protocole entraînerait son insécurité.

Passons d'un protocole maison le plus simple à SIGMA. L'opération de base qui nous intéresse est l'accord de clés: une fonction à l'issue de laquelle les deux participants obtiendront la même valeur, qui pourra être utilisée comme clé symétrique. Sans entrer dans les détails : chaque partie génère une paire de clés éphémères (utilisée uniquement dans le cadre d'une session) (clés publique et privée), échange des clés publiques, appelle la fonction d'accord à laquelle elle transmet sa clé privée et la clé publique de l'interlocuteur.

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

Tout le monde peut s’immiscer et remplacer les clés publiques par les siennes — ce protocole n’a pas d'authentification des interlocuteurs. Ajoutons une signature avec des clés de longue durée.

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

Une telle signature ne conviendra pas, car elle n'est pas liée à une session spécifique. De tels messages « conviendront » également pour des sessions avec d'autres participants. Tout le contexte doit être signé. Cela nécessite également d'ajouter l'envoi d'un autre message de A.

De plus, il est crucial d'ajouter sous la signature notre propre identifiant, sinon nous pouvons modifier IdXXX et resigner le message avec la clé d'un autre interlocuteur connu. Pour éviter des attaques de réflexion, il est nécessaire que les éléments sous la signature soient clairement positionnés selon leur signification : si A signe (PubA, PubB), alors B doit signer (PubB, PubA). Cela souligne également l'importance du choix de la structure et du format des données sérialisées. Par exemple, les ensembles en codage ASN.1 DER sont triés : SET OF(PubA, PubB) sera identique à SET OF(PubB, PubA).

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

Cependant, nous n'avons toujours pas « prouvé » que nous avons généré une clé secrète commune pour cette session. En principe, nous pourrions nous passer de cette étape — le premier message de transport serait invalide, mais nous voulons nous assurer qu'une fois l handshake terminé, tout est réellement validé. Pour l'instant, nous avons entre les mains le protocole ISO/IEC IS 9798-3.

Nous pourrions signer la clé générée elle-même. C'est dangereux, car il se peut que l'algorithme de signature utilisé présente des fuites (même si ce ne sont que des bits de signature, ce sont tout de même des fuites). On peut signer le hachage de la clé générée, mais une fuite même du hachage de la clé générée peut avoir de la valeur lors d'attaques par force brute sur la fonction de génération. SIGMA utilise une fonction MAC, authentifiant l'identifiant de l'expéditeur.

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

Dans un souci d'optimisation, certains pourraient souhaiter réutiliser leurs clés éphémères (ce qui, bien sûr, est désavantageux pour PFS). Par exemple, nous avons généré une paire de clés, tenté de nous connecter, mais TCP n'était pas disponible ou s'est interrompu en cours de protocole. Il est dommage de gaspiller l'entropie et les ressources processeur sur une nouvelle paire. Nous allons donc introduire ce que l'on appelle un cookie — une valeur pseudo-aléatoire qui protégera contre de potentielles attaques de répétition lors de la réutilisation des clés publiques éphémères. En raison de la liaison entre le cookie et la clé publique éphémère, la clé publique de l'autre participant peut être omise de la signature pour des raisons de non-utilité.

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

Enfin, nous souhaitons protéger la confidentialité de nos identifiants d'interlocuteurs vis-à-vis des observateurs passifs. Pour cela, SIGMA propose d'abord d'échanger des clés éphémères, de générer une clé commune sur laquelle seront chiffrés les messages d'authentification et d'identification. SIGMA décrit deux options :

• SIGMA-I — protège l'initiateur contre les attaques actives, le répondant contre les attaques passives : l'initiateur authentifie le répondant et s'il y a un problème, il ne révèle pas son identification. Le répondant, quant à lui, révèle son identification s'il commence un protocole actif. Un observateur passif n'apprendra rien ;
  SIGMA-R — protège le répondant contre les attaques actives, l'initiateur contre les attaques passives. Tout se fait exactement à l'inverse, mais dans ce protocole, il y a déjà quatre messages de poignée de main qui sont transmis.

  Nous choisissons SIGMA-I, qui ressemble davantage à ce que nous attendons des choses familières client-serveur : le client ne reconnaît que le serveur authentifié, tandis que le serveur connaît déjà tout. De plus, il est plus simple à mettre en œuvre en raison du nombre réduit de messages d'initialisation. Tout ce que nous ajoutons au protocole, c'est le chiffrement d'une partie du message et le transfert de l'identifiant A dans la partie chiffrée du dernier message :

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

  • Pour la signature, nous utilisons GOST R 34.10-2012 un algorithme avec des clés de 256 bits.
  • Pour la génération de la clé commune, le 34.10-2012 VKO est utilisé.
  • Comme MAC, CMAC est utilisé. Techniquement, c'est un mode de fonctionnement particulier d'un chiffre de bloc, décrit dans GOST R 34.13-2015. Comme fonction de chiffrement pour ce mode — Grasshopper (34.12-2015).
  • L'identifiant de l'interlocuteur est un hachage de sa clé publique. Pour le hachage, nous utilisons Stribog-256 (34.11-2012 256 bits).

  Après l'établissement de la connexion, nous aurons une clé commune convenue. Nous pouvons l'utiliser pour le chiffrement authentifié des messages de transport. Cette partie est assez simple et il est difficile de se tromper : nous incrémentons le compteur de messages, chiffrons le message, authentifions (MAC) le compteur et le texte chiffré, puis envoyons. Lors de la réception du message, nous vérifions que le compteur a la valeur attendue, authentifions le texte chiffré avec le compteur, puis déchiffrons. Quelle clé utiliser pour chiffrer les messages d'établissement, ceux de transport, et quelle clé pour l'authentification ? Utiliser une seule clé pour toutes ces tâches est dangereux et imprudent. Il est nécessaire de générer des clés en utilisant des fonctions spécialisées KDF (key derivation function). Encore une fois, ne compliquons pas les choses en inventant quelque chose : HKDF est bien connu, bien étudié et ne présente pas de problèmes connus. Malheureusement, la bibliothèque native de Python ne contient pas cette fonction, nous allons donc utiliser hkdf package. HKDF utilise à l'intérieur HMAC, qui utilise à son tour la fonction de hachage. Un exemple d'implémentation en Python sur la page Wikipedia ne nécessite que quelques lignes de code. Comme dans le cas de 34.10-2012, nous allons utiliser Stribog-256 comme fonction de hachage. La sortie de notre fonction de dérivation de clés sera appelée la clé de session, à partir de laquelle seront dérivées les clés symétriques manquantes :

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

  Structures / schémas

  Examinons quelles structures ASN.1 nous avons maintenant obtenues pour la transmission de toutes ces données :

  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 — ce qui sera signé (to be signed). HandshakeTBE — ce qui sera chiffré (to be encrypted). Je souligne le champ ukm dans MsgHandshake1. 34.10 VKO, pour une randomisation encore plus grande des clés générées, inclut le paramètre UKM (user keying material) — simplement de l'entropie supplémentaire.

  Ajout de la cryptographie dans le code

  Nous allons examiner uniquement les modifications apportées au code original, car la structure est restée la même (en réalité, l'implémentation finale a d'abord été écrite, puis toute la cryptographie a été retirée).

  Étant donné que l'authentification et l'identification des interlocuteurs se feront par des clés publiques, il faut désormais les stocker de manière durable. Pour simplifier, utilisons un JSON de ce type :

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

  our — notre paire de clés, clés privées et publiques hexadécimales. their — noms des interlocuteurs et leurs clés publiques. Modifions les arguments de la ligne de commande et ajoutons un post-traitement des données JSON :

  from pygost import gost3410
  from pygost.gost34112012256 import GOST34112012256
  
  CURVE = gost3410.GOST3410Curve(
      *gost3410.CURVE_PARAMS["GostR3410_2001_CryptoPro_A_ParamSet"]
  )
  
  parser = argparse.ArgumentParser(description="GOSTIM")
  parser.add_argument(
      "--keys-gen",
      action="store_true",
      help="Générez un JSON avec notre nouvelle paire de clés",
  )
  parser.add_argument(
      "--keys",
      default="keys.json",
      required=False,
      help="JSON avec nos clés et leurs clés",
  )
  parser.add_argument(
      "--bind",
      default="::1",
      help="Adresse à écouter",
  )
  parser.add_argument(
      "--port",
      type=int,
      default=6666,
      help="Port à écouter",
  )
  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)
  
  # Analysez et désérialisez nos clés et leurs clés {{{
  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),
      }
  # }}}
  

  La clé privée de l'algorithme 34.10 est un nombre aléatoire. De taille 256 bits pour les courbes elliptiques 256 bits. PyGOST ne fonctionne pas avec un ensemble d'octets, mais avec de grands nombres, c'est pourquoi notre clé privée (urandom(32)) doit être convertie en un nombre à l'aide de gost3410.prv_unmarshal(). La clé publique est déterminée à partir de la clé privée, à l'aide de gost3410.public_key(). La clé publique 34.10 est un couple de grands nombres qui doivent également être convertis en une séquence d'octets pour faciliter le stockage et le transfert, à l'aide de gost3410.pub_marshal().

  Après avoir lu le fichier JSON, les clés publiques doivent être converties à nouveau, à l'aide de gost3410.pub_unmarshal(). Comme nous allons recevoir les identifiants des interlocuteurs sous forme de hachage de la clé publique, nous pouvons les pré-calculer et les placer dans un dictionnaire pour une recherche rapide. Le hachage Streebog-256 est défini par gost34112012256.GOST34112012256(), qui satisfait complètement l'interface de hachage hashlib.

  Comment la coroutine de l'initiateur a-t-elle changé ? Tout comme dans le schéma de la poignée de main : nous générons un cookie (128 bits est amplement suffisant), une paire de clés éphémères 34.10, qui sera utilisée pour la fonction d'accord de clés VKO.

   395 async def initiator(host, port):
   396     _id = repr((host, port))
   397     logging.info("%s: dialing", _id)
   398     reader, writer = await asyncio.open_connection(host, port)
   399     # Générer notre clé publique éphémère et cookie, envoyer le message Handshake 0 {{{
   400     cookie_our = Cookie(urandom(16))
   401     prv = gost3410.prv_unmarshal(urandom(32))
   402     pub_our = gost3410.public_key(CURVE, prv)
   403     pub_our_raw = PubKey(gost3410.pub_marshal(pub_our))
   404     writer.write(Msg(("handshake0", MsgHandshake0((
   405         ("cookieInitiator", cookie_our),
   406         ("pubKeyInitiator", pub_our_raw),
   407     )))).encode())
   408     # }}}
   409     await writer.drain()
  

  • Nous attendons la réponse et décodons le message Msg reçu;
  • Nous assurons que nous avons reçu handshake1;
  • Nous décodons la clé publique éphémère de l'autre partie et calculons la clé de session;
  • Nous générons les clés symétriques nécessaires pour le traitement de la partie TBE du message.

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

  UKM est un nombre de 64 bits (urandom(8)), qui nécessite également une désérialisation à partir de la représentation par bytes, en utilisant gost3410_vko.ukm_unmarshal(). La fonction VKO pour 34.10-2012 256 bits est gost3410_vko.kek_34102012256() (KEK — clé de chiffrement).

  La clé de session générée est déjà une séquence aléatoire de bytes de 256 bits. Elle peut donc être immédiatement utilisée dans la fonction HKDF. Étant donné que GOST34112012256 satisfait à l'interface hashlib, elle peut être directement utilisée dans la classe Hkdf. Le sel (premier argument Hkdf) n'est pas spécifié, car la clé générée en raison de l'éphémérité des paires de clés impliquées sera différente pour chaque session et contient déjà suffisamment d'entropie. kdf.expand() génère par défaut des clés d'une longueur de 256 bits, requises pour le Kuznechik par la suite.

  Ensuite, les parties TBE et TBS du message reçu sont vérifiées :

  • MAC est calculé et vérifié sur le texte chiffré reçu ;
  • Le texte chiffré est déchiffré ;
  • La structure TBE est décodée ;
  • De celle-ci, l'identifiant de l'interlocuteur est pris et il est vérifié s'il nous est connu ;
  • MAC est calculé et vérifié sur cet identifiant ;
  • la signature est vérifiée au-dessus de la structure TBS, qui inclut les cookies des deux parties et la clé éphémère publique de l'autre partie. La signature est vérifiée avec la clé de signature à long terme de l'interlocuteur.

   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, déconnexion", _id, err)
   453         writer.close()
   454         return
   455     # }}}
  
   128 def validate_tbe(
   129         msg_handshake: Union[MsgHandshake1, MsgHandshake2],
   130         key_mac_identity: bytes,
   131         key_enc: bytes,
   132         key_mac: bytes,
   133         cookie_their: Cookie,
   134         cookie_our: Cookie,
   135         pub_key_our: PubKey,
   136 ) -> str:
   137     ciphertext = bytes(msg_handshake["ciphertext"])
   138     mac_tag = mac(GOST3412Kuznechik(key_mac).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext)
   139     if not compare_digest(mac_tag, bytes(msg_handshake["ciphertextMac"])):
   140         raise ValueError("MAC invalide")
   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("impossible de décoder TBE")
   151     key_sign_pub_hash = bytes(tbe["identity"])
   152     peer = KEYS.get(key_sign_pub_hash)
   153     if peer is None:
   154         raise ValueError("identité inconnue")
   155     mac_tag = mac(
   156         GOST3412Kuznechik(key_mac_identity).encrypt,
   157         KUZNECHIK_BLOCKSIZE,
   158         key_sign_pub_hash,
   159     )
   160     if not compare_digest(mac_tag, bytes(tbe["identityMac"])):
   161         raise ValueError("MAC d'identité invalide")
   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("signature invalide")
   174     return peer["name"]
  

  Comme mentionné précédemment, la norme 34.13-2015 décrit divers modes de fonctionnement des blocs de chiffrement de la norme 34.12-2015. Parmi eux se trouve le mode de génération de l'IMyD et de calcul du MAC. Dans PyGOST, il s'agit de gost3413.mac(). Ce mode nécessite le passage de la fonction de chiffrement (qui prend et renvoie un bloc de données), de la taille du bloc de chiffrement, et des données elles-mêmes. Pourquoi ne pas hardcoder la taille du bloc de chiffrement ? La norme 34.12-2015 ne décrit pas seulement le chiffre à 128 bits Kuznechik, mais aussi le 64 bits Magma — une version légèrement modifiée du GOST 28147-89, créée encore au KGB et ayant toujours l'un des niveaux de sécurité les plus élevés.

  Le GOST.3412.KUZNECHIK s'initialise en appelant GOST3412Kuznechik(key) et retourne un objet avec les méthodes .encrypt() et .decrypt(), pratiques pour la transmission aux fonctions 34.13. Le MAC est calculé comme suit : gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Pour comparer le MAC calculé et celui reçu, il ne faut pas utiliser la comparaison standard (==) des chaînes d'octets, car cela entraîne des fuites de temps de comparaison, ce qui peut, en général, conduire à des vulnérabilités fatales de type BEAST attaques sur TLS. En Python, il existe une fonction hmac.compare_digest spéciale pour cela.

  La fonction du chiffreur par blocs ne peut chiffrer qu'un seul bloc de données. Pour un nombre plus élevé, et aussi de longueur non multiple, il est nécessaire d'utiliser un mode de chiffrement. Les modes suivants sont décrits dans 34.13-2015 : ECB, CTR, OFB, CBC, CFB. Chacun a ses propres domaines d'application et caractéristiques. Malheureusement, il n'existe toujours pas de modes de chiffrement authentifiés, (comme CCM, OCB, GCM et autres) — nous sommes contraints d'ajouter nous-mêmes au moins un MAC. Je choisis le mode compteur (CTR) : il n'exige pas d'extension à la taille du bloc, peut être parallélisé, utilise uniquement la fonction de chiffrement, et peut être utilisé en toute sécurité pour chiffrer un grand nombre de messages (contrairement à CBC, où les collisions commencent relativement rapidement). Comme pour .mac(), .ctr() accepte des données similaires en entrée : ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). Il est nécessaire de définir un vecteur d'initialisation, d'une longueur exactement égale à la moitié de la taille du bloc de chiffrement. Si notre clé de chiffrement est utilisée uniquement pour chiffrer un seul message (même s'il comporte plusieurs blocs), il est sûr de définir un vecteur d'initialisation nul. Pour le chiffrement des messages de handshake, nous utilisons à chaque fois une clé distincte.

  La vérification de la signature avec gost3410.verify() est triviale : nous passons la courbe elliptique dans laquelle nous opérons (cela est simplement fixé dans notre protocole GOSTIM), la clé publique du signataire (n'oublions pas que cela doit être un tuple de deux grands nombres, et non une chaîne d'octets), le hachage 34.11-2012 et la signature reçue elle-même.

  Ensuite, dans l'initiateur, nous préparons et envoyons le message handshake2 de négociation, en effectuant les mêmes actions que nous avons faites lors de la vérification, seulement de manière symétrique : création de signatures sur nos propres clés au lieu de vérification, etc.

  ...

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

  Une fois la session établie, des clés de transport sont générées (une clé distincte pour le chiffrement, pour l'authentification, pour chaque partie), et le Kuznechik est initialisé pour le déchiffrement et la vérification du MAC :

   499     # Exécuter l'envoi de messages texte, initialiser le décodeur de transport {{{
   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     # Attendre les messages de test {{{
   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     # }}}
  

  La coroutine msg_sender chiffre désormais les messages avant de les envoyer dans la connexion TCP. Chaque message a un nonce croissant de manière monotone, qui sert également de vecteur d'initialisation lors du chiffrement en mode compteur. Chaque message et bloc de message auront des valeurs de compteur qui diffèrent assurément.

  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
  

  Les messages entrants sont traités par la coroutine msg_receiver, qui s'occupe de l'authentification et du déchiffrement :

  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("valeur de nonce inattendue")
      mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode())
      if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])):
          raise ValueError("MAC invalide")
      plaintext = ctr(
          encrypter,
          KUZNECHIK_BLOCKSIZE,
          bytes(payload["ciphertext"]),
          long2bytes(nonce_expected, 8),
      )
      text = plaintext.decode("utf-8")
      await OUT_QUEUES[peer_name].put(text)
  

  Conclusion

  GOSTIM est destiné à être utilisé uniquement à des fins éducatives (car il n'est pas couvert par des tests, au minimum) ! Le code source du programme peut être téléchargé ici (Hachage Stribog-256 : 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Comme tous mes projets, type GoGOST, (que j'ai déjà, NNCP, GoVPN, GOSTIM est entièrement un logiciel libre, distribué sous les conditions de GPLv3+.

  Sergueï Matveïev, cipherpunk, membre de la Fondation du logiciel libre, développeur Python/Go, expert principal au FGUP « NTC Atlas ».

Source : habr.com