Django CMSの公開記事が無権限で書き換えられた――認可を「画面」ではなく「状態遷移」で直す

Django CMSの公開記事が無権限で書き換えられた――認可を「画面」ではなく「状態遷移」で直す
目次

Django製CMSの認可バイパスを塞ぐ――自動保存・リビジョン復元・公開競合まで状態遷移で守る


CMSには、記事を書く人、内容を確認する人、公開する人がいます。

画面上では役割がきれいに分かれていても、裏側に一つでも権限確認の弱い更新経路が残っていれば、承認フロー全体を迂回できます。

今回見つかったのは、まさにその問題でした。

編集権限を剥奪された記事所有者でも、自動保存APIまたはリビジョン復元を利用すると、公開済み記事の本文を書き換えられる状態になっていました。

しかも、記事の状態は「公開中」のままです。

つまり、レビュー担当者が一度も確認していない文章が、そのまま公開ページへ表示される可能性がありました。

通常の編集画面を閉じるだけでは修正できません。

必要だったのは、CMSに存在する状態遷移を洗い出し、

誰が、どの記事を、どの状態から、どの状態へ変更してよいのか

という認可ルールを定義し、そのルールをすべての更新経路へ適用することでした。

さらに、その条件を回帰テストとして固定しました。

この記事の要点

  • CMSの認可は、画面やURLではなく「データを変更する操作」単位で設計する必要がある。
  • 記事所有者であることと、現在も記事を変更する権限を持つことは別の概念である。
  • 自動保存やリビジョン復元も、通常編集と同じ認可条件を通さなければならない。
  • 公開済み記事へ作業中の内容を直接保存すると、未承認文章が公開される危険がある。
  • レビュー後に本文が変更された場合は、versionを確認して公開を中止する必要がある。
  • HTTP 403や409を返すだけでは不十分で、拒否された後にDBが変更されていないことまでテストする。
  • 認可だけでなく、プロキシ、MFA、Redis、バックアップ、コンテナ境界も本番環境のセキュリティ境界として確認する。

結論

今回の問題の本質は、通常の編集画面ではなく、Articleを書き換えられる複数の経路で認可ルールが統一されていなかったことです。

修正後は、記事変更の条件を共通化しました。

変更可能
  = change_article を持つ
  AND
    (
      記事所有者である
      OR
      review_article を持つ
    )

この条件を、通常編集だけでなく、自動保存、プレビュー、リビジョン復元、Django AdminなどArticleへ触れる経路へ適用します。

さらに、公開操作ではレビュー時点のversionを記録します。

公開時に現在のversionと比較し、レビュー後に本文が変更されていれば公開しません。

これにより、

執筆
 ↓
レビュー
 ↓
内容変更なし
 ↓
公開

という承認済みの経路だけを成立させます。


何が起きていたのか

従来の編集判定では、次のいずれかに該当すれば記事を編集できる設計になっていました。

  • 記事の所有者である
  • staffユーザーである
  • レビュー権限を持っている

一見すると自然です。

しかし、この判定には異なる概念が混在しています。

「記事所有者」と「現在の編集権限」は別

「記事を所有している」という関係は、その記事とユーザーとの所属関係を示します。

しかし、

owner == request.user

だからといって、

現在もArticleを変更してよい

とは限りません。

管理者がユーザーから編集権限を剥奪したにもかかわらず、所有者判定だけで編集できるのであれば、権限剥奪が実質的に機能しません。

is_staff も万能な管理者権限ではない

Djangoのis_staffは、主にDjango Adminへのアクセス可否を判断する属性として利用されます。

実際のモデル操作については、ユーザーが持つpermissionやAdmin側の認可処理も関係します。

そのため、

if request.user.is_staff:
    allow()

のようにstaff属性だけでCMSの記事編集・承認・公開を許可すると、本来分離すべき権限をまとめて与えてしまう可能性があります。

自動保存とリビジョン復元が別経路になっていた

さらに問題だったのが、自動保存とリビジョン復元です。

通常の編集画面とは別の処理からArticleを直接更新していました。

その経路では、

change_article
publish_article
現在の記事状態

などの条件が十分に確認されていませんでした。

結果として、

通常編集
  ↓
認可チェック
  ↓
拒否

となっていても、

自動保存API
  ↓
弱い認可
  ↓
Article更新

という別経路から公開本文を書き換えられる状態になっていました。


この問題は自分のCMSにも関係するか

同様の問題は、独自CMSだけに限定されません。

次のような機能を持つCMSでは確認する価値があります。

  • 自動保存
  • 下書き保存
  • リビジョン復元
  • プレビュー
  • 一括更新
  • REST API
  • Django Admin
  • 公開予約
  • 承認
  • 差し戻し
  • 複製
  • インポート

重要なのは、画面の数ではありません。

ArticleやPostなどの公開データを変更できる入口が何個あるかを確認します。

たとえば、編集画面を閉じてもREST APIが残っていれば、認可不備は解消していません。


最初に「拒否された後も何も変わらない」テストを書く

セキュリティ修正では、HTTP 403 Forbiddenや409 Conflictが返るだけでは不十分です。

たとえば、次の処理では問題があります。

本文を保存
 ↓
権限チェック
 ↓
403

レスポンス上は拒否されています。

しかし、DBでは本文が変更されています。

攻撃者から見れば成功です。

