前言

在上个月我写了一个加密传输的Demo，相当的业余，这个东西只能做到从客户端向服务端的单向传输，而且因为只有一次请求，所以中间人攻击也非常的容易。
虽然我觉得那个应该足够我完成作业了，但是我想了想，太业余总显得我很菜 ~~（难道不是吗？）~~ ，所以我打算还是努力一下把完整的握手也做出来吧。

实现思路

和上次差不多，同样我打算通过RSA加密一个随机数作为AES的密钥，但是和上次不同，这次的这个密钥将在通信建立之后密钥就不再变化，传入传出都是这个密钥。
之前觉得握手阶段可能很麻烦，不过我画了一下图稍微理了理思路发现其实也没多复杂，也就是客户端生成密钥然后加密传输过来让服务端保存，完成之后之间的传输只要传被加密的数据就行了。这样我觉得应该也能起到一定的防止中间人攻击的作用吧……
不过我只用了一个文件作为传输的管道，为了避免冲突，我在每次写入数据的时候也都标明了数据的来源，这样就不会因为自己刚发完然后自己又重新接收自己发的东西了。
另外老师希望我的Demo有个简单的应用，所以我又另外加了一个极其简单的登录系统在里面，这样我就不用手动操作服务端了。

代码

server.py

# -*- coding: utf-8 -*-
print("加密通道服务端演示开始")
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5
from Crypto.Cipher import AES  
import base64
import hashlib
import json
import time

private_key = """-----BEGIN RSA PRIVATE KEY-----
MIICXQIBAAKBgQDfEQ82qUrto7h4BL3TsA/DFXSdM44cbeY4kPccD7gLGhaZRClz
YKIh5zYdfjBGF+0HXfMa1u9b7GNs2AjVIsx8Kx0QLnMfmtkmGWGhOXz/9IDLKJOx
0weKv61gysKItgzVKn2mbLool4R/PQBc3AjDyHw+io1KpVz+3kRTaGs1fQIDAQAB
AoGAWB4kFWLA/6k6OOcemd4mC9mQ7HyuOdrMJDJX+5TWDkSrArajbTmSMrRkczgj
F71h3BQn8cVQXs695ARfUNrjTbi2Y0LjN7ScK7ExzTLdoMEFw5JsHggJZ0zBQY6w
mwOdGfqzA6tZPXgkn+jqEha+CD6GrwnTM1oDGJC/aKG2OmECQQDkO9IhUhFc/PSU
0zvGE6AOcqk5wlOuvMg+oAFHJHJZ9XW7+X/Nx0ZoVDFq/cZQj+46t+fiwUwhdW7l
IfCvNGKFAkEA+jRQmWGKrbf1ns4S0SezJvysd5O6otRGJXr+Ex2uDhc39ZTeUsyg
kjrLhp8STLMOmql+8g5fghct17EuCX1EmQJBAJz9BNnEkIrst/OSpH/nyeWGOx6u
q077LaXd+2MLD9kO/O/Se3V5B9YFa4STkJCjoBMloswXd51gIGpdgSeSmd0CQQCL
PrwwcGmWfo+ynqs4PajlpK9zKQMwhYS4bTejedwZOXDKOtx0Ji+i0hfcxwCPMQOK
rZPZsIgUxUOdC508aLvZAkBDkHxunCzDm0w4DdTUN7S9YSpVvQEjK/xUQiWaKV12
8QgskhU2DNdYK2NxifnWrKtx3uQmqMxX5aLuJZ4493yr
-----END RSA PRIVATE KEY-----"""

# 公钥解密
def rsa_decode(cipher_text, private_key):
    rsakey = RSA.importKey(private_key)  # 导入读取到的私钥
    cipher = PKCS1_v1_5.new(rsakey)  # 生成对象
    # 将密文解密成明文，返回的是一个bytes类型数据，需要自己转换成str
    text = cipher.decrypt(base64.b64decode(cipher_text), "ERROR")
    return text.decode()
    
