201 lines
6.0 KiB
Python
Executable File
201 lines
6.0 KiB
Python
Executable File
"""CGI-savvy HTTP Server.
|
|
|
|
This module builds on SimpleHTTPServer by implementing GET and POST
|
|
requests to cgi-bin scripts.
|
|
|
|
"""
|
|
|
|
|
|
__version__ = "0.3"
|
|
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import socket
|
|
import string
|
|
import urllib
|
|
import BaseHTTPServer
|
|
import SimpleHTTPServer
|
|
|
|
|
|
class CGIHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
|
|
|
|
"""Complete HTTP server with GET, HEAD and POST commands.
|
|
|
|
GET and HEAD also support running CGI scripts.
|
|
|
|
The POST command is *only* implemented for CGI scripts.
|
|
|
|
"""
|
|
|
|
def do_POST(self):
|
|
"""Serve a POST request.
|
|
|
|
This is only implemented for CGI scripts.
|
|
|
|
"""
|
|
|
|
if self.is_cgi():
|
|
self.run_cgi()
|
|
else:
|
|
self.send_error(501, "Can only POST to CGI scripts")
|
|
|
|
def send_head(self):
|
|
"""Version of send_head that support CGI scripts"""
|
|
if self.is_cgi():
|
|
return self.run_cgi()
|
|
else:
|
|
return SimpleHTTPServer.SimpleHTTPRequestHandler.send_head(self)
|
|
|
|
def is_cgi(self):
|
|
"""test whether PATH corresponds to a CGI script.
|
|
|
|
Return a tuple (dir, rest) if PATH requires running a
|
|
CGI script, None if not. Note that rest begins with a
|
|
slash if it is not empty.
|
|
|
|
The default implementation tests whether the path
|
|
begins with one of the strings in the list
|
|
self.cgi_directories (and the next character is a '/'
|
|
or the end of the string).
|
|
|
|
"""
|
|
|
|
path = self.path
|
|
|
|
for x in self.cgi_directories:
|
|
i = len(x)
|
|
if path[:i] == x and (not path[i:] or path[i] == '/'):
|
|
self.cgi_info = path[:i], path[i+1:]
|
|
return 1
|
|
return 0
|
|
|
|
cgi_directories = ['/cgi-bin', '/htbin']
|
|
|
|
def run_cgi(self):
|
|
"""Execute a CGI script."""
|
|
dir, rest = self.cgi_info
|
|
i = string.rfind(rest, '?')
|
|
if i >= 0:
|
|
rest, query = rest[:i], rest[i+1:]
|
|
else:
|
|
query = ''
|
|
i = string.find(rest, '/')
|
|
if i >= 0:
|
|
script, rest = rest[:i], rest[i:]
|
|
else:
|
|
script, rest = rest, ''
|
|
scriptname = dir + '/' + script
|
|
scriptfile = self.translate_path(scriptname)
|
|
if not os.path.exists(scriptfile):
|
|
self.send_error(404, "No such CGI script (%s)" % `scriptname`)
|
|
return
|
|
if not os.path.isfile(scriptfile):
|
|
self.send_error(403, "CGI script is not a plain file (%s)" %
|
|
`scriptname`)
|
|
return
|
|
if not executable(scriptfile):
|
|
self.send_error(403, "CGI script is not executable (%s)" %
|
|
`scriptname`)
|
|
return
|
|
nobody = nobody_uid()
|
|
self.send_response(200, "Script output follows")
|
|
self.wfile.flush() # Always flush before forking
|
|
pid = os.fork()
|
|
if pid != 0:
|
|
# Parent
|
|
pid, sts = os.waitpid(pid, 0)
|
|
if sts:
|
|
self.log_error("CGI script exit status x%x" % sts)
|
|
return
|
|
# Child
|
|
try:
|
|
# Reference: http://hoohoo.ncsa.uiuc.edu/cgi/env.html
|
|
# XXX Much of the following could be prepared ahead of time!
|
|
env = {}
|
|
env['SERVER_SOFTWARE'] = self.version_string()
|
|
env['SERVER_NAME'] = self.server.server_name
|
|
env['GATEWAY_INTERFACE'] = 'CGI/1.1'
|
|
env['SERVER_PROTOCOL'] = self.protocol_version
|
|
env['SERVER_PORT'] = str(self.server.server_port)
|
|
env['REQUEST_METHOD'] = self.command
|
|
uqrest = urllib.unquote(rest)
|
|
env['PATH_INFO'] = uqrest
|
|
env['PATH_TRANSLATED'] = self.translate_path(uqrest)
|
|
env['SCRIPT_NAME'] = scriptname
|
|
if query:
|
|
env['QUERY_STRING'] = query
|
|
host = self.address_string()
|
|
if host != self.client_address[0]:
|
|
env['REMOTE_HOST'] = host
|
|
env['REMOTE_ADDR'] = self.client_address[0]
|
|
# AUTH_TYPE
|
|
# REMOTE_USER
|
|
# REMOTE_IDENT
|
|
if self.headers.typeheader is None:
|
|
env['CONTENT_TYPE'] = self.headers.type
|
|
else:
|
|
env['CONTENT_TYPE'] = self.headers.typeheader
|
|
length = self.headers.getheader('content-length')
|
|
if length:
|
|
env['CONTENT_LENGTH'] = length
|
|
accept = []
|
|
for line in self.headers.getallmatchingheaders('accept'):
|
|
if line[:1] in string.whitespace:
|
|
accept.append(string.strip(line))
|
|
else:
|
|
accept = accept + string.split(line[7:], ',')
|
|
env['HTTP_ACCEPT'] = string.joinfields(accept, ',')
|
|
ua = self.headers.getheader('user-agent')
|
|
if ua:
|
|
env['HTTP_USER_AGENT'] = ua
|
|
# XXX Other HTTP_* headers
|
|
decoded_query = string.replace(query, '+', ' ')
|
|
try:
|
|
os.setuid(nobody)
|
|
except os.error:
|
|
pass
|
|
os.dup2(self.rfile.fileno(), 0)
|
|
os.dup2(self.wfile.fileno(), 1)
|
|
print scriptfile, script, decoded_query
|
|
os.execve(scriptfile,
|
|
[script, decoded_query],
|
|
env)
|
|
except:
|
|
self.server.handle_error(self.request, self.client_address)
|
|
os._exit(127)
|
|
|
|
|
|
nobody = None
|
|
|
|
def nobody_uid():
|
|
"""Internal routine to get nobody's uid"""
|
|
global nobody
|
|
if nobody:
|
|
return nobody
|
|
import pwd
|
|
try:
|
|
nobody = pwd.getpwnam('nobody')[2]
|
|
except pwd.error:
|
|
nobody = 1 + max(map(lambda x: x[2], pwd.getpwall()))
|
|
return nobody
|
|
|
|
|
|
def executable(path):
|
|
"""Test for executable file."""
|
|
try:
|
|
st = os.stat(path)
|
|
except os.error:
|
|
return 0
|
|
return st[0] & 0111 != 0
|
|
|
|
|
|
def test(HandlerClass = CGIHTTPRequestHandler,
|
|
ServerClass = BaseHTTPServer.HTTPServer):
|
|
SimpleHTTPServer.test(HandlerClass, ServerClass)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
test()
|