そのため今回は、攻撃手順そのものを回帰テストとして再現しました。

  • 権限を剥奪した所有者が自動保存しても、本文と公開状態が変わらない。
  • 権限を剥奪した所有者がリビジョンを復元しても、本文と公開状態が変わらない。
  • 無関係なstaffユーザーは、他人の記事を編集・プレビューできない。
  • 公開中の記事へ自動保存しても、公開本文が上書きされない。
  • 正当な執筆者は、自分の下書きを従来どおり編集し、レビュー依頼できる。

まず、脆弱な状態でテストが失敗することを確認します。

その後に実装を修正します。

この順番にすることで、

本当に脆弱性を再現できているのか

と、

修正によって本当に攻撃経路が閉じたのか

をテストで確認できます。

拒否経路だけではなく、正規の執筆経路を残すことも重要です。

セキュリティを強化した結果、正当な利用者まで記事を書けなくなれば、CMSとして成立しません。


原因は「認可条件が入口ごとに違っていたこと」

今回の問題を単純化すると、原因は次の構造でした。

                    ┌─ 通常編集 ───── 強い認可
                    │
ユーザー ─ Article ├─ 自動保存 ───── 弱い認可
                    │
                    ├─ リビジョン ─── 弱い認可
                    │
                    └─ Admin ─────── 別の認可

Articleを変更するという結果は同じなのに、入口ごとに認可条件が異なっていました。

この状態では、一番弱い入口がシステム全体の認可強度になります。

そこで、認可条件そのものを共通化しました。


認可を共通の「不変条件」にする

修正後、記事を変更するには二つの条件を同時に満たす必要があります。

  1. 現在もchange_article権限を持っている。
  2. 自分の記事である、またはreview_article権限によって対象記事を扱える。

簡略化すると次のようになります。

変更可能
  = change_articleを持つ
  AND
    (
      記事所有者である
      OR
      review_articleを持つ
    )

重要なのは、この判定を通常フォームだけに書かないことです。

次の入口から同じ認可処理を呼び出します。

通常編集
自動保存
プレビュー
リビジョン
Django Admin
その他Article更新API
        │
        ▼
共通認可サービス
        │
        ▼
Article変更

公開操作は、さらに強い状態遷移として扱います。

今回のCMSでは、レビュー可能であることだけでは公開できません。

review_article
      AND
publish_article

の両方を要求します。

staff属性だけで、この条件を迂回することもできない設計にしました。

ポイントは、URLや画面単位で権限を考えないことです。

このURLへ入れるか

ではなく、

Articleを変更できるか
レビュー済みArticleを公開できるか

という操作単位で不変条件を定義します。


公開版を直接編集しない

現在のデータモデルには、公開中のスナップショットとは別に編集できる作業コピーがありません。

この構造で公開記事を自動保存すると、保存先は公開ページが参照しているArticleそのものになります。

つまり、

公開Article
   ↑
自動保存

という危険な構造になります。

そこで今回は、公開中の記事に対する自動保存をHTTP 409 Conflictで拒否しました。

リビジョン復元についても、本文へ触れる前に拒否します。

なぜ409なのか

今回のケースでは、ユーザーに編集権限がないというより、

現在のArticle状態では、その保存操作を成立させられない

という状態競合として扱っています。

そのため、公開中Articleへの自動保存では409 Conflictを利用しています。

将来は「公開版」と「作業コピー」を分離したい

公開したまま次の版を長期間編集する必要がある場合は、

公開スナップショット
        │
        ├──── 公開ページ
        │
作業コピー
        │
        └──── 編集・自動保存

という構造が必要です。

公開行そのものを一時的な下書きとして利用すると、未承認内容が漏れるのは本文だけとは限りません。

たとえば次の経路があります。

検索
RSS
キャッシュ
OGP
サイトマップ
API
全文検索インデックス

今回は、

安全に保存できる場所がないなら保存しない

という安全側の挙動を選びました。


レビュー後に差し替えられた本文も公開させない

認可を修正しても、もう一つ問題があります。

レビュー担当者が内容を確認してから公開ボタンを押すまでの間に、執筆者が本文を変更した場合です。

たとえば次の順番です。

version 10 をレビュー
        ↓
執筆者が自動保存
        ↓
version 11
        ↓
レビュー担当者が公開

公開処理が現在のArticleをそのまま公開すると、レビュー担当者が確認していないversion 11が公開されます。

権限上は正規の利用者しか操作していなくても、承認フローとしては破綻しています。

レビュー時点のversionを公開条件にする

そこで、レビュー画面を表示した時点のArticle.versionを承認フォームへ結び付けました。

レビュー時
Article.version = 10
        ↓
承認フォーム
reviewed_version = 10

公開時には記事行をロックします。

そのうえで、

現在version == reviewed_version

の場合だけ公開します。

レビュー後に本文が変更されていれば、公開を中止します。

確認済みversion = 10
現在version     = 11

        ↓

公開しない
再レビューを要求

自動保存とリビジョン復元についても、状態とversionの確認から保存までを同じ記事行のロック内で処理します。

これにより、

最後に保存した処理が勝つ

というLast Write Winsではなく、

レビューした版だけを公開できる

という挙動になります。


ログイン周辺のDoSと情報漏えいも見直す

今回の認可問題を調査する過程で、認証まわりにも改善点が見つかりました。

クライアントIPを誤認するとレート制限がDoSになる

リバースプロキシ配下では、アプリケーションが本当のクライアントIPを正しく取得できないと、複数の利用者が同じプロキシIPから接続しているように見える場合があります。

