Essendo uno sviluppatore di una libreria (primitivi crittografici GOST in puro Python), ricevo spesso domande su come implementare un semplice scambio di messaggi sicuro. Molti ritengono che la crittografia applicata sia abbastanza semplice e che una chiamata a .encrypt() a un cifratore a blocchi sia sufficiente per una comunicazione sicura. Altri, invece, credono che la crittografia applicata sia appannaggio di pochi e che sia accettabile che aziende ricche come Telegram con matematici olimpici un protocollo sicuro.
Tutto ciò mi ha spinto a scrivere questo articolo per dimostrare che implementare protocolli crittografici e un IM sicuro non è un compito così difficile. Tuttavia, non vale la pena inventare i propri protocolli di autenticazione e scambio chiavi.
Nell'articolo verrà scritto , , il protocollo SIGMA-I IPsec IKE PyDERASN scritto in precedenza Design dell'IM
Per cominciare, dobbiamo capire come sarà il nostro IM. Per semplicità, supponiamo che sia una rete peer-to-peer, senza alcuna rilevazione dei partecipanti. Indicheremo manualmente a quale indirizzo: porto collegarci per comunicare con il nostro interlocutore.
Capisco che, al momento, l'ipotesi sulla disponibilità di una connessione diretta tra due computer arbitrari rappresenta una limitazione significativa nell'applicabilità dell'IM nella pratica. Ma più sviluppatori implementeranno costrizioni di traversamento NAT, più a lungo rimarremo nella rete IPv4, con la deprimente probabilità di comunicazione tra computer arbitrari. Quanto ancora dobbiamo sopportare l'assenza di IPv6 a casa e al lavoro?
Capisco che, al momento, l'idea della disponibilità di una connessione diretta tra due computer qualsiasi rappresenti una significativa limitazione all'applicabilità di IM nella pratica. Ma più sviluppatori implementeranno vari stratagemmi di NAT-traversal, più a lungo continueremo a rimanere nell'Internet IPv4, con la deprimente probabilità di connessione tra computer qualsiasi. Quanto ancora dobbiamo sopportare 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 tutto: ci presentiamo, troviamo o non troviamo il nome/chiave, ci scollegiamo o continuiamo a lavorare, conoscendo l'interlocutore. In secondo luogo, in generale, questo è sicuro ed esclude molteplici attacchi.
L'interfaccia di 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 ogni interlocutore con tre socket di dominio Unix:
- in — in cui vengono scritti i messaggi inviati all'interlocutore;
- out — da cui vengono letti i messaggi ricevuti dall'interlocutore;
- state — leggendo da qui, scopriamo se l'interlocutore è attualmente connesso, indirizzo/porta di connessione.
Inoltre, viene creato un socket conn, scrivendo in cui l'host porta, iniziamo la connessione con l'interlocutore remoto.
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Questo approccio consente di realizzare implementazioni indipendenti del trasporto IM e dell'interfaccia utente, poiché a ciascuno il proprio gusto e colore, non si può accontentare tutti. Usando e/o , è possibile ottenere un'interfaccia multi-finestra con evidenziazione della sintassi. E con l'aiuto di , è possibile 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 un supporto manuale di thread dedicati (per tali cose utilizzo da tempo il linguaggio ). Perciò ho deciso di utilizzare i socket di dominio Unix. Purtroppo, questo elimina la possibilità di fare echo 2001:470:dead::babe 6666 > conn. Ho risolto questo problema, usando : echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
Protocollo iniziale non sicuro
Come trasporto viene utilizzato TCP: garantisce la consegna e il suo ordine. UDP non garantisce né l'uno né l'altro (cosa che sarebbe utile quando si applica la crittografia), e il supporto per in Python non è incluso di default.
Purtroppo, in TCP non esiste il concetto di messaggio, ma solo di flusso di byte. Pertanto, è necessario ideare un formato per i messaggi, in modo da poterli separare l'uno dall'altro in questo flusso. Possiamo convenire di utilizzare il carattere di ritorno a capo. Per iniziare va bene, tuttavia, quando iniziamo a crittografare i nostri messaggi, questo simbolo potrebbe apparire ovunque nel testo cifrato. Nei network, per questo motivo, i protocolli popolari inviano prima la lunghezza del messaggio in byte. Ad esempio, in Python esiste in bundle xdrlib che consente di lavorare con un formato simile. .
Non lavoreremo in modo corretto ed efficiente con la lettura TCP — semplificheremo il codice. Leggiamo in un ciclo infinito i dati dal socket, finché non decodifichiamo il messaggio completo. Come formato per questo approccio possiamo usare anche JSON e XML. Ma, quando verrà aggiunta la crittografia, i dati dovranno essere firmati e autenticati — e questo richiederà una rappresentazione byte-per-byte identica degli oggetti, cosa che JSON/XML non garantiscono (il risultato dei dumps può differire).
XDR è adatto per questo compito, tuttavia scelgo ASN.1 con codifica DER e una libreria, poiché avremo oggetti di alto livello con cui spesso è più gradevole e conveniente lavorare. A differenza di schemi senza schema , o , ASN.1 controllerà automaticamente i dati rispetto a uno schema rigorosamente definito.
# 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 testuale), o un messaggio di handshake MsgHandshake (in cui viene trasmesso il nome dell’interlocutore). Adesso sembra troppo complicato, ma è una preparazione 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 di peer",
)
parser.add_argument(
"--their-names",
required=True,
help="I loro nomi di 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(","))
Viene specificato un nome personale (—our-name alice). I nomi di tutti i peer attesi vengono elencati separati da virgole (—their-names bob,eve). Per ciascun peer, viene creata una directory con i socket Unix, oltre a una coroutine per ogni 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"))
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 peer 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()
Durante la lettura dal socket di stato, il programma cerca nell'archivio PEER_ALIVE l'indirizzo del peer. Se non ci sono ancora connessioni con il peer, viene registrata una 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 scrive l'indirizzo nel socket conn, viene avviata la funzione "initiator" 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))
Consideriamo l'initiator. Inizialmente, ovviamente, apre una connessione all'host/porta specificato e invia un messaggio di handshake con il suo nome:
130 async def initiator(host, port):
131 _id = repr((host, port))
132 logging.info("%s: composizione", _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()
Dopo, aspetta una risposta dalla parte remota. Cerca di decodificare la risposta ricevuta secondo lo schema Msg ASN.1. Supponiamo che l'intero messaggio sia inviato in un unico segmento TCP e che lo riceveremo in modo atomico con la chiamata a .read(). Controlliamo se abbiamo effettivamente ricevuto il messaggio di handshake.
141 # Aspetta il messaggio di handshake {{{
142 data = await reader.read(256)
143 if data == b"":
144 logging.warning("%s: nessuna risposta, disconnessione", _id)
145 writer.close()
146 return
147 try:
148 msg, _ = Msg().decode(data)
149 except ASN1Error:
150 logging.warning("%s: risposta illeggibile, disconnessione", _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 inatteso, disconnessione", _id)
156 writer.close()
157 return
158 # }}}
Controlliamo se il nome del peer ricevuto è a noi noto. Se non lo è, interrompiamo la connessione. Controlliamo se avevamo già stabilito una connessione con lui (il peer ha di nuovo richiesto di connettersi a noi) e la chiudiamo. Nella coda IN_QUEUES vengono inserite stringhe Python con il testo del messaggio, ma esiste un valore speciale None, che segnala alla coroutine msg_sender di interrompere il lavoro, per farle dimenticare il proprio writer collegato 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 l'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 riceve i messaggi in uscita (immessi nella coda dal socket in), li serializza in un messaggio MsgText e li invia tramite la connessione TCP. Essa può interrompersi in qualsiasi momento — questo lo gestiamo 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 nel socket out del corrispondente interlocutore. Perché non possiamo semplicemente utilizzare .read() e decodificare il messaggio? Perché non è escluso che più messaggi dell'utente vengano aggregati nel buffer del sistema operativo e inviati in un unico segmento TCP. Potremo decodificare il primo, ma nel buffer potrebbe rimanere una parte del successivo. In qualsiasi situazione anomala chiudiamo la connessione TCP e arrestiamo 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'iniziatore e compie esattamente le stesse azioni, ma il ciclo infinito di lettura dei messaggi viene avviato subito, per semplicità. Attualmente, il protocollo di handshake invia un messaggio da ciascun lato, ma in futuro, l'iniziatore della connessione invierà due messaggi, dopo i quali sarà subito possibile inviare messaggi di testo.
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 inaspettato", _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: disconnettendo", _id)
125 if msg_expected == "text":
126 IN_QUEUES[peer_name].put(None)
127 writer.close()
Protocollo sicuro
È tempo di rendere sicura la nostra comunicazione. Cosa intendiamo per sicurezza e cosa vogliamo:
- riservatezza dei messaggi trasmessi;
- autenticità e integrità dei messaggi trasmessi — eventuali modifiche devono essere rilevate;
- protezione dagli attacchi di riproduzione (replay attack) — deve essere rilevato il fatto che messaggi siano scomparsi o ripetuti (e decidiamo di interrompere la connessione);
- identificazione e autenticazione dei conversatori tramite chiavi pubbliche preimpostate — abbiamo già deciso in precedenza di realizzare una rete friend-to-friend. Solo dopo l'autenticazione capiremo con chi stiamo comunicando;
- presenza di proprietà (PFS) — la compromissione della nostra chiave di firma a lunga durata non deve consentire la lettura di tutta la corrispondenza precedente. La registrazione del traffico intercettato diventa inutile;
- la validità dei messaggi (di trasporto e di handshake) è limitata a una singola sessione TCP. L'inserimento di messaggi correttamente firmati/autenticati da un'altra sessione (anche dallo stesso interlocutore) non dovrebbe essere possibile;
- l'osservatore passivo non dovrebbe vedere né identificatori degli utenti, né chiavi pubbliche a lungo termine trasmesse, né gli hash di esse. Una certa anonimato dall'osservatore passivo.
Incredibile, ma praticamente tutti vogliono avere questo minimo in qualsiasi protocollo di handshake, e molto poco di quanto sopra viene effettivamente implementato per i protocolli "fatti in casa". E ora non inventiamo nulla di nuovo. Raccomanderei senza dubbio di utilizzare per costruire protocolli, ma scegliamo qualcosa di più semplice.
I due protocolli più popolari sono:
- — un protocollo complesso con una lunga storia di bug, problemi, vulnerabilità, mancanza di pianificazione, complessità e lacune (tuttavia, questo si applica poco a TLS 1.3). Ma non lo consideriamo a causa della sua eccessiva complessità.
- con — non hanno seri problemi crittografici, anche se non sono semplici. Se si legge di IKEv1 e IKEv2, la loro origine è data da , ISO/IEC IS 9798-3 e protocolli SIGMA (SIGn-and-MAc) — piuttosto facili da implementare in una sera.
Perché SIGMA, come ultimo anello di sviluppo dei protocolli STS/ISO, è buono? Soddisfa tutti i nostri requisiti (compresa la "nascondibilità" degli identificatori degli interlocutori), non ha problemi crittografici noti. È minimalista: la rimozione di almeno un elemento dal messaggio del protocollo porterà alla sua insicurezza.
Passiamo dal protocollo di base fatto in casa a SIGMA. L'operazione più basilare che ci interessa è : una funzione, in cui entrambi i partecipanti riceveranno lo stesso valore, che potrà essere utilizzato come chiave simmetrica. Senza entrare nel dettaglio: ciascuna delle parti genera una coppia di chiavi efemere (utilizzate solo all'interno di una singola sessione), scambia le chiavi pubbliche, chiama la funzione di negoziazione, alla quale forniscono la loro 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 nel mezzo e sostituire le chiavi pubbliche con le proprie — in questo protocollo non c'è autenticazione delle controparti. 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) ║ │
│ ╚═════════════════════╝ │
│ │
Tale firma non è appropriata, poiché non è legata a una sessione specifica. Questi messaggi "andranno bene" anche per sessioni con altri partecipanti. Deve essere firmato tutto il contesto. Questo costringe anche a inviare un ulteriore messaggio da A.
Inoltre, è fondamentale aggiungere alla firma anche il proprio identificatore, altrimenti potremmo sostituire IdXXX e risignare il messaggio con la chiave di un altro interlocutore noto. Per prevenire , è necessario che gli elementi sotto la firma si trovino in posizioni chiaramente definite in base al 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, gli insiemi nella codifica 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 comune condivisa per questa sessione. In linea di principio, potremmo anche fare a meno di questo passaggio: il primo messaggio di trasporto sarà non valido, ma vogliamo essere certi che, una volta concluso il handshake, tutto sia effettivamente concordato. Al momento abbiamo a disposizione il protocollo ISO/IEC IS 9798-3.
Potremmo firmare anche la chiave generata. Questo è rischioso, poiché non è escluso che nell'algoritmo di firma utilizzato possano esserci perdite (siano esse bit di firma o meno). È possibile firmare l'hash della chiave generata, ma una perdita anche dell'hash della chiave generata può avere valore in caso di attacco 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, ...)║ │ │ ╚═════════════════════╝ │ │
Come ottimizzazione, alcuni potrebbero voler riutilizzare le proprie chiavi effimere (cosa che, 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 a metà 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. Grazie al binding tra il cookie e la chiave pubblica effimera, la chiave pubblica dell'altra parte può essere rimossa dalla firma per inutilizzo.
┌─────┐ ┌─────┐ │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 da osservatori passivi. A questo proposito, SIGMA propone di scambiare prima chiavi effimere, generare una chiave comune sulla quale cifrare i messaggi di autenticazione e identificazione. SIGMA descrive due varianti:
- SIGMA-I — protegge l'iniziatore da attacchi attivi e il rispondente da attacchi passivi: l'iniziatore autentica il rispondente e, se qualcosa non torna, non rivela la propria identificazione. Il rispondente rivela la propria identificazione solo se inizia un protocollo attivo. Un osservatore passivo non apprende nulla;
SIGMA-R — protegge il rispondente da attacchi attivi, l'iniziatore da attacchi passivi. Tutto è esattamente all'opposto, ma in questo protocollo vengono già trasmessi quattro messaggi di handshake.Scegliamo SIGMA-I poiché è più simile a ciò che ci aspettiamo da elementi client-server familiari: solo il server autenticato può riconoscere il client, mentre il server conosce già tutto. Inoltre, è più semplice da implementare a causa del numero ridotto di messaggi di handshake. Tutto ciò 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))) │ ╔═══════════════════════════╗ │<─────────────────────────────────────────────────────────────────────────────│ ║SignPrvB, SignPubB = load()║ │ │ ║PrvB, PubB = DHgen() ║ │ │ ╚═══════════════════════════╝ │ │ ╔═════════════════════╗ │ Enc((IdA, sign(SignPrvA, (CookieB, CookieA, PubA)), MAC(IdA))) │ ║Key = DH(PrvA, PubB) ║ │─────────────────────────────────────────────────────────────────────────────>│ ║verify(Key, IdB) ║ │ │ ║verify(SignPubB, ...)║ │ │ ╚═════════════════════╝ │ │
- Per la firma si utilizza GOST R algoritmo con chiavi a 256 bit.
- Per la generazione della chiave condivisa si utilizza 34.10-2012 VKO.
- Come MAC si utilizza CMAC. Tecnicheamente, questo è un modo speciale di funzionamento di un cifrario a blocchi, descritto in GOST R 34.13-2015. Come funzione di crittografia per questo modo — (34.12-2015).
- Come identificatore del corrispondente viene utilizzato un hash della sua chiave pubblica. Come hash viene applicato (34.11-2012 256 bit).
Dopo il handshake avremo una chiave condivisa concordata. Possiamo usarla per la crittografia autenticata dei messaggi di trasporto. Questa parte è piuttosto semplice e difficile da sbagliare: incrementiamo il contatore dei messaggi, crittografiamo il messaggio, autenticizziamo (MAC) il contatore e il testo crittografato, e inviamo. All'arrivo del messaggio verifichiamo che il contatore abbia il valore atteso, autenticizziamo il testo crittografato con il contatore, e decrittografiamo. Quale chiave utilizzare per crittografare i messaggi di handshake, di trasporto, quale autenticizzare? Usare una sola chiave per tutti questi compiti è pericoloso e irragionevole. È necessario generare chiavi utilizzando funzioni specializzate (funzione di derivazione della chiave). Anche in questo caso, non complicheremo le cose o inventeremo nulla: è ben nota, ben studiata e non presenta problemi noti. Purtroppo, nella libreria nativa di Python non è presente questa funzione, quindi utilizziamo il pacchetto. HKDF utilizza internamente , che a sua volta utilizza una funzione hash. Un esempio di implementazione in Python sulla pagina Wikipedia richiede poche righe di codice. Come nel caso di 34.10-2012, utilizzeremo Stribog-256 come funzione hash. L'output della nostra funzione di concordanza delle chiavi sarà chiamato chiave di sessione, da cui saranno generate le chiavi 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/schema
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 (da firmare). HandshakeTBE — ciò che sarà crittografato (da crittografare). Faccio notare il campo ukm in MsgHandshake1. 34.10 VKO, per una maggiore randomizzazione delle chiavi generate, include il parametro UKM (materiale di chiave utente) — semplicemente un'ulteriore entropia.
Aggiunta di crittografia nel codice
Consideriamo solo le modifiche apportate al codice originale, poiché la struttura è rimasta la stessa (in realtà, è stata inizialmente scritta l'implementazione finale, e poi ne è stata estratta tutta la crittografia).
Poiché l'autenticazione e l'identificazione dei comunicanti saranno effettuate tramite chiavi pubbliche, ora devono essere memorizzate da qualche parte 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 private e pubbliche esadecimali. their — i nomi dei comunicanti e le loro chiavi pubbliche. Modifichiamo gli argomenti della riga di comando e aggiungiamo un post-processing dei dati JSON:
da pygost import gost3410 da pygost.gost34112012256 import GOST34112012256 CURVA = gost3410.GOST3410Curve( *gost3410.CURVE_PARAMS["GostR3410_2001_CryptoPro_A_ParamSet"] ) parser = argparse.ArgumentParser(description="GOSTIM") parser.add_argument( "--keys-gen", action="store_true", help="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(CURVA, gost3410.prv_unmarshal(prv_raw)) pub_raw = gost3410.pub_marshal(pub) print(json.dumps({ "nostro": {"prv": hexenc(prv_raw), "pub": hexenc(pub_raw)}, "loro": {}, })) exit(0) # Analizza e de-serializza 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["nostro"]["prv"])) _pub = hexdec(_keys["nostro"]["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["loro"].items(): _pub = hexdec(pub_raw) KEYS[GOST34112012256(_pub).digest()] = { "nome": peer_name, "pub": gost3410.pub_unmarshal(_pub), } # }}}La chiave privata dell'algoritmo 34.10 è un numero casuale. Ha una dimensione di 256 bit per curve elliptiche a 256 bit. PyGOST non lavora con una serie di byte, ma con , pertanto la nostra chiave privata (urandom(32)) deve essere convertita in un numero utilizzando gost3410.prv_unmarshal(). La chiave pubblica viene calcolata in modo deterministico dalla chiave privata utilizzando gost3410.public_key(). La chiave pubblica 34.10 è composta da due numeri grandi, che devono essere anch'essi convertiti in una sequenza di byte per comodità di memorizzazione e trasmissione, utilizzando gost3410.pub_marshal().
Dopo aver letto il file JSON, le chiavi pubbliche devono essere convertite di nuovo utilizzando gost3410.pub_unmarshal(). Poiché riceveremo gli identificatori dei partner come hash dalla chiave pubblica, possiamo calcolarli in anticipo e inserirli in un dizionario per una ricerca rapida. L'hash Streibog-256 è rappresentato da gost34112012256.GOST34112012256(), completamente compatibile con l'interfaccia hashlib delle funzioni di hash.
Come è cambiata la coroutine dell'inizializzatore? Tutto come nello schema di handshake: generiamo un cookie (128 bit è più che sufficiente), una coppia di chiavi efemere 34.10, che verrà utilizzata per la funzione di concordanza delle chiavi VKO.
395 async def initiator(host, port): 396 _id = repr((host, port)) 397 logging.info("%s: composizione", _id) 398 reader, writer = await asyncio.open_connection(host, port) 399 # Generiamo la nostra chiave pubblica ephemerale e cookie, inviamo il messaggio 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 una risposta e decodifichiamo il messaggio Msg ricevuto;
- ci assicuriamo di aver ricevuto handshake1;
- decodifichiamo la chiave pubblica ephemerale dell'altra parte e calcoliamo la chiave di sessione;
- generiamo le chiavi simmetriche necessarie per elaborare la parte TBE del messaggio.
423 logging.info("%s: ricevuto messaggio %s", _id, msg.choice) 424 if msg.choice != "handshake1": 425 logging.warning("%s: messaggio inaspettato, disconnessione", _id) 426 writer.close() 427 return 428 # }}} 429 msg_handshake1 = msg.value 430 # Convalida del messaggio di 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")L'UKM è un numero a 64 bit (urandom(8)), che richiede anch'esso la disserializzazione dalla rappresentazione byte, utilizzando gost3410_vko.ukm_unmarshal(). La funzione VKO per 34.10-2012 a 256 bit è gost3410_vko.kek_34102012256() (KEK — chiave di crittografia della chiave).
La chiave di sessione generata è già una sequenza di byte pseudo-casuale a 256 bit. Può quindi essere immediatamente utilizzata nella funzione HKDF. Poiché GOST34112012256 soddisfa l'interfaccia hashlib, può essere utilizzata direttamente nella classe Hkdf. Non specifichiamo il sale (primo argomento di Hkdf) poiché la chiave generata, a causa dell'ephemeralità delle coppie di chiavi coinvolte, sarà diversa per ogni sessione ed avrà già sufficiente entropia. kdf.expand() restituisce per impostazione predefinita chiavi di lunghezza 256 bit, necessarie per l'Uccello di Fuoco in seguito.
In seguito controlliamo le parti TBE e TBS del messaggio ricevuto:
- si calcola e si verifica il MAC sul testo cifrato ricevuto;
- si decripta il testo cifrato;
- si decodifica la struttura TBE;
- da essa si prende l'identificativo dell'interlocutore e si verifica se ci è noto;
- si calcola e si verifica il MAC su questo identificatore;
- viene verificata la firma sopra la struttura TBS, che include i cookie di entrambe le parti e la chiave pubblica effimera dell'altra parte. La firma viene verificata utilizzando la chiave di firma a lungo termine dell'interlocutore.
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, disconnessione", _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 non valido") 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("impossibile decodificare TBE") 151 key_sign_pub_hash = bytes(tbe["identity"]) 152 peer = KEYS.get(key_sign_pub_hash) 153 if peer is None: 154 raise ValueError("identità sconosciuta") 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 identità non valido") 162 tbs = HandshakeTBS(( 163 ("cookieTheir", cookie_their), 164 ("cookieOur", cookie_our), 165 ("pubKeyOur", pub_key_our), 166 )) 167 if not gost3410.verify( 168 CURVE, 169 peer["pub"], 170 GOST34112012256(tbs.encode()).digest(), 171 bytes(tbe["signature"]), 172 ): 173 raise ValueError("firma non valida") 174 return peer["name"]Come già scritto sopra, la 34.13-2015 descrive vari dalla 34.12-2015. Tra queste c'è la modalità per la generazione di un'imbottitura, il calcolo del MAC. In PyGOST si utilizza gost3413.mac(). Questa modalità richiede la trasmissione della funzione di crittografia (che accetta e restituisce un blocco di dati), la dimensione del blocco di crittografia e, appunto, i dati stessi. Perché non si può definire hardcoded la dimensione del blocco di crittografia? La 34.12-2015 descrive non solo il cifrario a 128 bit Kuznechik, ma anche quello a 64 bit — un'ulteriore modifica del GOST 28147-89, creato ancora in KGB e che mantiene ancora uno dei più alti standard di sicurezza.
Il grasso viene inizializzato con gost.3412.GOST3412Kuznechik(key) chiamando e restituisce un oggetto con metodi .encrypt()/.decrypt() adatti per il trasferimento in funzioni 34.13. La MAC viene calcolata come segue: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Per confrontare la MAC calcolata e quella ricevuta, non si può utilizzare un confronto normale (==) delle stringhe di byte, poiché questa operazione provoca perdite di tempo di confronto, il che, in generale, può portare a vulnerabilità fatali di tipo attacchi a TLS. In Python esiste una funzione speciale hmac.compare_digest per questo.
La funzione del cifratore a blocchi può crittografare solo un blocco di dati. Per quantità maggiori, 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. Ognuna ha i propri ambiti di applicazione e caratteristiche. Con grande dispiacere, non abbiamo ancora modalità di crittografia autentificate standardizzate la modalità contatore Come e .mac(), .ctr() accetta dati simili in ingresso: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). È necessario specificare un vettore di inizializzazione, lungo esattamente la metà della dimensione del blocco di cifratura. Se la nostra chiave di crittografia viene utilizzata solo per crittografare un messaggio (anche se composto da più blocchi), è possibile specificare in sicurezza un vettore di inizializzazione nullo. Per crittografare i messaggi di handshake utilizziamo ogni volta una chiave separata.
Il controllo della firma gost3410.verify() è triviale: passiamo la curva ellittica all'interno della quale operiamo (che fissiamo nel nostro protocollo GOSTIM), la chiave pubblica del firmatario (non dimentichiamo che deve essere una tupla di due 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 che abbiamo fatto durante il controllo, solo in modo simmetrico: la firma sulle nostre chiavi invece della verifica, e così via...
Successivamente, nell'iniziatore prepariamo e inviamo il messaggio handshake2 per il handshake, compiendo le stesse operazioni che abbiamo fatto durante la verifica, solo in modo simmetrico: firmando le nostre chiavi invece di verificarle, 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)Quando la sessione è stabilita, vengono generati le chiavi di trasporto (una chiave separata per la crittografia, per l'autenticazione, per ciascuna delle parti), si inizializza il Kuznechik per la decrittografia e la verifica del MAC:
499 # Esegui il mittente del messaggio 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 # Aspetta i messaggi di test {{{ 521 while True: 522 data = await reader.read(MaxMsgLen) ... 530 msg, tail = Msg().decode(buf) ... 537 try: 538 await msg_receiver( 539 msg.value, 540 nonce_expected, 541 macer, 542 encrypter, 543 peer_name, 544 ) 545 except ValueError as err: 546 logging.warning("%s: %s", err) 547 break 548 nonce_expected += 1 549 # }}}La coroutine msg_sender ora crittografa i messaggi prima di inviarli nella connessione TCP. Ogni messaggio ha un nonce che aumenta monotonamente, fungendo anche da vettore di inizializzazione per la crittografia in modalità contatore. Ad ogni messaggio e blocco di messaggi verranno garantiti valori diversi del contatore.
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 inaspettato") 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(text)Conclusione
GOSTIM è destinato esclusivamente a scopi didattici (dato che non ha copertura testuale, almeno)! È possibile scaricare il codice sorgente del programma (Hash Stribog-256: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Come tutti i miei progetti, tipo , , , , GOSTIM è completamente , distribuito secondo i termini della .
, , membro , sviluppatore Python/Go, specialista principale .
Fonte: habr.com
