Będąc programistą biblioteki (kryptograficzne prymitywy zgodne z GOST w czystym Pythonie), często otrzymuję pytania, jak na szybko zrealizować najprostszy bezpieczny wymianę wiadomości. Wielu uważa, że kryptografia aplikacyjna jest dość prostą sprawą i wywołanie .encrypt() w przypadku szyfru blokowego wystarczy do bezpiecznego przesyłania przez kanał komunikacyjny. Inni z kolei sądzą, że kryptografia aplikacyjna to domena nielicznych i akceptowalne jest to, że bogate firmy takie jak Telegram z matematykami-olimpijczykami bezpiecznego protokołu.
To wszystko skłoniło mnie do napisania tego artykułu, aby pokazać, że realizacja protokołów kryptograficznych i bezpiecznego IM nie jest tak skomplikowanym zadaniem. Jednak nie warto wynajdować własnych protokołów uwierzytelniania i wymiany kluczy.
W artykule zostanie opisany , , instant messenger z uwierzytelniania i wymiany kluczy (na podstawie którego zaimplementowano ), używając wyłącznie kryptograficznych algorytmów GOST biblioteki PyGOST oraz kodowania ASN.1 przy pomocy biblioteki (o której już ). Warunkiem koniecznym jest, aby był na tyle prosty, aby można go było napisać od podstaw w jedną noc (lub dzień roboczy), w przeciwnym razie to już nie jest prosta aplikacja. Z pewnością zawiera błędy, nadmierne złożoności, niedociągnięcia, ponadto jest to moja pierwsza aplikacja z wykorzystaniem biblioteki asyncio.
Projekt IM
Na początek musimy zrozumieć, jak będzie wyglądać nasz IM. Dla uproszczenia, niech to będzie sieć peer-to-peer, bez wykrywania uczestników. Osobiście będziemy podawać adres: port, do którego będziemy się łączyć, aby rozmawiać z rozmówcą.
Rozumiem, że w tej chwili przypuszczenie o dostępności bezpośredniego połączenia między dwoma dowolnymi komputerami jest poważnym ograniczeniem w praktyce IM. Ale im więcej programistów zaimplementuje różne obejścia NAT-traversal, tym dłużej będziemy trwać w Internecie IPv4, z niepokojącą prawdopodobieństwem połączenia między dowolnymi komputerami. Ileż można znosić brak IPv6 w domu i w pracy?
Będziemy mieć sieć friend-to-friend: wszyscy możliwi rozmówcy muszą być wcześniej znani. Po pierwsze, to znacznie upraszcza: przedstawiliśmy się, znaleźliśmy lub nie znaleźliśmy imienia/klucza, rozłączyliśmy się lub kontynuujemy pracę, znając rozmówcę. Po drugie, w ogólnym przypadku jest to bezpieczne i wyklucza wiele ataków.
Interfejs IM będzie zbliżony do klasycznych rozwiązań , które bardzo mi się podobają ze względu na minimalizm i filozofię Unix-way. Program IM dla każdego rozmówcy tworzy katalog z trzema gniazdami Unix domain socket:
- in — zapisywane są w nim wysyłane rozmówcy wiadomości;
- out — z niego czytane są wiadomości odebrane od rozmówcy;
- state — czytając z niego, dowiadujemy się, czy rozmówca jest obecnie połączony, adres/port połączenia.
Ponadto tworzony jest gniazdo conn, zapisując w nim host i port, inicjujemy połączenie zdalnym rozmówcą.
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Takie podejście pozwala na niezależne implementacje transportu IM i interfejsu użytkownika, bo jak mówi przysłowie, o gustach się nie dyskutuje, nie każdemu można dogodzić. Używając i/lub , można uzyskać wieloekranowy interfejs z podświetlaniem składni. A dzięki , można uzyskać linię do wprowadzania wiadomości kompatybilną z GNU Readline.
W rzeczywistości, projekty suckless wykorzystują pliki FIFO. Osobiście nie mogłem zrozumieć, jak w asyncio pracować z plikami współbieżnie bez ręcznego podkładu z przydzielonych wątków (od dawna do takich rzeczy używam języka ). Dlatego postanowiłem ograniczyć się do gniazd Unix domain. Niestety, to pozbawia możliwości wykonania echo 2001:470:dead::babe 6666 > conn. Rozwiązałem ten problem, używając : echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
Pierwotny niebezpieczny protokół
Jako transport używany jest TCP: gwarantuje dostarczenie i jego porządek. UDP nie gwarantuje ani tego, ani tamtego (co byłoby przydatne, gdy zastosowana zostanie kryptografia), a wsparcia w Pythonie nie ma z pudełka.
Niestety, TCP nie ma pojęcia wiadomości, a jedynie strumienia bajtów. Dlatego musimy wymyślić format do wiadomości, aby można je było oddzielać w tym strumieniu. Na początek możemy ustalić użycie znaku nowej linii. Na razie wystarczy, jednak gdy zaczniemy szyfrować nasze wiadomości, ten znak może pojawić się gdziekolwiek w szyfrogramie. Dlatego w sieciach popularne są protokoły, które najpierw wysyłają długość wiadomości w bajtach. Na przykład w Pythonie na pokładzie jest xdrlib, która pozwala pracować z takim formatem. .
Nie będziemy poprawnie i efektywnie pracować z czytaniem TCP — uprościmy kod. Odczytujemy dane z gniazda w nieskończonej pętli, aż odkodujemy pełną wiadomość. Możemy użyć JSON lub XML jako formatu dla takiego podejścia. Jednak gdy dodamy kryptografię, dane będą musiały być podpisane i uwierzytelnione — a to wymaga identycznej bajt-w-bajt reprezentacji obiektów, co nie jest zapewniane przez JSON/XML (wynik z dumps może się różnić).
XDR nadaje się do tego zadania, jednak wybieram ASN.1 z kodowaniem DER oraz bibliotekę, ponieważ będziemy mieć do czynienia z obiektami wysokiego poziomu, z którymi często przyjemniej i wygodniej się pracuje. W przeciwieństwie do schemaless , lub , ASN.1 automatycznie sprawdzi dane w odniesieniu do ściśle zdefiniowanej schemy.
# 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))),
))
Odebranym komunikatem będzie Msg: albo tekstowe MsgText (na razie z jednym polem tekstowym), albo komunikat ręcznego uścisku MsgHandshake (w którym przesyłane jest imię rozmówcy). Obecnie wydaje się to przesadnie skomplikowane, ale to inwestycja na przyszłość.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│MsgHandshake(IdA) │
│─────────────────>│
│ │
│MsgHandshake(IdB) │
││
│ │
│ MsgText() │
│<─────────────────│
│ │
IM bez kryptografii
Jak już wspomniałem, do wszystkich operacji z gniazdami będzie używana biblioteka asyncio. Ogłosimy, czego oczekujemy w momencie uruchomienia:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Nasza nazwa peer",
)
parser.add_argument(
"--their-names",
required=True,
help="Ich nazwy peer, oddzielone przecinkami",
)
parser.add_argument(
"--bind",
default="::1",
help="Adres, na którym nasłuchuje",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Port, na którym nasłuchuje",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
Ustalana jest własna nazwa (—our-name alice). Wymieniane są wszystkie oczekiwane rozmówcy (—their-names bob,eve), oddzielone przecinkami. Dla każdego z rozmówców tworzony jest katalog z gniazdami Unix oraz coroutine dla każdego 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"))
Wiadomości przychodzące od użytkownika z gniazda in są wysyłane do kolejki 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"))
Wiadomości przychodzące od rozmówców są wysyłane do kolejek OUT_QUEUES, z których dane są zapisywane w gnieździe 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()
Podczas odczytu z gniazda state program szuka w słowniku PEER_ALIVE adresu rozmówcy. Jeśli połączenia z rozmówcą jeszcze nie ma, zostaje zapisana pusta linia.
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()
Podczas zapisu adresu w gnieździe conn uruchamiana jest funkcja „inicjatora” połączenia:
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))
Rozważmy inicjatora. Najpierw otwiera on, oczywiście, połączenie z podanym hostem/portem i wysyła wiadomość handshake ze swoją nazwą:
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 message {{{
135 writer.write(Msg(("handshake", MsgHandshake((
136 ("peerName", OUR_NAME),
137 )))).encode())
138 # }}}
139 await writer.drain()
Następnie czeka na odpowiedź zdalnej strony. Próbuje zdekodować otrzymaną odpowiedź zgodnie z schematem Msg ASN.1. Zakładamy, że cała wiadomość zostanie wysłana w jednym segmencie TCP i otrzymamy ją atomowo podczas wywołania .read(). Sprawdzamy, czy otrzymaliśmy dokładnie wiadomość handshake.
141 # Wait for Handshake message {{{
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 # }}}
Sprawdzamy, czy przyjęta nazwa rozmówcy jest nam znana. Jeśli nie, to przerywamy połączenie. Sprawdzamy, czy już mieliśmy nawiązane połączenie z nim (rozmówca ponownie próbował się z nami połączyć) i zamykamy je. W kolejce IN_QUEUES umieszczane są ciągi Python z tekstem wiadomości, jednak istnieje specjalna wartość None, która sygnalizuje, by korutyna msg_sender zakończyła działanie, aby zapomniała o swoim writer związanym z przestarzałym połączeniem TCP.
159 msg_handshake = msg.value
160 peer_name = str(msg_handshake["peerName"])
161 if peer_name not in THEIR_NAMES:
162 logging.warning("unknown peer name: %s", peer_name)
163 writer.close()
164 return
165 logging.info("%s: session established: %s", _id, peer_name)
166 # Run text message sender, initialize transport decoder {{{
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 przyjmuje wychodzące wiadomości (wrzucane do kolejki z gniazda in), serializuje je w wiadomości MsgText i wysyła przez połączenie TCP. Może ono przerwać się w dowolnym momencie — to chcemy wyraźnie przechwycić.
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))
Na końcu inicjator wchodzi w nieskończoną pętlę odczytu wiadomości z gniazda. Sprawdza, czy są to wiadomości tekstowe i umieszcza je w kolejce OUT_QUEUES, skąd będą wysyłane do gniazda out odpowiedniego rozmówcy. Dlaczego nie można po prostu robić .read() i dekodować wiadomość? Ponieważ nie można wykluczyć sytuacji, w której kilka wiadomości od użytkownika zostanie zgromadzonych w buforze systemu operacyjnego i wysłanych w jednym segmencie TCP. Dekodować będziemy mogli pierwszą, a w buforze może pozostać część z kolejnej. W przypadku jakiejkolwiek nieprzewidzianej sytuacji zamykamy połączenie TCP i zatrzymujemy korutynę msg_sender (wysyłając None do kolejki OUT_QUEUES).
174 buf = b""
175 # Czekaj na wiadomości testowe {{{
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: przekroczono maksymalny rozmiar bufora", _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: nieoczekiwana wiadomość %s", _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: rozłączanie: %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: odebrano wiadomość o długości %d znaków", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Wróćmy do głównego kodu. Po utworzeniu wszystkich korutyn w momencie uruchamiania programu uruchamiamy serwer TCP. Na każde nawiązane połączenie tworzy korutinę responder (odpowiadający).
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("Nasłuchiwanie na: %s", server.sockets[0].getsockname())
loop.run_forever()
responder jest podobny do initiator i wykonuje wszystkie te same działania, ale nieskończona pętla odczytu wiadomości uruchamia się natychmiast, dla uproszczenia. Obecnie protokół handshake wysyła po jednej wiadomości z każdej strony, ale w przyszłości będzie wysyłane po dwie od inicjatora połączenia, po których natychmiast możliwe będzie wysyłanie wiadomości tekstowych.
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()
Bezpieczny protokół
Nadeszła pora, aby zabezpieczyć naszą komunikację. Co mamy na myśli przez bezpieczeństwo i czego się spodziewamy:
- prywatność przesyłanych wiadomości;
- autentyczność i integralność przesyłanych wiadomości — ich zmiana powinna być wykryta;
- ochrona przed atakami powtórzeniowymi (replay attack) — fakt zniknięcia lub powtórzenia wiadomości powinien być wykryty (i decydujemy o zerwaniu połączenia);
- identyfikacja i autoryzacja rozmówców na podstawie wcześniej ustalonych kluczy publicznych — wcześniej zdecydowaliśmy, że tworzymy sieć friend-to-friend. Tylko po autoryzacji dowiemy się, z kim rozmawiamy;
- obecność właściwości (PFS) — kompromitacja naszego długoterminowego klucza podpisu nie powinna prowadzić do możliwości odczytania całej wcześniejszej korespondencji. Nagranie przechwyconego ruchu staje się bezużyteczne;
- ważność wiadomości (transportowych i handshake) tylko w obrębie jednej sesji TCP. Wstawienie poprawnie podpisanych/uwierzytelnionych wiadomości z innej sesji (nawet z tym samym rozmówcą) nie powinno być możliwe;
- pasywny obserwator nie powinien widzieć ani identyfikatorów użytkowników, ani przekazywanych długoterminowych kluczy publicznych, ani ich skrótów. Pewna anonimowość od pasywnego obserwatora.
Zadziwiające, ale ten minimalny zestaw funkcji prawie wszyscy chcą mieć w każdym protokole handshake, a bardzo mało z wymienionych wymagań jest ostatecznie spełnionych przez „domowe” protokoły. Dlatego nie będziemy teraz wynajdować niczego nowego. Zdecydowanie poleciłbym użyć do budowy protokołów, ale wybierzemy coś prostszego.
Najbardziej popularne są dwa protokoły:
- — najtrudniejszy protokół z długą historią błędów, niedociągnięć, luk, słabej przemyślności, złożoności i niedociągnięć (jednakże w niewielkim stopniu dotyczy to TLS 1.3). Nie rozważamy go z powodu nadmiernej złożoności.
- z — nie mają poważnych problemów kryptograficznych, chociaż też nie są proste. Jeśli poczytać o IKEv1 i IKEv2, ich źródłem są , ISO/IEC IS 9798-3 oraz protokoły SIGMA (SIGn-and-MAc) — wystarczająco proste do implementacji w jeden wieczór.
Czym SIGMA, jako ostatnie ogniwo rozwoju protokołów STS/ISO, jest dobre? Spełnia wszystkie nasze wymagania (w tym „ukrywania” identyfikatorów rozmówców), nie ma znanych problemów kryptograficznych. Jest minimalistyczny — usunięcie choćby jednego elementu z wiadomości protokołu spowoduje jego niebezpieczeństwo.
Przejdźmy od najprostszych domowych protokołów do SIGMA. Najbardziej podstawową interesującą nas operacją jest : funkcja, na wyjściu której obaj uczestnicy uzyskają tę samą wartość, którą można będzie wykorzystać jako klucz symetryczny. Nie wnikając w szczegóły: każda ze stron generuje efemeryczną (używaną tylko w obrębie jednej sesji) parę kluczy (klucz publiczny i prywatny), wymieniają się kluczami publicznymi, wywołują funkcję ustalania, do której przekazują swój klucz prywatny i klucz publiczny rozmówcy.
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│ IdA, PubA │ ╔════════════════════╗
│───────────────>│ ║PrvA, PubA = DHgen()║
│ │ ╚════════════════════╝
│ IdB, PubB │ ╔════════════════════╗
│<───────────────│ ║PrvB, PubB = DHgen()║
│ │ ╚════════════════════╝
────┐ ╔═══════╧════════════╗
│ ║Key = DH(PrvA, PubB)║
<───┘ ╚═══════╤════════════╝
│ │
│ │
Każdy może wtrącić się i zamienić klucze publiczne na własne — w tym protokole nie ma uwierzytelnienia rozmówców. Dodajmy podpis przy użyciu długowiecznych kluczy.
┌─────┐ ┌─────┐
│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) ║ │
│ ╚═════════════════════╝ │
│ │
Taki podpis nie będzie odpowiedni, ponieważ nie jest powiązany z konkretną sesją. Tego typu wiadomości mogą być „odpowiednie” także dla sesji z innymi uczestnikami. Pełny kontekst powinien być podpisany. To zmusza nas również do dodania przesyłki jeszcze jednej wiadomości od A.
Ponadto krytyczne jest dodanie do podpisu własnego identyfikatora, ponieważ w przeciwnym razie możemy podmienić IdXXX i podpisać wiadomość kluczem innego znanego rozmówcy. W celu zapobieżenia , należy upewnić się, że elementy pod podpisem znajdują się w ściśle określonych miejscach zgodnie ze swoim znaczeniem: jeśli A podpisuje (PubA, PubB), to B musi podpisywać (PubB, PubA). To również mówi o znaczeniu wyboru struktury i formatu serializowanych danych. Na przykład zbiory w kodowaniu ASN.1 DER są sortowane: SET OF(PubA, PubB) będzie identyczny z 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) ║ │ │ ╚═════════════════════╝ │ │
Jednakże wciąż nie „udowodniliśmy”, że wypracowaliśmy taki sam wspólny klucz dla tej sesji. Zasadniczo można by się obejść bez tego kroku — pierwsza wiadomość transportowa byłaby nieważna, ale chcemy, aby po zakończeniu handshake'a mieć pewność, że wszystko zostało naprawdę uzgodnione. Obecnie posiadamy protokół ISO/IEC IS 9798-3.
Moglibyśmy podpisywać także sam wypracowany klucz. To niebezpieczne, ponieważ istnieje ryzyko, że w używanym algorytmie podpisu mogą wystąpić wycieki (nawet jeśli to bity-na-podpis, to jednak wycieki). Można podpisywać skrót wypracowanego klucza, ale wyciek nawet skrótu z wypracowanego klucza może mieć wartość w ataku brute-force na funkcję wytwarzania. SIGMA wykorzystuje funkcję MAC, która uwierzytelnia identyfikator nadawcy.
┌─────┐ ┌─────┐ │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 optymalizację, niektórzy mogą chcieć ponownie wykorzystać swoje efemeryczne klucze (co, oczywiście, jest szkodliwe dla PFS). Na przykład, wygenerowaliśmy parę kluczy, próbowaliśmy się połączyć, ale TCP nie było dostępne lub połączenie zostało przerwane w trakcie protokołu. Szkoda marnować entropię i zasoby procesora na nową parę. Dlatego wprowadzimy tzw. ciastko — pseudolosową wartość, która zabezpieczy przed potencjalnymi atakami typu replay podczas ponownego użycia efemerycznych kluczy publicznych. Z powodu powiązania między ciastkiem a efemerycznym kluczem publicznym, klucz publiczny przeciwnej strony można usunąć z podpisu jako zbędny.
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Na koniec chcemy uzyskać prywatność naszych identyfikatorów rozmówców przed pasywnym obserwatorem. W tym celu SIGMA proponuje najpierw wymianę kluczy efemerycznych, opracowanie wspólnego klucza, na którym zaszyfrujemy autoryzacyjne i identyfikujące wiadomości. SIGMA opisuje dwa warianty:
- SIGMA-I — chroni inicjatora przed aktywnymi atakami, a respondenta przed pasywnymi: inicjator autoryzuje respondenta i jeśli coś się nie zgadza, to nie ujawnia swojej identyfikacji. Respondent zaś ujawnia swoją identyfikację, jeśli rozpocznie aktywny protokół. Pasywny obserwator nic nie dowiaduje się;
SIGMA-R — chroni respondenta przed aktywnymi atakami, a inicjatora przed pasywnymi. Wszystko jest dokładnie odwrotnie, ale w tym protokole przekazywanych jest już cztery wiadomości uzgadniających.Wybieramy SIGMA-I jako bardziej zbliżony do tego, czego oczekujemy od tradycyjnych rozwiązań klient-serwer: klient zna tylko uwierzytelniony serwer, a serwer i tak zna wszystko. Dodatkowo jest prostszy do wdrożenia z powodu mniejszej liczby komunikatów ręcznej wymiany. Wszystko, co wnosimy do protokołu, to szyfrowanie części komunikatu i przeniesienie identyfikatora A do szyfrowanej części ostatniego komunikatu:
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
- Do podpisywania używa się GOST R algorytmu z kluczami 256-bitowymi.
- Do generowania wspólnego klucza używa się 34.10-2012 VKO.
- Jako MAC stosuje się CMAC. Technicznie jest to szczególny tryb pracy algorytmu szyfrowania blokowego, opisany w GOST R 34.13-2015. Jako funkcję szyfrowania dla tego trybu — (34.12-2015).
- Jako identyfikator rozmówcy używany jest skrót jego klucza publicznego. Skrótem jest (34.11-2012 256 bitów).
Po wykonaniu handshake'u ustalimy wspólny klucz. Możemy go używać do uwierzytelnionego szyfrowania komunikatów transportowych. Ta część jest dość prosta i trudno się w niej pomylić: inkrementujemy licznik wiadomości, szyfrujemy komunikat, uwierzytelniając (MAC) licznik i szyfrogram, wysyłamy. Podczas odbioru wiadomości sprawdzamy, czy licznik ma oczekiwaną wartość, uwierzytelniamy szyfrogram z licznikiem, deszyfrujemy. Jakim kluczem szyfrować wiadomości handshake, transportowe, oraz jak je uwierzytelniać? Używanie jednego klucza do wszystkich tych zadań jest niebezpieczne i nierozważne. Należy generować klucze, używając wyspecjalizowanych funkcji (funkcja generacji kluczy). Ponownie, nie będziemy kombinować i wymyślać czegoś nowego: jest dobrze znana, dobrze zbadana i nie ma znanych problemów. Niestety, w rodzimej bibliotece Pythona nie ma tej funkcji, więc użyjemy pakietu. HKDF wewnętrznie używa , który z kolei korzysta z funkcji skrótu. Przykład implementacji w Pythonie na stronie Wikipedii zajmuje zaledwie kilka linii kodu. Podobnie jak w przypadku 34.10-2012, jako funkcję skrótu użyjemy Stribog-256. Wynik naszej funkcji generowania kluczy będzie nazywany kluczem sesyjnym, z którego będą generowane brakujące klucze symetryczne:
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/schematy
Zobaczmy, jakie struktury ASN.1 otrzymaliśmy do przesyłania wszystkich tych danych:
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 będzie podpisywane (to be signed). HandshakeTBE – to, co będzie szyfrowane (to be encrypted). Zwracam uwagę na pole ukm w MsgHandshake1. 34.10 VKO, dla jeszcze większej losowości generowanych kluczy, obejmuje parametr UKM (user keying material) – po prostu dodatkowa entropia.
Dodanie kryptografii do kodu
Omówimy tylko wprowadzone zmiany do oryginalnego kodu, ponieważ struktura pozostała niezmieniona (tak naprawdę, najpierw stworzono ostateczną wersję, a potem usunięto całą kryptografię).
Ponieważ autoryzacja i identyfikacja rozmówców będą przeprowadzane za pomocą kluczy publicznych, teraz muszą być one przechowywane gdzieś na stałe. Dla uproszczenia użyjemy JSON w takim formacie:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }our – nasza para kluczy, szesnastkowe klucze prywatne i publiczne. their – imiona rozmówców i ich klucze publiczne. Zmienimy argumenty wiersza poleceń i dodamy postprocessowanie danych JSON:
from pygost import gost3410 from pygost.gost34112012256 import GOST34112012256 CURVE = gost3410.GOST3410Curve( *gost3410.CURVE_PARAMS["GostR3410_2001_CryptoPro_A_ParamSet"] ) parser = argparse.ArgumentParser(description="GOSTIM") parser.add_argument( "--keys-gen", action="store_true", help="Generuj JSON z naszym nowym kluczem", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON z naszymi i ich kluczami", ) parser.add_argument( "--bind", default="::1", help="Adres do nasłuchu", ) parser.add_argument( "--port", type=int, default=6666, help="Port do nasłuchu", ) 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) # Parsowanie i unmarshalowanie naszych i ich kluczy {{{ 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), } # }}}Klucz prywatny algorytmu 34.10 to losowa liczba. Ma 256-bitów dla 256-bitowych krzywych eliptycznych. PyGOST działa nie na zestawie bajtów, a na , dlatego nasz klucz prywatny (urandom(32)) musi być przekształcony w liczbę przy użyciu gost3410.prv_unmarshal(). Klucz publiczny jest wyznaczany deterministycznie z klucza prywatnego za pomocą gost3410.public_key(). Klucz publiczny 34.10 to dwie duże liczby, które również muszą być przekształcone w sekwencję bajtów dla łatwego przechowywania i przesyłania, używając gost3410.pub_marshal().
Po odczytaniu pliku JSON, klucze publiczne muszą być odpowiednio przekształcone z powrotem, używając gost3410.pub_unmarshal(). Ponieważ identyfikatory rozmówców przychodzą w postaci skrótu klucza publicznego, mogą być od razu wyliczone i umieszczone w słowniku w celu szybkiego wyszukiwania. Skrót Stribog-256 to gost34112012256.GOST34112012256(), w pełni spełniający interfejs funkcji skrótu hashlib.
Jak zmieniła się korutyna inicjatora? Wszystko, jak w schemacie rękoczynu: generujemy cookie (128-bit wystarczy), efemeryczny klucz pary 34.10, który będzie używany do funkcji VKO użycia klucza.
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 # Generuj nasz efemeryczny klucz publiczny i ciasteczko, wysyłamy wiadomość 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()- czekamy na odpowiedź i dekodujemy przychodzącą wiadomość Msg;
- upewniamy się, że otrzymaliśmy handshake1;
- dekodujemy efemeryczny klucz publiczny drugiej strony i obliczamy klucz sesyjny;
- wypracowujemy klucze symetryczne potrzebne do przetwarzania części wiadomości TBE.
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 # Walidacja wiadomości 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 to 64-bit liczba (urandom(8)), która również wymaga deserializacji z bajtowego przedstawienia, używając gost3410_vko.ukm_unmarshal(). Funkcja VKO dla 34.10-2012 256-bit to gost3410_vko.kek_34102012256() (KEK — klucz szyfrowania klucza).
Wypracowany klucz sesyjny jest już 256-bitową sekwencją bajtową o charakterze pseudolosowym. Dlatego można go od razu użyć w funkcji HKDF. Ponieważ GOST34112012256 spełnia interfejs hashlib, można go od razu użyć w klasie Hkdf. Sól (pierwszy argument Hkdf) nie jest podawana, ponieważ wypracowany klucz z powodu efemeryczności zaangażowanych par kluczy będzie różny dla każdej sesji i posiada wystarczającą entropię. kdf.expand() domyślnie już wydaje klucze o długości 256-bit, wymagane przez Kuczynka w dalszej części.
Następnie sprawdzane są części TBE i TBS przychodzącej wiadomości:
- obliczany jest i sprawdzany MAC dla przychodzącego tekstu szyfrowego;
- deszyfrowany jest tekst szyfrowy;
- dekodowana jest struktura TBE;
- z niej wyciągany jest identyfikator rozmówcy i sprawdzane jest, czy jest nam znany;
- obliczany jest i sprawdzany MAC dla tego identyfikatora;
- podpis jest weryfikowana nad strukturą TBS, która zawiera ciasteczka obu stron oraz publiczny efemeryczny klucz drugiej strony. Podpis jest weryfikowany długoterminowym kluczem podpisu rozmówcy.
441 spróbuj: 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 wyjątek ValueError jako err: 452 logging.warning("%s: %s, rozłączanie", _id, err) 453 writer.close() 454 zwróć 455 # }}} 128 def validate_tbe( 129 msg_handshake: Union[MsgHandshake1, MsgHandshake2], 130 key_mac_identity: bytes, 131 key_enc: bytes, 132 key_mac: bytes, 133 cookie_their: Cookie, 134 cookie_our: Cookie, 135 pub_key_our: PubKey, 136 ) -> str: 137 ciphertext = bytes(msg_handshake["ciphertext"]) 138 mac_tag = mac(GOST3412Kuznechik(key_mac).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext) 139 if not compare_digest(mac_tag, bytes(msg_handshake["ciphertextMac"])): 140 raise ValueError("nieprawidłowy MAC") 141 plaintext = ctr( 142 GOST3412Kuznechik(key_enc).encrypt, 143 KUZNECHIK_BLOCKSIZE, 144 ciphertext, 145 8 * b"x00", 146 ) 147 spróbuj: 148 tbe, _ = HandshakeTBE().decode(plaintext) 149 wyjątek ASN1Error: 150 raise ValueError("nie można dekodować TBE") 151 key_sign_pub_hash = bytes(tbe["identity"]) 152 peer = KEYS.get(key_sign_pub_hash) 153 if peer is None: 154 raise ValueError("nieznana tożsamość") 155 mac_tag = mac( 156 GOST3412Kuznechik(key_mac_identity).encrypt, 157 KUZNECHIK_BLOCKSIZE, 158 key_sign_pub_hash, 159 ) 160 if not compare_digest(mac_tag, bytes(tbe["identityMac"])): 161 raise ValueError("nieprawidłowy MAC tożsamości") 162 tbs = HandshakeTBS(( 163 ("cookieTheir", cookie_their), 164 ("cookieOur", cookie_our), 165 ("pubKeyOur", pub_key_our), 166 )) 167 if not gost3410.verify( 168 CURVE, 169 peer["pub"], 170 GOST34112012256(tbs.encode()).digest(), 171 bytes(tbe["signature"]), 172 ): 173 raise ValueError("nieprawidłowy podpis") 174 zwróć peer["name"]Jak już wspomniano powyżej, 34.13-2015 opisuje różne z 34.12-2015. Wśród nich jest tryb generowania imityzacji, obliczania MAC-a. W PyGOST jest to gost3413.mac(). Ten tryb wymaga przekazania funkcji szyfrowania (przyjmującej i zwracającej jeden blok danych), rozmiaru bloku szyfrującego oraz danych. Dlaczego nie można ustawiać na sztywno rozmiaru bloku szyfrującego? 34.12-2015 opisuje nie tylko 128-bitowy szyfr Kuzniek, ale także 64-bitowy — nieco zmodyfikowany GOST 28147-89, stworzony jeszcze w KGB i do dziś mający jeden z najwyższych progów bezpieczeństwa.
Kuzniecik inicjalizuje gost.3412.GOST3412Kuznechik(key) za pomocą wywołania i zwraca obiekt z metodami .encrypt()/.decrypt(), które nadają się do przekazywania w funkcjach 34.13. MAC oblicza się w następujący sposób: gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext). Do porównania wygenerowanego i otrzymanego MAC-a nie można używać zwykłego porównania (==) łańcuchów bajtów, ponieważ ta operacja powoduje wycieki czasowe, co w ogólnym przypadku może prowadzić do krytycznych podatności typu ataków na TLS. W Pythonie istnieje specjalna funkcja hmac.compare_digest do tego celu.
Funkcja szyfru blokowego może zaszyfrować tylko jeden blok danych. Aby obsłużyć więcej, a także o długości nie będącej wielokrotnością, należy użyć trybu szyfrowania. W 34.13-2015 opisane są następujące: ECB, CTR, OFB, CBC, CFB. Każdy z nich ma swoje akceptowalne obszary zastosowań i cechy. Niestety, wciąż nie mamy ustandaryzowanych (takich jak CCM, OCB, GCM i podobne) — jesteśmy zmuszeni przynajmniej do samodzielnego dodania MAC. Wybieram (CTR): nie wymaga uzupełnienia do rozmiaru bloku, może być równolegle przetwarzany, wykorzystuje tylko funkcję szyfrowania, może być bezpiecznie używany do szyfrowania dużej liczby wiadomości (w przeciwieństwie do CBC, w którym stosunkowo szybko zaczynają występować kolizje).
Tak jak .mac(), .ctr() przyjmuje podobne dane jako wejście: ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv). Wymagane jest określenie wektora inicjalizacji o długości dokładnie połowy bloku szyfrującego. Jeśli nasz klucz szyfrowania jest używany tylko do szyfrowania jednej wiadomości (nawet jeśli składa się z kilku bloków), można bezpiecznie ustawić zerowy wektor inicjalizacji. Do szyfrowania wiadomości handshake używamy za każdym razem osobnego klucza.
Weryfikacja podpisu gost3410.verify() jest trywialna: przekazujemy krzywą eliptyczną, w obrębie której działamy (po prostu ją ustalamy w naszym protokole GOSTIM), publiczny klucz podpisującego (nie zapominajmy, że powinien to być krotka dwóch dużych liczb, a nie łańcuch bajtów), hasz 34.11-2012 i sam przychodzący podpis.
Następnie w inicjatorze przygotowujemy i wysyłamy wiadomość handshake2, przeprowadzając te same czynności, co podczas weryfikacji, tylko symetrycznie: podpis na swoich kluczach zamiast weryfikacji, itd...
456 # Przygotuj i wyślij wiadomość Handshake 2 {{{ 457 tbs = HandshakeTBS(( 458 ("cookieTheir", cookie_their), 459 ("cookieOur", cookie_our), 460 ("pubKeyOur", pub_our_raw), 461 )) 462 signature = gost3410.sign( 463 CURVE, 464 KEY_OUR_SIGN_PRV, 465 GOST34112012256(tbs.encode()).digest(), 466 ) 467 key_handshake2_mac_identity = kdf.expand(b"handshake2-mac-identity") 468 mac_tag = mac( 469 GOST3412Kuznechik(key_handshake2_mac_identity).encrypt, 470 KUZNECHIK_BLOCKSIZE, 471 bytes(KEY_OUR_SIGN_PUB_HASH), 472 ) 473 tbe = HandshakeTBE(( 474 ("identity", KEY_OUR_SIGN_PUB_HASH), 475 ("signature", OctetString(signature)), 476 ("identityMac", MAC(mac_tag)), 477 )) 478 tbe_raw = tbe.encode() 479 key_handshake2_enc = kdf.expand(b"handshake2-enc") 480 key_handshake2_mac = kdf.expand(b"handshake2-mac") 481 ciphertext = ctr( 482 GOST3412Kuznechik(key_handshake2_enc).encrypt, 483 KUZNECHIK_BLOCKSIZE, 484 tbe_raw, 485 8 * b"x00", 486 ) 487 mac_tag = mac( 488 GOST3412Kuznechik(key_handshake2_mac).encrypt, 489 KUZNECHIK_BLOCKSIZE, 490 ciphertext, 491 ) 492 writer.write(Msg(("handshake2", MsgHandshake2(( 493 ("ciphertext", OctetString(ciphertext)), 494 ("ciphertextMac", MAC(mac_tag)), 495 )))).encode()) 496 # }}} 497 await writer.drain() 498 logging.info("%s: sesja nawiązana: %s", _id, peer_name)Gdy sesja zostanie nawiązana, generowane są klucze transportowe (osobny klucz do szyfrowania, do uwierzytelniania dla każdej ze stron), a Kucznik jest inicjalizowany do deszyfrowania i weryfikacji MAC-a:
499 # Uruchom wysyłacza wiadomości tekstowych, zainicjalizuj dekoder transportowy {{{ 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 # Czekaj na wiadomości testowe {{{ 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 # }}}Coroutine msg_sender teraz szyfruje wiadomości przed wysłaniem ich w połączeniu TCP. Każda wiadomość ma monotonicznie rosnący nonce, który jest również wektorem inicjalizacji do szyfrowania w trybie licznika. Każda wiadomość i blok wiadomości gwarantują, że będą miały różne wartości licznika.
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 += 1Przychodzące wiadomości są przetwarzane przez korutynę msg_receiver, która zajmuje się uwierzytelnianiem i deszyfrowaniem:
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("nieoczekiwana wartość nonce") mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode()) if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])): raise ValueError("nieprawidłowy 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)Podsumowanie
GOSTIM jest przeznaczony wyłącznie do celów edukacyjnych (ponieważ nie ma pokrycia testami, przynajmniej)! Kod źródłowy programu można pobrać (hash Stryboga-256: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Jak wszystkie moje projekty, typu , , , , GOSTIM jest całkowicie , rozpowszechnianym na warunkach .
, , członek , programista Python/Go, główny specjalista .
Źródło: habr.com
