導語：在進行網關、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>:
3.參考資料
- https://www.python-httpx.org
- pypi.org/project/httpx/
- requests.readthedocs.io
作者:ten
來源:微信公眾號:鵝廠架構師
出處
:https://mp.weixin.qq.com/s/xxP-MVgdASZ324MLkv1-Vw
