Jako vývojář knihovny (GOST kryptografické primitivy v čistém Pythonu) se často setkávám s dotazy, jak na koleni implementovat základní bezpečnou výměnu zpráv. Mnozí si myslí, že aplikovaná kryptografie je dost jednoduchá věc, a že vola .encrypt() u blokového šifrování bude dostačující pro bezpečný přenos přes komunikační kanál. Jiní se domnívají, že aplikovaná kryptografie je domenou několika málo, a je akceptovatelné, že bohaté firmy jako Telegram s olympijskými matematiky bezpečný protokol.
To mě vedlo k napsání tohoto článku, abych ukázal, že implementace kryptografických protokolů a bezpečného IM není tak složitý úkol. Nicméně, nevynalézejte vlastní autentizační a klíčové dohody.
V článku bude popsán , , instant messenger s protokolem autentizace a dohody o klíčích (na jehož základě je realizován ), přičemž používáme výhradně GOST kryptografické algoritmy z knihovny PyGOST a ASN.1 kódování zpráv pomocí knihovny (o které jsem již ). Nezbytným požadavkem je, že musí být dostatečně jednoduché, aby jej bylo možné napsat od nuly za jediný večer (nebo pracovní den), jinak to už není jednoduchý program. Jistě obsahuje chyby, zbytečné složitosti a nedostatky, navíc je to můj první program s využitím knihovny asyncio.
Design IM
Nejdříve je třeba pochopit, jak bude vypadat náš IM. Pro zjednodušení to bude peer-to-peer síť, bez jakéhokoli odhalování účastníků. Ručně budeme udávat, na jakou adresu: port se připojit pro komunikaci s partnerem.
Chápu, že v současnosti je předpoklad o dostupnosti přímého spojení mezi dvěma libovolnými počítači - zásadní omezení použitelnosti IM v praxi. Ale čím více vývojářů implementuje různé NAT-traversal triky, tím déle zůstaneme v IPv4 internetu, s deprimující pravděpodobností spojení mezi libovolnými počítači. Jak dlouho ještě budeme snášet absenci IPv6 doma a v práci?
Budeme mít síť friend-to-friend: všichni možní účastníci musí být předem známí. Za prvé, to všechno výrazně zjednodušuje: představíme se, najdeme nebo nenajdeme jméno/klíč, odpojíme se nebo pokračujeme v práci s vědomím, koho se chceme spojit. Za druhé, v obecném případě je to bezpečné a vylučuje spoustu útoků.
Rozhraní IM bude blízké klasickým řešením. , které se mi opravdu líbí svým minimalismem a filozofií Unix-way. Program IM pro každého účastníka vytváří adresář se třemi Unix domain sockety:
- in — do něj se zapisují odesílané zprávy účastníkovi;
- out — z něj se čtou zprávy přijímané od účastníka;
- state — čtením z něj zjistíme, zda je účastník aktuálně připojen, adresa/port připojení.
Kromě toho se vytváří conn socket, do kterého zapíšeme host port a iniciujeme připojení k vzdálenému účastníkovi.
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Takový přístup umožňuje nezávislé implementace IM transportu a uživatelského rozhraní, protože proti vkusu a barvě kolegy není. Použitím a/nebo , lze získat vícenásobné rozhraní s syntaktickým zvýrazněním. A pomocí můžeme získat řádek pro zadávání zpráv kompatibilní se GNU Readline.
Ve skutečnosti suckless projekty používají FIFO soubory. Osobně jsem nedokázal pochopit, jak pracovat s nelimitovanými soubory v asyncio bez ručně vytvořené podpory z dedikovaných vláken (pro takové věci už dlouho používám jazyk ). Proto jsem se rozhodl obejít se bez Unix domain socketů. Bohužel to zbavuje možnosti provést echo 2001:470:dead::babe 6666 > conn. Tento problém jsem vyřešil pomocí : echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
Původní nezabezpečený protokol
Jako transport se používá TCP: zaručuje dodání a jeho pořadí. UDP nezaručuje ani jedno, ani druhé (což by bylo užitečné, když se použije kryptografie), a podpora v Pythonu není přímo k dispozici.
Bohužel, v TCP neexistuje pojem zprávy, pouze tok bytů. Proto je nutné vymyslet formát pro zprávy, aby je bylo možné mezi sebou oddělovat v tomto toku. Můžeme se dohodnout na použití znaku nového řádku. Pro začátek to stačí, ale když začneme šifrovat naše zprávy, může se tento znak objevit kdekoli v šifrovaném textu. V sítích jsou proto populární protokoly, které nejprve zasílají délku zprávy v bytech. Například v Pythonu je k dispozici knihovna xdrlib, která umožňuje pracovat s takovým formátem. .
Nebudeme správně a efektivně pracovat se čtením TCP — zjednodušíme kód. V nekonečné smyčce čteme data ze socketu, dokud neodešleme celou zprávu. Jako formát pro tento přístup můžeme použít i JSON s XML. Ale když se přidá kryptografie, bude nutné data podepisovat a autentizovat — a to si vyžádá identické bajtové zastoupení objektů, což JSON/XML nezajišťují (výstup se může lišit).
XDR je vhodné pro tuto úlohu, ale já volím ASN.1 s DER kódováním a knihovnu, protože budeme mít k dispozici vysoceúrovňové objekty, se kterými se často lépe a pohodlněji pracuje. Na rozdíl od bezschémových , nebo , ASN.1 automaticky zkontroluje data v souladu s pevně danou schématem.
# 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))),
))
Přijatou zprávou bude Msg: buď textové MsgText (zatím s jedním textovým polem), nebo zpráva handshakování MsgHandshake (ve které je předáváno jméno protistrany). Nyní to vypadá příliš složitě, ale je to (investice) do budoucna.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│MsgHandshake(IdA) │
│─────────────────>│
│ │
│MsgHandshake(IdB) │
│<─────────────────│
│ │
│ MsgText() │
│─────────────────>│
│ │
│ MsgText() │
│<─────────────────│
│ │
IM bez kryptografie
Jak jsem již řekl, pro všechny operace se sockety bude použita knihovna asyncio. Vyhlásíme, co očekáváme v okamžiku spuštění:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Naše jméno peeru",
)
parser.add_argument(
"--their-names",
required=True,
help="Jejich jména peerů, oddělená čárkami",
)
parser.add_argument(
"--bind",
default="::1",
help="Adresa pro naslouchání",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Port pro naslouchání",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
Nastaví se vlastní jméno (—our-name alice). Všechny očekávané protějšky se vyjmenují čárkou (—their-names bob,eve). Pro každé z protějšků se vytvoří adresář s Unix sockety a také se vytvoří korutina pro každé 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"))
Příchozí zprávy od uživatele ze socketu in se odesílají do front 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"))
Příchozí zprávy od protějšků se odesílají do front OUT_QUEUES, ze kterých se data zapisují do socketu 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()
Při čtení ze socketu state program hledá v slovníku PEER_ALIVE adresu protějšku. Pokud připojení k protějšku dosud neexistuje, zapíše se prázdný řetězec.
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()
Při zápisu adresy do socketu conn se spouští funkce „iniciátor“ připojení:
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))
Podívejme se na iniciátora. Nejprve se samozřejmě otevře připojení k zadanému hostiteli/portu a odešle se handshake zpráva se svým jménem:
130 async def initiator(host, port):
131 _id = repr((host, port))
132 logging.info("%s: vytáčení", _id)
133 reader, writer = await asyncio.open_connection(host, port)
134 # Zpráva handshake {{{
135 writer.write(Msg(("handshake", MsgHandshake((
136 ("peerName", OUR_NAME),
137 )))).encode())
138 # }}}
139 await writer.drain()
Pak čeká na odpověď od vzdálené strany. Snaží se dekódovat přijatou odpověď podle schématu Msg ASN.1. Předpokládáme, že celá zpráva bude odeslána v jednom TCP segmentu a my ji atomicky obdržíme při volání .read(). Ověřujeme, že jsme obdrželi skutečně handshake zprávu.
141 # Čekání na zprávu handshake {{{
142 data = await reader.read(256)
143 if data == b"":
144 logging.warning("%s: žádná odpověď, odpojování", _id)
145 writer.close()
146 return
147 try:
148 msg, _ = Msg().decode(data)
149 except ASN1Error:
150 logging.warning("%s: nečitelná odpověď, odpojování", _id)
151 writer.close()
152 return
153 logging.info("%s: obdržel %s zprávu", _id, msg.choice)
154 if msg.choice != "handshake":
155 logging.warning("%s: nečekaná zpráva, odpojování", _id)
156 writer.close()
157 return
158 # }}}
Ověřujeme, že příchozí jméno partnera nám je známo. Pokud ne, tak ukončujeme spojení. Kontrolujeme, zda s ním již spojení nebylo navázáno (partner opět dal příkaz k připojení) a ukončujeme ho. Do FRONTY_IN se vkládají Python řetězce s textem zprávy, ale existuje zvláštní hodnota None, která SIGNALIZUJE, že korutina msg_sender by měla přestat pracovat, aby zapomněla na svůj writer spojený se zastaralým TCP spojením.
159 msg_handshake = msg.value
160 peer_name = str(msg_handshake["peerName"])
161 if peer_name not in THEIR_NAMES:
162 logging.warning("neznámé jméno partnera: %s", peer_name)
163 writer.close()
164 return
165 logging.info("%s: seance navázána: %s", _id, peer_name)
166 # Spustit odesílatele textových zpráv, inicializovat dekodér transportu {{{
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 přijímá odchozí zprávy (vkládané do fronty z in socketu), serializuje je do zprávy MsgText a odesílá je přes TCP spojení. To se může kdykoli přerušit — to zachycujeme explicitně.
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: odesláno %d znaků zprávy", peer_name, len(text))
Na konci iniciátor vstupuje do nekonečné smyčky pro čtení zpráv ze socketu. Kontroluje, zda se jedná o textové zprávy, a ukládá je do OUT_QUEUES, ze které budou odeslány do výstupního socketu odpovídajícího partnera. Proč prostě neprovést .read() a nerozluštit zprávu? Protože není vyloučena situace, kdy bude v operačním systému agregováno několik zpráv od uživatele do jednoho TCP segmentu. Rozluštit můžeme první, ale v bufferu může zůstat část následující zprávy. V případě jakékoli mimořádné situace zavíráme TCP spojení a zastavujeme korutinu msg_sender (odeslaním None do OUT_QUEUES).
174 buf = b""
175 # Čekání na testovací zprávy {{{
176 while True:
177 data = await reader.read(MaxMsgLen)
178 if data == b"":
179 break
180 buf += data
181 if len(buf) > MaxMsgLen:
182 logging.warning("%s: maximální velikost bufferu překročena", _id)
183 break
184 try:
185 msg, tail = Msg().decode(buf)
186 except ASN1Error:
187 continue
188 buf = tail
189 if msg.choice != "text":
190 logging.warning("%s: neočekávaná %s zpráva", _id, msg.choice)
191 break
192 try:
193 await msg_receiver(msg.value, peer_name)
194 except ValueError as err:
195 logging.warning("%s: %s", err)
196 break
197 # }}}
198 logging.info("%s: odpojení: %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: přijatá zpráva o délce %d znaků", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Vraťme se k hlavnímu kódu. Po vytvoření všech korutin, v okamžiku spuštění programu, spouštíme TCP server. Pro každé navázané spojení vytváří responder (odpovídač) korutinu.
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("Naslouchání na: %s", server.sockets[0].getsockname())
loop.run_forever()
responder je podobný initiatorovi a zrcadlově vykonává všechny stejná opatření, ale nekonečná smyčka pro čtení zpráv se spouští okamžitě, pro jednoduchost. Nyní protokol handshake odesílá jednou zprávu z každé strany, ale později bude odesílání dvou zpráv od iniciátora spojení, po kterém bude možné okamžitě odeslat textové zprávy.
72 async def responder(reader, writer):
73 _id = writer.get_extra_info("peername")
74 logging.info("%s: připojeno", _id)
75 buf = b""
76 msg_expected = "handshake"
77 peer_name = None
78 while True:
79 # Čtení, dokud nedostaneme Msg message {{{
80 data = await reader.read(MaxMsgLen)
81 if data == b"":
82 logging.info("%s: uzavřeno připojení", _id)
83 break
84 buf += data
85 if len(buf) > MaxMsgLen:
86 logging.warning("%s: překročena maximální velikost bufferu", _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: neočekávaná %s zpráva", _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 # Zpracování zprávy handshake {{{
104 elif msg_expected == "handshake":
105 logging.info("%s: obdržena %s zpráva", _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("neznámé jméno peeru: %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: relace navázána: %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: odpojení", _id)
125 if msg_expected == "text":
126 IN_QUEUES[peer_name].put(None)
127 writer.close()
Bezpečný protokol
Nastal čas zabezpečit naši komunikaci. Co přesně myslíme bezpečností a co očekáváme:
- důvěrnost předávaných zpráv;
- autenticita a integrita předávaných zpráv — změny musí být detekovány;
- ochrana proti útokům přehrání (replay attack) — ztráta nebo opakování zpráv musí být odhalena (a rozhodneme se ukončit spojení);
- identifikace a autentizace komunikačních partnerů na základě předem zadaných veřejných klíčů — již jsme se dříve rozhodli, že vytvoříme síť friend-to-friend. Teprve po autentizaci pochopíme, s kým komunikujeme;
- přítomnost vlastnosti (PFS) — kompromitace našeho dlouhotrvajícího podpisového klíče by neměla umožnit čtení celé předchozí korespondence. Záznam zachyceného provozu se stává k ničemu;
- platnost zpráv (transportních a handshake) pouze v rámci jedné TCP relace. Vkládání správně podepsaných/autentizovaných zpráv z jiné relace (i s tímto stejným protějškem) by nemělo být možné;
- pasivní pozorovatel by neměl vidět ani identifikátory uživatelů, ani předávané dlouhoživoucí veřejné klíče, ani jejich hashe. Jakási anonymita vůči pasivnímu pozorovateli.
Úžasné je, že toto minimum chce mít téměř každý v jakémkoli handshake protokolu, a z mála z uvedeného se nakonec skutečně plní pro "domácí" protokoly. Tak nyní nebudeme vymýšlet nové. Určitě bych doporučil použít pro budování protokolů, ale vyberme něco jednoduššího.
Nejpopulárnější jsou dva protokoly:
- — velmi složitý protokol s dlouhou historií chyb, nedostatků, zranitelností, slabého promyšlení, komplikovanosti a nedostatků (nicméně, to se týká jen málo TLS 1.3). Ale nezvažujeme ho kvůli přílišné složitosti.
- s — nemají vážné kryptografické problémy, i když také nejsou jednoduché. Pokud si přečtete o IKEv1 a IKEv2, jejich původními zdroji jsou , ISO/IEC IS 9798-3 a SIGMA (SIGn-and-MAc) protokoly — dostatečně jednoduché na implementaci za jeden večer.
Čím je SIGMA, jako poslední článek vývoje STS/ISO protokolů, dobrý? Splňuje všechny naše požadavky (včetně "skrytí" identifikátorů protějšků), nemá známé kryptografické problémy. Je minimalistický — odstranění alespoň jedné položky ze zprávy protokolu povede k jeho nebezpečnosti.
Pojďme projít od nejzákladnějšího domácího protokolu k SIGMA. Nejdůležitější operace, která nás zajímá, je : funkce, při které oba účastníci obdrží stejnou hodnotu, kterou lze použít jako symetrický klíč. Bez hlubokých podrobností: každá ze stran generuje efemérní (používající se pouze v rámci jedné relace) klíčový pár (veřejný a soukromý klíč), vyměňují si veřejné klíče, spouštějí funkci vyjednávání, do které předávají svůj soukromý klíč a veřejný klíč protějšku.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│ IdA, PubA │ ╔════════════════════╗
│───────────────〉│ ║PrvA, PubA = DHgen()║
│ │ ╚════════════════════╝
│ IdB, PubB │ ╔════════════════════╗
│〈───────────────│ ║PrvB, PubB = DHgen()║
│ │ ╚════════════════════╝
────┐ ╔═══════╧════════════╗
│ ║Key = DH(PrvA, PubB)║
〈───┘ ╚═══════╤════════════╝
│ │
│ │
Kdokoli může zasáhnout a vyměnit veřejné klíče za svoje vlastní - v tomto protokolu není autentizace účastníků. Přidáme podpis dlouhotrvajícími klíči.
┌─────┐ ┌─────┐
│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) ║ │
│ ╚═════════════════════╝ │
│ │
Takový podpis nebude vhodný, protože není vázán na konkrétní relaci. Takové zprávy budou „vyhovovat“ i pro relace s jinými účastníky. Podpis by měl zahrnovat celý kontext. To také nutí přidat odeslání další zprávy od A.
Kromě toho je kritické přidat pod podpis i svůj vlastní identifikátor, jinak můžeme zaměnit IdXXX a znovu podepsat zprávu klíčem jiného známého účastníka. Pro zamezení , je nutné, aby se prvky pod podpisem nacházely na přesně určených místech podle svého významu: pokud A podepisuje (PubA, PubB), pak B musí podepisovat (PubB, PubA). To také ukazuje na důležitost volby struktury a formátu serializovaných dat. Například množiny v kódování ASN.1 DER jsou seřazeny: SET OF(PubA, PubB) bude shodné s 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) ║ │ │ ╚═════════════════════╝ │ │
Nicméně stále jsme „neprokázali“, že jsme vyprodukovali stejný společný klíč pro tuto relaci. V zásadě se lze obejít i bez tohoto kroku — první transportní zpráva bude neplatná, ale chceme, aby po dokončení handshake bylo jisté, že je vše skutečně dohodnuto. V tuto chvíli máme k dispozici protokol ISO/IEC IS 9798-3.
Mohli bychom podepsat i samotný vygenerovaný klíč. To je nebezpečné, protože nelze vyloučit, že ve použitém podpisovém algoritmu mohou být úniky (i když jde o bity na podpis, stále jsou to úniky). Je možné podepisovat hash z vygenerovaného klíče, ale únik i hashe z vygenerovaného klíče může mít hodnotu při útoku hrubou silou na funkci generování. SIGMA používá MAC funkci, která autentizuje identifikátor odesílatele.
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Jako optimalizaci se někteří mohou rozhodnout znovu použít své efemérní klíče (což je samozřejmě nešťastné pro PFS). Například jsme vygenerovali pár klíčů, pokusili se připojit, ale TCP nebylo dostupné nebo se někde uprostřed protokolu přerušilo. Je škoda utrácet vynaloženou entropii a procesorové zdroje na nový pár. Proto zavedeme takzvaný cookie — pseudonáhodnou hodnotu, která ochrání před možnými náhodnými replay útoky při opakovaném používání efemérních veřejných klíčů. Díky vazbě mezi cookie a efemérním veřejným klíčem lze veřejný klíč protistrany vyjmout z podpisu jako nadbytečný.
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Nakonec chceme ochránit soukromí našich identifikátorů komunikujících stran před pasivním sledováním. K tomu SIGMA navrhuje nejprve vyměnit efemérní klíče, vypracovat společný klíč, kterým se zašifrují autentifikační a identifikační zprávy. SIGMA popisuje dvě varianty:
- SIGMA-I — chrání iniciátora před aktivními útoky, ale odpovědi před pasivními: iniciátor autentifikuje odpověď a pokud něco nesedí, svou identifikaci nezveřejňuje. Odpověď však poskytne svou identifikaci, pokud s ním začne aktivní protokol. Pasivní pozorovatel se nic nedozví;
SIGMA-R — chrání odpověď před aktivními útoky, iniciátora před pasivními. Všechno je přesně obráceně, ale v tomto protokolu se již přenášejí čtyři handshake zprávy.Vybrali jsme SIGMA-I, protože se více podobá tomu, co očekáváme od tradičních klient-serverových věcí: klient zná pouze autentizovaný server, zatímco server ví všechno. Navíc je jednodušší na implementaci kvůli menšímu počtu handshake zpráv. Vše, co do protokolu vkládáme, je šifrování části zprávy a přesun identifikátoru A do šifrované části poslední zprávy:
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
- Pro podepisování se používá GOST R algoritmus s 256bitovými klíči.
- Pro generování společného klíče se používá 34.10-2012 VKO.
- Jako MAC se používá CMAC. Technicky se jedná o speciální režim práce blokového šifru, popsaný v GOST R 34.13-2015. Jako šifrovací funkce pro tento režim — (34.12-2015).
- Jako identifikátor protistrany se používá hash z jejího veřejného klíče. Pro hash se aplikuje (34.11-2012 256 bitů).
Po zahájení spojení budeme mít dohodnutý sdílený klíč. Ten můžeme použít pro autentizované šifrování přenosových zpráv. Tato část je zcela jednoduchá a je těžké se v ní zmýlit: inkrementujeme čítač zpráv, šifrujeme zprávu, autentizujeme (MAC) čítač a šifrovaný text, odešleme. Při příjmu zprávy zkontrolujeme, že čítač má očekávanou hodnotu, autentizujeme šifrovaný text s čítačem, a dekódujeme. Jakým klíčem šifrovat zprávy po zahájení spojení, přenosové, jakým autentizovat? Použít jeden klíč pro všechny tyto úkoly je nebezpečné a nerozumné. Je nutné vytvářet klíče pomocí specializovaných funkcí (funkce odvozování klíče). Opět, nebudeme vymýšlet složitosti: je dlouho známa, dobře prozkoumaná a nemá známé problémy. Bohužel v nativní knihovně Python není tato funkce, takže použijeme balíček. HKDF uvnitř používá , který, na oplátku, používá hashovací funkci. Příklad implementace na Pythonu na stránce Wikipedia zabírá jen pár řádků kódu. Stejně jako u 34.10-2012 použijeme jako hashovací funkci Stрибог-256. Výstup naší funkce dohodování klíčů se bude nazývat sdílený klíč, ze kterého budou odvozeny chybějící symetrické:
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")Struktury/diagramy
Podíváme se, jaké ASN.1 struktury jsme nyní vytvořili pro přenos všech těchto dat:
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 — to, co bude podepsáno (to be signed). HandshakeTBE — to, co bude zašifrováno (to be encrypted). Všimněte si pole ukm v MsgHandshake1. 34.10 VKO, pro ještě větší randomizaci generovaných klíčů, obsahuje parametr UKM (user keying material) — to je prostě dodatečná entropie.
Přidání kryptografie do kódu
Zvažme pouze změny provedené v původním kódu, neboť rámec zůstal stejný (ve skutečnosti byla nejprve napsána konečná implementace a poté byla z ní odstraněna veškerá kryptografie).
Jelikož bude autentizace a identifikace účastníků provedena pomocí veřejných klíčů, je nyní potřeba je někde dlouhodobě uchovávat. Pro zjednodušení použijeme JSON takového typu:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }our — náš klíčový pár, hexadecimální soukromý a veřejné klíče. their — jména účastníků a jejich veřejné klíče. Změníme argumenty příkazové řádky a přidáme postprocessing JSON dat:
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="Generujte JSON s naším novým klíčovým párem", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON s našimi a jejich klíči", ) parser.add_argument( "--bind", default="::1", help="Adresa pro poslouchání", ) parser.add_argument( "--port", type=int, default=6666, help="Port pro poslouchání", ) 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) # Parsování a unmarshalling našich a jejich klíčů {{{ 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), } # }}}Soukromý klíč algoritmu 34.10 — náhodné číslo. O velikosti 256 bitů pro 256bitové elliptické křivky. PyGOST nepracuje s byte-array, ale s , takže náš soukromý klíč (urandom(32)) je třeba převést na číslo pomocí gost3410.prv_unmarshal(). Veřejný klíč je deterministicky počítán ze soukromého klíče pomocí gost3410.public_key(). Veřejný klíč 34.10 — dvě velká čísla, která je také třeba převést na byte-sekvenci pro snazší uchování a přenos, pomocí gost3410.pub_marshal().
Po přečtení JSON souboru je třeba veřejné klíče opět převést zpět pomocí gost3410.pub_unmarshal(). Jelikož nám přicházejí identifikátory partnerů ve formě hashe z veřejného klíče, můžeme je ihned předem spočítat a umístit do slovníku pro rychlé vyhledávání. Stribog-256 hash je gost34112012256.GOST34112012256(), plně vyhovující hashlib rozhraní hash funkcí.
Jak se změnila korutina iniciátora? Všechno, jak v schématu handshake: generujeme cookie (128 bitů je zcela dostačující), efemérní klíčový pár 34.10, který bude použit pro VKO funkci dohody o klíči.
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 # Vytvoříme náš efemérní veřejný klíč a cookie, pošleme zprávu 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()- čekáme na odpověď a dekódujeme přijatou zprávu Msg;
- ujistíme se, že jsme obdrželi handshake1;
- dekódujeme efemérní veřejný klíč protistrany a vypočítáme session key;
- vytváříme symetrické klíče potřebné pro zpracování TBE části zprávy.
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 # Ověření zprávy 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 je 64bitové číslo (urandom(8)), které také vyžaduje deserializaci z bytového zastoupení pomocí gost3410_vko.ukm_unmarshal(). VKO funkce pro 34.10-2012 256bit je gost3410_vko.kek_34102012256() (KEK — klíč pro šifrování klíčů).
Vytvořený session key je již 256bitová bytová pseudonáhodná posloupnost. Proto může být okamžitě použit v HKDF funkci. Protože GOST34112012256 splňuje hashlib rozhraní, může být ihned použit v Hkdf třídě. Sůl (první argument Hkdf) neuvádíme, protože vytvořený klíč kvůli efemérním účastnickým klíčovým párům bude různý pro každou relaci a již obsahuje dostatek entropie. kdf.expand() ve výchozím nastavení již vydává klíče o délce 256bitů potřebné pro Kyznečka dále.
Dále se kontrolují TBE a TBS části přijaté zprávy:
- ověřuje se a počítá MAC nad přijatým šifrovaným textem;
- šifrovaný text se dešifruje;
- struktura TBE se dekóduje;
- z ní se vezme identifikátor protistrany a kontroluje se, zda je nám vůbec znám;
- počítá se a kontroluje MAC nad tímto identifikátorem;
- ověřuje se podpis nad TBS strukturou, do které patří cookie obou stran a veřejný efemérní klíč opačné strany. Podpis se ověřuje dlouhodobým klíčem podpisu protistrany.
441 pokus: 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 kromě ValueError jako err: 452 logging.warning("%s: %s, odpojuji se", _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 pokud neporovnáte_digest(mac_tag, bytes(msg_handshake["ciphertextMac"])): 140 raise ValueError("neplatné MAC") 141 plaintext = ctr( 142 GOST3412Kuznechik(key_enc).encrypt, 143 KUZNECHIK_BLOCKSIZE, 144 ciphertext, 145 8 * b"x00", 146 ) 147 pokus: 148 tbe, _ = HandshakeTBE().decode(plaintext) 149 kromě ASN1Error: 150 raise ValueError("nelze dekódovat TBE") 151 key_sign_pub_hash = bytes(tbe["identity"]) 152 peer = KEYS.get(key_sign_pub_hash) 153 pokud je peer None: 154 raise ValueError("neznámá identita") 155 mac_tag = mac( 156 GOST3412Kuznechik(key_mac_identity).encrypt, 157 KUZNECHIK_BLOCKSIZE, 158 key_sign_pub_hash, 159 ) 160 pokud neporovnáte_digest(mac_tag, bytes(tbe["identityMac"])): 161 raise ValueError("neplatné identitní MAC") 162 tbs = HandshakeTBS(( 163 ("cookieTheir", cookie_their), 164 ("cookieOur", cookie_our), 165 ("pubKeyOur", pub_key_our), 166 )) 167 pokud neplatí gost3410.verify( 168 CURVE, 169 peer["pub"], 170 GOST34112012256(tbs.encode()).digest(), 171 bytes(tbe["signature"]), 172 ): 173 raise ValueError("neplatný podpis") 174 return peer["name"]Jak již bylo zmíněno výše, 34.13-2015 popisuje různé z 34.12-2015. Mezi nimi je režim generování HMAC, výpočtu MAC. V PyGOST to je gost3413.mac(). Tento režim vyžaduje přenos funkce šifrování (přijímající a vracející jeden blok dat), velikosti šifrovacího bloku a vlastně samotných dat. Proč nelze hardcodovat velikost šifrovacího bloku? 34.12-2015 popisuje nejen 128-bitový šifrovač Kuznechik, ale také 64-bitový — mírně upravený GOST 28147-89, vytvořený ještě v KGB a stále má jeden z nejvyšších bezpečnostních prahů.
Kuznečík se inicializuje pomocí gost.3412.GOST3412Kuznechik(key) a vrací objekt s metodami .encrypt() a .decrypt(), které jsou vhodné pro přenos do funkce 34.13. MAC se vypočítává následujícím způsobem: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Pro porovnání vypočteného a přijatého MACu nelze použít běžné porovnání (==) bajtových řetězců, protože tato operace způsobuje úniky časového porovnání, což může za určitých okolností vést k fatálním zranitelnostem typu útoky na TLS. V Pythonu existuje speciální funkce hmac.compare_digest pro toto.
Funkce blokového šifru může zašifrovat pouze jeden blok dat. Pro větší množství, a navíc ne násobné délky, je nutné použít režim šifrování. V normě 34.13-2015 jsou popsány následující: ECB, CTR, OFB, CBC, CFB. Každý z nich má své povolené oblasti použití a charakteristiky. K obrovskému politování, stále nemáme standardizované (jako CCM, OCB, GCM a podobné) — musíme sami alespoň přidat MAC. Já volím (CTR): nevyžaduje doplnění na velikost bloku, může být paralelizován, používá pouze funkci šifrování, může být bezpečně použit pro šifrování velkého množství zpráv (na rozdíl od CBC, kde relativně rychle začínají kolize).
Stejně jako .mac(), .ctr() přijímá podobná data jako vstup: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). Je třeba zadat inicializační vektor, jehož délka je přesně poloviční velikosti šifrovacího bloku. Pokud náš šifrovací klíč se používá pouze pro šifrování jedné zprávy (i když se skládá z několika bloků), je bezpečné nastavit nulový inicializační vektor. Pro šifrování handshake zpráv používáme pokaždé jiný klíč.
Ověření podpisu gost3410.verify() je triviální: předáme eliptickou křivku, v jejímž rámci pracujeme (tu jednoduše fixujeme v našem protokolu GOSTIM), veřejný klíč podepisovatele (nezapomínejte, že to musí být n-tice dvou velkých čísel, nikoli bajtový řetězec), 34.11-2012 hash a samotný přijatý podpis.
Dále v iniciátorovi připravujeme a odesíláme handshake2 zprávu handshake, provádějící stejné kroky, jaké jsme dělali při ověřování, pouze symetricky: podpis na svých klíčích místo ověření, atd...
456 # Připravte a odešlete Handshake 2 zprávu {{{ 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)Když je relace navázána, generují se transportní klíče (jednoduchý klíč pro šifrování, pro autentizaci, pro každou stranu), a inicializuje se Kuznechik pro dešifrování a kontrolu MAC:
499 # Spusťte odesílatele textových zpráv, inicializujte transportní dekodér {{{ 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 # Čekání na testovací zprávy {{{ 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 je coroutine, která nyní šifruje zprávy před jejich odesláním do TCP připojení. Každá zpráva má monotónně vzrůstající nonce, která je také vektorem inicializace při šifrování v režimu počítadla. Každá zpráva i blok zprávy mají zaručeně odlišné hodnoty počítadla.
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 += 1Příchozí zprávy zpracovává korutina msg_receiver, která se zabývá autentizací a dešifrováním:
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("neočekávaná hodnota nonce") mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode()) if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])): raise ValueError("neplatný 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ávěr
GOSTIM by měl být používán výhradně pro vzdělávací účely (protože není pokryt testy, minimálně)! Zdrojový kód programu si můžete stáhnout (Stribog-256 hash: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Stejně jako všechny moje projekty, typ , , , , GOSTIM je zcela , distribuovaný pod podmínkami .
, , člen , Python/Go vývojář, hlavní odborník .
Zdroj: habr.com
