Fejlesztőként a (GOST alapú kriptográfiai primitívek tiszta Python-ban) könyvtárral gyakran kapok kérdéseket arról, hogyan lehet gyorsan megvalósítani a legegyszerűbb biztonságos üzenetváltást. Sokan úgy vélik, hogy az alkalmazott kriptográfia meglehetősen egyszerű dolog, és egy .encrypt() hívás egy blokkalgoritmusnál elegendő a biztonságos adatátviteli csatornához. Mások viszont azt gondolják, hogy az alkalmazott kriptográfia csupán néhány ember kiváltsága, és elfogadható, hogy a gazdag vállalatok, mint például a Telegram, matematikai olimpiai versenyzőkkel a biztonságos protokollt.
Mindez arra ösztönzött, hogy megírjam ezt a cikket, hogy megmutassam, a kriptográfiai protokollok és a biztonságos IM megvalósítása nem olyan nehéz feladat. Azonban nem érdemes saját hitelesítési és kulcsmegosztási protokollokat feltalálni.
A cikkben , , rendelkező azonnali üzenetküldőt fogunk létrehozni hitelesítési és kulcsmegosztási protokollal (amely alapján ), kizárólag a PyGOST könyvtár GOST kriptográfiai algoritmusait és az ASN.1 üzenetkódolást a (amelyről korábban már ). A szükséges feltétel: annyira egyszerűnek kell lennie, hogy egy este alatt (vagy munkaidő alatt) megírható legyen, különben ez már nem egy egyszerű program. Biztosan lesznek hibák benne, felesleges bonyodalmak, hiányosságok, ráadásul ez az első programom az asyncio könyvtár használatával.
Az IM dizájnja
Elsőként meg kell értenünk, hogyan fog kinézni az IM. Az egyszerűség kedvéért legyen egy peer-to-peer hálózat, mindenféle résztvevő-azonosítás nélkül. Kézzel fogjuk megadni, hogy melyik címhez: portra kell csatlakozni a beszélgetőpartnerrel való kommunikációhoz.
Megértem, hogy a jelenlegi helyzetben a két tetszőleges számítógép közötti közvetlen kapcsolat elérhetősége – jelentős korlátozás az IM gyakorlati alkalmazásában. De minél több fejlesztő valósít meg különféle NAT-traversal megoldásokat, annál tovább fogunk ragaszkodni az IPv4 internethez, reménytelen valószínűséggel a tetszőleges számítógépek közötti kapcsolatra. Hányan bírják még elviselni az IPv6 hiányát otthon és a munkahelyen?
Képzelj el egy barátok közötti hálózatot: minden lehetséges beszélgetőtárs előre ismert. Először is, ez nagyban leegyszerűsíti a dolgokat: bemutatkoztunk, megtaláltuk vagy nem találtuk meg a nevet/kulcsot, elváltunk vagy folytatjuk a munkát, ismerve a beszélgetőtársunkat. Másodszor, általánosságban véve ez biztonságos és kizár egy csomó támadást.
Az IM felület hasonló lesz a klasszikus megoldásokhoz. , amelyek nagyon tetszenek a minimalizmusuk és a Unix-way filozófiájuk miatt. Az IM program minden beszélgetőtárs számára létrehoz egy könyvtárat három Unix domain socket-tel:
- in — ebbe kerülnek a beszélgetőtársnak küldött üzenetek;
- out — ebből olvashatók a beszélgetőtárstól érkező üzenetek;
- state — ebből megtudjuk, hogy a beszélgetőtársunk most csatlakozik-e, és mi az address/port.
Ezen kívül létrejön egy conn socket, amelybe beírva a host portot, elindítjuk a kapcsolatot a távoli beszélgetőtárssal.
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Ez a megközelítés lehetővé teszi független IM szállítók és felhasználói felületek megvalósítását, hiszen ízlés és szín kérdéséhez nem lehet mindenki kedvére tenni. Használva Services , egy többablakos felületet kaphatunk szintaxis kiemeléssel. A segítségével egy GNU Readline-kompatibilis üzenetbeviteli sort kaphatunk.
Valójában a suckless projektek FIFO fájlokat használnak. Személy szerint nem tudtam kitalálni, hogyan lehet az asyncio-t versenyképesen használni fájlokkal, saját kezűleg létrehozott háttérszálak nélkül (ilyen dolgokra régóta a ) nyelvet használom. Ezért döntöttem úgy, hogy Unix domain socketekre támaszkodom. Sajnos ez eltávolította a lehetőséget, hogy echo 2001:470:dead::babe 6666 > conn. Ezt a problémát úgy oldottam meg, hogy : echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
Az eredeti nem biztonságos protokoll
A transportként TCP-t használunk: garantálja a szállítást és annak sorrendjét. Az UDP nem garantálja sem az egyiket, sem a másikat (ami hasznos lenne, amikor kriptográfiát alkalmazunk), és a nincsen beépítve a Pythonba.
Sajnos a TCP-nél nincs üzenetfogalom, csak bájtfolyam. Ezért valamilyen formátumot kell kitalálnunk az üzenetek számára, hogy azokat el lehessen választani egymástól ebben a folyamban. Megállapodhatunk abban, hogy a sortörés szimbólumát használjuk. Eleinte megfelel, azonban amikor elkezdjük titkosítani az üzeneteinket, ez a szimbólum bárhol megjelenhet a titkos szövegben. A hálózatokban ezért népszerűek azok a protokollok, amelyek először a bájtokban mért üzenet hosszát küldik el. Például a Pythonban van egy beépített xdrlib, ami lehetővé teszi az ilyen formátummal való munkát. .
Nem fogunk helyesen és hatékonyan dolgozni a TCP olvasásával - leegyszerűsítjük a kódot. Végtelen ciklusban olvassuk a data-t a socket-ből, amíg nem dekódoljuk a teljes üzenetet. Egy ilyen megközelítéshez használhatunk JSON-t vagy XML-t is mint formátumot. De amikor a kriptográfia hozzáadódik, az adatokat alá kell írni és hitelesíteni - ami megköveteli az objektumok bitekben azonos ábrázolását, amit a JSON/XML nem biztosít (a dump-olt eredmények eltérhetnek).
Az XDR megfelelő ilyen feladatokra, azonban én az ASN.1-t választom DER-kódolással és könyvtárral, mivel magasan szintű objektumaink lesznek, amelyekkel gyakran kellemesebb és kényelmesebb dolgozni. A schemaless verziókkal ellentétben az , vagy , az ASN.1 automatikusan ellenőrzi az adatokat a szigorúan definiált séma ellen.
# Msg ::= CHOICE {
# text MsgText,
# handshake [0] EXPLICIT MsgHandshake }
class Msg(Choice):
schema = ((
("text", MsgText()),
("handshake", MsgHandshake(expl=tag_ctxc(0))),
))
# MsgText ::= SEQUENCE {
# text UTF8String (SIZE(1..MaxTextLen))}
class MsgText(Sequence):
schema = ((
("text", UTF8String(bounds=(1, MaxTextLen))),
))
# MsgHandshake ::= SEQUENCE {
# peerName UTF8String (SIZE(1..256)) }
class MsgHandshake(Sequence):
schema = ((
("peerName", UTF8String(bounds=(1, 256))),
))
A fogadott üzenet a Msg lesz: vagy egy szöveges MsgText (jelenleg egy szövegmezővel), vagy egy kézfogásüzenet MsgHandshake (amelyben a beszélgetőpartner neve érkezik). Jelenleg túlbonyolítottnak tűnik, de ez a jövőbeli terveink elősegítése.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│MsgHandshake(IdA) │
│─────────────────>│
│ │
│MsgHandshake(IdB) │
││
│ │
│ MsgText() │
│<─────────────────│
│ │
IM kriptográfia nélkül
Ahogy már mondtam, az összes socket-operációhoz az asyncio könyvtárat fogjuk használni. Meg fogjuk határozni, mit várunk a program indításakor:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Saját nevünk",
)
parser.add_argument(
"--their-names",
required=True,
help="Az ő neveik, vesszővel elválasztva",
)
parser.add_argument(
"--bind",
default="::1",
help="Cím, amelyen figyelni fogunk",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Port, amelyen figyelni fogunk",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
A saját név megadása (—our-name alice). Vesszővel elválasztva felsoroljuk a várt partnereket (—their-names bob,eve). Minden partner számára létrejön egy könyvtár Unix socketekkel, valamint egy aszinkron függvény minden in, out, state számára:
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"))
A felhasználótól érkező üzenetek az in socketből az IN_QUEUES sorokba kerülnek:
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"))
A partnerektől érkező üzenetek az OUT_QUEUES sorokba kerülnek, ahonnan az adatok az out socketbe íródnak:
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()
A state socketből való olvasás során a program megkeresi a PEER_ALIVE szótárban a partner címét. Ha még nincs kapcsolat a partnerrel, üres sor kerül beírásra.
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()
A cím conn socketbe történő írásakor elindul a kapcsolat "kezdeményező" függvénye:
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))
Nézzük meg a kezdeményezőt. Először is nyilvánvalóan megnyitja a kapcsolatot a megadott host/port címmel, és elküldi a handshake üzenetet a saját nevével:
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 # Handshake üzenet {{{
135 writer.write(Msg(("handshake", MsgHandshake((
136 ("peerName", OUR_NAME),
137 )))).encode())
138 # }}}
139 await writer.drain()
Ezután megvárja a távoli fél válaszát. Megpróbálja dekódolni a beérkezett választ az Msg ASN.1 megoldás szerint. Feltételezzük, hogy az üzenetet egyetlen TCP-szegmensben küldik el, és azt atomikusan kapjuk meg a .read() hívásakor. Ellenőrizzük, hogy valóban handshake üzenetet kaptunk.
141 # Várj a Handshake üzenetre {{{
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 # }}}
Ellenőrizzük, hogy a beérkezett beszélgetőpartner neve ismert-e számunkra. Ha nem, akkor bontjuk a kapcsolatot. Ellenőrizzük, hogy már volt-e kapcsolatunk vele (a beszélgetőpartner újra csatlakozási parancsot adott), és zárjuk le ezt. Az IN_QUEUES sorba Python karakterláncok kerülnek a beérkezett üzenet szövegével, de van egy különleges érték, a None, amely jelzi a msg_sender korutinának, hogy fejezze be a működést, hogy elfelejtse a kapcsolódó writer-t a régi TCP kapcsolathoz.
159 msg_handshake = msg.value
160 peer_name = str(msg_handshake["peerName"])
161 if peer_name not in THEIR_NAMES:
162 logging.warning("ismeretlen hozzáférési név: %s", peer_name)
163 writer.close()
164 return
165 logging.info("%s: session established: %s", _id, peer_name)
166 # Futtassuk a szöveges üzenet küldőt, inicializáljuk a szállító dekódolót {{{
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 # }}}
A msg_sender kezeli a kimenő üzeneteket (amelyeket az in socketből teszünk a sorba), a MsgText üzenetbe sorolja őket és elküldi a TCP kapcsolaton keresztül. Ez megszakadhat bármikor — ezt kifejezetten elkapjuk.
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: sent %d characters message", peer_name, len(text))
A végén az indító egy végtelen ciklusba lép, ahol üzeneteket olvas a socketből. Ellenőrzi, hogy ezek szöveges üzenetek-e, és elhelyezi az OUT_QUEUES sorba, ahonnan azokat a megfelelő partner out socketjébe fogják küldeni. Miért ne lehetne egyszerűen .read() és dekódolni az üzenetet? Mert nem zárható ki az a helyzet, amikor több üzenet a felhasználótól összegyűlik az operációs rendszer puffereiben, és egyetlen TCP-szegmensként kerül elküldésre. Az elsőt dekódolni tudjuk, de a bufferben ott maradhat a következőből egy rész. Bármilyen rendellenes helyzetben bezárjuk a TCP-kapcsolatot, és leállítjuk a msg_sender korutint (None küldésével az OUT_QUEUES sorba).
174 buf = b""
175 # Várakozás a tesztüzenetekre {{{
176 míg True:
177 data = await reader.read(MaxMsgLen)
178 ha data == b"":
179 break
180 buf += data
181 ha len(buf) > MaxMsgLen:
182 logging.warning("%s: a maximális pufferméret túllépve", _id)
183 break
184 próbáld meg:
185 msg, tail = Msg().decode(buf)
186 kivéve ASN1Error:
187 folytatás
188 buf = tail
189 ha msg.choice != "text":
190 logging.warning("%s: váratlan %s üzenet", _id, msg.choice)
191 break
192 próbáld meg:
193 await msg_receiver(msg.value, peer_name)
194 kivéve ValueError mint err:
195 logging.warning("%s: %s", err)
196 break
197 # }}}
198 logging.info("%s: bontás: %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: %d karakteres üzenet érkezett", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Térjünk vissza a fő kódhoz. Miután az összes korutint létrehoztuk a program indításakor, elindítjuk a TCP-szervert. Minden egyes létrehozott kapcsolat után létrehozza a responder (válaszoló) korutint.
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("Hallgatás: %s", server.sockets[0].getsockname())
loop.run_forever()
A responder hasonlít az initiatorra, és visszafelé végzi el ugyanezeket a műveleteket, de a végtelen üzenetolvasási ciklus azonnal elindul, a egyszerűség kedvéért. Jelenleg a kézfogó protokoll minden oldalról egy üzenetet küld, de a jövőben kettőt fog küldeni a kapcsolat indítójától, amelyek után azonnal lehetővé válik a szöveges üzenetek küldése.
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()
Biztonságos protokoll
Eljött az idő, hogy biztonságossá tegyük a kommunikációnkat. Mit értünk biztonság alatt, és mit akarunk:
- az átadott üzenetek titkosságát;
- az átadott üzenetek hitelessége és sértetlensége — bármilyen módosítást észlelni kell;
- védelem a replay támadásokkal szemben — az üzenetek eltűnését vagy megismétlését észlelnünk kell (és úgy döntünk, hogy megszakítjuk a kapcsolatot);
- a beszélgetőpartnerek az előre megadott nyilvános kulcsok alapján történő azonosítása és hitelesítése — korábban már eldöntöttük, hogy barátok közti hálózatot építünk. Csak a hitelesítés után tudjuk, kivel kommunkukálunk;
- létezik tulajdonságok (PFS) — a hosszú életű aláíró kulcsunk kompromittálása nem vezethet az összes korábbi üzenet olvashatóságához. A lehallgatott forgalom rögzítése haszontalanná válik;
- az üzenetek (szállítási és kézfogási) érvényessége csak egy TCP-szekció határain belül van. Másik szekcióból (még ha ugyanazzal a partnerrel is) megfelelően aláírt/azonosított üzenetek beillesztése nem lehetőség;
- a passzív megfigyelőnek nem szabad látnia sem a felhasználói azonosítókat, sem a továbbított hosszú élettartamú nyilvános kulcsokat, sem a róluk készült hash-eket. Valami anonimitás a passzív megfigyelő elől.
Meglepetés, de ezt a minimumot szinte mindenki szeretné látni bármely kézfogási protokollban, és rendkívül kevés a felsoroltak közül végül megvalósul. Most sem fogunk újat kitalálni. Határozottan ajánlanám a protokollok építésére, de válasszunk valami egyszerűbbet.
Két protokoll a legnépszerűbb:
- — egy bonyolult protokoll, hosszú hibák, hiányosságok, sebezhetőségek, rossz mérlegelés, bonyolultság és elmaradottság történetével (bár a TLS 1.3-ra ez nem igazán vonatkozik). De nem vesszük figyelembe a túlkomplettyezése miatt.
- a — nem rendelkezik komoly kriptográfiai problémákkal, bár nem is egyszerűek. Ha elolvassuk az IKEv1 és IKEv2 részleteit, eredetük , ISO/IEC IS 9798-3 és a SIGMA (SIGn-and-MAc) protokoll — elég egyszerűek ahhoz, hogy egy este alatt megvalósíthatók legyenek.
Mit tud nyújtani a SIGMA, mint az STS/ISO protokollok fejlődésének utolsó láncszeme? Megfelel minden követelményünknek (beleértve a partnerek azonosítójának „elrejtését” is), és nem rendelkezik ismert kriptográfiai problémákkal. Minimalista — bármely elemsornak a protokoll üzenetéből való eltávolítása a biztonság hiányához vezet.
Gyalogoljunk el a legegyszerűbb házi protokolltól a SIGMA-ig. A legalapvetőbb működés, ami érdekel minket, az : egy olyan funkció, amelyen keresztül mindkét résztvevő ugyanazt az értéket kapja, amelyet szimmetrikus kulcsként használhatnak. Részletekbe nem bonyolódva: mindkét fél generál egy efemer (csak egy szekció keretében használt) kulcspárt (nyilvános és privát kulcsok), kicserélik a nyilvános kulcsaikat, és meghívják az egyesítési funkciót, amelybe az ő privát kulcsukat és a partnerük nyilvános kulcsát adják meg.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│ IdA, PubA │ ╔════════════════════╗
│───────────────>│ ║PrvA, PubA = DHgen()║
│ │ ╚════════════════════╝
│ IdB, PubB │ ╔════════════════════╗
│<───────────────│ ║PrvB, PubB = DHgen()║
│ │ ╚════════════════════╝
────┐ ╔═══════╧════════════╗
│ ║Key = DH(PrvA, PubB)║
<───┘ ╚═══════╤════════════╝
│ │
│ │
Bárki beleavatkozhat, és saját nyilvános kulcsaira cserélheti azokat - ebben a protokollban nincsenek autentikációs lépések. Adjunk hozzá aláírást hosszú élettartamú kulcsokkal.
┌─────┐ ┌─────┐
│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) ║ │
│ ╚═════════════════════╝ │
│ │
Ez az aláírás nem megfelelő, mert nincs konkrét munkamenethez kötve. Az ilyen üzenetek 'megfelelnek' más résztvevőkkel folytatott munkamenetekhez is. Az aláírásnak az egész kontextust fel kell ölelnie. Ez miatt a másik üzenet küldése is szükséges A-tól.
Ezen kívül kritikus, hogy az aláíráshoz a saját azonosítónkat is hozzáadjuk, mert különben helyettesíthetjük az IdXXX-et, és egy másik ismert beszélgetőpartner kulcsával újbóli aláírást generálhatunk. A megelőzés érdekében , szükséges, hogy a felirat alatti elemek pontosan meghatározott helyeken legyenek a jelentésük szerint: ha A aláír (PubA, PubB), akkor B-nek (PubB, PubA) kell aláírnia. Ez azt is jelzi, hogy mennyire fontos a sorozatban lévő adatok struktúrájának és formátumának a megválasztása. Például az ASN.1 DER kódolásban a halmazokat rendezni kell: a SET OF(PubA, PubB) azonos lesz a SET OF(PubB, PubA)-val.
┌─────┐ ┌─────┐ │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) ║ │ │ ╚═════════════════════╝ │ │
Mindazonáltal még nem "bizonyítottuk", hogy az adott szakaszhoz azonos közös kulcsot generáltunk. Elvileg ez a lépés kihagyható — az első szállított üzenet érvénytelen lesz, de azt szeretnénk, hogy amikor a kézfogás befejeződik, biztosak legyünk abban, hogy minden valóban össze van hangolva. Jelenleg a kezünkben van az ISO/IEC IS 9798-3 protokoll.
Aláírhatnánk magát az előállított kulcsot is. Ez kockázatos, mivel nem zárható ki, hogy az használt aláírási algoritmus hibákat rejthet (bár legyenek is bitek-aláírásként, de mégis hibák). Aláírhatunk egy hash-t az előállított kulcsról, de a kulcs hash-ének kiszivárgása is értékes lehet a kulcs-generáló funkció brute-force támadása során. A SIGMA MAC funkciót használ, amely hitelesíti a küldő azonosítóját.
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Optimalizálásként néhányan szerethetik újra felhasználni az efemer kulcsaikat (ami természetesen hátrányos a PFS számára). Például, generáltunk egy kulcspárot, próbáltunk csatlakozni, de a TCP nem volt elérhető, vagy valahol a protokoll közepén megszakadt. Kár volna az elköltött entrópia és CPU-eredmények új párra való költésére. Ezért bevezetjük a cookie-t – egy ál-random értéket, amely megvédhet a véletlen replay támadásoktól az efemer nyilvános kulcsok újra felhasználásakor. A cookie és az efemer nyilvános kulcs közötti kötődés miatt a másik fél nyilvános kulcsát el lehet távolítani a digitális aláírásból, mert nem szükséges.
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Végül azt akarjuk, hogy a résztvevők azonosítói védve legyenek a passzív megfigyelőktől. Ehhez a SIGMA azt javasolja, hogy először cseréljünk efemer kulcsokat, hogy közös kulcsot alkossunk, amelyen az autentikáló és az azonosító üzenetek titkosítva lesznek. A SIGMA két lehetőséget ír le:
- SIGMA-I — védi az iniciátort az aktív támadásoktól, a válaszadót a passzívaktól: az iniciátor autentikálja a válaszadót, és ha valami nem stimmel, akkor nem adja meg az azonosítását. A válaszoló viszont megadja az azonosítását, ha aktív protokollba kezdenek vele. A passzív megfigyelő semmit sem tud meg;
SIGMA-R — védi a válaszolót az aktív támadásoktól, az iniciátort a passzívaktól. Minden éppen fordítva van, de ebben a protokollban már négy üzenet kézfogása történik.A SIGMA-I-t választjuk, mivel jobban hasonlít arra, amit a kliens-szerveres megszokott dolgoktól várunk: a kliens csak a hitelesített szervert ismeri, és a szerver mindent tud. Ráadásul egyszerűbben megvalósítható, mivel kevesebb kézfogás üzenet szükséges. Amit a protokollba bevezetünk, az a üzenet egy részének titkosítása és az A azonosító áthelyezése az utolsó üzenet titkos részébe:
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
- A aláíráshoz a GOST R szabványt használják. Algoritmus 256 bites kulcsokkal.
- A közös kulcs előállításához a 34.10-2012 VKO-t használják.
- MAC-ként a CMAC-ot alkalmazzák. Technikai szempontból ez egy speciális üzemmód a blokk titkosító számára, amelyet a GOST R 34.13-2015 ír le. Ezen üzemmód titkosító függvényeként - (34.12-2015).
- A beszélgetőpartner azonosítója a nyilvános kulcsának hash értéke. A hash érték a következőn alapul (34.11-2012 256 bit).
A kézfogás után egy közös kulcsot állapítunk meg. Ezt használhatjuk a szállított üzenetek hitelesített titkosítására. Ez a rész elég egyszerű, és nehéz benne hibázni: növeljük az üzenetszámlálót, titkosítjuk az üzenetet, hitelesítjük (MAC) a számlálót és a titkosított szöveget, majd elküldjük. Az üzenet fogadásakor ellenőrizzük, hogy a számláló a várt értéket mutatja, hitelesítjük a titkosított szöveget a számlálóval, majd visszatitkosítunk. Melyik kulccsal titkosítsuk a kézfogás és a szállított üzeneteket, és melyikkel hitelesítsük? Egy kulcs használata minden feladatra veszélyes és ésszerűtlen. Szakosodott funkciókat kell kifejlesztenünk a kulcsok előállításához. (kulcsképző funkció). Ismét csak, ne bonyolítsuk túl a dolgot: jól ismert, jól kutatott, és nincsenek ismert problémái. Sajnos a Python alapkönyvtárában nincs ilyen funkció, ezért használjuk a csomagot. Az HKDF belül a , amely viszont hash-funkciót használ. A Python példás implementációja a Wikipedia oldalán néhány sor kódot foglal magában. Ahogyan a 34.10-2012 esetében, a hash-funkcióhoz most is a Stribog-256-ot fogjuk használni. A kulcsmegállapodásunk kimenete a szekciós kulcs néven lesz ismert, amelyből a szükséges szimmetrikus kulcsok kinyerésre kerülnek:
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")Struktúrák/sémák
Nézzük meg, milyen ASN.1 struktúrák jöttek létre most a szükséges adatok átadására:
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 – az, ami aláírásra kerül (to be signed). HandshakeTBE – az, ami titkosítva lesz (to be encrypted). Figyelembe kell venni az ukm mezőt a MsgHandshake1-ben. A 34.10 VKO számára, további véletlenszerűsítés érdekében a generált kulcsokhoz UKM-t (user keying material) adunk – ez egyszerűen egy kiegészítő entropia.
Kriptográfia hozzáadása a kódhoz
Csak az eredeti kódba bevezetett változtatásokat vizsgáljuk, mivel az alapstruktúra változatlan maradt (valójában először a végső implementációt írták meg, majd abból távolították el a teljes kriptográfiát).
Mivel az azonosítást és az azonosító felek hitelesítését a nyilvános kulcsok alapján fogjuk végezni, most már valahol hosszútávon kell tárolnunk őket. Az egyszerűség kedvéért egy ilyen formátumú JSON-t használunk:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }our – a mi kulcspárosunk, hexadecimális privát és nyilvános kulcsok. their – a társalgók nevei és nyilvános kulcsaik. Módosítjuk a parancssori argumentumokat és hozzáadunk egy JSON adatfeldolgozást:
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="JSON generálása új kulcspárral", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON a mi és az ő kulcsaikkal", ) parser.add_argument( "--bind", default="::1", help="Cím, amelyen hallgatni fogunk", ) parser.add_argument( "--port", type=int, default=6666, help="Port, amelyen hallgatni fogunk", ) 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) # Kulcsaink és az ő kulcsaik XML kivonása {{{ with open(args.keys, "rb") as fd: _keys = json.loads(fd.read().decode("utf-8")) KEY_OUR_SIGN_PRV = gost3410.prv_unmarshal(hexdec(_keys["our"]["prv"])) _pub = hexdec(_keys["our"]["pub"]) KEY_OUR_SIGN_PUB = gost3410.pub_unmarshal(_pub) KEY_OUR_SIGN_PUB_HASH = OctetString(GOST34112012256(_pub).digest()) for peer_name, pub_raw in _keys["their"].items(): _pub = hexdec(pub_raw) KEYS[GOST34112012256(_pub).digest()] = { "name": peer_name, "pub": gost3410.pub_unmarshal(_pub), } # }}}A 34.10 algoritmus privát kulcsa — véletlen szám. 256-bites méretű, 256-bites elliptikus görbéhez. A PyGOST nem bájtokkal dolgozik, hanem , ezért a privát kulcsunkat (urandom(32)) számra kell alakítani a gost3410.prv_unmarshal() segítségével. A publikus kulcs determinisztikusan számítható a privát kulcsból a gost3410.public_key() használatával. A 34.10 publikus kulcs két nagy számot tartalmaz, amelyeket bájtok sorozatává kell alakítani a könnyebb tárolás és átvitel érdekében a gost3410.pub_marshal() segítségével.
A JSON fájl beolvasása után a publikus kulcsokat vissza kell alakítani a gost3410.pub_unmarshal() alkalmazásával. Mivel a partnereink azonosítói a publikus kulcs hash-éből érkeznek, azokat előre kiszámíthatjuk és elhelyezhetjük egy szótárban a gyors keresés érdekében. A Stribog-256 hash a gost34112012256.GOST34112012256(), amely teljes mértékben megfelel a hashlib hash-függvények interfészének.
Hogyan változott az indító korutina? Minden, mint a kézfogási sémában: cookie-t generálunk (a 128 bites bőven elegendő), egy efemer kulcspárt 34.10, amelyet a kulcsmegosztási VKO funkcióhoz fogunk használni.
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áljuk ephemerális nyilvános kulcsunkat és a sütit, küldjük a Handshake 0 üzenetet {{{ 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()- várjuk a választ és dekódoljuk a beérkező Msg üzenetet;
- meggyőződünk arról, hogy megkaptuk a handshake1-t;
- dekódoljuk a másik fél ephemerális nyilvános kulcsát és számítjuk ki a session kulcsot;
- szimmetrikus kulcsokat állítunk elő, amelyek szükségesek a TBE üzenet feldolgozásához.
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 # Ellenőrizzük a Handshake üzenetet {{{ 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")Az UKM egy 64-bites szám (urandom(8)), amely szintén deszerializálást igényel byte formátumból, a gost3410_vko.ukm_unmarshal() használatával. A VKO függvény a 34.10-2012 256-biteshez a gost3410_vko.kek_34102012256() (KEK — kulcskódoló kulcs).
A létrehozott session kulcs már egy 256-bites byte álomrandom sorozat, így ezt azonnal fel lehet használni az HKDF függvényben. Mivel a GOST34112012256 megfelel a hashlib interfésznek, az közvetlenül használható az Hkdf osztályban. A sót (az HKDF első argumentumát) nem adjuk meg, mivel a létrehozott kulcs az érintett kulcspárok ephemerális voltából kifolyólag minden sessionben más lesz és már elegendő entrópiát tartalmaz. A kdf.expand() alapértelmezés szerint már a 256-bites kulcsokat adja vissza, amelyek a Kuznyec számára szükségesek a továbbiakban.
Ezután ellenőrizzük a beérkező üzenet TBE és TBS részeit:
- MAC-et számítunk és ellenőrizzük a beérkező titkosított szövegen;
- megfejtjük a titkosított szöveget;
- dekódoljuk a TBE struktúrát;
- ebből levesszük a beszélgetőpartner azonosítóját, és ellenőrizzük, hogy ismerjük-e egyáltalán;
- MAC-et számítunk és ellenőrizzük ezen az azonosítón;
- A TBS struktúra aláírásának ellenőrzése történik, amely magában foglalja mindkét fél cookie-ját és a másik fél nyilvános efemer kulcsát. Az aláírást az illető hosszú életű aláíró kulcsával ellenőrzik.
441 próbáld: 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 kivéve ValueError err: 452 logging.warning("%s: %s, bontás", _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 ha nem egyezik a mac_tag, bytes(msg_handshake["ciphertextMac"]) -t: 140 raise ValueError("érvénytelen MAC") 141 plaintext = ctr( 142 GOST3412Kuznechik(key_enc).encrypt, 143 KUZNECHIK_BLOCKSIZE, 144 ciphertext, 145 8 * b"x00", 146 ) 147 próbáld: 148 tbe, _ = HandshakeTBE().decode(plaintext) 149 kivéve ASN1Error: 150 raise ValueError("nem tudom dekódolni a TBE-t") 151 key_sign_pub_hash = bytes(tbe["identity"]) 152 peer = KEYS.get(key_sign_pub_hash) 153 ha a peer None: 154 raise ValueError("ismeretlen identitás") 155 mac_tag = mac( 156 GOST3412Kuznechik(key_mac_identity).encrypt, 157 KUZNECHIK_BLOCKSIZE, 158 key_sign_pub_hash, 159 ) 160 ha nem egyezik a mac_tag, bytes(tbe["identityMac"]) -t: 161 raise ValueError("érvénytelen identitás MAC") 162 tbs = HandshakeTBS(( 163 ("cookieTheir", cookie_their), 164 ("cookieOur", cookie_our), 165 ("pubKeyOur", pub_key_our), 166 )) 167 ha nem validálja a gost3410. 168 CURVE, 169 peer["pub"], 170 GOST34112012256(tbs.encode()).digest(), 171 bytes(tbe["signature"]), 172 ): 173 raise ValueError("érvénytelen aláírás") 174 return peer["name"]Ahogy fentebb már említettem, a 34.13-2015 különféle a 34.12-2015-ből. Ezek között szerepel az üzenet integritási kód (MAC) előállításának módja is. A PyGOST-ban ez a gost3413.mac(). Ez a mód megköveteli a titkosító függvény (amely egyetlen adatblokkokat fogad el és ad vissza), a blokkméret és a tényleges adatok átvitelét. Miért nem lehet a blokkméretet hardcode-olni? A 34.12-2015 nemcsak a 128 bites Kuznecik titkosítót írja le, hanem a 64 bites — egy kicsit módosított ГОСТ 28147-89, amelyet még a KGB készített, és még mindig az egyik legmagasabb biztonsági küszöböt képviseli.
A Góliát inicializálja a gost.3412.GOST3412Kuznechik(key) hívással, és egy olyan objektumot ad vissza, amely rendelkezik az .encrypt() / .decrypt() módszerekkel, amelyek alkalmasak a 34.13 funkciókhoz. A MAC az alábbiak szerint számítódik: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). A számított és az érkezett MAC összehasonlítására nem használható a szokásos összehasonlítás (==) bájtok sztringjei között, mivel ez az eljárás időbeli összehasonlítási szivárgást eredményez, ami a legtöbb esetben halálos sebezhetőségeket okozhat. TLS elleni támadásokkal. A Pythonban létezik egy különleges hmac.compare_digest függvény erre.
A blokk titkosító függvény csak egy adatblokkot tud titkosítani. Több blokk, főleg ha nem többszöröse a méretnek, titkosítási mód alkalmazását igényli. A 34.13-2015 dokumentálja a következőket: ECB, CTR, OFB, CBC, CFB. Mindegyiknek megvannak a saját alkalmazási területei és jellemzői. Sajnos jelenleg nincsenek standardizált (mint a CCM, OCB, GCM és hasonlók) — nekünk magunknak legalább MAC-ot kell hozzáadnunk. Én a (CTR) választom: nem igényel kiegészítést blokk méretig, párhuzamosítható, csak titkosító függvényt használ, és biztonságosan használható nagy mennyiségű üzenet titkosítására (ellentétben a CBC-vel, ahol viszonylag gyorsan kezdődnek a ütközések).
Ahogyan a .mac() is, a .ctr() hasonló adatokat vár bemenetként: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). Meg kell adni egy inicializáló vektort, amely pontosan a titkosító blokk felének hosszú. Ha a titkosítási kulcsunkat csak egy üzenet titkosítására használjuk (még ha az több blokkból is áll), akkor biztonságosan megadható a nulla inicializáló vektor. A handshake üzenetek titkosításához mindig külön kulcsot használunk.
A gost3410.verify() aláírás ellenőrzése triviális: átadjuk az elliptikus görbét, amelyen dolgozunk (ezt egyszerűen rögzítjük a GOSTIM protokollban), a publikus kulcsot az aláírótól (ne feledjük, hogy ennek egy két nagy számot tartalmazó tömbnek kell lennie, nem bájtok sztringjének), a 34.11-2012 hash-t és a megérkezett aláírást.
Ezután az iniciátorban előkészítjük és elküldjük a handshake2 üzenetet, a kézfogás során, ugyanazokat a lépéseket követve, mint az érvényesítésnél, csak szimmetrikusan: aláírás a saját kulcsainkon a hitelesítés helyett, stb...
456 # Készítse el és küldje el a Handshake 2 üzenetet {{{ 457 tbs = HandshakeTBS(( 458 ("cookieTheir", cookie_their), 459 ("cookieOur", cookie_our), 460 ("pubKeyOur", pub_our_raw), 461 )) 462 signature = gost3410.sign( 463 CURVE, 464 KEY_OUR_SIGN_PRV, 465 GOST34112012256(tbs.encode()).digest(), 466 ) 467 key_handshake2_mac_identity = kdf.expand(b"handshake2-mac-identity") 468 mac_tag = mac( 469 GOST3412Kuznechik(key_handshake2_mac_identity).encrypt, 470 KUZNECHIK_BLOCKSIZE, 471 bytes(KEY_OUR_SIGN_PUB_HASH), 472 ) 473 tbe = HandshakeTBE(( 474 ("identity", KEY_OUR_SIGN_PUB_HASH), 475 ("signature", OctetString(signature)), 476 ("identityMac", MAC(mac_tag)), 477 )) 478 tbe_raw = tbe.encode() 479 key_handshake2_enc = kdf.expand(b"handshake2-enc") 480 key_handshake2_mac = kdf.expand(b"handshake2-mac") 481 ciphertext = ctr( 482 GOST3412Kuznechik(key_handshake2_enc).encrypt, 483 KUZNECHIK_BLOCKSIZE, 484 tbe_raw, 485 8 * b"x00", 486 ) 487 mac_tag = mac( 488 GOST3412Kuznechik(key_handshake2_mac).encrypt, 489 KUZNECHIK_BLOCKSIZE, 490 ciphertext, 491 ) 492 writer.write(Msg(("handshake2", MsgHandshake2(( 493 ("ciphertext", OctetString(ciphertext)), 494 ("ciphertextMac", MAC(mac_tag)), 495 )))).encode()) 496 # }}} 497 await writer.drain() 498 logging.info("%s: session established: %s", _id, peer_name)A session létrejötte után szállító kulcsok jönnek létre (külön kulcs az eljáráshoz, autenticate-hez, mindkét fél számára), és a Kuznyecik inicializálva van a visszafejtésre és a MAC ellenőrzésére:
499 # Futassa a szöveges üzenetküldőt, inicializálja a szállító dekódert {{{ 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 # Várjon a teszt üzenetekre {{{ 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 # }}}A msg_sender coroutine mostantól titkosítja az üzeneteket, mielőtt elküldené azokat a TCP-kapcsolaton. Minden üzenet monoton növekvő nonce-t tartalmaz, amely egyben inicializáló vektor is a számláló üzemmódban történő titkosításhoz. Minden üzenetnek és üzenetblokkán a számláló értékei garantáltan eltérőek lesznek.
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 += 1A bejövő üzeneteket a msg_receiver rutin dolgozza fel, amely az azonosítással és dekódolással foglalkozik:
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("váratlan nonce érték") mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode()) if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])): raise ValueError("érvénytelen MAC") plaintext = ctr( encrypter, KUZNECHIK_BLOCKSIZE, bytes(payload["ciphertext"]), long2bytes(nonce_expected, 8), ) text = plaintext.decode("utf-8") await OUT_QUEUES[peer_name].put(text)Zárás
A GOSTIM kizárólag oktatási célokra használható (mivel minimum tesztelés nem történt)! A program forráskódja letölthető (Stribog-256 hash: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Mint az összes projektem, típusú , , , , a GOSTIM teljes mértékben , amely a következő feltételekkel terjeszthető .
, , a következő tagja , Python/Go-fejlesztő, vezető szakember .
Forrás: habr.com
