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).
24 __all__
= ["CGIHTTPRequestHandler"]
30 import SimpleHTTPServer
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.
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.
54 """Serve a POST request.
56 This is only implemented for CGI scripts.
63 self
.send_error(501, "Can only POST to CGI scripts")
66 """Version of send_head that support CGI scripts"""
70 return SimpleHTTPServer
.SimpleHTTPRequestHandler
.send_head(self
)
73 """Test whether self.path corresponds to a CGI script.
75 Return a tuple (dir, rest) if self.path requires running a
76 CGI script, None if not. Note that rest begins with a
77 slash if it is not empty.
79 The default implementation tests whether the path
80 begins with one of the strings in the list
81 self.cgi_directories (and the next character is a '/'
82 or the end of the string).
88 for x
in self
.cgi_directories
:
90 if path
[:i
] == x
and (not path
[i
:] or path
[i
] == '/'):
91 self
.cgi_info
= path
[:i
], path
[i
+1:]
95 cgi_directories
= ['/cgi-bin', '/htbin']
97 def is_executable(self
, path
):
98 """Test whether argument path is an executable file."""
99 return executable(path
)
101 def is_python(self
, path
):
102 """Test whether argument path is a Python script."""
103 head
, tail
= os
.path
.splitext(path
)
104 return tail
.lower() in (".py", ".pyw")
107 """Execute a CGI script."""
109 dir, rest
= self
.cgi_info
111 i
= path
.find('/', len(dir) + 1)
114 nextrest
= path
[i
+1:]
116 scriptdir
= self
.translate_path(nextdir
)
117 if os
.path
.isdir(scriptdir
):
118 dir, rest
= nextdir
, nextrest
119 i
= path
.find('/', len(dir) + 1)
123 # find an explicit query string, if present.
126 rest
, query
= rest
[:i
], rest
[i
+1:]
130 # dissect the part after the directory name into a script name &
131 # a possible additional path, to be stored in PATH_INFO.
134 script
, rest
= rest
[:i
], rest
[i
:]
136 script
, rest
= rest
, ''
138 scriptname
= dir + '/' + script
139 scriptfile
= self
.translate_path(scriptname
)
140 if not os
.path
.exists(scriptfile
):
141 self
.send_error(404, "No such CGI script (%r)" % scriptname
)
143 if not os
.path
.isfile(scriptfile
):
144 self
.send_error(403, "CGI script is not a plain file (%r)" %
147 ispy
= self
.is_python(scriptname
)
149 if not (self
.have_fork
or self
.have_popen2
or self
.have_popen3
):
150 self
.send_error(403, "CGI script is not a Python script (%r)" %
153 if not self
.is_executable(scriptfile
):
154 self
.send_error(403, "CGI script is not executable (%r)" %
158 # Reference: http://hoohoo.ncsa.uiuc.edu/cgi/env.html
159 # XXX Much of the following could be prepared ahead of time!
161 env
['SERVER_SOFTWARE'] = self
.version_string()
162 env
['SERVER_NAME'] = self
.server
.server_name
163 env
['GATEWAY_INTERFACE'] = 'CGI/1.1'
164 env
['SERVER_PROTOCOL'] = self
.protocol_version
165 env
['SERVER_PORT'] = str(self
.server
.server_port
)
166 env
['REQUEST_METHOD'] = self
.command
167 uqrest
= urllib
.unquote(rest
)
168 env
['PATH_INFO'] = uqrest
169 env
['PATH_TRANSLATED'] = self
.translate_path(uqrest
)
170 env
['SCRIPT_NAME'] = scriptname
172 env
['QUERY_STRING'] = query
173 host
= self
.address_string()
174 if host
!= self
.client_address
[0]:
175 env
['REMOTE_HOST'] = host
176 env
['REMOTE_ADDR'] = self
.client_address
[0]
177 authorization
= self
.headers
.getheader("authorization")
179 authorization
= authorization
.split()
180 if len(authorization
) == 2:
181 import base64
, binascii
182 env
['AUTH_TYPE'] = authorization
[0]
183 if authorization
[0].lower() == "basic":
185 authorization
= base64
.decodestring(authorization
[1])
186 except binascii
.Error
:
189 authorization
= authorization
.split(':')
190 if len(authorization
) == 2:
191 env
['REMOTE_USER'] = authorization
[0]
193 if self
.headers
.typeheader
is None:
194 env
['CONTENT_TYPE'] = self
.headers
.type
196 env
['CONTENT_TYPE'] = self
.headers
.typeheader
197 length
= self
.headers
.getheader('content-length')
199 env
['CONTENT_LENGTH'] = length
200 referer
= self
.headers
.getheader('referer')
202 env
['HTTP_REFERER'] = referer
204 for line
in self
.headers
.getallmatchingheaders('accept'):
205 if line
[:1] in "\t\n\r ":
206 accept
.append(line
.strip())
208 accept
= accept
+ line
[7:].split(',')
209 env
['HTTP_ACCEPT'] = ','.join(accept
)
210 ua
= self
.headers
.getheader('user-agent')
212 env
['HTTP_USER_AGENT'] = ua
213 co
= filter(None, self
.headers
.getheaders('cookie'))
215 env
['HTTP_COOKIE'] = ', '.join(co
)
216 # XXX Other HTTP_* headers
217 # Since we're setting the env in the parent, provide empty
218 # values to override previously set values
219 for k
in ('QUERY_STRING', 'REMOTE_HOST', 'CONTENT_LENGTH',
220 'HTTP_USER_AGENT', 'HTTP_COOKIE', 'HTTP_REFERER'):
221 env
.setdefault(k
, "")
222 os
.environ
.update(env
)
224 self
.send_response(200, "Script output follows")
226 decoded_query
= query
.replace('+', ' ')
229 # Unix -- fork as we should
231 if '=' not in decoded_query
:
232 args
.append(decoded_query
)
233 nobody
= nobody_uid()
234 self
.wfile
.flush() # Always flush before forking
238 pid
, sts
= os
.waitpid(pid
, 0)
239 # throw away additional data [see bug #427345]
240 while select
.select([self
.rfile
], [], [], 0)[0]:
241 if not self
.rfile
.read(1):
244 self
.log_error("CGI script exit status %#x", sts
)
252 os
.dup2(self
.rfile
.fileno(), 0)
253 os
.dup2(self
.wfile
.fileno(), 1)
254 os
.execve(scriptfile
, args
, os
.environ
)
256 self
.server
.handle_error(self
.request
, self
.client_address
)
259 elif self
.have_popen2
or self
.have_popen3
:
260 # Windows -- use popen2 or popen3 to create a subprocess
267 if self
.is_python(scriptfile
):
268 interp
= sys
.executable
269 if interp
.lower().endswith("w.exe"):
270 # On Windows, use python.exe, not pythonw.exe
271 interp
= interp
[:-5] + interp
[-4:]
272 cmdline
= "%s -u %s" % (interp
, cmdline
)
273 if '=' not in query
and '"' not in query
:
274 cmdline
= '%s "%s"' % (cmdline
, query
)
275 self
.log_message("command: %s", cmdline
)
278 except (TypeError, ValueError):
280 files
= popenx(cmdline
, 'b')
285 if self
.command
.lower() == "post" and nbytes
> 0:
286 data
= self
.rfile
.read(nbytes
)
288 # throw away additional data [see bug #427345]
289 while select
.select([self
.rfile
._sock
], [], [], 0)[0]:
290 if not self
.rfile
._sock
.recv(1):
293 shutil
.copyfileobj(fo
, self
.wfile
)
298 self
.log_error('%s', errors
)
301 self
.log_error("CGI script exit status %#x", sts
)
303 self
.log_message("CGI script exited OK")
306 # Other O.S. -- execute script in this process
308 save_stdin
= sys
.stdin
309 save_stdout
= sys
.stdout
310 save_stderr
= sys
.stderr
312 save_cwd
= os
.getcwd()
314 sys
.argv
= [scriptfile
]
315 if '=' not in decoded_query
:
316 sys
.argv
.append(decoded_query
)
317 sys
.stdout
= self
.wfile
318 sys
.stdin
= self
.rfile
319 execfile(scriptfile
, {"__name__": "__main__"})
322 sys
.stdin
= save_stdin
323 sys
.stdout
= save_stdout
324 sys
.stderr
= save_stderr
326 except SystemExit, sts
:
327 self
.log_error("CGI script exit status %s", str(sts
))
329 self
.log_message("CGI script exited OK")
335 """Internal routine to get nobody's uid"""
344 nobody
= pwd
.getpwnam('nobody')[2]
346 nobody
= 1 + max(map(lambda x
: x
[2], pwd
.getpwall()))
350 def executable(path
):
351 """Test for executable file."""
356 return st
.st_mode
& 0111 != 0
359 def test(HandlerClass
= CGIHTTPRequestHandler
,
360 ServerClass
= BaseHTTPServer
.HTTPServer
):
361 SimpleHTTPServer
.test(HandlerClass
, ServerClass
)
364 if __name__
== '__main__':