| # (c) 2005 Ian Bicking and contributors; written for Paste (http://pythonpaste.org) |
| # Licensed under the MIT license: http://www.opensource.org/licenses/mit-license.php |
| |
| """ |
| Application that runs a CGI script. |
| """ |
| import os |
| import sys |
| import subprocess |
| from six.moves.urllib.parse import quote |
| try: |
| import select |
| except ImportError: |
| select = None |
| import six |
| |
| from paste.util import converters |
| |
| __all__ = ['CGIError', 'CGIApplication'] |
| |
| class CGIError(Exception): |
| """ |
| Raised when the CGI script can't be found or doesn't |
| act like a proper CGI script. |
| """ |
| |
| class CGIApplication(object): |
| |
| """ |
| This object acts as a proxy to a CGI application. You pass in the |
| script path (``script``), an optional path to search for the |
| script (if the name isn't absolute) (``path``). If you don't give |
| a path, then ``$PATH`` will be used. |
| """ |
| |
| def __init__(self, |
| global_conf, |
| script, |
| path=None, |
| include_os_environ=True, |
| query_string=None): |
| if global_conf: |
| raise NotImplemented( |
| "global_conf is no longer supported for CGIApplication " |
| "(use make_cgi_application); please pass None instead") |
| self.script_filename = script |
| if path is None: |
| path = os.environ.get('PATH', '').split(':') |
| self.path = path |
| if '?' in script: |
| assert query_string is None, ( |
| "You cannot have '?' in your script name (%r) and also " |
| "give a query_string (%r)" % (script, query_string)) |
| script, query_string = script.split('?', 1) |
| if os.path.abspath(script) != script: |
| # relative path |
| for path_dir in self.path: |
| if os.path.exists(os.path.join(path_dir, script)): |
| self.script = os.path.join(path_dir, script) |
| break |
| else: |
| raise CGIError( |
| "Script %r not found in path %r" |
| % (script, self.path)) |
| else: |
| self.script = script |
| self.include_os_environ = include_os_environ |
| self.query_string = query_string |
| |
| def __call__(self, environ, start_response): |
| if 'REQUEST_URI' not in environ: |
| environ['REQUEST_URI'] = ( |
| quote(environ.get('SCRIPT_NAME', '')) |
| + quote(environ.get('PATH_INFO', ''))) |
| if self.include_os_environ: |
| cgi_environ = os.environ.copy() |
| else: |
| cgi_environ = {} |
| for name in environ: |
| # Should unicode values be encoded? |
| if (name.upper() == name |
| and isinstance(environ[name], str)): |
| cgi_environ[name] = environ[name] |
| if self.query_string is not None: |
| old = cgi_environ.get('QUERY_STRING', '') |
| if old: |
| old += '&' |
| cgi_environ['QUERY_STRING'] = old + self.query_string |
| cgi_environ['SCRIPT_FILENAME'] = self.script |
| proc = subprocess.Popen( |
| [self.script], |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| env=cgi_environ, |
| cwd=os.path.dirname(self.script), |
| ) |
| writer = CGIWriter(environ, start_response) |
| if select and sys.platform != 'win32': |
| proc_communicate( |
| proc, |
| stdin=StdinReader.from_environ(environ), |
| stdout=writer, |
| stderr=environ['wsgi.errors']) |
| else: |
| stdout, stderr = proc.communicate(StdinReader.from_environ(environ).read()) |
| if stderr: |
| environ['wsgi.errors'].write(stderr) |
| writer.write(stdout) |
| if not writer.headers_finished: |
| start_response(writer.status, writer.headers) |
| return [] |
| |
| class CGIWriter(object): |
| |
| def __init__(self, environ, start_response): |
| self.environ = environ |
| self.start_response = start_response |
| self.status = '200 OK' |
| self.headers = [] |
| self.headers_finished = False |
| self.writer = None |
| self.buffer = b'' |
| |
| def write(self, data): |
| if self.headers_finished: |
| self.writer(data) |
| return |
| self.buffer += data |
| while b'\n' in self.buffer: |
| if b'\r\n' in self.buffer and self.buffer.find(b'\r\n') < self.buffer.find(b'\n'): |
| line1, self.buffer = self.buffer.split(b'\r\n', 1) |
| else: |
| line1, self.buffer = self.buffer.split(b'\n', 1) |
| if not line1: |
| self.headers_finished = True |
| self.writer = self.start_response( |
| self.status, self.headers) |
| self.writer(self.buffer) |
| del self.buffer |
| del self.headers |
| del self.status |
| break |
| elif b':' not in line1: |
| raise CGIError( |
| "Bad header line: %r" % line1) |
| else: |
| name, value = line1.split(b':', 1) |
| value = value.lstrip() |
| name = name.strip() |
| if six.PY3: |
| name = name.decode('utf8') |
| value = value.decode('utf8') |
| if name.lower() == 'status': |
| if ' ' not in value: |
| # WSGI requires this space, sometimes CGI scripts don't set it: |
| value = '%s General' % value |
| self.status = value |
| else: |
| self.headers.append((name, value)) |
| |
| class StdinReader(object): |
| |
| def __init__(self, stdin, content_length): |
| self.stdin = stdin |
| self.content_length = content_length |
| |
| @classmethod |
| def from_environ(cls, environ): |
| length = environ.get('CONTENT_LENGTH') |
| if length: |
| length = int(length) |
| else: |
| length = 0 |
| return cls(environ['wsgi.input'], length) |
| |
| def read(self, size=None): |
| if not self.content_length: |
| return b'' |
| if size is None: |
| text = self.stdin.read(self.content_length) |
| else: |
| text = self.stdin.read(min(self.content_length, size)) |
| self.content_length -= len(text) |
| return text |
| |
| def proc_communicate(proc, stdin=None, stdout=None, stderr=None): |
| """ |
| Run the given process, piping input/output/errors to the given |
| file-like objects (which need not be actual file objects, unlike |
| the arguments passed to Popen). Wait for process to terminate. |
| |
| Note: this is taken from the posix version of |
| subprocess.Popen.communicate, but made more general through the |
| use of file-like objects. |
| """ |
| read_set = [] |
| write_set = [] |
| input_buffer = b'' |
| trans_nl = proc.universal_newlines and hasattr(open, 'newlines') |
| |
| if proc.stdin: |
| # Flush stdio buffer. This might block, if the user has |
| # been writing to .stdin in an uncontrolled fashion. |
| proc.stdin.flush() |
| if input: |
| write_set.append(proc.stdin) |
| else: |
| proc.stdin.close() |
| else: |
| assert stdin is None |
| if proc.stdout: |
| read_set.append(proc.stdout) |
| else: |
| assert stdout is None |
| if proc.stderr: |
| read_set.append(proc.stderr) |
| else: |
| assert stderr is None |
| |
| while read_set or write_set: |
| rlist, wlist, xlist = select.select(read_set, write_set, []) |
| |
| if proc.stdin in wlist: |
| # When select has indicated that the file is writable, |
| # we can write up to PIPE_BUF bytes without risk |
| # blocking. POSIX defines PIPE_BUF >= 512 |
| next, input_buffer = input_buffer, b'' |
| next_len = 512-len(next) |
| if next_len: |
| next += stdin.read(next_len) |
| if not next: |
| proc.stdin.close() |
| write_set.remove(proc.stdin) |
| else: |
| bytes_written = os.write(proc.stdin.fileno(), next) |
| if bytes_written < len(next): |
| input_buffer = next[bytes_written:] |
| |
| if proc.stdout in rlist: |
| data = os.read(proc.stdout.fileno(), 1024) |
| if data == b"": |
| proc.stdout.close() |
| read_set.remove(proc.stdout) |
| if trans_nl: |
| data = proc._translate_newlines(data) |
| stdout.write(data) |
| |
| if proc.stderr in rlist: |
| data = os.read(proc.stderr.fileno(), 1024) |
| if data == b"": |
| proc.stderr.close() |
| read_set.remove(proc.stderr) |
| if trans_nl: |
| data = proc._translate_newlines(data) |
| stderr.write(data) |
| |
| try: |
| proc.wait() |
| except OSError as e: |
| if e.errno != 10: |
| raise |
| |
| def make_cgi_application(global_conf, script, path=None, include_os_environ=None, |
| query_string=None): |
| """ |
| Paste Deploy interface for :class:`CGIApplication` |
| |
| This object acts as a proxy to a CGI application. You pass in the |
| script path (``script``), an optional path to search for the |
| script (if the name isn't absolute) (``path``). If you don't give |
| a path, then ``$PATH`` will be used. |
| """ |
| if path is None: |
| path = global_conf.get('path') or global_conf.get('PATH') |
| include_os_environ = converters.asbool(include_os_environ) |
| return CGIApplication( |
| None, |
| script, path=path, include_os_environ=include_os_environ, |
| query_string=query_string) |
