common.py
186 lines
| 6.2 KiB
| text/x-python
|
PythonLexer
|
r2391 | # hgweb/common.py - Utility functions needed by hgweb_mod and hgwebdir_mod | ||
|
r2356 | # | ||
# Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net> | ||||
|
r2859 | # Copyright 2005, 2006 Matt Mackall <mpm@selenic.com> | ||
|
r2356 | # | ||
|
r8225 | # This software may be used and distributed according to the terms of the | ||
|
r10263 | # GNU General Public License version 2 or any later version. | ||
|
r2356 | |||
|
r5561 | import errno, mimetypes, os | ||
|
r5993 | HTTP_OK = 200 | ||
|
r12183 | HTTP_NOT_MODIFIED = 304 | ||
|
r5993 | HTTP_BAD_REQUEST = 400 | ||
|
r6926 | HTTP_UNAUTHORIZED = 401 | ||
|
r7029 | HTTP_FORBIDDEN = 403 | ||
|
r5993 | HTTP_NOT_FOUND = 404 | ||
|
r6926 | HTTP_METHOD_NOT_ALLOWED = 405 | ||
|
r5993 | HTTP_SERVER_ERROR = 500 | ||
|
r9910 | |||
def checkauthz(hgweb, req, op): | ||||
'''Check permission for operation based on request data (including | ||||
authentication info). Return if op allowed, else raise an ErrorResponse | ||||
exception.''' | ||||
user = req.env.get('REMOTE_USER') | ||||
deny_read = hgweb.configlist('web', 'deny_read') | ||||
if deny_read and (not user or deny_read == ['*'] or user in deny_read): | ||||
raise ErrorResponse(HTTP_UNAUTHORIZED, 'read not authorized') | ||||
allow_read = hgweb.configlist('web', 'allow_read') | ||||
result = (not allow_read) or (allow_read == ['*']) | ||||
if not (result or user in allow_read): | ||||
raise ErrorResponse(HTTP_UNAUTHORIZED, 'read not authorized') | ||||
if op == 'pull' and not hgweb.allowpull: | ||||
raise ErrorResponse(HTTP_UNAUTHORIZED, 'pull not authorized') | ||||
elif op == 'pull' or op is None: # op is None for interface requests | ||||
return | ||||
# enforce that you can only push using POST requests | ||||
if req.env['REQUEST_METHOD'] != 'POST': | ||||
msg = 'push requires POST request' | ||||
raise ErrorResponse(HTTP_METHOD_NOT_ALLOWED, msg) | ||||
# require ssl by default for pushing, auth info cannot be sniffed | ||||
# and replayed | ||||
scheme = req.env.get('wsgi.url_scheme') | ||||
if hgweb.configbool('web', 'push_ssl', True) and scheme != 'https': | ||||
|
r17456 | raise ErrorResponse(HTTP_FORBIDDEN, 'ssl required') | ||
|
r9910 | |||
deny = hgweb.configlist('web', 'deny_push') | ||||
if deny and (not user or deny == ['*'] or user in deny): | ||||
raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized') | ||||
allow = hgweb.configlist('web', 'allow_push') | ||||
result = allow and (allow == ['*'] or user in allow) | ||||
if not result: | ||||
raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized') | ||||
|
r14058 | # Hooks for hgweb permission checks; extensions can add hooks here. | ||
# Each hook is invoked like this: hook(hgweb, request, operation), | ||||
# where operation is either read, pull or push. Hooks should either | ||||
# raise an ErrorResponse exception, or just return. | ||||
# | ||||
# It is possible to do both authentication and authorization through | ||||
# this. | ||||
permhooks = [checkauthz] | ||||
|
r9910 | |||
|
r5561 | class ErrorResponse(Exception): | ||
|
r7741 | def __init__(self, code, message=None, headers=[]): | ||
|
r13444 | if message is None: | ||
message = _statusmessage(code) | ||||
|
r13640 | Exception.__init__(self) | ||
|
r5561 | self.code = code | ||
|
r13444 | self.message = message | ||
|
r7741 | self.headers = headers | ||
|
r13599 | def __str__(self): | ||
return self.message | ||||
|
r5563 | |||
|
r13570 | class continuereader(object): | ||
def __init__(self, f, write): | ||||
self.f = f | ||||
self._write = write | ||||
self.continued = False | ||||
def read(self, amt=-1): | ||||
if not self.continued: | ||||
self.continued = True | ||||
self._write('HTTP/1.1 100 Continue\r\n\r\n') | ||||
return self.f.read(amt) | ||||
def __getattr__(self, attr): | ||||
if attr in ('close', 'readline', 'readlines', '__iter__'): | ||||
return getattr(self.f, attr) | ||||
|
r16687 | raise AttributeError | ||
|
r5563 | |||
def _statusmessage(code): | ||||
from BaseHTTPServer import BaseHTTPRequestHandler | ||||
responses = BaseHTTPRequestHandler.responses | ||||
return responses.get(code, ('Error', 'Unknown error'))[0] | ||||
|
r5760 | |||
|
r9694 | def statusmessage(code, message=None): | ||
return '%d %s' % (code, message or _statusmessage(code)) | ||||
|
r2356 | |||
|
r13958 | def get_stat(spath): | ||
"""stat changelog if it exists, spath otherwise""" | ||||
|
r10078 | cl_path = os.path.join(spath, "00changelog.i") | ||
|
r3853 | if os.path.exists(cl_path): | ||
|
r13958 | return os.stat(cl_path) | ||
|
r2356 | else: | ||
|
r13958 | return os.stat(spath) | ||
def get_mtime(spath): | ||||
return get_stat(spath).st_mtime | ||||
|
r2356 | |||
|
r2514 | def staticfile(directory, fname, req): | ||
|
r5930 | """return a file inside directory with guessed Content-Type header | ||
|
r2356 | |||
fname always uses '/' as directory separator and isn't allowed to | ||||
contain unusual path components. | ||||
|
r5930 | Content-Type is guessed using the mimetypes module. | ||
|
r2356 | Return an empty string if fname is illegal or file not found. | ||
""" | ||||
parts = fname.split('/') | ||||
for part in parts: | ||||
if (part in ('', os.curdir, os.pardir) or | ||||
os.sep in part or os.altsep is not None and os.altsep in part): | ||||
return "" | ||||
|
r7288 | fpath = os.path.join(*parts) | ||
if isinstance(directory, str): | ||||
directory = [directory] | ||||
for d in directory: | ||||
path = os.path.join(d, fpath) | ||||
if os.path.exists(path): | ||||
break | ||||
|
r2356 | try: | ||
os.stat(path) | ||||
ct = mimetypes.guess_type(path)[0] or "text/plain" | ||||
|
r5993 | req.respond(HTTP_OK, ct, length = os.path.getsize(path)) | ||
|
r13400 | fp = open(path, 'rb') | ||
data = fp.read() | ||||
fp.close() | ||||
return data | ||||
|
r5561 | except TypeError: | ||
|
r8761 | raise ErrorResponse(HTTP_SERVER_ERROR, 'illegal filename') | ||
|
r5561 | except OSError, err: | ||
if err.errno == errno.ENOENT: | ||||
|
r5993 | raise ErrorResponse(HTTP_NOT_FOUND) | ||
|
r5561 | else: | ||
|
r5993 | raise ErrorResponse(HTTP_SERVER_ERROR, err.strerror) | ||
|
r3276 | |||
|
r4462 | def paritygen(stripecount, offset=0): | ||
"""count parity of horizontal stripes for easier reading""" | ||||
if stripecount and offset: | ||||
# account for offset, e.g. due to building the list in reverse | ||||
count = (stripecount + offset) % stripecount | ||||
parity = (stripecount + offset) / stripecount & 1 | ||||
else: | ||||
count = 0 | ||||
parity = 0 | ||||
while True: | ||||
yield parity | ||||
count += 1 | ||||
if stripecount and count >= stripecount: | ||||
parity = 1 - parity | ||||
count = 0 | ||||
|
r5779 | def get_contact(config): | ||
"""Return repo contact information or empty string. | ||||
web.contact is the primary source, but if that is not set, try | ||||
ui.username or $EMAIL as a fallback to display something useful. | ||||
""" | ||||
return (config("web", "contact") or | ||||
config("ui", "username") or | ||||
os.environ.get("EMAIL") or "") | ||||
|
r12183 | |||
def caching(web, req): | ||||
tag = str(web.mtime) | ||||
if req.env.get('HTTP_IF_NONE_MATCH') == tag: | ||||
raise ErrorResponse(HTTP_NOT_MODIFIED) | ||||
req.headers.append(('ETag', tag)) | ||||