これはzip bombs.tar.gzに関する質問に関連していますが、gzip または bzip2 圧縮を念頭に置いています (たとえば、ファイルを受け入れる Web サービス) 。
Python は便利なtarfile モジュールを提供していますが、zipbomb に対する保護は提供していないようです。
tarfile モジュールを使用する Python コードで、zip 爆弾を検出する最もエレガントな方法は何でしょうか?
そして、もう少し簡単にするために、実際のファイルは関係ありません。入力はファイルのようなオブジェクトです (Web フレームワークによって提供され、ユーザーがアップロードしたファイルを表します)。
resourcemoduleを使用して、プロセスとその子プロセスが利用できるリソースを制限できます。
メモリ内で解凍する必要がある場合は、 resource.RLIMIT_AS(またはRLIMIT_DATA, RLIMIT_STACK) を設定できます。たとえば、コンテキスト マネージャーを使用して以前の値に自動的に復元します。
import contextlib
import resource
@contextlib.contextmanager
def limit(limit, type=resource.RLIMIT_AS):
soft_limit, hard_limit = resource.getrlimit(type)
resource.setrlimit(type, (limit, hard_limit)) # set soft limit
try:
yield
finally:
resource.setrlimit(type, (soft_limit, hard_limit)) # restore
with limit(1 << 30): # 1GB
# do the thing that might try to consume all memory
制限に達した場合。MemoryError上げられます。
これにより、限られたメモリを使用しながら、gzip ストリームの圧縮されていないサイズが決まります。
#!/usr/bin/python
import sys
import zlib
f = open(sys.argv[1], "rb")
z = zlib.decompressobj(15+16)
total = 0
while True:
buf = z.unconsumed_tail
if buf == "":
buf = f.read(1024)
if buf == "":
break
got = z.decompress(buf, 4096)
if got == "":
break
total += len(got)
print total
if z.unused_data != "" or f.read(1024) != "":
print "warning: more input after end of gzip stream"
tar ファイル内のすべてのファイルを抽出すると、必要なスペースが少し過大に見積もられます。長さには、これらのファイルと tar ディレクトリ情報が含まれます。
gzip.py コードは、入力データのサイズを除いて、解凍されるデータの量を制御しません。gzip.py では、圧縮された 1024 バイトを一度に読み取ります。したがって、圧縮されていないデータのメモリ使用量が最大約 1056768 バイト (1032 * 1024、1032:1 は deflate の最大圧縮率) であれば、gzip.py を使用できます。ここでのソリューションではzlib.decompress、圧縮されていないデータの量を制限する 2 番目の引数を使用します。gzip.py はありません。
これにより、tar 形式をデコードすることにより、抽出された tar エントリの合計サイズが正確に決定されます。
#!/usr/bin/python
import sys
import zlib
def decompn(f, z, n):
"""Return n uncompressed bytes, or fewer if at the end of the compressed
stream. This only decompresses as much as necessary, in order to
avoid excessive memory usage for highly compressed input.
"""
blk = ""
while len(blk) < n:
buf = z.unconsumed_tail
if buf == "":
buf = f.read(1024)
got = z.decompress(buf, n - len(blk))
blk += got
if got == "":
break
return blk
f = open(sys.argv[1], "rb")
z = zlib.decompressobj(15+16)
total = 0
left = 0
while True:
blk = decompn(f, z, 512)
if len(blk) < 512:
break
if left == 0:
if blk == "\0"*512:
continue
if blk[156] in ["1", "2", "3", "4", "5", "6"]:
continue
if blk[124] == 0x80:
size = 0
for i in range(125, 136):
size <<= 8
size += blk[i]
else:
size = int(blk[124:136].split()[0].split("\0")[0], 8)
if blk[156] not in ["x", "g", "X", "L", "K"]:
total += size
left = (size + 511) // 512
else:
left -= 1
print total
if blk != "":
print "warning: partial final block"
if left != 0:
print "warning: tar file ended in the middle of an entry"
if z.unused_data != "" or f.read(1024) != "":
print "warning: more input after end of gzip stream"
これの変形を使用して、爆弾の tar ファイルをスキャンできます。これには、データを解凍する前にヘッダー情報で大きなサイズを見つけることができるという利点があります。
.tar.bz2 アーカイブに関しては、Python bz2 ライブラリ (少なくとも 3.3 の時点) は、大量のメモリを消費する bz2 爆弾に対して不可避的に安全ではありません。このbz2.decompress関数は、2 番目の引数を提供していませんzlib.decompress。これは、ランレングス コーディングにより、bz2 形式の最大圧縮率が zlib よりもはるかに高いという事実によってさらに悪化します。bzip2 は、1 GB のゼロを 722 バイトに圧縮します。したがって、2 番目の引数がなくてもbz2.decompress実行できるように、入力を計測して の出力を計測することはできません。zlib.decompress解凍後の出力サイズに制限がないことは、Python インターフェイスの根本的な欠陥です。
3.3 の _bz2module.c を調べて、この問題を回避するための文書化されていない使用方法があるかどうかを確認しました。それを回避する方法はありません。そこdecompressにある関数は、提供されたすべての入力を解凍できるまで、結果バッファーを拡大し続けます。_bz2module.c を修正する必要があります。
Linux 用に開発する場合は、別のプロセスで解凍を実行し、ulimit を使用してメモリ使用量を制限できます。
import subprocess
subprocess.Popen("ulimit -v %d; ./decompression_script.py %s" % (LIMIT, FILE))
decompression_script.py は、ディスクに書き込む前にメモリ内のファイル全体を解凍する必要があることに注意してください。
答えは次のとおりだと思います。簡単で既製のソリューションはありません。これが私が今使っているものです：
class SafeUncompressor(object):
"""Small proxy class that enables external file object
support for uncompressed, bzip2 and gzip files. Works transparently, and
supports a maximum size to avoid zipbombs.
"""
blocksize = 16 * 1024
class FileTooLarge(Exception):
pass
def __init__(self, fileobj, maxsize=10*1024*1024):
self.fileobj = fileobj
self.name = getattr(self.fileobj, "name", None)
self.maxsize = maxsize
self.init()
def init(self):
import bz2
import gzip
self.pos = 0
self.fileobj.seek(0)
self.buf = ""
self.format = "plain"
magic = self.fileobj.read(2)
if magic == '\037\213':
self.format = "gzip"
self.gzipobj = gzip.GzipFile(fileobj = self.fileobj, mode = 'r')
elif magic == 'BZ':
raise IOError, "bzip2 support in SafeUncompressor disabled, as self.bz2obj.decompress is not safe"
self.format = "bz2"
self.bz2obj = bz2.BZ2Decompressor()
self.fileobj.seek(0)
def read(self, size):
b = [self.buf]
x = len(self.buf)
while x < size:
if self.format == 'gzip':
data = self.gzipobj.read(self.blocksize)
if not data:
break
elif self.format == 'bz2':
raw = self.fileobj.read(self.blocksize)
if not raw:
break
# this can already bomb here, to some extend.
# so disable bzip support until resolved.
# Also monitor http://stackoverflow.com/questions/13622706/how-to-protect-myself-from-a-gzip-or-bzip2-bomb for ideas
data = self.bz2obj.decompress(raw)
else:
data = self.fileobj.read(self.blocksize)
if not data:
break
b.append(data)
x += len(data)
if self.pos + x > self.maxsize:
self.buf = ""
self.pos = 0
raise SafeUncompressor.FileTooLarge, "Compressed file too large"
self.buf = "".join(b)
buf = self.buf[:size]
self.buf = self.buf[size:]
self.pos += len(buf)
return buf
def seek(self, pos, whence=0):
if whence != 0:
raise IOError, "SafeUncompressor only supports whence=0"
if pos < self.pos:
self.init()
self.read(pos - self.pos)
def tell(self):
return self.pos
bzip2 ではうまく機能しないため、コードの一部が無効になっています。その理由はbz2.BZ2Decompressor.decompress、不要な大量のデータチャンクがすでに生成されている可能性があるためです。