快捷導(dǎo)航

python實現(xiàn)簡易SSL的項目實踐

更新時間：2025年02月10日 10:30:14 作者：mileeees

本文主要介紹了python實現(xiàn)簡易SSL的項目實踐,包括CA.py、server.py和client.py三個模塊,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧

運行環(huán)境

操作系統(tǒng)：浪潮云啟操作系統(tǒng) InLinux 23.12 LTS SP1
python版本：3.9.9
使用的python包：cryptography、ownca、signify

浪潮云啟操作系統(tǒng)（InLinux）面向企業(yè)級業(yè)務(wù)，提供自主可控、安全可靠的新一代服務(wù)器操作系統(tǒng)，全面支持云計算、大數(shù)據(jù)、人工智能、物聯(lián)網(wǎng)等新型場景，具備性能高效、擴展便捷、管理智能、內(nèi)生安全等特性。

運行前準備

# 安裝python
yum install -y python

# 安裝python包
pip install cryptography ownca signify

程序?qū)崿F(xiàn)與流程說明

本程序?qū)崿F(xiàn)了一個簡易的SSL，共分為三個模塊：CA.py，server.py，client.py。

CA.py負責簽發(fā)證書，server.py與client.py通信，過程中會實現(xiàn)生成公私鑰對、會話密鑰等過程。為了簡易性，server能夠從本地直接獲取證書，且只有server對client檢查證書。

程序流程如下：

CA生成根證書、公私鑰對
client生成公私鑰對、向CA發(fā)送CSR請求
CA收到CSR請求，用私鑰簽名，向client發(fā)送簽名證書，client拿到證書
client第一次向server發(fā)送數(shù)據(jù)，并附帶證書信息
server檢驗證書信息，并生成session_key，利用client公鑰加密session_key，將密文發(fā)回給client，之后的對話用session_key驗證。
client解密session_key，利用消息和session_key生成MAC，向server發(fā)送消息并附帶MAC
server收到消息并驗證MAC，對話結(jié)束。

運行截圖

CA.py

請?zhí)砑訄D片描述

server.py

在這里插入圖片描述

client.py

在這里插入圖片描述

證書目錄：

在這里插入圖片描述

代碼

CA.py

import ownca.ownca
import socket
from cryptography import x509
from cryptography.x509 import NameOID

ca = ownca.CertificateAuthority(ca_storage='./myCA',common_name='myCA')
print("myCA initialized")

HOST = "127.0.0.1"  # Standard loopback interface address (localhost)
PORT = 11111  # Port to listen on (non-privileged ports are > 1023)

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((HOST, PORT))
server_socket.listen(5)
print("myCA server is running...")
while True:
    conn, addr = server_socket.accept()
    print(f"Connected by {addr}")
    data = conn.recv(2048)
    csr = x509.load_pem_x509_csr(data)
    try:
        # sign the CSR, if success, the certificate will generate and store
        ca.sign_csr(csr,csr.public_key(),maximum_days=825)
    except Exception as e:
        print(e)
    # extract CN from CSR
    common_name = csr.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
    # load the issued certificate from existing file
    load_cert = ca.load_certificate(common_name)
    print('Successfully sign for ' + common_name)
    # Send the certificate bytes
    send_data = load_cert.cert_bytes
    conn.sendall(send_data)
    conn.close()

client.py

import socket
import pickle
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, hmac
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding


def generateCSR(common_name, private_key):
    csr = x509.CertificateSigningRequestBuilder().subject_name(x509.Name([
        # Provide various details about who we are.
        x509.NameAttribute(NameOID.COUNTRY_NAME, u"CN"),
        x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, u"Beijing"),
        # x509.NameAttribute(NameOID.LOCALITY_NAME, u"Richmond"),
        # x509.NameAttribute(NameOID.ORGANIZATION_NAME, u"My Organization"),
        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
    ])).sign(private_key, hashes.SHA256())
    return csr

