Fix duplicate test numbers in extra.decTest
[python.git] / Lib / CGIHTTPServer.py
blobad49e93e04507d7b620bf5da01238bbb5d2fc0c1
1 """CGI-savvy HTTP Server.
3 This module builds on SimpleHTTPServer by implementing GET and POST
4 requests to cgi-bin scripts.
6 If the os.fork() function is not present (e.g. on Windows),
7 os.popen2() is used as a fallback, with slightly altered semantics; if
8 that function is not present either (e.g. on Macintosh), only Python
9 scripts are supported, and they are executed by the current process.
11 In all cases, the implementation is intentionally naive -- all
12 requests are executed sychronously.
14 SECURITY WARNING: DON'T USE THIS CODE UNLESS YOU ARE INSIDE A FIREWALL
15 -- it may execute arbitrary Python code or external programs.
17 Note that status code 200 is sent prior to execution of a CGI script, so
18 scripts cannot send other status codes such as 302 (redirect).
19 """
22 __version__ = "0.4"
24 __all__ = ["CGIHTTPRequestHandler"]
26 import os
27 import sys
28 import urllib
29 import BaseHTTPServer
30 import SimpleHTTPServer
31 import select
34 class CGIHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
36 """Complete HTTP server with GET, HEAD and POST commands.
38 GET and HEAD also support running CGI scripts.
40 The POST command is *only* implemented for CGI scripts.
42 """
44 # Determine platform specifics
45 have_fork = hasattr(os, 'fork')
46 have_popen2 = hasattr(os, 'popen2')
47 have_popen3 = hasattr(os, 'popen3')
49 # Make rfile unbuffered -- we need to read one line and then pass
50 # the rest to a subprocess, so we can't use buffered input.
51 rbufsize = 0
53 def do_POST(self):
54 """Serve a POST request.
56 This is only implemented for CGI scripts.
58 """
60 if self.is_cgi():
61 self.run_cgi()
62 else:
63 self.send_error(501, "Can only POST to CGI scripts")
65 def send_head(self):
66 """Version of send_head that support CGI scripts"""
67 if self.is_cgi():
68 return self.run_cgi()
69 else:
70 return SimpleHTTPServer.SimpleHTTPRequestHandler.send_head(self)
72 def is_cgi(self):
73 """Test whether self.path corresponds to a CGI script.
75 Returns True and updates the cgi_info attribute to the tuple
76 (dir, rest) if self.path requires running a CGI script.
77 Returns False otherwise.
79 If any exception is raised, the caller should assume that
80 self.path was rejected as invalid and act accordingly.
82 The default implementation tests whether the normalized url
83 path begins with one of the strings in self.cgi_directories
84 (and the next character is a '/' or the end of the string).
85 """
86 splitpath = _url_collapse_path_split(self.path)
87 if splitpath[0] in self.cgi_directories:
88 self.cgi_info = splitpath
89 return True
90 return False
92 cgi_directories = ['/cgi-bin', '/htbin']
94 def is_executable(self, path):
95 """Test whether argument path is an executable file."""
96 return executable(path)
98 def is_python(self, path):
99 """Test whether argument path is a Python script."""
100 head, tail = os.path.splitext(path)
101 return tail.lower() in (".py", ".pyw")
103 def run_cgi(self):
104 """Execute a CGI script."""
105 path = self.path
106 dir, rest = self.cgi_info
108 i = path.find('/', len(dir) + 1)
109 while i >= 0:
110 nextdir = path[:i]
111 nextrest = path[i+1:]
113 scriptdir = self.translate_path(nextdir)
114 if os.path.isdir(scriptdir):
115 dir, rest = nextdir, nextrest
116 i = path.find('/', len(dir) + 1)
117 else:
118 break
120 # find an explicit query string, if present.
121 i = rest.rfind('?')
122 if i >= 0:
123 rest, query = rest[:i], rest[i+1:]
124 else:
125 query = ''
127 # dissect the part after the directory name into a script name &
128 # a possible additional path, to be stored in PATH_INFO.
129 i = rest.find('/')
130 if i >= 0:
131 script, rest = rest[:i], rest[i:]
132 else:
133 script, rest = rest, ''
135 scriptname = dir + '/' + script
136 scriptfile = self.translate_path(scriptname)
137 if not os.path.exists(scriptfile):
138 self.send_error(404, "No such CGI script (%r)" % scriptname)
139 return
140 if not os.path.isfile(scriptfile):
141 self.send_error(403, "CGI script is not a plain file (%r)" %
142 scriptname)
143 return
144 ispy = self.is_python(scriptname)
145 if not ispy:
146 if not (self.have_fork or self.have_popen2 or self.have_popen3):
147 self.send_error(403, "CGI script is not a Python script (%r)" %
148 scriptname)
149 return
150 if not self.is_executable(scriptfile):
151 self.send_error(403, "CGI script is not executable (%r)" %
152 scriptname)
153 return
155 # Reference: http://hoohoo.ncsa.uiuc.edu/cgi/env.html
156 # XXX Much of the following could be prepared ahead of time!
157 env = {}
158 env['SERVER_SOFTWARE'] = self.version_string()
159 env['SERVER_NAME'] = self.server.server_name
160 env['GATEWAY_INTERFACE'] = 'CGI/1.1'
161 env['SERVER_PROTOCOL'] = self.protocol_version
162 env['SERVER_PORT'] = str(self.server.server_port)
163 env['REQUEST_METHOD'] = self.command
164 uqrest = urllib.unquote(rest)
165 env['PATH_INFO'] = uqrest
166 env['PATH_TRANSLATED'] = self.translate_path(uqrest)
167 env['SCRIPT_NAME'] = scriptname
168 if query:
169 env['QUERY_STRING'] = query
170 host = self.address_string()
171 if host != self.client_address[0]:
172 env['REMOTE_HOST'] = host
173 env['REMOTE_ADDR'] = self.client_address[0]
174 authorization = self.headers.getheader("authorization")
175 if authorization:
176 authorization = authorization.split()
177 if len(authorization) == 2:
178 import base64, binascii
179 env['AUTH_TYPE'] = authorization[0]
180 if authorization[0].lower() == "basic":
181 try:
182 authorization = base64.decodestring(authorization[1])
183 except binascii.Error:
184 pass
185 else:
186 authorization = authorization.split(':')
187 if len(authorization) == 2:
188 env['REMOTE_USER'] = authorization[0]
189 # XXX REMOTE_IDENT
190 if self.headers.typeheader is None:
191 env['CONTENT_TYPE'] = self.headers.type
192 else:
193 env['CONTENT_TYPE'] = self.headers.typeheader
194 length = self.headers.getheader('content-length')
195 if length:
196 env['CONTENT_LENGTH'] = length
197 referer = self.headers.getheader('referer')
198 if referer:
199 env['HTTP_REFERER'] = referer
200 accept = []
201 for line in self.headers.getallmatchingheaders('accept'):
202 if line[:1] in "\t\n\r ":
203 accept.append(line.strip())
204 else:
205 accept = accept + line[7:].split(',')
206 env['HTTP_ACCEPT'] = ','.join(accept)
207 ua = self.headers.getheader('user-agent')
208 if ua:
209 env['HTTP_USER_AGENT'] = ua
210 co = filter(None, self.headers.getheaders('cookie'))
211 if co:
212 env['HTTP_COOKIE'] = ', '.join(co)
213 # XXX Other HTTP_* headers
214 # Since we're setting the env in the parent, provide empty
215 # values to override previously set values
216 for k in ('QUERY_STRING', 'REMOTE_HOST', 'CONTENT_LENGTH',
217 'HTTP_USER_AGENT', 'HTTP_COOKIE', 'HTTP_REFERER'):
218 env.setdefault(k, "")
219 os.environ.update(env)
221 self.send_response(200, "Script output follows")
223 decoded_query = query.replace('+', ' ')
225 if self.have_fork:
226 # Unix -- fork as we should
227 args = [script]
228 if '=' not in decoded_query:
229 args.append(decoded_query)
230 nobody = nobody_uid()
231 self.wfile.flush() # Always flush before forking
232 pid = os.fork()
233 if pid != 0:
234 # Parent
235 pid, sts = os.waitpid(pid, 0)
236 # throw away additional data [see bug #427345]
237 while select.select([self.rfile], [], [], 0)[0]:
238 if not self.rfile.read(1):
239 break
240 if sts:
241 self.log_error("CGI script exit status %#x", sts)
242 return
243 # Child
244 try:
245 try:
246 os.setuid(nobody)
247 except os.error:
248 pass
249 os.dup2(self.rfile.fileno(), 0)
250 os.dup2(self.wfile.fileno(), 1)
251 os.execve(scriptfile, args, os.environ)
252 except:
253 self.server.handle_error(self.request, self.client_address)
254 os._exit(127)
256 elif self.have_popen2 or self.have_popen3:
257 # Windows -- use popen2 or popen3 to create a subprocess
258 import shutil
259 if self.have_popen3:
260 popenx = os.popen3
261 else:
262 popenx = os.popen2
263 cmdline = scriptfile
264 if self.is_python(scriptfile):
265 interp = sys.executable
266 if interp.lower().endswith("w.exe"):
267 # On Windows, use python.exe, not pythonw.exe
268 interp = interp[:-5] + interp[-4:]
269 cmdline = "%s -u %s" % (interp, cmdline)
270 if '=' not in query and '"' not in query:
271 cmdline = '%s "%s"' % (cmdline, query)
272 self.log_message("command: %s", cmdline)
273 try:
274 nbytes = int(length)
275 except (TypeError, ValueError):
276 nbytes = 0
277 files = popenx(cmdline, 'b')
278 fi = files[0]
279 fo = files[1]
280 if self.have_popen3:
281 fe = files[2]
282 if self.command.lower() == "post" and nbytes > 0:
283 data = self.rfile.read(nbytes)
284 fi.write(data)
285 # throw away additional data [see bug #427345]
286 while select.select([self.rfile._sock], [], [], 0)[0]:
287 if not self.rfile._sock.recv(1):
288 break
289 fi.close()
290 shutil.copyfileobj(fo, self.wfile)
291 if self.have_popen3:
292 errors = fe.read()
293 fe.close()
294 if errors:
295 self.log_error('%s', errors)
296 sts = fo.close()
297 if sts:
298 self.log_error("CGI script exit status %#x", sts)
299 else:
300 self.log_message("CGI script exited OK")
302 else:
303 # Other O.S. -- execute script in this process
304 save_argv = sys.argv
305 save_stdin = sys.stdin
306 save_stdout = sys.stdout
307 save_stderr = sys.stderr
308 try:
309 save_cwd = os.getcwd()
310 try:
311 sys.argv = [scriptfile]
312 if '=' not in decoded_query:
313 sys.argv.append(decoded_query)
314 sys.stdout = self.wfile
315 sys.stdin = self.rfile
316 execfile(scriptfile, {"__name__": "__main__"})
317 finally:
318 sys.argv = save_argv
319 sys.stdin = save_stdin
320 sys.stdout = save_stdout
321 sys.stderr = save_stderr
322 os.chdir(save_cwd)
323 except SystemExit, sts:
324 self.log_error("CGI script exit status %s", str(sts))
325 else:
326 self.log_message("CGI script exited OK")
329 # TODO(gregory.p.smith): Move this into an appropriate library.
330 def _url_collapse_path_split(path):
332 Given a URL path, remove extra '/'s and '.' path elements and collapse
333 any '..' references.
335 Implements something akin to RFC-2396 5.2 step 6 to parse relative paths.
337 Returns: A tuple of (head, tail) where tail is everything after the final /
338 and head is everything before it. Head will always start with a '/' and,
339 if it contains anything else, never have a trailing '/'.
341 Raises: IndexError if too many '..' occur within the path.
343 # Similar to os.path.split(os.path.normpath(path)) but specific to URL
344 # path semantics rather than local operating system semantics.
345 path_parts = []
346 for part in path.split('/'):
347 if part == '.':
348 path_parts.append('')
349 else:
350 path_parts.append(part)
351 # Filter out blank non trailing parts before consuming the '..'.
352 path_parts = [part for part in path_parts[:-1] if part] + path_parts[-1:]
353 if path_parts:
354 tail_part = path_parts.pop()
355 else:
356 tail_part = ''
357 head_parts = []
358 for part in path_parts:
359 if part == '..':
360 head_parts.pop()
361 else:
362 head_parts.append(part)
363 if tail_part and tail_part == '..':
364 head_parts.pop()
365 tail_part = ''
366 return ('/' + '/'.join(head_parts), tail_part)
369 nobody = None
371 def nobody_uid():
372 """Internal routine to get nobody's uid"""
373 global nobody
374 if nobody:
375 return nobody
376 try:
377 import pwd
378 except ImportError:
379 return -1
380 try:
381 nobody = pwd.getpwnam('nobody')[2]
382 except KeyError:
383 nobody = 1 + max(map(lambda x: x[2], pwd.getpwall()))
384 return nobody
387 def executable(path):
388 """Test for executable file."""
389 try:
390 st = os.stat(path)
391 except os.error:
392 return False
393 return st.st_mode & 0111 != 0
396 def test(HandlerClass = CGIHTTPRequestHandler,
397 ServerClass = BaseHTTPServer.HTTPServer):
398 SimpleHTTPServer.test(HandlerClass, ServerClass)
401 if __name__ == '__main__':
402 test()