Sendo um desenvolvedor da biblioteca (primitivos criptográficos de acordo com o GOST em Python puro), frequentemente recebo perguntas sobre como implementar uma troca segura de mensagens de forma simples. Muitos acreditam que a criptografia aplicada é algo bastante simples, e apenas a chamada .encrypt() de um cifrador de bloco é suficiente para o envio seguro através de um canal de comunicação. Outros acreditam que a criptografia aplicada é um domínio de poucos, e é aceitável que empresas ricas como Telegram, com matemáticos olímpicos, um protocolo seguro.
Tudo isso me motivou a escrever este artigo, para mostrar que a implementação de protocolos criptográficos e de IM seguro não é uma tarefa tão complicada. No entanto, não vale a pena inventar seus próprios protocolos de autenticação e troca de chaves.
O artigo abordará , , com protocolo de autenticação e troca de chaves (no qual é baseado ), usando exclusivamente algoritmos criptográficos do GOST da biblioteca PyGOST e codificação ASN.1 das mensagens pela biblioteca (sobre a qual eu já ). A condição necessária: ele deve ser simples o suficiente para que possa ser escrito do zero em uma noite (ou dia de trabalho), caso contrário, já não é um programa simples. Haverá certamente erros, complexidades desnecessárias e falhas, além disso, este é o meu primeiro programa utilizando a biblioteca asyncio.
Design do IM
Para começar, precisamos entender como será o nosso IM. Para simplificar, vamos supor que será uma rede peer-to-peer, sem qualquer descoberta de participantes. Faremos a conexão manualmente ao endereço e porta desejados para comunicação com o interlocutor.
Eu entendo que, neste momento, a suposição de que existe uma conexão direta entre dois computadores arbitrários é uma limitação significativa para a aplicabilidade do IM na prática. Mas quanto mais desenvolvedores implementarem soluções de NAT-traversal, mais tempo permaneceremos na Internet IPv4, com a desanimadora probabilidade de conexão entre computadores aleatórios. Até quando poderemos suportar a falta de IPv6 em casa e no trabalho?
Teremos uma rede amigo-a-amigo: todos os possíveis interlocutores devem ser conhecidos de antemão. Primeiro, isso simplifica muito tudo: apresentamo-nos, encontramos ou não o nome/chave, desconectamos ou continuamos a trabalhar, sabendo quem é o interlocutor. Em segundo lugar, em geral, isso é seguro e exclui várias ataques.
A interface IM será semelhante às soluções clássicas , que eu gosto muito pelo seu minimalismo e pela filosofia Unix-way. O programa IM cria um diretório para cada interlocutor com três soquetes de domínio Unix:
- in — para onde são registradas as mensagens enviadas ao interlocutor;
- out — de onde são lidas as mensagens recebidas do interlocutor;
- state — ao ler dele, sabemos se o interlocutor está conectado atualmente, endereço/porta de conexão.
Além disso, um soquete conn é criado, escrevendo nele o host e a porta, iniciamos a conexão com o interlocutor remoto.
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Essa abordagem permite criar implementações independentes do transporte IM e da interface do usuário, pois sobre gosto e cor não há discussão, não se pode agradar a todos. Usando e/ou , é possível obter uma interface de múltiplas janelas com destaque de sintaxe. E com o é possível obter uma linha de entrada compatível com GNU Readline para mensagens.
Na verdade, projetos suckless usam arquivos FIFO. Pessoalmente, eu não consegui entender como trabalhar com arquivos de forma concorrente no asyncio sem um suporte manual de threads dedicadas (para essas coisas, há tempos uso a linguagem ). Portanto, decidi usar soquetes de domínio Unix. Infelizmente, isso impossibilita fazer echo 2001:470:dead::babe 6666 > conn. Resolvi esse problema, usando : echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
Protocolo inicial não seguro
O transporte utilizado é o TCP: ele garante a entrega e sua ordem. O UDP não garante nenhum dos dois (o que seria útil quando a criptografia for aplicada), e o suporte não está disponível no Python por padrão.
Infelizmente, no TCP não existe o conceito de mensagem, apenas um fluxo de bytes. Portanto, é necessário inventar um formato para as mensagens, de modo que possam ser separadas nesse fluxo. Podemos concordar em usar o caractere de nova linha. Para começar, isso funciona, no entanto, quando começarmos a criptografar nossas mensagens, esse caractere pode aparecer em qualquer lugar no texto cifrado. Por isso, em redes, são populares os protocolos que enviam primeiro o comprimento da mensagem em bytes. Por exemplo, em Python, existe a biblioteca xdrlib que permite trabalhar com esse formato. .
Não vamos trabalhar de forma correta e eficiente com a leitura de TCP — vamos simplificar o código. Lemos em um loop infinito os dados do socket até decodificarmos a mensagem completa. Como formato para essa abordagem, podemos usar tanto JSON quanto XML. Mas, quando a criptografia for adicionada, os dados precisarão ser assinados e autenticados — e isso exigirá uma representação idêntica byte a byte dos objetos, o que JSON/XML não garantem (o resultado de dumps pode ser diferente).
O XDR se adequa a essa tarefa, porém eu escolho ASN.1 com codificação DER e a biblioteca, pois teremos objetos de alto nível com os quais muitas vezes é mais agradável e conveniente trabalhar. Ao contrário do schemaless , ou , o ASN.1 verificará automaticamente os dados em relação a um esquema rigidamente definido.
# 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))),
))
A mensagem recebida será Msg: ou MsgText de texto (por enquanto com um único campo de texto), ou uma mensagem de handshake MsgHandshake (na qual é transmitido o nome do interlocutor). Agora parece excessivamente complicado, mas isso é um investimento para o futuro.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│MsgHandshake(IdA) │
│─────────────────>│
│ │
│MsgHandshake(IdB) │
││
│ │
│ MsgText() │
│<─────────────────│
│ │
IM sem criptografia
Como já mencionei, para todas as operações com sockets, será utilizada a biblioteca asyncio. Vamos declarar o que estamos esperando no momento da execução:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Nosso nome de par",
)
parser.add_argument(
"--their-names",
required=True,
help="Nomes dos pares deles, separados por vírgula",
)
parser.add_argument(
"--bind",
default="::1",
help="Endereço para escutar",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Porta para escutar",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
Define um nome próprio (—our-name alice). Os nomes de todos os pares esperados são listados, separados por vírgula (—their-names bob,eve). Para cada um dos pares, um diretório é criado com sockets Unix, bem como uma corrotina 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"))
As mensagens recebidas do usuário no socket in são enviadas para as filas 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"))
As mensagens recebidas dos pares são enviadas para as filas OUT_QUEUES, das quais os dados são gravados no 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()
Ao ler do socket state, o programa procura no dicionário PEER_ALIVE o endereço do par. Se ainda não houver conexão com o par, uma string vazia é gravada.
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()
Ao gravar o endereço no socket conn, a função "iniciadora" da conexão é acionada:
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))
Vamos considerar a iniciativa. Primeiro, ele abre a conexão com o host/porta especificados e envia uma mensagem de handshake com seu nome:
130 async def initiator(host, port):
131 _id = repr((host, port))
132 logging.info("%s: discando", _id)
133 reader, writer = await asyncio.open_connection(host, port)
134 # Mensagem de handshake {{{
135 writer.write(Msg(("handshake", MsgHandshake((
136 ("peerName", OUR_NAME),
137 )))).encode())
138 # }}}
139 await writer.drain()
Em seguida, aguarda uma resposta da parte remota. Tenta decodificar a resposta recebida de acordo com o esquema Msg ASN.1. Supomos que toda a mensagem será enviada em um único segmento TCP e a receberemos atomicamente ao chamar .read(). Verificamos se recebemos de fato a mensagem de handshake.
141 # Esperar pela mensagem de Handshake {{{
142 data = await reader.read(256)
143 if data == b"":
144 logging.warning("%s: sem resposta, desconectando", _id)
145 writer.close()
146 return
147 try:
148 msg, _ = Msg().decode(data)
149 except ASN1Error:
150 logging.warning("%s: resposta indecifrável, desconectando", _id)
151 writer.close()
152 return
153 logging.info("%s: recebeu a mensagem %s", _id, msg.choice)
154 if msg.choice != "handshake":
155 logging.warning("%s: mensagem inesperada, desconectando", _id)
156 writer.close()
157 return
158 # }}}
Verificamos se o nome do interlocutor recebido nos é conhecido. Se não, encerramos a conexão. Verificamos se já estabelecemos uma conexão com ele (o interlocutor deu novamente a ordem para se conectar a nós) e a fechamos. Na fila IN_QUEUES, são colocadas strings Python com o texto da mensagem, mas há um valor especial None, indicando que a corrotina msg_sender deve parar, fazendo com que ela esqueça seu writer associado à conexão TCP obsoleta.
159 msg_handshake = msg.value
160 peer_name = str(msg_handshake["peerName"])
161 if peer_name not in THEIR_NAMES:
162 logging.warning("nome do peer desconhecido: %s", peer_name)
163 writer.close()
164 return
165 logging.info("%s: sessão estabelecida: %s", _id, peer_name)
166 # Enviar mensagem de texto, inicializar decodificador de transporte {{{
167 peer_alive = PEER_ALIVES.pop(peer_name, None)
168 if peer_alive is not None:
169 peer_alive.close()
170 await IN_QUEUES[peer_name].put(None)
171 PEER_ALIVES[peer_name] = writer
172 asyncio.ensure_future(msg_sender(peer_name, writer))
173 # }}}
msg_sender recebe mensagens de saída (colocadas na fila a partir do socket de entrada), as serializa em mensagens MsgText e as envia pela conexão TCP. Esta pode ser interrompida a qualquer momento — isso capturamos explicitamente.
async def msg_sender(peer_name: str, writer) -> None:
in_queue = IN_QUEUES[peer_name]
while True:
text = await in_queue.get()
if text is None:
break
writer.write(Msg(("text", MsgText((
("text", UTF8String(text)),
)))).encode())
try:
await writer.drain()
except ConnectionResetError:
del PEER_ALIVES[peer_name]
return
logging.info("%s: envio de mensagem de %d caracteres", peer_name, len(text))
No final, o iniciador entra em um ciclo infinito de leitura de mensagens do soquete. Ele verifica se são mensagens de texto e as coloca na fila OUT_QUEUES, de onde elas serão enviadas para o soquete de saída do correspondente. Por que não podemos simplesmente fazer .read() e decodificar a mensagem? Porque não se pode excluir a situação em que várias mensagens do usuário são agregadas no buffer do sistema operacional e enviadas em um único segmento TCP. Nós conseguiremos decodificar a primeira, mas no buffer pode permanecer parte da seguinte. Em qualquer situação de exceção, fechamos a conexão TCP e paramos a corrotina msg_sender (enviando None para a fila OUT_QUEUES).
174 buf = b""
175 # Aguarde por mensagens de teste {{{
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: tamanho máximo do buffer excedido", _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: mensagem %s inesperada", _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: mensagem de %d caracteres recebida", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Vamos voltar ao código principal. Após a criação de todas as corrotinas no momento do início do programa, iniciamos o servidor TCP. Para cada conexão estabelecida, ele cria uma corrotina 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("Ouvindo em: %s", server.sockets[0].getsockname())
loop.run_forever()
responder é semelhante ao iniciador e executa todas as mesmas ações de forma inversa, mas o ciclo infinito de leitura de mensagens é iniciado imediatamente, por simplicidade. Agora o protocolo de handshake envia uma mensagem de cada lado, mas, posteriormente, serão duas do iniciador da conexão, após as quais o envio de texto será imediatamente possível.
72 async def responder(reader, writer):
73 _id = writer.get_extra_info("peername")
74 logging.info("%s: connected", _id)
75 buf = b""
76 msg_expected = "handshake"
77 peer_name = None
78 while True:
79 # Read until we get Msg message {{{
80 data = await reader.read(MaxMsgLen)
81 if data == b"":
82 logging.info("%s: closed connection", _id)
83 break
84 buf += data
85 if len(buf) > MaxMsgLen:
86 logging.warning("%s: max buffer size exceeded", _id)
87 break
88 try:
89 msg, tail = Msg().decode(buf)
90 except ASN1Error:
91 continue
92 buf = tail
93 # }}}
94 if msg.choice != msg_expected:
95 logging.warning("%s: unexpected %s message", _id, msg.choice)
96 break
97 if msg_expected == "text":
98 try:
99 await msg_receiver(msg.value, peer_name)
100 except ValueError as err:
101 logging.warning("%s: %s", err)
102 break
103 # Process Handshake message {{{
104 elif msg_expected == "handshake":
105 logging.info("%s: got %s message", _id, msg_expected)
106 msg_handshake = msg.value
107 peer_name = str(msg_handshake["peerName"])
108 if peer_name not in THEIR_NAMES:
109 logging.warning("unknown peer name: %s", peer_name)
110 break
111 writer.write(Msg(("handshake", MsgHandshake((
112 ("peerName", OUR_NAME),
113 )))).encode())
114 await writer.drain()
115 logging.info("%s: session established: %s", _id, peer_name)
116 peer_alive = PEER_ALIVES.pop(peer_name, None)
117 if peer_alive is not None:
118 peer_alive.close()
119 await IN_QUEUES[peer_name].put(None)
120 PEER_ALIVES[peer_name] = writer
121 asyncio.ensure_future(msg_sender(peer_name, writer))
122 msg_expected = "text"
123 # }}}
124 logging.info("%s: disconnecting", _id)
125 if msg_expected == "text":
126 IN_QUEUES[peer_name].put(None)
127 writer.close()
Protocolo Seguro
É hora de garantir nossa comunicação. O que entendemos por segurança e o que desejamos:
- privacidade das mensagens transmitidas;
- autenticidade e integridade das mensagens transmitidas — qualquer alteração deve ser detectada;
- proteção contra ataques de repetição (replay attack) — a perda ou repetição de mensagens deve ser detectada (e decidimos encerrar a conexão);
- identificação e autenticação dos interlocutores com chaves públicas previamente definidas — decidimos anteriormente que faremos uma rede friend-to-friend. Apenas após a autenticação entenderemos com quem estamos conversando;
- a existência (PFS) — a comprometimento de nossa chave de assinatura de longa duração não deve permitir a leitura de toda a comunicação anterior. O registro do tráfego interceptado torna-se inútil;
- a validade das mensagens (transporte e handshake) é restrita a uma única sessão TCP. A inserção de mensagens corretamente assinadas/autenticadas de outra sessão (mesmo com o mesmo interlocutor) não deve ser possível;
- um observador passivo não deve ver nem identificadores de usuários, nem chaves públicas de longa duração transmitidas, nem seus hashes. Alguma anonimidade em relação ao observador passivo.
Surpreendentemente, esse mínimo é praticamente desejado por todos em qualquer protocolo de handshake, e muito pouco do elenco listado é finalmente cumprido para protocolos 'caseiros'. Portanto, não vamos inventar nada novo agora. Eu definitivamente recomendaria usar para a construção de protocolos, mas vamos escolher algo mais simples.
Os dois protocolos mais populares são:
- — um protocolo muito complexo com uma longa história de bugs, falhas, vulnerabilidades, má elaboração, complexidade e erros (embora isso se aplique pouco ao TLS 1.3). Mas não o consideramos devido à sua complexidade excessiva.
- com — não têm problemas criptográficos sérios, embora também não sejam simples. Se você ler sobre IKEv1 e IKEv2, verá que sua origem é , os protocolos ISO/IEC IS 9798-3 e SIGMA (SIGn-and-MAc) — suficientemente simples para implementação em uma noite.
O que torna o SIGMA, como o último estágio de desenvolvimento dos protocolos STS/ISO, bom? Ele atende a todos os nossos requisitos (incluindo o 'ocultamento' dos identificadores dos interlocutores), não tem problemas criptográficos conhecidos. É minimalista — a remoção de pelo menos um elemento da mensagem do protocolo tornará sua segurança comprometida.
Vamos passar do protocolo doméstico mais simples até o SIGMA. A operação básica que nos interessa é : uma função em que ambos os participantes obterão o mesmo valor que pode ser usado como chave simétrica. Sem entrar em detalhes: cada parte gera um par de chaves efêmeras (usadas apenas dentro de uma sessão), troca chaves públicas, chama a função de acordo, na qual ambas as partes fornecem sua chave privada e a chave pública do interlocutor.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│ IdA, PubA │ ╔════════════════════╗
│───────────────►│ ║PrvA, PubA = DHgen()║
│ │ ╚════════════════════╝
│ IdB, PubB │ ╔════════════════════╗
│◄───────────────│ ║PrvB, PubB = DHgen()║
│ │ ╚════════════════════╝
────┐ ╔═══════╧════════════╗
│ ║Key = DH(PrvA, PubB)║
◄────┘ ╚═══════╤════════════╝
│ │
│ │
Qualquer um pode se intrometer e substituir as chaves públicas pelas suas próprias — neste protocolo, não há autenticação das partes. Vamos adicionar uma assinatura com chaves de longa duração.
┌─────┐ ┌─────┐
│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) ║ │
│ ╚═════════════════════╝ │
│ │
Essa assinatura não é adequada, pois não está vinculada a uma sessão específica. Essas mensagens "servem" também para sessões com outros participantes. Todo o contexto deve ser assinado. Isso também obriga a adicionar o envio de uma nova mensagem de A.
Além disso, é crítico adicionar a assinatura e seu próprio identificador, pois, do contrário, podemos substituir IdXXX e re-assinar a mensagem com a chave de outro interlocutor conhecido. Para evitar , é necessário que os elementos sob a assinatura estejam localizados em lugares claramente definidos de acordo com seu significado: se A assina (PubA, PubB), então B deve assinar (PubB, PubA). Isso também ressalta a importância de escolher a estrutura e o formato dos dados serializados. Por exemplo, conjuntos na codificação ASN.1 DER são ordenados: 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) ║ │ │ ╚═════════════════════╝ │ │
No entanto, ainda não "provamos" que produzimos a mesma chave comum para esta sessão. Em princípio, poderíamos dispensar essa etapa — a primeira mensagem de transporte seria inválida, mas queremos que, quando o handshake estiver concluído, tenhamos certeza de que tudo está realmente concordado. No momento, temos em mãos o protocolo ISO/IEC IS 9798-3.
Poderíamos assinar a própria chave gerada. Isso é arriscado, pois não se pode descartar que o algoritmo de assinatura utilizado possa ter vazamentos (mesmo que apenas bits para a assinatura, ainda assim são vazamentos). É possível assinar o hash da chave gerada, mas o vazamento até mesmo do hash da chave gerada pode ter valor em um ataque de força bruta à função de geração. O SIGMA utiliza uma função MAC, autenticando a identificação do remetente.
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Como otimização, alguns podem querer reutilizar suas chaves efêmeras (o que, claro, é lamentável para PFS). Por exemplo, geramos um par de chaves, tentamos nos conectar, mas o TCP não estava disponível ou foi interrompido em algum lugar no meio do protocolo. É um desperdício gastar a entropia e os recursos da CPU em um novo par. Portanto, vamos introduzir o que chamamos de cookie — um valor pseudo-aleatório que protege contra possíveis ataques de replay ao reutilizar chaves públicas efêmeras. Devido à ligação entre o cookie e a chave pública efêmera, a chave pública do participante oposto pode ser removida da assinatura, pois não é mais necessária.
┌─────┐ ┌─────┐ │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 garantir a privacidade dos nossos identificadores de interlocutores contra a observação passiva. Para isso, a SIGMA sugere primeiro trocar chaves efêmeras, e então elaborar uma chave comum, com a qual serão criptografadas as mensagens de autenticação e identificação. A SIGMA descreve duas opções:
- SIGMA-I — protege o iniciador contra ataques ativos e o respondente contra passivos: o iniciador autentica o respondente e, se algo não coincidir, não revela sua identificação. O respondente, por sua vez, revela sua identificação se um protocolo ativo for iniciado com ele. Um observador passivo não descobrirá nada;
SIGMA-R — protege o respondente contra ataques ativos e o iniciador contra passivos. Tudo é preciso ao contrário, mas neste protocolo já são trocadas quatro mensagens de handshake.Escolhemos o SIGMA-I como mais semelhante ao que esperamos de interações cliente-servidor habituais: o cliente só conhece o servidor autenticado, enquanto o servidor já possui todas as informações. Além disso, é mais simples de implementar devido ao menor número de mensagens de handshake. Tudo que introduzimos no protocolo é a criptografia de parte da mensagem e a transferência do identificador A para a parte criptografada da última mensagem:
┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └──┬──┘ │ PubA, CookieA │ ╔═══════════════════════════╗ │─────────────────────────────────────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚═══════════════════════════╝ │PubB, CookieB, Enc((IdB, sign(SignPrvB, (CookieA, CookieB, PubB)), MAC(IdB))) │ ╔═══════════════════════════╗ │<─────────────────────────────────────────────────────────────────────────────│ ║SignPrvB, SignPubB = load()║ │ │ ║PrvB, PubB = DHgen() ║ │ │ ╚═══════════════════════════╝ │ │ ╔═════════════════════╗ │ Enc((IdA, sign(SignPrvA, (CookieB, CookieA, PubA)), MAC(IdA))) │ ║Key = DH(PrvA, PubB) ║ │─────────────────────────────────────────────────────────────────────────────>│ ║verify(Key, IdB) ║ │ │ ║verify(SignPubB, ...)║ │ │ ╚═════════════════════╝ │ │
- Para a assinatura, é utilizado o GOST R algoritmo com chaves de 256 bits.
- Para a geração da chave compartilhada, utiliza-se o 34.10-2012 VKO.
- Como MAC, usa-se o CMAC. Tecnicamente, este é um modo especial de funcionamento de um cifrador de bloco, descrito no GOST R 34.13-2015. Como função de criptografia para este modo — (34.12-2015).
- Como identificador do interlocutor, é utilizado um hash da sua chave pública. O hash utilizado é (34.11-2012 256 bits).
Após o handshake, teremos uma chave comum acordada. Podemos usar essa chave para a criptografia autenticada das mensagens de transporte. Esta parte é bem simples e é difícil errar: incrementamos o contador de mensagens, criptografamos a mensagem, autenticamos (MAC) o contador e o texto criptografado, enviamos. Ao receber a mensagem, verificamos se o contador tem o valor esperado, autenticamos o texto criptografado com o contador e descriptografamos. Qual chave usar para criptografar mensagens de handshake e de transporte, e qual usar para autenticar? Usar uma única chave para todas essas tarefas é arriscado e imprudente. É necessário gerar chaves usando funções especializadas (função de derivação de chave). Novamente, não vamos complicar nem inventar nada: é bem conhecida, bem pesquisada e não possui problemas conhecidos. Infelizmente, na biblioteca padrão do Python não há essa função, portanto, vamos usar o pacote. O HKDF utiliza internamente , que, por sua vez, utiliza uma função de hash. Um exemplo de implementação em Python na página da Wikipedia ocupa poucas linhas de código. Assim como no caso de 34.10-2012, vamos usar Streebog-256 como função de hash. A saída de nossa função de acordo de chaves será chamada de chave de sessão, a partir da qual serão derivadas as chaves simétricas necessárias:
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")Estruturas/esquemas
Vamos considerar quais estruturas ASN.1 resultaram para a transmissão de todos esses dados:
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 é o que será assinado (to be signed). HandshakeTBE é o que será criptografado (to be encrypted). Observe o campo ukm em MsgHandshake1. 34.10 VKO, para uma ainda maior aleatorização das chaves geradas, inclui o parâmetro UKM (user keying material) — apenas uma entropia adicional.
Adicionando criptografia ao código
Vamos considerar apenas as alterações feitas ao código original, uma vez que a estrutura permaneceu a mesma (na verdade, primeiro foi escrita a implementação final e depois dela foi removida toda a criptografia).
Como a autenticação e identificação dos parceiros serão realizadas através de chaves públicas, agora elas precisam ser armazenadas de forma duradoura. Para simplificar, usaremos um JSON deste tipo:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }our — nosso par de chaves, chaves privadas e públicas em hexadecimal. their — nomes dos parceiros e suas chaves públicas. Vamos alterar os argumentos da linha de comando e adicionar um pós-processamento dos dados 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="Gere JSON com nosso novo par de chaves", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON com nossas e suas chaves", ) parser.add_argument( "--bind", default="::1", help="Endereço para escutar", ) parser.add_argument( "--port", type=int, default=6666, help="Porta para escutar", ) 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) # Fazer o parse e deserializar nossas e suas chaves {{{ 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), } # }}}A chave privada do algoritmo 34.10 é um número aleatório. Com um tamanho de 256 bits para curvas elípticas de 256 bits. O PyGOST trabalha não com um conjunto de bytes, mas com , portanto, nossa chave privada (urandom(32)) deve ser convertida em um número usando gost3410.prv_unmarshal(). A chave pública é calculada deterministicamente a partir da chave privada, usando gost3410.public_key(). A chave pública 34.10 consiste em dois grandes números, que também precisam ser convertidos em uma sequência de bytes para facilitar o armazenamento e a transmissão, usando gost3410.pub_marshal().
Após ler o arquivo JSON, as chaves públicas devem ser convertidas de volta, usando gost3410.pub_unmarshal(). Como receberemos os identificadores dos interlocutores na forma de hash da chave pública, podemos calculá-los antecipadamente e armazená-los em um dicionário para busca rápida. O hash Streebog-256 é gost34112012256.GOST34112012256(), totalmente compatível com a interface hashlib de funções de hash.
Como a corrotina do iniciador mudou? Tudo está como no esquema de handshake: geramos um cookie (128 bits é mais do que suficiente), um par de chaves efêmeras 34.10, que será utilizado para a função VKO de troca de chaves.
395 async def initiador(host, port): 396 _id = repr((host, port)) 397 logging.info("%s: discando", _id) 398 reader, writer = await asyncio.open_connection(host, port) 399 # Gere nossa chave pública efêmera e cookie, envie a mensagem Handshake 0 {{{ 400 cookie_nosso = Cookie(urandom(16)) 401 prv = gost3410.prv_unmarshal(urandom(32)) 402 pub_nosso = gost3410.public_key(CURVE, prv) 403 pub_nosso_raw = PubKey(gost3410.pub_marshal(pub_nosso)) 404 writer.write(Msg(("handshake0", MsgHandshake0(( 405 ("cookieIniciador", cookie_nosso), 406 ("pubKeyIniciador", pub_nosso_raw), 407 )))).encode()) 408 # }}} 409 await writer.drain()- aguardando a resposta e decodificando a mensagem Msg recebida;
- assegurando que recebemos handshake1;
- decodificando a chave pública efêmera do lado oposto e calculando a chave de sessão;
- gerando as chaves simétricas necessárias para processar a parte TBE da mensagem.
423 logging.info("%s: recebeu %s mensagem", _id, msg.choice) 424 if msg.choice != "handshake1": 425 logging.warning("%s: mensagem inesperada, desconectando", _id) 426 writer.close() 427 return 428 # }}} 429 msg_handshake1 = msg.value 430 # Validar mensagem Handshake {{{ 431 cookie_deles = msg_handshake1["cookieRespondente"] 432 pub_deles_raw = msg_handshake1["pubKeyRespondente"] 433 pub_deles = gost3410.pub_unmarshal(bytes(pub_deles_raw)) 434 ukm_raw = bytes(msg_handshake1["ukm"]) 435 ukm = ukm_unmarshal(ukm_raw) 436 key_sessao = kek_34102012256(CURVE, prv, pub_deles, ukm, mode=2001) 437 kdf = Hkdf(None, key_sessao, hash=GOST34112012256) 438 key_handshake1_mac_identidade = kdf.expand(b"handshake1-mac-identidade") 439 key_handshake1_enc = kdf.expand(b"handshake1-enc") 440 key_handshake1_mac = kdf.expand(b"handshake1-mac")UKM é um número de 64 bits (urandom(8)), que também requer desserialização de sua representação em bytes, usando gost3410_vko.ukm_unmarshal(). A função VKO para 34.10-2012 de 256 bits é gost3410_vko.kek_34102012256() (KEK — chave de criptografia).
A chave de sessão gerada já é uma sequência pseudoaleatória de bytes de 256 bits. Por isso, ela pode ser usada imediatamente na função HKDF. Como GOST34112012256 atende à interface hashlib, ela pode ser usada diretamente na classe Hkdf. Não especificamos o sal (primeiro argumento Hkdf), pois a chave gerada, devido à efemeridade das chaves públicas envolvidas, será diferente para cada sessão e já possui entropia suficiente. kdf.expand() por padrão já fornece chaves de 256 bits, necessárias para o Kucher em seguida.
Em seguida, verificam-se as partes TBE e TBS da mensagem recebida:
- calculando e verificando o MAC sobre o texto cifrado recebido;
- decifrando o texto cifrado;
- decodificando a estrutura TBE;
- dela é extraído o identificador do interlocutor e verificado se já o conhecemos;
- calculando e verificando o MAC sobre esse identificador;
- a assinatura é verificada sobre a estrutura TBS, que inclui cookies de ambas as partes e a chave efêmera pública da parte oposta. A assinatura é verificada com a chave de assinatura de longo prazo do 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("não é possível 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("identidade desconhecida") 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 identidade 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("assinatura inválida") 174 return peer["name"]Como já mencionado anteriormente, a norma 34.13-2015 descreve diferentes da norma 34.12-2015. Entre eles, há o modo de geração de autenticação, o cálculo do MAC. No PyGOST, isso é feito com gost3413.mac(). Este modo requer a transmissão de uma função de criptografia (que aceita e retorna um bloco de dados), o tamanho do bloco de criptografia e, efetivamente, os próprios dados. Por que não se pode definir o tamanho do bloco de criptografia de forma fixa? A norma 34.12-2015 descreve não apenas a cifra Kuznechik de 128 bits, mas também a de 64 bits. — uma versão levemente modificada do GOST 28147-89, criada ainda na KGB e que ainda possui um dos mais altos padrões de segurança.
O Kuznechik é inicializado usando gost.3412.GOST3412Kuznechik(key) e retorna um objeto com métodos .encrypt()/.decrypt() adequados para passar para as funções 34.13. O MAC é calculado da seguinte forma: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Para comparar o MAC calculado com o recebido, não se pode usar a comparação comum (==) de strings de bytes, pois essa operação causa vazamentos de tempo de comparação, o que, em geral, pode levar a vulnerabilidades fatais do tipo ataques ao TLS. No Python, existe uma função especial hmac.compare_digest para isso.
A função de cifra de bloco pode criptografar apenas um bloco de dados. Para quantidades maiores, especialmente se não forem múltiplas desse tamanho, é necessário usar um modo de criptografia. Na norma 34.13-2015, estão descritos os seguintes modos: ECB, CTR, OFB, CBC, CFB. Cada um tem suas áreas de aplicação permitidas e características. Para nosso grande pesar, ainda não temos modos de criptografia autenticada o modo contador Assim como .mac(), .ctr() aceita dados de entrada semelhantes: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). É necessário definir um vetor de inicialização, com comprimento exatamente igual à metade do tamanho do bloco. Se nossa chave de criptografia for usada apenas para criptografar uma mensagem (mesmo que seja de vários blocos), é seguro definir um vetor de inicialização nulo. Para a criptografia das mensagens de handshake, usamos a cada vez uma chave separada.
A verificação da assinatura gost3410.verify() é trivial: enviamos a curva elíptica dentro da qual estamos operando (que simplesmente fixamos em nosso protocolo GOSTIM), a chave pública do signatário (lembrando que deve ser uma tupla de dois grandes números, e não uma string de bytes), o hash 34.11-2012 e a própria assinatura recebida.
Em seguida, no iniciador, preparamos e enviamos a mensagem handshake2 de handshake, realizando as mesmas ações que fizemos na verificação, apenas simetricamente: assinatura com nossas chaves em vez de verificação, etc...
Em seguida, no iniciador, preparamos e enviamos a mensagem handshake2 do handshake, realizando as mesmas ações que fizemos na verificação, apenas de forma simétrica: assinatura com nossas chaves em vez de verificação, e assim por diante...
456 # Preparar e enviar mensagem de handshake 2 {{{ 457 tbs = HandshakeTBS(( 458 ("cookieTheir", cookie_their), 459 ("cookieOur", cookie_our), 460 ("pubKeyOur", pub_our_raw), 461 )) 462 assinatura = gost3410.sign( 463 CURVE, 464 KEY_OUR_SIGN_PRV, 465 GOST34112012256(tbs.encode()).digest(), 466 ) 467 chave_handshake2_mac_identity = kdf.expand(b"handshake2-mac-identity") 468 mac_tag = mac( 469 GOST3412Kuznechik(chave_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(assinatura)), 476 ("identityMac", MAC(mac_tag)), 477 )) 478 tbe_raw = tbe.encode() 479 chave_handshake2_enc = kdf.expand(b"handshake2-enc") 480 chave_handshake2_mac = kdf.expand(b"handshake2-mac") 481 ciphertext = ctr( 482 GOST3412Kuznechik(chave_handshake2_enc).encrypt, 483 KUZNECHIK_BLOCKSIZE, 484 tbe_raw, 485 8 * b"x00", 486 ) 487 mac_tag = mac( 488 GOST3412Kuznechik(chave_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: sessão estabelecida: %s", _id, peer_name)Quando a sessão é estabelecida, as chaves de transporte são geradas (uma chave separada para criptografia, uma para autenticação, para cada uma das partes), e o Kuznechik é inicializado para descriptografar e verificar o MAC:
499 # Executar o remetente de mensagens de texto, inicializar o decodificador de transporte {{{ 500 chave_iniciador_enc = kdf.expand(b"transport-initiator-enc") 501 chave_iniciador_mac = kdf.expand(b"transport-initiator-mac") 502 chave_responder_enc = kdf.expand(b"transport-responder-enc") 503 chave_responder_mac = kdf.expand(b"transport-responder-mac") ... 509 asyncio.ensure_future(msg_sender( 510 peer_name, 511 chave_iniciador_enc, 512 chave_iniciador_mac, 513 writer, 514 )) 515 encrypter = GOST3412Kuznechik(chave_responder_enc).encrypt 516 macer = GOST3412Kuznechik(chave_responder_mac).encrypt 517 # }}} 519 nonce_esperado = 0 520 # Aguardar por mensagens de teste {{{ 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_esperado, 541 macer, 542 encrypter, 543 peer_name, 544 ) 545 except ValueError as err: 546 logging.warning("%s: %s", err) 547 break 548 nonce_esperado += 1 549 # }}}A corrotina msg_sender agora criptografa mensagens antes de enviá-las na conexão TCP. Cada mensagem possui um nonce que aumenta monotonamente, que também serve como vetor de inicialização na criptografia em modo contador. Garantido que cada mensagem e bloco de mensagem terão valores de contador 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 += 1As mensagens recebidas são processadas pela corrotina msg_receiver, que se ocupa da autenticação e descriptografia:
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)Conclusão
O GOSTIM destina-se exclusivamente a fins educacionais (uma vez que não é coberto por testes, pelo menos)! O código-fonte do programa pode ser baixado (hash Stribog-256: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Assim como todos os meus projetos, do tipo , , , , o GOSTIM é completamente , distribuído sob os termos da .
, , membro , desenvolvedor Python/Go, especialista chefe .
Fonte: habr.com