その状態でIP単位のレート制限を行うと、攻撃者がログインやパスワードリセットのレート枠を使い切ることで、無関係な利用者まで制限される可能性があります。

そこで、アプリ側と認証処理が同じプロキシ構成を前提としてクライアントIPを判定するようにしました。

構成値が実際の配備経路と一致しない場合は、起動時に検出できるようにします。

回帰試験では、

異なるクライアントIP
      ↓
異なる利用者として認識

されることと、

利用者が追加した偽のX-Forwarded-For
      ↓
信頼しない

ことを確認します。

ただし、信頼すべきプロキシ構成は配備環境によって変わります。

CDNやロードバランサーを追加した場合は、必ず再評価が必要です。

オリジンへの直接接続を許可したままX-Forwarded-Forを信用すると、IP判定そのものを迂回される可能性もあります。

管理ダッシュボードの表示範囲も権限で制限する

ログインしているだけで閲覧できた管理ダッシュボードについても見直しました。

一般ユーザーには、次のような内部情報を返しません。

予約記事タイトル
他ユーザーの執筆者情報
公開予定日
内部集計件数

CMS権限を入口で要求します。

さらに、レビュー担当者以外の一覧や集計は本人の記事へ限定します。

コメント保留件数についても、コメントを管理できる利用者だけに表示します。

MFA対象をstaffだけで決めない

MFAの対象もis_staffだけでは決めません。

公開、承認、削除など、高影響の操作権限を持つユーザーには、staffでなくてもMFAを要求する設計としました。

重要なのは属性名ではなく、

そのアカウントが侵害された場合に
どこまで変更できるか

です。


認可修正を契機に、本番環境の境界も再監査した

アプリケーションの認可が正しくても、それだけで安全な本番環境になるわけではありません。

平文バックアップ、認証されていないRedis、毎回内容が変わるコンテナイメージなどが残っていれば、別の経路からセキュリティ境界が崩れます。

そこで今回、認可修正とは別の防御層として、本番候補構成も再確認しました。

通信とコンテナ境界

  • NginxでTLS 1.2およびTLS 1.3を終端する。
  • HTTPアクセスはHTTPSへリダイレクトする。
  • 公開HTMLへリクエストごとのnonceを利用したCSPを付与する。
  • Redisへ認証を要求する。
  • Nginxはfrontend、DBとRedisはinternalなbackendへ分離する。
  • NginxコンテナからRedisへ直接接続できないネットワーク構成にする。

Redisについては、利用バージョンと要件に応じてACLによる最小権限化も検討対象になります。

起動とreadiness

  • migrationとcollectstaticは一度だけ実行するrelease jobへ分離する。
  • 通常のweb再起動では未適用migrationの有無を確認する。
  • migrationが失敗した場合はwebを起動しない。
  • readinessでDBとRedisへの実接続を確認する。
  • 依存サービス障害時は内部情報を出さずHTTP 503を返す。
  • Nginx自身のhealthcheckも、本番で利用する経路に近いHTTPS経路で確認する。

バックアップ

  • バックアップディレクトリを0700にする。
  • 暗号化済みバックアップファイルを0600にする。
  • DBとmediaをAES-256-CBCで暗号化する。
  • PBKDF2から必要な鍵素材を導出する。
  • 暗号文へHMAC-SHA-256を付与し、復号前に完全性を検証する。
  • 暗号化用とMAC用の鍵は適切に分離する。
  • HMAC鍵そのものをプロセス引数へ載せず、権限制限した鍵ファイルなどから読み込む。
  • 復旧訓練では本番DBや本番mediaを直接上書きせず、使い捨て環境へ復元する。

AES-CBCとHMACを組み合わせる場合は、暗号化と完全性検証を正しい順序で実装する必要があります。

新規設計であれば、AES-GCMなどのAEADを利用して暗号化と認証を一体化する方法も候補になります。

サプライチェーン

  • Pythonの推移的依存までlockファイルへ固定する。
  • 可能な範囲でパッケージハッシュを検証する。
  • Python、PostgreSQL、Redis、Nginxなどのコンテナイメージをdigestで固定する。
  • 秘密鍵、ローカル証明書、バックアップをDocker build contextから除外する。

依存関係スキャナーが、

既知の脆弱性 0件

と報告しても、今回のようなアプリケーション固有の認可不備まで安全であることは意味しません。

依存関係スキャン
        +
認可・状態遷移テスト
        +
実環境の構成確認

は、それぞれ別の防御です。


どこまで検証したか

今回の修正時点では、Djangoの全363テストが成功しました。

さらに、単体テストだけでなく、ローカルの実Compose構成でも次の項目を確認しています。

  • migration用jobが正常終了した後にwebとNginxが起動する。
  • HTTPSでトップページとhealthcheckへアクセスできる。
  • CSP、HSTS、X-Content-Type-Options: nosniffなど必要なセキュリティヘッダーが付与される。
  • HTTPアクセスがHTTPSへリダイレクトされる。
  • TLS 1.1は拒否され、TLS 1.2およびTLS 1.3では接続できる。
  • Redisは未認証アクセスを拒否し、認証後だけ操作できる。
  • Redis停止中はreadinessが503になり、復旧後は200へ戻る。
  • NginxからRedisへ直接接続できない。
  • DBダンプとmediaを使い捨て領域へ復元できる。
  • 暗号文を改ざんするとHMAC検証で拒否される。

さらに別視点で再レビューを行い、今回設定したレビュー範囲では、追加で修正すべき具体的な問題が残っていないことを確認しました。

