導(dǎo)語：在進(jìn)行網(wǎng)關(guān)、CDN類型產(chǎn)品的轉(zhuǎn)發(fā)測試過程中，除了普通的HTTP|HTTPS請求，通常我們還會涉及到一些協(xié)議以及TLS層面的測試。在敏捷環(huán)境下，通常到一定時間段我們的測試工具會五花八門，不便于管理的同時，也增加了其他同事使用的學(xué)習(xí)成本。本文將介紹使用Python/ target=_blank class=infotextkey>Python的httpx庫以及curl來支持絕大部分HTTP請求場景，希望能給大家?guī)硪恍椭瑲g迎大家留言討論，一起進(jìn)一步完善工具。
1.準(zhǔn)備工作
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
在執(zhí)行./configure時，可以從終端觀察到HTTP2是enabled的狀態(tài)：
2.開始
2.1.基本請求
2.1.1.httpx.Client
從上文的示例中，我們可以看到httpx庫可以在import后，直接通過調(diào)用httpx.get、httpx.post等調(diào)用發(fā)起對應(yīng)method的請求。在小工具中，此類調(diào)用方式無疑是方便、快捷且實用的。而在工程中，則推薦使用httpx.Client或httpx.AsyncClient，并通過調(diào)用示例化對象client的request方法代替直接調(diào)用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端構(gòu)造chunked請求，來驗證反向代理的模塊對此類請求處理的準(zhǔn)確性。但是現(xiàn)有的工具卻沒有一個明確的參數(shù)方便我們構(gòu)造此類請求。
首先，chunked是指分塊傳輸編碼（Chunked transfer encoding），允許客戶端或服務(wù)端將body分成不確定的多塊進(jìn)行傳輸。
而通過help，可以看到httpx請求方法中的content參數(shù)是允許傳入一個byte iterator的。
那么我們只要實現(xiàn)一個生成器，將content進(jìn)行切塊，再將生成器作為參數(shù)傳給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協(xié)議。
由于httpx庫對HTTP2有較完善的支持，發(fā)起HTTP2請求也非常方便，只需在實例化Client時指定參數(shù)http2=True即可。
with httpx.Client(http2=True) as client:
client.request(
method=method,
url=req_url,
headers=req_headers,
content=content
)
curl命令則只需帶上--http2參數(shù)即可。
curl <url> --http2
2.3.1.構(gòu)造多路復(fù)用請求
多路復(fù)用 —— 一個連接中的請求是非阻塞的，即同連接中的請求可并發(fā)，存在多個stream。我們通過httpx.AsyncClient來構(gòu)造：
async with httpx.AsyncClient(http2=http2) as client:
for i in range(req_num):
task_list = list()
for m in range(multiplexing): # multiplexing，指定stream的數(shù)量
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協(xié)議相關(guān)測試過程中，通常需要驗證系統(tǒng)的各種功能在和不同的Cipher suite（密碼套件）以及SSL/TLS版本結(jié)合的情況下，運作能力是否付符合設(shè)計預(yù)期。
在Python中，存在一個內(nèi)置的庫：ssl，它能滿足測試驗證的大部分述求。需要注意的是，ssl庫依賴于設(shè)備上的OpenSSL。
在curl中，則可以通過--ciphers、--tls-max、--tlsv1.x等參數(shù)來滿足述求。
2.4.1.Ciphers
httpx+ssl：
import httpx
import ssl
### 創(chuàng)建默認(rèn)SSL上下文 ###
ssl_ctx = ssl.create_default_context()
### 關(guān)閉單向認(rèn)證場景的證書校驗 ###
# 修改校驗標(biāo)志位
ssl_ctx.verify_flags = ssl.VerifyFlags.VERIFY_DEFAULT
# 關(guān)閉域名校驗
ssl_ctx.check_hostname = False
# 關(guān)閉證書校驗
ssl_ctx.verify_mode=ssl.VerifyMode.CERT_NONE
### 設(shè)置SSL上下文的cipher suite ###
ssl_ctx.set_ciphers("AES128-GCM-SHA256")
### 發(fā)起HTTPS單向認(rèn)證請求 ###
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，如果調(diào)用過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
查詢系統(tǒng)支持哪些cipher suite：
openssl ciphers
2.4.2.指定SSL/TLS版本
httpx+ssl：
import httpx
import ssl
### 創(chuàng)建默認(rèn)SSL上下文 ###
ssl_ctx = ssl.create_default_context()
### 關(guān)閉單向認(rèn)證場景的證書校驗 ###
# 修改校驗標(biāo)志位
ssl_ctx.verify_flags = ssl.VerifyFlags.VERIFY_DEFAULT
# 關(guān)閉域名校驗
ssl_ctx.check_hostname = False
# 關(guān)閉證書校驗
ssl_ctx.verify_mode=ssl.VerifyMode.CERT_NONE
### 設(shè)置SSL協(xié)議版本 ###
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_1)
# 校驗SSL上下文對象的協(xié)議版本號
ssl_ctx.protocol
"""
<_SSLMethod.PROTOCOL_TLSv1_1: 4>
"""
### 發(fā)起HTTPS單向認(rèn)證請求 ###
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，否則請求會優(yōu)先按支持的最新版本協(xié)商。
curl <https url> -k --tlsv1.2 --tls-max 1.2
2.5.Resolve/SNI
SNI（Server Name Indication）是為了解決一個服務(wù)器使用多個域名和證書的TLS擴(kuò)展，主要解決一臺服務(wù)器只能使用一個證書的缺點。
在驗證域名查找與SNI的場景時，需要Client支持域名解析。通常的臨時做法是，去修改系統(tǒng)中的/etc/hosts，固定被測IP地址與域名的映射。這么做雖然方便快捷，但是也存在Client共用時的解析優(yōu)先級沖突，驗證IP地址與域名多對多時需要頻繁修改調(diào)整等問題。所以，接下來我們來一起看看怎么在Python和curl的運行時動態(tài)地去resolve，以解決上述問題。
Python，通過重寫socket底層getaddrinfo方法，將域名與IP地址在運行時動態(tài)篡改為預(yù)期的映射關(guān)系。然后使用httpx，直接對域名發(fā)起請求即可。
重寫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參數(shù)，域名與IP地址的映射關(guān)系僅在當(dāng)次請求時生效：
curl <url> --resolve <domain>:<port>:
3.參考資料
- https://www.python-httpx.org
- pypi.org/project/httpx/
- requests.readthedocs.io
作者:ten
來源:微信公眾號:鵝廠架構(gòu)師
出處
:https://mp.weixin.qq.com/s/xxP-MVgdASZ324MLkv1-Vw
