Siendo un desarrollador de la biblioteca (primitivas criptográficas GOST en Python puro), a menudo recibo preguntas sobre cómo implementar un intercambio seguro de mensajes de manera sencilla. Muchos piensan que la criptografía aplicada es bastante simple, y que una llamada .encrypt() de un cifrador por bloques es suficiente para enviar datos de forma segura a través de un canal de comunicación. Otros creen que la criptografía aplicada es cosa de unos pocos, y que es aceptable que empresas ricas como Telegram tengan matemáticos de élite. un protocolo seguro.
Todo esto me llevó a escribir este artículo para mostrar que implementar protocolos criptográficos y un servicio de mensajería instantánea seguro no es una tarea tan complicada. Sin embargo, no es aconsejable inventar protocolos propios de autenticación y negociación de claves.
En este artículo se describirá , , el protocolo de autenticación y negociación de claves SIGMA-I (en el que se basa ), utilizando exclusivamente algoritmos criptográficos GOST de la biblioteca PyGOST y la codificación de mensajes ASN.1 de la biblioteca (de la que ya he ). La condición necesaria: debe ser lo suficientemente simple como para poder ser escrito desde cero en una noche (o un día de trabajo), de lo contrario, ya no sería un programa sencillo. Seguramente tendrá errores, complejidades innecesarias y fallos, además, este es mi primer programa usando la biblioteca asyncio. Para comenzar, necesitamos entender cómo se verá nuestro IM. Para simplificar, será una red peer-to-peer, sin ningún tipo de descubrimiento de participantes. Conectaremos manualmente a qué dirección: puerto conectarnos para comunicarnos con el interlocutor.
Entiendo que, en este momento, suponer la disponibilidad de una conexión directa entre dos computadoras arbitrarias es una limitación significativa en la aplicabilidad del IM en la práctica. Pero cuanto más desarrolladores implementen todo tipo de trucos de NAT-traversal, más tiempo seguiremos en el Internet IPv4, con una probabilidad desalentadora de conexión entre computadoras arbitrarias. ¿Hasta cuándo vamos a soportar la falta de IPv6 en casa y en el trabajo?
Entiendo que, en este momento, la suposición de que existe una conexión directa entre dos computadoras arbitrarias es una limitación significativa en la aplicabilidad del IM en la práctica. Pero cuanto más desarrolladores implementen diversos trucos de NAT-traversal, más tiempo permaneceremos en una Internet IPv4, con la desalentadora probabilidad de conexión entre computadoras arbitrarias. ¿Cuánto tiempo más podemos soportar la falta de IPv6 en casa y en el trabajo?
Entiendo que, en este momento, suponer la disponibilidad de una conexión directa entre dos computadoras arbitrarias es una limitación significativa para la aplicabilidad de la IM en la práctica. Pero cuántos más desarrolladores implementen costuras para sortear NAT, más tiempo permaneceremos en el Internet IPv4, con una desalentadora probabilidad de conexión entre computadoras arbitrarias. ¿Hasta cuándo podremos soportar la falta de IPv6 en casa y en el trabajo?
Tendremos una red de amigo-a-amigo: todos los posibles interlocutores deben ser conocidos de antemano. En primer lugar, esto simplifica mucho las cosas: nos presentamos, buscamos o no encontramos un nombre/clave, nos desconectamos o seguimos trabajando, sabiendo quién es nuestro interlocutor. En segundo lugar, en general, esto es seguro y excluye muchos ataques.
La interfaz de IM será similar a las soluciones clásicas. , que me gustan mucho por su minimalismo y filosofía al estilo Unix. La programa de IM crea un directorio para cada interlocutor con tres sockets de dominio Unix:
- in — se graban los mensajes enviados al interlocutor;
- out — se leen los mensajes recibidos del interlocutor;
- state — al leer de él, sabemos si el interlocutor está conectado actualmente, dirección/puerto de conexión.
Además, se crea un socket conn; al grabar en él el puerto del host, iniciamos la conexión con el interlocutor remoto.
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Este enfoque permite realizar implementaciones independientes del transporte IM y la interfaz de usuario, ya que sobre gustos y colores no hay compañeros, a cada uno no se le puede complacer. Usando y/o , se puede obtener una interfaz en múltiples ventanas con resaltado de sintaxis. Y con ayuda de , se puede obtener una línea de entrada compatible con GNU Readline para mensajes.
En realidad, los proyectos suckless utilizan archivos FIFO. Personalmente, no pude entender cómo trabajar con archivos de manera concurrente en asyncio sin mi propia base de subprocesos dedicados (para esas cosas hace tiempo que uso el lenguaje ). Por lo tanto, decidí utilizar sockets de dominio Unix. Desafortunadamente, esto limita la posibilidad de hacer echo 2001:470:dead::babe 6666 > conn. Resolví este problema usando : echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
Protocolo inicialmente inseguro
Se utiliza TCP como transporte: garantiza la entrega y su orden. UDP no garantiza ni lo uno ni lo otro (lo que sería útil cuando se aplica la criptografía), y no hay soporte de en Python por defecto.
Lamentablemente, en TCP no existe el concepto de mensaje, solo un flujo de bytes. Por lo tanto, es necesario idear un formato para los mensajes, de modo que se puedan separar entre sí en este flujo. Podemos acordar usar el símbolo de salto de línea. Esto servirá al principio, sin embargo, cuando comencemos a cifrar nuestros mensajes, este símbolo podría aparecer en cualquier lugar del texto cifrado. Por eso, en las redes son populares los protocolos que primero envían la longitud del mensaje en bytes. Por ejemplo, en Python, existe xdrlib que permite trabajar con tal formato. .
No trabajaremos de manera correcta y eficiente con la lectura de TCP; simplificaremos el código. Leemos en un ciclo infinito los datos del socket, hasta que decodificamos el mensaje completo. Como formato para este enfoque, se puede usar tanto JSON como XML. Pero, cuando se agregue criptografía, se requerirá firmar y autenticar los datos, lo que exigirá una representación idéntica byte por byte de los objetos, algo que JSON/XML no garantiza (el resultado de los volcadors puede variar).
XDR es adecuado para esta tarea, sin embargo, elijo ASN.1 con codificación DER y biblioteca, ya que tendremos objetos de alto nivel con los que a menudo es más agradable y conveniente trabajar. A diferencia de sin esquema. , o , ASN.1 comprobará automáticamente los datos en comparación con un esquema rígido asignado.
# 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))),
))
El mensaje recibido será Msg: ya sea MsgText de texto (por ahora con un solo campo de texto), o un mensaje de saludo MsgHandshake (en el que se transmite el nombre del interlocutor). Ahora parece una complicación innecesaria, pero es una base para el futuro.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│MsgHandshake(IdA) │
│─────────────────>│
│ │
│MsgHandshake(IdB) │
││
│ │
│ MsgText() │
│<─────────────────│
│ │
IM sin criptografía
Como ya he mencionado, para todas las operaciones con sockets se utilizará la biblioteca asyncio. Declaremos lo que esperamos en el momento del inicio:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Nuestro nombre de par",
)
parser.add_argument(
"--their-names",
required=True,
help="Nombres de sus pares, separados por comas",
)
parser.add_argument(
"--bind",
default="::1",
help="Dirección a la que escuchar",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Puerto para escuchar",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
Se establece el nombre propio (—our-name alice). Se enumeran todos los interlocutores esperados (—their-names bob,eve), separados por comas. Para cada interlocutor, se crea un directorio con sockets Unix, así como una corutina para cada 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"))
Los mensajes entrantes del usuario desde el socket in se envían a las colas 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"))
Los mensajes entrantes de los interlocutores se envían a las colas OUT_QUEUES, desde las cuales los datos se escriben en el 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()
Al leer desde el socket state, el programa busca en el diccionario PEER_ALIVE la dirección del interlocutor. Si aún no hay conexión con el interlocutor, se escribe una cadena vacía.
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()
Al escribir la dirección en el socket conn, se activa la función "iniciador" de la conexión:
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))
Consideremos el iniciador. Primero, claramente, abre la conexión al host/puerto indicado y envía un mensaje de handshake con su nombre:
130 async def initiador(host, puerto):
131 _id = repr((host, puerto))
132 logging.info("%s: marcando", _id)
133 lector, escritor = await asyncio.open_connection(host, puerto)
134 # Mensaje de inicio {{{
135 escritor.write(Msg(("inicio", MsgHandshake((
136 ("nombrePeer", NUESTRO_NOMBRE),
137 )))).encode())
138 # }}}
139 await escritor.drain()
Luego, espera una respuesta del lado remoto. Intenta decodificar la respuesta recibida según el esquema Msg ASN.1. Suponemos que todo el mensaje se enviará en un solo segmento TCP y lo recibiremos de manera atómica al llamar a .read(). Verificamos que hemos recibido el mensaje de inicio.
141 # Espera el mensaje de inicio {{{
142 datos = await lector.read(256)
143 if datos == b"":
144 logging.warning("%s: sin respuesta, desconectando", _id)
145 escritor.close()
146 return
147 try:
148 msg, _ = Msg().decode(datos)
149 except ASN1Error:
150 logging.warning("%s: respuesta indescifrable, desconectando", _id)
151 escritor.close()
152 return
153 logging.info("%s: recibió mensaje %s", _id, msg.choice)
154 if msg.choice != "inicio":
155 logging.warning("%s: mensaje inesperado, desconectando", _id)
156 escritor.close()
157 return
158 # }}}
Verificamos que el nombre del interlocutor recibido sea conocido. Si no, cortamos la conexión. Comprobamos si ya habíamos establecido una conexión con él (el interlocutor dio nuevamente el comando para conectarse a nosotros) y la cerramos. En IN_QUEUES se colocan las cadenas Python con el texto del mensaje, pero hay un valor especial None, que indica a la corutina msg_sender que debe dejar de trabajar, para que se olvide de su escritor relacionado con la conexión TCP obsoleta.
159 msg_handshake = msg.value
160 nombre_peer = str(msg_handshake["nombrePeer"])
161 if nombre_peer not in SUS_NOMBRES:
162 logging.warning("nombre de peer desconocido: %s", nombre_peer)
163 escritor.close()
164 return
165 logging.info("%s: sesión establecida: %s", _id, nombre_peer)
166 # Ejecutar el emisor de mensajes de texto, inicializar decodificador de transporte {{{
167 peer_alive = PEER_ALIVES.pop(nombre_peer, None)
168 if peer_alive is not None:
169 peer_alive.close()
170 await IN_QUEUES[nombre_peer].put(None)
171 PEER_ALIVES[nombre_peer] = escritor
172 asyncio.ensure_future(msg_sender(nombre_peer, escritor))
173 # }}}
msg_sender recibe los mensajes salientes (que se colocan en la cola del socket de entrada), los serializa en un mensaje MsgText y los envía a través de la conexión TCP. Esta puede cortarse en cualquier momento, lo cual capturamos explícitamente.
async def msg_sender(nombre_peer: str, escritor) -> None:
cola_entrada = IN_QUEUES[nombre_peer]
while True:
texto = await cola_entrada.get()
if texto is None:
break
escritor.write(Msg(("texto", MsgText((
("texto", UTF8String(texto)),
)))).encode())
try:
await escritor.drain()
except ConnectionResetError:
del PEER_ALIVES[nombre_peer]
return
logging.info("%s: envió un mensaje de %d caracteres", nombre_peer, len(texto))
Al final, el iniciador entra en un ciclo infinito de lectura de mensajes del socket. Verifica si se trata de mensajes de texto y los coloca en la cola OUT_QUEUES, de donde serán enviados al socket correspondiente del interlocutor. ¿Por qué no simplemente hacer .read() y decodificar el mensaje? Porque no se excluye la posibilidad de que varios mensajes del usuario se agreguen en el búfer del sistema operativo y se envíen en un solo segmento TCP. Podremos decodificar el primero, pero en el búfer puede quedar parte del siguiente. Ante cualquier situación anómala, cerramos la conexión TCP y detenemos la corutina msg_sender (enviando None a la cola OUT_QUEUES).
174 buf = b""
175 # Esperar mensajes de prueba {{{
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: se ha excedido el tamaño máximo del búfer", _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: mensaje %s inesperado", _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: desconectando: %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: mensaje de %d caracteres recibido", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Volvamos al código principal. Después de crear todas las corutinas en el momento de iniciar el programa, iniciamos el servidor TCP. Para cada conexión establecida, crea una corutina responder (respondedor).
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("Escuchando en: %s", server.sockets[0].getsockname())
loop.run_forever()
El responder es similar al initiator y realiza exactamente las mismas acciones, pero el ciclo infinito de lectura de mensajes se inicia de inmediato, por simplicidad. Actualmente, el protocolo de apretón de manos envía un mensaje de cada lado, pero, en el futuro, el iniciador de conexión enviará dos mensajes, después de lo cual se podrá enviar texto de inmediato.
72 async def responder(reader, writer):
73 _id = writer.get_extra_info("peername")
74 logging.info("%s: conectado", _id)
75 buf = b""
76 msg_expected = "handshake"
77 peer_name = None
78 while True:
79 # Leer hasta que obtengamos un mensaje Msg {{{
80 data = await reader.read(MaxMsgLen)
81 if data == b"":
82 logging.info("%s: conexión cerrada", _id)
83 break
84 buf += data
85 if len(buf) > MaxMsgLen:
86 logging.warning("%s: tamaño máximo del búfer excedido", _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: mensaje %s inesperado", _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 # Procesar mensaje Handshake {{{
104 elif msg_expected == "handshake":
105 logging.info("%s: recibió mensaje %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("nombre de par desconocido: %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: sesión establecida: %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: desconectando", _id)
125 if msg_expected == "text":
126 IN_QUEUES[peer_name].put(None)
127 writer.close()
Protocolo seguro
Ha llegado el momento de asegurar nuestra comunicación. ¿Qué entendemos por seguridad y qué deseamos:
- la confidencialidad de los mensajes transmitidos;
- la autenticidad e integridad de los mensajes transmitidos: cualquier modificación debe ser detectada;
- protección contra ataques de repetición (replay attack): la pérdida o repetición de mensajes debe ser detectada (y decidimos romper la conexión);
- identificación y autenticación de interlocutores mediante claves públicas previamente establecidas: ya decidimos anteriormente que formamos una red de amigo a amigo. Solo después de la autenticación sabremos con quién estamos comunicando;
- presencia de propiedades (PFS): la compromisión de nuestra clave de firma de larga duración no debe permitir la posibilidad de leer toda la conversación anterior. La grabación del tráfico interceptado se vuelve inútil;
- La validez de los mensajes (de transporte y del apretón de manos) solo se mantiene dentro de una misma sesión TCP. La inserción de mensajes firmados/authenticados correctamente de otra sesión (incluso con el mismo interlocutor) no debería ser posible;
- un observador pasivo no debería ver ni identificadores de usuario, ni claves públicas de larga duración transmitidas, ni sus hashes. Cierta anonimidad frente a un observador pasivo.
Es sorprendente, pero este mínimo es algo que prácticamente todos quieren tener en cualquier protocolo de apretón de manos, y muy poco de lo mencionado se logra finalmente en los protocolos "domésticos". Así que esta vez no vamos a inventar algo nuevo. Sin duda recomendaría usar para construir protocolos, aunque elijamos algo más sencillo.
Los dos protocolos más populares son:
- — un protocolo extremadamente complicado con una larga historia de errores, fallas, vulnerabilidades, falta de planificación, complejidad y defectos (sin embargo, esto se aplica poco a TLS 1.3). Pero no lo consideramos debido a su sobrecomplicación.
- con — no tienen problemas criptográficos serios, aunque tampoco son simples. Si se lee sobre IKEv1 y IKEv2, sus orígenes son , ISO/IEC IS 9798-3 y los protocolos SIGMA (SIGn-and-MAc), que son lo suficientemente simples de implementar en una sola noche.
¿Qué hace que SIGMA, como el último eslabón en el desarrollo de los protocolos STS/ISO, sea tan bueno? Satisface todos nuestros requisitos (incluida la "ocultación" de los identificadores de los interlocutores) y no tiene problemas criptográficos conocidos. Es minimalista: eliminar al menos un elemento del mensaje del protocolo lo hará inseguro.
Vamos a recorrer el camino desde el protocolo doméstico más básico hasta el SIGMA. La operación más fundamental que nos interesa es : una función en la que ambas partes obtendrán el mismo valor que se podrá utilizar como clave simétrica. Sin entrar en detalles: cada parte genera un par de claves efímeras (que se utilizan solo dentro de una sesión) (claves pública y privada), intercambian claves públicas, y llaman a la función de acuerdo, a la que pasan su clave privada y la clave pública del interlocutor.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│ IdA, PubA │ ╔════════════════════╗
│───────────────>│ ║PrvA, PubA = DHgen()║
│ │ ╚════════════════════╝
│ IdB, PubB │ ╔════════════════════╗
│<───────────────│ ║PrvB, PubB = DHgen()║
│ │ ╚════════════════════╝
────┐ ╔═══════╧════════════╗
│ ║Key = DH(PrvA, PubB)║
<───┘ ╚═══════╤════════════╝
│ │
│ │
Cualquiera puede interferir y reemplazar las claves públicas por las suyas propias: no hay autenticación de los interlocutores en este protocolo. Agregaremos una firma con claves de larga duración.
┌─────┐ ┌─────┐
│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) ║ │
│ ╚═════════════════════╝ │
│ │
Esta firma no es adecuada, ya que no está vinculada a una sesión específica. Estos mensajes "serían adecuados" también para sesiones con otros participantes. Todo el contexto debe ser firmado. Esto también obliga a agregar el envío de otro mensaje por parte de A.
Además, es crítico agregar bajo la firma también su propio identificador, ya que, de lo contrario, podríamos reemplazar IdXXX y volver a firmar el mensaje con la clave de otro interlocutor conocido. Para prevenir , es necesario que los elementos bajo la firma se encuentren en lugares claramente establecidos según su significado: si A firma (PubA, PubB), entonces B debe firmar (PubB, PubA). Esto también resalta la importancia de elegir la estructura y el formato de los datos serializados. Por ejemplo, los conjuntos en codificación ASN.1 DER se ordenan: SET OF(PubA, PubB) será idéntico a 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) ║ │ │ ╚═════════════════════╝ │ │
Sin embargo, aún no hemos "demostrado" que hemos generados la misma clave compartida para esta sesión. En principio, se podría omitir este paso: el primer mensaje de transporte será inválido, pero queremos asegurarnos de que cuando se complete el apretón de manos, realmente todo esté acordado. Por ahora, tenemos en nuestras manos el protocolo ISO/IEC IS 9798-3.
Podríamos firmar la clave generada. Esto es peligroso, ya que no se puede descartar que en el algoritmo de firma utilizado puedan existir filtraciones (aunque sean bits-para-firma, siguen siendo filtraciones). Se puede firmar el hash de la clave generada, pero la filtración incluso de este hash puede ser valiosa en un ataque de fuerza bruta a la función de generación. SIGMA utiliza una función MAC que autentica el identificador del remitente.
┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └──┬──┘ │ IdA, PubA │ ╔═══════════════════════════╗ │─────────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚═══════════════════════════╝ │IdB, PubB, sign(SignPrvB, (PubA, PubB)), MAC(IdB) │ ╔═══════════════════════════╗ ││ ║verify(Key, IdB) ║ │ │ ║verify(SignPubB, ...)║ │ │ ╚═════════════════════╝ │ │
Como optimización, algunos pueden querer reutilizar sus claves efímeras (lo cual, por supuesto, es perjudicial para PFS). Por ejemplo, generamos un par de claves, intentamos conectarnos, pero TCP no estaba disponible o se interrumpió en medio del protocolo. Es lamentable desperdiciar la entropía gastada y los recursos del procesador en un nuevo par. Por lo tanto, introduciremos un cookie, un valor pseudoaleatorio que protegerá contra posibles ataques de repetición al reutilizar claves públicas efímeras. Debido al enlace entre el cookie y la clave pública efímera, la clave pública de la otra parte se puede eliminar de la firma por innecesaria.
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Finalmente, queremos proteger la privacidad de nuestros identificadores ante un observador pasivo. Para ello, SIGMA propone primero intercambiar claves efímeras, generar una clave común con la que se cifren los mensajes de autenticación e identificación. SIGMA describe dos variantes:
- SIGMA-I — protege al iniciador de ataques activos, y al respondedor de ataques pasivos: el iniciador autentica al respondedor y, si hay algún desacuerdo, no revela su identificación. El respondedor, por su parte, proporciona su identificación solo si se inicia un protocolo activo. Un observador pasivo no obtendrá información alguna;
SIGMA-R — protege al respondedor de ataques activos, y al iniciador de ataques pasivos. Todo es a la inversa, pero en este protocolo se transmiten cuatro mensajes de apretón de manos.Elegimos SIGMA-I como una opción más alineada con lo que esperamos de las cosas habituales cliente-servidor: el cliente solo es conocido por el servidor autenticado, mientras que el servidor ya conoce todo. Además, es más sencillo de implementar debido a la menor cantidad de mensajes de apretón de manos. Todo lo que incorporamos al protocolo es el cifrado de parte del mensaje y la transferencia del identificador A a la parte cifrada del último mensaje:
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
- Se utiliza GOST R para la firma algoritmo con claves de 256 bits.
- Para la generación de la clave compartida se utiliza 34.10-2012 VKO.
- Se utiliza CMAC como MAC. Técnicamente, este es un modo especial de operación del cifrador de bloques, descrito en GOST R 34.13-2015. Como función de cifrado para este modo — (34.12-2015).
- Como identificador del interlocutor se utiliza el hash de su clave pública. Se aplica el hash (34.11-2012 256 bits).
Después del apretón de manos tendremos una clave compartida acordada. Podemos utilizarla para el cifrado autenticado de mensajes en tránsito. Esta parte es bastante simple y es difícil equivocarse: incrementamos el contador de mensajes, ciframos el mensaje, autenticamos (MAC) el contador y el texto cifrado, y lo enviamos. Al recibir un mensaje, verificamos que el contador tiene el valor esperado, autenticamos el texto cifrado con el contador y lo desciframos. ¿Con qué clave ciframos los mensajes de apretón de manos y de transporte, y qué clave utilizamos para la autenticación? Usar una sola clave para todas estas tareas es peligroso y poco razonable. Es necesario generar claves utilizando funciones especializadas (función de derivación de clave). Nuevamente, no vamos a complicarlo ni a inventar nada: es bien conocida, ha sido bien investigada y no tiene problemas conocidos. Desafortunadamente, en la biblioteca nativa de Python no existe esta función, por lo que utilizamos paquete. HKDF utiliza internamente , que a su vez utiliza la función hash. Un ejemplo de implementación en Python en la página de Wikipedia ocupa unas pocas líneas de código. Al igual que con 34.10-2012, utilizaremos Stribog-256 como función hash. La salida de nuestra función de acuerdo de claves se llamará clave de sesión, a partir de la cual se generarán las simétricas que faltan:
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")Estructuras/esquemas
Veamos qué estructuras ASN.1 hemos obtenido para la transmisión de todos estos datos:
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 — lo que será firmado (to be signed). HandshakeTBE — lo que será cifrado (to be encrypted). Llamo la atención sobre el campo ukm en MsgHandshake1. 34.10 VKO, para una mayor aleatorización de las claves generadas, incluye el parámetro UKM (user keying material) — simplemente entropía adicional.
Agregar criptografía al código
Consideremos solo los cambios realizados al código original, ya que la estructura se mantuvo igual (de hecho, primero se escribió la implementación final y luego se eliminó toda la criptografía).
Dado que la autenticación e identificación de los interlocutores se llevarán a cabo mediante claves públicas, ahora deben almacenarse de forma duradera en algún lugar. Para simplificar, utilizamos JSON de este tipo:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }our — nuestro par de claves, claves privadas y públicas en hexadecimal. their — nombres de los interlocutores y sus claves públicas. Cambiaremos los argumentos de la línea de comandos y agregaremos un procesamiento posterior de los datos JSON:
from pygost import gost3410 from pygost.gost34112012256 import GOST34112012256 CURVA = gost3410.GOST3410Curve( *gost3410.CURVE_PARAMS["GostR3410_2001_CryptoPro_A_ParamSet"] ) parser = argparse.ArgumentParser(description="GOSTIM") parser.add_argument( "--keys-gen", action="store_true", help="Generar JSON con nuestro nuevo par de claves", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON con nuestras y sus claves", ) parser.add_argument( "--bind", default="::1", help="Dirección para escuchar", ) parser.add_argument( "--port", type=int, default=6666, help="Puerto para escuchar", ) args = parser.parse_args() if args.keys_gen: prv_raw = urandom(32) pub = gost3410.public_key(CURVA, gost3410.prv_unmarshal(prv_raw)) pub_raw = gost3410.pub_marshal(pub) print(json.dumps({ "nuestro": {"prv": hexenc(prv_raw), "pub": hexenc(pub_raw)}, "suyo": {}, })) exit(0) # Analizar y deserializar nuestras y sus claves {{{ with open(args.keys, "rb") as fd: _keys = json.loads(fd.read().decode("utf-8")) KEY_NUESTRO_FIRMA_PRV = gost3410.prv_unmarshal(hexdec(_keys["nuestro"]["prv"])) _pub = hexdec(_keys["nuestro"]["pub"]) KEY_NUESTRO_FIRMA_PUB = gost3410.pub_unmarshal(_pub) KEY_NUESTRO_FIRMA_PUB_HASH = OctetString(GOST34112012256(_pub).digest()) for peer_name, pub_raw in _keys["suyo"].items(): _pub = hexdec(pub_raw) KEYS[GOST34112012256(_pub).digest()] = { "nombre": peer_name, "pub": gost3410.pub_unmarshal(_pub), } # }}}La clave privada del algoritmo 34.10 es un número aleatorio. Tiene un tamaño de 256 bits para curvas elípticas de 256 bits. PyGOST no trabaja con un conjunto de bytes, sino con , por lo que nuestra clave privada (urandom(32)) debe ser convertida en un número utilizando gost3410.prv_unmarshal(). La clave pública se calcula determinísticamente a partir de la privada, usando gost3410.public_key(). La clave pública 34.10 consiste en dos números grandes, que también deben ser convertidos a una secuencia de bytes para facilitar su almacenamiento y transmisión, utilizando gost3410.pub_marshal().
Después de leer el archivo JSON, las claves públicas, en consecuencia, deben ser convertidas de nuevo, usando gost3410.pub_unmarshal(). Dado que recibiremos identificadores de los interlocutores en forma de hash de la clave pública, podemos calcularlos de antemano y almacenarlos en un diccionario para una búsqueda rápida. El hash Streebog-256 es gost34112012256.GOST34112012256(), que satisface completamente la interfaz de funciones hash de hashlib.
¿Cómo ha cambiado la corutina del iniciador? Todo sigue el esquema de apretón de manos: generamos una cookie (128 bits son más que suficientes), un par de claves efímeras 34.10 que se utilizará para la función VKO de acuerdo con el protocolo de intercambio de claves.
395 async def initiador(host, port): 396 _id = repr((host, port)) 397 logging.info("%s: marcando", _id) 398 lector, escritor = await asyncio.open_connection(host, port) 399 # Generar nuestra clave pública efímera y cookie, enviar mensaje de Handshake 0 {{{ 400 cookie_nuestra = Cookie(urandom(16)) 401 prv = gost3410.prv_unmarshal(urandom(32)) 402 pub_nuestra = gost3410.public_key(CURVE, prv) 403 pub_nuestra_cruda = PubKey(gost3410.pub_marshal(pub_nuestra)) 404 escritor.write(Msg(("handshake0", MsgHandshake0(( 405 ("cookieIniciador", cookie_nuestra), 406 ("pubKeyIniciador", pub_nuestra_cruda), 407 )))).encode()) 408 # }}} 409 await escritor.drain()- esperamos la respuesta y decodificamos el mensaje Msg recibido;
- aseguramos que hemos recibido handshake1;
- decodificamos la clave pública efímera de la otra parte y calculamos la clave de sesión;
- generamos las claves simétricas necesarias para procesar la parte TBE del mensaje.
423 logging.info("%s: obtuvo %s mensaje", _id, msg.choice) 424 if msg.choice != "handshake1": 425 logging.warning("%s: mensaje inesperado, desconectando", _id) 426 escritor.close() 427 return 428 # }}} 429 msg_handshake1 = msg.value 430 # Validar mensaje de Handshake {{{ 431 cookie_suya = msg_handshake1["cookieRespondedor"] 432 pub_suya_cruda = msg_handshake1["pubKeyRespondedor"] 433 pub_suya = gost3410.pub_unmarshal(bytes(pub_suya_cruda)) 434 ukm_crudo = bytes(msg_handshake1["ukm"]) 435 ukm = ukm_unmarshal(ukm_crudo) 436 key_session = kek_34102012256(CURVE, prv, pub_suya, ukm, mode=2001) 437 kdf = Hkdf(None, key_session, hash=GOST34112012256) 438 key_handshake1_mac_identidad = kdf.expand(b"handshake1-mac-identidad") 439 key_handshake1_enc = kdf.expand(b"handshake1-enc") 440 key_handshake1_mac = kdf.expand(b"handshake1-mac")UKM es un número de 64 bits (urandom(8)), que también requiere deserialización desde la representación de bytes, utilizando gost3410_vko.ukm_unmarshal(). La función VKO para 34.10-2012 de 256 bits es gost3410_vko.kek_34102012256() (KEK — clave de cifrado).
La clave de sesión generada ya es una secuencia pseudoaleatoria de bytes de 256 bits. Por lo tanto, se puede usar inmediatamente en la función HKDF. Dado que GOST34112012256 cumple con la interfaz hashlib, se puede usar directamente en la clase Hkdf. No especificamos la sal (primer argumento de Hkdf), ya que la clave generada, debido a la efímera naturaleza de los pares de claves involucrados, será diferente para cada sesión y ya tiene suficiente entropía. kdf.expand() por defecto ya genera claves de 256 bits necesarias para el próximo uso en el Cucaracha.
A continuación, se verifican las partes TBE y TBS del mensaje recibido:
- se calcula y verifica el MAC sobre el texto cifrado recibido;
- se descifra el texto cifrado;
- se decodifica la estructura TBE;
- de ella se toma el identificador del interlocutor y se verifica si nos es conocido;
- se calcula y verifica el MAC sobre este identificador;
- se verifica la firma sobre la estructura TBS, que incluye cookies de ambas partes y la clave pública efímera de la parte opuesta. La firma se verifica con la clave de firma de largo plazo del interlocutor.
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, desconectando", _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 inválido") 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("no se puede decodificar 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("identidad desconocida") 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 de identidad inválido") 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("firma inválida") 174 return peer["name"]Como se mencionó anteriormente, 34.13-2015 describe varios de 34.12-2015. Entre ellos se encuentra el modo de generación de inserciones de MAC. En PyGOST, esto es gost3413.mac(). Este modo requiere la transmisión de una función de cifrado (que toma y devuelve un bloque de datos), el tamaño del bloque de cifrado y, propiamente dicho, los datos mismos. ¿Por qué no se puede definir de manera fija el tamaño del bloque de cifrado? 34.12-2015 describe no solo el cifrado de 128 bits Kuznechik, sino también el de 64 bits — una versión ligeramente modificada del GOST 28147-89, creado en la KGB y que aún posee uno de los niveles de seguridad más altos.
Grasshopper se inicializa con gost.3412.GOST3412Kuznechik(key) llamando y devuelve un objeto con los métodos .encrypt()/.decrypt() adecuados para la función 34.13. El MAC se calcula de la siguiente manera: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Para comparar el MAC calculado y el recibido, no se puede utilizar una comparación simple (==) de cadenas de bytes, ya que esta operación provoca fugas de tiempo en la comparación, lo que, en general, puede llevar a vulnerabilidades fatales de tipo ataques a TLS. En Python hay una función especial hmac.compare_digest para esto.
La función de cifrado por bloques solo puede cifrar un único bloque de datos. Para un mayor número, y además de longitud no múltiple, es necesario usar un modo de cifrado. En 34.13-2015 se describen los siguientes: ECB, CTR, OFB, CBC, CFB. Cada uno tiene sus ámbitos de aplicación y características permitidas. Con gran pesar, todavía no tenemos modos de cifrado autenticados el modo contador Al igual que .mac(), .ctr() acepta datos de entrada similares: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). Se requiere establecer un vector de inicialización, que tenga exactamente la mitad del tamaño del bloque de cifrado. Si nuestra clave de cifrado se utiliza solo para cifrar un mensaje (aunque sea de varios bloques), es seguro establecer un vector de inicialización nulo. Para cifrar los mensajes de handshake, usamos cada vez una clave separada.
La verificación de la firma gost3410.verify() es trivial: se pasa la curva elíptica dentro de la cual trabajamos (que simplemente fijamos en nuestro protocolo GOSTIM), la clave pública del firmante (no olvidemos que debe ser una tupla de dos grandes números, no una cadena de bytes), el hash 34.11-2012 y la firma recibida en sí.
Luego, en el iniciador preparamos y enviamos el mensaje handshake2 de apretón de manos, realizando las mismas acciones que hicimos durante la verificación, solo que simétricamente: firma en nuestras claves en lugar de verificación, etc.
A continuación, en el iniciador, preparamos y enviamos el mensaje handshake2 de la negociación, realizando las mismas acciones que llevamos a cabo durante la verificación, solo que simétricamente: firmando con nuestras claves en lugar de verificar, etc.
456 # Preparar y enviar el mensaje 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: sesión establecida: %s", _id, peer_name)Una vez establecida la sesión, se generan las claves de transporte (una clave separada para la encriptación, otra para la autenticación, para cada una de las partes), y se inicializa Kuznechik para la desencriptación y verificación de MAC:
499 # Ejecutar el envío de mensajes de texto, inicializar el decodificador de transporte {{{ 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 # Esperar mensajes de prueba {{{ 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 corutina msg_sender ahora cifra los mensajes antes de enviarlos por la conexión TCP. Cada mensaje tiene un nonce que aumenta monotonamente, que también actúa como vector de inicialización al cifrar en modo contador. Cada mensaje y bloque de mensajes tendrá valores de contador que serán garantizados como diferentes.
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 += 1Los mensajes entrantes son procesados por la corutina msg_receiver, que se encarga de la autenticación y cifra:
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("valor nonce inesperado") mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode()) if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])): raise ValueError("MAC inválido") plaintext = ctr( encrypter, KUZNECHIK_BLOCKSIZE, bytes(payload["ciphertext"]), long2bytes(nonce_expected, 8), ) text = plaintext.decode("utf-8") await OUT_QUEUES[peer_name].put(text)Conclusión
GOSTIM está destinado exclusivamente a fines educativos (ya que no está cubierto por pruebas, al menos) ¡El código fuente del programa se puede descargar! (Hash Stribog-256: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Al igual que todos mis proyectos, tipo , , , , GOSTIM es completamente , distribuido bajo los términos de .
, , miembro , desarrollador de Python/Go, especialista principal .
Fuente: habr.com