ただし、これは将来の脆弱性が存在しないことや、本番環境全体の安全性を保証するものではありません。


それでも実環境での最終確認は残る

ここまでで、リポジトリ内の本番候補構成とローカル再現環境は整いました。

しかし、

ローカルでテストが通る

ことと、

本番環境が安全である

ことは同じではありません。

公開前には、実際の環境で少なくとも次の項目を確認します。

  • 実証明書の発行、更新、期限監視
  • DNS、ファイアウォール、CDN、オリジン間の到達制御
  • 実際のプロキシ経路とクライアントIP取得
  • メール送信
  • パスワードリセット
  • MFA、WebAuthnなどの認証フロー
  • バックアップ鍵の保管
  • オフホストバックアップ
  • 定期的な復元訓練
  • コンテナOSパッケージ
  • 実際に稼働しているコンテナイメージ
  • デプロイ後の公開記事
  • 管理画面
  • readiness
  • アプリケーションログ
  • セキュリティログ

再発防止で重要なのは「URL一覧」ではなく「状態遷移一覧」

今回の問題から得られた重要な教訓は、認可テストをURL一覧だけで管理しないことです。

たとえばArticleには、次の状態遷移があります。

下書き
  │
  ├─ 編集
  ├─ 自動保存
  ├─ リビジョン復元
  │
  ▼
レビュー待ち
  │
  ├─ 差し戻し
  ├─ 再編集
  │
  ▼
レビュー済み
  │
  ├─ 本文変更
  └─ 公開
      │
      ▼
    公開中
      │
      ├─ 更新
      ├─ 公開停止
      └─ リビジョン復元

この一つひとつについて、

誰が実行できるか
どのpermissionが必要か
どの状態から実行できるか
成功後にどの状態になるか
拒否時にデータが変化しないか

を定義します。

これをテストへ落とし込むことで、新しいAPIや管理機能を追加したときにも、既存の認可ルールを壊していないか確認できます。


まとめ

セキュリティ修正の完了条件は、パッチを書いたことではありません。

HTTP 403や409を返せるようになったことだけでもありません。

重要なのは、

元の攻撃手順が失敗する
        ↓
拒否後もDBが変化しない
        ↓
正規の執筆フローが動く
        ↓
レビューした版だけが公開される
        ↓
実際の配備環境でも同じ境界が守られる

ところまで確認することです。

今回の修正で最も大きかった学びは、認可を

「この画面を開けるか」

で考えるのではなく、

「誰が、どのデータに対して、どの状態遷移を実行してよいか」

で設計することでした。

CMSでは、通常編集、自動保存、リビジョン、API、Adminなど、同じデータへ到達する経路が増えやすくなります。

そのため、一つの画面を安全にするだけでは足りません。

最も弱い更新経路が、CMS全体の認可強度になります。

画面ではなく操作を見る。

ユーザー属性ではなくpermissionを見る。

レスポンスコードだけではなく、拒否後のデータを見る。

そして、レビューした内容と実際に公開される内容が同じであることを確認する。

今回の修正は、単なる権限チェックの追加ではなく、CMSの状態遷移そのものをセキュリティ境界として捉え直す作業になりました。


実際のDjangoコードではどう実装するか

ここからは、今回の認可ルールをDjangoへ実装する例を示します。

サンプルではアプリ名をcms、記事モデルをArticleとします。

実際のプロジェクトでは、モデル名、ステータス名、URL構成などを置き換えてください。

Djangoではモデルごとにaddchangedeleteviewのデフォルトpermissionが作成されます。

一方、review_articlepublish_articleのようなCMS固有の権限は、カスタムpermissionとして定義できます。

Articleモデルにversionとカスタム権限を持たせる

# cms/models.py

from django.conf import settings
from django.db import models


class Article(models.Model):
    class Status(models.TextChoices):
        DRAFT = "draft", "下書き"
        IN_REVIEW = "in_review", "レビュー待ち"
        PUBLISHED = "published", "公開中"

    owner = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.PROTECT,
        related_name="articles",
    )

    title = models.CharField(max_length=255)
    body = models.TextField()

    status = models.CharField(
        max_length=20,
        choices=Status.choices,
        default=Status.DRAFT,
    )

    # 本文など公開内容へ影響する変更のたびに増加させる
    version = models.PositiveBigIntegerField(default=1)

    published_at = models.DateTimeField(
        null=True,
        blank=True,
    )

    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        permissions = [
            (
                "review_article",
                "Can review articles",
            ),
            (
                "publish_article",
                "Can publish articles",
            ),
        ]

ここで重要なのは、versionを単なる表示用番号にしないことです。

version = レビューした内容と
          現在の内容が同じか判定するための値

として利用します。

本文、タイトル、OGP、SEO情報など、公開内容へ影響するフィールドを変更した場合はversionを増やします。

逆に、アクセス数など公開内容に影響しない情報までversionへ含めると、不必要な再レビューが増えるため注意が必要です。


記事の認可判定を1か所へ集約する

次に、

通常編集
自動保存
プレビュー
リビジョン復元
Admin

から共通して利用する認可関数を作ります。

# cms/services/permissions.py

from django.contrib.auth.models import AbstractBaseUser

from cms.models import Article


CHANGE_ARTICLE = "cms.change_article"
REVIEW_ARTICLE = "cms.review_article"
PUBLISH_ARTICLE = "cms.publish_article"


