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

main #13

Merged
ccyj merged 4 commits from sangge/tpre-python:main into main 2023-10-18 16:08:04 +08:00
Conversation 0 Commits 4 Files Changed 1 +184 -175
1 changed files with 184 additions and 175 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 02fbfbbd70 - Show all commits

147
src/tpre.py
View File

@ -6,6 +6,7 @@ import traceback
point = Tuple[int, int] point = Tuple[int, int]
capsule = Tuple[point, point, int] capsule = Tuple[point, point, int]
# 生成密钥对模块 # 生成密钥对模块
class CurveFp: class CurveFp:
def __init__(self, A, B, P, N, Gx, Gy, name): def __init__(self, A, B, P, N, Gx, Gy, name):
@ -17,6 +18,7 @@ class CurveFp:
self.Gy = Gy self.Gy = Gy
self.name = name self.name = name
sm2p256v1 = CurveFp( sm2p256v1 = CurveFp(
name="sm2p256v1", name="sm2p256v1",
A=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFC, A=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFC,
@ -24,7 +26,7 @@ sm2p256v1 = CurveFp(
P=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFF, P=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFF,
N=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFF7203DF6B21C6052B53BBF40939D54123, N=0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFF7203DF6B21C6052B53BBF40939D54123,
Gx=0x32C4AE2C1F1981195F9904466A39C9948FE30BBFF2660BE1715A4589334C74C7, Gx=0x32C4AE2C1F1981195F9904466A39C9948FE30BBFF2660BE1715A4589334C74C7,
Gy=0xBC3736A2F4F6779C59BDCEE36B692153D0A9877CC62A474002DF32E52139F0A0 Gy=0xBC3736A2F4F6779C59BDCEE36B692153D0A9877CC62A474002DF32E52139F0A0,
) )
# 椭圆曲线 # 椭圆曲线
@ -33,17 +35,20 @@ G = sm2p256v1
# 生成元 # 生成元
g = (sm2p256v1.Gx, sm2p256v1.Gy) g = (sm2p256v1.Gx, sm2p256v1.Gy)
def multiply(a: point, n: int) -> point: 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: point, b: point) -> point: 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
@ -55,16 +60,21 @@ def inv(a: int, n: int) -> int:
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: point) -> 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) -> point: 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)
@ -76,11 +86,9 @@ def jacobianDouble(Xp_Yp_Zp: Tuple[int, int, int], A: int, P: int) -> Tuple[int,
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
@ -106,14 +114,10 @@ def jacobianAdd(
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)
@ -124,45 +128,50 @@ 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 hash2(double_G: Tuple[point, point]) -> 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[point,
point, def hash3(triple_G: Tuple[point, point, point]) -> int:
point]) -> 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[point,
point, def hash4(triple_G: Tuple[point, point, point], Zp: int) -> int:
point],
Zp: 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))
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: point) -> int: def KDF(G: point) -> int:
sm3 = Sm3() # pylint: disable=e0602 sm3 = Sm3() # pylint: disable=e0602
print(G) print(G)
@ -170,25 +179,23 @@ def KDF(G: point) -> int:
sm3.update(i.to_bytes(32)) sm3.update(i.to_bytes(32))
digest = sm3.digest() 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 mask_128bit = (1 << 128) - 1
digest = digest & mask_128bit digest = digest & mask_128bit
print('key =',digest) print("key =", digest)
traceback.print_stack() 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[point, int]:
'''
params: params:
lamda_param: an init safety param 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()
@ -196,29 +203,30 @@ def GenerateKeyPair(
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的公钥和私钥 # 生成A和B的公钥和私钥
# pk_A, sk_A = GenerateKeyPair(0, ()) # pk_A, sk_A = GenerateKeyPair(0, ())
# pk_B, sk_B = GenerateKeyPair(0, ()) # pk_B, sk_B = GenerateKeyPair(0, ())
def Encrypt(pk: point, m: bytes) -> Tuple[Tuple[
point,point, int], bytes]: def Encrypt(pk: point, m: bytes) -> Tuple[Tuple[point, point, int], bytes]:
enca = Encapsulate(pk) enca = Encapsulate(pk)
K = enca[0].to_bytes(16) 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
enc_Data = sm4_enc.update(m) enc_Data = sm4_enc.update(m)
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: capsule) -> int: def Decapsulate(ska: int, capsule: capsule) -> int:
E, V, s = capsule E, V, s = capsule
EVa = multiply(add(E, V), ska) # (E*V)^ska EVa = multiply(add(E, V), ska) # (E*V)^ska
@ -226,57 +234,57 @@ def Decapsulate(ska:int,capsule:capsule) -> int:
return K return K
def Decrypt(sk_A: int,C:Tuple[Tuple[
point, point, int], bytes]) ->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 hash5(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 return hash
def hash6(triple_G: Tuple[point,
point, def hash6(triple_G: Tuple[point, point, point]) -> int:
point]) -> 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))
hash = sm3.digest() hash = sm3.digest()
hash = int.from_bytes(hash,'big') % G.P hash = int.from_bytes(hash, "big") % G.P
return hash return hash
def f(x: int, f_modulus: list, T: int) -> int:
'''
''' def f(x: int, f_modulus: list, T: int) -> int:
""" """
res = 0 res = 0
for i in range(T): for i in range(T):
res += f_modulus[i] * pow(x, i) res += f_modulus[i] * pow(x, i)
return res return res
def GenerateReKey(sk_A: int, pk_B: point, 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)
@ -309,6 +317,7 @@ def GenerateReKey(sk_A: int, pk_B: point, N: int, T: int) -> list:
return KF return KF
def Encapsulate(pk_A: point) -> Tuple[int, capsule]: 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)
@ -320,6 +329,7 @@ def Encapsulate(pk_A: point) -> Tuple[int, capsule]:
capsule = (E, V, s) capsule = (E, V, s)
return (K, capsule) return (K, capsule)
def Checkcapsule(capsule: capsule) -> bool: # 验证胶囊的有效性 def Checkcapsule(capsule: capsule) -> bool: # 验证胶囊的有效性
E, V, s = capsule E, V, s = capsule
h2 = hash2((E, V)) h2 = hash2((E, V))
@ -334,11 +344,12 @@ def Checkcapsule(capsule:capsule) -> bool: # 验证胶囊的有效性
return flag return flag
def ReEncapsulate(kFrag: list, capsule: capsule) -> Tuple[point, point, int, point]: def ReEncapsulate(kFrag: list, capsule: capsule) -> Tuple[point, point, int, point]:
id, rk, Xa, U1 = kFrag id, rk, Xa, U1 = kFrag
E, V, s = capsule 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)
@ -347,12 +358,17 @@ def ReEncapsulate(kFrag:list,capsule:capsule) -> Tuple[point,point,int,point] :
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[capsule,int])->Tuple[Tuple[point,point,int,point],int] :
def ReEncrypt(
kFrag: list, C: Tuple[capsule, int]
) -> Tuple[Tuple[point, point, int, point], int]:
capsule, enc_Data = C capsule, enc_Data = C
cFrag = ReEncapsulate(kFrag, capsule) cFrag = ReEncapsulate(kFrag, capsule)
return (cFrag, enc_Data) # 输出密文 return (cFrag, enc_Data) # 输出密文
# capsule, enc_Data = C # capsule, enc_Data = C
@ -369,16 +385,11 @@ def mergecfrag(cfrag_cts:list)->list:
return cfrags return cfrags
def DecapsulateFrags(sk_B: int, pk_B: point, pk_A: point, cFrags: list) -> int:
def DecapsulateFrags(sk_B:int, """
pk_B: point,
pk_A:point,
cFrags:list
) -> int:
'''
return: return:
K: sm4 key K: sm4 key
''' """
Elist = [] Elist = []
Vlist = [] Vlist = []
@ -426,18 +437,16 @@ def DecapsulateFrags(sk_B:int,
return K return K
# M = IAEAM(K,enc_Data) # M = IAEAM(K,enc_Data)
# cfrags = {{capsule1,capsule2,...},ct} ,ct->en_Data # cfrags = {{capsule1,capsule2,...},ct} ,ct->en_Data
def DecryptFrags(sk_B: int, def DecryptFrags(sk_B: int, pk_B: point, pk_A: point, cfrags: list) -> bytes:
pk_B: point,
pk_A: point,
cfrags:list
) -> bytes:
capsules, enc_Data = cfrags # 加密后的密文 capsules, enc_Data = cfrags # 加密后的密文
K = DecapsulateFrags(sk_B, pk_B, pk_A, capsules) K = DecapsulateFrags(sk_B, pk_B, pk_A, capsules)
K = K.to_bytes(16) K = K.to_bytes(16)
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
try: try:
dec_Data = sm4_dec.update(enc_Data) dec_Data = sm4_dec.update(enc_Data)
@ -445,5 +454,5 @@ def DecryptFrags(sk_B: int,
except Exception as e: except Exception as e:
print(e) print(e)
print("key error") print("key error")
dec_Data = b'' dec_Data = b""
return dec_Data return dec_Data