实验：动态口令认证

大家好，欢迎来到IT知识分享网。

 一、设计目标

编写程序实现用动态口令认证：

（1）动态口令生成算法生成口令

（2）通过2台机器演示动态口令并比较 （两个程序也行）

二、动态令牌口令生成算具体要求

1.概述

本系统中使用的算法主要有两个：动态口令生成算法和生成种子密钥的密钥分散算法。

下面具体表述这两个算法。

2.动态口令生成算法

2.1综述

        动态口令生成算法是由交易信息随机数产生动态口令的算法。使用动态口令生成算法，在输入挑战信息后生成动态口令。用户持有的动态令牌和后台服务器同时使用这一算法生成动态口令完成验证。

        动态口令生成时将参与运算的事件因子和双方协商输入的挑战因子通过私有的种子密钥加密后截位产生最终令牌显示的口令。

2.2算法概述

2.2.1整体流程

本程序的动态口令生成方式概述如下：

T = T0 / TC 

S = F(K ,T)

OD = Truncate(S)

P = OD % (10^N)

T 是参与运算的时间因子，是一个8 字节整数，T0是以UTC 时间或用户选择的时间标准为计量标准的一个8字节整数，TC是以秒为单位的口令变化周期，最大长度为60s；

K 是长度不少于 128 比特的运算密钥，只有认证双方持有，F()是算法函数，在本程序中使用SM3算法。

S是算法函数输出结果，本程序中使用的SM3算法的输出结果（即杂凑值）长度为256 比特。

Truncate()是截位函数，OD是其输出结果，长度为32比特。

N是令牌或其他终端显示口令的位数，N不小于6。P是最终显示的动态口令。

<<为循环左移符号。|为连接符，表示将两组数据根据左右顺序拼接。

2.2.2截位算法

Truncate()是动态口令生成过程中，使用的截位函数。本程序中使用的杂凑结果截位实现方式如下：

定义S1,S2,S3,S4,S5,S6,S7,S8，表示8个4字节整数，通过如下方法赋值：

S1 = S[0] << 24 | S[1] << 16 | S[2] << 8 | S[3]

S2 = S[4] << 24 | S[5] << 16 | S[6] << 8 | S[7]

S3 = S[8] << 24 | S[9] << 16 | S[10] << 8 | S[11]

S4 = S[12] << 24 | S[13] << 16 | S[14] << 8 | S[15]

S5 = S[16] << 24 | S[17] << 16 | S[18] << 8 | S[19]

S6 = S[20] << 24 | S[21] << 16 | S[22] << 8 | S[23]

S7 = S[24] << 24 | S[25] << 16 | S[26] << 8 | S[27]

S8 = S[28] << 24 | S[29] << 16 | S[30] << 8 | S[31]

OD= (S1 + S2 + S3 +S4 +S5 +S6 + S7 +S8) MOD 2^32

2.2.3 算法使用要求

2.2.3.1 使用要求

        本程序涉及动态口令生成方式的杂凑算法，应选用SM3杂凑算法，杂凑值长度为256比特。

2.2.3.2 SM3杂凑算法使用说明

在S = F(K , T)环节，使用SM3算法时，K|T为输入参数。

三、代码实现

server.py

import socket from gmssl import sm3, func import time def generate_dynamic_password(secret, period=30, digits=6): T = int(time.time() / period) # 计算时间步长，32位整数 T_bytes = T.to_bytes((T.bit_length() + 7) // 8, byteorder='big') # 将时间步长转换为字节，4字节数组 KT = secret.encode() + T_bytes # 连接密钥和时间步长 hash_result = sm3.sm3_hash(func.bytes_to_list(KT)) # 使用SM3算法进行哈希，64字符的十六进制字符串 S = [int(hash_result[i:i + 2], 16) for i in range(0, len(hash_result), 2)] # 将哈希结果转换为整数数组，长度为32的整数数组 # 进行截位函数计算 S1 = S[0] << 24 | S[1] << 16 | S[2] << 8 | S[3] S2 = S[4] << 24 | S[5] << 16 | S[6] << 8 | S[7] S3 = S[8] << 24 | S[9] << 16 | S[10] << 8 | S[11] S4 = S[12] << 24 | S[13] << 16 | S[14] << 8 | S[15] S5 = S[16] << 24 | S[17] << 16 | S[18] << 8 | S[19] S6 = S[20] << 24 | S[21] << 16 | S[22] << 8 | S[23] S7 = S[24] << 24 | S[25] << 16 | S[26] << 8 | S[27] S8 = S[28] << 24 | S[29] << 16 | S[30] << 8 | S[31] OD = (S1 + S2 + S3 + S4 + S5 + S6 + S7 + S8) % (2 32) # 计算最终的OD值，32位整数。 return f"{OD % (10 digits):0{digits}d}" # 计算并返回最终的动态口令 def main(): host = 'localhost' port = 12345 server_socket = socket.socket() server_socket.bind((host, port)) server_socket.listen(1) print("服务端已启动，等待客户端连接...") while True: client_socket, addr = server_socket.accept() print(f"接收到来自 {addr} 的连接") data = client_socket.recv(1024).decode() print(f"来自客户端的数据：{data}") secret = '' + data password = generate_dynamic_password(secret) client_socket.send(password.encode()) client_socket.close() if __name__ == '__main__': main() 