def can_change_article(
    user: AbstractBaseUser,
    article: Article,
) -> bool:
    """
    Articleを変更できるか判定する。

    条件:
    1. change_article を持つ
    2. 所有者、または review_article を持つ
    """

    if not user.is_authenticated:
        return False

    if not user.has_perm(CHANGE_ARTICLE):
        return False

    if article.owner_id == user.pk:
        return True

    return user.has_perm(REVIEW_ARTICLE)


def can_review_article(
    user: AbstractBaseUser,
    article: Article,
) -> bool:
    """
    Articleをレビューできるか判定する。
    """

    if not user.is_authenticated:
        return False

    return (
        user.has_perm(CHANGE_ARTICLE)
        and user.has_perm(REVIEW_ARTICLE)
    )


def can_publish_article(
    user: AbstractBaseUser,
    article: Article,
) -> bool:
    """
    Articleを公開できるか判定する。
    """

    if not user.is_authenticated:
        return False

    return (
        user.has_perm(CHANGE_ARTICLE)
        and user.has_perm(REVIEW_ARTICLE)
        and user.has_perm(PUBLISH_ARTICLE)
    )

ポイントは、

if user.is_staff:
    return True

を書いていないことです。

Django Adminへアクセスできる属性と、CMSの記事を変更・レビュー・公開できる権限を分離します。

is_superuserには別途注意する

一方、Django標準のpermissionシステムでは、activeなsuperuserは実質的にすべてのpermissionを持つユーザーとして扱われます。

そのため、

user.has_perm("cms.publish_article")

はsuperuserに対して通常Trueになります。

つまり、

staff
  ↓
自動的には許可しない

superuser
  ↓
Django標準ではpermissionを持つ扱い

という違いがあります。

superuserにもCMSの職務分離を強制したい場合は、Django標準permissionとは別のポリシー層を設計する必要があります。


権限マトリクスで認可仕様を固定する

文章だけで認可仕様を管理すると、後から条件が曖昧になります。

そこで、CMSの権限を表にします。

ユーザーchange_articlereview_articlepublish_article自分の下書き編集他人の記事編集レビュー公開
一般ユーザー×××××××
執筆者×××××
権限剥奪済み所有者×××××××
staffのみ×××××××
レビュー担当××
公開担当
superuserDjango標準では○Django標準では○Django標準では○

この表が、そのままテスト仕様になります。

特に重要なのは、

「所有者」や「staff」という属性だけでは○にならない

ことです。


自動保存をtransaction.atomic()select_for_update()で守る

次に、自動保存を実装します。

ここでは二種類の競合を同時に防ぎます。

認可競合
    +
同時更新競合

です。

# cms/services/article_service.py

from django.core.exceptions import PermissionDenied
from django.db import transaction

from cms.models import Article
from cms.services.permissions import can_change_article


class ArticleConflict(Exception):
    """Articleの現在状態では処理できない。"""


class ArticleVersionConflict(ArticleConflict):
    """クライアントが見ているversionが古い。"""


class PublishedArticleConflict(ArticleConflict):
    """公開Articleを直接変更しようとした。"""


@transaction.atomic
def autosave_article(
    *,
    user,
    article_id: int,
    body: str,
    expected_version: int,
) -> Article:

    article = (
        Article.objects
        .select_for_update(of=("self",))
        .get(pk=article_id)
    )

    # ロック取得後のArticleに対して認可する
    if not can_change_article(user, article):
        raise PermissionDenied

    # 公開版を作業コピーとして使わせない
    if article.status == Article.Status.PUBLISHED:
        raise PublishedArticleConflict

    # ブラウザが古いArticleを編集していないか確認
    if article.version != expected_version:
        raise ArticleVersionConflict

    article.body = body
    article.version += 1

    article.save(
        update_fields=[
            "body",
            "version",
            "updated_at",
        ]
    )

    return article

transaction.atomic()は、この処理全体を一つのトランザクションとして扱います。

select_for_update()は、対応するデータベースでは取得した行をトランザクション終了までロックします。

そのため、同じArticleに対する別の更新処理が途中へ割り込みにくくなります。

処理のイメージは次のとおりです。

BEGIN

SELECT Article
FOR UPDATE

    ↓

認可確認

    ↓

状態確認

    ↓

version確認

    ↓

本文変更

    ↓

version + 1

COMMIT

重要なのは、

article = Article.objects.get(pk=article_id)

# いろいろ処理する

with transaction.atomic():
    ...

のように、Articleをロックする前に取得しないことです。

認可・状態・versionを判断するときには、ロック後に取得した最新の行を使用します。


なぜversion確認と行ロックを両方使うのか

一見すると、

select_for_update()

があるならversionは不要に思えるかもしれません。

しかし、役割が違います。

select_for_update()

同時に実行されているDBトランザクション同士の競合を制御します。

version

利用者が古い画面を開いたまま編集していた場合の競合を検出します。

たとえば次の状況です。

12:00
ユーザーA
version 10を表示

12:01
ユーザーB
version 10を編集
        ↓
version 11

12:10
ユーザーA
古いversion 10を保存

12:10時点では、ユーザーBのトランザクションはすでに終了しています。

行ロックだけでは、

version 11
    ↓
version 10時点の本文で上書き

を防げません。

そこでリクエストに、

{
  "body": "変更後本文",
  "version": 10
}

のように、利用者が編集を開始したversionを含めます。

DB側がversion 11なら、

expected_version = 10
current_version  = 11

        ↓

409 Conflict

