HTTP請求工具之「拿來吧你」

由 閃念基因 發佈 2024-04-29T19:40:38.529193+00:00

為了讓 curl 支持 HTTP2 我們需要安裝 nghttp2:升級curl版本:在執行./configure時，可以從終端觀察到HTTP2是enabled的狀態:

導語：在進行網關、CDN類型產品的轉發測試過程中，除了普通的HTTP|HTTPS請求，通常我們還會涉及到一些協議以及TLS層面的測試。在敏捷環境下，通常到一定時間段我們的測試工具會五花八門，不便於管理的同時，也增加了其他同事使用的學習成本。本文將介紹使用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_INET,
                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.參考資料

  1. https://www.python-httpx.org
  2. pypi.org/project/httpx/
  3. requests.readthedocs.io

作者:ten

來源:微信公眾號:鵝廠架構師

出處:https://mp.weixin.qq.com/s/xxP-MVgdASZ324MLkv1-Vw

關鍵字：
#code #程式開發

【amz獨家組合】Relove_私密淨味_胺基酸潔淨凝露

售價 NT$ 560-2,380

MAGICOM美肌之鏡_專業質感_髮雕(英式油蠟/日系灰蠟)

售價 NT$ 900-1,550

【amz嚴選】可愛創意_動物疊疊樂餐具

售價 NT$ 249

reDance瑞丹絲_植萃草本_抗痘洗面皂

售價 NT$ 290

EVELINE_巴西_蔬果酵素_清暢素膠囊

售價 NT$ 450-1,791

【amz嚴選】吸水速乾_珊瑚絨舒服毛巾

售價 NT$ 290-590

reDance_肌膚光溜溜淨白霜

售價 NT$ 1,180

【E‧Heart】機能型壓力褲裙 S-M/L-LL/3L

售價 NT$ 799