Si një zhvillues i bibliotekës (Primitive kriptografike GOST në Python të pastër), shpesh marr pyetje se si mund të realizohet një shkëmbim mesazhesh i sigurt. Shumë njerëz e konsiderojnë kriptografinë aplikative një gjë mjaft të thjeshtë, dhe thirrja .encrypt() e një algoritmi bllokues do të ishte e mjaftueshme për dërgimin e sigurt përmes një kanali komunikimi. Të tjerë mendojnë se kriptografia aplikative është një fushë për pak, dhe është e pranueshme që kompanitë e pasura si Telegram me matematicienët olimpikë një protokoll të sigurt.
E gjithë kjo më nxit të shkruaj këtë artikull, për të treguar se realizimi i protokolleve kriptografike dhe IM-eve të sigurta nuk është një detyrë aq e vështirë. Sidoqoftë, nuk ia vlen të shpikni protokolle të reja verifikimi dhe shkëmbimi çelesh.
Në artikull do të shkruhet , , me të verifikimit dhe shkëmbimit të çelesh (në bazë të cilit është realizuar ), duke përdorur ekskluzivisht algoritmet kriptografike GOST të bibliotekës PyGOST dhe kodimin ASN.1 të mesazheve nga biblioteka (për të cilën kam ). Kushti i nevojshëm: ai duhet të jetë aq i thjeshtë, saqë të mund të shkruhet nga e para për një mbrëmje (ose një ditë pune), ndryshe nuk është një program i thjeshtë. Sigurisht që do të ketë gabime, komplekse të panevojshme, mangësi, dhe plus është programi im i parë që përdor bibliotekën asyncio.
Dizajni i IM
Për fillim, duhet të kuptojmë si do të duket IM ynë. Për thjeshtësi, le të jetë një rrjet peer-to-peer, pa ndonjë zbulim të pjesëmarrësve. Ne do të specifikojmë manualisht se në cilin adresë: port duhet të lidhemi për të komunikuar me biseduesin.
E kuptoj që, në këtë moment, supozimi për disponueshmërinë e një lidhjeje të drejtpërdrejtë midis dy kompjuterëve të rastësishëm është një kufizim i rëndësishëm në praktikën e IM. Por sa më shumë zhvillues do të zbatonjnë truket e NAT-traversal, aq më gjatë do të mbetemi në internetin IPv4, me një probabilitet shqetësues të lidhjes midis kompjuterëve të rastësishëm. Sa kohë mund të durojmë mungesën e IPv6 në shtëpi dhe punë?
Ne do kemi një rrjet miqësor të kushteve: të gjithë bisedimtarët e mundshëm duhet të njihen paraprakisht. Së pari, kjo e thjeshton shumë gjithçka: prezantohemi, gjejmë ose nuk gjejmë emrin/kyçin, çmendemi ose vazhdojmë punën duke e ditur bisedimtarin. Së dyti, në përgjithësi, është e sigurt dhe përjashton shumë sulme.
Interfejsi i IM-së do të jetë i afërt me zgjidhjet klasike. , të cilat më pëlqejnë shumë për minimalizmin e tyre dhe filozofinë Unix-way. Programi IM krijon një dosje për çdo bisedimtar me tre sokete Unix domain:
- in â nĂ« tĂ« shkruhen mesazhet qĂ« dĂ«rgohen bisedimtarit;
- out â nga ai lexohen mesazhet qĂ« pranohen nga bisedimtari;
- state â duke lexuar nga ai, mĂ«sojmĂ« nĂ«se bisedimtari Ă«shtĂ« aktualisht i lidhur, adresa/porta e lidhjes.
Për më tepër, krijohet soketi conn, duke regjistruar në të host port, ne fillojmë një lidhje me bisedimtarin e largët.
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Ky qasje lejon realizime të pavarura të transportit IM dhe ndërfaqes së përdoruesit, sepse për shijen dhe ngjyrën nuk ka shok, gjithkujt nuk i pëlqen. Duke përdorur dhe/ose , mund të marrim një ndërfaqe me shumë dritare me ndriçim sintaksor. Dhe përmes mund të kemi një linjë të input-it të përputhshme me GNU Readline për dërgimin e mesazheve.
NĂ« tĂ« vĂ«rtetĂ«, projektet suckless pĂ«rdorin skedarĂ« FIFO. Personalish, nuk arrita tĂ« kuptoj si tĂ« punoj me skedarĂ« nĂ« asyncio nĂ« mĂ«nyrĂ« konkurruese pa njĂ« mbĂ«shtetje tĂ« veçantĂ« nga njĂ« tread i veçantĂ« (pĂ«r ato gjĂ«ra pĂ«rdor tashmĂ« njĂ« gjuhĂ« ). Prandaj vendosa tĂ« pĂ«rmbahem nga soketĂ«t Unix domain. FatkeqĂ«sisht, kjo privon mundĂ«sinĂ« pĂ«r tĂ« bĂ«rĂ« echo 2001:470:dead::babe 6666 > conn. Zgjodha kĂ«tĂ« problem, duke pĂ«rdorur : echo 2001:470:dead::babe 6666 | socat â UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
Protokolli fillestar i pasigurt
Sikurse transport përdoret TCP: ai garanton dorëzimin dhe rendin e saj. UDP nuk garanton asnjërin prej tyre (çka do të ishte e dobishme kur të aplikohej kriptografia), dhe mbështetja në Python nuk është në paketën e tij.
Fatke, në TCP nuk ka koncept mesazhi, por vetëm rrjedha bajtash. Prandaj, është e nevojshme të shpikim një format për mesazhet në mënyrë që t'i ndarim ato nga njëra-tjetra në këtë rrjedhë. Mund të bien dakord të përdorim simbolin e kalimit në rresht. Për fillim, kjo do të mjaftojë, megjithatë, kur të fillojmë të enkriptojmë mesazhet tona, ky simbol mund të shfaqet kudo në tekstin e enkriptuar. Në rrjetet, prandaj, janë të njohura protokollet që dërgojnë fillimisht gjatë e mesazhit në bajta. Për shembull, në Python, menjëherë ekziston xdrlib që mundëson të punojë me një format të tillë. .
Ne nuk do tĂ« punojmĂ« nĂ« mĂ«nyrĂ« tĂ« saktĂ« dhe efektive me leximin TCP â do ta thjeshtojmĂ« kodin. LexojmĂ« tĂ« dhĂ«nat nga socket-i nĂ« njĂ« cikĂ«l tĂ« pafund, derisa tĂ« dekodojmĂ« mesazhin e plotĂ«. NjĂ« format i tillĂ« mund tĂ« pĂ«rdorĂ« JSON me XML. Por, kur tĂ« shtohet kriptografia, tĂ« dhĂ«nat do tĂ« duhet tĂ« nĂ«nshkruhen dhe autentikohen â dhe kjo do tĂ« kĂ«rkojĂ« njĂ« pĂ«rfaqĂ«sim tĂ« identik bajt pĂ«r bajt tĂ« objekteve, qĂ« nuk e sigurojnĂ« JSON/XML (rezultati i dumps mund tĂ« jetĂ« ndryshe).
XDR është i përshtatshëm për një detyrë të tillë, megjithatë, unë zgjidh ASN.1 me kodimin DER dhe bibliotekën, pasi ne do të kemi objekte me nivel të lartë me të cilat shpesh është më e këndshme dhe më e lehtë të punosh. Në dallim nga schemaless , ose , ASN.1 automatikisht do të verifikojë të dhënat përballë një skeme të caktuar me forcë.
# 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))),
))
Mesazhi i pranuar do të jetë Msg: ose MsgText tekstor (për tani me një fushë teksti), ose MsgHandshake (në të cilin dërgohet emri i biseduesit). Tani duket e komplikuar, por kjo është një përgatitje për të ardhmen.
âââââââ âââââââ
âPeerAâ âPeerBâ
ââââŹâââ ââââŹâââ
âMsgHandshake(IdA) â
ââââââââââââââââââ>â
â â
âMsgHandshake(IdB) â
ââ
â â
â MsgText() â
â<ââââââââââââââââââ
â â
IM pa kriptografi
Siç kam thënë më parë, për gjithë operacionet me socket-at do të përdoret biblioteka asyncio. Do të shpallim se çfarë presim në momentin e fillimit:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Emri ynë",
)
parser.add_argument(
"--their-names",
required=True,
help="Emrat e tyre, të ndarë me presje",
)
parser.add_argument(
"--bind",
default="::1",
help="Adresa për të dëgjuar",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Porti për të dëgjuar",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
Caktohet emri ynĂ« (âour-name alice). TĂ« gjithĂ« biseduesit e pritur shĂ«nohen me presje (âtheir-names bob,eve). PĂ«r secilin nga biseduesit, krijohet njĂ« dosje me sokete Unix, si dhe njĂ« korutinĂ« pĂ«r secilin 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"))
Mesazhet e pranuara nga përdoruesi nga soketi in dërgohen në radhët 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"))
Mesazhet e pranuara nga biseduesit dërgohen në radhët OUT_QUEUES, nga të cilat të dhënat shkruhen në soketin 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()
Gjatë leximit nga soketi state, programi kërkon në fjalorin PEER_ALIVE adresën e biseduesit. Nëse nuk ka akoma lidhje me biseduesin, regjistrohet një varg i zbrazët.
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()
Gjatë shkrimit të adresës në soketin conn, aktivizohet funksioni 'inisiator' të lidhjes:
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))
Le të shqyrtojmë iniciatorin. Fillimisht, ai natyrisht hap një lidhje me host-in/port-in e caktuar dhe dërgon një mesazh handshake me emrin e tij:
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 # Mesazhi i handshake {{{
135 writer.write(Msg(("handshake", MsgHandshake((
136 ("emriIshokut", OUR_NAME),
137 )))).encode())
138 # }}}
139 await writer.drain()
Pastaj, pret përgjigjen nga ana e largët. Provon të dekodojë përgjigjen e ardhur sipas skemës Msg ASN.1. Supozojmë se të gjithë mesazhi do të dërgohet në një segment TCP dhe ne do ta marrim atë atomikisht kur të thërrasim .read(). Kontrollojmë që kemi marrë pikërisht mesazhin e handshake.
141 # Prit për mesazh 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 # }}}
Kontrollojmë që emri i ardhur i bashkëbiseduesit është i njohur për ne. Nëse jo, atëherë shkëputim lidhjen. Kontrollojmë nëse kemi pasur tashmë një lidhje me të (bashkëbiseduesi përsëri jep urdhrin për t'u lidhur me ne) dhe e mbyllim atë. Në IN_QUEUES vendosen vargje Python me tekstin e mesazhit, por ka një vlerë speciale None, që sinjalizon korutinën msg_sender të ndalojë punën, në mënyrë që ajo të harrojë mbi writer-in e saj, të lidhur me lidhjen TCP të vjetruar.
159 msg_handshake = msg.value
160 peer_name = str(msg_handshake["peerName"])
161 if peer_name not in THEIR_NAMES:
162 logging.warning("emri i panjohur i bashkëbiseduesit: %s", peer_name)
163 writer.close()
164 return
165 logging.info("%s: seanca e vendosur: %s", _id, peer_name)
166 # Ekzekuto mesazhin e dërguesit, inicializo dekoduesin e transportit {{{
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 merr mesazhet e daljes (tĂ« vendosura nĂ« radhĂ« nga socket-i in), i serializon ato nĂ« mesazhin MsgText dhe i dĂ«rgon pĂ«rmes lidhjes TCP. Ajo mund tĂ« ndĂ«rpritet nĂ« çdo moment â kĂ«tĂ« ne e kapim nĂ« mĂ«nyrĂ« tĂ« qartĂ«.
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((
("tekst", UTF8String(text)),
)))).encode())
try:
await writer.drain()
except ConnectionResetError:
del PEER_ALIVES[peer_name]
return
logging.info("%s: e dërguar mesazhi prej %d karakteresh", peer_name, len(text))
Në fund, iniciatori kalon në një cikël të pafund të leximit të mesazheve nga soketi. Kontrollon nëse janë mesazhe tekstuale dhe i vendos në radhën OUT_QUEUES, nga e cila ato do të dërgohen në soketin përkatës të biseduesit. Pse nuk mund të bëjmë thjesht .read() dhe të dekodojmë mesazhin? Sepse nuk përjashtohet mundësia që disa mesazhe nga përdoruesi të agregohen në tamponin e sistemit operativ dhe të dërgohen me një TCP-segment të vetëm. Ne mund të dekodojmë të parin, por pastaj në tampon mund të mbetet një pjesë nga i dyti. Në çdo situatë emergjente, ne mbyllim lidhjen TCP dhe ndalojmë korutinë msg_sender (duke dërguar None në radhën OUT_QUEUES).
174 buf = b""
175 # Prit për mesazhet e provës {{{
176 ndërsa e vërtetë:
177 data = await reader.read(MaxMsgLen)
178 nëse data == b"":
179 break
180 buf += data
181 nëse len(buf) > MaxMsgLen:
182 logging.warning("%s: madhësia maksimale e tamponit u tejkalua", _id)
183 break
184 provo:
185 msg, tail = Msg().decode(buf)
186 përjashtim ASN1Error:
187 vazhdo
188 buf = tail
189 nëse msg.choice != "text":
190 logging.warning("%s: mesazh i papritur %s", _id, msg.choice)
191 break
192 provo:
193 await msg_receiver(msg.value, peer_name)
194 përjashtim ValueError si err:
195 logging.warning("%s: %s", err)
196 break
197 # }}}
198 logging.info("%s: duke u shkëputur: %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: marrë një mesazh me %d karaktere", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Le të kthehemi në kodin kryesor. Pas krijimit të të gjitha korutinave në momentin e nisjes së programit, ne fillojmë serverin TCP. Për çdo lidhje të vendosur, ai krijon një responder (përgjigjës) korutinë.
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("Duke dëgjuar në: %s", server.sockets[0].getsockname())
loop.run_forever()
responder është i ngjashëm me iniciatorin dhe kryen të gjitha ato të njëjtat veprime, por cikli i pafund i leximit të mesazheve nis menjëherë, për thjeshtësi. Tani protokolli i dorëheqjes dërgon një mesazh nga çdo anë, por më vonë, do të dërgojë dy nga iniciatori i lidhjes, pas të cilave është menjëherë e mundur të dërgosh tekstual.
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()
Protokolli i sigurt
Ka ardhur koha pĂ«r tĂ« siguruar komunikimin tonĂ«. ĂfarĂ« nĂ«nkuptojmĂ« me sigurinĂ« dhe çfarĂ« dĂ«shirojmĂ«:
- privatësia e mesazheve të dërguara;
- autenticiteti dhe integriteti i mesazheve tĂ« dĂ«rguara â çdo ndryshim duhet tĂ« zbulohet;
- mbrojtja nga sulmet e riprodhimit (replay attack) â fakti i mungesĂ«s ose pĂ«rsĂ«ritjes sĂ« mesazheve duhet tĂ« zbulohet (dhe ne vendosim tĂ« ndĂ«rpresim lidhjen);
- identifikimi dhe autentifikimi i biseduesve pĂ«rmes çelĂ«save publikĂ« tĂ« regjistruar paraprakisht â ne tashmĂ« e kemi vendosur mĂ« parĂ« se do tĂ« krijojmĂ« njĂ« rrjet friend-to-friend. VetĂ«m pas autentifikimit do tĂ« kuptojmĂ« me kĂ« kemi komunikuar;
- praninĂ« pronĂ«sitĂ« (PFS) â kompremitimi i çelĂ«sit tonĂ« tĂ« nĂ«nshkrimit qĂ« jeton gjatĂ« nuk duhet tĂ« çojĂ« nĂ« mundĂ«sinĂ« e leximit tĂ« gjithĂ« korrespondencĂ«s sĂ« mĂ«parshme. Regjistrimi i trafikut tĂ« kapur bĂ«het i padobishĂ«m;
- valabiliteti i mesazheve (transport dhe dorëzimi) është i kufizuar vetëm në një sesion TCP. Injektimi i mesazheve të nënshkruara/identifikuara siç duhet nga një sesion tjetër (edhe me të njëjtin bisedues) nuk duhet të jetë i mundur;
- një vëzhgues pasiv nuk duhet të shikojë as identifikuesit e përdoruesve, as çelësat publikë të gjatë që dërgohen, as hash-et nga ata. Njëfarë anonimie nga një vëzhgues pasiv.
Mrekulli, por ky minimum është diçka që praktikisht të gjithë e duan në çdo protokoll dorëzimi, dhe ka shumë pak nga e lista e përmendur që në fund zbatohet për protokollet "e bërë në shtëpi". Pra tani, le të mos shpikim diçka të re. Unë do të rekomandoja qartë përdorimin e për ndërtimin e protokolleve, por le të zgjidhim diçka më të thjeshtë.
Dy protokole më të njohura janë:
- â njĂ« protokoll shumĂ« kompleks me njĂ« histori tĂ« gjatĂ« tĂ« defekteve, gabimeve, dobĂ«sive, planifikimit tĂ« dobĂ«t, kompleksitetit dhe mangĂ«sive ( megjithatĂ«, kjo nuk e pĂ«rfshin shumĂ« TLS 1.3). Por ne nuk e shqyrtojmĂ« atĂ« pĂ«r shkak tĂ« kompleksitetit tĂ« tepruar.
- me â nuk kanĂ« probleme tĂ« rĂ«nda kriptografike, megjithatĂ« sĂ«rish nuk janĂ« tĂ« thjeshta. NĂ«se i lexoni pĂ«r IKEv1 dhe IKEv2, burimi i tyre Ă«shtĂ« , ISO/IEC IS 9798-3 dhe protokollet SIGMA (SIGn-and-MAc) â mjaft tĂ« thjeshta pĂ«r t'u zbatuar brenda njĂ« mbrĂ«mjeje.
ĂfarĂ« e bĂ«n SIGMA, si lidhja e fundit nĂ« zhvillimin e protokolleve STS/ISO, tĂ« mirĂ«n? Ai pĂ«rmbush tĂ« gjitha kĂ«rkesat tona (pĂ«rfshirĂ« "fshehjen" e identifikuesve tĂ« biseduesve), nuk ka probleme tĂ« njohura kriptografike. Ai Ă«shtĂ« minimalist â heqja e madhe tĂ« paktĂ«n njĂ« elementi nga mesazhi i protokollit do tĂ« çonte nĂ« pasigurinĂ« e tij.
Le të kalojmë nga protokolli më i thjeshtë i bërë në shtëpi në SIGMA. Operacioni më bazik që na intereson është : një funksion, në daljen e të cilit të dy pjesëmarrësit do të marrin të njëjtin vlerë, e cila mund të përdoret si një çelës simetrik. Pa u angazhuar në detaje: secila palë gjeneron një çift çelësesh efemero (të përdorur vetëm brenda një sesioni) (çelësi publik dhe privat), shkëmbejnë çelësat publik, thërrasin funksionin e pajtimit, në të cilin përfshijnë çelësin e tyre privat dhe çelësin publik të biseduesit.
âââââââ âââââââ
âPeerAâ âPeerBâ
ââââŹâââ ââââŹâââ
â IdA, PubA â ââââââââââââââââââââââ
âââââââââââââââââșâ âPrvA, PubA = DHgen()â
â â ââââââââââââââââââââââ
â IdB, PubB â ââââââââââââââââââââââ
ââââââââââââââââââ âPrvB, PubB = DHgen()â
â â ââââââââââââââââââââââ
âââââ âââââââââ§âââââââââââââ
â âKey = DH(PrvA, PubB)â
ââââââ âââââââââ€âââââââââââââ
â â
â â
Kushdo mund tĂ« pĂ«rzihet dhe tĂ« zĂ«vendĂ«sojĂ« çelĂ«sat publikĂ« me tĂ« tijat â nĂ« kĂ«tĂ« protokoll nuk ka autentikimi pĂ«r biseduesit. Le tĂ« shtojmĂ« nĂ«nshkrimin me çelĂ«sa tĂ« gjatĂ«.
âââââââ âââââââ
â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) â â
â âââââââââââââââââââââââ â
â â
Një nënshkrim i tillë nuk do të funksionojë, pasi ai nuk është i lidhur me një seancë të caktuar. Të tilla mesazhe "do të funksionojnë" edhe për seanca me pjesëmarrës të tjerë. Të gjithë konteksti duhet të nënshkruhet. Kjo gjithashtu detyron shtimin e një mesazhi tjetër nga A.
Për më tepër, është kritik të shtohet nënshkrimi dhe identifikuesi personal, në të kundërt, mund të zëvendësojmë IdXXX dhe të nënshkruajmë mesazhin me çelësin e një biseduesi tjetër të njohur. Për parandalimin e , është e nevojshme që elementet nën nënshkrimin të vendosen në vende të qarta sipas kuptimit të tyre: nëse A nënshkruan (PubA, PubB), atëherë B duhet të nënshkruajë (PubB, PubA). Kjo gjithashtu tregon rëndësinë e zgjedhjes së struktures dhe formatit të të dhënave të serializuara. Për shembull, shumicat në kodimin ASN.1 DER renditen: SET OF(PubA, PubB) do të jetë identike me 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) â â â âââââââââââââââââââââââ â â
MegjithatĂ«, ne ende nuk e kemi "provuara" qĂ« kemi krijuar njĂ« çelĂ«s tĂ« pĂ«rbashkĂ«t tĂ« njĂ«jtĂ« pĂ«r kĂ«tĂ« sesion. NĂ« parim, mund tĂ« kalojmĂ« pa kĂ«tĂ« hap â mesazhi i parĂ« tĂ« transportit do tĂ« jetĂ« i pavlefshĂ«m, por ne duam qĂ« kur dorĂ«heqja tĂ« pĂ«rfundojĂ«, tĂ« jemi tĂ« sigurt se gjithçka Ă«shtĂ« dakorduar nĂ« tĂ« vĂ«rtetĂ«. NĂ« kĂ«tĂ« moment, ne kemi protokollin ISO/IEC IS 9798-3.
Ne mund të nënshkruajmë edhe çelësin e krijuar. Kjo është e rrezikshme, pasi nuk përjashtohet që në algoritmin e përdorur për nënshkrimin mund të ketë rrjedhje (edhe sikur të jetë bita-për-nënshkrim, megjithatë janë rrjedhje). Mund të nënshkruajmë hash-in e çelësit të krijuar, por rrjedhja edhe e hash-it të çelësit të krijuar mund të ketë vlerë gjatë sulmeve brute-force në funksionin e krijimit. SIGMA përdor funksionin MAC, që autentifikon identifikuesin e dërguesit.
âââââââ âââââââ â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, ...)â â â âââââââââââââââââââââââ â â
Si pĂ«r optimizim, disa mund tĂ« duan tĂ« rishfrytĂ«zojnĂ« çelĂ«sat e tyre efemĂ«r (gjĂ« qĂ«, sigurisht, Ă«shtĂ« e dĂ«mshme pĂ«r PFS). PĂ«r shembull, ne gjeneruam njĂ« çift çelĂ«sash, pĂ«rpiqĂ«m tĂ« lidhemi, por TCP nuk ishte e disponueshme ose u ndĂ«rpre diku nĂ« mes tĂ« protokollit. ĂshtĂ« e pakuptimtĂ« tĂ« humbasĂ«sh entropinĂ« e pĂ«rdorur dhe burimet e procesorit pĂ«r njĂ« çift tĂ« ri. Prandaj, do tĂ« prezantojmĂ« njĂ« cookie â njĂ« vlerĂ« pseudo tĂ« rastĂ«sishme, e cila do tĂ« mbrojĂ« nga sulmet e mundshme tĂ« ripĂ«rdorimit tĂ« rastĂ«sishĂ«m gjatĂ« ripĂ«rdorimit tĂ« çelĂ«save publikĂ« efemĂ«r. PĂ«r shkak tĂ« lidhjes midis cookie dhe çelĂ«sit publik efemĂ«r, çelĂ«si publik i palĂ«s tjetĂ«r mund tĂ« hiqet nga nĂ«nshkrimi si i panevojshĂ«m.
âââââââ âââââââ â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ë fund, ne duam të sigurojmë privatësinë e identifikuesve tanë nga një vëzhgues pasiv. Për këtë, SIGMA propozon fillimisht shkëmbimin e çelësave efemerë, për të krijuar një çelës të përbashkët mbi të cilin do të kodojmë mesazhet e autentifikimit dhe identifikimit. SIGMA përshkruan dy mundësi:
- SIGMA-I â mbron iniciatorin nga sulmet aktive, dhe pĂ«rgjigjĂ«sin nga ato pasive: iniciatori autentifikon pĂ«rgjigjĂ«sin dhe nĂ«se diçka nuk Ă«shtĂ« nĂ« rregull, ai nuk e zbulohet identitetin e tij. PĂ«rgjigjĂ«si ndan identitetin e tij nĂ«se fillon njĂ« protokoll aktiv. NjĂ« vĂ«zhgues pasiv nuk do tĂ« dijĂ« asgjĂ«;
SIGMA-R â mbron pĂ«rgjigjĂ«sin nga sulmet aktive, dhe iniciatorin nga ato pasive. E gjithĂ« kjo Ă«shtĂ« e kundĂ«rt, por nĂ« kĂ«tĂ« protokoll transmetohen tashmĂ« katĂ«r mesazhe dorĂ«shkrimi.Ne zgjedhim SIGMA-I si mĂ« tĂ« afĂ«rt me ato qĂ« presim nga gjĂ«rat tradicionale klient-server: klienti njeh vetĂ«m serverin e autentikuar, ndĂ«rsa serveri e njeh gjithçka. Plus, Ă«shtĂ« mĂ« e lehtĂ« pĂ«r t'u realizuar pĂ«r shkak tĂ« numrit tĂ« vogĂ«l tĂ« mesazheve tĂ« dorĂ«zimit. E vetmja gjĂ« qĂ« po e shtojmĂ« nĂ« protokoll Ă«shtĂ« enkriptimi i njĂ« pjesĂ« tĂ« mesazhit dhe transferimi i identifikuesit A nĂ« pjesĂ«n e enkriptuar tĂ« mesazhit tĂ« fundit:
âââââââ âââââââ â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, ...)â â â âââââââââââââââââââââââ â â
- Për nënshkrimin përdoret GOST R algoritmi me çelësa 256-bit.
- Për krijimin e çelësit të përbashkët përdoret 34.10-2012 VKO.
- Si MAC pĂ«rdoret CMAC. Teknikisht, ky Ă«shtĂ« njĂ« mod i veçantĂ« i punĂ«s sĂ« algoritmit tĂ« blokut, i pĂ«rshkruar nĂ« GOST R 34.13-2015. Si funksion enkriptimi pĂ«r kĂ«tĂ« mod â (34.12-2015).
- Si identifikues i biseduesit përdoret hash-i nga çelësi i tij publik. Përdoret si hash (34.11-2012 256 bit).
Pas dorëzimit do të kemi një çelës të përbashkët të marrëveshjes. Atë mund ta përdorim për enkriptimin e autentifikimit të mesazheve të transportit. Kjo pjesë është krejt e thjeshtë dhe është e vështirë të bësh gabim: inkrementojmë numëruesin e mesazheve, enkriptojmë mesazhin, autentifikojmë (MAC) numëruesin dhe tekstin e enkriptuar, dërgojmë. Kur pranojmë një mesazh, kontrollojmë që numëruesi të ketë vlerën e pritur, autentifikojmë tekstin e enkriptuar me numëruesin, dekriptojmë. Me çfarë çelësi të enkriptojmë mesazhet e dorëzimit, ato të transportit, si të autentifikojmë? Të përdorim një çelës për të gjitha këto detyra është e rrezikshme dhe e paarsyeshme. Nevojitet të krijojmë çelësa, duke përdorur funksione të specializuara (funksioni i nxjerrjes së çelësit). Edhe një herë, nuk do të jemi ndërlikuar dhe të shpikim diçka: është e njohur prej kohësh, e studiuar mirë dhe nuk ka probleme të njohura. Fatkeqësisht, në bibliotekën native të Python nuk ekziston ky funksion, prandaj përdorim paketën. HKDF brenda përdor përdor , i cili, nga ana e tij, përdor funksionin hash. Një shembull i zbatimit në Python në faqen Wikipedia zë disa rreshta kodi. Ashtu si në rastin e 34.10-2012, si funksion hash do të përdorim Stribog-256. Dalja e funksionit tonë të pajtimit të çelësave do të quhet çelësi sesionit, nga i cili do të prodhohen çelësat simetrikë të nevojshëm:
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")Struktura/ Schema
Le të shqyrtojmë cilat struktura ASN.1 kemi tani për dërgimin e të gjitha këtyre të dhënave:
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 â ajo do tĂ« nĂ«nshkruhet (do tĂ« nĂ«nshkruhet). HandshakeTBE â ajo qĂ« do tĂ« kriptohet (do tĂ« kriptohet). VĂ«rej se nĂ« fushĂ«n ukm nĂ« MsgHandshake1. 34.10 VKO, pĂ«r njĂ« shĂ«nim mĂ« tĂ« madh tĂ« rastĂ«sishĂ«m te çelĂ«sat e prodhuar, pĂ«rfshin parametrin UKM (materiali i çelĂ«sit tĂ« pĂ«rdoruesit) â thjesht entropi shtesĂ«.
Shtimi i kriptografisë në kod
Të shqyrtojmë vetëm ndryshimet e bëra në kodin origjinal, pasi struktura ka mbetur e njëjtë (në të vërtetë, fillimisht u shkrua implementimi përfundimtar, dhe pastaj u hoqën të gjitha kriptografitë).
Pasi identifikimi dhe autentikimi i biseduesve do të kryhet përmes çelësave publikë, tani duhet t'i ruajmë ato në një mënyrë afatgjatë. Për thjeshtësi, përdorim një JSON të tillë:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }our â çifti ynĂ« i çelĂ«save, çelĂ«sat privat dhe publik nĂ« gjashtĂ«mbĂ«dhjetĂ«sh. their â emrat e biseduesve dhe çelĂ«sat e tyre publik. Do tĂ« ndryshojmĂ« argumentet e komandĂ«s dhe do tĂ« shtojmĂ« pastrimin e tĂ« dhĂ«nave JSON:
nga pygost import gost3410 nga 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="Generoni JSON me çiftin tonĂ« tĂ« çelĂ«sit tĂ« ri", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON me çelĂ«sat tanĂ« dhe ata", ) parser.add_argument( "--bind", default="::1", help="Adresa pĂ«r tĂ« pĂ«r tĂ« dĂ«gjuar", ) parser.add_argument( "--port", type=int, default=6666, help="Porti pĂ«r tĂ« dĂ«gjuar", ) 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) # Analizo dhe deshifro çelĂ«sat tanĂ« dhe ata {{{ mbi open(args.keys, "rb") si 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()) per peer_name, pub_raw nĂ« _keys["their"].items(): _pub = hexdec(pub_raw) KEYS[GOST34112012256(_pub).digest()] = { "name": peer_name, "pub": gost3410.pub_unmarshal(_pub), } # }}}ĂelĂ«si privat i algoritmit 34.10 Ă«shtĂ« njĂ« numĂ«r i rastĂ«sishĂ«m. Me madhĂ«si 256-bit pĂ«r kthesat eliptike 256-bit. PyGOST nuk punon me njĂ« grup bajtesh, por me , prandaj çelĂ«si ynĂ« privat (urandom(32)) duhet tĂ« konvertohet nĂ« numĂ«r duke pĂ«rdorur gost3410.prv_unmarshal(). ĂelĂ«si publik llogaritet nĂ« mĂ«nyrĂ« deterministe nga ai privat, duke pĂ«rdorur gost3410.public_key(). ĂelĂ«si publik 34.10 pĂ«rbĂ«het nga dy numra tĂ« mĂ«dhenj, tĂ« cilĂ«t gjithashtu duhet tĂ« konvertohen nĂ« njĂ« sekuencĂ« bajtesh pĂ«r ruajtje dhe transferim tĂ« lehtĂ«, duke pĂ«rdorur gost3410.pub_marshal().
Pasi të lexohet skedari JSON, çelësat publikë, përkatësisht, duhet të kthehen prapa, duke përdorur gost3410.pub_unmarshal(). Duke qenë se do të marrim identifikuesit e biseduesve në formën e hash-it nga çelësi publik, ato mund të llogariten paraprakisht dhe të vendosen në një fjalor për kërkimin e shpejtë. Hash-i Stribog-256 është gost34112012256.GOST34112012256(), plotësisht duke përmbushur ndërfaqen e funksioneve hash të hashlib.
Si Ă«shtĂ« ndryshuar korutina e iniciatorit? Ădo gjĂ«, si sipas skemĂ«s sĂ« dorĂ«zimit: gjenerojmĂ« cookie (128-bit Ă«shtĂ« mjaft) dhe njĂ« çift çelĂ«sash efemer 34.10, i cili do tĂ« pĂ«rdoret pĂ«r funksionin VKO tĂ« rĂ«nĂ«shĂ«s sĂ« çelĂ«sit.
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 # Gjeneroni çelësin tonë publik efemer dhe cookie, dërgoni mesazhin 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()- po presim përgjigje dhe dekodojmë mesazhin Msg që ka ardhur;
- sigurohemi që kemi marrë handshake1;
- dekodojmë çelësin publik efemer të palës tjetër dhe llogarisim çelësin e sesionit;
- gjenerojmë çelësat simetrikë të nevojshëm për përpunimin e pjesës TBE të mesazhit.
423 logging.info("%s: got %s message", _id, msg.choice) 424 if msg.choice != "handshake1": 425 logging.warning("%s: mesazh i papritur, po shkĂ«putemi", _id) 426 writer.close() 427 return 428 # }}} 429 msg_handshake1 = msg.value 430 # Validoni mesazhin e 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 Ă«shtĂ« njĂ« numĂ«r 64-bit (urandom(8)), i cili gjithashtu kĂ«rkon deserializim nga paraqitja e byte-ve, duke pĂ«rdorur gost3410_vko.ukm_unmarshal(). Funksioni VKO pĂ«r 34.10-2012 256-bit Ă«shtĂ« gost3410_vko.kek_34102012256() (KEK â çelsi i enkriptimit tĂ« çelĂ«sit).
ĂelĂ«si i sesionit tĂ« gjeneruar tashmĂ« Ă«shtĂ« njĂ« sekuencĂ« 256-bit e rastĂ«sishme byte. Prandaj ai mund tĂ« pĂ«rdoret menjĂ«herĂ« nĂ« funksionin HKDF. Duke qenĂ« se GOST34112012256 pĂ«rmbush ndĂ«rfaqen hashlib, mund tĂ« pĂ«rdoret direkt nĂ« klasĂ«n Hkdf. Kripa (argumenti i parĂ« Hkdf) nuk e specifikojmĂ«, pasi çelĂ«si i gjeneruar pĂ«r shkak tĂ« efemĂ«risĂ« sĂ« çelĂ«save do tĂ« jetĂ« i ndryshĂ«m pĂ«r çdo sesion dhe ai ka mjaft entropi. kdf.expand() pĂ«r herĂ« tĂ« parĂ« jep çelĂ«sat me gjatĂ«si 256-bit, tĂ« nevojshĂ«m pĂ«r Kuzneçikun nĂ« vijim.
Më pas kontrollohen pjesët TBE dhe TBS të mesazhit që ka ardhur:
- llogaritet dhe verifikohet MAC mbi tekstin e enkriptuar që ka ardhur;
- teksti i enkriptuar dekodifikohet;
- struktura TBE dekodifikohet;
- nga ajo merret identifikuesi i komunikuesit dhe kontrollohet nëse është njohur nga ne;
- llogaritet dhe kontrollohet MAC mbi këtë identifikues;
- nënshkrimi verifikohet mbi strukturën TBS, e cila përfshin cookie-t e të dyja palëve dhe çelësin publik efemer të palës tjetër. Nënshkrimi verifikohet me çelësin e gjatë të nënshkrimit të bashkëbiseduesit.
441 provo: 442 emri_i_palĂ«s = verifiko_tbe( 443 msg_handshake1, 444 key_handshake1_mac_identity, 445 key_handshake1_enc, 446 key_handshake1_mac, 447 cookie_tona, 448 cookie_tĂ«_tyres, 449 pub_tyre_raw, 450 ) 451 pĂ«rjashto vleraError si err: 452 logging.warning("%s: %s, duke shkĂ«putur", _id, err) 453 writer.close() 454 kthe 455 # }}} 128 def verifiko_tbe( 129 msg_handshake: Union[MsgHandshake1, MsgHandshake2], 130 key_mac_identiteti: bytes, 131 key_enc: bytes, 132 key_mac: bytes, 133 cookie_tyre: Cookie, 134 cookie_tona: Cookie, 135 pub_key_tona: PubKey, 136 ) -> str: 137 ciphertext = bytes(msg_handshake["ciphertext"]) 138 mac_tag = mac(GOST3412Kuznechik(key_mac).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext) 139 nĂ«se jo compare_digest(mac_tag, bytes(msg_handshake["ciphertextMac"])): 140 ngrihet vleraError("MAC i pavlefshĂ«m") 141 plaintext = ctr( 142 GOST3412Kuznechik(key_enc).encrypt, 143 KUZNECHIK_BLOCKSIZE, 144 ciphertext, 145 8 * b"x00", 146 ) 147 provo: 148 tbe, _ = HandshakeTBE().decode(plaintext) 149 pĂ«rjashto ASN1Error: 150 ngrihet vleraError("nuk mund tĂ« dekodoj TBE") 151 key_sign_pub_hash = bytes(tbe["identity"]) 152 palĂ« = KEY.get(key_sign_pub_hash) 153 nĂ«se palĂ« Ă«shtĂ« None: 154 ngrihet vleraError("identitet i panjohur") 155 mac_tag = mac( 156 GOST3412Kuznechik(key_mac_identity).encrypt, 157 KUZNECHIK_BLOCKSIZE, 158 key_sign_pub_hash, 159 ) 160 nĂ«se jo compare_digest(mac_tag, bytes(tbe["identityMac"])): 161 ngrihet vleraError("MAC i identitetit i pavlefshĂ«m") 162 tbs = HandshakeTBS(( 163 ("cookieTyre", cookie_tyre), 164 ("cookieTona", cookie_tona), 165 ("pubKeyTona", pub_key_tona), 166 )) 167 nĂ«se jo gost3410.verify( 168 CURVE, 169 palĂ«["pub"], 170 GOST34112012256(tbs.encode()).digest(), 171 bytes(tbe["signature"]), 172 ): 173 ngrihet vleraError("nĂ«nshkrim i pavlefshĂ«m") 174 kthe palĂ«["name"]Siç e thashĂ« mĂ« parĂ«, 34.13-2015 pĂ«rshkruan modet e ndryshme nĂ« 34.12-2015. NdĂ«r ta ka njĂ« mod tĂ« prodhimit tĂ« imitoshtesĂ«, llogaritjes sĂ« MAC-it. NĂ« PyGOST, kjo Ă«shtĂ« gost3413.mac(). Ky mod kĂ«rkon transmetimin e njĂ« funksioni tĂ« enkriptimit (qĂ« merr dhe kthen njĂ« bllok tĂ« dhĂ«nash), madhĂ«sinĂ« e bllokut tĂ« enkriptimit dhe, me siguri, vetĂ« tĂ« dhĂ«nat. Pse nuk mund tĂ« hardcode-ojmĂ« madhĂ«sinĂ« e bllokut tĂ« enkriptimit? 34.12-2015 pĂ«rshkruan jo vetĂ«m enkriptimin 128-bit Kunechnik, por gjithashtu edhe 64-bitin â njĂ« modifikim i vogĂ«l i GOST 28147-89, i krijuar mĂ« parĂ« nga KGB dhe ende qĂ« mban njĂ« nga standardet mĂ« tĂ« larta tĂ« sigurisĂ«.
Kuznechik inicializohet gost.3412.GOST3412Kuznechik(key) duke e thirrur dhe kthen një objekt me metodat .encrypt()/.decrypt() të përshtatshme për dërgim në funksionet 34.13. MAC llogaritet në këtë mënyrë: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Për të krahasuar MAC-in e llogaritur dhe atë të mbërritur nuk mund të përdorim një krahasim të zakonshëm (==) të vargjeve të bajtëve, pasi kjo operacion sjell rrjedhje të kohës së krahasimit, e cila, në përgjithësi, mund të çojë në dobësi fatale të tipit sulmet ndaj TLS. Në Python ka një funksion të veçantë hmac.compare_digest për këtë.
Funksioni i blokut tĂ« enkriptimit mund tĂ« enkriptojĂ« vetĂ«m njĂ« bllok tĂ« dhĂ«nash. PĂ«r mĂ« shumĂ«, dhe me gjatĂ«si qĂ« nuk Ă«shtĂ« shumĂ«fish, Ă«shtĂ« e nevojshme tĂ« pĂ«rdoret njĂ« mod i enkriptimit. NĂ« 34.13-2015 pĂ«rshkruhen si mĂ«poshtĂ«: ECB, CTR, OFB, CBC, CFB. Ădo njĂ«ri ka fushat e tij tĂ« lejuara dhe karakteristikat e tij. PĂ«r fat tĂ« keq, ne ende nuk kemi moda tĂ« standardizuara (tipa CCM, OCB, GCM dhe tĂ« ngjashme) â ne jemi tĂ« detyruar tĂ« shtojmĂ« vetĂ« MAC-in. UnĂ« zgjedh (CTR): ai nuk kĂ«rkon plotĂ«sim deri nĂ« madhĂ«sinĂ« e bllokut, mund tĂ« pĂ«rparojĂ« nĂ« mĂ«nyrĂ« paralele, pĂ«rdor vetĂ«m funksionin e enkriptimit, mund tĂ« pĂ«rdoret nĂ« mĂ«nyrĂ« tĂ« sigurt pĂ«r enkriptimin e njĂ« sasi tĂ« madhe mesazhesh (ndryshe nga CBC, ku relativisht shpejt fillojnĂ« kolizionet).
Siç është .mac(), .ctr() merr të dhëna të ngjashme si hyrje: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). Kërkohet caktimi i një vektori inicializimi, që ka një gjatësi saktësisht sa gjysma e bllokut të enkriptimit. Nëse çelësi ynë i enkriptimit përdoret vetëm për enkriptimin e një mesazhi (edhe pse nga disa blloqe), atëherë është e sigurt të caktohet një vektor inicializimi të zeruar. Për enkriptimin e mesazheve të dorëzimit, ne përdorim gjithmonë një çelës të ndarë çdo herë.
Kontrolli i nënshkrimit gost3410.verify() është triviale: ne dërgojmë kurbën eliptike brenda së cilës punojmë (atë e fiksojmë thjesht në protokollin tonë GOSTIM), çelësi publik i nënshkruese (mos harroni, se kjo duhet të jetë një çift nga dy numra të mëdhenj, dhe jo një varg bajtësh), hapi 34.11-2012 dhe nënshkrimi i mbërritur.
Pastaj, ne në iniciues përgatitim dhe dërgojmë mesazhin handshake2 të dorëzimit, duke kryer të njëjtat veprime që kemi bërë gjatë kontrollit, vetëm në mënyrë simetrike: nënshkrimi në çelësat tanë përveç kontrollit, etj...
456 # Përgatit dhe dërgo Mesazhin 2 Handshake {{{ 457 tbs = HandshakeTBS(( 458 ("cookieTheir", cookie_their), 459 ("cookieOur", cookie_our), 460 ("pubKeyOur", pub_our_raw), 461 )) 462 signature = gost3410.sign( 463 CURVE, 464 KEY_OUR_SIGN_PRV, 465 GOST34112012256(tbs.encode()).digest(), 466 ) 467 key_handshake2_mac_identity = kdf.expand(b"handshake2-mac-identity") 468 mac_tag = mac( 469 GOST3412Kuznechik(key_handshake2_mac_identity).encrypt, 470 KUZNECHIK_BLOCKSIZE, 471 bytes(KEY_OUR_SIGN_PUB_HASH), 472 ) 473 tbe = HandshakeTBE(( 474 ("identity", KEY_OUR_SIGN_PUB_HASH), 475 ("signature", OctetString(signature)), 476 ("identityMac", MAC(mac_tag)), 477 )) 478 tbe_raw = tbe.encode() 479 key_handshake2_enc = kdf.expand(b"handshake2-enc") 480 key_handshake2_mac = kdf.expand(b"handshake2-mac") 481 ciphertext = ctr( 482 GOST3412Kuznechik(key_handshake2_enc).encrypt, 483 KUZNECHIK_BLOCKSIZE, 484 tbe_raw, 485 8 * b"x00", 486 ) 487 mac_tag = mac( 488 GOST3412Kuznechik(key_handshake2_mac).encrypt, 489 KUZNECHIK_BLOCKSIZE, 490 ciphertext, 491 ) 492 writer.write(Msg(("handshake2", MsgHandshake2(( 493 ("ciphertext", OctetString(ciphertext)), 494 ("ciphertextMac", MAC(mac_tag)), 495 )))).encode()) 496 # }}} 497 await writer.drain() 498 logging.info("%s: sesi e ngjitur: %s", _id, peer_name)Kur sesioni është krijuar, janë gjeneruar çelësat e transportit (një çelës të veçantë për enkriptimin, për autentifikimin, për çdo palë), dhe është iniciuar Kuznechik për dekodimin dhe verifikimin e MAC-it:
499 # Ekzekuto dĂ«rguesin e mesazheve tekstuale, inicializo dekoderin e transportit {{{ 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 # Prit pĂ«r mesazhet e testit {{{ 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 # }}}msg_sender korutina tani enkripton mesazhet, para se tĂ« dĂ«rgohet nĂ« lidhjen TCP. Ădo mesazh ka njĂ« nonce qĂ« rritet monotone, i cili gjithashtu shĂ«rben si njĂ« vektor inicializimi gjatĂ« enkriptimit nĂ« modin e numĂ«ratorit. Ădo mesazh dhe bllok mesazhi garanton qĂ« tĂ« ketĂ« vlera tĂ« ndryshme pĂ«r numĂ«ratorin.
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 += 1Mesazhet e ardhshme përpunohen nga korutina msg_receiver, e cila merret me autentifikimin dhe dekriptimin:
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("vlera e papritur e nonce") mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode()) if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])): raise ValueError("MAC e pavlefshme") plaintext = ctr( encrypter, KUZNECHIK_BLOCKSIZE, bytes(payload["ciphertext"]), long2bytes(nonce_expected, 8), ) text = plaintext.decode("utf-8") await OUT_QUEUES[peer_name].put(text)PĂ«rfundim
GOSTIM pritet të përdoret ekskluzivisht për qëllime mësimore (pasi nuk është mbuluar me teste, të paktën)! Kodi burimor i programit mund të shkarkohet (Hashi Stribog-256: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Ashtu si të gjithë projektet e mia, tip , , , , GOSTIM është plotësisht , i shpërndarë sipas kushteve .
, , anëtar i , zhvilluesi Python/Go, specialist kryesor .
Burimi: habr.com
