root/browser.py

#! /usr/bin/env python
# encoding: UTF-8
# Copyright (C) 2007, Stefan Schwarzer
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import BaseHTTPServer
import cgi
import email.Utils
import httplib
import mimetypes
import os
import sys
import time
import urllib
import urlparse

# Websourcebrowser modules
import coding
import config
import converter
import pygmentsfinder
import session
import template
import urlpath


def actual_css_path():
    """
    Return the file system path of the CSS file to use.
    """
    py_path = os.path.abspath(sys.modules[__name__].__file__)
    # assume the CSS file is in the same directory as the Python file
    css_path = os.path.join(os.path.dirname(py_path), config.CSS_FILE_NAME)
    return css_path


class SourceBrowserHandler(BaseHTTPServer.BaseHTTPRequestHandler):
    #protocol_version = "HTTP/1.1"
    server_version = "Websourcebrowser 0.x"
    sys_version = ""

    #
    # deal with builtin commands
    #
    def handle_builtin(self, url):
        """
        Handle a special command which is built into Websourcebrowser.

        Currently, there are the following commands:

        - the CSS file name: return the CSS file of Websourcebrowser
          to the web client

        - raw_data: returns the raw data of a file to the web client
        """
        # determine builtin command/file; skip to string after the
        #  special directory
        builtin_url = url[2+len(config.SPECIAL_DIR):]
        parse_result = urlparse.urlparse(builtin_url)
        builtin = parse_result[2]
        query_mapping = cgi.parse_qs(parse_result[4])
        if builtin == config.CSS_FILE_NAME:
            self.emit_data(self.get_file(actual_css_path()),
                           content_type="text/css",
                           mtime=os.path.getmtime(actual_css_path()))
        elif builtin == "help":
            self.emit_data(template.HELP, title=u"Websourcebrowser help",
                           h1_class="Other")
        elif builtin == "source_placeholder":
            self.emit_data(template.SOURCE_PLACEHOLDER,
                           title=u"Source code placeholder",
                           h1_class="Other")
        elif builtin == "raw_data":
            url = query_mapping.get('url', ["/"])[0]
            # `urlpath.to_file_system` takes an encoded URL, so provide it
            url = urllib.quote(url)
            try:
                source_path = urlpath.to_file_system(config.root, url)
            except urlpath.NotUnderRoot:
                self.emit_data(http_status=httplib.NOT_FOUND)
            else:
                content_type = mimetypes.guess_type(source_path)[0]
                if content_type is None:
                    content_type = "application/octet-stream"
                self.emit_data(data=self.get_file(source_path, raw=True),
                               content_type=content_type)

    #
    # HTTP-specific processing
    #
    def log_message(self, *args, **kwargs):
        if config.logging:
            return BaseHTTPServer.BaseHTTPRequestHandler.log_message(
                   self, *args, **kwargs)

    def emit_data(self, data=u"", title=u"", content_type="text/html",
                   h1_class="Path", http_status=httplib.OK, mtime=None,
                   refresh_html=u"", use_header_and_footer=True):
        """
        Emit the `data` on the HTTP output stream with a content type
        of `content_type`. If `h1_class` is set, it's used to set the
        CSS class of the h1 heading. If the integer `http_status` is
        set, it's the HTTP status code of the response. The default is
        200 (OK). If `mtime` is set, it's taken as a floating point
        value like from `time.time()` to set a Last-Modified header.
        If `use_header_and_footer` is true (the default),
        prepend/append the header and footer templates from the
        `template` module.
        """
        self.send_response(http_status)
        if content_type.startswith("text/"):
            header_content_type = "%s; charset=%s" % (content_type,
                                                      coding.DEFAULT_ENCODING)
        else:
            header_content_type = content_type
        self.send_header("Content-type", header_content_type)
        if mtime:
            self.send_header("Last-Modified", email.Utils.formatdate(mtime))
        if content_type != "text/css":
            self.send_header("Cache-Control", "no-cache")
        self.end_headers()
        if content_type == "text/html":
            http_status_group = str(http_status)[:1]
            if http_status_group in ("4", "5"):
                h1_class = "Error"
                title = httplib.responses[http_status]
                data = self.responses[http_status][1]
            if use_header_and_footer:
                self.wfile.write(coding.encode(template.HEADER % {
                  'title': cgi.escape(title),
                  'project_title': cgi.escape(config.project_title),
                  'h1_class': h1_class,
                  'special_dir': config.SPECIAL_DIR,
                  'refresh_html': refresh_html}))
        if content_type.startswith("text/"):
            self.wfile.write(coding.encode(data))
        else:
            self.wfile.write(data)
        if content_type == "text/html" and use_header_and_footer:
            self.wfile.write(coding.encode(template.FOOTER))

    def _url_and_params(self, url):
        """
        Return a tuple, splitting the URL `url` into an URL without
        query string and a query mapping as returned by `cgi.parse_qs`.
        """
        params = session.default_session.get_from_url(url)
        parsed_url = list(urlparse.urlparse(url))
        parsed_url[4] = ""
        return urlparse.urlunparse(parsed_url), params

    def do_GET(self):
        """
        Handle HTTP GET request.
        """
        # reject forbidden clients
        if config.allowed_clients is not config.ALL_CLIENTS and \
          self.client_address[0] not in config.allowed_clients and \
          self.path != "/%s/%s" % (config.SPECIAL_DIR, config.CSS_FILE_NAME):
            self.emit_data(http_status=httplib.FORBIDDEN)
            return
        # handle builtin command
        if self.path.startswith("/%s/" % config.SPECIAL_DIR):
            self.handle_builtin(self.path)
            return
        # separate URL and query parameters
        url, params = self._url_and_params(self.path)
        try:
            path = urlpath.to_file_system(config.root, url)
        except urlpath.NotUnderRoot:
            # don't show items "above" the current directory; thereby
            #  avoiding information disclosure (double dot attack)
            self.emit_data(http_status=httplib.BAD_REQUEST)
        # defaults
        use_header_and_footer = True
        h1_class = "File"
        if not os.path.exists(path):
            self.emit_data(http_status=httplib.NOT_FOUND)
            return
        if os.path.isdir(path):
            # by default, use frames
            use_frames = (params.get('frames', "yes") == "yes")
            if use_frames:
                # emit the frameset
                # don't use frames in windows which are part of the frameset
                params['frames'] = "no"
                dir_url = session.default_session.add_to_url(url, params)
                html = template.FRAMESET % {
                  'project_title': cgi.escape(config.project_title),
                  'special_dir': config.SPECIAL_DIR,
                  'dir_url':
                    cgi.escape('%s#%s' % (dir_url, params.get('anchor', ""))),
                  'dir_target': config.DIR_WINDOW_TARGET,
                  'source_target': config.SOURCE_WINDOW_TARGET}
                use_header_and_footer = False
            else:
                # emit the frame
                html = converter.dir_to_html(path, params)
                h1_class = "Dir"
        elif self.is_image_path(path):
            html = converter.image_to_html(url)
        else:
            data = self.get_file(path, raw=True,
                                 size=self._binary_test_size)
            if self.is_binary(data):
                data = self.get_file(path, raw=True)
                html = converter.binary_to_html(data)
            else:
                text = self.get_file(path)
                html = converter.text_to_html(text, path)
        # assume title is UTF-8-encoded though we don't know for sure
        title = coding.decode(url[1:]) or u"."
        title = urllib.unquote(title)
        self.emit_data(html, title=title, mtime=os.path.getmtime(path),
                       h1_class=h1_class,
                       use_header_and_footer=use_header_and_footer)

    #
    # utility methods
    #
    def get_file(self, path, raw=False, size=None):
        """
        Read the file identified by `path` and return its contents
        decoded as unicode string unless `raw` is true (by default,
        `raw` is false). If `size` is given, that many
        bytes/characters are read; by default, the whole file is read.
        """
        mode = ['r', 'rb'][bool(raw)]
        f = open(path, mode)
        try:
            if size is None:
                data = f.read()
            else:
                data = f.read(size)
        finally:
            f.close()
        if raw:
            return data
        else:
            return coding.decode(data)

    def is_image_path(self, path):
        """
        Return `True` if the `path` presumably represents an image,
        else return `False`.
        """
        mime_type = mimetypes.guess_type(path)[0]
        if mime_type is None:
            return False
        return mime_type.split("/", 1)[0] == "image"

    # number of bytes to read to test for binary data
    _binary_test_size = 1024

    def is_binary(self, data):
        """
        Return `True`, if `data` assumedly represents binary data,
        not text data. Else return `False`.
        """
        if not data:
            return False
        # assume binary data if over 5 % control codes
        threshold = 0.05
        control_codes = [byte for byte in data
                              if ord(byte) < 32 and not byte in "\n\r\t"]
        return (len(control_codes) / float(len(data)) > threshold)


