导语：在进行网关、CDN类型产品的转发测试过程中，除了普通的HTTP|HTTPS请求，通常我们还会涉及到一些协议以及TLS层面的测试。在敏捷环境下，通常到一定时间段我们的测试工具会五花八门，不便于管理的同时，也增加了其他同事使用的学习成本。本文将介绍使用Python/ target=_blank class=infotextkey>Python的httpx库以及curl来支持绝大部分HTTP请求场景，希望能给大家带来一些帮助，欢迎大家留言讨论，一起进一步完善工具。
1.准备工作
1.1.httpx
httpx库仅支持Python3，可通过pip命令安装支持HTTP2与命令行工具。
python -m pip install httpx
# 支持HTTP2
python -m pip install 'httpx[http2]'
# 安装命令行客户端
python -m pip install 'httpx[cli]'
Python中使用示例：
>>> import httpx
>>> r = httpx.get('https://www.example.org/')
>>> r
<Response [200 OK]>
>>> r.status_code
200
>>> r.headers['content-type']
'text/html; charset=UTF-8'
>>> r.text
'<!doctype html>n<html>n<head>n<title>Example DomAIn</title>...'
命令行使用示例：
1.2.curl
建议将curl版本升级至7.68.0或以上，本文中以7.68.0为例。
为了让 curl 支持 HTTP2 我们需要安装 nghttp2（http2 的 C 语言库）：
# 从git直接拷贝项目或下载压缩包：
git clone https://Github.com/tatsuhiro-t/nghttp2.git
cd nghttp2-master
autoreconf -i
automake
autoconf
./configure
make && make install
echo '/usr/local/lib' > /etc/ld.so.conf.d/local.conf
ldconfig
升级curl版本：
yum install build-dep curl
wget http://curl.haxx.se/download/curl-7.68.0.zip
unzip curl-7.68.0.zip
cd curl-7.68.0
./configure --with-nghttp2=/usr/local --with-ssl
make && make install
ldconfig
在执行./configure时，可以从终端观察到HTTP2是enabled的状态：
2.开始
2.1.基本请求
2.1.1.httpx.Client
从上文的示例中，我们可以看到httpx库可以在import后，直接通过调用httpx.get、httpx.post等调用发起对应method的请求。在小工具中，此类调用方式无疑是方便、快捷且实用的。而在工程中，则推荐使用httpx.Client或httpx.AsyncClient，并通过调用示例化对象client的request方法代替直接调用get、post等来使代码更灵活。
with httpx.Client() as client:
client.request(
method=method,
url=req_url,
headers=req_headers,
content=content
)
2.1.2.curl
curl <url> -X <method> -d <data> -H <header>
# POST或PUT大文件时，建议使用 -F 'file=@<filename>'
# HEAD请求建议直接使用-I，而不是 -X HEAD
2.2.chunked
有时候我们需要通过在Client端构造chunked请求，来验证反向代理的模块对此类请求处理的准确性。但是现有的工具却没有一个明确的参数方便我们构造此类请求。
首先，chunked是指分块传输编码（Chunked transfer encoding），允许客户端或服务端将body分成不确定的多块进行传输。
而通过help，可以看到httpx请求方法中的content参数是允许传入一个byte iterator的。
那么我们只要实现一个生成器，将content进行切块，再将生成器作为参数传给httpx.request即可。
if req_chunked:
_content = content
_middle = _content.__len__() // 2
async def content():
yield _content[:_middle]
yield _content[_middle:]
else:
pass
async with httpx.AsyncClient(http2=http2) as client:
task = asyncio.create_task(
client.request(
method=method,
url=req_url,
headers=req_headers,
content=content() if callable(content) else content
)
)
try:
await task
except asyncio.CancelledError:
pass
2.3.HTTP2
备注：成功完成一个HTTP2请求，需要Server端也支持HTTP2协议。
由于httpx库对HTTP2有较完善的支持，发起HTTP2请求也非常方便，只需在实例化Client时指定参数http2=True即可。
with httpx.Client(http2=True) as client:
client.request(
method=method,
url=req_url,
headers=req_headers,
content=content
)
curl命令则只需带上--http2参数即可。
curl <url> --http2
2.3.1.构造多路复用请求
多路复用 —— 一个连接中的请求是非阻塞的，即同连接中的请求可并发，存在多个stream。我们通过httpx.AsyncClient来构造：
async with httpx.AsyncClient(http2=http2) as client:
for i in range(req_num):
task_list = list()
for m in range(multiplexing): # multiplexing，指定stream的数量
task = asyncio.create_task(
client.request(
method=method,
url=req_url,
headers=req_headers,
content=content
)
)
task_list.Append(task)
await asyncio.wait(task_list)
for t in task_list:
try:
await t
except _err_type:
pass
for t in task_list:
if t.exception():
logger.warning(t.exception())
else:
pass
response_list.append(t.result())
2.4.SSL/TLS
在SSL/TLS协议相关测试过程中，通常需要验证系统的各种功能在和不同的Cipher suite（密码套件）以及SSL/TLS版本结合的情况下，运作能力是否付符合设计预期。
在Python中，存在一个内置的库：ssl，它能满足测试验证的大部分述求。需要注意的是，ssl库依赖于设备上的OpenSSL。
在curl中，则可以通过--ciphers、--tls-max、--tlsv1.x等参数来满足述求。
2.4.1.Ciphers
httpx+ssl：
import httpx
import ssl
### 创建默认SSL上下文 ###
ssl_ctx = ssl.create_default_context()
### 关闭单向认证场景的证书校验 ###
# 修改校验标志位
ssl_ctx.verify_flags = ssl.VerifyFlags.VERIFY_DEFAULT
# 关闭域名校验
ssl_ctx.check_hostname = False
# 关闭证书校验
ssl_ctx.verify_mode=ssl.VerifyMode.CERT_NONE
### 设置SSL上下文的cipher suite ###
ssl_ctx.set_ciphers("AES128-GCM-SHA256")
### 发起HTTPS单向认证请求 ###
url = "https://<vip>:<vport>/"
rsp = httpx.get(url, verify=ssl_ctx)
"""
>>> rsp.status_code
200
>>> rsp.headers
Headers({'date': 'Fri, 01 Jul 2022 08:25:55 GMT', 'content-type': 'application/octet-stream', 'transfer-encoding': 'chunked', 'connection': 'keep-alive', 'server': 'openresty/1.17.8.1', 'cid': '215253869784', 'cipher': 'AES128-GCM-SHA256'})
"""
# 查询cipher suite，如果调用过set_ciphers()，则只会返回配置的ciphers信息
ssl_ctx.get_ciphers()
"""
[{'id': 50380848,
'name': 'ECDHE-RSA-AES256-GCM-SHA384',
'protocol': 'TLSv1/SSLv3',
'description': 'ECDHE-RSA-AES256-GCM-SHA384 TLSv1.2 Kx=ECDH Au=RSA Enc=AESGCM(256) mac=AEAD',
'strength_bits': 256,
'alg_bits': 256},
{'id': 50380844,
'name': 'ECDHE-ECDSA-AES256-GCM-SHA384',
'protocol': 'TLSv1/SSLv3',
'description': 'ECDHE-ECDSA-AES256-GCM-SHA384 TLSv1.2 Kx=ECDH Au=ECDSA Enc=AESGCM(256) Mac=AEAD',
'strength_bits': 256,
'alg_bits': 256},
...
]
"""
curl：
curl <https url> -k --cipher ECDHE-RSA-AES256-GCM-SHA384
查询系统支持哪些cipher suite：
openssl ciphers
2.4.2.指定SSL/TLS版本
httpx+ssl：
import httpx
import ssl
### 创建默认SSL上下文 ###
ssl_ctx = ssl.create_default_context()
### 关闭单向认证场景的证书校验 ###
# 修改校验标志位
ssl_ctx.verify_flags = ssl.VerifyFlags.VERIFY_DEFAULT
# 关闭域名校验
ssl_ctx.check_hostname = False
# 关闭证书校验
ssl_ctx.verify_mode=ssl.VerifyMode.CERT_NONE
### 设置SSL协议版本 ###
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_1)
# 校验SSL上下文对象的协议版本号
ssl_ctx.protocol
"""
<_SSLMethod.PROTOCOL_TLSv1_1: 4>
"""
### 发起HTTPS单向认证请求 ###
url = "https://<vip>:<vport>/"
rsp = httpx.get(url, verify=ssl_ctx)
"""
>>> rsp.status_code
200
>>> rsp.headers
Headers({'date': 'Mon, 04 Jul 2022 06:28:08 GMT', 'content-type': 'application/octet-stream', 'transfer-encoding': 'chunked', 'connection': 'keep-alive', 'server': 'openresty/1.17.8.1', 'cid': '218525668279', 'cipher': 'ECDHE-RSA-AES128-SHA', 'ssl_protocol': 'TLSv1.1'})
"""
curl：
# 示例为TLSv1.2，其他版本以此类推。
# 注意：最好指定--tls-max，否则请求会优先按支持的最新版本协商。
curl <https url> -k --tlsv1.2 --tls-max 1.2
2.5.Resolve/SNI
SNI（Server Name Indication）是为了解决一个服务器使用多个域名和证书的TLS扩展，主要解决一台服务器只能使用一个证书的缺点。
在验证域名查找与SNI的场景时，需要Client支持域名解析。通常的临时做法是，去修改系统中的/etc/hosts，固定被测IP地址与域名的映射。这么做虽然方便快捷，但是也存在Client共用时的解析优先级冲突，验证IP地址与域名多对多时需要频繁修改调整等问题。所以，接下来我们来一起看看怎么在Python和curl的运行时动态地去resolve，以解决上述问题。
Python，通过重写socket底层getaddrinfo方法，将域名与IP地址在运行时动态篡改为预期的映射关系。然后使用httpx，直接对域名发起请求即可。
重写socket底层getaddrinfo方法示例代码：
import ipaddress
import socket
from loguru import logger
class DNS(object):
DNS_CACHE = dict()
def __init__(self):
super(DNS, self).__init__()
self.socket_get_address_info = socket.getaddrinfo
socket.getaddrinfo = self.custom_get_address_info
def add_custom_dns(
self,
domain: str,
port: int,
ip: str
):
key = (domain.encode("utf-8"), port)
if ipaddress.ip_address(ip).version == 4:
value = (
socket.AddressFamily.AF_.NET,
socket.SocketKind.SOCK_STREAM,
socket.IPPROTO_TCP,
'',
(ip, port)
)
else:
value = (
socket.AddressFamily.AF_INET6,
socket.SocketKind.SOCK_STREAM,
socket.IPPROTO_TCP,
'',
(ip, port, 0, 0)
)
self.DNS_CACHE[key] = [value]
logger.debug(f"DNS_Cache: {self.DNS_CACHE}")
return None
def custom_get_address_info(
self,
*args
):
logger.debug(
f"Args: {args}"
)
try:
if isinstance(args[0], str):
key = (args[0].encode("utf8"), args[1])
else:
key = args[:2]
return self.DNS_CACHE[key]
except KeyError:
return self.socket_get_address_info(*args)
curl，使用--resolve参数，域名与IP地址的映射关系仅在当次请求时生效：
curl <url> --resolve <domain>:<port>:
作者:ten
出处
:https://mp.weixin.qq.com/s/xxP-MVgdASZ324MLkv1-Vw
京公网安备 11010802035086号