として保存を拒否します。

つまり、

select_for_update
    =
同時実行中の競合対策

version
    =
古い画面からの上書き対策

です。


自動保存APIでは403と409を分ける

Viewでは、認可エラーと状態競合を区別します。

# cms/views/autosave.py

from django.contrib.auth.decorators import login_required
from django.core.exceptions import PermissionDenied
from django.http import JsonResponse
from django.views.decorators.http import require_POST

from cms.services.article_service import (
    ArticleConflict,
    autosave_article,
)


@login_required
@require_POST
def autosave(request, article_id):

    try:
        body = request.POST["body"]
        expected_version = int(
            request.POST["version"]
        )

        article = autosave_article(
            user=request.user,
            article_id=article_id,
            body=body,
            expected_version=expected_version,
        )

    except PermissionDenied:
        return JsonResponse(
            {
                "error": "permission_denied",
            },
            status=403,
        )

    except ArticleConflict:
        return JsonResponse(
            {
                "error": "article_conflict",
                "message": (
                    "記事が変更されたか、"
                    "現在の状態では保存できません。"
                ),
            },
            status=409,
        )

    return JsonResponse(
        {
            "status": "saved",
            "version": article.version,
        }
    )

ここでは、

403
=
あなたには操作権限がない

409
=
権限はあっても現在のArticle状態では操作できない

と分離しています。

ただし、403を返すことで「そのIDの記事が存在する」という情報が漏れるシステムでは、404へ統一する設計も検討が必要です。


リビジョン復元も必ずArticleを先にロックする

リビジョン復元でも同じ原則を使います。

# cms/services/revision_service.py

from django.core.exceptions import PermissionDenied
from django.db import transaction

from cms.models import Article, ArticleRevision
from cms.services.article_service import (
    ArticleVersionConflict,
    PublishedArticleConflict,
)
from cms.services.permissions import can_change_article


@transaction.atomic
def restore_revision(
    *,
    user,
    article_id: int,
    revision_id: int,
    expected_version: int,
) -> Article:

    article = (
        Article.objects
        .select_for_update(of=("self",))
        .get(pk=article_id)
    )

    if not can_change_article(user, article):
        raise PermissionDenied

    if article.status == Article.Status.PUBLISHED:
        raise PublishedArticleConflict

    if article.version != expected_version:
        raise ArticleVersionConflict

    revision = ArticleRevision.objects.get(
        pk=revision_id,
        article_id=article.id,
    )

    article.title = revision.title
    article.body = revision.body
    article.version += 1

    article.save(
        update_fields=[
            "title",
            "body",
            "version",
            "updated_at",
        ]
    )

    return article

ここでも、

リビジョン取得
    ↓
Article更新
    ↓
最後に権限チェック

にはしません。

Articleをロックしたうえで、

認可
 ↓
状態
 ↓
version
 ↓
revision所属
 ↓
復元

の順番にします。


レビューしたversionをサーバー側へ記録する

次に、レビュー後の差し替え対策です。

ここでは、クライアントから送られてくるhidden inputだけを信頼しません。

たとえば、

<input
    type="hidden"
    name="reviewed_version"
    value="10"
>

だけに依存すると、その値はブラウザから変更できます。

そこで、レビュー結果をDBへ記録します。

# cms/models.py

class ArticleApproval(models.Model):
    article = models.ForeignKey(
        Article,
        on_delete=models.CASCADE,
        related_name="approvals",
    )

    reviewer = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.PROTECT,
    )

    article_version = models.PositiveBigIntegerField()

    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        indexes = [
            models.Index(
                fields=[
                    "article",
                    "article_version",
                ]
            ),
        ]

レビュー完了時には、Articleをロックした状態で現在versionを記録します。

# cms/services/review_service.py

from django.core.exceptions import PermissionDenied
from django.db import transaction

from cms.models import Article, ArticleApproval
from cms.services.permissions import can_review_article


@transaction.atomic
def approve_article(
    *,
    user,
    article_id: int,
) -> ArticleApproval:

    article = (
        Article.objects
        .select_for_update(of=("self",))
        .get(pk=article_id)
    )

    if not can_review_article(user, article):
        raise PermissionDenied

    if article.status != Article.Status.IN_REVIEW:
        raise ArticleConflict(
            "レビュー可能な状態ではありません。"
        )

    approval = ArticleApproval.objects.create(
        article=article,
        reviewer=user,
        article_version=article.version,
    )

    return approval

これで、

reviewer
   ↓
version 10を確認
   ↓
ArticleApproval
article_version = 10

という監査可能な記録が残ります。


公開時にselect_for_update()してレビューversionを比較する

公開処理では、Articleをロックしてから最新状態を確認します。

# cms/services/publish_service.py

from django.core.exceptions import PermissionDenied
from django.db import transaction
from django.utils import timezone

from cms.models import Article, ArticleApproval
from cms.services.article_service import ArticleConflict
from cms.services.permissions import can_publish_article


class ReviewRequired(ArticleConflict):
    pass


@transaction.atomic
def publish_article(
    *,
    user,
    article_id: int,
) -> Article:

    article = (
        Article.objects
        .select_for_update(of=("self",))
        .get(pk=article_id)
    )

    if not can_publish_article(user, article):
        raise PermissionDenied

    if article.status == Article.Status.PUBLISHED:
        raise ArticleConflict(
            "すでに公開されています。"
        )

    approval = (
        ArticleApproval.objects
        .filter(
            article=article,
            article_version=article.version,
        )
        .order_by("-created_at")
        .first()
    )

    if approval is None:
        raise ReviewRequired(
            "現在のversionはレビューされていません。"
        )

    article.status = Article.Status.PUBLISHED
    article.published_at = timezone.now()

    article.save(
        update_fields=[
            "status",
            "published_at",
            "updated_at",
        ]
    )

    return article