def RSADecryption(cipher, private_key):
    msg = private_key.decrypt(
        cipher,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    return msg

# Generate the RSA private key
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
cn = input("Input common name: ")
csr = generateCSR(cn, private_key)
public_key = private_key.public_key()
# public_key = csr.public_key()
csr_bytes = csr.public_bytes(serialization.Encoding.PEM)
print(cn + ": Key Pair and CSR generated!")

HOST = "127.0.0.1"  # The server's hostname or IP address

CA_PORT = 11111  # The CA server port
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect((HOST, CA_PORT))
    s.sendall(csr_bytes) # send the generated CSR
    cert_data = s.recv(2048)
print(cn + " signing finished!, part of the cert:")
print(cert_data[:40] + b'...')

print('================================================')
Server_PORT = 22222  # The Server port
session_key = b''
msg = b''
hMAC = b''
public_key_pem = public_key.public_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PublicFormat.SubjectPublicKeyInfo
)
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect((HOST, Server_PORT))
    # send all params to the Server
    params = [cn, cert_data, public_key_pem, session_key, msg, hMAC]
    data = pickle.dumps(params)
    s.sendall(data)
    data = s.recv(2048)
print("Received from Server: ")
session_key = RSADecryption(data, private_key)
print("Session key decrypted:",session_key)

msg = b'This is messgae from ' + cn.encode('utf-8') + b'.\n'
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect((HOST, Server_PORT))
    msg = msg * 10
    h = hmac.HMAC(session_key,hashes.SHA256())
    h.update(msg)
    hMAC = h.finalize()
    # Send all params with valid session_key, msg and hMAC
    params = [cn, cert_data, public_key_pem, session_key, msg, hMAC]
    data = pickle.dumps(params)
    s.sendall(data)
    data = s.recv(2048)
    print(data)

server.py

import os
import pickle
import socket
import ownca
from cryptography.hazmat.primitives import hashes, hmac
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.serialization import load_pem_public_key
from signify.x509 import CertificateStore, VerificationContext, Certificate

ca = ownca.CertificateAuthority(ca_storage='./myCA',common_name='myCA')
print("Successfully load the CA certificate")

HOST = "127.0.0.1"
PORT = 22222
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((HOST, PORT))
server_socket.listen()
print("Server is running...")

# generate a list of random bytes as Session key
def generateSessionKey(byte_num=16):
    return os.urandom(byte_num)

# Encrypt the msg using RSA
def RSAEncryption(msg, public_key):
    cipher = public_key.encrypt(
        msg,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    return cipher

while True:
    conn, addr = server_socket.accept()
    print(f"Connected by {addr}")
    data = conn.recv(2048)
    # find all params from the received data
    params = pickle.loads(data)
    cn = params[0]
    cert_from_cli_pem = params[1]
    public_key_pem = params[2] # for encrypting the session_key
    session_key = params[3]
    msg = params[4]
    hMAC_to_check = params[5]
    # if no session built
    if not session_key:
        try:
            # Using Signify to verify the certificate
            # Raise exception if verification error
            trust_store = CertificateStore([Certificate.from_pem(ca.cert_bytes)], trusted=True)
            context = VerificationContext(trust_store)
            Certificate.from_pem(cert_from_cli_pem).verify(context)
        except Exception as e:
            print(cn + "Verification error: ", e)
            conn.sendall(b"Verification error!\n")
            conn.close()
            continue
        finally:
            print(cn + " Verification success!")
        # load cert from stu_pem and find the public_key
        public_key = load_pem_public_key(public_key_pem)
        session_key = generateSessionKey()
        print(cn + " Session key generated:",session_key)
        cipher = RSAEncryption(session_key,public_key)
        conn.sendall(cipher)
    # valid session key
    else:
        h = hmac.HMAC(session_key, hashes.SHA256())
        h.update(msg)
        try:
            h.verify(hMAC_to_check)
        except Exception as e:
            print(cn + " MAC Verification wrong:",e)
            conn.close()
            continue
        finally:
            print(cn + " MAC Verification success!")
            print(msg)
            conn.sendall(b'MAC Verification success!')
    conn.close()