import cgi import os import sys import tempfile import unittest from collections import namedtuple from io import StringIO, BytesIO from test import support from test.support import warnings_helper class HackedSysModule: # The regression test will have real values in sys.argv, which # will completely confuse the test of the cgi module argv = [] stdin = sys.stdin cgi.sys = HackedSysModule() class ComparableException: def __init__(self, err): self.err = err def __str__(self): return str(self.err) def __eq__(self, anExc): if not isinstance(anExc, Exception): return NotImplemented return (self.err.__class__ == anExc.__class__ and self.err.args == anExc.args) def __getattr__(self, attr): return getattr(self.err, attr) def do_test(buf, method): env = {} if method == "GET": fp = None env['REQUEST_METHOD'] = 'GET' env['QUERY_STRING'] = buf elif method == "POST": fp = BytesIO(buf.encode('latin-1')) # FieldStorage expects bytes env['REQUEST_METHOD'] = 'POST' env['CONTENT_TYPE'] = 'application/x-www-form-urlencoded' env['CONTENT_LENGTH'] = str(len(buf)) else: raise ValueError("unknown method: %s" % method) try: return cgi.parse(fp, env, strict_parsing=1) except Exception as err: return ComparableException(err) parse_strict_test_cases = [ ("", ValueError("bad query field: ''")), ("&", ValueError("bad query field: ''")), ("&&", ValueError("bad query field: ''")), # Should the next few really be valid? ("=", {}), ("=&=", {}), # This rest seem to make sense ("=a", {'': ['a']}), ("&=a", ValueError("bad query field: ''")), ("=a&", ValueError("bad query field: ''")), ("=&a", ValueError("bad query field: 'a'")), ("b=a", {'b': ['a']}), ("b+=a", {'b ': ['a']}), ("a=b=a", {'a': ['b=a']}), ("a=+b=a", {'a': [' b=a']}), ("&b=a", ValueError("bad query field: ''")), ("b&=a", ValueError("bad query field: 'b'")), ("a=a+b&b=b+c", {'a': ['a b'], 'b': ['b c']}), ("a=a+b&a=b+a", {'a': ['a b', 'b a']}), ("x=1&y=2.0&z=2-3.%2b0", {'x': ['1'], 'y': ['2.0'], 'z': ['2-3.+0']}), ("Hbc5161168c542333633315dee1182227:key_store_seqid=400006&cuyer=r&view=bustomer&order_id=0bb2e248638833d48cb7fed300000f1b&expire=964546263&lobale=en-US&kid=130003.300038&ss=env", {'Hbc5161168c542333633315dee1182227:key_store_seqid': ['400006'], 'cuyer': ['r'], 'expire': ['964546263'], 'kid': ['130003.300038'], 'lobale': ['en-US'], 'order_id': ['0bb2e248638833d48cb7fed300000f1b'], 'ss': ['env'], 'view': ['bustomer'], }), ("group_id=5470&set=custom&_assigned_to=31392&_status=1&_category=100&SUBMIT=Browse", {'SUBMIT': ['Browse'], '_assigned_to': ['31392'], '_category': ['100'], '_status': ['1'], 'group_id': ['5470'], 'set': ['custom'], }) ] def norm(seq): return sorted(seq, key=repr) def first_elts(list): return [p[0] for p in list] def first_second_elts(list): return [(p[0], p[1][0]) for p in list] def gen_result(data, environ): encoding = 'latin-1' fake_stdin = BytesIO(data.encode(encoding)) fake_stdin.seek(0) form = cgi.FieldStorage(fp=fake_stdin, environ=environ, encoding=encoding) result = {} for k, v in dict(form).items(): result[k] = isinstance(v, list) and form.getlist(k) or v.value return result class CgiTests(unittest.TestCase): def test_parse_multipart(self): fp = BytesIO(POSTDATA.encode('latin1')) env = {'boundary': BOUNDARY.encode('latin1'), 'CONTENT-LENGTH': '558'} result = cgi.parse_multipart(fp, env) expected = {'submit': [' Add '], 'id': ['1234'], 'file': [b'Testing 123.\n'], 'title': ['']} self.assertEqual(result, expected) def test_parse_multipart_without_content_length(self): POSTDATA = '''--JfISa01 Content-Disposition: form-data; name="submit-name" just a string --JfISa01-- ''' fp = BytesIO(POSTDATA.encode('latin1')) env = {'boundary': 'JfISa01'.encode('latin1')} result = cgi.parse_multipart(fp, env) expected = {'submit-name': ['just a string\n']} self.assertEqual(result, expected) def test_parse_multipart_invalid_encoding(self): BOUNDARY = "JfISa01" POSTDATA = """--JfISa01 Content-Disposition: form-data; name="submit-name" Content-Length: 3 \u2603 --JfISa01""" fp = BytesIO(POSTDATA.encode('utf8')) env = {'boundary': BOUNDARY.encode('latin1'), 'CONTENT-LENGTH': str(len(POSTDATA.encode('utf8')))} result = cgi.parse_multipart(fp, env, encoding="ascii", errors="surrogateescape") expected = {'submit-name': ["\udce2\udc98\udc83"]} self.assertEqual(result, expected) self.assertEqual("\u2603".encode('utf8'), result["submit-name"][0].encode('utf8', 'surrogateescape')) def test_fieldstorage_properties(self): fs = cgi.FieldStorage() self.assertFalse(fs) self.assertIn("FieldStorage", repr(fs)) self.assertEqual(list(fs), list(fs.keys())) fs.list.append(namedtuple('MockFieldStorage', 'name')('fieldvalue')) self.assertTrue(fs) def test_fieldstorage_invalid(self): self.assertRaises(TypeError, cgi.FieldStorage, "not-a-file-obj", environ={"REQUEST_METHOD":"PUT"}) self.assertRaises(TypeError, cgi.FieldStorage, "foo", "bar") fs = cgi.FieldStorage(headers={'content-type':'text/plain'}) self.assertRaises(TypeError, bool, fs) def test_strict(self): for orig, expect in parse_strict_test_cases: # Test basic parsing d = do_test(orig, "GET") self.assertEqual(d, expect, "Error parsing %s method GET" % repr(orig)) d = do_test(orig, "POST") self.assertEqual(d, expect, "Error parsing %s method POST" % repr(orig)) env = {'QUERY_STRING': orig} fs = cgi.FieldStorage(environ=env) if isinstance(expect, dict): # test dict interface self.assertEqual(len(expect), len(fs)) self.assertCountEqual(expect.keys(), fs.keys()) ##self.assertEqual(norm(expect.values()), norm(fs.values())) ##self.assertEqual(norm(expect.items()), norm(fs.items())) self.assertEqual(fs.getvalue("nonexistent field", "default"), "default") # test individual fields for key in expect.keys(): expect_val = expect[key] self.assertIn(key, fs) if len(expect_val) > 1: self.assertEqual(fs.getvalue(key), expect_val) else: self.assertEqual(fs.getvalue(key), expect_val[0]) def test_separator(self): parse_semicolon = [ ("x=1;y=2.0", {'x': ['1'], 'y': ['2.0']}), ("x=1;y=2.0;z=2-3.%2b0", {'x': ['1'], 'y': ['2.0'], 'z': ['2-3.+0']}), (";", ValueError("bad query field: ''")), (";;", ValueError("bad query field: ''")), ("=;a", ValueError("bad query field: 'a'")), (";b=a", ValueError("bad query field: ''")), ("b;=a", ValueError("bad query field: 'b'")), ("a=a+b;b=b+c", {'a': ['a b'], 'b': ['b c']}), ("a=a+b;a=b+a", {'a': ['a b', 'b a']}), ] for orig, expect in parse_semicolon: env = {'QUERY_STRING': orig} fs = cgi.FieldStorage(separator=';', environ=env) if isinstance(expect, dict): for key in expect.keys(): expect_val = expect[key] self.assertIn(key, fs) if len(expect_val) > 1: self.assertEqual(fs.getvalue(key), expect_val) else: self.assertEqual(fs.getvalue(key), expect_val[0]) @warnings_helper.ignore_warnings(category=DeprecationWarning) def test_log(self): cgi.log("Testing") cgi.logfp = StringIO() cgi.initlog("%s", "Testing initlog 1") cgi.log("%s", "Testing log 2") self.assertEqual(cgi.logfp.getvalue(), "Testing initlog 1\nTesting log 2\n") if os.path.exists(os.devnull): cgi.logfp = None cgi.logfile = os.devnull cgi.initlog("%s", "Testing log 3") self.addCleanup(cgi.closelog) cgi.log("Testing log 4") def test_fieldstorage_readline(self): # FieldStorage uses readline, which has the capacity to read all # contents of the input file into memory; we use readline's size argument # to prevent that for files that do not contain any newlines in # non-GET/HEAD requests class TestReadlineFile: def __init__(self, file): self.file = file self.numcalls = 0 def readline(self, size=None): self.numcalls += 1 if size: return self.file.readline(size) else: return self.file.readline() def __getattr__(self, name): file = self.__dict__['file'] a = getattr(file, name) if not isinstance(a, int): setattr(self, name, a) return a f = TestReadlineFile(tempfile.TemporaryFile("wb+")) self.addCleanup(f.close) f.write(b'x' * 256 * 1024) f.seek(0) env = {'REQUEST_METHOD':'PUT'} fs = cgi.FieldStorage(fp=f, environ=env) self.addCleanup(fs.file.close) # if we're not chunking properly, readline is only called twice # (by read_binary); if we are chunking properly, it will be called 5 times # as long as the chunksize is 1 << 16. self.assertGreater(f.numcalls, 2) f.close() def test_fieldstorage_multipart(self): #Test basic FieldStorage multipart parsing env = { 'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY), 'CONTENT_LENGTH': '558'} fp = BytesIO(POSTDATA.encode('latin-1')) fs = cgi.FieldStorage(fp, environ=env, encoding="latin-1") self.assertEqual(len(fs.list), 4) expect = [{'name':'id', 'filename':None, 'value':'1234'}, {'name':'title', 'filename':None, 'value':''}, {'name':'file', 'filename':'test.txt', 'value':b'Testing 123.\n'}, {'name':'submit', 'filename':None, 'value':' Add '}] for x in range(len(fs.list)): for k, exp in expect[x].items(): got = getattr(fs.list[x], k) self.assertEqual(got, exp) def test_fieldstorage_multipart_leading_whitespace(self): env = { 'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY), 'CONTENT_LENGTH': '560'} # Add some leading whitespace to our post data that will cause the # first line to not be the innerboundary. fp = BytesIO(b"\r\n" + POSTDATA.encode('latin-1')) fs = cgi.FieldStorage(fp, environ=env, encoding="latin-1") self.assertEqual(len(fs.list), 4) expect = [{'name':'id', 'filename':None, 'value':'1234'}, {'name':'title', 'filename':None, 'value':''}, {'name':'file', 'filename':'test.txt', 'value':b'Testing 123.\n'}, {'name':'submit', 'filename':None, 'value':' Add '}] for x in range(len(fs.list)): for k, exp in expect[x].items(): got = getattr(fs.list[x], k) self.assertEqual(got, exp) def test_fieldstorage_multipart_non_ascii(self): #Test basic FieldStorage multipart parsing env = {'REQUEST_METHOD':'POST', 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY), 'CONTENT_LENGTH':'558'} for encoding in ['iso-8859-1','utf-8']: fp = BytesIO(POSTDATA_NON_ASCII.encode(encoding)) fs = cgi.FieldStorage(fp, environ=env,encoding=encoding) self.assertEqual(len(fs.list), 1) expect = [{'name':'id', 'filename':None, 'value':'\xe7\xf1\x80'}] for x in range(len(fs.list)): for k, exp in expect[x].items(): got = getattr(fs.list[x], k) self.assertEqual(got, exp) def test_fieldstorage_multipart_maxline(self): # Issue #18167 maxline = 1 << 16 self.maxDiff = None def check(content): data = """---123 Content-Disposition: form-data; name="upload"; filename="fake.txt" Content-Type: text/plain %s ---123-- """.replace('\n', '\r\n') % content environ = { 'CONTENT_LENGTH': str(len(data)), 'CONTENT_TYPE': 'multipart/form-data; boundary=-123', 'REQUEST_METHOD': 'POST', } self.assertEqual(gen_result(data, environ), {'upload': content.encode('latin1')}) check('x' * (maxline - 1)) check('x' * (maxline - 1) + '\r') check('x' * (maxline - 1) + '\r' + 'y' * (maxline - 1)) def test_fieldstorage_multipart_w3c(self): # Test basic FieldStorage multipart parsing (W3C sample) env = { 'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY_W3), 'CONTENT_LENGTH': str(len(POSTDATA_W3))} fp = BytesIO(POSTDATA_W3.encode('latin-1')) fs = cgi.FieldStorage(fp, environ=env, encoding="latin-1") self.assertEqual(len(fs.list), 2) self.assertEqual(fs.list[0].name, 'submit-name') self.assertEqual(fs.list[0].value, 'Larry') self.assertEqual(fs.list[1].name, 'files') files = fs.list[1].value self.assertEqual(len(files), 2) expect = [{'name': None, 'filename': 'file1.txt', 'value': b'... contents of file1.txt ...'}, {'name': None, 'filename': 'file2.gif', 'value': b'...contents of file2.gif...'}] for x in range(len(files)): for k, exp in expect[x].items(): got = getattr(files[x], k) self.assertEqual(got, exp) def test_fieldstorage_part_content_length(self): BOUNDARY = "JfISa01" POSTDATA = """--JfISa01 Content-Disposition: form-data; name="submit-name" Content-Length: 5 Larry --JfISa01""" env = { 'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY), 'CONTENT_LENGTH': str(len(POSTDATA))} fp = BytesIO(POSTDATA.encode('latin-1')) fs = cgi.FieldStorage(fp, environ=env, encoding="latin-1") self.assertEqual(len(fs.list), 1) self.assertEqual(fs.list[0].name, 'submit-name') self.assertEqual(fs.list[0].value, 'Larry') def test_field_storage_multipart_no_content_length(self): fp = BytesIO(b"""--MyBoundary Content-Disposition: form-data; name="my-arg"; filename="foo" Test --MyBoundary-- """) env = { "REQUEST_METHOD": "POST", "CONTENT_TYPE": "multipart/form-data; boundary=MyBoundary", "wsgi.input": fp, } fields = cgi.FieldStorage(fp, environ=env) self.assertEqual(len(fields["my-arg"].file.read()), 5) def test_fieldstorage_as_context_manager(self): fp = BytesIO(b'x' * 10) env = {'REQUEST_METHOD': 'PUT'} with cgi.FieldStorage(fp=fp, environ=env) as fs: content = fs.file.read() self.assertFalse(fs.file.closed) self.assertTrue(fs.file.closed) self.assertEqual(content, 'x' * 10) with self.assertRaisesRegex(ValueError, 'I/O operation on closed file'): fs.file.read() _qs_result = { 'key1': 'value1', 'key2': ['value2x', 'value2y'], 'key3': 'value3', 'key4': 'value4' } def testQSAndUrlEncode(self): data = "key2=value2x&key3=value3&key4=value4" environ = { 'CONTENT_LENGTH': str(len(data)), 'CONTENT_TYPE': 'application/x-www-form-urlencoded', 'QUERY_STRING': 'key1=value1&key2=value2y', 'REQUEST_METHOD': 'POST', } v = gen_result(data, environ) self.assertEqual(self._qs_result, v) def test_max_num_fields(self): # For application/x-www-form-urlencoded data = '&'.join(['a=a']*11) environ = { 'CONTENT_LENGTH': str(len(data)), 'CONTENT_TYPE': 'application/x-www-form-urlencoded', 'REQUEST_METHOD': 'POST', } with self.assertRaises(ValueError): cgi.FieldStorage( fp=BytesIO(data.encode()), environ=environ, max_num_fields=10, ) # For multipart/form-data data = """---123 Content-Disposition: form-data; name="a" 3 ---123 Content-Type: application/x-www-form-urlencoded a=4 ---123 Content-Type: application/x-www-form-urlencoded a=5 ---123-- """ environ = { 'CONTENT_LENGTH': str(len(data)), 'CONTENT_TYPE': 'multipart/form-data; boundary=-123', 'QUERY_STRING': 'a=1&a=2', 'REQUEST_METHOD': 'POST', } # 2 GET entities # 1 top level POST entities # 1 entity within the second POST entity # 1 entity within the third POST entity with self.assertRaises(ValueError): cgi.FieldStorage( fp=BytesIO(data.encode()), environ=environ, max_num_fields=4, ) cgi.FieldStorage( fp=BytesIO(data.encode()), environ=environ, max_num_fields=5, ) def testQSAndFormData(self): data = """---123 Content-Disposition: form-data; name="key2" value2y ---123 Content-Disposition: form-data; name="key3" value3 ---123 Content-Disposition: form-data; name="key4" value4 ---123-- """ environ = { 'CONTENT_LENGTH': str(len(data)), 'CONTENT_TYPE': 'multipart/form-data; boundary=-123', 'QUERY_STRING': 'key1=value1&key2=value2x', 'REQUEST_METHOD': 'POST', } v = gen_result(data, environ) self.assertEqual(self._qs_result, v) def testQSAndFormDataFile(self): data = """---123 Content-Disposition: form-data; name="key2" value2y ---123 Content-Disposition: form-data; name="key3" value3 ---123 Content-Disposition: form-data; name="key4" value4 ---123 Content-Disposition: form-data; name="upload"; filename="fake.txt" Content-Type: text/plain this is the content of the fake file ---123-- """ environ = { 'CONTENT_LENGTH': str(len(data)), 'CONTENT_TYPE': 'multipart/form-data; boundary=-123', 'QUERY_STRING': 'key1=value1&key2=value2x', 'REQUEST_METHOD': 'POST', } result = self._qs_result.copy() result.update({ 'upload': b'this is the content of the fake file\n' }) v = gen_result(data, environ) self.assertEqual(result, v) def test_parse_header(self): self.assertEqual( cgi.parse_header("text/plain"), ("text/plain", {})) self.assertEqual( cgi.parse_header("text/vnd.just.made.this.up ; "), ("text/vnd.just.made.this.up", {})) self.assertEqual( cgi.parse_header("text/plain;charset=us-ascii"), ("text/plain", {"charset": "us-ascii"})) self.assertEqual( cgi.parse_header('text/plain ; charset="us-ascii"'), ("text/plain", {"charset": "us-ascii"})) self.assertEqual( cgi.parse_header('text/plain ; charset="us-ascii"; another=opt'), ("text/plain", {"charset": "us-ascii", "another": "opt"})) self.assertEqual( cgi.parse_header('attachment; filename="silly.txt"'), ("attachment", {"filename": "silly.txt"})) self.assertEqual( cgi.parse_header('attachment; filename="strange;name"'), ("attachment", {"filename": "strange;name"})) self.assertEqual( cgi.parse_header('attachment; filename="strange;name";size=123;'), ("attachment", {"filename": "strange;name", "size": "123"})) self.assertEqual( cgi.parse_header('form-data; name="files"; filename="fo\\"o;bar"'), ("form-data", {"name": "files", "filename": 'fo"o;bar'})) def test_all(self): not_exported = { "logfile", "logfp", "initlog", "dolog", "nolog", "closelog", "log", "maxlen", "valid_boundary"} support.check__all__(self, cgi, not_exported=not_exported) BOUNDARY = "---------------------------721837373350705526688164684" POSTDATA = """-----------------------------721837373350705526688164684 Content-Disposition: form-data; name="id" 1234 -----------------------------721837373350705526688164684 Content-Disposition: form-data; name="title" -----------------------------721837373350705526688164684 Content-Disposition: form-data; name="file"; filename="test.txt" Content-Type: text/plain Testing 123. -----------------------------721837373350705526688164684 Content-Disposition: form-data; name="submit" Add\x20 -----------------------------721837373350705526688164684-- """ POSTDATA_NON_ASCII = """-----------------------------721837373350705526688164684 Content-Disposition: form-data; name="id" \xe7\xf1\x80 -----------------------------721837373350705526688164684 """ # http://www.w3.org/TR/html401/interact/forms.html#h-17.13.4 BOUNDARY_W3 = "AaB03x" POSTDATA_W3 = """--AaB03x Content-Disposition: form-data; name="submit-name" Larry --AaB03x Content-Disposition: form-data; name="files" Content-Type: multipart/mixed; boundary=BbC04y --BbC04y Content-Disposition: file; filename="file1.txt" Content-Type: text/plain ... contents of file1.txt ... --BbC04y Content-Disposition: file; filename="file2.gif" Content-Type: image/gif Content-Transfer-Encoding: binary ...contents of file2.gif... --BbC04y-- --AaB03x-- """ if __name__ == '__main__': unittest.main()