[OpenSPARC-T2-SAM] / sam-t2 / devtools / v9 / lib / python2.4 / CGIHTTPServer.py

"""CGI-savvy HTTP Server.

This module builds on SimpleHTTPServer by implementing GET and POST
requests to cgi-bin scripts.

If the os.fork() function is not present (e.g. on Windows),
os.popen2() is used as a fallback, with slightly altered semantics; if
that function is not present either (e.g. on Macintosh), only Python
scripts are supported, and they are executed by the current process.

In all cases, the implementation is intentionally naive -- all
requests are executed sychronously.

SECURITY WARNING: DON'T USE THIS CODE UNLESS YOU ARE INSIDE A FIREWALL
-- it may execute arbitrary Python code or external programs.

"""


__version__ = "0.4"

__all__ = ["CGIHTTPRequestHandler"]

import os
import sys
import urllib
import BaseHTTPServer
import SimpleHTTPServer
import select


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.

    """

    # Determine platform specifics
    have_fork = hasattr(os, 'fork')
    have_popen2 = hasattr(os, 'popen2')
    have_popen3 = hasattr(os, 'popen3')

    # Make rfile unbuffered -- we need to read one line and then pass
    # the rest to a subprocess, so we can't use buffered input.
    rbufsize = 0

    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 self.path corresponds to a CGI script.

        Return a tuple (dir, rest) if self.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 True
        return False

    cgi_directories = ['/cgi-bin', '/htbin']

    def is_executable(self, path):
        """Test whether argument path is an executable file."""
        return executable(path)

    def is_python(self, path):
        """Test whether argument path is a Python script."""
        head, tail = os.path.splitext(path)
        return tail.lower() in (".py", ".pyw")

    def run_cgi(self):
        """Execute a CGI script."""
        dir, rest = self.cgi_info
        i = rest.rfind('?')
        if i >= 0:
            rest, query = rest[:i], rest[i+1:]
        else:
            query = ''
        i = rest.find('/')
        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 (%r)" % scriptname)
            return
        if not os.path.isfile(scriptfile):
            self.send_error(403, "CGI script is not a plain file (%r)" %
                            scriptname)
            return
        ispy = self.is_python(scriptname)
        if not ispy:
            if not (self.have_fork or self.have_popen2 or self.have_popen3):
                self.send_error(403, "CGI script is not a Python script (%r)" %
                                scriptname)
                return
            if not self.is_executable(scriptfile):
                self.send_error(403, "CGI script is not executable (%r)" %
                                scriptname)
                return

        # 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]
        authorization = self.headers.getheader("authorization")
        if authorization:
            authorization = authorization.split()
            if len(authorization) == 2:
                import base64, binascii
                env['AUTH_TYPE'] = authorization[0]
                if authorization[0].lower() == "basic":
                    try:
                        authorization = base64.decodestring(authorization[1])
                    except binascii.Error:
                        pass
                    else:
                        authorization = authorization.split(':')
                        if len(authorization) == 2:
                            env['REMOTE_USER'] = authorization[0]
        # XXX 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 "\t\n\r ":
                accept.append(line.strip())
            else:
                accept = accept + line[7:].split(',')
        env['HTTP_ACCEPT'] = ','.join(accept)
        ua = self.headers.getheader('user-agent')
        if ua:
            env['HTTP_USER_AGENT'] = ua
        co = filter(None, self.headers.getheaders('cookie'))
        if co:
            env['HTTP_COOKIE'] = ', '.join(co)
        # XXX Other HTTP_* headers
        # Since we're setting the env in the parent, provide empty
        # values to override previously set values
        for k in ('QUERY_STRING', 'REMOTE_HOST', 'CONTENT_LENGTH',
                  'HTTP_USER_AGENT', 'HTTP_COOKIE'):
            env.setdefault(k, "")
        os.environ.update(env)

        self.send_response(200, "Script output follows")

        decoded_query = query.replace('+', ' ')

        if self.have_fork:
            # Unix -- fork as we should
            args = [script]
            if '=' not in decoded_query:
                args.append(decoded_query)
            nobody = nobody_uid()
            self.wfile.flush() # Always flush before forking
            pid = os.fork()
            if pid != 0:
                # Parent
                pid, sts = os.waitpid(pid, 0)
                # throw away additional data [see bug #427345]
                while select.select([self.rfile], [], [], 0)[0]:
                    if not self.rfile.read(1):
                        break
                if sts:
                    self.log_error("CGI script exit status %#x", sts)
                return
            # Child
            try:
                try:
                    os.setuid(nobody)
                except os.error:
                    pass
                os.dup2(self.rfile.fileno(), 0)
                os.dup2(self.wfile.fileno(), 1)
                os.execve(scriptfile, args, os.environ)
            except:
                self.server.handle_error(self.request, self.client_address)
                os._exit(127)

        elif self.have_popen2 or self.have_popen3:
            # Windows -- use popen2 or popen3 to create a subprocess
            import shutil
            if self.have_popen3:
                popenx = os.popen3
            else:
                popenx = os.popen2
            cmdline = scriptfile
            if self.is_python(scriptfile):
                interp = sys.executable
                if interp.lower().endswith("w.exe"):
                    # On Windows, use python.exe, not pythonw.exe
                    interp = interp[:-5] + interp[-4:]
                cmdline = "%s -u %s" % (interp, cmdline)
            if '=' not in query and '"' not in query:
                cmdline = '%s "%s"' % (cmdline, query)
            self.log_message("command: %s", cmdline)
            try:
                nbytes = int(length)
            except (TypeError, ValueError):
                nbytes = 0
            files = popenx(cmdline, 'b')
            fi = files[0]
            fo = files[1]
            if self.have_popen3:
                fe = files[2]
            if self.command.lower() == "post" and nbytes > 0:
                data = self.rfile.read(nbytes)
                fi.write(data)
            # throw away additional data [see bug #427345]
            while select.select([self.rfile._sock], [], [], 0)[0]:
                if not self.rfile._sock.recv(1):
                    break
            fi.close()
            shutil.copyfileobj(fo, self.wfile)
            if self.have_popen3:
                errors = fe.read()
                fe.close()
                if errors:
                    self.log_error('%s', errors)
            sts = fo.close()
            if sts:
                self.log_error("CGI script exit status %#x", sts)
            else:
                self.log_message("CGI script exited OK")

        else:
            # Other O.S. -- execute script in this process
            save_argv = sys.argv
            save_stdin = sys.stdin
            save_stdout = sys.stdout
            save_stderr = sys.stderr
            try:
                save_cwd = os.getcwd()
                try:
                    sys.argv = [scriptfile]
                    if '=' not in decoded_query:
                        sys.argv.append(decoded_query)
                    sys.stdout = self.wfile
                    sys.stdin = self.rfile
                    execfile(scriptfile, {"__name__": "__main__"})
                finally:
                    sys.argv = save_argv
                    sys.stdin = save_stdin
                    sys.stdout = save_stdout
                    sys.stderr = save_stderr
                    os.chdir(save_cwd)
            except SystemExit, sts:
                self.log_error("CGI script exit status %s", str(sts))
            else:
                self.log_message("CGI script exited OK")


nobody = None

def nobody_uid():
    """Internal routine to get nobody's uid"""
    global nobody
    if nobody:
        return nobody
    try:
        import pwd
    except ImportError:
        return -1
    try:
        nobody = pwd.getpwnam('nobody')[2]
    except KeyError:
        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 False
    return st.st_mode & 0111 != 0