class PrpCrypt(object):
 
    def __init__(self, key):
        self.key = key.encode('utf-8')
        self.mode = AES.MODE_CBC
        
    def encrypt(self, text):
        text = text.encode('utf-8')
        cryptor = AES.new(self.key, self.mode, b'0000000000000000')
        # 这里密钥key 长度必须为16（AES-128）,
        # 24（AES-192）,或者32 （AES-256）Bytes 长度
        # 目前AES-128 足够目前使用
        length = 16
        count = len(text)
        if count < length:
            add = (length - count)
            # \0 backspace
            # text = text + ('\0' * add)
            text = text + ('\0' * add).encode('utf-8')
        elif count > length:
            add = (length - (count % length))
            # text = text + ('\0' * add)
            text = text + ('\0' * add).encode('utf-8')
        self.ciphertext = cryptor.encrypt(text)
        # 因为AES加密时候得到的字符串不一定是ascii字符集的，输出到终端或者保存时候可能存在问题
        # 所以这里统一把加密后的字符串转化为16进制字符串
        return base64.b64encode(self.ciphertext)
 
    # 解密后，去掉补足的空格用strip() 去掉
    def decrypt(self, text):
        cryptor = AES.new(self.key, self.mode, b'0000000000000000')
        plain_text = cryptor.decrypt(base64.b64decode(text))
        # return plain_text.rstrip('\0')
        return bytes.decode(plain_text).rstrip('\0')

#初始化pipe
file = open('pipe.txt','w')
print("",file = file)
file.close()
#协商密钥
while True:
    pipe = open("pipe.txt", mode='r')
    data = pipe.read()
    if not data == "\n":
        data = json.loads(data)
        pipe.close()
        break
    pipe.close()
    time.sleep(1)
file = open('pipe.txt','w')
print("",file = file)
file.close()
key = rsa_decode(data["key"], private_key)
akey = PrpCrypt(key)
print("成功获取密钥",key)
print("已建立连接")
userInfo={"Name":"Mayx","Passwd":'25d55ad283aa400af464c76d713c07ad',"Book":"《会说话就多说点》，《在哪里能买到您的著作》"}

#处理阶段
#Waiting for Quest
while True:
    pipe = open("pipe.txt", mode='r')
    data = pipe.read()
    if not data == "\n":
        data = json.loads(data)
        if data["from"] == "Client":
            pipe.close()
            break
    pipe.close()
    time.sleep(1)
print("已收到请求")
data = json.loads(akey.decrypt(data["data"]))
if data["name"] == userInfo["Name"] and data["pwd"] == userInfo["Passwd"]:
    file = open('pipe.txt','w')
    print(json.dumps({"from":"Server","data":akey.encrypt("登录成功，您的书单如下："+userInfo["Book"]).decode("utf-8")}),file = file)
    file.close()
else:
    file = open('pipe.txt','w')
    print(json.dumps({"from":"Server","data":akey.encrypt("登录失败").decode("utf-8")}),file = file)
    file.close()
print("请求已回应")
print("加密通道服务端演示结束")

client.py

# -*- coding: utf-8 -*-
print("加密通道客户端演示开始")
import random
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5
from Crypto.Cipher import AES  
import base64
import hashlib
import json
import time

public_key = """-----BEGIN PUBLIC KEY-----
MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDfEQ82qUrto7h4BL3TsA/DFXSd
M44cbeY4kPccD7gLGhaZRClzYKIh5zYdfjBGF+0HXfMa1u9b7GNs2AjVIsx8Kx0Q
LnMfmtkmGWGhOXz/9IDLKJOx0weKv61gysKItgzVKn2mbLool4R/PQBc3AjDyHw+
io1KpVz+3kRTaGs1fQIDAQAB
-----END PUBLIC KEY-----
"""

