Fiind dezvoltator al bibliotecii (primitivi criptografici GOST pe Python pur), primesc adesea întrebări despre cum să implementez rapid un schimb de mesaje sigur. Mulți consideră criptografia aplicată o chestiune destul de simplă și că apelul .encrypt() al unui algoritm de criptare va fi suficient pentru a trimite in siguranță pe un canal de comunicare. Alții consideră că criptografia aplicată este o preocupare pentru câțiva, și că este acceptabil ca companii mari precum Telegram, cu matematicieni olimpici un protocol sigur.
Toate acestea m-au determinat să scriu acest articol pentru a arăta că implementarea protocoalelor criptografice și a IM-ului sigur nu este o sarcină atât de complicată. Totuși, nu ar trebui să inventați protocoale proprii de autentificare și de schimb de chei.
În articol se va detalia , , SIGMA-I IPsec IKE (despre care am mai scris anterior Designul IM-ului
Pentru început, trebuie să înțelegem cum va arăta IM-ul nostru. Pentru simplificare, să presupunem că va fi o rețea peer-to-peer, fără niciun fel de detectare a participanților. Vom specifica manual la ce adresă: port să ne conectăm pentru a comunica cu interlocutorul.
Înțeleg că, în acest moment, presupunerea privind disponibilitatea unei conexiuni directe între două computere arbitrare este o limitare semnificativă a aplicabilității IM-ului în practică. Dar cu cât mai mulți dezvoltatori vor implementa soluții NAT-traversal, cu atât mai mult vom rămâne în Internetul IPv4, cu o probabilitate descurajantă de conectare între computere arbitrare. Cât timp ne va mai lipsi IPv6 acasă și la muncă?
Înțeleg că, în prezent, presupunerea existenței unei conexiuni directe între două calculatoare aleatorii reprezintă o limitare semnificativă a aplicabilității IM în practică. Cu cât mai mulți dezvoltatori vor implementa diverse soluții de traversare a NAT-ului, cu atât mai mult vom rămâne în Internetul IPv4, cu o probabilitate frustrantă de conectivitate între calculatoare aleatorii. Cât timp mai trebuie să suportăm absența IPv6 acasă și la birou?
Vom avea o rețea friend-to-friend: toți interlocutorii posibili trebuie să fie cunoscuți dinainte. Pe de o parte, acest lucru simplifică foarte mult lucrurile: ne prezentăm, găsim sau nu numele/cheia, ne deconectăm sau continuăm activitatea, știind cine este interlocutorul. Pe de altă parte, în general, este sigur și exclude multe atacuri.
Interfața IM va fi similară cu soluțiile clasice. , care îmi plac foarte mult pentru minimalismul și filosofia Unix-way. Programul IM creează pentru fiecare interlocutor un director cu trei socket-uri Unix domain:
- in — în care se înregistrează mesajele trimise interlocutorului;
- out — din care se citesc mesajele primite de la interlocutor;
- state — citind din el, aflăm dacă interlocutorul este conectat acum, adresa/portul de conexiune.
În plus, se creează un socket conn, înregistrând în care host port, inițiem conexiunea cu interlocutorul de la distanță.
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Această abordare permite realizarea unor implementări independente ale transportului IM și ale interfeței utilizatorului, deoarece la gusto și culori colegii nu se potrivesc, nu poți mulțumi pe toată lumea. Folosind și / sau , poți obține o interfață multi-fereastră cu evidențiere sintactică. Iar cu ajutorul poți obține un șir compatibil cu GNU Readline pentru introducerea mesajelor.
În realitate, proiectele suckless folosesc fișiere FIFO. Personal, nu am reușit să înțeleg cum să lucrez cu fișierele în asyncio concurent fără a folosi fire dedicate (pentru astfel de lucruri folosesc de mult timp un limbaj ). De aceea am decis să mă descurc cu socket-uri unix domain. Din păcate, aceasta elimină posibilitatea de a face echo 2001:470:dead::babe 6666 > conn. Am rezolvat această problemă folosind : echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
Protocolul inițial nesigur
Ca transport este utilizat TCP: acesta garantează livrarea și ordinea acesteia. UDP nu garantează nici una, nici alta (ce ar fi util atunci când se aplică criptografia), iar suportul nu este disponibil în Python din boxă.
Din păcate, TCP nu are noțiunea de mesaj, ci doar un flux de bytes. Prin urmare, este necesar să inventăm un format pentru mesaje, astfel încât acestea să poată fi separate în acest flux. Putem conveni să folosim caracterul de sfârșit de linie. Pentru început, va fi suficient, totuși, când vom începe să criptăm mesajele noastre, acest caracter poate apărea oriunde în textul criptat. De aceea, în rețele sunt populare protocoalele care trimit mai întâi lungimea mesajului în bytes. De exemplu, în Python există din cutie xdrlib care permite lucrul cu un astfel de format. .
Nu vom lucra corect și eficient cu citirea din TCP — vom simplifica codul. Citim datele din socket într-un ciclu infinit, până când decodificăm mesajul complet. Ca format pentru această abordare, putem folosi atât JSON, cât și XML. Dar, atunci când va fi adăugată criptografia, datele vor trebui să fie semnate și autentificate — și acest lucru va necesita o reprezentare identică, byte la byte a obiectelor, lucru pe care JSON/XML nu îl garantează (rezultatul dumps-ului poate diferi).
XDR este potrivit pentru această sarcină, totuși aleg ASN.1 cu codare DER și biblioteca, deoarece vom avea la dispoziție obiecte de nivel înalt cu care este adesea mai plăcut și convenabil de lucrat. Spre deosebire de schemaless , sau , ASN.1 va verifica automat datele în raport cu o schemă rigid definită.
# 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))),
))
Mesajul acceptat va fi Msg: fie MsgText text (încă cu un singur câmp de text), fie mesajul de handshake MsgHandshake (în care se transmite numele interlocutorului). Acum pare complicat, dar acesta este un fundament pentru viitor.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│MsgHandshake(IdA) │
│─────────────────>│
│ │
│MsgHandshake(IdB) │
│<─────────────────│
│ │
│ MsgText() │
│─────────────────>│
│ │
│ MsgText() │
│<─────────────────│
│ │
IM fără criptografie
După cum am menționat deja, pentru toate operațiile cu socket-uri va fi utilizată biblioteca asyncio. Să declarăm ce așteptăm în momentul lansării:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Numele nostru peer",
)
parser.add_argument(
"--their-names",
required=True,
help="Numele peer-urilor lor, separate prin virgulă",
)
parser.add_argument(
"--bind",
default="::1",
help="Adresa pe care ascultăm",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Portul pe care ascultăm",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
Se setează un nume propriu (—our-name alice). Numele tuturor interlocutorilor așteptați sunt enumerate prin virgulă (—their-names bob,eve). Pentru fiecare dintre interlocutori, se creează un director cu socket-uri Unix, precum și câte o corutină pentru fiecare 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"))
Mesajele primite de la utilizator prin socket-ul in sunt trimise în cozi 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"))
Mesajele primite de la interlocutori sunt trimise în cozi OUT_QUEUES, din care datele sunt scrise în socket-ul 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()
Când se citește din socket-ul state, programul caută în dictionarul PEER_ALIVE adresa interlocutorului. Dacă nu există conexiune cu interlocutorul, se scrie un șir gol.
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()
Când se scrie adresa în socket-ul conn, se lansează funcția „inițiatorului” conexiunii:
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))
Să analizăm inițiatorul. La început, el deschide conexiunea către gazda/portul specificat și trimite un mesaj handshake cu numele său:
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 # Messaj de handshake {{{
135 writer.write(Msg(("handshake", MsgHandshake((
136 ("numePeer", OUR_NAME),
137 )))).encode())
138 # }}}
139 await writer.drain()
Apoi, așteaptă un răspuns de la partea remote. Încearcă să decodifice răspunsul primit conform schemei Msg ASN.1. Presupunem că mesajul întreg va fi trimis într-un singur segment TCP și îl vom primi atomic la apelul .read(). Verificăm că mesajul primit este cu adevărat un mesaj de handshake.
141 # Așteptând mesajul Handshake {{{
142 data = await reader.read(256)
143 if data == b"":
144 logging.warning("%s: no answer, disconnecting", _id)
145 writer.close()
146 return
147 try:
148 msg, _ = Msg().decode(data)
149 except ASN1Error:
150 logging.warning("%s: undecodable answer, disconnecting", _id)
151 writer.close()
152 return
153 logging.info("%s: got %s message", _id, msg.choice)
154 if msg.choice != "handshake":
155 logging.warning("%s: unexpected message, disconnecting", _id)
156 writer.close()
157 return
158 # }}}
Verificăm dacă numele interlocutorului primit ne este cunoscut. Dacă nu, atunci rupem conexiunea. Verificăm dacă aveam deja o conexiune stabilită cu el (interlocutorul a dat din nou comanda de conectare la noi) și o închidem. În IN_QUEUES sunt plasate șiruri de caractere Python cu textul mesajului, dar există o valoare specială None, care semnalizează corutina msg_sender să înceteze activitatea, astfel încât să uite de writer-ul său asociat cu vechea conexiune TCP.
159 msg_handshake = msg.value
160 peer_name = str(msg_handshake["numePeer"])
161 if peer_name not in THEIR_NAMES:
162 logging.warning("nume peer necunoscut: %s", peer_name)
163 writer.close()
164 return
165 logging.info("%s: sesiune stabilită: %s", _id, peer_name)
166 # Rulăm trimisătorul de mesaj text, inițializăm decoder-ul de transport {{{
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 primește mesajele ieșitoare (plasate în coadă din socketul de intrare), le serializează într-un mesaj MsgText și le trimite prin conexiunea TCP. Aceasta poate fi ruptă în orice moment — vom prinde asta în mod explicit.
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: a trimis un mesaj de %d caractere", peer_name, len(text))
La final, initiatorul intră într-un ciclu infinit de citire a mesajelor din socket. Verifică dacă acestea sunt mesaje text și le pune în OUT_QUEUES, din care vor fi trimise către socketul out al corespondenților respectivi. De ce nu putem pur și simplu să facem .read() și să decodificăm mesajul? Pentru că nu putem exclude situația în care mai multe mesaje de la utilizator sunt agregate în buffer-ul sistemului de operare și trimise într-un singur segment TCP. Decodificăm primul, dar în buffer poate rămâne o parte din următorul. În orice situație neprevăzută, închidem conexiunea TCP și oprim coroutinele msg_sender (prin trimiterea None în OUT_QUEUES).
174 buf = b""
175 # Așteaptă mesaje de test {{{
176 while True:
177 data = await reader.read(MaxMsgLen)
178 if data == b"":
179 break
180 buf += data
181 if len(buf) > MaxMsgLen:
182 logging.warning("%s: dimensiunea maximă a buffer-ului a fost depășită", _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: mesaj %s neașteptat", _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: deconectare: %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: mesaj de %d caractere primit", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Să ne întoarcem la codul principal. După ce am creat toate coroutinele, în momentul în care pornesc programul, pornim serverul TCP. Pentru fiecare conexiune stabilită, creează o corutină responder.
logging.basicConfig(
level=logging.INFO,
format="%(levelname)s %(asctime)s: %(funcName)s: %(message)s",
)
loop = asyncio.get_event_loop()
server = loop.run_until_complete(asyncio.start_server(responder, args.bind, args.port))
logging.info("Ascult pe: %s", server.sockets[0].getsockname())
loop.run_forever()
responderul este similar cu initiatorul și execută în oglindă toate aceleași acțiuni, dar ciclul infinit de citire a mesajelor este lansat imediat, pentru simplitate. Momentan, protocolul de handshake trimite câte un mesaj de fiecare parte, dar, ulterior, vor fi două din partea initiatorului conexiunii, după care poate urma imediat trimiterea mesajelor text.
72 async def responder(reader, writer):
73 _id = writer.get_extra_info("peername")
74 logging.info("%s: connected", _id)
75 buf = b""
76 msg_expected = "handshake"
77 peer_name = None
78 while True:
79 # Read until we get Msg message {{{
80 data = await reader.read(MaxMsgLen)
81 if data == b"":
82 logging.info("%s: closed connection", _id)
83 break
84 buf += data
85 if len(buf) > MaxMsgLen:
86 logging.warning("%s: max buffer size exceeded", _id)
87 break
88 try:
89 msg, tail = Msg().decode(buf)
90 except ASN1Error:
91 continue
92 buf = tail
93 # }}}
94 if msg.choice != msg_expected:
95 logging.warning("%s: unexpected %s message", _id, msg.choice)
96 break
97 if msg_expected == "text":
98 try:
99 await msg_receiver(msg.value, peer_name)
100 except ValueError as err:
101 logging.warning("%s: %s", err)
102 break
103 # Process Handshake message {{{
104 elif msg_expected == "handshake":
105 logging.info("%s: got %s message", _id, msg_expected)
106 msg_handshake = msg.value
107 peer_name = str(msg_handshake["peerName"])
108 if peer_name not in THEIR_NAMES:
109 logging.warning("unknown peer name: %s", peer_name)
110 break
111 writer.write(Msg(("handshake", MsgHandshake((
112 ("peerName", OUR_NAME),
113 )))).encode())
114 await writer.drain()
115 logging.info("%s: session established: %s", _id, peer_name)
116 peer_alive = PEER_ALIVES.pop(peer_name, None)
117 if peer_alive is not None:
118 peer_alive.close()
119 await IN_QUEUES[peer_name].put(None)
120 PEER_ALIVES[peer_name] = writer
121 asyncio.ensure_future(msg_sender(peer_name, writer))
122 msg_expected = "text"
123 # }}}
124 logging.info("%s: disconnecting", _id)
125 if msg_expected == "text":
126 IN_QUEUES[peer_name].put(None)
127 writer.close()
Protocolul sigur
A venit timpul să securizăm comunicația noastră. Ce înțelegem prin securitate și ce dorim:
- confidențialitatea mesajelor transmise;
- autenticitatea și integritatea mesajelor transmise — orice modificare ar trebui să fie detectată;
- protecția împotriva atacurilor de tip replay — dispariția sau repetarea mesajelor ar trebui să fie detectată (și decidem să întrerupem conexiunea);
- identificarea și autentificarea interlocutorilor pe baza cheilor publice preîncărcate — am decis anterior că construim o rețea friend-to-friend. Numai după autentificare vom înțelege cu cine comunicăm;
- existența proprietăți (PFS) — compromiterea cheii noastre de semnătură pe termen lung nu ar trebui să ducă la posibilitatea de a citi întreaga corespondență anterioară. Înregistrarea traficului interceptat devine inutilă;
- valabilitatea mesajelor (de transport și de handshake) este valabilă doar în cadrul unei sesiuni TCP. Inserarea mesajelor corect semnate/autentificate dintr-o altă sesiune (chiar și cu aceleași părți) nu ar trebui să fie posibilă;
- un observator pasiv nu ar trebui să vadă nici identificatorii utilizatorilor, nici cheile publice pe termen lung transmise, nici hash-urile acestora. O oarecare anonimitate față de un observator pasiv.
Este surprinzător, dar acest minimum este dorit practic de toată lumea în orice protocol de handshake, iar extrem de puțin din ceea ce este enumerat este în cele din urmă realizat pentru protocoalele „home-made”. Așadar, nici acum nu vom inventa ceva nou. Aș recomanda cu tărie să folosim pentru construirea protocoalelor, dar să alegem ceva mai simplu.
Cele mai populare sunt două protocoale:
- — un protocol extrem de complex, cu o istorie lungă de erori, defecte, vulnerabilități, o planificare slabă, complexitate și neajunsuri (deși, asta nu se aplică mult lui TLS 1.3). Dar nu îl considerăm din cauza complexității excesive.
- de — nu au probleme criptografice grave, deși nu sunt nici ușoare. Dacă citim despre IKEv1 și IKEv2, sursa lor este , standardele ISO/IEC IS 9798-3 și protocoalele SIGMA (SIGn-and-MAc) — destul de simple pentru implementare în seara unei zile.
Ce are SIGMA, ca ultima verigă a dezvoltării protocoalelor STS/ISO, care este bun? Îndeplinește toate cerințele noastre (inclusiv „ascunderea” identificatorilor interlocutorilor), nu are probleme criptografice cunoscute. Este minimalist — eliminarea oricărui element din mesajul protocolului va duce la nesiguranța acestuia.
Să ne îndreptăm de la protocolul home-made cel mai simplu către SIGMA. Cea mai fundamentală operațiune care ne interesează este : o funcție, la ieșire, care le va oferi ambelor părți aceeași valoare, care poate fi folosită ca cheie simetrică. Fără a intra în detalii: fiecare parte generează un a pair efemer (folosit doar în cadrul unei sesiuni) de chei (o cheie publică și una privată), își schimbă cheile publice, invocă funcția de negociere, la care introduc cheia privată și cheia publică a interlocutorului.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│ IdA, PubA │ ╔════════════════════╗
│───────────────│ ║PrvA, PubA = DHgen()║
│ │ ╚════════════════════╝
│ IdB, PubB │ ╔════════════════════╗
│<───────────────│ ║PrvB, PubB = DHgen()║
│ │ ╚════════════════════╝
────┐ ╔═══════╧════════════╗
│ ║Key = DH(PrvA, PubB)║
<───┘ ╚═══════╤════════════╝
│ │
│ │
Oricine poate interveni și poate înlocui cheile publice cu ale sale, deoarece acest protocol nu are autentificare a partenerilor. Să adăugăm o semnătură folosind cheile de lungă durată.
┌─────┐ ┌─────┐
│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) ║ │
│ ╚═════════════════════╝ │
│ │
Această semnătură nu este adecvată, deoarece nu este legată de o sesiune specifică. Astfel de mesaje „se potrivesc” și pentru sesiuni cu alți participanți. Tot contextul trebuie să fie semnat. Aceasta impune, de asemenea, adăugarea unei noi trimiterea de mesaj din partea lui A.
În plus, este critic să adăugăm, sub semnătură, și propriul identificator, deoarece altfel putem schimba IdXXX și resemna mesajul cu cheia unui alt cunoscut. Pentru a preveni , este necesar ca elementele de sub titlu să se afle în locuri clar definite, conform semnificației lor: dacă A semnează (PubA, PubB), atunci B trebuie să semneze (PubB, PubA). Aceasta subliniază și importanța alegerii structurii și formatului datelor serializate. De exemplu, mulțimile în codificarea ASN.1 DER sunt sortate: SET OF(PubA, PubB) va fi identic cu SET OF(PubB, PubA).
┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └──┬──┘ │ IdA, PubA │ ╔═══════════════════════════╗ │────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚═══════════════════════════╝ │IdB, PubB, sign(SignPrvB, (IdB, PubA, PubB)) │ ╔═══════════════════════════╗ │<────────────────────────────────────────────│ ║SignPrvB, SignPubB = load()║ │ │ ║PrvB, PubB = DHgen() ║ │ │ ╚═══════════════════════════╝ │ sign(SignPrvA, (IdA, PubB, PubA)) │ ╔═════════════════════╗ │────────────────────────────────────────────>│ ║verify(SignPubB, ...)║ │ │ ║Key = DH(PrvA, PubB) ║ │ │ ╚═════════════════════╝ │ │
Cu toate acestea, încă nu am „dovedit” că am generat aceeași cheie comună pentru această sesiune. De principiu, am putea să sărim peste acest pas — primul mesaj de transport va fi invalid, dar dorim ca, la finalizarea handshake-ului, să avem siguranța că totul este realmente convenit. La momentul actual, avem în mâinile noastre protocolul ISO/IEC IS 9798-3.
Am putea semna și cheia generată. Acest lucru este riscant, deoarece algoritmul de semnătură pe care îl folosim ar putea avea scurgeri (chiar dacă sunt doar biți pentru semnătură, totuși sunt scurgeri). Putem semna hash-ul cheii generate, dar scurgerea chiar și a hash-ului din cheia generată poate fi valoroasă în cazul unei atacuri de tip brute-force asupra funcției de generare. SIGMA utilizează o funcție MAC care autentifică identificatorul expeditorului.
┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └──┬──┘ │ IdA, PubA │ ╔═══════════════════════════╗ │─────────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚═══════════════════════════╝ │IdB, PubB, sign(SignPrvB, (PubA, PubB)), MAC(IdB) │ ╔═══════════════════════════╗ │<─────────────────────────────────────────────────│ ║SignPrvB, SignPubB = load()║ │ │ ║PrvB, PubB = DHgen() ║ │ │ ╚═══════════════════════════╝ │ │ ╔═════════════════════╗ │ sign(SignPrvA, (PubB, PubA)), MAC(IdA) │ ║Key = DH(PrvA, PubB) ║ │─────────────────────────────────────────────────>│ ║verify(Key, IdB) ║ │ │ ║verify(SignPubB, ...)║ │ │ ╚═════════════════════╝ │ │
Ca optimizare, unii ar putea dori să reutilizeze cheile lor efemere (ceea ce, desigur, este defavorabil pentru PFS). De exemplu, am generat o pereche de chei, am încercat să ne conectăm, dar TCP nu a fost disponibil sau s-a întrerupt undeva în timpul protocolului. Este păcat să pierdem entropia și resursele procesorului pe o nouă pereche. Prin urmare, vom introduce așa-numitul cookie - o valoare pseudo-randomă care va proteja împotriva posibilelor atacuri de tip replay atunci când reutilizăm cheile publice efemere. Din cauza legăturii între cookie și cheia publică efemeră, cheia publică a participanților opuși poate fi omisă din semnătură, nefiind necesară.
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
În final, dorim să obținem confidențialitatea identificatorilor interlocutorilor noștri față de un observator pasiv. Pentru aceasta, SIGMA propune mai întâi schimbul de chei efemere, pentru a genera o cheie comună, cu care se vor cripta mesajele de autentificare și identificare. SIGMA descrie două variante:
- SIGMA-I — protejează inițiatorul de atacuri active, iar respondentul de atacuri pasive: inițiatorul autentifică respondentul și, dacă ceva nu se potrivește, nu își dezvăluie identitatea. Respondentul își dezvăluie identitatea doar dacă se începe un protocol activ. Un observator pasiv nu va afla nimic;
SIGMA-R — protejează respondentul de atacuri active, iar inițiatorul de atacuri pasive. Totul este invers, dar în acest protocol sunt transmise deja patru mesaje de tip handshake.Alegem SIGMA-I, deoarece este mai asemănător cu ceea ce ne așteptăm de la lucrurile obișnuite client-server: clientul recunoaște doar serverul autentificat, iar serverul deja cunoaște totul. În plus, este mai ușor de implementat din cauza numărului mai mic de mesaje de negociere. Tot ce introducem în protocol este criptarea unei părți a mesajului și transferul identificatorului A în partea criptată a ultimului mesaj:
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
- Pentru semnare se utilizează ГОСТ Р algoritmul cu chei de 256 biți.
- Pentru generarea cheii comune se folosește 34.10-2012 VKO.
- Ca MAC se utilizează CMAC. Tehnic, acesta este un mod special de funcționare al unui cifrator pe blocuri, descris în ГОСТ Р 34.13-2015. Ca funcție de criptare pentru acest mod — (34.12-2015).
- Ca identificator pentru interlocutor se folosește un hash al cheii sale publice. Ca hash se utilizează (34.11-2012 256 biți).
După strângerea mâinilor, vom avea o cheie comună convenită. O putem folosi pentru criptarea autentificată a mesajelor de transport. Această parte este foarte simplă și este greu să greșim: incrementăm numărătorul mesajelor, criptăm mesajul, autentificăm (MAC) numărătorul și textul criptat, apoi trimitem. Când primim mesajul, verificăm că numărătorul are o valoare așteptată, autentificăm textul criptat cu numărătorul, apoi decriptăm. Cu ce cheie să criptăm mesajele de strângere a mâinilor, cele de transport, cu ce să autentificăm? Folosirea unei singure chei pentru toate aceste sarcini este periculoasă și nerațională. Este necesar să generăm chei folosind funcții specializate (funcția de derivare a cheii). Din nou, să nu ne complicăm și să inventăm ceva: este bine cunoscută, bine studiată și nu are probleme cunoscute. Din păcate, în biblioteca nativă Python nu există această funcție, așa că vom folosi pachet. HKDF folosește în interior , care, la rândul său, folosește o funcție hash. Un exemplu de implementare în Python pe pagina Wikipedia ocupă câteva linii de cod. La fel ca în cazul 34.10-2012, vom folosi pentru funcția hash Stribog-256. Ieșirea funcției noastre de negociere a cheilor se va numi cheie de sesiune, din care vor fi generate cheile simetrice lipsă:
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")Structuri/schema
Să examinăm ce structuri ASN.1 am obținut pentru a transmite toate aceste date:
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 — ceea ce va fi semnat (to be signed). HandshakeTBE — ceea ce va fi criptat (to be encrypted). Aș dori să atrag atenția asupra câmpului ukm din MsgHandshake1. 34.10 VKO, pentru o și mai mare randomizare a cheilor generate, include parametrul UKM (material de cheia utilizatorului) — pur și simplu o entropie suplimentară.
Adăugarea criptografiei în cod
Vom analiza doar modificările aduse codului original, deoarece structura a rămas aceeași (de fapt, inițial a fost scrisă implementarea finală, iar apoi din aceasta a fost eliminată întreaga criptografie).
Având în vedere că autentificarea și identificarea interlocutorilor se va face pe baza cheilor publice, acum trebuie să le stocăm undeva pe termen lung. Pentru simplitate, folosim un JSON de acest tip:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }our — perechea noastră de chei, cheile private și publice în hexazecimal. their — numele interlocutorilor și cheile lor publice. Vom modifica argumentele din linia de comandă și vom adăuga postprocesarea datelor 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="Generate JSON with our new keypair", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON with our and their keys", ) parser.add_argument( "--bind", default="::1", help="Address to listen on", ) parser.add_argument( "--port", type=int, default=6666, help="Port to listen on", ) 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) # Parse and unmarshal our and their keys {{{ 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), } # }}}Cheia privată a algoritmului 34.10 este un număr aleator. Are dimensiunea de 256 de biți pentru curbele eliptice de 256 de biți. PyGOST nu funcționează cu un set de octeți, ci cu , de aceea cheia noastră privată (urandom(32)) trebuie să fie transformată în număr, folosind gost3410.prv_unmarshal(). Cheia publică este calculată determinist din cheia privată folosind gost3410.public_key(). Cheia publică 34.10 este formată din două numere mari, care trebuie de asemenea transformate într-o secvență de octeți pentru a facilita stocarea și transmiterea, folosind gost3410.pub_marshal().
După citirea fișierului JSON, cheile publice trebuie, respectiv, transformate înapoi, folosind gost3410.pub_unmarshal(). Deoarece vom primi identificatorii interlocutorilor sub formă de hash din cheia publică, aceștia pot fi calculați din timp și plasați într-un dicționar pentru o căutare rapidă. Hash-ul Stribog-256 este gost34112012256.GOST34112012256(), care satisface pe deplin interfața funcțiilor de hash din hashlib.
Cum s-a modificat corutina inițiatorului? Totul, conform schemei de strângere de mână: generăm un cookie (128 de biți sunt mai mult decât suficienți), o pereche de chei efemere 34.10, care va fi utilizată pentru funcția VKO de negociere a cheilor.
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 # Generăm cheia noastră publică efemeră și cookie-ul, trimitem mesajul 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()- așteptăm un răspuns și decodificăm mesajul Msg primit;
- ne asigurăm că am primit handshake1;
- decodificăm cheia publică efemeră a părții opuse și calculăm cheia de sesiune;
- generăm cheile simetrice necesare pentru procesarea părții TBE a mesajului.
423 logging.info("%s: got %s message", _id, msg.choice) 424 if msg.choice != "handshake1": 425 logging.warning("%s: unexpected message, disconnecting", _id) 426 writer.close() 427 return 428 # }}} 429 msg_handshake1 = msg.value 430 # Validăm mesajul de 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 este un număr de 64 de biți (urandom(8)), care necesită de asemenea deserializare din formatul de bytes, utilizând gost3410_vko.ukm_unmarshal(). Funcția VKO pentru 34.10-2012 de 256 biți este gost3410_vko.kek_34102012256() (KEK — cheie de criptare).
Cheia de sesiune generată este deja o secvență pseudo-aleatoare de 256 de biți. De aceea, poate fi folosită imediat în funcția HKDF. Deoarece GOST34112012256 satisface interfața hashlib, poate fi utilizată direct în clasa Hkdf. Nu specificăm sare (primul argument Hkdf), deoarece cheia generată, din cauza efemerității perechilor de chei implicate, va fi diferită pentru fiecare sesiune și deja are suficientă entropie. kdf.expand() returnează în mod implicit chei de 256 de biți, necesare pentru Kuznechik ulterior.
Apoi, se verifică părțile TBE și TBS ale mesajului primit:
- se calculează și se verifică MAC-ul pe textul cifrat primit;
- se decriptează textul cifrat;
- se decodifică structura TBE;
- din aceasta se extrage identificatorul interlocutorului și se verifică dacă este cunoscut;
- se calculează și se verifică MAC-ul pe acest identificator;
- se verifică semnătura deasupra structurii TBS, care include cookie-urile ambelor părți și cheia publică efemeră a celeilalte părți. Semnătura este verificată cu cheia de semnătură pe termen lung a interlocutorului.
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, disconnecting", _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("invalid MAC") 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("can not decode 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("unknown identity") 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("invalid identity MAC") 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("invalid signature") 174 return peer["name"]Așa cum am menționat anterior, 34.13-2015 descrie diferite din 34.12-2015. Printre acestea se numără modul de generare a autentificării, calcularea MAC-ului. În PyGOST, acesta este gost3413.mac(). Acest mod necesită transmiterea funcției de criptare (care primește și returnează un singur bloc de date), dimensiunea blocului de criptare și, desigur, datele în sine. De ce nu se poate hardcode dimensiunea blocului de criptare? 34.12-2015 descrie nu doar cifra de 128 de biți Kuznechik, ci și cifra de 64 de biți — o variantă puțin modificată a GOST 28147-89, creată încă în KGB și care are în continuare unul dintre cele mai ridicate praguri de securitate.
Grasshopper se inițiază cu gost.3412.GOST3412Kuznechik(key) prin apel și returnează un obiect cu metodele .encrypt() și .decrypt() potrivite pentru a fi transmise în funcțiile 34.13. MAC-ul se calculează după cum urmează: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Pentru a compara MAC-ul calculat cu cel primit, nu se poate folosi compararea obișnuită (==) a șirurilor de octeți, deoarece această operație poate provoca scurgeri de timp în comparație, ceea ce, în general, poate duce la vulnerabilități fatale de tip atacuri asupra TLS-ului. În Python există o funcție specială hmac.compare_digest pentru asta.
Funcția de cifrare pe bloc poate cripta doar un singur bloc de date. Pentru un număr mai mare, chiar și de lungimi necruțate, este necesar să se utilizeze un mod de criptare. În 34.13-2015 sunt descrise următoarele: ECB, CTR, OFB, CBC, CFB. Fiecare are domeniile și caracteristicile sale de aplicare. Din păcate, nu avem încă moduri de criptare standardizate modul de contador (CTR): nu necesită completarea la dimensiunea blocului, poate fi paralelizat, folosește doar funcția de criptare și poate fi utilizat în siguranță pentru a cripta un număr mare de mesaje (spre deosebire de CBC, unde coliziunile încep relativ repede). Verificarea semnăturii gost3410.verify() este trivială: trimitem curba eliptică în limita căreia lucrăm (pe care o fixăm în protocolul nostru GOSTIM), cheia publică a semnatarului (nu uitați că aceasta trebuie să fie un tuplu format din două numere mari, nu un șir de octeți), hash-ul 34.11-2012 și semnătura primit.
Apoi, initiatorul pregătește și trimite un mesaj handshake2, efectuând aceleași acțiuni ca și atunci când am verificat, doar simetric: semnătură pe cheile sale în loc de verificare, etc...
Verificarea semnăturii gost3410.verify() este trivială: transmitem curba eliptică în cadrul căreia lucrăm (pe care o fixăm pur și simplu în protocolul nostru GOSTIM), cheia publică a semnatarului (să nu uităm că aceasta trebuie să fie un tuplu format din două numere mari, nu un șir de bytes), hash-ul conform standardului 34.11-2012 și semnătura primită.
Apoi, în inițiator, pregătim și trimitem mesajul handshake2 de negociere, efectuând aceleași acțiuni pe care le-am făcut la verificare, doar simetric: semnătura pe cheile noastre în loc de verificare, etc…
456 # Pregătiți și trimiteți mesajul 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: sesiune stabilită: %s", _id, peer_name)Când sesiunea este stabilită, se generează cheile de transport (o cheie separată pentru criptare, pentru autentificare, pentru fiecare dintre părți), se inițiază Kuznechik pentru decriptare și verificarea MAC-ului:
499 # Rulați trimitătorul de mesaje text, inițializați decodorul de transport {{{ 500 key_initiator_enc = kdf.expand(b"transport-initiator-enc") 501 key_initiator_mac = kdf.expand(b"transport-initiator-mac") 502 key_responder_enc = kdf.expand(b"transport-responder-enc") 503 key_responder_mac = kdf.expand(b"transport-responder-mac") ... 509 asyncio.ensure_future(msg_sender( 510 peer_name, 511 key_initiator_enc, 512 key_initiator_mac, 513 writer, 514 )) 515 encrypter = GOST3412Kuznechik(key_responder_enc).encrypt 516 macer = GOST3412Kuznechik(key_responder_mac).encrypt 517 # }}} 519 nonce_expected = 0 520 # Așteptați mesaje de test {{{ 521 while True: 522 data = await reader.read(MaxMsgLen) ... 530 msg, tail = Msg().decode(buf) ... 537 try: 538 await msg_receiver( 539 msg.value, 540 nonce_expected, 541 macer, 542 encrypter, 543 peer_name, 544 ) 545 except ValueError as err: 546 logging.warning("%s: %s", err) 547 break 548 nonce_expected += 1 549 # }}}corutina msg_sender acum criptează mesajele înainte de a le trimite prin conexiunea TCP. Fiecare mesaj are un nonce care crește monoton, acționând și ca vector de inițializare pentru criptarea în modul de contor. Fiecare mesaj și bloc de mesaj vor avea garantat valori diferite ale contorului.
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 += 1Mesajele primite sunt procesate de corutina msg_receiver, care se ocupă de autentificare și decriptare:
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("valoare nonce neașteptată") mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode()) if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])): raise ValueError("MAC invalid") plaintext = ctr( encrypter, KUZNECHIK_BLOCKSIZE, bytes(payload["ciphertext"]), long2bytes(nonce_expected, 8), ) text = plaintext.decode("utf-8") await OUT_QUEUES[peer_name].put(text)Concluzie
GOSTIM este destinat exclusiv pentru scopuri educaționale (deoarece nu este acoperit de teste, cel puțin)! Codul sursă al programului poate fi descărcat (Hash Stribog-256: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Ca și toate proiectele mele, tip: , , , , GOSTIM este complet , distribuit conform termenilor .
, , membru , dezvoltator Python/Go, specialist principal .
Sursa: habr.com
