shallow's blog
0%

[bytesctf2020 final]Crypto WP

发表于 更新于 分类于

文章首发于安全客

去年去北京打的bytesctf线下，质量还是挺高的。ak很爽

impersonation

Alice：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from KeyExchange import KeyExchange
from gmssl import func
from flag import FLAG
import socket
import time
import signal
import sys

sys.stderr = open('/dev/null', 'w')
signal.alarm(5)

n = 'FFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFF7203DF6B21C6052B53BBF40939D54123'
public_key_B = 'e52f035c340267a2ee2c57de87db9acf443d1fb98f0b7abbc55d9f332f4f823e' \
               '0f81e7dde971b1e4d02981fc5741eb30f71bf6bcd0c02b06e5c857eedc58cae5'

G = '32c4ae2c1f1981195f9904466a39c9948fe30bbff2660be1715a4589334c74c7' \
    'bc3736a2f4f6779c59bdcee36b692153d0a9877cc62a474002df32e52139f0a0'

print("Hi, Can you tell me bob's address?")
print("I have a message to tell him.")
try:
    addr = input('>>>').strip()
    ip, port = [x.strip() for x in addr.split(':')]
    port = int(port)
except:
    ip, port = 'bob', 1337

print(f"OK! {ip}:{port}, I got it!")
try:
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((ip, port))
    data = s.recv(1024).strip()
    assert data == b'Hi, Alice?'
except:
    print("Oh noooooo! Where did Bob go?")
    exit()

KE = KeyExchange()
r_str = func.random_hex(len(n))
P_a, T_a = KE.send(r_str)
s.sendall(f'{P_a.zfill(128)}{T_a}\n'.encode())
time.sleep(0.1)
data = s.recv(1024).strip().decode()
P_b, T_b = data[:128],  data[128:]
if P_b != public_key_B:
    print("Oh noooooo! Fake Bob!")
    exit()
msg = KE.transport(P_b, T_b, FLAG)
s.sendall(f'{msg}\n'.encode())
time.sleep(0.1)
data = s.recv(1024).strip().decode()
msg = KE.decrypt(data)
if FLAG == msg:
    print("My feelings are transmitted, Bye~")
else:
    print("Oh noooooo! Fake Bob!")
exit()

中间人攻击

场景大概是Alice想与Bob通信，可以通过本地起一个socket当服务器，给他自己的ip, port后再连接Bob即可拦截与修改信息，进行中间人攻击。很有趣的背景，这个场景是有一定现实意义的，由于网站都有证书，所以Alice作为一个用户往往能够判断对方的公钥正确性，而服务器却不行，而若密钥协商协议实现有问题，则很有可能仍然有安全隐患。

Bob的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from KeyExchange import KeyExchange
from gmssl import func
import signal
import sys
sys.stderr = open('/dev/null', 'w')
signal.alarm(5)

n = 'FFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFF7203DF6B21C6052B53BBF40939D54123'
private_key = '????????????????????????????????????????????????????????????????'
public_key = 'e52f035c340267a2ee2c57de87db9acf443d1fb98f0b7abbc55d9f332f4f823e' \
             '0f81e7dde971b1e4d02981fc5741eb30f71bf6bcd0c02b06e5c857eedc58cae5'

KE = KeyExchange(public_key, private_key)
print('Hi, Alice?')
data = input().strip()
P_a, T_a = data[:128],  data[128:]
r_str = func.random_hex(len(n))
P_b, T_b = KE.respond(r_str, P_a, T_a)
print(f'{P_b.zfill(128)}{T_b}')
data = input().strip()
msg = KE.reencrypt(data)
print(msg)

KeyExchange是一个修改过的ecdh，交换后Alice通过OTP发送flag，Bob再通过Alice的公钥发回。

主要的困难（与教科书最基本的中间人攻击的区别）是Alice知道Bob的公钥，无法给她自己的公钥来实现转发。

而可能的漏洞有三个：

  • Bob不知道Alice的公钥
  • Alice有对flag的decrypt功能，并且有回显
  • Alice与Bob均不对点进行检查

对于第二个漏洞，需要能找到一个改flag一位的地方，是初赛中类似题利用的漏洞。但是这次并没有这样的漏洞，加密时用与一个随机数异或，没办法确定对或者错对应的flag位的值。