たとえばレビュー済みversionが10だったとします。

ArticleApproval
article_version = 10

その後、執筆者が本文を変更すると、

Article.version = 11

になります。

公開処理では、

approval.version = 10
article.version  = 11

        ↓

一致する承認なし

        ↓

公開拒否

となります。

レビュー担当者が見ていないversion 11は公開できません。


修正前の攻撃経路をシーケンス図で見る

修正前は、自動保存APIが所有者判定だけでArticleを更新していました。


通常編集画面側の認可が正しくても、自動保存という別入口からArticleを変更できています。

この場合、

通常編集 = 安全

でも、

CMS全体 = 安全

とは言えません。


修正後は共通認可と行ロックを通る

修正後は次の流れになります。

sequenceDiagram
    autonumber

    actor User as 権限剥奪済み所有者
    participant API as 自動保存API
    participant Service as ArticleService
    participant DB as PostgreSQL

    User->>API: POST /autosave/
    API->>Service: autosave_article()

    Service->>DB: BEGIN
    Service->>DB: SELECT ... FOR UPDATE
    DB-->>Service: Article

    Service->>Service: change_article確認

    Note over Service: 権限なし

    Service-->>API: PermissionDenied
    Service->>DB: ROLLBACK

    API-->>User: 403 Forbidden

    Note over DB: body変更なし<br/>version変更なし<br/>status変更なし

ここで重要なのは403そのものではありません。

拒否
 ↓
ROLLBACK
 ↓
Articleが変化していない

ことです。


レビュー後の差し替え攻撃もversionで止める

次の競合も考えます。

sequenceDiagram
    autonumber

    actor Writer as 執筆者
    actor Reviewer as レビュー担当
    actor Publisher as 公開担当
    participant DB as PostgreSQL

    Reviewer->>DB: version 10をレビュー
    DB-->>Reviewer: Article version=10

    Reviewer->>DB: ArticleApproval(version=10)
    DB-->>Reviewer: 承認保存

    Writer->>DB: SELECT FOR UPDATE
    DB-->>Writer: version=10

    Writer->>DB: 本文更新 version=11
    DB-->>Writer: COMMIT

    Publisher->>DB: SELECT FOR UPDATE
    DB-->>Publisher: Article version=11

    Publisher->>DB: version=11のApproval検索
    DB-->>Publisher: 存在しない

    DB-->>Publisher: 公開拒否

    Note over Publisher,DB: version 11を再レビューするまで<br/>公開できない

この仕組みによって、

レビューした文章

と、

実際に公開される文章

を結び付けます。


Django Adminも同じ認可関数を使う

通常画面だけを修正しても、Django Adminから変更できれば抜け道になります。

Adminにも同じ判定を適用します。

# cms/admin.py

from django.contrib import admin

from cms.models import Article
from cms.services.permissions import can_change_article


@admin.register(Article)
class ArticleAdmin(admin.ModelAdmin):

    def has_change_permission(
        self,
        request,
        obj=None,
    ):
        if obj is None:
            return request.user.has_perm(
                "cms.change_article"
            )

        return can_change_article(
            request.user,
            obj,
        )

Django Adminもモデルに対するpermissionを利用してアクセスを制御します。

また、ModelAdmin.has_change_permission()などを利用すると、オブジェクト単位の追加条件を実装できます。


「拒否された後にDBが変わっていない」ことをテストする

回帰テストではレスポンスコードだけを確認しません。

重要なのはDBです。

# cms/tests/test_article_authorization.py

from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.test import TestCase

from cms.models import Article


User = get_user_model()


class ArticleAuthorizationTests(TestCase):

    def setUp(self):
        self.user = User.objects.create_user(
            username="writer",
            password="test-password",
        )

        self.article = Article.objects.create(
            owner=self.user,
            title="安全な本文",
            body="レビュー済み本文",
            status=Article.Status.DRAFT,
            version=10,
        )

        # いったんchange権限を付与
        permission = Permission.objects.get(
            codename="change_article",
        )

        self.user.user_permissions.add(
            permission
        )

        # 権限キャッシュの影響を避けるため再取得
        self.user = User.objects.get(
            pk=self.user.pk
        )

    def test_revoked_owner_cannot_autosave(self):

        permission = Permission.objects.get(
            codename="change_article",
        )

        self.user.user_permissions.remove(
            permission
        )

        # has_perm()のキャッシュを残さない
        self.user = User.objects.get(
            pk=self.user.pk
        )

        self.client.force_login(self.user)

        response = self.client.post(
            f"/cms/articles/{self.article.pk}/autosave/",
            {
                "body": "攻撃者が変更した本文",
                "version": 10,
            },
        )

        self.assertEqual(
            response.status_code,
            403,
        )

        self.article.refresh_from_db()

        self.assertEqual(
            self.article.body,
            "レビュー済み本文",
        )

        self.assertEqual(
            self.article.version,
            10,
        )

        self.assertEqual(
            self.article.status,
            Article.Status.DRAFT,
        )

Djangoの標準ModelBackendでは、取得したpermission情報がユーザーオブジェクト上にキャッシュされる場合があります。

