在flask框架中实现mTLS双向加密通信

前言

mTLS，全称为双向TLS（Mutual Transport Layer Security），是一种安全通信协议，确保通信双方在传输层进行身份验证。与单向HTTPS不同，吗TLS不仅要求客户端验证服务端的身份，还要求服务端验证客户端的身份。

生成证书

常见的证书生成方式就是自签名证书，有条件的话还是自建CA机构或者购买专门的证书。

<li>生成ca根证书。生成过程中会提示填写密码，记住密码，后续会用到。这里直接声明Common Name为<code>qw.er.com，可根据实际需求修改ON、OU、CN等信息。li>

openssl req -x509 -newkey rsa:4096 -keyout ca.key -out ca.crt -subj "/CN=qw.er.com" -days 3650

<li>新建并编辑<code>openssl.cnf文件，根据实际需求修改CN、CN等信息li>

[req]

req_extensions = v3_req

distinguished_name = req_distinguished_name

prompt = no

[req_distinguished_name]

countryName = CN

stateOrProvinceName = Anhui

localityName = Hefei

organizationName = zhangsan

commonName = qw.er.com

[v3_req]

subjectAltName = @alt_names

[alt_names]

DNS.1 = qw.er.com

<li>创建服务端证书。生成私钥文件时会提示输入根证书密码。

<code>openssl req -newkey rsa:2048 -nodes -keyout server.key -out server.csr -subj "/CN=qw.er.com" -config openssl.cnf

openssl x509 -req -in server.csr -out server.crt -CA ca.crt -CAkey ca.key -CAcreateserial -days 365 -extensions v3_req -extfile openssl.cnf

<li>创建客户端证书。生成私钥文件时会提示输入根证书密码

<code>openssl req -newkey rsa:2048 -nodes -keyout client.key -out client.csr -subj "/CN=qw.er.com" -config openssl.cnf

openssl x509 -req -in client.csr -out client.crt -CA ca.crt -CAkey ca.key -CAcreateserial -days 365 -extensions v3_req -extfile openssl.cnf

服务端示例

除了flask本身，只需要导入标准库ssl。注意证书路径需要是绝对路径。

from flask import Flask

import ssl

app = Flask(__name__)

@app.get("/health")

def health():

return "ok"

def get_ssl_context():

ca_crt_path = "/home/atlas/workspace/flask-learn/certs/ca.crt"

server_crt = "/home/atlas/workspace/flask-learn/certs/server.crt"

server_key = "/home/atlas/workspace/flask-learn/certs/server.key"

password = "qwerasdf"

context = ssl.SSLContext(protocol=ssl.PROTOCOL_TLS_SERVER)

context.verify_mode = ssl.CERT_REQUIRED

context.load_verify_locations(ca_crt_path)

context.load_cert_chain(certfile=server_crt, keyfile=server_key, password=password)

return context

if __name__ == "__main__":

ssl_context = get_ssl_context()

app.run(host="0.0.0.0", port=5000, ssl_context=ssl_context)code>

客户端示例

<li>方式1：使用curl命令

curl -k --cacert ca.crt --key client.key --cert client.crt https://127.0.0.1:5000/health

• 方式2：使用requests库。注意修改hosts文件将qw.er.com映射到IP

import requests

ca_crt_path = "/home/atlas/workspace/flask-learn/certs/ca.crt"

client_crt = "/home/atlas/workspace/flask-learn/certs/client.crt"

client_key = "/home/atlas/workspace/flask-learn/certs/client.key"

url = "https://qw.er.com:5000/health"

resp = requests.get(url, cert=(client_crt, client_key), verify=ca_crt_path)

print(resp.status_code)

print(resp.text)

• 方式3：使用httpx库。注意修改hosts文件将qw.er.com映射到IP

import httpx

import ssl

def get_ssl_context():

password = "qwerasdf"

ca_crt_path = "/home/atlas/workspace/flask-learn/certs/ca.crt"

client_crt = "/home/atlas/workspace/flask-learn/certs/client.crt"

client_key = "/home/atlas/workspace/flask-learn/certs/client.key"

context = ssl.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT)

context.verify_mode = ssl.CERT_REQUIRED

context.load_verify_locations(ca_crt_path)

context.load_cert_chain(certfile=client_crt, keyfile=client_key, password=password)

return context

url = "https://qw.er.com:5000/health"

context = get_ssl_context()

with httpx.Client(verify=context) as client:

resp = client.get(url)

print(resp.status_code)

print(resp.text)