因此先考虑第一点与第三点的利用

KeyExchange的流程是

  • Alice -> Bob : hash(ra * G) * AlicePublickey + ra * G
  • Bob -> Alice : hash(rb * G) * BobPublickey + rb * G
1
2
3
4
5
(hash(rb * G) * BobPrivatekey + rb) * (hash(ra * G) * AlicePublickey + ra * G)

= (hash(rb * G) * BobPrivatekey + rb) * (hash(ra * G) * AlicePrivatekey + ra) * G

= (hash(rb * G) * BobPublickey + rb * G) * (hash(ra * G) * AlicePrivatekey + ra)

而这个共享密钥是通过这个点经过SM3得到，两个不同的点得到的密钥之间没有任何关系

因此在消息转发时要让Alice与Bob计算的共享密钥需相同，并且由于Alice的密钥每次更换，若不知道给Bob发送的公钥对应的私钥，也不可能解密成功。

我们可控的点为：

  • 给Alice的Bob rb
  • 给Bob的Alice ra
  • 给Bob的Alice publickey

因为Bob的公钥无法改变，给alice发送的消息只有rb可控，但由于还要进行哈希计算，所以难以控制，所以尝试直接将Bob发送的rb转发，试图通过给Bob的ra与publickey来实现相等

则目标就是

1
hash(ra * G) * AlicePublickey + ra * G = hash(R) * Publickey + R

（以下为我在比赛时第一时间的想法）

将左边看作一个整体点P

若先选取R，则

1
Publickey = (P - R) * invert(hash(R) , n) = xG

但无法知道x的值，则最后一步Bob发送的flag无法解密。

若先选取Publickey，几乎没法确定R。

所以想到找低阶点，若Publickey的阶低，则x可以通过爆破得到，同时，也可以通过爆破来确定R。

1
R = P - i * Publickey

如果hash(R) % order = i，则能得到满足条件的R。

对于选取的一个低阶点，计算order次，碰撞不成功的概率为(1 - 1/order)^order > 1/e = 0.37

所以大概成功率会大于60%

但SM2使用的曲线安全性很高，阶为素数，无法在这条曲线上找低阶点。

但是Bob并不检查点的合法性，可以寻找不在曲线上的低阶点

