3. HTTPS安全下载

HTTPS安全下载 学习卡片

任务目标

在现有OTA客户端框架中，将固件下载通道从HTTP升级为HTTPS，实现服务器证书校验，确保固件传输过程的机密性与完整性。本案例聚焦于下载通道的安全改造，不涉及固件签名校验（属于另一独立环节）。

环境依赖

  • 运行环境：Linux（Ubuntu 20.04+）或 macOS，需具备网络访问能力
  • Python 3.8+，依赖库：requests（HTTP客户端）、cryptography（生成测试证书）
  • 安装命令：pip install requests cryptography

核心实现

以下代码实现一个完整的HTTPS固件下载客户端，包含：证书固定校验（使用预置CA证书）、下载进度回调、断点续传支持，以及针对证书验证失败、连接异常、校验和不匹配三类错误的明确处理。服务器地址通过环境变量注入，便于切换测试环境。

import os
import ssl
import hashlib
import tempfile
from pathlib import Path
import requests
from requests.adapters import HTTPAdapter
from urllib3.poolmanager import PoolManager

class FirmwareDownloader:
    """HTTPS固件下载客户端，支持证书固定与断点续传"""

    def __init__(self, ca_cert_path: str, firmware_url: str, dest_path: str):
        self.ca_cert = ca_cert_path
        self.url = firmware_url
        self.dest = Path(dest_path)
        self.session = self._create_secure_session()

    def _create_secure_session(self) -> requests.Session:
        """创建强制证书校验的会话，禁用证书警告"""
        session = requests.Session()
        session.verify = self.ca_cert  # 固定CA证书，不使用系统默认
        session.headers.update({"User-Agent": "OTA-Client/3.0"})

        # 自定义适配器：仅允许TLS1.2+，禁用不安全重定向
        class TlsAdapter(HTTPAdapter):
            def init_poolmanager(self, *args, **kwargs):
                ctx = ssl.create_default_context(cafile=self.ca_cert)
                ctx.minimum_version = ssl.TLSVersion.TLSv1_2
                kwargs["ssl_context"] = ctx
                return super().init_poolmanager(*args, **kwargs)

        session.mount("https://", TlsAdapter())
        return session

    def download(self, expected_sha256: str = None) -> bool:
        """执行下载，返回是否成功。支持断点续传与完整性校验"""
        headers = {}
        if self.dest.exists():
            headers["Range"] = f"bytes={self.dest.stat().st_size}-"

        try:
            with self.session.get(self.url, headers=headers, stream=True, timeout=(5, 30)) as resp:
                resp.raise_for_status()

                # 处理断点续传：服务器返回206或200
                mode = "ab" if resp.status_code == 206 else "wb"
                if resp.status_code == 200 and self.dest.exists():
                    self.dest.unlink()  # 服务器不支持续传，重新下载

                sha = hashlib.sha256()
                with open(self.dest, mode) as f:
                    for chunk in resp.iter_content(chunk_size=64 * 1024):
                        if chunk:
                            f.write(chunk)
                            sha.update(chunk)

                if expected_sha256:
                    actual = sha.hexdigest()
                    if actual != expected_sha256:
                        raise ValueError(f"SHA256 mismatch: expected {expected_sha256}, got {actual}")
                return True

        except requests.exceptions.SSLError as e:
            print(f"[证书错误] 服务器证书验证失败: {e}")
        except requests.exceptions.ConnectionError as e:
            print(f"[连接错误] 无法建立安全连接: {e}")
        except requests.exceptions.HTTPError as e:
            print(f"[HTTP错误] 服务器返回异常状态: {e}")
        except ValueError as e:
            print(f"[完整性错误] {e}")
        return False

# ---------- 最小验证入口 ----------
if __name__ == "__main__":
    # 通过环境变量注入配置，避免硬编码
    CA_PATH = os.environ.get("OTA_CA_CERT", "test_ca.pem")
    FW_URL = os.environ.get("OTA_FW_URL", "https://localhost:8443/firmware.bin")
    DEST_FILE = os.environ.get("OTA_DEST", "downloaded_fw.bin")
    EXPECTED_SHA = os.environ.get("OTA_EXPECTED_SHA", "")

    downloader = FirmwareDownloader(CA_PATH, FW_URL, DEST_FILE)
    ok = downloader.download(expected_sha256=EXPECTED_SHA or None)
    print(f"下载{'成功' if ok else '失败'}: {DEST_FILE}")

运行与验证

本案例需要本地HTTPS测试服务器。使用以下命令生成自签名CA与服务器证书，并启动一个简单的HTTPS文件服务（需另存为test_server.py）：

# 1. 生成测试证书（仅用于本地验证）
openssl req -x509 -newkey rsa:2048 -nodes -keyout ca_key.pem -out test_ca.pem -days 30 -subj "/CN=TestOTA_CA"
openssl req -newkey rsa:2048 -nodes -keyout server_key.pem -out server.csr -subj "/CN=localhost"
openssl x509 -req -in server.csr -CA test_ca.pem -CAkey ca_key.pem -CAcreateserial -out server_cert.pem -days 7 -extfile <(echo "subjectAltName=DNS:localhost,IP:127.0.0.1")

# 2. 启动HTTPS文件服务器（Python 3.8+）
python3 -c "
import http.server, ssl, os
os.chdir('/tmp')  # 放置firmware.bin的目录
with open('firmware.bin','wb') as f: f.write(b'OTA_TEST_FIRMWARE_'*1000)  # 生成测试固件
srv = http.server.HTTPServer(('localhost', 8443), http.server.SimpleHTTPRequestHandler)
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.load_cert_chain('server_cert.pem', 'server_key.pem')
srv.socket = ctx.wrap_socket(srv.socket, server_side=True)
print('HTTPS server on https://localhost:8443')
srv.serve_forever()
" &
sleep 1

# 3. 运行下载客户端（使用生成的CA证书）
export OTA_CA_CERT=test_ca.pem
export OTA_FW_URL=https://localhost:8443/firmware.bin
export OTA_DEST=/tmp/downloaded_fw.bin
export OTA_EXPECTED_SHA=$(sha256sum /tmp/firmware.bin | awk '{print $1}')
python3 firmware_downloader.py

通过条件：

  • 终端输出下载成功，且/tmp/downloaded_fw.bin与/tmp/firmware.bin的SHA256一致
  • 将OTA_CA_CERT指向错误CA文件（如server_cert.pem）重新运行，应输出[证书错误]且下载失败——证明证书校验生效
  • 将OTA_EXPECTED_SHA改为任意错误值，应输出[完整性错误]——证明校验和验证生效

常见问题

  • 证书验证失败：确认客户端使用的CA文件与服务器证书由同一CA签发；服务器地址必须匹配证书中的subjectAltName（使用localhost而非127.0.0.1可避免IP匹配问题）
  • 连接被拒绝：确认服务器已启动且端口未被占用；检查防火墙设置
  • 断点续传不生效：部分简易服务器（如Python内置）不支持Range请求，此时客户端会自动回退为全量下载（代码已处理）
← 上一章 📖 返回目录 下一章 →