让加密对用例透明:继承 requests.Session 做一层切面

工程 ·

这是接口自动化系列的第二篇。第一篇讲用例设计的四条原则,这篇讲整个项目里技术上最有意思的一块。

问题

做接口自动化做到第三周,撞上一件挺麻烦的事:测试环境的接口是明文的,预发和线上是 AES 加密的。

具体说,加密环境下请求体要先 AES 加密再包一层:

{ "params": "<base64 密文>" }

响应也是加密的:

{ "result": "<base64 密文>" }

而测试环境就是普通的 JSON 进、JSON 出。

问题来了:这时候已经有 300 多条用例了,它们都是按明文写的。要支持在预发环境跑,难道每条用例里都加一段“判断环境 → 决定要不要加密”?

那会是一场灾难。300 处重复代码,每处都可能写错,而且以后新增用例还得记着这回事。更根本的是,加解密跟“这个接口的业务逻辑对不对”毫无关系——它不该出现在用例层。

思路:找一个能拦截所有请求的位置

我用的是 requests,所有请求都从 Session 发出去:

resp = auth_session.post(url, json=body)

翻了下 requests 的源码,发现一个很关键的事实:Session 的 get / post / put / delete 全部最终调用同一个方法——Session.request()。

那答案就明确了:继承 Session,重写 request()。

所有请求都会经过这一个点,在它前后做加解密,用例层完全感知不到。这就是切面(AOP)的思路,只不过 requests 没有官方的中间件机制,继承重写是最贴合它设计的切入点。

实现

import json, base64
import requests
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad


class EncryptedSession(requests.Session):
    def __init__(self, cfg):
        super().__init__()
        self._cfg = cfg

    def request(self, method, url, **kwargs):
        # ① 请求方向:明文 dict → 加密 → 包成 {"params": 密文}
        if self._cfg.USE_AES and "json" in kwargs:
            ciphertext = self._encrypt(kwargs.pop("json"))
            kwargs["data"] = json.dumps({"params": ciphertext})
            kwargs.setdefault("headers", {})["Content-Type"] = "application/json"

        # ② 真正发请求,交给父类
        resp = super().request(method, url, **kwargs)

        # ③ 响应方向:{"result": 密文} → 解密 → 偷换响应体
        if self._cfg.USE_AES:
            outer = json.loads(resp.content)
            if outer.get("result"):
                plain = self._decrypt(outer["result"])
                resp._content = json.dumps(plain).encode()

        return resp

    def _encrypt(self, data: dict) -> str:
        c = AES.new(self._cfg.AES_KEY.encode(), AES.MODE_CBC, self._cfg.AES_IV.encode())
        return base64.b64encode(c.encrypt(pad(json.dumps(data).encode(), 16))).decode()

    def _decrypt(self, b64: str) -> dict:
        c = AES.new(self._cfg.AES_KEY.encode(), AES.MODE_CBC, self._cfg.AES_IV.encode())
        return json.loads(unpad(c.decrypt(base64.b64decode(b64)), 16))

三十来行,但每一段都有讲究。

① 为什么是 pop("json") 而不是读一下

ciphertext = self._encrypt(kwargs.pop("json"))
kwargs["data"] = json.dumps({"params": ciphertext})

必须用 pop 把 json 参数摘掉。因为 requests 看到 json= 参数会自己做一遍序列化——如果留着它,就会变成“你加密的密文”和“requests 序列化的明文”同时存在,服务端收到的是一团乱麻。

摘掉 json、换成 data 传密文字符串,这是一次参数形态的转换:用例层交出来的是“我想发的业务数据”,网络层收到的是“实际该发的字节”。 这层转换正是这个类存在的意义。

② super().request() 是分寸所在

resp = super().request(method, url, **kwargs)

真正的网络请求还是交给父类。我只在它前后各做一件事,不重新实现网络层。

这条边界很重要。连接池、重试、超时、cookie 管理、代理——requests 处理得比我好得多,我没有任何理由碰它们。继承的价值在于“插进去一段”,不在于“重写一遍”。

③ resp._content —— 偷换响应体

resp._content = json.dumps(plain).encode()

这是全篇最“脏”也最关键的一行。

resp.json() 本质上是解析 resp.content,而 content 背后是 _content 这个属性。把解密后的明文写回 _content,用例拿到的 resp.json() 就是正常的明文结构——它根本不知道中间加密过。

动私有属性是有代价的,得认:这依赖了 requests 的内部实现,理论上库升级可能失效。我接受这个代价,因为替代方案(自己构造一个 Response 对象)要复制的东西更多、更容易出错。做这类取舍时我倾向于:依赖面越小越好,但可读性和可控性优先。 一行 _content 赋值,出问题时定位成本极低。

