Entwéckler ze sinn Bibliothéiken (GOST cryptographic primitives am pure Python), Ech kréien dacks Froen iwwert wéi déi einfachst sécher Messagerie op de Knéi ëmsetzen. Vill Leit betruechten applizéiert Kryptografie als ganz einfach, a ruffen .encrypt () op engem Block Chiffer ass genuch fir se sécher iwwer e Kommunikatiounskanal ze schécken. Anerer gleewen datt applizéiert Kryptografie d'Schicksal vun de puer ass, an et ass akzeptabel datt räich Firmen wéi Telegram mat Olympiaden-Mathematiker sécher Protokoll.
All dëst huet mech gefrot dësen Artikel ze schreiwen fir ze weisen datt d'Ëmsetzung vun kryptografesche Protokoller a séchere IM net sou eng schwiereg Aufgab ass. Wéi och ëmmer, et ass net derwäert Är eege Authentifikatioun a Schlësselvertragsprotokoller ze erfannen.
Den Artikel wäert schreiwen , , Instant Messenger mat Authentifikatioun a Schlëssel Accord Protokoll (op Basis vun deem et ëmgesat gëtt ), benotzt exklusiv GOST kryptografesch Algorithmen PyGOST Bibliothéik an ASN.1 Message Kodéierungsbibliothéik (iwwer deem ech scho ). Eng Viraussetzung: et muss sou einfach sinn, datt et an engem Owend (oder Aarbechtsdag) vun Null geschriwwe ka ginn, soss ass et keen einfache Programm méi. Et huet méiglecherweis Feeler, onnéideg Komplikatiounen, Mängel, plus dëst ass mäin éischte Programm mat der Asyncio-Bibliothéik.
IM Design
Als éischt musse mir verstoen wat eis IM ausgesäit. Fir Einfachheet, loosst et e Peer-to-Peer Netzwierk sinn, ouni Entdeckung vun de Participanten. Mir wäerte perséinlech uginn wéi eng Adress: Port fir ze verbannen fir mat dem Gespréichspartner ze kommunizéieren.
Ech verstinn datt, zu dësem Zäitpunkt, d'Annahme datt direkt Kommunikatioun tëscht zwee arbiträr Computeren verfügbar ass eng bedeitend Limitatioun fir d'Uwendbarkeet vum IM an der Praxis. Awer méi Entwéckler implementéieren all Zorte vun NAT-Traversal Krutchen, dest méi laang wäerte mir um IPv4 Internet bleiwen, mat enger depriméierender Wahrscheinlechkeet vun der Kommunikatioun tëscht arbiträre Computeren. Wéi laang kënnt Dir de Mangel un IPv6 doheem an op der Aarbecht toleréieren?
Mir wäerten e Frënd-zu-Frënd Netzwierk hunn: all méiglech Gespréichspartner mussen am Viraus bekannt sinn. Als éischt vereinfacht dat alles immens: Mir hunn eis virgestallt, den Numm/Schlëssel fonnt oder net fonnt, ofgekoppelt oder weider schaffen, de Gespréichspartner kennen. Zweetens, am Allgemengen, ass et sécher an eliminéiert vill Attacken.
D'IM Interface wäert no bei klassesche Léisungen sinn , déi ech wierklech gär fir hir Minimalismus an Unix-Wee Philosophie. Den IM Programm erstellt e Verzeichnis mat dräi Unix Domain Sockets fir all Gespréichspartner:
- an-Messagen un d'Interlocutor geschéckt ginn an et opgeholl;
- eraus - Messagen, déi vum Gespréichspartner kritt ginn, ginn dovun gelies;
- Staat - andeems mir dovunner liesen, fanne mir eraus ob de Gespréichspartner am Moment ugeschloss ass, d'Verbindungsadress / Hafen.
Zousätzlech gëtt e Conn Socket erstallt, andeems Dir den Hostport schreift, an deen mir eng Verbindung mam Remote-Interlocutor initiéieren.
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Dës Approche erlaabt Iech onofhängeg Implementatioune vun IM Transport an User Interface ze maachen, well et kee Frënd ass, Dir kënnt net jiddereen weg. Benotzt an / oder , Dir kënnt e Multi-Fënster Interface mat Syntax Highlight kréien. A mat der Hëllef Dir kënnt eng GNU Readline-kompatibel Message Input Linn kréien.
Tatsächlech benotzen suckless Projeten FIFO Dateien. Perséinlech konnt ech net verstoen wéi ech mat Dateien kompetitiv an Asyncio schaffen ouni en handgeschriwwenen Hannergrond aus engagéierten Threads (ech hunn d'Sprooch fir sou Saachen fir eng laang Zäit benotzt ). Dofir hunn ech decidéiert mat Unix Domain Sockets ze maachen. Leider mécht dëst et onméiglech fir echo 2001:470:dead::babe 6666 > conn. Ech geléist dëse Problem benotzt : echo 2001:470: dout:: Puppelchen 6666 | socat - UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
D'Original onsécher Protokoll
TCP gëtt als Transport benotzt: et garantéiert d'Liwwerung a seng Bestellung. UDP garantéiert weder (wat nëtzlech wier wann d'Kryptographie benotzt gëtt), mee Ënnerstëtzung Python kënnt net aus der Këscht.
Leider gëtt et am TCP kee Konzept vun engem Message, nëmmen e Stroum vu Bytes. Dofir ass et néideg mat engem Format fir Messagen ze kommen, fir datt se an dësem Thread ënnerenee gedeelt kënne ginn. Mir kënnen averstanen der Linn fidderen Charakter ze benotzen. Et ass gutt fir Ufänger, awer wa mir ufänken eis Messagen ze verschlësselen, kann dëse Charakter iwwerall am Chiffertext optrieden. An Netzwierker sinn dofir populär Protokoller déi, déi als éischt d'Längt vum Message a Bytes schécken. Zum Beispill, aus der Këscht Python huet xdrlib, wat Iech erlaabt mat engem ähnlechen Format ze schaffen .
Mir schaffen net korrekt an effizient mat TCP Liesen - mir vereinfachen de Code. Mir liesen Daten aus der Socket an enger endloser Loop bis mir de komplette Message decodéieren. JSON mat XML kann och als Format fir dës Approche benotzt ginn. Awer wann d'Kryptographie bäigefüügt gëtt, mussen d'Donnéeën ënnerschriwwen an authentifizéiert ginn - an dëst erfuerdert eng Byte-fir-Byte identesch Representatioun vun Objeten, déi JSON / XML net ubitt (Dumpresultater kënne variéieren).
XDR gëeegent fir dës Aufgab, Ee wielen ech ASN.1 mat DER Kodéierung an Bibliothéik, well mir wäerten héich-Niveau Objeten op der Hand hunn, mat deenen et dacks méi agreabel a praktesch ass ze schaffen. Géigesaz schemaless , oder , ASN.1 wäert automatesch d'Donnéeën géint eng schwéier-kodéiert Schema kontrolléieren.
# 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))),
))
De kritt Message wäert Msg sinn: entweder en Text MsgText (mat engem Textfeld fir de Moment) oder e MsgHandshake Handshake Message (deen den Numm vum Gespréichspartner enthält). Elo gesäit et iwwerkomplizéiert aus, awer dëst ass e Fundament fir d'Zukunft.
Ich A) │ ───────── ────────>│ │ │ │MsgHandshake(IdB) │ │<─────────────────── │ │ MsgText() │ │──── MsgText() │ │ │
IM ouni Kryptografie
Wéi ech scho gesot hunn, gëtt d'Asyncio-Bibliothéik fir all Socket-Operatiounen benotzt. Loosst eis matdeelen wat mir beim Start erwaarden:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Our peer name",
)
parser.add_argument(
"--their-names",
required=True,
help="Their peer names, comma-separated",
)
parser.add_argument(
"--bind",
default="::1",
help="Address to listen on",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Port to listen on",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
Setzt Ären eegene Numm (--eisem Numm Alice). All erwaart Gespréichspartner sinn opgezielt getrennt duerch Komma (-hir-Nimm bob, Eva). Fir jiddereng vun den Gespréichspartner gëtt e Verzeechnes mat Unix Sockets erstallt, souwéi eng Coroutine fir all 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"))
Messagen, déi vum Benotzer aus dem Socket kommen, ginn an d'IN_QUEUES Schlaang geschéckt:
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"))
Messagen, déi vu Gespréichspartner kommen, ginn an OUT_QUEUES Schlaangen geschéckt, aus deenen d'Donnéeën an d'Out Socket geschriwwe ginn:
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()
Wann Dir aus engem Staatssocket liest, sicht de Programm d'Adress vum Gespréichspartner am PEER_ALIVE Wierderbuch. Wann et nach keng Verbindung mam Gespréichspartner ass, da gëtt eng eidel Linn geschriwwen.
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()
Wann Dir eng Adress an e Conn Socket schreift, gëtt d'Verbindung "Initiator" Funktioun gestart:
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))
Loosst eis den Initiator betruechten. Als éischt mécht se offensichtlech eng Verbindung mam spezifizéierte Host / Hafen op a schéckt en Handshake Message mat sengem Numm:
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()
Duerno waart et op eng Äntwert vun der Remote Party. Probéiert déi erakommen Äntwert ze dekodéieren mam Msg ASN.1 Schema. Mir huelen un datt de ganze Message an engem TCP-Segment geschéckt gëtt a mir kréien et atomesch wann Dir .read () rufft. Mir kontrolléieren ob mir d'Handshake Message kritt hunn.
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 # }}}
Mir kontrolléieren ob den Numm vum Gespréichspartner un eis bekannt ass. Wann net, da briechen mir d'Verbindung. Mir kontrolléieren ob mir schonn eng Verbindung mat him etabléiert hunn (de Gespréichspartner huet nach eng Kéier de Kommando ginn fir mat eis ze verbannen) an zoumaachen. D'IN_QUEUES Schlaang hält Python Saiten mam Text vun der Noriicht, awer huet e spezielle Wäert vun None, deen d'msg_sender Coroutine signaliséiert fir opzehalen ze schaffen, sou datt et iwwer säi Schrëftsteller vergiesst, deen mat der Legacy TCP Verbindung ass.
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 akzeptéiert erausginn Messagen (an der Schlaang vun engem Socket), serialiséiert se an e MsgText Message a schéckt se iwwer eng TCP Verbindung. Et kann zu all Moment briechen - mir interceptéieren dat kloer.
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))
Um Enn geet den Initiator an eng onendlech Loop vu Liese vu Messagen aus der Socket. Iwwerpréift ob dës Messagen SMSen sinn a plazéiert se an der OUT_QUEUES Schlaang, vun där se an den Out Socket vum entspriechende Gespréichspartner geschéckt ginn. Firwat kënnt Dir net just maachen .read () an decodéieren de Message? Well et ass méiglech datt verschidde Messagen vum Benotzer am Betribssystembuffer aggregéiert ginn an an engem TCP-Segment geschéckt ginn. Mir kënnen den éischten decodéieren, an dann kann en Deel vun der nächster am Puffer bleiwen. Am Fall vun enger anormaler Situatioun schloe mir d'TCP-Verbindung zou a stoppen d'msg_sender Coroutine (andeems Dir Keen an d'OUT_QUEUES Schlaang schéckt).
174 buf = b""
175 # Wait for test messages {{{
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: max buffer size exceeded", _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: unexpected %s message", _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: disconnecting: %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: received %d characters message", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Loosst eis zréck op den Haaptcode. Nodeems Dir all d'Koroutinen erstallt an der Zäit wou de Programm ufänkt, starten mir den TCP Server. Fir all etabléiert Verbindung erstellt en Äntwert Coroutine.
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("Listening on: %s", server.sockets[0].getsockname())
loop.run_forever()
Äntwert ass ähnlech wéi Initiator a spigelt all déiselwecht Handlungen, awer déi onendlech Schleife vu Messagen liesen fänkt direkt un, fir Einfachheet. De Moment schéckt de Handshakeprotokoll eng Noriicht vun all Säit, awer an Zukunft ginn et zwee vum Verbindungsinitiator, duerno kënnen SMSen direkt geschéckt ginn.
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()
Séchert Protokoll
Et ass Zäit eis Kommunikatiounen ze sécheren. Wat menge mir mat Sécherheet a wat wëlle mir:
- Confidentialitéit vun iwwerdroen Messagen;
- Authentizitéit an Integritéit vun iwwerdroen Messagen - hir Ännerungen muss festgestallt ginn;
- Schutz géint Replay Attacken - d'Tatsaach vu fehlend oder widderholl Messagen muss festgestallt ginn (a mir décidéieren d'Verbindung opzehalen);
- Identifikatioun an Authentifikatioun vun Interlocutoren mat pre-aginnen ëffentleche Schlësselen - mir hu scho virdru beschloss datt mir e Frënd-zu-Frënd-Netzwierk maachen. Nëmmen no der Authentifikatioun wäerte mir verstoen mat wiem mir kommunizéieren;
- Disponibilitéit Eegeschaften (PFS) - d'Kompromittéiere vun eisem laangliewege Ënnerschrëftschlëssel sollt net zur Fäegkeet féieren all fréier Korrespondenz ze liesen. Enregistréiere vun ofgefaangen Verkéier gëtt nëtzlos;
- Validitéit / Validitéit vun Messagen (Transport an Handshake) nëmmen bannent engem TCP Sëtzung. Korrekt ënnerschriwwen/authentifizéiert Messagen aus enger anerer Sessioun (och mam selwechte Gespréichspartner) anzeginn soll net méiglech sinn;
- e passive Beobachter soll weder Benotzeridentifizéierer gesinn, iwwerdroe laanglieweg ëffentlech Schlësselen oder Hashes vun hinnen. Eng gewëssen Anonymitéit vun engem passiven Observateur.
Iwwerraschend wëll bal jiddereen dëse Minimum an all Handshake Protokoll hunn, a ganz wéineg vun den uewe genannte gëtt schlussendlech fir "hausgemaachte" Protokoller erfëllt. Elo wäerte mir näischt Neies erfannen. Ech géif definitiv recommandéieren ze benotzen fir Protokoller ze bauen, awer loosst eis eppes méi einfach wielen.
Déi zwee populärste Protokoller sinn:
- - e ganz komplexe Protokoll mat enger laanger Geschicht vu Bugs, Jambs, Schwachstelle, schlecht Gedanken, Komplexitéit a Mängel (awer dat huet wéineg mat TLS 1.3 ze dinn). Awer mir betruechten et net well et iwwerkomplizéiert ass.
- с - hu keng sérieux kryptographesch Problemer, obwuel se och net einfach sinn. Wann Dir iwwer IKEv1 an IKEv2 liest, dann ass hir Quell , ISO/IEC IS 9798-3 a SIGMA (SIGn-and-MAc) Protokoller - einfach genuch fir an engem Owend ëmzesetzen.
Wat ass gutt iwwer SIGMA, als de leschte Link an der Entwécklung vu STS / ISO Protokoller? Et entsprécht all eis Ufuerderungen (inklusiv "verstoppen" Interlocutor Identifizéierer) an huet keng bekannte kryptographesch Probleemer. Et ass minimalistesch - op d'mannst een Element aus dem Protokoll Message ewechzehuelen féiert zu senger Onsécherheet.
Loosst eis vum einfachsten hausgemaachte Protokoll op SIGMA goen. De stäerkste Basis Operatioun mir interesséiert sinn ass : Eng Funktioun déi béid Participanten dee selwechte Wäert erausginn, deen als symmetresche Schlëssel benotzt ka ginn. Ouni an Detailer ze goen: jidderee vun de Parteien generéiert en ephemerescht (nëmmen an enger Sessioun benotzt) Schlësselpaar (ëffentlech a privat Schlësselen), tauscht ëffentlech Schlësselen, rufft d'Accordfunktioun un, op den Input vun deem se hire private Schlëssel an de Public passéieren Schlëssel vum Gespréichspartner.
A ╔══════════ ══════════╗ │───────────────>│ │ ║ ║PrvA,(┕) ════════ ═══════════╝ │ IdB, PubB │ ╔═══════════╝ ═══╝ <───────── ──────│ ║PrvB, PubB = DHgen()║ │ │ ╚══════════════════ ───┐ ╔════ ═══╧════════════╗ │ ║Key = DH(PrvA, PubB)║ <───┘═════════ ═══════ ════╝ │ │ │ │
Jiddereen kann an der Mëtt sprangen an ëffentlech Schlësselen mat hiren eegenen ersetzen - et gëtt keng Authentifikatioun vu Gespréichspartner an dësem Protokoll. Loosst eis eng Ënnerschrëft mat laangliewege Schlësselen addéieren.
XNUMX ignPrvA, (PubA)) │ ╔═ │──────────── ────────── ───────────── PubA = load()║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚═══════ ═══════ ═════════════ I, B (SignPrvB, (PubB)) │ ╔══════════════ ═══════ ════┐└ ─────────── ───────────── ──│ ║SignPrvB, SignPubB = load( )║ │ │ ║PrvB, │ ║PrvB, PubB = DH ═════════ ══════════════ ══╝ ────┐ ╔ ════════════ ═════╗ │ │ ║verifizéieren( SignPubB, ...)║ │ <───┘ ║Key = DH(Pr vA, PubB) ║ │ │ ╚═══════════════════════ ═╝ │ │ │
Sou eng Ënnerschrëft funktionnéiert net, well se net un eng spezifesch Sessioun gebonnen ass. Esou Messagen sinn och "gëeegent" fir Sessiounen mat anere Participanten. De ganze Kontext muss abonnéieren. Dëst forcéiert eis och en anere Message vum A.
Zousätzlech ass et kritesch Ären eegenen Identifizéierer ënner der Ënnerschrëft ze addéieren, well soss kënne mir IdXXX ersetzen an de Message nei ënnerschreiwen mam Schlëssel vun engem anere bekannte Gespréichspartner. Ze verhënneren , et ass néideg datt d'Elementer ënner der Ënnerschrëft op kloer definéiert Plazen no hirer Bedeitung sinn: wann A Zeechen (PubA, PubB), da muss B ënnerschreiwen (PubB, PubA). Dëst schwätzt och iwwer d'Wichtegkeet vun der Auswiel vun der Struktur an dem Format vun serialiséierten Donnéeën. Zum Beispill, Sets am ASN.1 DER Kodéierung sinn zortéiert: SET OF (PubA, PubB) wäert identesch mat SET OF (PubB, PubA) sinn.
A ╔══════════ ═════════════════╗ │──────────────────── ────────── ─────────────>│ ║SignPrvA, SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ║┕┕┕┕ ═ ═══════ ═══════════════╝ │IdB, PubB, sign(SignPrvB, (IdB, PubA, PubB)) │ ╔════════ ═════ ════════════╗ │<───────────────────────── ────────── ─────────│ ║SignPrvB, SignPubB = load()║ │ │ ║PrvB, PubB = DHgen() ║ │ ════════ ═ ════════ ══════════╝ │ Zeechen(SignPrvA, (IdA, PubB, PubA)) │ ╔═════════════════ ════╗ │─ ──────────────────────────────────────── ───>│ ║verify(SignPubB, ...) ║ │ │ ║key = dh (prva, PUBB) ║ │ │ │
Wéi och ëmmer, mir hunn nach ëmmer net "bewisen" datt mir dee selwechte gemeinsame Schlëssel fir dës Sessioun generéiert hunn. Prinzipiell kënne mir ouni dëse Schrëtt verzichten - déi éischt Transportverbindung wäert ongëlteg sinn, awer mir wëllen datt wann d'Handschlag fäerdeg ass, mir sécher sinn datt alles wierklech ofgeschloss ass. Am Moment hu mir den ISO/IEC IS 9798-3 Protokoll op der Hand.
Mir kënnen de generéierte Schlëssel selwer ënnerschreiwen. Dëst ass geféierlech, well et méiglech ass datt et Leckage am benotzte Signaturalgorithmus ka ginn (wann och Bits-per-Ënnerschrëft, awer nach ëmmer Leckage). Et ass méiglech en Hash vum Derivatiounsschlëssel z'ënnerschreiwen, awer och den Hash vum derivéierten Schlëssel ze lecken ka wäertvoll sinn an engem brute-force Attack op der Derivatiounsfunktioun. SIGMA benotzt eng MAC Funktioun déi d'Sender ID authentifizéiert.
A ╔══════════ ═════════════════╗ │──────────────────── ────────── XNUMX ╚ ═══════ ════════════════════╝ │IdB, PubB, sign(SignPrvB, (PubA,┕═══════╝) ═══ │<───────────────── ──────────────────── ────────── ─│ ║SignPrvB, SignPubB = load()║ │ │ ║PrvB, PubB = DHgen() ║ │ │ ╚════ ═══════════ ═════════ XNUMX │ ║Schlëssel = DH( PrvA, PubB) ║ │───────────────────── ──────────────── ───────── ─────>│ ║verify(Key, IdB) ║ │ │ ║verify(SignPubB, ...)║ │ │ ╚══════════════ ═════ ═╝ │ │
Als Optimisatioun kënnen e puer wëllen hir ephemeral Schlësselen nei benotzen (wat natierlech leider fir PFS ass). Zum Beispill hu mir e Schlësselpaar generéiert, probéiert ze verbannen, awer TCP war net verfügbar oder gouf iergendwou an der Mëtt vum Protokoll ënnerbrach. Et ass schued fir verschwenden Entropie a Prozessorressourcen op en neit Pair ze verschwenden. Dofir wäerte mir de sougenannte Cookie aféieren - e pseudo-zoufälleg Wäert, deen géint méiglech zoufälleg Replay Attacke schützt wann Dir ephemeral ëffentlech Schlësselen nei benotzt. Wéinst der Bindung tëscht dem Cookie an dem ephemeralen ëffentleche Schlëssel kann den ëffentleche Schlëssel vun der Géigendeel aus der Ënnerschrëft als onnéideg ewechgeholl ginn.
Cookien │ ╔════════ ═══════════════════╗ │─────────────────── ────────── ──────────────────────────────────────── ─>│ ║SignPrvA, SignPubA = load( )║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚══════════════════════════ ══╝ │IdB, PubB, CookieB , sign(SignPrvB, (CookieA, CookieB, PubB)), MAC(IdB) │ ╔════════════════════════════ ╗ │< ──────────────────────────────────────── ────────── │ │ ╚══════ ═════════════════════╝ │ │ ╔═════════════ ═══════╗ │ Zeechen( SignPrvA, (CookieB, CookieA, PubA)), MAC(IdA) │ ║Key = DH(PrvA, PubB) ║ │──────────────────────── ── ───────────────────────────────────────── ───────>│ ║ verify(Key, IdB) ║ │ │ ║verify(SignPubB, ...)║ │ │ ╚═════════════════════❕ │ │
Schlussendlech wëlle mir d'Privatsphär vun eise Gespréichspartner vun engem passiven Observateur kréien. Fir dëst ze maachen, proposéiert SIGMA fir d'éischt ephemeresch Schlësselen auszetauschen an e gemeinsame Schlëssel z'entwéckelen, op deem d'Authentifikatioun an d'Identifikatioun vun Messagen verschlësselt ginn. SIGMA beschreift zwou Optiounen:
- SIGMA-I - schützt den Initiator virun aktiven Attacken, de Respekter vu passiven: den Initiator authentifizéiert den Respekter a wann eppes net passt, da gëtt et seng Identifikatioun net. De Bekloten gëtt seng Identifikatioun eraus wann en aktive Protokoll mat him gestart gëtt. De passive Beobachter léiert näischt;
SIGMA-R - schützt den Äntwerte vu aktive Attacken, den Initiator vu passiven. Alles ass genau de Géigendeel, awer an dësem Protokoll gi véier Handshake Messagen scho vermëttelt.Mir wielen SIGMA-I well et méi ähnlech ass wéi wat mir vu Client-Server vertraute Saachen erwaarden: de Client gëtt nëmmen vum authentifizéierten Server unerkannt, a jidderee weess de Server schonn. Plus ass et méi einfach ze implementéieren wéinst manner Handshake Messagen. Alles wat mir zum Protokoll addéieren ass en Deel vum Message ze verschlësselen an den Identifizéierer A an de verschlësselten Deel vum leschte Message ze transferéieren:
PubA, CookieA │ ╔══════════ ═══════════════════ ────────── ───── ────────── ──────────────────────── ─────────── ───── ──────>│ ║SignPrvA , SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ║┕┐┕ ═════════ ═════════ ════╝ │ PubB, CookieB, Enc((IdB, sign(SignPrvB, (CookieA, CookieB, PubB)), MAC(IdB)) ┕┕┕┕ ═══════════════ ═══════╗ │<───────────── ────────── ───── ────────── ║SignP rvB, SignPubB = load()║ │ │ ║ │ │ ║ │ │ ────── ║║┐gen() ════════ ════════════════╝ │ │ ╔════════ ══════ ══╗ │ Enc((IdA, sign( SignPrvA, (CookieB, CookieA, PubA)), MAC(IdA))) │ ║Key = DH(PrvA, PubB) ║ │──────────────────── ────────────────── ────────────────────── ─────────── ──────>│ ║verify(Key, IdB) ║ │ │ ║verify( SignPubB, ...)║ │ │ ╚═══════════ ═════ ══╝ │ │
- GOST R gëtt fir Ënnerschrëft benotzt Algorithmus mat 256-Bit Schlësselen.
- Fir den ëffentleche Schlëssel ze generéieren, gëtt 34.10/2012/XNUMX VKO benotzt.
- CMAC gëtt als MAC benotzt. Technesch ass dëst e spezielle Modus vun der Operatioun vun engem Block Chiffer, beschriwwen am GOST R 34.13-2015. Als Verschlësselungsfunktioun fir dëse Modus - (34.12-2015).
- Den Hash vu sengem ëffentleche Schlëssel gëtt als Identifizéierer vum Gespréichspartner benotzt. Benotzt als Hash (34.11/2012/256 XNUMX Bits).
Nom Handschlag stëmme mir iwwer e gemeinsame Schlëssel. Mir kënnen et fir authentifizéiert Verschlësselung vun Transportmeldungen benotzen. Dësen Deel ass ganz einfach a schwéier e Feeler ze maachen: mir erhéijen de Message Konter, verschlësselen de Message, authentifizéieren (MAC) de Konter a Chiffertext, schécken. Wann Dir e Message kritt, kontrolléiere mir datt de Comptoir den erwuessene Wäert huet, authentifizéieren de Chiffertext mam Comptoir an entschlësselen. Wéi ee Schlëssel soll ech benotzen fir Handshake Messagen ze verschlësselen, Messagen ze transportéieren a wéi ech se authentifizéieren? Ee Schlëssel fir all dës Aufgaben ze benotzen ass geféierlech an onkloer. Et ass néideg Schlësselen mat spezialiséiert Funktiounen ze generéieren (Schlëssel Derivatioun Funktioun). Nach eng Kéier, loosst eis keng Hoer opgedeelt an eppes erfannen: ass laang bekannt, gutt recherchéiert an huet keng bekannte Problemer. Leider huet déi gebierteg Python Bibliothéik dës Funktioun net, also benotze mir Plastikstut. HKDF intern benotzt , déi am Tour eng Hash-Funktioun benotzt. E Beispill Implementatioun am Python op der Wikipedia Säit brauch just e puer Zeilen Code. Wéi am Fall vun 34.10/2012/256 wäerte mir Stribog-XNUMX als Hash Funktioun benotzen. D'Ausgab vun eiser Schlësselvertragsfunktioun gëtt de Sessiounsschlëssel genannt, aus deem déi fehlend symmetresch generéiert ginn:
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")Strukturen / Schemaen
Loosst eis kucken wéi eng ASN.1 Strukture mir elo hunn fir all dës Donnéeën ze vermëttelen:
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 ass wat ënnerschriwwe gëtt. HandshakeTBE - wat wäert verschlësselte ginn. Ech zéien Är Opmierksamkeet op d'ukm Feld am MsgHandshake1. 34.10 VKO, fir nach méi randomization vun der generéiert Schlësselen, ëmfaasst de UKM (Benotzer Schlëssel Material) Parameter - just zousätzlech Entropie.
Füügt Kryptographie zum Code
Loosst eis nëmmen d'Ännerungen, déi am Originalcode gemaach goufen, betruechten, well de Kader d'selwecht bliwwen ass (tatsächlech gouf d'endgülteg Ëmsetzung fir d'éischt geschriwwe, an dunn ass all d'Kryptographie ausgeschnidden).
Well d'Authentifikatioun an d'Identifikatioun vun de Gespréichspartner mat ëffentleche Schlësselen duerchgefouert gëtt, musse se elo iergendwou fir eng laang Zäit gespäichert ginn. Fir Einfachheet benotze mir JSON esou:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }eis - eis Schlësselpaar, hexadezimal privat an ëffentlech Schlësselen. hir - Nimm vun Gespréichspartner an hir ëffentlech Schlësselen. Loosst eis d'Argumenter vun der Kommandozeil änneren an d'Postveraarbechtung vun JSON Daten addéieren:
from pygost import gost3410 from pygost.gost34112012256 import GOST34112012256 CURVE = gost3410.GOST3410Curve( *gost3410.CURVE_PARAMS["GostR3410_2001_CryptoPro_A_ParamSet"] ) parser = argparse.ArgumentParser(description="GOSTIM") parser.add_argument( "--keys-gen", action="store_true", help="Generate JSON with our new keypair", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON with our and their keys", ) parser.add_argument( "--bind", default="::1", help="Address to listen on", ) parser.add_argument( "--port", type=int, default=6666, help="Port to listen on", ) args = parser.parse_args() if args.keys_gen: prv_raw = urandom(32) pub = gost3410.public_key(CURVE, gost3410.prv_unmarshal(prv_raw)) pub_raw = gost3410.pub_marshal(pub) print(json.dumps({ "our": {"prv": hexenc(prv_raw), "pub": hexenc(pub_raw)}, "their": {}, })) exit(0) # Parse and unmarshal our and their keys {{{ with open(args.keys, "rb") as fd: _keys = json.loads(fd.read().decode("utf-8")) KEY_OUR_SIGN_PRV = gost3410.prv_unmarshal(hexdec(_keys["our"]["prv"])) _pub = hexdec(_keys["our"]["pub"]) KEY_OUR_SIGN_PUB = gost3410.pub_unmarshal(_pub) KEY_OUR_SIGN_PUB_HASH = OctetString(GOST34112012256(_pub).digest()) for peer_name, pub_raw in _keys["their"].items(): _pub = hexdec(pub_raw) KEYS[GOST34112012256(_pub).digest()] = { "name": peer_name, "pub": gost3410.pub_unmarshal(_pub), } # }}}De private Schlëssel vum 34.10 Algorithmus ass eng zoufälleg Zuel. 256-Bit Gréisst fir 256-Bit elliptesch Kéiren. PyGOST funktionnéiert net mat enger Rei vu Bytes, awer mat , also muss eise private Schlëssel (urandom(32)) an eng Zuel ëmgewandelt ginn mat gost3410.prv_unmarshal(). Den ëffentleche Schlëssel gëtt deterministesch vum private Schlëssel bestëmmt mat gost3410.public_key (). Den ëffentleche Schlëssel 34.10 ass zwou grouss Zuelen déi och an eng Bytesequenz ëmgewandelt musse ginn fir d'Lagerung an d'Transmissioun mat gost3410.pub_marshal ().
Nodeems Dir d'JSON-Datei gelies hutt, mussen d'ëffentlech Schlësselen deementspriechend mat gost3410.pub_unmarshal () ëmgewandelt ginn. Well mir d'Identifikatoren vun de Gespréichspartner a Form vun engem Hash vum ëffentleche Schlëssel kréien, kënnen se direkt am Viraus berechent ginn an an engem Wierderbuch fir séier Sich gesat ginn. Stribog-256 Hash ass gost34112012256.GOST34112012256(), deen den Hashlib-Interface vun Hashfunktiounen voll erfëllt.
Wéi huet d'Initiator Coroutine geännert? Alles ass wéi am Handshake Schema: mir generéieren e Cookie (128-Bit ass vill), en ephemerescht Schlësselpaar 34.10, dat fir d'VKO Schlësselvertragsfunktioun benotzt gëtt.
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 # Generate our ephemeral public key and cookie, send Handshake 0 message {{{ 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()- mir waarden op eng Äntwert an decodéieren den erakommende Msg Message;
- sécherstellen Dir Handshake1 kréien;
- decodéieren den ephemeralen ëffentleche Schlëssel vun der Géigendeel Partei a berechnen de Sessiounsschlëssel;
- Mir generéieren symmetresch Schlësselen déi néideg sinn fir den TBE Deel vum Message ze veraarbecht.
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 # Validate Handshake message {{{ 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 ass eng 64-bëssen Zuel (urandom (8)), déi verlaangt och deserialization vu senger Byte Representatioun mat gost3410_vko.ukm_unmarshal (). VKO Funktioun fir 34.10/2012/256 3410-bëssen ass gost34102012256_vko.kek_XNUMX () (KEK - Verschlësselungsschlëssel).
De generéierte Sessiounsschlëssel ass schonn eng 256-Bit pseudo-zoufälleg Byte Sequenz. Dofir kann et direkt an HKDF Funktiounen benotzt ginn. Zënter GOST34112012256 den Hashlib-Interface erfëllt, kann et direkt an der Hkdf Klass benotzt ginn. Mir spezifizéieren net d'Salz (dat éischt Argument vun Hkdf), well de generéierte Schlëssel, wéinst der Ephemeralitéit vun den deelhuelende Schlësselpaaren, fir all Sessioun anescht ass a scho genuch Entropie enthält. kdf.expand () produzéiert par défaut schonn déi 256-Bit Schlësselen déi spéider fir Grasshopper néideg sinn.
Als nächst ginn d'TBE- an TBS-Deeler vum erakommen Message iwwerpréift:
- de MAC iwwer den erakommen Chiffertext gëtt berechent a kontrolléiert;
- de Chiffertext gëtt entschlësselt;
- TBE Struktur gëtt dekodéiert;
- den Identifikateur vum Gespréichspartner gëtt dovun geholl an et gëtt gepréift ob hien eis iwwerhaapt bekannt ass;
- MAC iwwer dësen Identifizéierer gëtt berechent a kontrolléiert;
- d'Ënnerschrëft iwwer d'TBS Struktur gëtt verifizéiert, wat de Cookie vu béide Parteien an den ëffentlechen ephemeresche Schlëssel vun der Géigendeel Partei enthält. D'Ënnerschrëft gëtt duerch de laangliewege Ënnerschrëftschlëssel vum Gespréichspartner verifizéiert.
441 try: 442 peer_name = validate_tbe( 443 msg_handshake1, 444 key_handshake1_mac_identity, 445 key_handshake1_enc, 446 key_handshake1_mac, 447 cookie_our, 448 cookie_their, 449 pub_their_raw, 450 ) 451 except ValueError as err: 452 logging.warning("%s: %s, disconnecting", _id, err) 453 writer.close() 454 return 455 # }}} 128 def validate_tbe( 129 msg_handshake: Union[MsgHandshake1, MsgHandshake2], 130 key_mac_identity: bytes, 131 key_enc: bytes, 132 key_mac: bytes, 133 cookie_their: Cookie, 134 cookie_our: Cookie, 135 pub_key_our: PubKey, 136 ) -> str: 137 ciphertext = bytes(msg_handshake["ciphertext"]) 138 mac_tag = mac(GOST3412Kuznechik(key_mac).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext) 139 if not compare_digest(mac_tag, bytes(msg_handshake["ciphertextMac"])): 140 raise ValueError("invalid MAC") 141 plaintext = ctr( 142 GOST3412Kuznechik(key_enc).encrypt, 143 KUZNECHIK_BLOCKSIZE, 144 ciphertext, 145 8 * b"x00", 146 ) 147 try: 148 tbe, _ = HandshakeTBE().decode(plaintext) 149 except ASN1Error: 150 raise ValueError("can not decode TBE") 151 key_sign_pub_hash = bytes(tbe["identity"]) 152 peer = KEYS.get(key_sign_pub_hash) 153 if peer is None: 154 raise ValueError("unknown identity") 155 mac_tag = mac( 156 GOST3412Kuznechik(key_mac_identity).encrypt, 157 KUZNECHIK_BLOCKSIZE, 158 key_sign_pub_hash, 159 ) 160 if not compare_digest(mac_tag, bytes(tbe["identityMac"])): 161 raise ValueError("invalid identity MAC") 162 tbs = HandshakeTBS(( 163 ("cookieTheir", cookie_their), 164 ("cookieOur", cookie_our), 165 ("pubKeyOur", pub_key_our), 166 )) 167 if not gost3410.verify( 168 CURVE, 169 peer["pub"], 170 GOST34112012256(tbs.encode()).digest(), 171 bytes(tbe["signature"]), 172 ): 173 raise ValueError("invalid signature") 174 return peer["name"]Wéi ech uewen geschriwwen, 34.13/2015/XNUMX beschreift verschidde aus 34.12/2015/3413. Dorënner gëtt et e Modus fir Imitatioun Inserts a MAC Berechnungen ze generéieren. An PyGOST ass dëst gost34.12.mac (). Dëse Modus erfuerdert d'Verschlësselungsfunktioun ze passéieren (ee Block vun Daten kréien an zréckginn), d'Gréisst vum Verschlësselungsblock an, tatsächlech, d'Donnéeën selwer. Firwat kënnt Dir d'Gréisst vum Verschlësselungsblock net hardcode? 2015/128/64 beschreift net nëmmen den XNUMX-Bit Grasshopper Chiffer, awer och de XNUMX-Bit - e liicht geännert GOST 28147-89, erstallt zréck am KGB an huet nach ëmmer ee vun den héchste Sécherheetsschwellen.
Kuznechik initialized vun engem Opruff gost.3412.GOST3412Kuznechik (Schlëssel) a gëtt en Objet mat .encrypt () / .decrypt () Methoden gëeegent fir Passe ze 34.13 Funktiounen. MAC gëtt wéi follegt berechent: gost3413.mac (GOST3412Kuznechik (Schlëssel).encrypt, KUZNECHIK_BLOCKSIZE, Chiffertext). Fir de berechent a kritt MAC ze vergläichen, kënnt Dir den übleche Verglach (==) vu Byte Strings net benotzen, well dës Operatioun d'Vergläichszäit leeft, wat am allgemenge Fall zu fatale Schwachstelle féiere kann wéi Attacken op TLS. Python huet eng speziell Funktioun, hmac.compare_digest, fir dës.
D'Block Chiffer Funktioun kann nëmmen ee Block vun Daten verschlësselen. Fir eng méi grouss Zuel, an och net e Multiple vun der Längt, ass et néideg de Verschlësselungsmodus ze benotzen. 34.13-2015 beschreift déi folgend: EZB, CTR, OFB, CBC, CFB. Jiddereen huet seng eege akzeptabel Uwendungsberäicher a Charakteristiken. Leider hu mir nach ëmmer net standardiséiert (wéi CCM, OCB, GCM an dergläiche) - mir si gezwongen op d'mannst MAC selwer ze addéieren. Ech wielen (CTR): et erfuerdert keng Polsterung op d'Blockgréisst, kann paralleliséiert ginn, benotzt nëmmen d'Verschlësselungsfunktioun, ka sécher benotzt ginn fir eng grouss Zuel vu Messagen ze verschlësselen (am Géigesaz zu CBC, déi relativ séier Kollisiounen huet).
Wéi .mac (), hëlt .ctr () ähnlechen Input: Chiffertext = gost3413.ctr (GOST3412Kuznechik (Schlëssel). Encrypt, KUZNECHIK_BLOCKSIZE, Kloertext, iv). Et ass erfuerderlech en Initialiséierungsvektor ze spezifizéieren dee genee d'Halschent vun der Längt vum Verschlësselungsblock ass. Wann eise Verschlësselungsschlëssel nëmme benotzt gëtt fir e Message ze verschlësselen (och wann aus e puer Blocks), dann ass et sécher en Null Initialisierungsvektor ze setzen. Fir Handshake Messagen ze verschlësselen, benotze mir all Kéier e separate Schlëssel.
D'Ënnerschrëft gost3410.verify() z'iwwerpréiwen ass trivial: mir passéieren déi elliptesch Curve an där mir schaffen (mir notéieren se einfach an eisem GOSTIM Protokoll), den ëffentleche Schlëssel vum Ënnerschreiwer (vergiesst net datt dëst en Tupel vun zwee sollt sinn grouss Zuelen, an net eng Byte String), 34.11/2012/XNUMX Hash an der Ënnerschrëft selwer.
Als nächst, am Initiator preparéieren a schécken mir eng Handshake Message un Handshake2, déi selwecht Handlungen ausféieren wéi mir während der Verifizéierung gemaach hunn, nëmme symmetresch: Ënnerschreiwe op eise Schlësselen amplaz ze kontrolléieren, etc ...
456 # Prepare and send Handshake 2 message {{{ 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)Wann d'Sessioun etabléiert ass, ginn Transportschlëssel generéiert (e separate Schlëssel fir Verschlësselung, fir Authentifikatioun, fir jidderee vun de Parteien), an de Grasshopper gëtt initialiséiert fir de MAC ze entschlësselen an ze kontrolléieren:
499 # Run text message sender, initialize transport decoder {{{ 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 # Wait for test messages {{{ 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 # }}}D'msg_sender Coroutine verschlësselt elo Messagen ier se op eng TCP Verbindung schéckt. All Message huet eng monoton Erhéijung Nonce, déi och d'Initialiséierung Vecteure ass wann am Konter Modus verschlësselte. All Message a Message Block ass garantéiert eng aner Konter Wäert ze hunn.
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 += 1Entréeën Messagen gi vun der msg_receiver Coroutine veraarbecht, déi d'Authentifikatioun an d'Entschlësselung behandelt:
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("unexpected nonce value") mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode()) if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])): raise ValueError("invalid 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)Konklusioun
GOSTIM soll exklusiv fir pädagogesch Zwecker benotzt ginn (well et op d'mannst net vun Tester ofgedeckt ass)! De Quellcode vum Programm kann erofgeluede ginn (Стрибог-256 хэш: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Как и все мои проекты, типа , , , , GOSTIM ass komplett , ënner de Konditioune verdeelt .
, , Member , Python / Go Entwéckler, Chef Spezialist .
Source: will.com