def startup_info():
    print "Trying to listen on host %s, port %d." % \
          (config.http_host, config.http_port)
    print ("Type http://%s:%d/ into the address field of your " \
           "webbrowser.") % (config.http_host, config.http_port)
    print "Access is allowed from these IP addresses:",
    if config.allowed_clients is config.ALL_CLIENTS:
        print "all (running as public server)."
    else:
        print "%s." % (", ".join(sorted(config.allowed_clients)))
    if config.invalid_clients:
        print "Ignored invalid client addresses: %s." % \
              ", ".join(config.invalid_clients)
    if pygmentsfinder.found_pygments:
        print "Using Pygments library for syntax highlighting."
    else:
        print "Pygments library not found - no syntax highlighting possible."
        if config.line_numbers:
            print "Pygments library not found - omitting line numbers."
    if 'win' in sys.platform:
        exit_key = "Ctrl-Break"
    else:
        exit_key = "Ctrl-C"
    print "Press %s to exit." % exit_key


def _main():
    config.set_from_environment()
    config.set_from_args()
    startup_info()
    server_address = (config.http_host, config.http_port)
    httpd = BaseHTTPServer.HTTPServer(server_address, SourceBrowserHandler)
    httpd.serve_forever()

def main():
    try:
        _main()
    except KeyboardInterrupt:
        print "Aborted by keyboard interrupt from user."


if __name__ == '__main__':
    main()