def test(HandlerClass = CGIHTTPRequestHandler,
         ServerClass = BaseHTTPServer.HTTPServer):
    SimpleHTTPServer.test(HandlerClass, ServerClass)


if __name__ == '__main__':
    test()
Commit	Line	Data
920dae64 AT	1	"""CGI-savvy HTTP Server.
	2
	3	This module builds on SimpleHTTPServer by implementing GET and POST
	4	requests to cgi-bin scripts.
	5
	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.
	10
	11	In all cases, the implementation is intentionally naive -- all
	12	requests are executed sychronously.
	13
	14	SECURITY WARNING: DON'T USE THIS CODE UNLESS YOU ARE INSIDE A FIREWALL
	15	-- it may execute arbitrary Python code or external programs.
	16
	17	"""
	18
	19
	20	__version__ = "0.4"
	21
	22	__all__ = ["CGIHTTPRequestHandler"]
	23
	24	import os
	25	import sys
	26	import urllib
	27	import BaseHTTPServer
	28	import SimpleHTTPServer
	29	import select
	30
	31
	32	class CGIHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
	33
	34	"""Complete HTTP server with GET, HEAD and POST commands.
	35
	36	GET and HEAD also support running CGI scripts.
	37
	38	The POST command is only implemented for CGI scripts.
	39
	40	"""
	41
	42	# Determine platform specifics
	43	have_fork = hasattr(os, 'fork')
	44	have_popen2 = hasattr(os, 'popen2')
	45	have_popen3 = hasattr(os, 'popen3')
	46
	47	# Make rfile unbuffered -- we need to read one line and then pass
	48	# the rest to a subprocess, so we can't use buffered input.
	49	rbufsize = 0
	50
	51	def do_POST(self):
	52	"""Serve a POST request.
	53
	54	This is only implemented for CGI scripts.
	55
	56	"""
	57
	58	if self.is_cgi():
	59	self.run_cgi()
	60	else:
	61	self.send_error(501, "Can only POST to CGI scripts")
	62
	63	def send_head(self):
	64	"""Version of send_head that support CGI scripts"""
