Home | Trees | Indices | Help |
|
---|
|
1 """Tests for managing HTTP issues (malformed requests, etc).""" 2 3 import errno 4 import mimetypes 5 import socket 6 import sys 7 8 import cherrypy 9 from cherrypy._cpcompat import HTTPConnection, HTTPSConnection, ntob, py3k 10 1113 """Return (content_type, body) ready for httplib.HTTP instance. 14 15 files: a sequence of (name, filename, value) tuples for multipart uploads. 16 """ 17 BOUNDARY = '________ThIs_Is_tHe_bouNdaRY_$' 18 L = [] 19 for key, filename, value in files: 20 L.append('--' + BOUNDARY) 21 L.append('Content-Disposition: form-data; name="%s"; filename="%s"' % 22 (key, filename)) 23 ct = mimetypes.guess_type(filename)[0] or 'application/octet-stream' 24 L.append('Content-Type: %s' % ct) 25 L.append('') 26 L.append(value) 27 L.append('--' + BOUNDARY + '--') 28 L.append('') 29 body = '\r\n'.join(L) 30 content_type = 'multipart/form-data; boundary=%s' % BOUNDARY 31 return content_type, body32 33 34 35 36 from cherrypy.test import helper 37 48 no_body.exposed = True 49 no_body._cp_config = {'request.process_request_body': False} 50 51 def post_multipart(self, file): 52 """Return a summary ("a * 65536\nb * 65536") of the uploaded file.""" 53 contents = file.file.read() 54 summary = [] 55 curchar = None 56 count = 0 57 for c in contents: 58 if c == curchar: 59 count += 1 60 else: 61 if count: 62 if py3k: curchar = chr(curchar) 63 summary.append("%s * %d" % (curchar, count)) 64 count = 1 65 curchar = c 66 if count: 67 if py3k: curchar = chr(curchar) 68 summary.append("%s * %d" % (curchar, count)) 69 return ", ".join(summary) 70 post_multipart.exposed = True 71 72 cherrypy.tree.mount(Root()) 73 cherrypy.config.update({'server.max_request_body_size': 30000000}) 74 setup_server = staticmethod(setup_server) 7577 # "The presence of a message-body in a request is signaled by the 78 # inclusion of a Content-Length or Transfer-Encoding header field in 79 # the request's message-headers." 80 # 81 # Send a message with neither header and no body. Even though 82 # the request is of method POST, this should be OK because we set 83 # request.process_request_body to False for our handler. 84 if self.scheme == "https": 85 c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT)) 86 else: 87 c = HTTPConnection('%s:%s' % (self.interface(), self.PORT)) 88 c.request("POST", "/no_body") 89 response = c.getresponse() 90 self.body = response.fp.read() 91 self.status = str(response.status) 92 self.assertStatus(200) 93 self.assertBody(ntob('Hello world!')) 94 95 # Now send a message that has no Content-Length, but does send a body. 96 # Verify that CP times out the socket and responds 97 # with 411 Length Required. 98 if self.scheme == "https": 99 c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT)) 100 else: 101 c = HTTPConnection('%s:%s' % (self.interface(), self.PORT)) 102 c.request("POST", "/") 103 response = c.getresponse() 104 self.body = response.fp.read() 105 self.status = str(response.status) 106 self.assertStatus(411)107109 alphabet = "abcdefghijklmnopqrstuvwxyz" 110 # generate file contents for a large post 111 contents = "".join([c * 65536 for c in alphabet]) 112 113 # encode as multipart form data 114 files=[('file', 'file.txt', contents)] 115 content_type, body = encode_multipart_formdata(files) 116 body = body.encode('Latin-1') 117 118 # post file 119 if self.scheme == 'https': 120 c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT)) 121 else: 122 c = HTTPConnection('%s:%s' % (self.interface(), self.PORT)) 123 c.putrequest('POST', '/post_multipart') 124 c.putheader('Content-Type', content_type) 125 c.putheader('Content-Length', str(len(body))) 126 c.endheaders() 127 c.send(body) 128 129 response = c.getresponse() 130 self.body = response.fp.read() 131 self.status = str(response.status) 132 self.assertStatus(200) 133 self.assertBody(", ".join(["%s * 65536" % c for c in alphabet]))134136 if getattr(cherrypy.server, "using_apache", False): 137 return self.skip("skipped due to known Apache differences...") 138 139 # Test missing version in Request-Line 140 if self.scheme == 'https': 141 c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT)) 142 else: 143 c = HTTPConnection('%s:%s' % (self.interface(), self.PORT)) 144 c._output(ntob('GET /')) 145 c._send_output() 146 if hasattr(c, 'strict'): 147 response = c.response_class(c.sock, strict=c.strict, method='GET') 148 else: 149 # Python 3.2 removed the 'strict' feature, saying: 150 # "http.client now always assumes HTTP/1.x compliant servers." 151 response = c.response_class(c.sock, method='GET') 152 response.begin() 153 self.assertEqual(response.status, 400) 154 self.assertEqual(response.fp.read(22), ntob("Malformed Request-Line")) 155 c.close()156158 if self.scheme == 'https': 159 c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT)) 160 else: 161 c = HTTPConnection('%s:%s' % (self.interface(), self.PORT)) 162 c.putrequest('GET', '/') 163 c.putheader('Content-Type', 'text/plain') 164 # See http://www.cherrypy.org/ticket/941 165 c._output(ntob('Re, 1.2.3.4#015#012')) 166 c.endheaders() 167 168 response = c.getresponse() 169 self.status = str(response.status) 170 self.assertStatus(400) 171 self.body = response.fp.read(20) 172 self.assertBody("Illegal header line.")173175 if self.scheme != 'https': 176 return self.skip("skipped (not running HTTPS)... ") 177 178 # Try connecting without SSL. 179 conn = HTTPConnection('%s:%s' % (self.interface(), self.PORT)) 180 conn.putrequest("GET", "/", skip_host=True) 181 conn.putheader("Host", self.HOST) 182 conn.endheaders() 183 response = conn.response_class(conn.sock, method="GET") 184 try: 185 response.begin() 186 self.assertEqual(response.status, 400) 187 self.body = response.read() 188 self.assertBody("The client sent a plain HTTP request, but this " 189 "server only speaks HTTPS on this port.") 190 except socket.error: 191 e = sys.exc_info()[1] 192 # "Connection reset by peer" is also acceptable. 193 if e.errno != errno.ECONNRESET: 194 raise195197 # Connect without SSL regardless of server.scheme 198 c = HTTPConnection('%s:%s' % (self.interface(), self.PORT)) 199 c._output(ntob('gjkgjklsgjklsgjkljklsg')) 200 c._send_output() 201 response = c.response_class(c.sock, method="GET") 202 try: 203 response.begin() 204 self.assertEqual(response.status, 400) 205 self.assertEqual(response.fp.read(22), ntob("Malformed Request-Line")) 206 c.close() 207 except socket.error: 208 e = sys.exc_info()[1] 209 # "Connection reset by peer" is also acceptable. 210 if e.errno != errno.ECONNRESET: 211 raise212
Home | Trees | Indices | Help |
|
---|
Generated by Epydoc 3.0.1 on Fri Sep 27 14:39:47 2013 | http://epydoc.sourceforge.net |