heechee / heechee / webdav /

Full commit

import logging

from lxml.etree import fromstring, tostring, Element, SubElement

from werkzeug import Request, Response, cached_property
from werkzeug.exceptions import HTTPException, MethodNotAllowed, NotFound, BadRequest, Forbidden

from heechee.constants import *
from heechee.exceptions import *
from heechee.repo import get_repository, File, Directory, tree_get
from heechee.webdav.exceptions import UnauthorizedAuthenticate
from heechee.webdav.props import PropsMixin
from import ReportMixin
from heechee.webdav.commit import CommitMixin
from import ReaderMixin
from import FileStorage
from heechee.webdav.caching.noop import NoopCache

class DAVRequest(Request):
    Overridden request object which has an xml_body function.

    def xml_body(self):
        # would be nicer to have fromstring( but that requires
        # an unreleased version of Werkzeug.
        return fromstring(

class RepositoryServer(PropsMixin, ReportMixin, CommitMixin, ReaderMixin):
    Main WSGI app for serving repositories. Only serves one; you'll need to use
    your own routing/creation logic and the Power Of WSGI to serve more.
    verbs = frozenset([
    write_verbs = frozenset([
    def __init__(self, repo, cache=None, storage=None):
        self.repo = repo
        self.cache = cache or NoopCache() = storage or FileStorage()
    def authorization_check(self, username, password, access_type):
        Takes the username, password and access type and returns if the user is
        allowed to access the repository. Username and password may be None.
        True means they're allowed to use the method.
        False means they're identified, and not allowed.
        None means their identification didn't match, and re-sends 401.
        access_type is one of "read" or "write".
        return True
    def __call__(self, request):
        "WSGI Entry point."
            # First check if they're even allowed to call that method ever.
            if request.method not in self.verbs:
                raise MethodNotAllowed(self.verbs)
            # Do some auth checking
            if request.method in self.write_verbs:
                self.access_check(request, "write")
                self.access_check(request, "read")
            return getattr(self, request.method)(request)
        except HTTPException, e:
            return e
    def access_check(self, request, accesss_type):
        # Get their status
        if request.authorization is None:
            allowed = self.authorization_check(None, None, accesss_type)
            allowed = self.authorization_check(
        # Act on that.
        if allowed is True:
        elif allowed is False:
            raise Forbidden("You are not allowed to call %s on this repository." % request.method)
            raise UnauthorizedAuthenticate("Please authorize to access this repository.")
    def OPTIONS(self, request):
        Look, Ma, we support ALL THESE methods. Really.
        # OPTIONS is also used for XML queries
        if request.content_length:
            # Make the response
            opr = Element(DAV_NS+"options-response")
            acs = SubElement(opr, DAV_NS+"activity-collection-set")
            acs.append(self._href(request.script_root + "/!svn/act/"))
            return Response(tostring(opr))
            return Response(headers = {
                "Allow": ",".join(self.verbs),
    def _href(self, target):
        "Single-call making of DAV:href tags."
        elem = Element(DAV_NS+"href")
        elem.text = target
        return elem

# As-script behaviour
if __name__ == "__main__":
    import optparse
    parser = optparse.OptionParser("%prog [options] REPOPATH")
    parser.add_option("-p", "--port", type="int", default=8080,
        help="Port on which the server should listen. Defaults to 8080.")
    parser.add_option("-t", "--repo-type", default=None,
        help="Repository type, 'hg' or 'git'. Autodetected by default.")
    parser.add_option("--debug", action='store_true', default=False,
        help="Display debug output (warning: there's lots)")

    options, args = parser.parse_args()
    if len(args) < 1:
        raise Exception('Not enough arguments; repository path expected')
    repopath = args[0]
    # Set up logging.
        format="%(asctime)s - %(levelname)8s - %(message)s",
        level=options.debug and logging.DEBUG or logging.INFO,
        datefmt="%Y-%m-%d %H:%M:%S",

    from werkzeug import run_simple
        RepositoryServer(get_repository(repopath, options.repo_type)),