在現代 Web 開發中，資料序列化是不可或缺的一環。本文將探討如何在 Python 環境下安全地處理序列化，並結合 API 版本控制策略，確保系統的穩定性和安全性。我們將以 Sensor 生態系統為例，逐步說明如何使用 JSON 和 Pickle 進行序列化，以及如何使用 HMAC 進行簽章驗證以防止資料竄改。同時，我們也會探討如何設計和實作多版本 API，以因應不斷變化的需求，並確保向後相容性。最後，我們將提供程式碼重構和測試的最佳實踐，以提升程式碼品質和可維護性。
序列化問題的解決方案與實作細節
在前面的章節中，我們探討瞭如何利用WSGI伺服器將現有的Sensor生態系統進行整合，並著重於解決序列化問題。本章節將探討序列化問題的解決方案，並提供詳細的實作細節。
序列化格式的選擇
在我們的例子中，主要使用了JSON協定進行序列化。然而，並非所有的類別都能與JSON相容。因此，我們需要考慮使用更通用的序列化器，如pickle。
使用Pickle的風險
雖然pickle是一個強大的序列化工具，但它也存在安全風險。未經驗證的pickle資料可能會導致任意程式碼的執行。因此，在使用pickle時，我們必須謹慎處理。
使用HMAC進行簽章驗證
為了提高pickle的安全性，我們可以使用HMAC（Keyed-Hash Message Authentication Code）進行簽章驗證。這樣可以確保pickle資料的完整性和真實性。
簽章與驗證的實作
import hashlib
import hmac
import pickle
# 定義金鑰
secret = bytearray([0xb2, 0x56, 0xc4, 0x88, 0x09, 0xa0, 0x8a, 0x1e, 0x28, 0xe3, 0xa3, 0x25, 0xe9, 0x2b, 0x98, 0x6f, 0x13, 0x60, 0xfb, 0x26, 0x06, 0x9b, 0x9d, 0x6f, 0x3a, 0x01, 0x2c, 0x3f, 0x9d, 0x9f, 0x72, 0xcd])
# 簽章
def sign_pickle(data):
untrusted_pickle = pickle.dumps(data)
digest = hmac.digest(secret, untrusted_pickle, hashlib.sha256)
signed_pickle = digest + b":" + untrusted_pickle
return signed_pickle
# 驗證
def verify_pickle(signed_pickle):
digest, untrusted = signed_pickle.split(b":", 1)
expected_digest = hmac.digest(secret, untrusted, hashlib.sha256)
if not hmac.compare_digest(digest, expected_digest):
raise ValueError("Bad Signature")
else:
return pickle.loads(untrusted)
#### 簽章與驗證的邏輯解析
簽章過程中，我們首先使用`pickle.dumps()`將資料序列化。接著，使用`hmac.digest()`計算序列化資料的HMAC值。最後，將HMAC值與序列化資料結合，形成簽章後的pickle資料。
在驗證過程中，我們首先將簽章後的pickle資料分割為HMAC值和序列化資料。然後，重新計算序列化資料的HMAC值，並與原始HMAC值進行比較。如果兩者一致，則驗證透過，否則丟擲`ValueError`。
### 將所有內容整合在一起
在我們的平行世界中，我們成功地將WSGI伺服器整合到現有的Sensor生態系統中。主要的變更在於`sensor_values()`檢視函式中增加了if-else陳述式，以處理序列化問題。
#### WSGI伺服器與Fallback編碼的實作
```python
from abc import ABC, abstractmethod
import typing as t
import json
import flask
from apd.sensors.sensors import Sensor
from apd.sensors.cli import get_sensors
from apd.sensors.wsgi import require_api_key, set_up_config
app = flask.Flask(__name__)
T_value = t.TypeVar("T_value")
class SerializableSensor(ABC, t.Generic[T_value]):
title: str
@abstractmethod
def value(self) -> T_value:
pass
@classmethod
@abstractmethod
def serialize(cls, value: T_value) -> str:
pass
@classmethod
@abstractmethod
def deserialize(cls, serialized: str) -> T_value:
pass
class JSONSerializedSensor(SerializableSensor[t.Any]):
@classmethod
def serialize(cls, value: t.Any) -> str:
try:
return json.dumps(value)
except TypeError:
return json.dumps(None)
@classmethod
def deserialize(cls, serialized: str) -> t.Any:
return json.loads(serialized)
class JSONWrappedSensor(JSONSerializedSensor):
def __init__(self, sensor: Sensor[t.Any]):
self.wrapped = sensor
self.title = sensor.title
def value(self) -> t.Any:
return self.wrapped.value()
def get_serializable_sensors() -> t.Iterable[SerializableSensor[t.Any]]:
sensors = get_sensors()
found = []
for sensor in sensors:
if isinstance(sensor, SerializableSensor):
found.append(sensor)
else:
found.append(JSONWrappedSensor(sensor))
return found
@app.route("/sensors/")
@require_api_key
def sensor_values() -> t.Tuple[t.Dict[str, t.Any], int, t.Dict[str, str]]:
headers = {"Content-Security-Policy": "default-src 'none'"}
data = {}
for sensor in get_serializable_sensors():
data[sensor.title] = sensor.serialize(sensor.value())
return data, 200, headers
if __name__ == "__main__":
import wsgiref.simple_server
set_up_config(None, app)
with wsgiref.simple_server.make_server("", 8000, app) as server:
server.serve_forever()
#### 程式碼邏輯解析：
1. **定義抽象基礎類別`SerializableSensor`**：該類別定義了序列化和反序列化的介面。
2. **實作`JSONSerializedSensor`**：該類別提供了根據JSON的序列化和反序列化實作。
3. **建立`JSONWrappedSensor`**：該類別用於包裝不支援序列化的Sensor，使其能夠被序列化。
4. **`get_serializable_sensors()`函式**：該函式負責取得所有可序列化的Sensor。
5. **`sensor_values()`檢視函式**：該函式處理對`/sensors/`的請求，傳回所有Sensor的值。
## 軟體介面設計與JSON序列化實作
身為軟體的維護者，我們在決定介面設計上有更大的彈性。雖然仍需考慮實作的簡易性，但我們有更強的權威來定義所需的函式功能。本文將探討如何透過限制使用JSON API來提高資料的可讀性，並實作相關的序列化和反序列化方法。
### JSON序列化方法的設計與實作
為了使原始資料更容易被理解，我們決定限制使用JSON API。這樣做可以避免使用`serialize(...)`和`deserialize(...)`函式可能導致的不可讀取輸出。相反地，我們將建立能夠將值轉換為JSON可序列化格式的方法，以及從該格式重建值的方法。
#### 在Sensor基礎類別中新增JSON序列化方法
我們可以在`Sensor`基礎類別中定義這些方法，並提供預設實作。目前尚未保證任何感測器都相容於JSON序列化，因此預設實作應當是引發例外。
```python
@classmethod
def to_json_compatible(cls, value: T_value) -> Any:
raise NotImplementedError
@classmethod
def from_json_compatible(cls, json_version: Any) -> T_value:
raise NotImplementedError
為現有感測器實作JSON序列化方法
接下來，我們需要為現有的感測器提供這對方法的實作。主要分為三種不同的程式碼路徑需要更新。
1. 已經與JSON相容的感測器
對於大多數已經與JSON相容的感測器，我們可以建立一個新的混合類別JSONSensor：
class JSONSensor(Sensor[T_value]):
@classmethod
def to_json_compatible(cls, value: T_value) -> t.Any:
return value
@classmethod
def from_json_compatible(cls, json_version: t.Any) -> T_value:
return cast(JSONT_value, json_version)
2. PythonVersion感測器
PythonVersion感測器使用了一個自定義類別，該類別無法直接例項化。因此，我們需要稍微修改感測器以便能夠從JSON轉換回原始值。
version_info_type = NamedTuple(
"version_info_type",
[
("major", int),
("minor", int),
("micro", int),
("releaselevel", str),
("serial", int),
],
)
class PythonVersion(JSONSensor[version_info_type]):
title = "Python Version"
def value(self) -> version_info_type:
return version_info_type(*sys.version_info)
@classmethod
def format(cls, value: version_info_type) -> str:
if value.micro == 0 and value.releaselevel == "alpha":
return "{0.major}.{0.minor}.{0.micro}a{0.serial}".format(value)
return "{0.major}.{0.minor}".format(value)
3. 溫度與太陽能功率感測器
這兩個感測器使用物理量作為其值型別，因此需要自定義的JSON方法。
class Temperature(Sensor[Optional[Any]]):
...
@classmethod
def to_json_compatible(cls, value: Optional[Any]) -> Any:
#### 內容解密：
# 將溫度值轉換為JSON可序列化的格式。
# 如果值不為None，則傳回一個包含magnitude和unit的字典。
if value is not None:
return {"magnitude": value.magnitude, "unit": str(value.units)}
else:
return None
@classmethod
def from_json_compatible(cls, json_version: Any) -> Optional[Any]:
#### 內容解密：
# 從JSON可序列化的格式重建溫度值。
# 如果json_version不為空，則使用magnitude和unit重建ureg.Quantity物件。
if json_version:
return ureg.Quantity(json_version["magnitude"], ureg[json_version["unit"]])
else:
return None
程式碼整理
目前sensors.py檔案包含了兩個基礎類別和一些實際的感測器。為了使程式碼函式庫更容易導覽，我們將支援程式碼移至base.py。
定義Sensor基礎類別
在base.py中，我們定義了Sensor基礎類別，如清單5-13所示。
import typing as t
T_value = t.TypeVar("T_value")
class Sensor(t.Generic[T_value]):
name: str
title: str
def value(self) -> T_value:
raise NotImplementedError
JSON API的改進
為了使JSON API與感測器入口點使用相同的鍵，我們在Sensor基礎類別中新增了一個name屬性。這樣可以更容易地反序列化資料，因為我們可以輕易地查詢定義它的感測器類別。
版本控制與API管理
在軟體開發過程中，版本控制是確保API穩定性和向後相容性的關鍵。隨著功能的增加和技術的演進，API需要進行更新以滿足新的需求，但同時也要考慮到已有的使用者和舊版本的相容性。
版本控制的重要性
當API需要進行重大變更時，直接修改現有的API可能會導致依賴該API的使用者端出現問題。因此，建立多個版本的API成為了一個必要的選擇。
版本控制策略
URL版本控制：將版本號直接包含在URL中，如
/v/1.0/sensors和/v/2.0/sensors。這種方法簡單明瞭，使用者可以根據需要選擇合適的版本。版本號管理：可以使用簡單的數字版本（如1.0、2.0），也可以採用日曆版本控制（Calendar Versioning）。選擇合適的版本控制策略取決於專案的需求和團隊的偏好。
實作多版本API
為了支援多個版本的API，需要對程式碼進行重構，將不同版本的API邏輯分離到不同的檔案中，並使用Flask的Blueprint功能來註冊和管理不同版本的API。
程式碼結構範例
# v10.py
from flask import Blueprint
version = Blueprint(__name__, __name__)
@version.route("/sensors/")
@require_api_key
def sensor_values() -> t.Tuple[t.Dict[str, t.Any], int, t.Dict[str, str]]:
# API實作細節
...
# __init__.py
from flask import Flask
from .v10 import version as v10_blueprint
app = Flask(__name__)
app.register_blueprint(v10_blueprint, url_prefix="/v/1.0")
檔案結構
src/apd/sensors/wsgi/
├── __init__.py
├── base.py
├── serve.py
├── v10.py
└── v20.py
測試多版本API
支援多版本API的同時，也需要確保每個版本的API都經過充分的測試。可以透過為每個版本建立獨立的測試類別來實作這一點。
測試類別範例
# test_v10_api.py
from apd.sensors.wsgi import v10
class TestV10API:
@pytest.fixture
def subject(self, api_key):
app = flask.Flask("testapp")
app.register_blueprint(v10.version)
set_up_config({"APD_SENSORS_API_KEY": api_key}, to_configure=app)
return app
@pytest.fixture
def api_server(self, subject):
return TestApp(subject)
class CommonTests:
@pytest.mark.functional
def test_sensor_values_fails_on_missing_api_key(self, api_server):
response = api_server.get("/sensors/", expect_errors=True)
assert response.status_code == 403
assert response.json["error"] == "Supply API key in X-API-Key header"
重構Sensor類別以支援JSON序列化
為了使Sensor類別能夠支援JSON序列化，需要實作to_json_compatible和from_json_compatible方法。
Sensor類別範例
class Sensor:
@classmethod
def to_json_compatible(cls, value: T_value) -> t.Any:
raise NotImplementedError
@classmethod
def from_json_compatible(cls, json_version: t.Any) -> T_value:
raise NotImplementedError
class JSONSensor(Sensor[T_value]):
@classmethod
def to_json_compatible(cls, value: T_value) -> t.Any:
return value
@classmethod
def from_json_compatible(cls, json_version: t.Any) -> T_value:
return t.cast(T_value, json_version)
內容解密：
to_json_compatible方法：將感測器的值轉換為可JSON序列化的格式。這使得感測器的資料可以輕鬆地被轉換為JSON格式，以便於網路傳輸或儲存。from_json_compatible方法：將JSON相容的資料轉換回感測器的值。這是to_json_compatible的逆操作，用於從JSON資料中還原感測器的值。JSONSensor類別：繼承自Sensor，並為其提供了一個預設的JSON序列化實作。這意味著任何繼承自JSONSensor的感測器類別都自動獲得了JSON序列化的能力。