④ 开关:让同一套用例两边都能跑

if self._cfg.USE_AES and "json" in kwargs:

USE_AES 来自环境配置。测试环境是 False,这时候这个类的行为跟普通 Session 一模一样——两个 if 都不进,纯透传。

于是环境切换就退化成了一个命令行参数:

pytest --env=test    # USE_AES=False,明文
pytest --env=pre     # USE_AES=True,自动加解密

用例代码一个字都不用改。这是整个设计想达到的效果。

用例层长什么样

resp = auth_session.post(url, json={"groupID": "g"})
data = get_data(resp)          # 不管加不加密,拿到的都是明文 dict
assert data["status"] == 0

就这样。没有一行代码提到加密。

同一个位置还能挂别的东西

打通这个切面之后我发现,“所有请求的必经之路”是个很值钱的位置。除了加解密,还顺手在附近解决了两件事。

一是鉴权 header。 这个服务需要三个鉴权头:用户 ID、防重放签名、登录 token。签名是 md5(userID + 密钥) 算出来的。这些如果让每条用例自己拼,同样是 300 处重复。做法是在建 Session 的时候一次性设进 session.headers,后续所有请求自动携带:

def _make_session(cfg, base_headers, login_data):
    s = EncryptedSession(cfg)
    s.headers.update({
        **base_headers,
        "x-user-id": login_data["userID"],
        "x-sign-id": gen_sign(login_data["userID"], cfg.SIGN_KEY),
        "x-token":   login_data["token"],
    })
    return s

按角色建几个常驻 Session(管理员 / 普通成员 / 非团队成员),用例要用哪个身份就注入哪个 fixture——越权测试因此变得非常好写,换个 session 参数就是换个身份。

顺带一个 requests 的细节,用好了很省事:per-request 的 headers= 会覆盖 session 级的同名 header,且只对这一次请求生效。

# session 级默认是低版本号
auth_session.post(url, json=body)

# 这一次请求用高版本号,测 V2 分支;session 级不受影响
auth_session.post(url, json=body, headers={"versionCode": "100010001"})

需要临时切版本、切身份时,走 per-request 覆盖,不要 s.headers.update()——后者会污染 session,让后面所有用例躺枪,而且这种问题排查起来极其痛苦。

二是 SSL 校验。 预发环境用的是内部证书链,certifi 不认,请求直接报错。这个用 autouse fixture 加一次性的 monkey-patch 解决:

@pytest.fixture(scope="session", autouse=True)
def _setup_ssl_verify(cfg):
    if getattr(cfg, "VERIFY_SSL", True):
        yield
        return
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    _orig = requests.Session.request
    def _patched(self, method, url, **kwargs):
        kwargs.setdefault("verify", False)
        return _orig(self, method, url, **kwargs)
    requests.Session.request = _patched
    yield
    requests.Session.request = _orig                # 结束后还原

autouse=True 让它不用被任何用例显式声明就生效,yield 之后老实还原猴子补丁。同样,用例层零感知。

一个必须说的前提:先读服务端代码

这套东西能对上,靠的不是猜。

我是把服务端的加密实现翻出来读了一遍才动手的,读完发现一个非常规的细节:这个服务的 IV 直接复用了 Key(正常应该每次随机生成 IV 并随密文传输)。如果按教科书写法去实现客户端,怎么都对不上,只会得到一堆 padding error。

同理,签名密钥是分环境的,测试环境和线上不是一套;服务端在测试环境会优先校验测试密钥、再兜底校验线上密钥——这个逻辑也是读代码才知道的。

所以第一篇里那条建议在这儿再强调一次:接口自动化虽然是黑盒,但写之前一定要读服务端代码。 不读代码,你的框架层只能靠试错去逼近真相,而试错的成本远高于读一遍实现。

⚠️ 另外提醒一句:读到的密钥、内网地址、数据库连接串,写文档和贴代码时务必抹掉。这篇文章里所有的 key、域名、路径都是占位符。

一点总结

这个类总共三十行,但它是整个项目里我最满意的一段代码,因为它体现了一个我越来越认同的原则:

把复杂度下沉到框架层,让用例层保持简单。

用例层应该只有业务语义——发什么请求、期望什么结果。凡是“跟业务无关但每条用例都得干”的事(加解密、鉴权、签名、证书、重试、日志),都该往下沉,沉到一个用例看不见的地方。

判断一层抽象做得好不好,我现在的标准很简单:上层能不能假装它不存在。

按这个标准,EncryptedSession 是合格的——300 多条用例里,没有一条知道自己的请求被加密过。