寻找方法为将b+1,计算一个点g的阶，若阶order有小质因子k，则计算(order // k) * g 则为想要的k阶点

这样，虽然是两个不在一条曲线上的点运算，但是它也一定能算出一个结果R，只不过这个结果R在第三条曲线上，但这里的R仅仅只用到了这一次，因此不会有其他麻烦。

而Publickey还会用于下面对flag的加密，用不在曲线上的点仍能够实现加密，但是解密时就得不到k* Publickey了

原本解密时，通过privatekeyC1可以得到加密使用的点，但因为我们的Publickey根本就不是从G得到的，因此无法直接得到加密使用的点。但由于Publickey的阶并不大，k Publickey的可能值仅有order个，爆破order次，即可从中找到明文。

exp：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
import socketserver
import os, signal
import threading
import string, binascii
from hashlib import sha256
from Crypto.Util import number
from KeyExchange import KeyExchange
from pwn import *
from gmssl import func, sm3
import string
KE = KeyExchange()
def sm3_hash_str(msg):
    return sm3.sm3_hash(func.bytes_to_list(msg.encode()))

def to_point(x , y):
    return hex(x)[2:].rjust(64 , '0') + hex(y)[2:].rjust(64 , '0')
def to_int(g):
    x = int(g[:64] , 16)
    y = int(g[64:] , 16)
    return x , y
def fu(g):
    x ,y = to_int(g)
    y = 0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFF - y
    return to_point(x , y)
class Task(socketserver.BaseRequestHandler):
    index = 0
    times = 0
    _key = None
    _rand_iter = None

    def _recvall(self):
        data = self.request.recv(1024)
        return data.strip()

    def send(self, msg, newline=True):
        if newline:
            msg += b'\n'
        self.request.sendall(msg)

    def get_target(self,peer_pk , peer_T):
        peer_h = int(sm3_hash_str(peer_T), 16) % int(KE.ecc_table['n'], base=16)
        peer_hP = KE._kg(peer_h, peer_pk)
        peer_ThP = KE._add_point(peer_T, peer_hP)
        peer_ThP = KE._convert_jacb_to_nor(peer_ThP)
        return peer_ThP
    def attack(self, tempg):
        tempx = 108369860142849285519512209608941191631731475365137473063138228543102305365193
        tempy = 17711224374516220167029583974392567739229554851195153447898777402617425143845
        temp = to_point(tempx , tempy)
        for i in range(1783):
            
            tempc = KE._kg(i , temp)
            tempc = fu(tempc)
            r = KE._convert_jacb_to_nor(KE._add_point(tempg , tempc))
            if int(sm3_hash_str(r), 16) % int(KE.ecc_table['n'], base=16) % 1783 == i:
                if KE._kg(int(sm3_hash_str(r), 16) % int(KE.ecc_table['n'], base=16), temp) != None:
                    rg = KE._convert_jacb_to_nor(KE._add_point(r , KE._kg(int(sm3_hash_str(r), 16) % int(KE.ecc_table['n'], base=16), temp)))
                    break
        assert self.get_target(temp , r) == tempg
        return temp , r
    def decrypt(msg):
        tempx = 108369860142849285519512209608941191631731475365137473063138228543102305365193
        tempy = 17711224374516220167029583974392567739229554851195153447898777402617425143845
        temp = to_point(tempx , tempy)
        len_2 = 64 * 2
        len_3 = len_2 + 64
        C2 = msg[len_3:]
        for k in range(1783):
            xy = KE._kg(k, temp)
            if xy != None:
                x2 = xy[0:64]
                y2 = xy[64:2 * 64]
                ml = len(C2)
                t = sm3.sm3_kdf(xy.encode('utf8'), ml / 2)
                temp = number.long_to_bytes(int(C2 , 16) ^ int(t , 16))
                if temp[0] == ord('B'):
                    print(temp)
    def recv(self, prompt=b''):
        self.send(prompt, newline=False)
        return self._recvall()
    def handle(self):
            self.send(b'Hi, Alice?')
            p = remote('172.16.9.45' , 13334)
            data = self._recvall()
            print('Alice-sl' , data)
            P_a, T_a = data[:128].decode(),  data[128:].decode()
            Alice_THP = self.get_target(P_a , T_a)

            P_sl , T_sl = self.attack(Alice_THP)

            p.recvline()
            p.sendline(P_sl + T_sl)
            print('sl-Bob',P_sl + T_sl)
            
            
            temp = p.recvline()[:-1]
            print('Bob-sl',temp)
            
            
            self.send(temp)
            print('sl-Alice' , temp)
            
            
            msg = self._recvall()
            print('Alice-sl' , msg)
            
            p.sendline(msg)
            print('sl-Bob' , msg)
            temp = p.recvline()
            print('Bob-sl' , temp)
            self.decrypt(temp)

            self.request.close()
            p.close()

class ForkedServer(socketserver.ForkingMixIn, socketserver.TCPServer):
    pass

if __name__ == "__main__":
    HOST, PORT = '0.0.0.0', 1118
    server = ForkedServer((HOST, PORT), Task)
    server.allow_reuse_address = True
    server.serve_forever()

但这种思路看着太乱了，完全不按题目正常的流程走，比赛时打了一整页草稿才写出来。 但是赛后跟V师傅交流了一下，发现上面有一步出了问题 当先任意选取R时，

1
Publickey = (P - R) * invert(hash(R) , n) = xG

这条式子其实是可以算出x的 最初认为P – R是一个未知的值，但是完全可以取R = P – G 则x = invert(hash(R) , n) 接着直接解密即可。

agid

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
import agid
import re
from gmssl.optimized_field_elements import FQ2, FQ


'''
def check_auth():
    bob = agid.Verifier()
    identity = bob.allow_identity[0]
    alice = agid.Prover(identity)
    Ppub, Qpub, H = bob.setup()
    Rid = bob.keygen(identity)
    V = bob.gpkgen()
    W = bob.gskgen(identity)
    U1, U2, U3, S1, S2, T1, T2, T3 = alice.commit(Rid, W, Ppub, Qpub, H)
    c = bob.challenge(U1, U2, U3, S1, S2, T1, T2, T3)
    s1, s2, s3, s4 = alice.prove(c)
    assert bob.verify(s1, s2, s3, s4)
check_server()
'''


def safe_eval(data):
    data = data.replace(' ', '').strip()
    if re.match(r"^[0-9,()\[\]]+$", data) is None:
        print('This is just for simpler data transmission,\nPlease do not consider attacking here.')
        exit()
    try:
        # This is just for simpler data transmission, please do not consider attacking here
        return eval(data)
    except:
        print('eval fault.')
    exit()


server = agid.Verifier()

Ppub, Qpub, H = server.setup()
V = server.gpkgen()
print((Ppub, Qpub, H, V))


print("Give me your identity")
identity = input(">>>").strip()
Rid = server.keygen(identity)
W = server.gskgen(identity)
print((Rid, W))

print("Give me U1, U2, U3, S1, S2, T1, T2, T3")
try:
    U1, U2, U3, S1, S2, T1, T2, T3 = [[FQ2(y) if type(y) is list else FQ(y) for y in x] for x in safe_eval(input(">>>"))]
except:
    print('eval error.')
    exit()

c = server.challenge(U1, U2, U3, S1, S2, T1, T2, T3)
print(c)


print("Give me s1, s2, s3, s4")
data = input(">>>")
try:
    s1, s2, s3, s4 = safe_eval(data)
except:
    print('format error.')
    exit()
if server.verify(s1, s2, s3, s4):
    print(FLAG)
else:
    print('NoNoNo...')

题目中用的gmssl并不是pip直接下载的那个，需要去github找另一个

题目是一个认证系统，如果有identity即可多得到两个数据

上面check_auth为正常的认证流程

我们需要做的是没有identity的情况下认证成功

看上去需要给12个变量，满足5个方程。十分复杂。

但实际上全写出来，真的很简单

1
2
3
eq1_left = ate.pairing(U2, U1)
eq1_right = ate.pairing(ec.G2, ec.G1)
assert eq1_left == eq1_right

虽然有双线性对，一开始还以为需要用到双线性对的限制，但是因为懒，就直接U2 = G2 ， U1 = G1就完事了

1
2
3
eq2_left = ate.pairing(S2, S1)
eq2_right = ate.pairing(self.V, ec.G1)
assert eq2_left == eq2_right

第二个同理，S2 = V ， S1 = G1

下面的则是要求

1
2
3
T1 = (s1 + s2*sm + h*s3)G1 - c * U3
T2 = (s3 * h)G1 + s4 * U1 - c * U3 = (s3 * h + s4)G1 - c * U3
T3 = (s1 + s * s2)G1 - c * S1 = (s1 + s * s2 - c)G1

由于预先不知道c的值，所以在T的构造里必须把c全消掉。

点为0可能出现一些神秘问题（上一道题是这样的，这题库不一样，也不知道会不会）。但是不为0完全也能构造

首先T3处，s未知，需要用s2 * Qpub 得到 s * s2 * G1，取s1 = c抵消c。s2可任意

因此T3 = s2 * Ppub

T2处方便起见U3 = G1，s3 * h G1 用 s3 H得到，同样让s4 = c抵消c，s3可任意

因此T2 = s3 * H

T1处几个参数已固定

T1 = s3 * H + s2 * Qpub

然后任意取值即可

exp中任意参数全取了1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import gmssl.optimized_curve as ec
import gmssl.optimized_pairing as ate
import agid
import re
from gmssl.optimized_field_elements import FQ2, FQ
from pwn import *
p = remote('172.16.9.45' , 18585)
recv = p.recvline()[:-1]
Ppub , Qpub , H , V = [[FQ2(y) if type(y) is list else FQ(y) for y in x] for x in eval(recv)]
p.recvuntil('>>>')
p.sendline('1')
p.recvuntil('>>>')
U1 = ec.G1
U2 = ec.G2
U3 = ec.G1
S1 = ec.G1
S2 = V
s2 = 1
s3 = 1
T3 = ec.multiply(Ppub , s2)
T2 = ec.multiply(H , s3)
T1 = ec.add(ec.multiply(Qpub , s2) , T2)
temp = (U1,U2,U3,S1,S2,T1,T2,T3)
print(str(temp))
p.sendline(str(temp))
c = int(p.recvline()[:-1])
s1 = c
s4 = c
slist = (s1,s2,s3,s4)
p.sendline(str(slist))
p.interactive()