1. Andrew Godwin
  2. heechee


heechee / heechee / webdav / __init__.py

import logging

    from hashlib import md5
except ImportError:
    from md5 import md5

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 heechee.webdav.report import ReportMixin
from heechee.webdav.commit import CommitMixin
from heechee.webdav.read import ReaderMixin
from heechee.webdav.storage import FileStorage
from heechee.webdav.caching.noop import NoopCache
from heechee.webdav.caching.disk import DiskCache

class DAVRequest(Request):
    Overridden request object which has an xml_body function.
    def raw_body(self):
        return self.input_stream.read(self.content_length)

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

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()
        self.storage = 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 _cache_key(self, request):
        Uses the request and the repo to make a cache key.
        return "%s:%s:%s:%s" % (request.method, self.repo.uuid, str(self.repo.mtime()), md5(request.path + ":" + request.raw_body).hexdigest())
    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)")
    parser.add_option("--cache", action='store_true', default=False,
        help="Turns on simple disk caching.")

    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",
    if options.cache:
        cache = DiskCache()
        #from heechee.webdav.caching.memcached import MemcachedCache
        #cache = MemcachedCache([""])
        cache = NoopCache()

    from werkzeug import run_simple
            repo = get_repository(repopath, options.repo_type),
            cache = cache,