Essendo uno sviluppatore libreria (primitivi crittografici GOST in puro Python), ricevo spesso domande su come implementare una semplice comunicazione sicura. Molti pensano che la crittografia applicativa sia piuttosto semplice, e che una chiamata a .encrypt() di un cifrario a blocchi sia sufficiente per un invio sicuro attraverso un canale di comunicazione. Altri, invece, credono che la crittografia applicativa sia riservata a pochi, e va bene che aziende ricche come Telegram, con matematici olimpionici un protocollo sicuro.
Tutto ciò mi ha spinto a scrivere questo articolo, per dimostrare che l'implementazione di protocolli crittografici e di un IM sicuro non è un compito così complicato. Tuttavia, non vale la pena inventare protocolli di autenticazione e di scambio di chiavi propri.
Nell'articolo verrà descritto , , instant messenger con protocollo di autenticazione e scambio di chiavi (sulla base del quale è realizzato ), utilizzando esclusivamente algoritmi crittografici GOST della libreria PyGOST e codifica ASN.1 dei messaggi tramite la libreria (di cui ho già ). Una condizione necessaria: deve essere così semplice da poter essere scritto da zero in una sola sera (o in una giornata lavorativa), altrimenti non è più un programma semplice. Probabilmente contiene errori, complessità superflue e imperfezioni, e inoltre è il mio primo programma utilizzando la libreria asyncio.
Design IM
Per cominciare, dobbiamo capire come apparirà il nostro IM. Per semplicità, consideriamo una rete peer-to-peer, senza alcuna rilevazione dei partecipanti. Indicheremo manualmente a quale indirizzo: porta connetterci per comunicare con l'interlocutore.
Capisco che, al momento, l'assunzione di una connessione diretta tra due computer qualsiasi rappresenti una limitazione significativa all'applicabilità dell'IM nella pratica. Ma più sviluppatori implementeranno vari escamotage per il NAT-traversal, più a lungo rimarremo nell'Internet IPv4, con la deprimente probabilità di connettersi tra computer casuali. Quanto a lungo possiamo tollerare l'assenza di IPv6 a casa e al lavoro?
Avremo una rete friend-to-friend: tutti i possibili interlocutori devono essere noti in anticipo. In primo luogo, questo semplifica notevolmente le cose: ci si presenta, si trova o non si trova il nome/chiave, ci si disconnette o si continua a lavorare, conoscendo l'interlocutore. In secondo luogo, in generale, è sicuro e esclude molteplici attacchi.
L'interfaccia dell'IM sarà simile a soluzioni classiche , che mi piacciono molto per il loro minimalismo e la filosofia Unix-way. Il programma IM crea una directory per ciascun interlocutore con tre socket di dominio Unix:
- in — in essa vengono registrati i messaggi inviati all'interlocutore;
- out — da essa vengono letti i messaggi ricevuti dall'interlocutore;
- stato — leggendo da esso, possiamo sapere se il contatto è attualmente connesso, indirizzo/porta di connessione.
Inoltre, si crea un socket conn, nel quale memorizziamo l'host e la porta, per avviare la connessione con il contatto remoto.
|-- alice
| |-- in
| |-- out
| `-- stato
|-- bob
| |-- in
| |-- out
| `-- stato
`- conn
Questo approccio consente di realizzare implementazioni indipendenti del trasporto IM e dell'interfaccia utente, perché non si può accontentare tutti i gusti. Utilizzando e/o , è possibile ottenere un'interfaccia multi-finestra con evidenziazione della sintassi. E con si può ottenere una riga di input compatibile con GNU Readline per l'invio dei messaggi.
In realtà, i progetti suckless utilizzano file FIFO. Personalmente non sono riuscito a capire come lavorare con i file in asyncio in modo concorrente senza una base manuale di thread dedicati (per queste cose uso da tempo il linguaggio ). Pertanto, ho deciso di utilizzare i socket Unix domain. Sfortunatamente, questo impedisce di eseguire echo 2001:470:dead::babe 6666 > conn. Ho risolto questo problema utilizzando : echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
protocollo iniziale non sicuro
Il trasporto utilizza TCP: garantisce la consegna e il suo ordine. UDP non garantisce né l'uno né l'altro (cosa utile quando si applica la crittografia), mentre supporta non è disponibile di default in Python.
Sfortunatamente, in TCP non esiste il concetto di messaggio, solo di flusso di byte. Pertanto, è necessario inventare un formato per i messaggi in modo da poterli separare all'interno di questo flusso. Possiamo convenire di utilizzare il carattere di nuova riga. Per iniziare va bene, tuttavia, quando cominciamo a crittografare i nostri messaggi, questo carattere potrebbe apparire ovunque nel testo cifrato. Pertanto, nei network, sono popolari i protocolli che inviano prima la lunghezza del messaggio in byte. Ad esempio, in Python, è disponibile la libreria xdrlib che consente di lavorare con questo tipo di formato. .
Non riusciremo a lavorare in modo corretto ed efficiente con la lettura TCP — semplifichiamo il codice. Leggiamo i dati dal socket in un ciclo infinito, finché non decodifichiamo il messaggio completo. Per un approccio del genere, possiamo utilizzare sia JSON che XML come formato. Tuttavia, quando si aggiunge la crittografia, i dati dovranno essere firmati e autenticati — il che richiederà una rappresentazione degli oggetti identica byte per byte, cosa che JSON/XML non garantisce (i risultati dei dump possono differire).
L'XDR è adatto per questo compito, tuttavia scelgo ASN.1 con codifica DER e una libreria, poiché avremo a disposizione oggetti di alto livello con cui è spesso più piacevole e conveniente lavorare. A differenza di schemaless , o , ASN.1 verificherà automaticamente i dati rispetto a uno schema rigido.
# 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))),
))
Il messaggio accettato sarà Msg: o un MsgText testuale (per ora con un solo campo di testo), oppure un messaggio di handshake MsgHandshake (in cui viene trasmesso il nome dell'interlocutore). Al momento appare troppo complicato, ma è un investimento per il futuro.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│MsgHandshake(IdA) │
│─────────────────>│
│ │
│MsgHandshake(IdB) │
││
│ │
│ MsgText() │
│<─────────────────│
│ │
IM senza crittografia
Come ho già detto, per tutte le operazioni con i socket verrà utilizzata la libreria asyncio. Dichiareremo cosa ci aspettiamo al momento dell'avvio:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Il nostro nome peer",
)
parser.add_argument(
"--their-names",
required=True,
help="I loro nomi peer, separati da virgole",
)
parser.add_argument(
"--bind",
default="::1",
help="Indirizzo su cui ascoltare",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Porta su cui ascoltare",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
Si imposta il proprio nome (—our-name alice). Vengono elencati separati da virgola tutti i conversatori attesi (—their-names bob,eve). Per ciascuno dei conversatori, viene creata una directory con socket Unix e una coroutine per ogni in, out, stato:
per 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"))
I messaggi in arrivo dagli utenti dal socket in vengono inviati alle code 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"))
I messaggi in arrivo dai contatti vengono inviati alle code OUT_QUEUES, da cui i dati vengono scritti nel 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()
Quando si legge dal socket state, il programma cerca nell'indice PEER_ALIVE l'indirizzo del contatto. Se non c'è ancora una connessione con il contatto, viene scritto un stringa vuota.
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()
Quando si registra un indirizzo nel socket conn, si attiva la funzione di "iniziatore" della connessione:
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))
Esaminiamo l'iniziatore. Prima di tutto, apre, ovviamente, una connessione all'host/porta specificati e invia un messaggio di handshake con il proprio nome:
130 async def initiator(host, port):
131 _id = repr((host, port))
132 logging.info("%s: dialing", _id)
133 reader, writer = await asyncio.open_connection(host, port)
134 # Messaggio di handshake {{{
135 writer.write(Msg(("handshake", MsgHandshake((
136 ("peerName", OUR_NAME),
137 )))).encode())
138 # }}}
139 await writer.drain()
Poi, aspetta una risposta dalla parte remota. Cerca di decodificare la risposta ricevuta secondo lo schema Msg ASN.1. Presumiamo che l'intero messaggio venga inviato in un singolo segmento TCP e che lo riceveremo in modo atomico quando si chiama .read(). Verifichiamo di aver ricevuto esattamente il messaggio di handshake.
141 # Aspettare il messaggio di handshake {{{
142 data = await reader.read(256)
143 if data == b"":
144 logging.warning("%s: nessuna risposta, disconnettendo", _id)
145 writer.close()
146 return
147 try:
148 msg, _ = Msg().decode(data)
149 except ASN1Error:
150 logging.warning("%s: risposta non decodificabile, disconnettendo", _id)
151 writer.close()
152 return
153 logging.info("%s: ricevuto messaggio %s", _id, msg.choice)
154 if msg.choice != "handshake":
155 logging.warning("%s: messaggio inaspettato, disconnettendo", _id)
156 writer.close()
157 return
158 # }}}
Controlliamo che il nome del corrispondente ricevuto sia conosciuto. Se no, chiudiamo la connessione. Verifichiamo se avevamo già stabilito una connessione con lui (il corrispondente ha nuovamente inviato il comando di connessione) e la chiudiamo. Nella coda IN_QUEUES vengono inserite stringhe Python con il testo del messaggio, ma è presente un valore speciale None, che segnala alla coroutine msg_sender di interrompere il lavoro, affinché dimentichi il suo writer associato alla connessione 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 peer sconosciuto: %s", peer_name)
163 writer.close()
164 return
165 logging.info("%s: sessione stabilita: %s", _id, peer_name)
166 # Esegui invio di messaggi di testo, inizializza il decoder di trasporto {{{
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 gestisce i messaggi in uscita (immessi nella coda dal socket in), li serializza in un messaggio MsgText e li invia tramite connessione TCP. Questa connessione può interrompersi in qualsiasi momento — lo catturiamo esplicitamente.
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: inviato messaggio di %d caratteri", peer_name, len(text))
Alla fine, l'iniziatore entra in un ciclo infinito di lettura dei messaggi dal socket. Controlla se si tratta di messaggi di testo e li inserisce nella coda OUT_QUEUES, da cui saranno inviati al socket out del corrispondente interlocutore. Perché non possiamo semplicemente usare .read() e decodificare il messaggio? Perché non si può escludere che più messaggi dall'utente vengano aggregati nel buffer del sistema operativo e inviati in un unico segmento TCP. Possiamo decodificare solo il primo, mentre nel buffer potrebbe rimanere parte del successivo. In caso di qualsiasi situazione anomala, chiudiamo la connessione TCP e fermiamo la coroutine msg_sender (inviando None nella coda OUT_QUEUES).
174 buf = b""
175 # Aspetta i messaggi di test {{{
176 while True:
177 data = await reader.read(MaxMsgLen)
178 if data == b"":
179 break
180 buf += data
181 if len(buf) > MaxMsgLen:
182 logging.warning("%s: dimensione massima del buffer superata", _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: messaggio %s inaspettato", _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: disconnessione: %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: messaggio ricevuto di %d caratteri", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Torniamo al codice principale. Dopo aver creato tutte le coroutine al momento dell'avvio del programma, avviamo il server TCP. Per ogni connessione stabilita, crea una coroutine responder (risponditore).
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("In ascolto su: %s", server.sockets[0].getsockname())
loop.run_forever()
Il responder è simile all'initiator e compie esattamente le stesse azioni, ma il ciclo infinito di lettura dei messaggi viene avviato immediatamente, per semplicità. Attualmente, il protocollo di handshake invia un messaggio da ciascuna parte, ma in futuro ne saranno inviati due dall'initiator della connessione, dopo i quali sarà possibile inviare immediatamente testi.
72 async def responder(reader, writer):
73 _id = writer.get_extra_info("peername")
74 logging.info("%s: connesso", _id)
75 buf = b""
76 msg_expected = "handshake"
77 peer_name = None
78 while True:
79 # Leggi fino a ottenere il messaggio Msg {{{
80 data = await reader.read(MaxMsgLen)
81 if data == b"":
82 logging.info("%s: connessione chiusa", _id)
83 break
84 buf += data
85 if len(buf) > MaxMsgLen:
86 logging.warning("%s: dimensione massima del buffer superata", _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: messaggio %s inatteso", _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 # Elabora il messaggio Handshake {{{
104 elif msg_expected == "handshake":
105 logging.info("%s: ricevuto messaggio %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("nome peer sconosciuto: %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: sessione stabilita: %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: disconnessione", _id)
125 if msg_expected == "text":
126 IN_QUEUES[peer_name].put(None)
127 writer.close()
Protocollo sicuro
È giunto il momento di garantire la sicurezza nella nostra comunicazione. Cosa intendiamo per sicurezza e cosa desideriamo:
- riservatezza dei messaggi trasmessi;
- autenticità e integrità dei messaggi trasmessi: eventuali modifiche devono essere rilevate;
- protezione contro attacchi di riproduzione (replay attack): la perdita o la ripetizione dei messaggi devono essere rilevate (e decidiamo di interrompere la connessione);
- identificazione e autenticazione dei comunicanti tramite chiavi pubbliche precedentemente impostate: abbiamo già deciso di creare una rete friend-to-friend. Solo dopo l'autenticazione sapremo con chi stiamo comunicando;
- presenza proprietà (PFS): la compromissione della nostra chiave di firma a lungo termine non deve comportare la possibilità di leggere tutta la corrispondenza precedente. La registrazione del traffico intercettato diventa inutile;
- validità dei messaggi (di trasporto e di handshake) solo all'interno di una sessione TCP. L'inserimento di messaggi correttamente firmati/authenticati da una sessione diversa (anche con lo stesso interlocutore) non deve essere possibile;
- un osservatore passivo non dovrebbe vedere né gli identificatori degli utenti, né le chiavi pubbliche a lungo termine trasmesse, né i loro hash. Una certa anonimato dall'osservatore passivo.
Incredibilmente, questa è una base che praticamente tutti desiderano avere in qualsiasi protocollo di handshake, e molto poco di quanto elencato alla fine viene effettivamente realizzato per i protocolli "fatti in casa". Anche adesso non inventiamo niente di nuovo. Consiglio vivamente di usare per costruire protocolli, ma scegliamo qualcosa di più semplice.
I due protocolli più popolari sono:
- — un protocollo complesso con una lunga storia di bug, imperfezioni, vulnerabilità, scarsa progettazione, complessità e difetti (in effetti, ciò che riguarda TLS 1.3 non è molto rilevante). Ma non lo consideriamo a causa della sua eccessiva complessità.
- con — non presentano seri problemi crittografici, anche se non sono semplici. Se si legge di IKEv1 e IKEv2, le loro origini sono , standard ISO/IEC IS 9798-3 e i protocolli SIGMA (SIGn-and-MAc) — abbastanza semplici da implementare in una sola serata.
Cosa rende SIGMA, come ultima evoluzione dei protocolli STS/ISO, così vantaggioso? Soddisfa tutte le nostre esigenze (compresa la 'nascita' degli identificatori delle controparti), e non presenta problemi crittografici noti. È minimalista: l'eliminazione anche solo di un elemento dal messaggio del protocollo ne comprometterebbe la sicurezza.
Facciamo un viaggio dal protocollo più semplice a SIGMA. L'operazione di base che ci interessa è : una funzione che consente ad entrambe le parti di ottenere lo stesso valore, utilizzabile come chiave simmetrica. Senza entrare troppo nei dettagli: ciascuna parte genera una coppia di chiavi effimere (utilizzate solo per una singola sessione), scambia chiavi pubbliche, e chiama la funzione di accordo, alla quale fornisce la propria chiave privata e la chiave pubblica dell'interlocutore.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│ IdA, PubA │ ╔════════════════════╗
│───────────────►│ ║PrvA, PubA = DHgen()║
│ │ ╚════════════════════╝
│ IdB, PubB │ ╔════════════════════╗
│◄───────────────│ ║PrvB, PubB = DHgen()║
│ │ ╚════════════════════╝
────┐ ╔═══════╧════════════╗
│ ║Key = DH(PrvA, PubB)║
◄───┘ ╚═══════╤════════════╝
│ │
│ │
Chiunque può intervenire e sostituire le chiavi pubbliche con le proprie: in questo protocollo non c'è autenticazione tra le parti. Aggiungiamo una firma con chiavi a lungo termine.
┌─────┐ ┌─────┐
│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) ║ │
│ ╚═════════════════════╝ │
│ │
Questa firma non è valida poiché non è collegata a una sessione specifica. Messaggi di questo tipo possono essere utilizzati anche per sessioni con altri partecipanti. L'intero contesto deve essere firmato. Questo costringe a includere anche l'invio di un ulteriore messaggio da A.
Inoltre, è fondamentale aggiungere sotto la firma e un identificatore personale, altrimenti potremmo sostituire IdXXX e riscrivere il messaggio con la chiave di un altro interlocutore conosciuto. Per prevenire , è necessario che gli elementi sotto la firma si trovino in posizioni chiaramente definite per il loro significato: se A firma (PubA, PubB), allora B deve firmare (PubB, PubA). Questo sottolinea anche l'importanza della scelta della struttura e del formato dei dati serializzati. Ad esempio, i set nell'encoding ASN.1 DER sono ordinati: SET OF(PubA, PubB) sarà identico a SET OF(PubB, PubA).
┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └──┬──┘ │ IdA, PubA │ ╔═══════════════════════════╗ │────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚═══════════════════════════╝ │IdB, PubB, sign(SignPrvB, (IdB, PubA, PubB)) │ ╔═══════════════════════════╗ ││ ║verify(SignPubB, ...)║ │ │ ║Key = DH(PrvA, PubB) ║ │ │ ╚═════════════════════╝ │ │
Tuttavia, non abbiamo ancora "dimostrato" di aver generato una chiave condivisa per questa sessione. In linea di principio, si può anche fare a meno di questo passaggio: il primo messaggio di trasporto sarà considerato non valido, ma vogliamo assicurarci che, al termine del handshake, tutto sia veramente concordato. Al momento, abbiamo in mano il protocollo ISO/IEC IS 9798-3.
Potremmo anche firmare la chiave generata. Questo è rischioso, poiché non è escluso che nell'algoritmo di firma utilizzato possano esserci delle perdite (anche se solo bit di firma, ma comunque perdite). È possibile firmare l'hash della chiave generata, ma una fuga anche solo dell'hash della chiave generata può avere valore durante attacchi brute-force sulla funzione di generazione. SIGMA utilizza una funzione MAC per autenticare l'identificatore del mittente.
┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └──┬──┘ │ IdA, PubA │ ╔═══════════════════════════╗ │─────────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚═══════════════════════════╝ │IdB, PubB, sign(SignPrvB, (PubA, PubB)), MAC(IdB) │ ╔═══════════════════════════╗ ││ ║verify(Key, IdB) ║ │ │ ║verify(SignPubB, ...)║ │ │ ╚═════════════════════╝ │ │
Per ottimizzare, alcuni potrebbero voler riutilizzare le proprie chiavi effimere (cosa, ovviamente, dannosa per il PFS). Ad esempio, abbiamo generato una coppia di chiavi, abbiamo tentato di connetterci, ma il TCP non era disponibile o si è interrotto da qualche parte nel protocollo. È un peccato sprecare l'entropia e le risorse della CPU per una nuova coppia. Pertanto, introduciamo il cosiddetto cookie — un valore pseudocasuale che proteggerà da possibili attacchi di replay casuali durante il riutilizzo delle chiavi pubbliche effimere. A causa del binding tra il cookie e la chiave pubblica effimera, la chiave pubblica dell'altro partecipante può essere rimossa dalla firma per mancanza di necessità.
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Infine, vogliamo garantire la privacy dei nostri identificatori agli osservatori passivi. A tale scopo, SIGMA propone di scambiare prima chiavi effimere, generare una chiave condivisa per crittografare i messaggi di autenticazione e identificazione. SIGMA descrive due varianti:
- SIGMA-I — protegge l'iniziatore dagli attacchi attivi e il rispondente da quelli passivi: l'iniziatore autentica il rispondente e, se qualcosa non corrisponde, non rivela la propria identificazione. Il rispondente, invece, fornisce la propria identificazione se viene avviato un protocollo attivo con lui. Un osservatore passivo non apprenderà nulla;
SIGMA-R — protegge il rispondente dagli attacchi attivi e l'iniziatore da quelli passivi. Il tutto è esattamente inverso, ma in questo protocollo vengono trasmessi già quattro messaggi di handshake.Scegliamo SIGMA-I, poiché è più simile a ciò che ci aspettiamo dai normali aspetti client-server: il client riconosce solo il server autenticato, mentre il server conosce già tutto. Inoltre, è più semplice da implementare grazie a un numero ridotto di messaggi di handshake. L'unica cosa che introduciamo nel protocollo è la crittografia di una parte del messaggio e il trasferimento dell'identificatore A nella parte crittografata dell'ultimo messaggio:
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
- Per la firma viene utilizzato il GOST R algoritmo con chiavi a 256 bit.
- Per la generazione della chiave condivisa viene utilizzato 34.10-2012 VKO.
- Come MAC viene utilizzato CMAC. Questa è tecnicamente una modalità speciale di funzionamento di un cifrario a blocchi, descritta nel GOST R 34.13-2015. Come funzione di crittografia per questa modalità — (34.12-2015).
- Come identificatore del partner viene utilizzata l'hash del suo chiave pubblica. Come hash viene applicato (34.11-2012 256 bit).
Dopo la stretta di mano avremo una chiave condivisa concordata. Questa può essere utilizzata per la crittografia autenticata dei messaggi trasportati. Questa parte è piuttosto semplice e difficile da sbagliare: incrementiamo il contatore dei messaggi, crittografiamo il messaggio, autentichiamo (MAC) il contatore e il testo crittografato, inviamo. Al ricevimento del messaggio verifichiamo che il contatore abbia il valore atteso, autentichiamo il testo crittografato con il contatore, e decrittografiamo. Quale chiave utilizzare per crittografare i messaggi di stretta di mano, i messaggi trasportati, quale per autenticare? Usare una sola chiave per tutte queste attività è pericoloso e poco saggio. È necessario generare chiavi utilizzando funzioni specializzate (key derivation function). Non complicheremo le cose e non inveniamo nulla: è ben nota, ampiamente studiata e non presenta problemi noti. Sfortunatamente, la libreria standard di Python non include questa funzione, quindi utilizziamo come pacchetto. Internamente, HKDF utilizza , che a sua volta si basa su una funzione hash. Un esempio di implementazione in Python disponibile sulla pagina Wikipedia richiede poche righe di codice. Come nel caso di 34.10-2012, utilizzeremo la funzione hash Stribog-256. L'output della nostra funzione di derivazione della chiave sarà chiamato chiave di sessione, da cui saranno generate le simmetriche mancanti:
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")Strutture/schemi
Esaminiamo quali strutture ASN.1 abbiamo ora ottenuto per la trasmissione di tutti questi dati:
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 — ciò che sarà firmato (to be signed). HandshakeTBE — ciò che sarà cifrato (to be encrypted). Si fa notare il campo ukm in MsgHandshake1. 34.10 VKO, per una maggiore casualità delle chiavi generate, include il parametro UKM (user keying material) — solo ulteriore entropia.
Aggiunta di crittografia nel codice
Considereremo solo le modifiche apportate al codice originale, poiché la struttura è rimasta la stessa (in realtà, prima è stata scritta l'implementazione finale, e poi è stata rimossa tutta la crittografia).
Poiché l'autenticazione e l'identificazione dei partecipanti avverrà tramite le chiavi pubbliche, ora dobbiamo memorizzarle in modo duraturo. Per semplicità, utilizziamo un JSON di questo tipo:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }our — la nostra coppia di chiavi, chiavi esadecimali private e pubbliche. their — i nomi dei partecipanti e le loro chiavi pubbliche. Modifichiamo gli argomenti della riga di comando e aggiungiamo il post-trattamento dei dati 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="Genera JSON con la nostra nuova coppia di chiavi", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON con le nostre e le loro chiavi", ) parser.add_argument( "--bind", default="::1", help="Indirizzo su cui ascoltare", ) parser.add_argument( "--port", type=int, default=6666, help="Porta su cui ascoltare", ) 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) # Elenca e deserializza le nostre e le loro chiavi {{{ with open(args.keys, "rb") as fd: _keys = json.loads(fd.read().decode("utf-8")) KEY_OUR_SIGN_PRV = gost3410.prv_unmarshal(hexdec(_keys["our"]["prv"])) _pub = hexdec(_keys["our"]["pub"]) KEY_OUR_SIGN_PUB = gost3410.pub_unmarshal(_pub) KEY_OUR_SIGN_PUB_HASH = OctetString(GOST34112012256(_pub).digest()) for peer_name, pub_raw in _keys["their"].items(): _pub = hexdec(pub_raw) KEYS[GOST34112012256(_pub).digest()] = { "name": peer_name, "pub": gost3410.pub_unmarshal(_pub), } # }}}La chiave privata dell'algoritmo 34.10 è un numero casuale. Di dimensioni 256 bit per curve ellittiche a 256 bit. PyGOST non opera con un insieme di byte, ma con , quindi la nostra chiave privata (urandom(32)) deve essere convertita in un numero utilizzando gost3410.prv_unmarshal(). La chiave pubblica è calcolata in modo deterministico dalla chiave privata, usando gost3410.public_key(). La chiave pubblica 34.10 è composta da due numeri grandi, che devono anch'essi essere convertiti in una sequenza di byte per facilitarne la memorizzazione e la trasmissione, utilizzando gost3410.pub_marshal().
Dopo aver letto il file JSON, le chiavi pubbliche devono essere convertite nuovamente, utilizzando gost3410.pub_unmarshal(). Poiché riceveremo gli identificatori dei contatti come hash della chiave pubblica, possiamo calcolarli in anticipo e inserirli in un dizionario per una ricerca rapida. L'hash Stribog-256 è gost34112012256.GOST34112012256(), che soddisfa completamente l'interfaccia delle funzioni hash di hashlib.
Come è cambiata la coroutine dell'initiator? Tutto procede come nello schema del handshake: generiamo un cookie (128 bit sono più che sufficienti), una coppia di chiavi effimere 34.10, che sarà utilizzata per la funzione VKO di accordo delle chiavi.
395 async def initiator(host, port): 396 _id = repr((host, port)) 397 logging.info("%s: dialing", _id) 398 reader, writer = await asyncio.open_connection(host, port) 399 # Generiamo la nostra chiave pubblica effimera e cookie, inviamo il messaggio di handshake 0 {{{ 400 cookie_our = Cookie(urandom(16)) 401 prv = gost3410.prv_unmarshal(urandom(32)) 402 pub_our = gost3410.public_key(CURVE, prv) 403 pub_our_raw = PubKey(gost3410.pub_marshal(pub_our)) 404 writer.write(Msg(("handshake0", MsgHandshake0(( 405 ("cookieInitiator", cookie_our), 406 ("pubKeyInitiator", pub_our_raw), 407 )))).encode()) 408 # }}} 409 await writer.drain()- Aspettiamo la risposta e decodifichiamo il messaggio Msg ricevuto;
- Verifichiamo di aver ricevuto handshake1;
- Decodifichiamo la chiave pubblica effimera dell'altra parte e calcoliamo la chiave di sessione;
- Generiamo le chiavi simmetriche necessarie per gestire la parte TBE del messaggio.
423 logging.info("%s: ricevuto il messaggio %s", _id, msg.choice) 424 se msg.choice != "handshake1": 425 logging.warning("%s: messaggio inaspettato, disconnessione", _id) 426 writer.close() 427 return 428 # }}} 429 msg_handshake1 = msg.value 430 # Convalida il messaggio Handshake {{{ 431 cookie_their = msg_handshake1["cookieResponder"] 432 pub_their_raw = msg_handshake1["pubKeyResponder"] 433 pub_their = gost3410.pub_unmarshal(bytes(pub_their_raw)) 434 ukm_raw = bytes(msg_handshake1["ukm"]) 435 ukm = ukm_unmarshal(ukm_raw) 436 key_session = kek_34102012256(CURVE, prv, pub_their, ukm, mode=2001) 437 kdf = Hkdf(None, key_session, hash=GOST34112012256) 438 key_handshake1_mac_identity = kdf.expand(b"handshake1-mac-identity") 439 key_handshake1_enc = kdf.expand(b"handshake1-enc") 440 key_handshake1_mac = kdf.expand(b"handshake1-mac")UKM è un numero a 64 bit (urandom(8)), che richiede anche la deserializzazione dalla rappresentazione byte, utilizzando gost3410_vko.ukm_unmarshal(). La funzione VKO per 34.10-2012 a 256 bit è gost3410_vko.kek_34102012256() (KEK — key encryption key).
La chiave di sessione generata è già una sequenza pseudo-casuale di 256 bit. Pertanto, può essere immediatamente utilizzata nella funzione HKDF. Poiché GOST34112012256 soddisfa l'interfaccia hashlib, può essere utilizzato immediatamente nella classe Hkdf. Non specificiamo il sale (il primo argomento di Hkdf) poiché la chiave generata, a causa della transitorietà delle coppie di chiavi coinvolte, sarà diversa per ogni sessione e conterrà già sufficiente entropia. kdf.expand() restituisce già per impostazione predefinita chiavi lunghe 256 bit, necessarie per il Kuznets.
Successivamente, vengono controllate le parti TBE e TBS del messaggio ricevuto:
- si calcola e verifica il MAC sul testo cifrato ricevuto;
- si decritta il testo cifrato;
- si decodifica la struttura TBE;
- da essa viene estratto l'identificatore dell'interlocutore e si verifica se ci è noto;
- si calcola e verifica il MAC su questo identificatore;
- si verifica la firma sulla struttura TBS, che include i cookie di entrambe le parti e la chiave pubblica effimera dell'altra parte. La firma viene verificata con la chiave di firma a lungo termine dell'interlocutore.
441 prova: 442 nome_peer = 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 eccetto ValueError come err: 452 logging.warning("%s: %s, disconnesso", _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 se non compare_digest(mac_tag, bytes(msg_handshake["ciphertextMac"])): 140 raise ValueError("MAC non valido") 141 plaintext = ctr( 142 GOST3412Kuznechik(key_enc).encrypt, 143 KUZNECHIK_BLOCKSIZE, 144 ciphertext, 145 8 * b"x00", 146 ) 147 prova: 148 tbe, _ = HandshakeTBE().decode(plaintext) 149 eccetto ASN1Error: 150 raise ValueError("impossibile decodificare TBE") 151 key_sign_pub_hash = bytes(tbe["identity"]) 152 peer = KEYS.get(key_sign_pub_hash) 153 se peer è None: 154 raise ValueError("identità sconosciuta") 155 mac_tag = mac( 156 GOST3412Kuznechik(key_mac_identity).encrypt, 157 KUZNECHIK_BLOCKSIZE, 158 key_sign_pub_hash, 159 ) 160 se non compare_digest(mac_tag, bytes(tbe["identityMac"])): 161 raise ValueError("MAC identità non valido") 162 tbs = HandshakeTBS(( 163 ("cookieTheir", cookie_their), 164 ("cookieOur", cookie_our), 165 ("pubKeyOur", pub_key_our), 166 )) 167 se non gost3410.verify( 168 CURVE, 169 peer["pub"], 170 GOST34112012256(tbs.encode()).digest(), 171 bytes(tbe["signature"]), 172 ): 173 raise ValueError("firma non valida") 174 return peer["name"]Come già accennato in precedenza, la 34.13-2015 descrive diversi dalla 34.12-2015. Tra di essi c'è la modalità di generazione della firma, calcolo del MAC. In PyGOST si usa gost3413.mac(). Questa modalità richiede la trasmissione della funzione di crittografia (che accetta e restituisce un blocco di dati), della dimensione del blocco di crittografia e, naturalmente, dei dati stessi. Perché non si può hardcodare la dimensione del blocco di crittografia? La 34.12-2015 descrive non solo il cifrario a 128 bit Kuznechik, ma anche il 64 bit – una versione leggermente modificata del GOST 28147-89, creata ancora al KGB e tuttora dotata di uno dei più alti livelli di sicurezza.
Kuznechik si inizializza con la chiamata gost.3412.GOST3412Kuznechik(key) e restituisce un oggetto con metodi .encrypt()/.decrypt() idonei per essere passati alla funzione 34.13. Il MAC si calcola nel seguente modo: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Per confrontare il MAC calcolato e quello ricevuto non si può usare il normale confronto (==) delle stringhe di byte, poiché questa operazione potrebbe generare perdite di tempo di confronto, che, in generale, possono portare a vulnerabilità fatali come gli attacchi TLS. In Python esiste una funzione specifica hmac.compare_digest per questo.
La funzione di cifratura a blocchi può crittografare solo un singolo blocco di dati. Per una maggiore quantità, e non di lunghezza multipla, è necessario utilizzare una modalità di crittografia. Nella norma 34.13-2015 sono descritti i seguenti: ECB, CTR, OFB, CBC, CFB. Ognuno ha i propri ambiti e caratteristiche d'uso. Purtroppo, non abbiamo ancora modalità di crittografia standardizzate (come CCM, OCB, GCM e simili) — siamo costretti almeno a implementare noi un MAC. Scelgo (CTR): non richiede un'integrazione alla dimensione del blocco, può essere parallelizzata, utilizza solo la funzione di crittografia, può essere usata in sicurezza per crittografare un gran numero di messaggi (a differenza del CBC, che inizia a presentare collisioni relativamente rapidamente).
Come .mac(), anche .ctr() accetta dati simili in input: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). È necessario impostare un vettore di inizializzazione, lungo esattamente la metà della dimensione del blocco di cifratura. Se la nostra chiave di cifratura viene utilizzata solo per cifrare un singolo messaggio (anche se composto da più blocchi), è sicuro specificare un vettore di inizializzazione nullo. Per cifrare i messaggi handshake utilizziamo di volta in volta una chiave separata.
Il controllo della firma gost3410.verify() è triviale: dobbiamo fornire la curva ellittica entro la quale operiamo (che fissiamo nel nostro protocollo GOSTIM), la chiave pubblica del firmatario (ricordando che deve essere una coppia di numeri grandi, non una stringa di byte), l'hash 34.11-2012 e la firma ricevuta.
Successivamente, nell'iniziatore prepariamo e inviamo il messaggio handshake2 di handshake, eseguendo le stesse operazioni fatte durante il controllo, ma simmetricamente: firmiamo con le nostre chiavi invece di controllare, e così via...
456 # Prepara e invia il messaggio Handshake 2 {{{ 457 tbs = HandshakeTBS(( 458 ("cookieTheir", cookie_their), 459 ("cookieOur", cookie_our), 460 ("pubKeyOur", pub_our_raw), 461 )) 462 signature = gost3410.sign( 463 CURVE, 464 KEY_OUR_SIGN_PRV, 465 GOST34112012256(tbs.encode()).digest(), 466 ) 467 key_handshake2_mac_identity = kdf.expand(b"handshake2-mac-identity") 468 mac_tag = mac( 469 GOST3412Kuznechik(key_handshake2_mac_identity).encrypt, 470 KUZNECHIK_BLOCKSIZE, 471 bytes(KEY_OUR_SIGN_PUB_HASH), 472 ) 473 tbe = HandshakeTBE(( 474 ("identity", KEY_OUR_SIGN_PUB_HASH), 475 ("signature", OctetString(signature)), 476 ("identityMac", MAC(mac_tag)), 477 )) 478 tbe_raw = tbe.encode() 479 key_handshake2_enc = kdf.expand(b"handshake2-enc") 480 key_handshake2_mac = kdf.expand(b"handshake2-mac") 481 ciphertext = ctr( 482 GOST3412Kuznechik(key_handshake2_enc).encrypt, 483 KUZNECHIK_BLOCKSIZE, 484 tbe_raw, 485 8 * b"x00", 486 ) 487 mac_tag = mac( 488 GOST3412Kuznechik(key_handshake2_mac).encrypt, 489 KUZNECHIK_BLOCKSIZE, 490 ciphertext, 491 ) 492 writer.write(Msg(("handshake2", MsgHandshake2(( 493 ("ciphertext", OctetString(ciphertext)), 494 ("ciphertextMac", MAC(mac_tag)), 495 )))).encode()) 496 # }}} 497 await writer.drain() 498 logging.info("%s: session established: %s", _id, peer_name)Una volta stabilita la sessione, vengono generati i chiavi di trasporto (una per la crittografia, una per l'autenticazione, per ciascuna delle parti), e viene inizializzato Kuznechik per la decrittazione e la verifica del MAC:
499 # Esegui il mittente di messaggi di testo, inizializza il decodificatore di trasporto {{{ 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 # Attendi i messaggi di prova {{{ 521 while True: 522 data = await reader.read(MaxMsgLen) ... 530 msg, tail = Msg().decode(buf) ... 537 try: 538 await msg_receiver( 539 msg.value, 540 nonce_expected, 541 macer, 542 encrypter, 543 peer_name, 544 ) 545 except ValueError as err: 546 logging.warning("%s: %s", err) 547 break 548 nonce_expected += 1 549 # }}}La coroutine msg_sender ora crittografa i messaggi prima di inviarli nella connessione TCP. Ogni messaggio ha un nonce che cresce in modo monotono, che è anche un vettore di inizializzazione quando si crittografa in modalità contatore. Ogni messaggio e blocco di messaggi garantirà valori del contatore distinti.
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 += 1I messaggi in arrivo vengono elaborati dalla coroutine msg_receiver, che si occupa dell'autenticazione e della decrittazione:
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("valore nonce inatteso") mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode()) if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])): raise ValueError("MAC non valido") plaintext = ctr( encrypter, KUZNECHIK_BLOCKSIZE, bytes(payload["ciphertext"]), long2bytes(nonce_expected, 8), ) text = plaintext.decode("utf-8") await OUT_QUEUES[peer_name].put(textConclusione
GOSTIM è previsto esclusivamente per scopi didattici (poiché non coperto da test, almeno)! Il codice sorgente del programma può essere scaricato (Stribog-256 hash: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Come tutti i miei progetti, di tipo , , , , GOSTIM è completamente , distribuito sotto i termini .
, , membro , sviluppatore Python/Go, esperto principale .
Fonte: habr.com