# 公钥加密
def rsa_encode(message, public_key):
    rsakey = RSA.importKey(public_key)  # 导入读取到的公钥
    cipher = PKCS1_v1_5.new(rsakey)  # 生成对象
    # 通过生成的对象加密message明文，注意，在python3中加密的数据必须是bytes类型的数据，不能是str类型的数据
    cipher_text = base64.b64encode(cipher.encrypt(message.encode(encoding="utf-8")))
    # 公钥每次加密的结果不一样跟对数据的padding（填充）有关
    return cipher_text.decode()

class PrpCrypt(object):
 
    def __init__(self, key):
        self.key = key.encode('utf-8')
        self.mode = AES.MODE_CBC
 
    # 加密函数，如果text不足16位就用空格补足为16位，
    # 如果大于16当时不是16的倍数，那就补足为16的倍数。
    def encrypt(self, text):
        text = text.encode('utf-8')
        cryptor = AES.new(self.key, self.mode, b'0000000000000000')
        # 这里密钥key 长度必须为16（AES-128）,
        # 24（AES-192）,或者32 （AES-256）Bytes 长度
        # 目前AES-128 足够目前使用
        length = 16
        count = len(text)
        if count < length:
            add = (length - count)
            # \0 backspace
            # text = text + ('\0' * add)
            text = text + ('\0' * add).encode('utf-8')
        elif count > length:
            add = (length - (count % length))
            # text = text + ('\0' * add)
            text = text + ('\0' * add).encode('utf-8')
        self.ciphertext = cryptor.encrypt(text)
        # 因为AES加密时候得到的字符串不一定是ascii字符集的，输出到终端或者保存时候可能存在问题
        # 所以这里统一把加密后的字符串转化为16进制字符串
        return base64.b64encode(self.ciphertext)
        
        # 解密后，去掉补足的空格用strip() 去掉
    def decrypt(self, text):
        cryptor = AES.new(self.key, self.mode, b'0000000000000000')
        plain_text = cryptor.decrypt(base64.b64decode(text))
        # return plain_text.rstrip('\0')
        return bytes.decode(plain_text).rstrip('\0')

#初始化阶段
while True:
    try:
        pipe = open("pipe.txt", mode='r')
    except:
        time.sleep(1)
    else:
        break
pipe.close()
#协商密钥
key = str(random.randint(1000000000000000,9999999999999999))
akey = PrpCrypt(key)
cipher = rsa_encode(key, public_key)
data = json.dumps({"key":cipher})
file = open('pipe.txt','w')
print(data,file = file)
file.close()
while True:
    pipe = open("pipe.txt", mode='r')
    if pipe.read() == "\n":
        pipe.close()
        break
    pipe.close()
    time.sleep(1)
print("成功发送密钥",key)
print("已建立连接")

#处理阶段
#Single Quest
name = input("请输入用户名：")
pwd = input("请输入密码：")
hash = hashlib.md5(pwd.encode(encoding='UTF-8')).hexdigest()
data = json.dumps({"from":"Client","data":akey.encrypt(json.dumps({"name":name,"pwd":hash})).decode("utf-8")})
file = open('pipe.txt','w')
print(data,file = file)
file.close()
print("已发出，等待回应")
while True:
    pipe = open("pipe.txt", mode='r')
    data = pipe.read()
    if not data == "\n":
        data = json.loads(data)
        if data["from"] == "Server":
            pipe.close()
            break
    pipe.close()
    time.sleep(1)
print("已收到回应")
data = akey.decrypt(data["data"])
print(data)
print("加密通道客户端演示结束")

后记

关于SSL/TLS的具体内容我也没有好好看一遍，我打算回头有时间仔细看一看，然后再回来看这个代码是有多么的糟糕🤪。

10 KiB Raw Blame History Unescape Escape

前言

实现思路

代码

server.py

client.py

后记

10 KiB

Raw Blame History Unescape Escape