CSVファイルのアップロードと配布
受入・保存・配布のそれぞれに必要な条件と、拒否した後の保存状態を確認します。
説明
アップロードでは、拡張子やContent-Typeは送信者が指定する情報です。ファイルを受け取った後に内容を解析し、アプリが扱う形式と業務上の範囲を検証します。
商品と数量を持つCSVを使います。見かけだけをCSVにした入力を拒否し、正しい内容を再出力して別IDへ保存します。別利用者による取得と返却ヘッダーまで追います。
受入・保存・配布のそれぞれに必要な条件と、拒否した後の保存状態を確認します。
1CSVの受入条件を用意する
列はitem,quantity、商品はpenまたはbook、数量は1〜5、最大3行と128バイトです。
from sd08 import *
lab = Upload()
body = b"item,quantity\r\npen,02\r\n"
print(len(body),list(csv.readerCSVの列構造を解析する(io.StringIO(body.decode()))))標準CSVパーサーは02を文字列として返します。業務条件の検査と数量の整数化は、パースとは別に必要です。
2拡張子と申告形式を確認する
同じ内容に異なる名前と形式を付けます。
print(lab.save("alice","orders.html"受入外の拡張子を渡す,"text/csv",body))
print(lab.save("alice","orders.csv","text/html"受入外の申告形式を渡す,body))extension_rejectedとmedia_type_rejectedです。これらは入口の条件であり、両方が正しくても内容検証は省略しません。
3内容と業務範囲を検査する
見かけがCSVでもHTML本文や想定外の列・数量を受け付けません。
print(lab.save("alice","orders.csv","text/csv",b"<p>report</p>"形式を偽った本文を渡す))
print(lab.save("alice","orders.csv","text/csv",b"item,quantity\npen,6業務上限を超える数量を渡す\n"))
print(lab.db.execute("SELECT COUNT(*) FROM files").fetchone())構造違いはschema_rejected、数量6はrow_rejected。保存件数は0です。CSVにはPNGのような固定マジック値がないため、対象の構文と業務規則を調べています。
4内容を再出力して管理IDへ保存する
送信名にパスが含まれていても保存先には使わず、検査したCSVをアプリ側で再出力します。
id = lab.save検査済みの値をCSVへ再出力する("alice","../orders.csv","text/csv",body)
print(id)
print(lab.db.execute("SELECT id,owner,body管理IDと保存した内容を読む FROM files").fetchall())upload-1へ保存され、02は2、改行はLFへ揃います。元の名前によるパス操作は起きず、ここではDBのBLOBとして管理します。
5別主体による取得を拒否する
同じupload-1をbobとして要求します。
print(lab.get("bob",id)所有者以外で同じIDを取得する)
print(lab.get("alice","upload-missing"))どちらも404、空ヘッダー、空本文です。推測しにくいIDに頼らず、取得SQLにownerを含めています。
6所有者への配布ヘッダーを確認する
所有者aliceとして取得します。
status,headers,saved = lab.get("alice",id)認可された本文を取得する
print(status,headers返却形式と配布方法を調べる)
print(saved.decode(),end="")200で、CSVのContent-Type、attachment、nosniffを返す設定です。配布名は固定のexport.csvで、送信者の名前をヘッダーへ直接挿入していません。
7大きい入力を拒否して保存値を照合する
容量の条件を超えた129バイトを送り、元の内容が維持されるか調べます。
print(lab.save("alice","large.csv","text/csv",b"x"*129内容処理の容量上限を超える))
print(lab.db.execute("SELECT COUNT(*),length(body),digest FROM files").fetchone())
print(hashlib.sha256(saved)返却したバイト列を保存値と照合する.hexdigest())too_largeで1件のままです。保存した本文のハッシュと再取得した本文のハッシュも一致します。HTTP受信時の総量制限は、この内容検査より前段でも必要です。
まとめ
申告された形式に加え、内容の構文と業務条件を検査します。
送信名を保存先にせず、管理IDで保存します。
取得時の認可と返却ヘッダーまで設計します。
操作履歴と解説を振り返れます。
Python 3.15 / 固定の公開試料
from sd08 import *
lab = Upload()
body = b"item,quantity\r\npen,02\r\n"
print(len(body),list(csv.reader(io.StringIO(body.decode()))))23 [['item', 'quantity'], ['pen', '02']]
print(lab.save("alice","orders.html","text/csv",body))
print(lab.save("alice","orders.csv","text/html",body))extension_rejected media_type_rejected
print(lab.save("alice","orders.csv","text/csv",b"<p>report</p>"))
print(lab.save("alice","orders.csv","text/csv",b"item,quantity\npen,6\n"))
print(lab.db.execute("SELECT COUNT(*) FROM files").fetchone())schema_rejected row_rejected (0,)
id = lab.save("alice","../orders.csv","text/csv",body)
print(id)
print(lab.db.execute("SELECT id,owner,body FROM files").fetchall())upload-1
[('upload-1', 'alice', b'item,quantity\npen,2\n')]print(lab.get("bob",id))
print(lab.get("alice","upload-missing"))(404, {}, b'')
(404, {}, b'')status,headers,saved = lab.get("alice",id)
print(status,headers)
print(saved.decode(),end="")200 {'Content-Type': 'text/csv; charset=utf-8', 'Content-Disposition': 'attachment; filename="export.csv"', 'X-Content-Type-Options': 'nosniff'}
item,quantity
pen,2print(lab.save("alice","large.csv","text/csv",b"x"*129))
print(lab.db.execute("SELECT COUNT(*),length(body),digest FROM files").fetchone())
print(hashlib.sha256(saved).hexdigest())too_large (1, 20, '95ee0cf5a7ce498ad92dc52f43e8ea5eda49134453785ce04272565b115f7f5c') 95ee0cf5a7ce498ad92dc52f43e8ea5eda49134453785ce04272565b115f7f5c
演習完了