そのため、テスト中にpermissionを追加・削除した直後は、ユーザーをDBから再取得して確認した方が安全です。


古いversionからの自動保存も409にする

次は正常な執筆者ですが、ブラウザ側のversionが古いケースです。

def test_stale_version_cannot_overwrite_article(self):

    self.article.version = 11
    self.article.body = "新しい本文"

    self.article.save(
        update_fields=[
            "body",
            "version",
        ]
    )

    self.client.force_login(self.user)

    response = self.client.post(
        f"/cms/articles/{self.article.pk}/autosave/",
        {
            "body": "version 10から保存した古い本文",
            "version": 10,
        },
    )

    self.assertEqual(
        response.status_code,
        409,
    )

    self.article.refresh_from_db()

    self.assertEqual(
        self.article.body,
        "新しい本文",
    )

    self.assertEqual(
        self.article.version,
        11,
    )

ここでも重要なのは、

self.assertEqual(
    response.status_code,
    409,
)

だけではありません。

必ず、

refresh_from_db()

した後の本文とversionを確認します。


select_for_update()のテストではTransactionTestCaseを使う

ここは重要な注意点です。

Djangoの通常のTestCaseは、各テストをトランザクションでラップします。

そのため、本来トランザクション外なら問題になるselect_for_update()が、テストでは動いているように見える場合があります。

Django公式ドキュメントでも、select_for_update()のトランザクション動作を正しくテストする場合はTransactionTestCaseを使うよう注意されています。

from django.test import TransactionTestCase


class ArticleConcurrencyTests(
    TransactionTestCase
):
    reset_sequences = True

    def test_publish_checks_current_version(self):
        ...

さらに重要なのがDBです。

SQLiteではSELECT ... FOR UPDATEが利用されないため、

Article.objects.select_for_update()

と書いてあっても、PostgreSQLと同じ行ロック動作を検証できません。

本番がPostgreSQLなら、競合制御の統合テストもPostgreSQLで実行する必要があります。


テストの実行方法

今回の例で認可・競合テストだけを実行する場合は、たとえば次のようにします。

python manage.py test \
  cms.tests.test_article_authorization \
  cms.tests.test_article_concurrency

目的

認可バイパスとArticleの競合制御に関する回帰テストを実行します。

実行場所

manage.pyが存在するDjangoプロジェクトのルートディレクトリです。

正常例

Ran 18 tests in 2.341s

OK

異常例

FAIL: test_revoked_owner_cannot_autosave

または、

FAIL: test_publish_checks_current_version

などが表示されます。

結果の判断

すべてのテストがOKであることに加えて、

拒否時にArticleが変更されない
古いversionで上書きされない
未レビューversionを公開できない
正当な執筆者は保存できる

という各不変条件がassertされていることを確認します。


transaction.atomic()を付ければ安全、ではない

最後に、この実装にも注意点があります。

1. 行ロックを長時間保持すると性能問題になる

select_for_update()で取得した行は、トランザクション終了までロックされます。

そのため、atomicブロックの中で、

外部API通信
メール送信
重い画像処理
LLM呼び出し
大量集計

などを実行すると、Articleのロック時間が長くなります。

トランザクション内では、

取得
認可
状態確認
DB更新

までを短時間で行う方が安全です。

2. DBトランザクションはRedisや外部サービスを巻き戻せない

たとえば次の処理は注意が必要です。

with transaction.atomic():
    article.save()

    redis.delete(cache_key)

    send_message_to_queue()

後からDBがROLLBACKされても、Redis削除や外部キュー送信まで元には戻りません。

DBコミット後に実行したい処理は、

transaction.on_commit(...)

を利用する設計も検討します。

3. SQLiteだけでは本番の競合を再現できない

SQLite上で363テストが成功しても、本番PostgreSQLの行ロックが正しいことまでは証明できません。

select_for_update()を利用するなら、本番と同じDBエンジンを使った統合試験を用意した方が安全です。

4. superuserの扱いは別途ポリシーが必要

is_staffによる迂回を削除しても、Djangoのsuperuserは別です。

CMSで、

システム管理者

と、

コンテンツ公開責任者

を完全に分離したい場合、superuserの運用ルールまで含めて設計する必要があります。


最終的な防御構造

今回の実装をまとめると、Article更新は次の構造になります。

               ┌─ 通常編集
               │
               ├─ 自動保存
               │
               ├─ リビジョン
               │
               ├─ プレビュー
               │
               └─ Django Admin
                       │
                       ▼
                共通認可サービス
                       │
             change_article
                       AND
            owner OR review_article
                       │
                       ▼
             transaction.atomic()
                       │
                       ▼
             select_for_update()
                       │
                       ▼
                  状態確認
                       │
                       ▼
                  version確認
                       │
                       ▼
                    更新

公開ではさらに、

change_article
       +
review_article
       +
publish_article
       +
現在versionのレビュー記録
       +
行ロック

を要求します。

この構造なら、

「このURLに認可チェックを書いた」

ではなく、

「Articleを変更するすべての経路が、同じ不変条件を満たさなければ更新できない」

という設計になります。

CMSのセキュリティで本当に守りたいのは画面ではありません。

Articleの状態遷移そのものです。

読んだ内容を10問練習と実技で確認

記事で理解した用語を、StudyQuestの演習とクラウド実技ラボで定着させます。

10問練習 実技ラボ

コメント(0件)

まだコメントはありません。最初のコメントを投稿してください!

コメントを投稿