Դիվանագետ լինելով գրադարանի (ԳՈՍՏ-ին հատուկ криптոգրաֆիական պրիմիտիվներ մաքուր Python-ում), ես հաճախ ստանում եմ հարցեր, թե ինչպես կարելի է մակերեսային անվտանգության հաղորդակցություն իրականացնել: Շատերը կարծում են, որ կիրառական криптոգրաֆիան բավականին հեշտ խնդիր է և .encrypt() կոչը բլոկային шифրանքի համար բավարար կլինի անվտանգ հաղորդման համար: Իսկ մյուսները հարցնում են, թե կիրառական криптոգրաֆիան բացառապես մի քանի մարդու ոլորտն է, և ընդունելի է, որ հարուստ ընկերություններ, ինչպիսիք են Telegram-ը, չունեն математիկայի օլիմպիական մասնագետներ անվտանգ պրոտոկոլ:
Այս ամենը ինձ հարկադրեց գրել այս հոդվածը՝ ցույց տալու, որ крипտոգրաֆիական պրոտոկոլների և անվտանգ IM-ի իրականացման խնդիրը այնքան էլ բարդ չի: Բայց ինքնուրույն ավտենտիկացիայի և բանալիների համաձայնեցման պրոտոկոլներ մշակելը հիմնավոր չէ:
Հոդվածում կգրվի , , հանդիպումների համար ավտենտիկացիայի և բանալիների համաձայնեցման պրոտոկոլ (որի վրա կառուցված է ), օգտագործելով բացառապես PyGOST գրադարանի ԳՈՍՏային крипտոգրաֆիական ալգորիթմներ և ASN.1 հաղորդագրությունների կոդավորումը PyDERASN գրադարանով (որի մասին ես արդեն ). Անհրաժեշտ պայմանը. այն պետք է բավականաչափ պարզ լինի, որպեսզի կարողանանք այն գրել զրոյից մեկ երեկո (կամ աշխատանքային օր) ընթացքում, այլապես դա այլևս պարզ ծրագիր չէ: Այդ ծրագրում անպայման կլինեն սխալներ, ավելորդ բարդություններ, թերություններ, Plus, սա իմ առաջին ծրագիրն է, որը օգտագործում է asyncio գրադարանից:
IM-ի դիզայն
Բառ պետք է հասկանալ, թե ինչպես պետք է տեսք ունենա մեր IM-ը: Պարզության համար, թող դա լինի peer-to-peer ցանց, առանց մասնակիցների հայտնաբերման: Մենք ինքներս պետք է նշենք, թե որ հասցեին: պորտ ինք նկատմամբ կապվելու համար զրուցակիցին:
Ես հասկանում եմ, որ ներկայիս պահին մեջի շատ լավ ճանաչում է, որ երկու մատչելի համակարգիչների միջև անմիջական կապի պատրաստակամությունը՝ գրեթե իրական լինելու սահմանափակում է՝ IM-ի կիրառման մեջ: Բայց որքան շատ ծրագրավորողներ անում են NAT-traversal լուծումներ, այնքան երկար ենք մնում IPv4-ի համացանցում, տխուր հավանականություններով երկկողմ դասը ունենալու համար: Ասեք, հնարավոր չէ ու մենակ IPv6-ը տան և աշխատավայրերի մեջ:
Մենք կունենանք friend-to-friend ցանց. բոլոր հնարավոր զրուցակիցները նախապես պետք է լինեն հայտնի: Առաջինը, դա զանազան ավելացնում է՝ դեմքերից, գտանք և չգտանք անուն/ բանալի, անջատվեցինք, թե շարունակենք աշխատանքը, գիտենալով զրուցակցին: Երկրորդ, ընդհանուր առմամբ, դա անվտանգ է և վերացնում է բազմաթիվ հարձակումներ:
IM-ի ինտերֆեյսը մոտ կլինի ավանդական լուծումներին , որոնք ինձ շատ են դուր գալիս իրենց նվազագույնությամբ և Unix-way פילիսոֆիայի: IM ծրագիրը յուրաքանչյուր զրուցակցի համար ստեղծում է ֆայլային շատյուն՝ երեք Unix domain socket-ով:
- in — այնտեղ գրառվում են ուղարկվող զրուցակցին հասցեագրված հաղորդագրություններ:
- out — այստեղ ընթերցվում են դիմակից ստացված հաղորդագրությունները;
- state — դրանում ընթերցելով մենք իմանում ենք, արդյոք դիմակը այժմ միացված է, կապի հասցեն/պորտը:
Բացի այդ, ստեղծվում է conn սոկետ, որի մեջ մենք գրանցում ենք հոսթ պորտը՝ սկսելու համար միացում հեռավոր դիմակին:
|-- alice
| |-- in
| |-- out
| `-- state
|-- bob
| |-- in
| |-- out
| `-- state
`- conn
Այս մոտեցումը թույլ է տալիս անկախ իրականացնել IM տրանսպորտը և օգտվողի ինտերֆեյսը, քանի որ համտեսի ու գույնի համար ընկեր չկա, յուրաքանչյուրին հաճելի չի լինի: Օգտագործելով և/կամ , կարելի է ստանալ բազմաթիվ պատուհաններով ինտերֆեյս սինտաքսի շեշտադրմամբ: Իսկ օգնությամբ կարող ենք ստանալ GNU Readline համահունչ տող հաղորդագրությունների մուտքագրության համար:
واقعում, suckless նախագծերը օգտագործում են FIFO-ֆայլեր: Միայն ես չեմ կարողացել հասկանալ, թե ինչպես asyncio-ով աշխատել ֆայլերի հետ մրցակցաբար առանց ձեռքով ստացված թերի թելերի նախապատրաստական հենաշերտի (այնպիսի բաների համար վաղուց օգտագործում եմ լեզու ). Հետևաբար, որոշեցի обходиться UNIX domain սոկետներով: Կ desgrակի, դա սահմանափակում է հնարավորությունը անել echo 2001:470:dead::babe 6666 > conn: Ես լուծեցի այս խնդիրը՝ օգտագործելով : echo 2001:470:dead::babe 6666 | socat — UNIX-CONNECT:conn, socat READLINE UNIX-CONNECT:alice/in.
Նախնական անապահով արձանագրություն
Որպես տրանսպորտ օգտագործվում է TCP. դա երաշխավորում է առաքումը և նրա հերթականությունը: UDP-ի դեպքում ոչինչ չի երաշխավորում, և դա կարող է օգտակար լինել, երբ կիրառվի կրիպտոգրաֆիան, իսկ աջակցություն Python-ում քանզի չկա:
Ցավոք, TCP-ում հաղորդագրության գաղափար չկա, միայն բիթերի հոսք: Հետևաբար, անհրաժեշտ է մտածել հաղորդագրությունների ձևաչափ, որպեսզի դրանք կարողանանք բաժանել միմյանց մեջ այս հոսքում: Կարող ենք պայմանավորվել օգտագործել նոր շարքի նշան: Նախնականում համապատասխան կլինի, սակայն երբ մենք սկսենք կոդավորել մեր հաղորդագրությունները, այս նշանը կարող է հայտնվել ցանկացած վայրում կոդավորված տեքստում: Այնպես որ, ցանցերում լայն տարածում են ստանում արձանագրությունները, որոնք նախ ուղարկում են հաղորդագրության երկարությունը բիթերով: Օրինակ, Python-ում կա xdrlib, որը թույլ է տալիս աշխատել նման ձևաչափով: .
Մենք չենք կարող բավարար և արդյունավետ աշխատել TCP կարդալու հետ՝ հեշտացնելով կոդը: Ընթերցում ենք անվերջ օղակում տվյալները սոկետից, մինչև որ մենք դեկոդируем ամբողջական հաղորդագրություն: Բացի այդ, նման մոտեցման համար կարելի է օգտագործել JSON կամ XML: Բայց երբ կհարցվի կրիպտոգրաֆիա, ապա տվյալները պետք է ստորագրվեն և ავტենտիկացվեն՝ և դա կպահանջի բիթ-վ բիթ նույնական պատկերի ներկայացումում, ինչպիսիք չեն ապահովում JSON/XML (dumps արդյունքը կարող է տարբեր լինել):
XDR-ը հարմար է նման խնդիրների համար, սակայն ես ընտրում եմ ASN.1 մոդուլային և գիրապահություն, քանի որ մեզ մոտ կլինեն բարձր մակարդակի օբյեկտներ, որոնց հետ հաճախ հաճելի և հարմարավետ է աշխատել: Հատուկ schemaless , կամ , ASN.1 ինքնավար ստուգելու է տվյալները խիստ նշանակված սխեմայի դեմ:
# 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))),
))
Ստացվող հաղորդումը կլինի Msg: թե TextMsg (ն tuyến մեկ տեքստային դաշտով), թե MsgHandshake հաղորդում (որտեղ փոխանցվում է զրուցակցի անունը): Այստեղ դա թվում է բարդ, բայց դա ապագայի համար նախապատրաստում է:
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│MsgHandshake(IdA) │
│─────────────────>|
│ │
│MsgHandshake(IdB) │
│|
│ │
│ MsgText() │
│<─────────────────|
│ │
IM առանց կոդավորման
Ինչպես արդեն ասել եմ, բոլոր վանդակների գործողություններից օգտագործվելու է asyncio գրադարանը: Հռչակենք, թե ինչ է սպասվում բաշխման ժամանակ:
parser = argparse.ArgumentParser(description="GOSTIM")
parser.add_argument(
"--our-name",
required=True,
help="Մեր զրուցակցի անուն",
)
parser.add_argument(
"--their-names",
required=True,
help="Նրանց զրուցակցի անունները, բաժանված 쉐",
)
parser.add_argument(
"--bind",
default="::1",
help="Այստեղ լսելու հասցեն",
)
parser.add_argument(
"--port",
type=int,
default=6666,
help="Լսելու պորտը",
)
args = parser.parse_args()
OUR_NAME = UTF8String(args.our_name)
THEIR_NAMES = set(args.their_names.split(","))
Հատկացվում է սեփական անուն (—our-name alice): Խոսակցուների անունները ցուցակվում են կոնյուկցիայի՝ (—their-names bob,eve): Եվ յուրաքանչյուր զրուցակցի համար ստեղծվում է կատեգորիաներ Unix վանդակներով, ինչպես նաև մի coroutine յուրաքանչյուր 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"))
Օգտագործողի կողմից ստացվող հաղորդումները in վանդակից ուղարկվում են 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"))
Զրուցակցներից ստացվող հաղորդումները փոխանցվում են OUT_QUEUES հերթերին, որոնցից տվյալները գրանցվում են 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()
state սուզելում կարդալիս ծրագիրը փնտրում է PEER_ALIVE բառարանում զրուցակցի հասցեն: Եթե միացման դեռ չկա, ապա գրանցվում է դատարկ ստեղ.
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()
conn վանդակին հասցեն գրանցելիս սկսվում է միացման «արտո՞ւռչ» ֆունկցիան:
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))
Որպես առաջին քայլ, անշուշտ, նա բացում է կապը նշված հյուրը/պորտի հետ և ուղարկում տնային հաղորդագրություն իր անունով:
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()
Այնուհետև սպասում է հեռավոր կողմի պատասխանին: Փորձում է մոդելավորել ստացված պատասխանն ըստ Msg ASN.1 սխեմայի: Ծանոթանում են, որ ողջ հաղորդագրությունն ուղարկվելու է մեկ TCP բաժնում և երբ պարզապես կանչում ենք .read()-ը, որևիցե շրջանաձև ստանում ենք:
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 # }}}
Ավելի ուշ ստուգում ենք, թե արդյոք ստացված զրուցակցի անունը ճանաչում ենք: Եթե ոչ, ապա պատռում ենք կապը: Ստուգում ենք, արդյո՞ք արդեն կապած ենք եղել նրան (զրուցակիցը կրկին հրահանգել է կապակցվել մեզ հետ) և փակել այն: IN_QUEUES շարքերում տեղադրվում են Python-յան տողեր հաղորդագրության հաղորդագրմամբ, բայց կա հատուկ արժեք None, որը ազդանշանում է msg_sender մամուլին դադար տալ և նրան մոռանալ իր writer-ի մասին, կապված հին 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-ը ստանում է ելքային հաղորդագրությունները (տեղադրված in սոճու մեջ) և սևում է նրանց MsgText հաղորդարգի մեջ և ուղարկում TCP կապով: Այն կարող է խզվել ցանկացած պահի, և դա մենք իհարկե կանխում ենք.
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))
Վերջում նախաձեռնողը մտնում է հաղորդագրությունները թերթելու անդադար ցիկլի մեջ socket-ից: Նա ստուգում է, արդյոք այս文本ային հաղորդագրություններ են և OUT_QUEUES հերթում է դնում, որի միջոցով նրանք կնվիրվեն առաքման socket-ին համապատասխան զրույցի հետ։ Ինչու չի կարելի պարզապես կատարել .read() և դեկոդավորում հաղորդագրությունը: Որովհետև բացառված չէ այն իրավիճակը, երբ մի քանի հաղորդագրություններ օգտվողից կհավաքվեն օպերացիոն համակարգի buffers-ում եւ կուղարկվեն մեկ TCP-հիմքի միջոցով: Մենք կարող ենք դեկոդավորել միայն առաջինը, իսկ հետո buffer-ում կարող է մնալ հաջորդի մաս: Որոշակի ոչ ստանդարտ իրավիճակում մենք փակել ենք TCP-հաղորդակցությունը և կանգնեցնում msg_sender coroutine-ը (None ուղարկելով OUT_QUEUES հերթում):
174 buf = b""
175 # Ստիպված ենք սպասել փորձարկման հաղորդագրություններին {{{
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: գերազանցել է առավելագույն buffer չափը", _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: սպասված չէ %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: կտրվում է: %s", _id, peer_name)
199 IN_QUEUES[peer_name].put(None)
200 writer.close()
66 async def msg_receiver(msg_text: MsgText, peer_name: str) -> None:
67 text = str(msg_text["text"])
68 logging.info("%s: ստացել է %d նիշ հաղորդագրություն", peer_name, len(text))
69 await OUT_QUEUES[peer_name].put(text)
Վերադառնանք հիմնական кодին։ Ամեն coroutine ստեղծելուց հետո ծրագրի մեկնարկի պահին սկսում ենք TCP-ծառը։ Ամեն մեկ հաստատված միանալու ժամանակ այն ստեղծում է responder (պատասխանողի) 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("Լսում է՝ %s", server.sockets[0].getsockname())
loop.run_forever()
responder-ը նման է initiator-ին և զուգորդված իրականացնում է բոլոր նույն գործողությունները, բայց հաղորդագրությունները թերթելու անդադար ցիկլը սկսվում է անմիջապես, պարզության համար։ Այժմ ձեռքսեղմման պրոտոկոլը ուղարկում է մեկ հաղորդագրություն յուրաքանչյուր կողմից, սակայն, հետագայում, հնարավոր կլինի ուղարկել երկուսը կապի նախաձեռնողից, այնուհետև անմիջապես կարող է լինել տեքստային հաղորդագրությունների ուղարկում։
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()
Անվտանգ պրոտոկոլ
Ժամանակն է ապահովել մեր շփումը: Ինչ է ապահովությունը և ինչ ենք ուզում:
- փոխանցվող հաղորդագրությունների գաղտնիությունը;
- փոխանցվող հաղորդագրությունների իսկություն և ամբողջականություն` դրանց փոփոխությունը պետք է հայտնաբերվի;
- վերադարձի հարձակումներից պաշտպանություն (replay attack)` հաղորդագրությունների կորուստն կամ կրկնությունը պետք է հայտնաբերվի (և մենք որոշում ենք խզել կապը);
- հանդիպողների հավաստում և հաստատում նախօրոք ներմուծված հանրային բանալուներով` մենք արդեն որոշել ենք, որ ստեղծենք ընկեր-մարտիկներ ցանց: Միայն հաստատումից հետո կիմանանք, թե ով է մեր հանդիպողը;
- առկայությունը հատկություններ (PFS)` մեր երկարաժամկետ ստորագրության բանալու զեղումը չպետք է թույլ տա ամբողջ նախորդ հաղորդակցությունը ընթերցել: Գրանցված տրանսպորտի տրաֆիկը դառնում է անօգուտ;
- հաղորդագրությունների (անդամների և ձեռակցման) գոյություն/վավերություն միայն մեկ TCP-ցուցադրման ընթացքում: Այլ ցուցադրությունում (տրանսպորտում նույնիսկ նույն հանդիպողի) ճիշտ ստորագրված/հաստատված հաղորդագրությունների մուտքը չի պետք է լինի հնարավոր;
- մոլոր ախուակահրողը չպետք է տեսնի ոչ օգտվողների նույնականացնողները, ոչ փոխանցվող երկարաժամկետ հանրային բանալուները, ոչ էլ դրանց հեշը: Որոշ անանունություն մոլոր ախուակահրողից:
Հեռուստացույցը, սակայն, այս նվազագույնը գրեթե բոլորն ուզում են ունենալ ցանկացած ձեռնհասության ալգորիթմում, և քչերն են նրանից վերջում կատարում «համոդելված» ալգորիթմների համար: Այժմ էլ նոր բան չպիտի հնարենք: Ինձ համար հստակորեն խորհուրդ կտամ օգտագործել ալգորիթմների կառուցման համար, բայց ընտրենք ինչ-որ ավելի պարզ:
Սա երկու ամենահայտնի ալգորիթմներ են:
- — առանցքային այնքան բարդ ալգորիթմ, որ ունի երկար պատմություն սխալների, խնդիրների, խոցելիությունների, վատ մտածվածության, բարդության և թերությունների ( սակայն, TLS 1.3-ի հետ կապ չունի): Բայց չենք դիտարկում այն բարդության պատճառով:
- с — չունեն լուրջ կտրող խնդիրներ, չնայած նույնպես բարդ են: Եթե կարդաք IKEv1-ի և IKEv2-ի մասին, ապա նրանց սկզբնատարը , ISO/IEC IS 9798-3 և SIGMA (SIGn-and-MAc) ալգորիթմները, բավականին պարզ են իրականացնելու համար մեկ երեկո:
Իսկ ինչով է SIGMA, որպես STS/ISO ալգորիթմների իդեալական փուլ, լավ? Այն բավարարում է մեր բոլոր պահանջներին (այդ թվում «կողմերի» ինքնությունը շեղելու), չունի հանրային քաղցկեղի խնդիրներ: Այն ծայրահեղMinimal է — գոնե մեկ բանի ապամոնտաժումը հաղորդագրությունից կպատճառի նրա անհանգստություն:
Եկեք անցնենք ամենապարզ մոդելավորված ալգորիթմից SIGMA: Մեր ամենառարձակ հետաքրքրող գործողությունը ՝ գործառույթ, որի ելքը երկու կողմը կստանան նույն արժեքը, որը կարող են օգտագործել սիմետրիական բանալիների համար: Յուրաքանչյուր կողմը ստեղծում է անցումային (մանկաստան մեկ սեսիայի շրջանակներում) բանալիների զույգ (հանրային և մասնավոր բանալիներ), փոխանակում են հանրային բանալիներով, զանգահարում են համաձայնեցման գործառույթը, որի մուտքի ենթադրում են իրենց մասնավոր բանալին և մյուս կողմի հանրային բանալին:
┌─────┐ ┌─────┐
│PeerA│ │PeerB│
└──┬──┘ └──┬──┘
│ IdA, PubA │ ╔════════════════════╗
│───────────────>│ ║PrvA, PubA = DHgen()║
│ │ ╚════════════════════╝
│ IdB, PubB │ ╔════════════════════╗
│<───────────────│ ║PrvB, PubB = DHgen()║
│ │ ╚════════════════════╝
────┐ ╔═══════╧════════════╗
│ ║Key = DH(PrvA, PubB)║
<───┘ ╚═══════╤════════════╝
│ │
│ │
Մ cualquiera puede քրքրել միջանկյալ և փոխարինել հանրային բանալիները իրենց սեփականով — այս ալգորիթմում չկա կողմերի հարմարեցման անվտանգություն: Ավելացնենք ստորագրություն երկարաժամկետ բանալիներով:
┌─────┐ ┌─────┐
│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) ║ │
│ ╚═════════════════════╝ │
│ │
Այս ստորագրությունը չի համապատասխանում, քանի որ այն կապված չէ կոնկրետ սեսիայի։ Այդպիսի հաղորդագրությունները «հարմար» կլինեն նաև այլ մասնակցուների սեսիաների համար։ Ստորագրությունն անհրաժեշտ է ամբողջ համատեքստին։ Սա նաև ստիպում է ավելացնել ևս մեկ հաղորդագրություն A-ից։
Ավելին, շատ կարևոր է ստորագրության տակ ավելացնել նաև սեփական նույնականացուցիչ, քանի որ, հակառակ դեպքում, մենք կարող ենք փոխարինել IdXXX-ները և մթերել հաղորդագրությունը մեկ այլ հայտնի զրուցակցի բանալիով։ Այս ամենը կանխելու համար , անհրաժեշտ է, որպեսզի ստորագրման տակ գտնվող առիթները գտնվեն ճշտաբար նշված վայրերում իրենց իմաստով։ Եթե A-ն ստորագրում է (PubA, PubB), ապա B-ն պետք է ստորագրի (PubB, PubA)։ Սա նաև ընդգծում է սարի կառուցվածքի և շարադրանքային տվյալների ձևաչափի ընտրության կարևորությունը։ Օրինակ, ASN.1 DER կոդավորման դեպքում խմբերը դասավորվում են։ SET OF(PubA, PubB) հավասար կլինի 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) ║ │ │ ╚═════════════════════╝ │ │
Արդեն մենք դեռ չենք «գալիս» ինչ-որ բան համար սեսիան եզրեր միասնական համատեղ բանալին: Իրականում, կարելի է շրջանցել այս քայլը — առաջին փոխադրման հաղորդագրությունը կլինի վավերական չէ, բայց ուզում ենք, որ երբ ձեռքի խնդիրը ավարտվի, իմանանք, որ ամեն ինչ իսկապես համաձայնեցված է: Այս պահին, մենք օգտվում ենք ISO/IEC IS 9798-3 արձանագրությունից:
Մենք կարող էինք ստորագրել նաև ինքներս մշակված բանալին: Սա վտանգավոր է, քանի որ միևնույն ժամանակ չի բացառվում, որ օգտագործվող ստորագրության ալգորիթմի մեջ կարող են լինել արտահոսքեր (չնայած դրանք կարող են լինել ստորագրության բիթեր, բայց, այնուամենայնիվ, արտահոսքեր): Կարող ենք ստորագրել մշակված բանալիի հանշը, սակայն նույնիսկ հանշի արտահոսքը մշակված բանալիից կարող է նշանակություն ունենալ brute-force հարձակման ժամանակ մշակման ֆունկցիայի դեմ: SIGMA-ն օգտագործում է MAC ֆունկցիա, որը հաստատում է ուղարկողի նույնականությունը:
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Ինչպես օպտիմիզումի մի մաս, որոշ մարդիկ կարող են ցանկանալ վերամշակել իրենց Էֆեմերային բանալիները (ինչը, իհարկե, ցավալի է PFS-ի համար): Օրինակ, մենք ստեղծել ենք բանալիների զույգ, փորձել ենք միանալ, բայց TCP-ն հասանելի չէր կամ ընդհատվել էր սահմանափակ ժամանակում: Ցավալի է ծախսել ծախսված էնդրոպիան և պրոցեսորի ռեսուրսները նոր զույգի համար: Ուստի, ներարկենք, ասվածը, փողկապ — կեղծ պատահական արժեք, որը կպաշտպանի հնարավոր պատահական կրկնուղի հարձակումներից, երբ կրկին օգտագործում ենք էնֆեմերային հանրային բանալիներ: Բանալիների փողկապի և էնֆեմերային հանրային բանալիի միջև կապվածության հետևանքով, հակառակ մասնակիցի հանրային բանալին կարելի է հեռացնել ստորագրությունից` առանց անհրաժեշտության։
┌─────┐ ┌─────┐ │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, ...)║ │ │ ╚═════════════════════╝ │ │
Finally, we want to ensure the privacy of our interlocutors' identifiers from passive observers. For this, SIGMA suggests first exchanging ephemeral keys to derive a common key, which will be used to encrypt authenticating and identifying messages. SIGMA describes two options:
- SIGMA-I — protects the initiator from active attacks and the responder from passive ones: the initiator authenticates the responder, and if something goes wrong, he does not reveal his identification. The responder, on the other hand, provides his identification if the active protocol is initiated. A passive observer learns nothing;
SIGMA-R — protects the responder from active attacks and the initiator from passive ones. Everything is exactly the opposite here, but this protocol involves four handshake messages.We choose SIGMA-I as it is more aligned with what we expect from typical client-server interactions: only the authenticated server recognizes the client, while the server already knows everything. Additionally, it is simpler to implement due to the smaller number of handshake messages. All we incorporate into the protocol is the encryption of part of the message and the transfer of identifier A into the encrypted part of the last message:
┌─────┐ ┌─────┐ │PeerA│ │PeerB│ └──┬──┘ └──┬──┘ │ PubA, CookieA │ ╔═══════════════════════════╗ │─────────────────────────────────────────────────────────────────────────────>│ ║SignPrvA, SignPubA = load()║ │ │ ║PrvA, PubA = DHgen() ║ │ │ ╚═══════════════════════════╝ │PubB, CookieB, Enc((IdB, sign(SignPrvB, (CookieA, CookieB, PubB)), MAC(IdB))) │ ╔═══════════════════════════╗ ││ ║verify(Key, IdB) ║ │ │ ║verify(SignPubB, ...)║ │ │ ╚═════════════════════╝ │ │
- Բա՛ց, ստորագրության համար կիրառվում է ГОСТ Р ալգորիթմը 256-բիթ ստեղների հետ:
- Երկարացված հանրագումարը ընդհանուր բանալու ստացման համար կիրառվում է 34.10-2012 VKO:
- MAC-ի համար կիրառվում է CMAC: տեխնիկապես դա բլոկային ծածուկի մոդոս հատուկ ռեժիմ է, որը նկարագրված է ГОСТ Р 34.13-2015: Այս ռեժիմի համար ծածկման ֆունկցիան՝ (34.12-2015).
- Մերձավորի ամենագրավիչի որպես նույնականացնող կիրառվում է նրա հանրային բանալու համար տեսակավորումը: Այս տեսակավորման համար կիրառվում է (34.11-2012 256 բիթ):
ՀHandshake-ի ավարտից հետո մենք կունենանք համաձայնեցված ընդհանուր բանալու: Այս բանալուն կարող ենք օգտագործել գովազդային ծածկման համար: Այս մասը բավականին պարզ է ու այստեղ դժվարագույն չեն սխալվել. ավելացնում ենք հաղորդագրությունների հաշիվը, ծածքում հաղորդագրությունը, գովազդում ենք (MAC) հաշիվը և ծածկագիրը, ուղարկում: Կրկին հաղորդագրություն ստանալու դեպքում ստուգում ենք, որ հաշիվն ունի սպասվող բանալի, ու իրեն ծածկագիրը համադրաբար ստուգում ենք: (հիմնական բանալի դերակատարի ֆունկցիա): Մեր կողմից, չպետք է խոստովանել ու տոչորավորել. մինչև հիմա հայտնի, լավ ուսումնասիրված և հայտնի խնդիրներ չունեցող է: Պարտադիր չէ, որ Python-ի բերիչ գրադարանում նման ֆունկցիա լինի, այնպես որ մենք օգտագործում ենք փաթեթը: HKDF-ը ներսում օգտագործում է , որը wiederum օգտագործում է հ.message ֆունկցիա: Python-ի օրինակների իրագործումը Վիքիպեդիայի էջում զբաղեցնում է մի քանի linhas կոդ: Ինչպես 34.10-2012 դեպքում, որպես hash ֆունկցիա մենք կիրառում ենք Стрибог-256: Մեր ֆունկցիայի իրենցեի կոդ՝ հանգույցի բանալին կոչվում է սեսիայի բանալին, որի հիման վրա կստեղծվեն բացակայող սիմետրիկներ:
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")Կառույցներ/սքեմաներ
Պահանջվում է քննարկել, թե որոնք են այժմ ASN.1 կառուցվածքները, որոնք մենք ստացել ենք բոլոր այս տվյալների փոխանցման համար:
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 be signed). HandshakeTBE — այն, ինչ պետք է ավտոմատացվի (to be encrypted). Նշում եմ MsgHandshake1-ում ukm դաշտը: 34.10 VKO, ավելի մեծ էլեկտրաշանույթների ստացման համար, պարունակում է UKM պարամետրը (user keying material) — պարզապես լրացուցիչ էնտրոպիա:
Կոդում criptography ավելացնելը
Համոզվենք՝ միայն օրիգինալ կոդում կատարված փոփոխությունները, քանի որ շրջանակը մնացել է նույնը (իրականում, ի սկզբանե գրել է վերջնական իրագործումը, և հետո այդ ամենը դուրս է գրվել):
Որպեսզի սահմանապակենք և ճանաչենք զրուցակցիները հանրային բանալիների միջոցով, դրանք այժմ պետք է երկարաժամկետ պահպանվել: Ումիության համար օգտագործենք JSON այս չափերի տարատեսակ:
{ "our": { "prv": "21254cf66c15e0226ef2669ceee46c87b575f37f9000272f408d0c9283355f98", "pub": "938c87da5c55b27b7f332d91b202dbef2540979d6ceaa4c35f1b5bfca6df47df0bdae0d3d82beac83cec3e353939489d9981b7eb7a3c58b71df2212d556312a1" }, "their": { "alice": "d361a59c25d2ca5a05d21f31168609deeec100570ac98f540416778c93b2c7402fd92640731a707ec67b5410a0feae5b78aeec93c4a455a17570a84f2bc21fce", "bob": "aade1207dd85ecd283272e7b69c078d5fae75b6e141f7649ad21962042d643512c28a2dbdc12c7ba40eb704af920919511180c18f4d17e07d7f5acd49787224a" } }մեր — մեր հիմնական զույգը, վեցանիշ այսպես կոչված անձնական և հանրային բանալիները։ նրանց — զրուցակիցների անունները և նրանց հանրային բանալիները։ Փոփոխենք հրամանի արկղի արգումենտները և ավելացնենք 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="Ստեղծել JSON մեր նոր բանալիների զույգով", ) parser.add_argument( "--keys", default="keys.json", required=False, help="JSON մեր և նրանց բանալիներով", ) parser.add_argument( "--bind", default="::1", help="Ադրես, որը պետք է լսել", ) parser.add_argument( "--port", type=int, default=6666, help="Լսելու համար պորտ", ) 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) # Փրկել և չմեկուսացնել մեր և նրանց բանալիները {{{ 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), } # }}}Անձնական բանալին 34.10 ալգորիթմի համար — պատահական թիվ։ 256-բիտ չափսով 256-բիտ էլիպտիկ կորերի համար։ PyGOST-ը չի աշխատում բայթի հավաքածուի հետ, այլ՝ , հետևաբար մեր անձնական բանալին (urandom(32)) պետք է փոխակերպվի թվի, օգտագործելով gost3410.prv_unmarshal()։ Հանրային բանալին վերաբերվում է անձնականին, օգտագործելով gost3410.public_key()։ Հանրային բանալին 34.10 — երկու մեծ թվեր, որոնք նույնպես պետք է փոխակերպվեն բայթային հաջորդականության համար՝ պահելու և տեղափոխելու հարմարավետության համար, օգտագործելով gost3410.pub_marshal()։
JSON ֆայլից կարդալուց հետո, հանրային բանալիները, համապատասխանաբար, պետք է վերածվեն, օգտագործելով gost3410.pub_unmarshal()։ Քանի որ մեզ կհասնեն զրուցակիցների նույնականացուցիչները հանրային բանալիից հաշվարկված հեշի տեսքով, ապա դրանք կարելի է անմիջապես նախապես հաշվարկել և տեղադրել բառարանում արագ որոնման համար։ Стрибог-256 հեշը դա gost34112012256.GOST34112012256()-ը, ամբողջությամբ բավարարելով hashlib-ի ֆունկցիաների հեշի հարցարկման ինտերֆեյսին։
Ինչպես փոխվեց նախաձեռնողի կորուտինը: Ամեն ինչ, ինչպես ձեռ shake shake-ը՝ ստեղծում ենք cookie (128-բիտ չափը բավական է), էլիպտիկ կույտ 34.10, որը կիրագործվի VKO բանալու համաձայնեցման ֆունկցիայի համար։
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 # Մենք ստեղծում ենք մեր ընդեղային հանրային բանալին և փաթեթը, ուղարկում ենք 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()- սպասում ենք պատասխանին և թվարկում ստացված Msg հաղորդագրությունը;
- հաստատում ենք, որ ստացել ենք handshake1;
- թվարկում ենք հակառակ կողմի դեռահաս հանրային բանալին ու հաշվարկում սեսիայի բանալին;
- ստեղծում ենք սիմետրիկ բանալիներ, որոնք անհրաժեշտ են 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 # Հաստատել 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-ն 64-բիթանոց թիվ է (urandom(8)), որը նույնպես պահանջում է կրկնօրինակել բայթային ներկայացումից, օգտագործելով gost3410_vko.ukm_unmarshal(): VKO ֆունկցիան 34.10-2012 256-բիթանոց համար հանդիսանում է gost3410_vko.kek_34102012256() (KEK — բանալիների এনkripsiyon բանալին):
Ստեղծված սեսիայի բանալին արդեն 256-բիթանոց բայթային թվային բաշխում է։ Այսպիսով, այն անմիջապես կարելի է օգտագործել HKDF ֆունկցիայում։ Քանի որ GOST34112012256 բավարարում է hashlib ինտերֆեյսին, այն կարելի է անմիջապես օգտագործել Hkdf դասում։ Այլ, առաջին արկածում է Hkdf, որը մենք չենք նշում, քանի որ ստեղծված բանալին կտրուկորեն կլինեն տարբեր յուրաքանչյուր սեսիայի համար և այն արդեն ունի բավարար սրբություն։ kdf.expand() արդեն արդեն թողարկում է 256-բիթանոց բանալիներ, որոնք անհրաժեշտ են Բերձորում հետագայում:
Այնուհետև ստուգվում են TBE և TBS բաժինները ստացված հաղորդագրության՝
- հաշվարկվում և ստուգվում է MAC ստացված սինվրա տեքստի վրա;
- հանել է սինվրա տեքստը;
- թիվ 1-ը գալիս է TBE կառուցվածքից;
- այստեղ վերցվում է բերանը դիմացինիդ և ստուգվում է, արդյոք նա մեզ հայտնի է ընդհանրապես;
- հաշվարկվում և ստուգվում է MAC այս ID-ի վրա;
- ստուգվում է կհաշվարկվի ստորագրությունը TBS կառուցվածքի վրա, որի մեջ մտնում են երկու կողմերի cookie-ները և հակառակորդի հանրային դեռահաս բանալին։ Ստորագրությունը ստուգվում է դիմավորող կողմի երկարատև ստորագրությամբ։
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"]Ինչպես արդեն նշվեց վերևում, 34.13-2015 նկարագրում է տարբեր 34.12-2015-ից: Այդ թվում կա, թե ինչպես պետք է արտադրվի IME մուտք, MAC-ի հաշվարկ: PyGOST-ում դա ագոստ3413.mac() է: Այս ռեժիմը պահանջում է գաղտնագրելու ֆունկցիայի փոխանցումը (ընդունող և վերադարձնող մեկ տվյալների բլոկ), գաղտնագրային բլոկի չափը և, սեփական, տվյալները: Ինչու չի կարելի խստորեն գրել բլոկի չափը: 34.12-2015 նկարագրում է ոչ միայն 128 բիթանոց Կուզնեչիկ գաղտնագիրը, այլ նաև 64 բիթանոց — մի փոքր փոփոխված ГОСТ 28147-89, որը ստեղծվել է դեռևս KGB-n և դեռևս ունի ամենաբարձր անվտանգության շեմերից մեկը:
Կուզնեչիկը ևнициалիզируется gost.3412.GOST3412Kuznechik(key) զանգով և վերադարձնում է օբյեկտ, որը ունի .encrypt() / .decrypt() մեթոդներ, որոնք կարելի են փոխանցել 34.13 գործառույթին: MAC-ը հաշվարկվում է հետևյալ կերպ. gost3413.mac(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, ciphertext): Համեմատելու համար հաշվարկված և ստացված MAC-ը չի կարելի օգտագործել սովորական համեմատմամբ (==) բայթային տողերի, քանի որ այս գործողությունը տալիս է ժամանակի համեմատության դժվարություններ, ինչը ընդհանուր առմամբ կարող է հանգեցնել մահացու թույլատրելիության, ինչպես նախա-հարձակումներին TLS-ի վրա: Python-ում for this-type specially hmac.compare_digest ֆունկցիան կա:
Բլոկային գաղտնագրի ֆունկցիան կարող է գաղտնագրել միայն մեկ տվյալների բլոկ: Մեծ քանակի համար, ապա ոչ էլ չափի թվով, անհրաժեշտ է օգտագործել գաղտնագրային ռեժիմ: 34.13-2015–ում նկարագրված են հետևյալները: ECB, CTR, OFB, CBC, CFB: Յուրաքանչյուր հենց իր թույլատրված օգտագործման ոլորտներն ու հատկությունները: Շատ մեծ ցավ է, որ մենք դեռ չունենք ստանդարտացված (օրինակ՝ CCM, OCB, GCM և նրանց նման) — մենք ստիպված ենք ինքնուրույն, понее, ավելացնել MAC։ Ես ընտրում եմ (CTR): նա չի պահանջում լրացում բլոկի չափին, կարող է նաև զուգահեռ գործել, օգտագործում է միայն ծածկագրման ֆունկցիան, անվտանգ կարող է օգտագործվել մեծ քանակությամբ հաղորդագրությունների ծածկագրման համար (CBC-ի համեմատ, որտեղ հարաբերականապես արագ սկսվում են հարթեցումները):
Ինչպես և .mac(), .ctr() ընդունում է նման տվյալներ՝ մուտքում։ ciphertext = gost3413.ctr(GOST3412Kuznechik(key).encrypt, KUZNECHIK_BLOCKSIZE, plaintext, iv): Պահանջվում է որոշել սկզբնական վեկտորը, որը ունի ճիշտ կես բլոկի երկարություն։ Եթե մեր ծածկագրումը օգտագործվում է միայն մեկ հաղորդագրության (թեկուզ և մի քանի բլոկներից) համար, ապա կարելի է անվտանգ կիրառել զրո սկզբնական վեկտոր։ Քանի որ մեզ համար հHandshake հաղորդագրությունները օգտագործում ենք յուրաքաչուր տարբեր բանալի։
gost3410.verify() նշակի ստուգումը պարզ է։ փոխանցում ենք էլիպտիկ ծալքի շրջանակներում, որով աշխատում ենք (այս ծալքը մենք պարզապես фиксируем ենք մեր GOSTIM պրոտոկոլում), կնքողի հանրային բանալին (չմոռանանք, որ սա պետք է լինի երկու մեծ թվերի_tuple, ոչ թե բայթային տող), 34.11-2012 համարժեքը և ինքը ստացող նշանը։
Քանի որ, նախաձեռնող մենք պատրաստում ենք և ուղարկում handshake2 հաղորդակցման նշանը, կատարելով այն նույն գործողությունները, ինչ մենք արեցինք ստուգման ժամանակ, պարզապես սիմետրիկ։ Մեր բանալիների վրա նշակը փոխանակելու փոխարեն, և այլն…
456 # Պատրաստել և ուղարկել 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: նստաշրջանը հաստատված է: %s", _id, peer_name)Երբ նստաշրջանը հաստատվի, արտադրվում են փոխադրումների բանալիները (միավորման բանալի, դեմքի բանալի, յուրաքանչյուր կողմի համար), և յուրացված է Կուզնիչկը՝ ծածկագրման և MAC-ի ստուգման համար:
499 # Տեքստի հաղորդագրություն ուղարկող, տրանսպորտի 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 # Տեսակների ստուգման սպասում {{{ 521 while True: 522 data = await reader.read(MaxMsgLen) ... 530 msg, tail = Msg().decode(buf) ... 537 try: 538 await msg_receiver( 539 msg.value, 540 nonce_expected, 541 macer, 542 encrypter, 543 peer_name, 544 ) 545 except ValueError as err: 546 logging.warning("%s: %s", err) 547 break 548 nonce_expected += 1 549 # }}}msg_sender coroutine-ը այժմ անվտանգ հաղորդագրություններ է encrypt անում, նախքան TCP կապի մեջ ուղարկելը: Ամեն հաղորդագրություն ունի մոնոտոնորեն աճող nonce, որը նաև հանդիսանում է ներմուծման վեկտոր ի շիֆրման համար հաշվողական ռեժիմում: Ամեն հաղորդագրության և հաղորդագրությունների բլոկի համար երաշխավորված են տարբեր հաշվիչ արժեքներ:
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 += 1Գործող հաղորդագրությունները մշակվում են msg_receiver coroutine-ով, որը զբաղվում է վավերացումը և հրահանգումը:
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("անհասանելի nonce արժեք") mac_tag = mac(macer, KUZNECHIK_BLOCKSIZE, payload.encode()) if not compare_digest(mac_tag, bytes(msg_text["payloadMac"])): raise ValueError("ոչ վավեր 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)Ավարտ
GOSTIM-ը նախատեսված է բացառապես ուսումնական նպատակներով (քանի որ չի ծածկաոդվում թեստերով, առնվազն)! Ծրագրի աղբյուրը կարող եք ներբեռնել (Ստրիբոգ-256 hash: 995bbd368c04e50a481d138c5fa2e43ec7c89bc77743ba8dbabee1fde45de120). Ինչպես բոլոր իմ նախագծերը, ինչպես , , , , GOSTIM-ը ինտեգրված է ամբողջությամբ , տարածվում է պայմաններով .
, , անդամ , Python/Go-разработчик, գլխավոր մասնագետ .
Ընտանիք: habr.com
