導(dǎo)語：在進(jìn)行網(wǎng)關(guān)、CDN類型產(chǎn)品的轉(zhuǎn)發(fā)測(cè)試過程中，除了普通的HTTP|HTTPS請(qǐng)求，通常我們還會(huì)涉及到一些協(xié)議以及TLS層面的測(cè)試。在敏捷環(huán)境下，通常到一定時(shí)間段我們的測(cè)試工具會(huì)五花八門，不便于管理的同時(shí)，也增加了其他同事使用的學(xué)習(xí)成本。本文將介紹使用Python/ target=_blank class=infotextkey>Python的httpx庫以及curl來支持絕大部分HTTP請(qǐng)求場(chǎng)景，希望能給大家?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版本升級(jí)至7.68.0或以上，本文中以7.68.0為例。

為了讓 curl 支持 HTTP2 我們需要安裝 nghttp2（http2 的 C 語言庫）：

# 從git直接拷貝項(xiàng)目或下載壓縮包：
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

升級(jí)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時(shí)，可以從終端觀察到HTTP2是enabled的狀態(tài)：

2.開始

2.1.基本請(qǐng)求

2.1.1.httpx.Client

從上文的示例中，我們可以看到httpx庫可以在import后，直接通過調(diào)用httpx.get、httpx.post等調(diào)用發(fā)起對(duì)應(yīng)method的請(qǐng)求。在小工具中，此類調(diào)用方式無疑是方便、快捷且實(shí)用的。而在工程中，則推薦使用httpx.Client或httpx.AsyncClient，并通過調(diào)用示例化對(duì)象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大文件時(shí)，建議使用 -F 'file=@<filename>'
# HEAD請(qǐng)求建議直接使用-I，而不是 -X HEAD

2.2.chunked

有時(shí)候我們需要通過在Client端構(gòu)造chunked請(qǐng)求，來驗(yàn)證反向代理的模塊對(duì)此類請(qǐng)求處理的準(zhǔn)確性。但是現(xiàn)有的工具卻沒有一個(gè)明確的參數(shù)方便我們構(gòu)造此類請(qǐng)求。

首先，chunked是指分塊傳輸編碼（Chunked transfer encoding），允許客戶端或服務(wù)端將body分成不確定的多塊進(jìn)行傳輸。

而通過help，可以看到httpx請(qǐng)求方法中的content參數(shù)是允許傳入一個(gè)byte iterator的。

那么我們只要實(shí)現(xiàn)一個(gè)生成器，將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

備注：成功完成一個(gè)HTTP2請(qǐng)求，需要Server端也支持HTTP2協(xié)議。

由于httpx庫對(duì)HTTP2有較完善的支持，發(fā)起HTTP2請(qǐng)求也非常方便，只需在實(shí)例化Client時(shí)指定參數(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ù)用請(qǐng)求

多路復(fù)用 —— 一個(gè)連接中的請(qǐng)求是非阻塞的，即同連接中的請(qǐng)求可并發(fā)，存在多個(gè)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)測(cè)試過程中，通常需要驗(yàn)證系統(tǒng)的各種功能在和不同的Cipher suite（密碼套件）以及SSL/TLS版本結(jié)合的情況下，運(yùn)作能力是否付符合設(shè)計(jì)預(yù)期。

在Python中，存在一個(gè)內(nèi)置的庫：ssl，它能滿足測(cè)試驗(yàn)證的大部分述求。需要注意的是，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)證場(chǎng)景的證書校驗(yàn) ###
# 修改校驗(yàn)標(biāo)志位
ssl_ctx.verify_flags = ssl.VerifyFlags.VERIFY_DEFAULT
# 關(guān)閉域名校驗(yàn)
ssl_ctx.check_hostname = False
# 關(guān)閉證書校驗(yàn)
ssl_ctx.verify_mode=ssl.VerifyMode.CERT_NONE


### 設(shè)置SSL上下文的cipher suite ###
ssl_ctx.set_ciphers("AES128-GCM-SHA256")


### 發(fā)起HTTPS單向認(rèn)證請(qǐng)求 ###
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()，則只會(huì)返回配置的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)證場(chǎng)景的證書校驗(yàn) ###
# 修改校驗(yàn)標(biāo)志位
ssl_ctx.verify_flags = ssl.VerifyFlags.VERIFY_DEFAULT
# 關(guān)閉域名校驗(yàn)
ssl_ctx.check_hostname = False
# 關(guān)閉證書校驗(yàn)
ssl_ctx.verify_mode=ssl.VerifyMode.CERT_NONE


### 設(shè)置SSL協(xié)議版本 ###
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_1)
# 校驗(yàn)SSL上下文對(duì)象的協(xié)議版本號(hào)
ssl_ctx.protocol
"""
<_SSLMethod.PROTOCOL_TLSv1_1: 4>
"""


### 發(fā)起HTTPS單向認(rèn)證請(qǐng)求 ###
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，否則請(qǐng)求會(huì)優(yōu)先按支持的最新版本協(xié)商。
curl <https url> -k --tlsv1.2 --tls-max 1.2

2.5.Resolve/SNI

SNI（Server Name Indication）是為了解決一個(gè)服務(wù)器使用多個(gè)域名和證書的TLS擴(kuò)展，主要解決一臺(tái)服務(wù)器只能使用一個(gè)證書的缺點(diǎn)。

在驗(yàn)證域名查找與SNI的場(chǎng)景時(shí)，需要Client支持域名解析。通常的臨時(shí)做法是，去修改系統(tǒng)中的/etc/hosts，固定被測(cè)IP地址與域名的映射。這么做雖然方便快捷，但是也存在Client共用時(shí)的解析優(yōu)先級(jí)沖突，驗(yàn)證IP地址與域名多對(duì)多時(shí)需要頻繁修改調(diào)整等問題。所以，接下來我們來一起看看怎么在Python和curl的運(yùn)行時(shí)動(dòng)態(tài)地去resolve，以解決上述問題。

Python，通過重寫socket底層getaddrinfo方法，將域名與IP地址在運(yùn)行時(shí)動(dòng)態(tài)篡改為預(yù)期的映射關(guān)系。然后使用httpx，直接對(duì)域名發(fā)起請(qǐng)求即可。

重寫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)次請(qǐng)求時(shí)生效：