ccyj/mimajingsai
1
0
Fork 0
forked from sangge/tpre-python
Code Pull Requests Actions Activity

refactor: refactor: 添加f函数注释；f、Encapsulate增加模P

Browse Source
This commit is contained in:
dqy 2023-10-18 16:59:22 +08:00
parent e4ce861ec7
commit c8975a104b
3 changed files with 258 additions and 222 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -6,4 +6,3 @@ test.py
example.py example.py
ReEncrypt.py ReEncrypt.py
src/demo.py

31
src/demo.py Normal file
View File

@ -0,0 +1,31 @@
from tpre import *
# 1
pk_a, sk_a = GenerateKeyPair(1, ())
m = b'hello world'
m = int.from_bytes(m)
# 2
capsule_ct = Encrypt(pk_a, m)
# 3
pk_b, sk_b = GenerateKeyPair(1, ())
N = 20
T = 10
# 5
rekeys = GenerateReKey(sk_a, pk_b, N, T)
# 7
cfrag_cts = []
for rekey in rekeys:
cfrag_ct = ReEncrypt(rekey, capsule_ct)
cfrag_cts.append(cfrag_ct)
# 9
cfrags = mergecfrag(cfrag_cts)
m = DecryptFrags(sk_b, pk_b, pk_a, cfrags)

448
src/tpre.py
View File