1. 客户端通过 import socket 导入了 socket 模块，并从 server 模块中导入了 generate_dynamic_password 函数。
2. 在 main() 函数中，首先指定了服务器的主机名和端口号，并创建了一个客户端套接字对象 client_socket。
3. 客户端调用 client_socket.connect((host, port)) 来连接到服务器。
4. 客户端要求用户输入口令，并将口令编码后发送给服务器。
5. 客户端生成动态口令的过程是：首先将用户口令与固定密钥 wjy 连接起来，然后调用 generate_dynamic_password 函数生成动态口令 password1。
6. 客户端接收从服务器返回的动态口令 password2。
7. 最后，客户端将在本地生成的动态口令 password1 和从服务器接收到的动态口令 password2 进行比较，如果两者相同则认为连接成功，否则连接失败。
8. 最后关闭客户端套接字。

关键函数：generate_dynamic_password函数

这个函数 generate_dynamic_password用于生成一个动态密码（一次性密码，OTP），其生成过程依赖于一个密钥和当前时间。密码在规定时间段（默认为30秒）内不断变化。

1. 函数定义：

generate_dynamic_password(secret, period=30, digits=6)：

     -secret：一个字符串，作为共享的密钥。

     -period：一个可选的整数参数，表示时间周期（以秒为单位），默认值为30秒。

     -digits：一个可选的整数参数，表示生成的动态口令的位数，默认值为6位。

2. 计算时间步长：

T = int(time.time() / period)：

     -time.time() 返回当前的时间戳（以秒为单位）。

     -将当前时间戳除以周期period并取整，得到时间步长T。这个时间步长T表示从某个起始点（通常是1970年1月1日）以来经过的时间段数量。

3. 将时间步长转换为字节：

T_bytes = T.to_bytes((T.bit_length() + 7) // 8, byteorder=’big’)：

     -T.bit_length() 返回整数 T的二进制位数。

     -(T.bit_length() + 7) // 8计算将 T表示为字节所需的字节数。

     -T.to_bytes将 T转换为字节序列，byteorder=’big’表示使用大端字节序。

4. 连接密钥和时间步长：

KT = secret.encode() + T_bytes：

     – secret.encode() 将密钥字符串编码为字节。

     – 将密钥的字节表示与时间步长的字节表示连接在一起，得到 KT。

5. 使用SM3算法进行哈希：

hash_result = sm3.sm3_hash(func.bytes_to_list(KT))：

     -sm3.sm3_hash是一个SM3哈希函数，将输入的字节列表KT转换为哈希值。

     -func.bytes_to_list(KT)将 KT转换为字节列表。

6. 将哈希结果转换为整数数组：

S = [int(hash_result[i:i + 2], 16) for i in range(0, len(hash_result), 2)]：

     – 遍历哈希结果字符串，每两个字符作为一个16进制数，转换为整数并存入列表 S。

7. 截位函数计算：

计算 S中前32个字节的组合值：

     – S1 = S[0] << 24 | S[1] << 16 | S[2] << 8 | S[3]

     – S2 = S[4] << 24 | S[5] << 16 | S[6] << 8 | S[7]

     – S3 = S[8] << 24 | S[9] << 16 | S[10] << 8 | S[11]

     – S4 = S[12] << 24 | S[13] << 16 | S[14] << 8 | S[15]

     – S5 = S[16] << 24 | S[17] << 16 | S[18] << 8 | S[19]

     – S6 = S[20] << 24 | S[21] << 16 | S[22] << 8 | S[23]

     – S7 = S[24] << 24 | S[25] << 16 | S[26] << 8 | S[27]

     – S8 = S[28] << 24 | S[29] << 16 | S[30] << 8 | S[31]

     这些操作将连续的四个字节组合成一个32位整数。

8. 计算最终的OD值：

OD = (S1 + S2 + S3 + S4 + S5 + S6 + S7 + S8) % (2 32)：

     -将上述八个32位整数相加，取模 2 32，得到最终的OD值。

9. 计算并返回最终的动态口令：

return f”{OD % (10 digits):0{digits}d}”：

     -将OD值对 10 digits 取模，并格式化为一个长度为 digits（这里是6）的字符串，返回这个动态口令。

通过以上步骤，这个函数生成了一个基于密钥和时间步长的动态口令，该口令每隔指定的时间周期（这里设置为30秒）自动变化。

client.py

import socket from server import generate_dynamic_password def main(): host = 'localhost' port = 12345 client_socket = socket.socket() client_socket.connect((host, port)) userkey = input("请输入口令: ") client_socket.send(userkey.encode()) secret = '' + userkey password1 = generate_dynamic_password(secret) password2 = client_socket.recv(1024).decode() print(f"在客户端生成的口令是：{password1}") print(f"从服务端接收到的动态口令是：{password2}") if password1 != password2: print("连接失败") else: print("连接成功") client_socket.close() if __name__ == '__main__': main() 

1. 创建一个服务器套接字并绑定到指定的主机和端口上。
2. 监听连接请求，等待客户端连接。
3. 一旦接受到客户端连接，接收客户端发送的数据，这里是用户口令。
4. 将收到的口令与预定义的密钥连接起来，然后使用动态口令生成算法生成动态口令。
5. 将生成的动态口令发送回客户端。
6. 关闭客户端套接字，等待下一个连接。

四、演示

 server端

client端