65	if self.is_cgi():
66	return self.run_cgi()
67	else:
68	return SimpleHTTPServer.SimpleHTTPRequestHandler.send_head(self)
69
70	def is_cgi(self):
71	"""Test whether self.path corresponds to a CGI script.
72
73	Return a tuple (dir, rest) if self.path requires running a
74	CGI script, None if not. Note that rest begins with a
75	slash if it is not empty.
76
77	The default implementation tests whether the path
78	begins with one of the strings in the list
79	self.cgi_directories (and the next character is a '/'
80	or the end of the string).
81
82	"""
83
84	path = self.path
85
86	for x in self.cgi_directories:
87	i = len(x)
88	if path[:i] == x and (not path[i:] or path[i] == '/'):
89	self.cgi_info = path[:i], path[i+1:]
90	return True
91	return False
92
93	cgi_directories = ['/cgi-bin', '/htbin']
94
95	def is_executable(self, path):
96	"""Test whether argument path is an executable file."""
97	return executable(path)
98
99	def is_python(self, path):
100	"""Test whether argument path is a Python script."""
101	head, tail = os.path.splitext(path)
102	return tail.lower() in (".py", ".pyw")
103
104	def run_cgi(self):
105	"""Execute a CGI script."""
106	dir, rest = self.cgi_info
107	i = rest.rfind('?')
108	if i >= 0:
109	rest, query = rest[:i], rest[i+1:]
110	else:
111	query = ''
112	i = rest.find('/')
113	if i >= 0:
114	script, rest = rest[:i], rest[i:]
115	else:
116	script, rest = rest, ''
117	scriptname = dir + '/' + script
118	scriptfile = self.translate_path(scriptname)
119	if not os.path.exists(scriptfile):
120	self.send_error(404, "No such CGI script (%r)" % scriptname)
121	return
122	if not os.path.isfile(scriptfile):
123	self.send_error(403, "CGI script is not a plain file (%r)" %
124	scriptname)
125	return
126	ispy = self.is_python(scriptname)
127	if not ispy:
128	if not (self.have_fork or self.have_popen2 or self.have_popen3):
129	self.send_error(403, "CGI script is not a Python script (%r)" %
130	scriptname)
131	return
132	if not self.is_executable(scriptfile):
133	self.send_error(403, "CGI script is not executable (%r)" %
134	scriptname)
135	return
136
137	# Reference: http://hoohoo.ncsa.uiuc.edu/cgi/env.html
138	# XXX Much of the following could be prepared ahead of time!
139	env = {}
140	env['SERVER_SOFTWARE'] = self.version_string()
141	env['SERVER_NAME'] = self.server.server_name
142	env['GATEWAY_INTERFACE'] = 'CGI/1.1'
143	env['SERVER_PROTOCOL'] = self.protocol_version
144	env['SERVER_PORT'] = str(self.server.server_port)
145	env['REQUEST_METHOD'] = self.command
146	uqrest = urllib.unquote(rest)
147	env['PATH_INFO'] = uqrest
148	env['PATH_TRANSLATED'] = self.translate_path(uqrest)
149	env['SCRIPT_NAME'] = scriptname
150	if query:
151	env['QUERY_STRING'] = query
152	host = self.address_string()
153	if host != self.client_address[0]:
154	env['REMOTE_HOST'] = host
155	env['REMOTE_ADDR'] = self.client_address[0]
156	authorization = self.headers.getheader("authorization")
157	if authorization:
158	authorization = authorization.split()
159	if len(authorization) == 2:
160	import base64, binascii
161	env['AUTH_TYPE'] = authorization[0]
162	if authorization[0].lower() == "basic":
163	try:
164	authorization = base64.decodestring(authorization[1])
165	except binascii.Error:
166	pass
167	else:
168	authorization = authorization.split(':')
169	if len(authorization) == 2:
170	env['REMOTE_USER'] = authorization[0]
171	# XXX REMOTE_IDENT
172	if self.headers.typeheader is None:
173	env['CONTENT_TYPE'] = self.headers.type
174	else:
175	env['CONTENT_TYPE'] = self.headers.typeheader
176	length = self.headers.getheader('content-length')
177	if length:
178	env['CONTENT_LENGTH'] = length
179	accept = []
180	for line in self.headers.getallmatchingheaders('accept'):
181	if line[:1] in "\t\n\r ":
182	accept.append(line.strip())
183	else:
184	accept = accept + line[7:].split(',')
185	env['HTTP_ACCEPT'] = ','.join(accept)
186	ua = self.headers.getheader('user-agent')
187	if ua:
188	env['HTTP_USER_AGENT'] = ua
189	co = filter(None, self.headers.getheaders('cookie'))
190	if co:
191	env['HTTP_COOKIE'] = ', '.join(co)
192	# XXX Other HTTP_* headers
193	# Since we're setting the env in the parent, provide empty
194	# values to override previously set values
195	for k in ('QUERY_STRING', 'REMOTE_HOST', 'CONTENT_LENGTH',
196	'HTTP_USER_AGENT', 'HTTP_COOKIE'):
197	env.setdefault(k, "")
198	os.environ.update(env)
199
200	self.send_response(200, "Script output follows")
201
202	decoded_query = query.replace('+', ' ')
203
204	if self.have_fork:
205	# Unix -- fork as we should
206	args = [script]
207	if '=' not in decoded_query:
208	args.append(decoded_query)
209	nobody = nobody_uid()
210	self.wfile.flush() # Always flush before forking
211	pid = os.fork()
212	if pid != 0:
213	# Parent
214	pid, sts = os.waitpid(pid, 0)
215	# throw away additional data [see bug #427345]
216	while select.select([self.rfile], [], [], 0)[0]:
217	if not self.rfile.read(1):
218	break
219	if sts:
220	self.log_error("CGI script exit status %#x", sts)
221	return
222	# Child
223	try:
224	try:
225	os.setuid(nobody)
226	except os.error:
227	pass
228	os.dup2(self.rfile.fileno(), 0)
229	os.dup2(self.wfile.fileno(), 1)
230	os.execve(scriptfile, args, os.environ)
231	except:
232	self.server.handle_error(self.request, self.client_address)
233	os._exit(127)
234
235	elif self.have_popen2 or self.have_popen3:
236	# Windows -- use popen2 or popen3 to create a subprocess
237	import shutil
238	if self.have_popen3:
239	popenx = os.popen3
240	else:
241	popenx = os.popen2
242	cmdline = scriptfile
243	if self.is_python(scriptfile):
244	interp = sys.executable
245	if interp.lower().endswith("w.exe"):
246	# On Windows, use python.exe, not pythonw.exe
247	interp = interp[:-5] + interp[-4:]
248	cmdline = "%s -u %s" % (interp, cmdline)
249	if '=' not in query and '"' not in query:
250	cmdline = '%s "%s"' % (cmdline, query)
251	self.log_message("command: %s", cmdline)
252	try:
253	nbytes = int(length)
254	except (TypeError, ValueError):
255	nbytes = 0
256	files = popenx(cmdline, 'b')
257	fi = files[0]
258	fo = files[1]
259	if self.have_popen3:
260	fe = files[2]
261	if self.command.lower() == "post" and nbytes > 0:
262	data = self.rfile.read(nbytes)
263	fi.write(data)
264	# throw away additional data [see bug #427345]
265	while select.select([self.rfile._sock], [], [], 0)[0]:
266	if not self.rfile._sock.recv(1):
267	break
268	fi.close()
269	shutil.copyfileobj(fo, self.wfile)
270	if self.have_popen3:
271	errors = fe.read()
272	fe.close()
273	if errors:
274	self.log_error('%s', errors)
275	sts = fo.close()
276	if sts:
277	self.log_error("CGI script exit status %#x", sts)
278	else:
279	self.log_message("CGI script exited OK")
280
281	else:
282	# Other O.S. -- execute script in this process
283	save_argv = sys.argv
284	save_stdin = sys.stdin
285	save_stdout = sys.stdout
286	save_stderr = sys.stderr
287	try:
288	save_cwd = os.getcwd()
289	try:
290	sys.argv = [scriptfile]
291	if '=' not in decoded_query:
292	sys.argv.append(decoded_query)
293	sys.stdout = self.wfile
294	sys.stdin = self.rfile
295	execfile(scriptfile, {"__name__": "__main__"})
296	finally:
297	sys.argv = save_argv
298	sys.stdin = save_stdin
299	sys.stdout = save_stdout
300	sys.stderr = save_stderr
301	os.chdir(save_cwd)
302	except SystemExit, sts:
303	self.log_error("CGI script exit status %s", str(sts))
304	else:
305	self.log_message("CGI script exited OK")
306
307
308	nobody = None
309
310	def nobody_uid():
311	"""Internal routine to get nobody's uid"""
312	global nobody
313	if nobody:
314	return nobody
315	try:
316	import pwd
317	except ImportError:
318	return -1
319	try:
320	nobody = pwd.getpwnam('nobody')[2]
321	except KeyError:
322	nobody = 1 + max(map(lambda x: x[2], pwd.getpwall()))
323	return nobody
324
325
326	def executable(path):
327	"""Test for executable file."""
328	try:
329	st = os.stat(path)
330	except os.error:
331	return False
332	return st.st_mode & 0111 != 0
333
334
335	def test(HandlerClass = CGIHTTPRequestHandler,
336	ServerClass = BaseHTTPServer.HTTPServer):
337	SimpleHTTPServer.test(HandlerClass, ServerClass)
338
339
340	if __name__ == '__main__':
341	test()