@ -1,6 +1,11 @@
from gmssl import * #pylint: disable = e0401 from gmssl import * # pylint: disable = e0401
from typing import Tuple, Callable from typing import Tuple, Callable
import random import random
import traceback
point = Tuple[int, int]
capsule = Tuple[point, point, int]
# 生成密钥对模块 # 生成密钥对模块
class CurveFp: class CurveFp:
@ -12,7 +17,8 @@ class CurveFp:
self.Gx = Gx self.Gx = Gx
self.Gy = Gy self.Gy = Gy
self.name = name self.name = name
sm2p256v1 = CurveFp( sm2p256v1 = CurveFp(
name="sm2p256v1", name="sm2p256v1",
A=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFC, A=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFC,
@ -20,7 +26,7 @@ sm2p256v1 = CurveFp(
P=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFF, P=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFF,
N=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFF7203DF6B21C6052B53BBF40939D54123, N=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFF7203DF6B21C6052B53BBF40939D54123,
Gx=0x32C4AE2C1F1981195F9904466A39C9948FE30BBFF2660BE1715A4589334C74C7, Gx=0x32C4AE2C1F1981195F9904466A39C9948FE30BBFF2660BE1715A4589334C74C7,
Gy=0xBC3736A2F4F6779C59BDCEE36B692153D0A9877CC62A474002DF32E52139F0A0 Gy=0xBC3736A2F4F6779C59BDCEE36B692153D0A9877CC62A474002DF32E52139F0A0,
) )
# 椭圆曲线 # 椭圆曲线
@ -29,17 +35,20 @@ G = sm2p256v1
# 生成元 # 生成元
g = (sm2p256v1.Gx, sm2p256v1.Gy) g = (sm2p256v1.Gx, sm2p256v1.Gy)
def multiply(a: Tuple[int, int], n: int) -> Tuple[int, int]:
def multiply(a: point, n: int) -> point:
N = sm2p256v1.N N = sm2p256v1.N
A = sm2p256v1.A A = sm2p256v1.A
P = sm2p256v1.P P = sm2p256v1.P
return fromJacobian(jacobianMultiply(toJacobian(a), n, N, A, P), P) return fromJacobian(jacobianMultiply(toJacobian(a), n, N, A, P), P)
def add(a: Tuple[int, int], b: Tuple[int, int]) -> Tuple[int, int]:
def add(a: point, b: point) -> point:
A = sm2p256v1.A A = sm2p256v1.A
P = sm2p256v1.P P = sm2p256v1.P
return fromJacobian(jacobianAdd(toJacobian(a), toJacobian(b), A, P), P) return fromJacobian(jacobianAdd(toJacobian(a), toJacobian(b), A, P), P)
def inv(a: int, n: int) -> int: def inv(a: int, n: int) -> int:
if a == 0: if a == 0:
return 0 return 0
@ -50,33 +59,36 @@ def inv(a: int, n: int) -> int:
nm, new = hm - lm * r, high - low * r nm, new = hm - lm * r, high - low * r
lm, low, hm, high = nm, new, lm, low lm, low, hm, high = nm, new, lm, low
return lm % n return lm % n
def toJacobian(Xp_Yp: Tuple[int, int]) -> Tuple[int, int, int]:
def toJacobian(Xp_Yp: point) -> Tuple[int, int, int]:
Xp, Yp = Xp_Yp Xp, Yp = Xp_Yp
return (Xp, Yp, 1) return (Xp, Yp, 1)
def fromJacobian(Xp_Yp_Zp: Tuple[int, int, int], P: int) -> Tuple[int, int]:
def fromJacobian(Xp_Yp_Zp: Tuple[int, int, int], P: int) -> point:
Xp, Yp, Zp = Xp_Yp_Zp Xp, Yp, Zp = Xp_Yp_Zp
z = inv(Zp, P) z = inv(Zp, P)
return ((Xp * z ** 2) % P, (Yp * z ** 3) % P) return ((Xp * z**2) % P, (Yp * z**3) % P)
def jacobianDouble(Xp_Yp_Zp: Tuple[int, int, int], A: int, P: int) -> Tuple[int, int, int]:
def jacobianDouble(
Xp_Yp_Zp: Tuple[int, int, int], A: int, P: int
) -> Tuple[int, int, int]:
Xp, Yp, Zp = Xp_Yp_Zp Xp, Yp, Zp = Xp_Yp_Zp
if not Yp: if not Yp:
return (0, 0, 0) return (0, 0, 0)
ysq = (Yp ** 2) % P ysq = (Yp**2) % P
S = (4 * Xp * ysq) % P S = (4 * Xp * ysq) % P
M = (3 * Xp ** 2 + A * Zp ** 4) % P M = (3 * Xp**2 + A * Zp**4) % P
nx = (M ** 2 - 2 * S) % P nx = (M**2 - 2 * S) % P
ny = (M * (S - nx) - 8 * ysq ** 2) % P ny = (M * (S - nx) - 8 * ysq**2) % P
nz = (2 * Yp * Zp) % P nz = (2 * Yp * Zp) % P
return (nx, ny, nz) return (nx, ny, nz)
def jacobianAdd( def jacobianAdd(
Xp_Yp_Zp: Tuple[int, int, int], Xp_Yp_Zp: Tuple[int, int, int], Xq_Yq_Zq: Tuple[int, int, int], A: int, P: int
Xq_Yq_Zq: Tuple[int, int, int],
A: int,
P: int
) -> Tuple[int, int, int]: ) -> Tuple[int, int, int]:
Xp, Yp, Zp = Xp_Yp_Zp Xp, Yp, Zp = Xp_Yp_Zp
Xq, Yq, Zq = Xq_Yq_Zq Xq, Yq, Zq = Xq_Yq_Zq
@ -84,10 +96,10 @@ def jacobianAdd(
return (Xq, Yq, Zq) return (Xq, Yq, Zq)
if not Yq: if not Yq:
return (Xp, Yp, Zp) return (Xp, Yp, Zp)
U1 = (Xp * Zq ** 2) % P U1 = (Xp * Zq**2) % P
U2 = (Xq * Zp ** 2) % P U2 = (Xq * Zp**2) % P
S1 = (Yp * Zq ** 3) % P S1 = (Yp * Zq**3) % P
S2 = (Yq * Zp ** 3) % P S2 = (Yq * Zp**3) % P
if U1 == U2: if U1 == U2:
if S1 != S2: if S1 != S2:
return (0, 0, 1) return (0, 0, 1)
@ -97,19 +109,15 @@ def jacobianAdd(
H2 = (H * H) % P H2 = (H * H) % P
H3 = (H * H2) % P H3 = (H * H2) % P
U1H2 = (U1 * H2) % P U1H2 = (U1 * H2) % P
nx = (R ** 2 - H3 - 2 * U1H2) % P nx = (R**2 - H3 - 2 * U1H2) % P
ny = (R * (U1H2 - nx) - S1 * H3) % P ny = (R * (U1H2 - nx) - S1 * H3) % P
nz = (H * Zp * Zq) % P nz = (H * Zp * Zq) % P
return (nx, ny, nz) return (nx, ny, nz)
def jacobianMultiply(
Xp_Yp_Zp: Tuple[int, int, int],
n: int,
N: int,
A: int,
P: int
) -> Tuple[int, int, int]:
def jacobianMultiply(
Xp_Yp_Zp: Tuple[int, int, int], n: int, N: int, A: int, P: int
) -> Tuple[int, int, int]:
Xp, Yp, Zp = Xp_Yp_Zp Xp, Yp, Zp = Xp_Yp_Zp
if Yp == 0 or n == 0: if Yp == 0 or n == 0:
return (0, 0, 1) return (0, 0, 1)
@ -120,159 +128,147 @@ def jacobianMultiply(
if (n % 2) == 0: if (n % 2) == 0:
return jacobianDouble(jacobianMultiply((Xp, Yp, Zp), n // 2, N, A, P), A, P) return jacobianDouble(jacobianMultiply((Xp, Yp, Zp), n // 2, N, A, P), A, P)
if (n % 2) == 1: if (n % 2) == 1:
return jacobianAdd(jacobianDouble(jacobianMultiply((Xp, Yp, Zp), n // 2, N, A, P), A, P), (Xp, Yp, Zp), A, P) return jacobianAdd(
jacobianDouble(jacobianMultiply((Xp, Yp, Zp), n // 2, N, A, P), A, P),
(Xp, Yp, Zp),
A,
P,
)
raise ValueError("jacobian Multiply error") raise ValueError("jacobian Multiply error")
# 生成元
# 生成元
U = multiply(g, random.randint(0, sm2p256v1.P)) U = multiply(g, random.randint(0, sm2p256v1.P))
# def Setup(sec: int) -> Tuple[CurveFp, Tuple[int, int],
# Tuple[int, int]]:
# '''
# params:
# sec: an init safety param
# return:
# G: sm2 curve
# g: generator
# U: another generator
# '''
# G = sm2p256v1
# g = (sm2p256v1.Gx, sm2p256v1.Gy)
# tmp_u = random.randint(0, sm2p256v1.P)
# U = multiply(g, tmp_u)
# return G, g, U
def hash2(double_G: Tuple[Tuple[int, int], Tuple[int, int]]) -> int: def hash2(double_G: Tuple[point, point]) -> int:
sm3 = Sm3() #pylint: disable=e0602 sm3 = Sm3() # pylint: disable=e0602
for i in double_G: for i in double_G:
for j in i: for j in i:
sm3.update(j.to_bytes(32)) sm3.update(j.to_bytes(32))
digest = sm3.digest() digest = sm3.digest()
digest = int.from_bytes(digest,'big') % sm2p256v1.P digest = int.from_bytes(digest, "big") % sm2p256v1.P
return digest return digest
def hash3(triple_G: Tuple[Tuple[int, int],
Tuple[int, int], def hash3(triple_G: Tuple[point, point, point]) -> int:
Tuple[int, int]]) -> int: sm3 = Sm3() # pylint: disable=e0602
sm3 = Sm3() #pylint: disable=e0602
for i in triple_G: for i in triple_G:
for j in i: for j in i:
sm3.update(j.to_bytes(32)) sm3.update(j.to_bytes(32))
digest = sm3.digest() digest = sm3.digest()
digest = int.from_bytes(digest, 'big') % sm2p256v1.P digest = int.from_bytes(digest, "big") % sm2p256v1.P
return digest return digest
def hash4(triple_G: Tuple[Tuple[int, int],
Tuple[int, int], def hash4(triple_G: Tuple[point, point, point], Zp: int) -> int:
Tuple[int, int]], sm3 = Sm3() # pylint: disable=e0602
Zp: int) -> int:
sm3 = Sm3() #pylint: disable=e0602
for i in triple_G: for i in triple_G:
for j in i: for j in i:
sm3.update(j.to_bytes(32)) sm3.update(j.to_bytes(32))
sm3.update(Zp.to_bytes(32)) sm3.update(Zp.to_bytes(32))
digest = sm3.digest() digest = sm3.digest()
digest = int.from_bytes(digest, 'big') % sm2p256v1.P digest = int.from_bytes(digest, "big") % sm2p256v1.P
return digest return digest
def KDF(G: Tuple[int, int]) -> int:
sm3 = Sm3() #pylint: disable=e0602 def KDF(G: point) -> int:
sm3 = Sm3() # pylint: disable=e0602
print(G)
for i in G: for i in G:
sm3.update(i.to_bytes(32)) sm3.update(i.to_bytes(32))
digest = sm3.digest(32) digest = sm3.digest()
digest = digest digest = digest
digest = int.from_bytes(digest, 'big') % sm2p256v1.P digest = int.from_bytes(digest, "big") % sm2p256v1.P
mask_128bit = (1 << 128) - 1
digest = digest & mask_128bit
print("key =", digest)
traceback.print_stack()
return digest return digest
def GenerateKeyPair(
lamda_parma: int, def GenerateKeyPair(lamda_parma: int, public_params: tuple) -> Tuple[point, int]:
public_params: tuple """
) -> Tuple[Tuple[int, int], int]: params:
''' lamda_param: an init safety param
params:
lamda_param: an init safety param
public_params: curve params public_params: curve params
return: return:
public_key, secret_key public_key, secret_key
''' """
sm2 = Sm2Key() #pylint: disable=e0602 sm2 = Sm2Key() # pylint: disable=e0602
sm2.generate_key() sm2.generate_key()
public_key_x = int.from_bytes(bytes(sm2.public_key.x),"big") public_key_x = int.from_bytes(bytes(sm2.public_key.x), "big")
public_key_y = int.from_bytes(bytes(sm2.public_key.y),"big") public_key_y = int.from_bytes(bytes(sm2.public_key.y), "big")
public_key = (public_key_x, public_key_y) public_key = (public_key_x, public_key_y)
secret_key = int.from_bytes(bytes(sm2.private_key), "big")
secret_key = int.from_bytes(bytes(sm2.private_key),"big")
return public_key, secret_key return public_key, secret_key
# 生成A和B的公钥和私钥
pk_A, sk_A = GenerateKeyPair(0, ())
pk_B, sk_B = GenerateKeyPair(0, ())
def Encrypt(pk: Tuple[int, int], m: int) -> Tuple[Tuple[ # 生成A和B的公钥和私钥
Tuple[int, int],Tuple[int, int], int], int]: # pk_A, sk_A = GenerateKeyPair(0, ())
# pk_B, sk_B = GenerateKeyPair(0, ())
def Encrypt(pk: point, m: bytes) -> Tuple[capsule, bytes]:
enca = Encapsulate(pk) enca = Encapsulate(pk)
K = enca[0].to_bytes() K = enca[0].to_bytes(16)
capsule = enca[1] capsule = enca[1]
if len(K) != 16: if len(K) != 16:
raise ValueError("invalid key length") raise ValueError("invalid key length")
iv = b'tpretpretpretpre' iv = b"tpretpretpretpre"
sm4_enc = Sm4Cbc(K, iv, DO_ENCRYPT) #pylint: disable=e0602 sm4_enc = Sm4Cbc(K, iv, DO_ENCRYPT) # pylint: disable=e0602
plain_Data = m.to_bytes(32) enc_Data = sm4_enc.update(m)
enc_Data = sm4_enc.update(plain_Data)
enc_Data += sm4_enc.finish() enc_Data += sm4_enc.finish()
enc_message = (capsule, enc_Data) enc_message = (capsule, enc_Data)
return enc_message return enc_message
def Decapsulate(ska:int,capsule:Tuple[Tuple[int,int],Tuple[int,int],int]) -> int:
E,V,s = capsule def Decapsulate(ska: int, capsule: capsule) -> int:
EVa=multiply(add(E,V), ska) # (E*V)^ska E, V, s = capsule
EVa = multiply(add(E, V), ska) # (E*V)^ska
K = KDF(EVa) K = KDF(EVa)
return K return K
def Decrypt(sk_A: int,C:Tuple[Tuple[
Tuple[int, int],Tuple[int, int], int], int]) ->int: def Decrypt(sk_A: int, C: Tuple[Tuple[point, point, int], bytes]) -> int:
''' """
params: params:
sk_A: secret key sk_A: secret key
C: (capsule, enc_data) C: (capsule, enc_data)
''' """
capsule,enc_Data = C capsule, enc_Data = C
K = Decapsulate(sk_A,capsule) K = Decapsulate(sk_A, capsule)
iv = b'tpretpretpretpre' iv = b"tpretpretpretpre"
sm4_dec = Sm4Cbc(K, iv, DO_DECRYPT) #pylint: disable= e0602 sm4_dec = Sm4Cbc(K, iv, DO_DECRYPT) # pylint: disable= e0602
dec_Data = sm4_dec.update(enc_Data) dec_Data = sm4_dec.update(enc_Data)
dec_Data += sm4_dec.finish() dec_Data += sm4_dec.finish()
return dec_Data return dec_Data
# GenerateRekey # GenerateRekey
def H5(id: int, D: int) -> int: def hash5(id: int, D: int) -> int:
sm3 = Sm3() #pylint: disable=e0602 sm3 = Sm3() # pylint: disable=e0602
sm3.update(id.to_bytes(32)) sm3.update(id.to_bytes(32))
sm3.update(D.to_bytes(32)) sm3.update(D.to_bytes(32))
hash = sm3.digest() hash = sm3.digest()
hash = int.from_bytes(hash,'big') % G.P hash = int.from_bytes(hash, "big") % G.P
return hash
def hash6(triple_G: Tuple[point, point, point]) -> int:
sm3 = Sm3() # pylint: disable=e0602
for i in triple_G:
for j in i:
sm3.update(j.to_bytes(32))
hash = sm3.digest()
hash = int.from_bytes(hash, "big") % G.P
return hash return hash
def H6(triple_G: Tuple[Tuple[int, int],
Tuple[int, int],
Tuple[int, int]]) -> int:
sm3 = Sm3() #pylint: disable=e0602
for i in triple_G:
for j in i:
sm3.update(j.to_bytes(32))
hash = sm3.digest()
hash = int.from_bytes(hash,'big') % G.P
return hash
def f(x: int, f_modulus: list, T: int) -> int: def f(x: int, f_modulus: list, T: int) -> int:
''' '''
@ -289,20 +285,21 @@ def f(x: int, f_modulus: list, T: int) -> int:
res = res % sm2p256v1.P res = res % sm2p256v1.P
return res return res
def GenerateReKey(sk_A, pk_B, N: int, T: int) -> list:
''' def GenerateReKey(sk_A: int, pk_B: point, N: int, T: int) -> list:
param: """
param:
skA, pkB, N(节点总数), T(阈值) skA, pkB, N(节点总数), T(阈值)
return: return:
rki(0 <= i <= N-1) rki(0 <= i <= N-1)
''' """
# 计算临时密钥对(x_A, X_A) # 计算临时密钥对(x_A, X_A)
x_A = random.randint(0, G.P - 1) x_A = random.randint(0, G.P - 1)
X_A = multiply(g, x_A) X_A = multiply(g, x_A)
# d是Bob的密钥对与临时密钥对的非交互式Diffie-Hellman密钥交换的结果 # d是Bob的密钥对与临时密钥对的非交互式Diffie-Hellman密钥交换的结果
d = hash3((X_A, pk_B, multiply(pk_B, x_A))) d = hash3((X_A, pk_B, multiply(pk_B, x_A)))
# 计算多项式系数, 确定代理节点的ID(一个点) # 计算多项式系数, 确定代理节点的ID(一个点)
f_modulus = [] f_modulus = []
# 计算f0 # 计算f0
@ -313,22 +310,23 @@ def GenerateReKey(sk_A, pk_B, N: int, T: int) -> list:
f_modulus.append(random.randint(0, G.P - 1)) f_modulus.append(random.randint(0, G.P - 1))
# 计算D # 计算D
D = H6((X_A, pk_B, multiply(pk_B, sk_A))) D = hash6((X_A, pk_B, multiply(pk_B, sk_A)))
# 计算KF # 计算KF
KF = [] KF = []
for i in range(N): for i in range(N):
y = random.randint(0, G.P - 1) y = random.randint(0, G.P - 1)
Y = multiply(g, y) Y = multiply(g, y)
s_x = H5(i, D) # id需要设置 s_x = hash5(i, D) # id需要设置
r_k = f(s_x, f_modulus, T) r_k = f(s_x, f_modulus, T)
U1 = multiply(U, r_k) U1 = multiply(U, r_k)
kFrag = (i, r_k, X_A, U1) kFrag = (i, r_k, X_A, U1)
KF.append(kFrag) KF.append(kFrag)
return KF return KF
def Encapsulate(pk_A: Tuple[int, int]) -> Tuple[int, Tuple[Tuple[int, int], Tuple[int, int], int]]:
def Encapsulate(pk_A: point) -> Tuple[int, capsule]:
r = random.randint(0, G.P - 1) r = random.randint(0, G.P - 1)
u = random.randint(0, G.P - 1) u = random.randint(0, G.P - 1)
E = multiply(g, r) E = multiply(g, r)
@ -340,120 +338,128 @@ def Encapsulate(pk_A: Tuple[int, int]) -> Tuple[int, Tuple[Tuple[int, int], Tupl
capsule = (E, V, s) capsule = (E, V, s)
return (K, capsule) return (K, capsule)
def Checkcapsule(capsule:Tuple[Tuple[int,int],Tuple[int,int],int]) -> bool: # 验证胶囊的有效性
E,V,s = capsule def Checkcapsule(capsule: capsule) -> bool: # 验证胶囊的有效性
h2 = hash2((E,V)) E, V, s = capsule
h2 = hash2((E, V))
g = (sm2p256v1.Gx, sm2p256v1.Gy) g = (sm2p256v1.Gx, sm2p256v1.Gy)
result1 = multiply(g,s) result1 = multiply(g, s)
temp = multiply(E,h2) # 中间变量 temp = multiply(E, h2) # 中间变量
result2 =add(V,temp) # result2=V*E^H2(E,V) result2 = add(V, temp) # result2=V*E^H2(E,V)
if result1 == result2: if result1 == result2:
flag =True flag = True
else: else:
flag = False flag = False
return flag
def ReEncapsulate(kFrag:list,capsule:Tuple[Tuple[int,int],Tuple[int,int],int]) -> Tuple[Tuple[int,int],Tuple[int,int],int,Tuple[int,int]] : return flag
id,rk,Xa,U1 = kFrag
E,V,s = capsule
def ReEncapsulate(kFrag: list, capsule: capsule) -> Tuple[point, point, int, point]:
id, rk, Xa, U1 = kFrag
E, V, s = capsule
if not Checkcapsule(capsule): if not Checkcapsule(capsule):
raise ValueError('Invalid capsule') raise ValueError("Invalid capsule")
flag = Checkcapsule(capsule) flag = Checkcapsule(capsule)
assert flag == True # 断言，判断胶囊capsule的有效性 assert flag == True # 断言,判断胶囊capsule的有效性
E1 = multiply(E,rk) E1 = multiply(E, rk)
V1 = multiply(V,rk) V1 = multiply(V, rk)
cfrag = E1,V1,id,Xa cfrag = E1, V1, id, Xa
return cfrag # cfrag=(E1,V1,id,Xa) E1= E^rk V1=V^rk return cfrag # cfrag=(E1,V1,id,Xa) E1= E^rk V1=V^rk
# 重加密函数 # 重加密函数
def ReEncrypt(kFrag:list,
C:Tuple[Tuple[Tuple[int,int],Tuple[int,int],int],int])->Tuple[Tuple[Tuple[int,int],Tuple[int,int],int,Tuple[int,int]],int] :
capsule,enc_Data = C def ReEncrypt(
kFrag: list, C: Tuple[capsule, bytes]
) -> Tuple[Tuple[point, point, int, point], bytes]:
capsule, enc_Data = C
cFrag = ReEncapsulate(kFrag, capsule)
return (cFrag, enc_Data) # 输出密文
cFrag = ReEncapsulate(kFrag,capsule)
return (cFrag,enc_Data) # 输出密文
# capsule, enc_Data = C # capsule, enc_Data = C
# N 是加密节点的数量，t是阈值 # 将加密节点加密后产生的t个（capsule,ct）合并在一起,产生cfrags = {{capsule1,capsule2,...},ct}
def mergecfrag(N:int,t:int)->tuple[Tuple[Tuple[int,int],Tuple[int,int] def mergecfrag(cfrag_cts: list) -> list:
,int,Tuple[int,int]], ...]: ct_list = []
cfrags = () cfrags_list = []
kfrags = GenerateReKey(sk_A,pk_B,N,t) cfrags = []
result = Encapsulate(pk_A) for cfrag_ct in cfrag_cts:
K,capsule = result cfrags_list.append(cfrag_ct[0])
for kfrag in kfrags: ct_list.append(cfrag_ct[1])
cfrag = ReEncapsulate(kfrag,capsule) cfrags.append(cfrags_list)
cfrags = cfrags + (cfrag,) cfrags.append(ct_list[0])
return cfrags return cfrags
def DecapsulateFrags(sk_B:int,pk_A:Tuple[int,int],cFrags:Tuple[Tuple[Tuple[int,int],Tuple[int,int],int,Tuple[int,int]]] def DecapsulateFrags(sk_B: int, pk_B: point, pk_A: point, cFrags: list) -> int:
,capsule:Tuple[Tuple[int,int],Tuple[int,int],int]) -> int: """
'''
return: return:
K: sm4 key K: sm4 key
''' """
Elist = [] Elist = []
Vlist = [] Vlist = []
idlist = [] idlist = []
X_Alist = [] X_Alist = []
t = 0 for cfrag in cFrags: # Ei,Vi,id,Xa = cFrag
for cfrag in cFrags: # Ei,Vi,id,Xa = cFrag
Elist.append(cfrag[0]) Elist.append(cfrag[0])
Vlist.append(cfrag[1]) Vlist.append(cfrag[1])
idlist.append(cfrag[2]) idlist.append(cfrag[2])
X_Alist.append(cfrag[3]) X_Alist.append(cfrag[3])
t = t+1 # 总共有t个片段，t为阈值
pkab = multiply(pk_A, sk_B) # pka^b
pkab = multiply(pk_A,sk_B) # pka^b D = hash6((pk_A, pk_B, pkab))
D = H6((pk_A,pk_B,pkab))
Sx = [] Sx = []
for id in idlist: # 从1到t for id in idlist: # 从1到t
sxi = H5(id,D) # id 节点的编号 sxi = hash5(id, D) # id 节点的编号
Sx.append(sxi) Sx.append(sxi)
bis= [] # b ==> λ bis = [] # b ==> λ
j = 1 j = 1
i = 1 i = 1
bi =1 bi = 1
for i in range(t): for i in range(len(cFrags)):
for j in range(t): for j in range(len(cFrags)):
if j == i: if j != i:
j=j+1 # bi = bi * (Sx[j] // (Sx[j] - Sx[i])) # 暂定整除
else: Sxj_sub_Sxi = (Sx[j] - Sx[i]) % sm2p256v1.P
bi = bi * (Sx[j]//(Sx[j]-Sx[i])) # 暂定整除 Sxj_sub_Sxi_inv = inv(Sxj_sub_Sxi, sm2p256v1.P)
bi = (bi * Sx[j] * Sxj_sub_Sxi_inv) % sm2p256v1.P
bis.append(bi) bis.append(bi)
E2=multiply(Elist[0],bis[0]) # E^ 便于计算 E2 = multiply(Elist[0], bis[0]) # E^ 便于计算
V2=multiply(Vlist[0],bis[0]) # V^ V2 = multiply(Vlist[0], bis[0]) # V^
for k in range(1,t): for k in range(1, len(cFrags)):
Ek = multiply(Elist[k],bis[k]) # EK/Vk 是个列表 Ek = multiply(Elist[k], bis[k]) # EK/Vk 是个列表
Vk = multiply(Vlist[k],bis[k]) Vk = multiply(Vlist[k], bis[k])
E2 = add(Ek,E2) E2 = add(Ek, E2)
V2 = add(Vk,V2) V2 = add(Vk, V2)
X_Ab = multiply(Xalist[0],b) # X_A^b X_A 的值是随机生成的xa，通过椭圆曲线上的倍点运算生成的固定的值 X_Ab = multiply(X_Alist[0], sk_B) # X_A^b X_A 的值是随机生成的xa,通过椭圆曲线上的倍点运算生成的固定的值
d = hash3((Xalist[0],pk_B,X_Ab)) d = hash3((X_Alist[0], pk_B, X_Ab))
EV = add(E2,V2) # E2 + V2 EV = add(E2, V2) # E2 + V2
EVd = multiply(EV,d) # (E2 + V2)^d EVd = multiply(EV, d) # (E2 + V2)^d
K = KDF(EVd) K = KDF(EVd)
return K return K
# M = IAEAM(K,enc_Data) # M = IAEAM(K,enc_Data)
def DecryptFrags(sk_B:int,
pk_A:Tuple[int,int], # cfrags = {{capsule1,capsule2,...},ct} ,ct->en_Data
cFrags:Tuple[Tuple[Tuple[int,int],Tuple[int,int],int,Tuple[int,int]]], def DecryptFrags(sk_B: int, pk_B: point, pk_A: point, cfrags: list) -> bytes:
C:Tuple[Tuple[Tuple[int,int],Tuple[int,int],int],int] capsules, enc_Data = cfrags # 加密后的密文
)->int: K = DecapsulateFrags(sk_B, pk_B, pk_A, capsules)
capsule,enc_Data = C # 加密后的密文 K = K.to_bytes(16)
K = DecapsulateFrags(sk_B,pk_A,cFrags,capsule) iv = b"tpretpretpretpre"
sm4_dec = Sm4Cbc(K, iv, DO_DECRYPT) # pylint: disable= e0602
iv = b'tpretpretpretpre' try:
sm4_dec = Sm4Cbc(K, iv, DO_DECRYPT) #pylint: disable= e0602 dec_Data = sm4_dec.update(enc_Data)
dec_Data = sm4_dec.update(enc_Data) dec_Data += sm4_dec.finish()
dec_Data += sm4_dec.finish() except Exception as e:
return dec_Data print(e)
print("key error")
dec_Data = b""
return dec_Data