mercurial/hgweb/common.py

43075:57875cf423c9

    Can be overridden by extensions to provide more complex authorization
    schemes.
    """
    return userlist == ['*'] or username in userlist

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.'''

    if allow_read and (not ismember(hgweb.repo.ui, user, 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

    # Allow LFS uploading via PUT requests
    if op == 'upload':
        if req.method != 'PUT':

        raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized')

    allow = hgweb.configlist('web', 'allow-push')
    if not (allow and ismember(hgweb.repo.ui, user, allow)):
        raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized')

# 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, push or upload. Hooks should either
# raise an ErrorResponse exception, or just return.

        if headers is None:
            headers = []
        self.headers = headers
        self.message = message

class continuereader(object):
    """File object wrapper to handle HTTP 100-continue.

    This is used by servers so they automatically handle Expect: 100-continue
    request headers. On first read of the request body, the 100 Continue
    response is sent. This should trigger the client into actually sending
    the request body.
    """
    def __init__(self, f, write):
        self.f = f
        self._write = write
        self.continued = False

    def __getattr__(self, attr):
        if attr in ('close', 'readline', 'readlines', '__iter__'):
            return getattr(self.f, attr)
        raise AttributeError

def _statusmessage(code):
    responses = httpserver.basehttprequesthandler.responses
    return pycompat.bytesurl(
        responses.get(code, (r'Error', r'Unknown error'))[0])

def statusmessage(code, message=None):
    return '%d %s' % (code, message or _statusmessage(code))

def get_stat(spath, fn):
    """stat fn if it exists, spath otherwise"""
    cl_path = os.path.join(spath, fn)
    if os.path.exists(cl_path):
        return os.stat(cl_path)
    else:
        return os.stat(spath)

def get_mtime(spath):
    return get_stat(spath, "00changelog.i")[stat.ST_MTIME]

def ispathsafe(path):
    """Determine if a path is safe to use for filesystem access."""
    parts = path.split('/')
    for part in parts:
        if (part in ('', pycompat.oscurdir, pycompat.ospardir) or
            pycompat.ossep in part or
            pycompat.osaltsep is not None and pycompat.osaltsep in part):
            return False

    return True

def staticfile(directory, fname, res):
    """return a file inside directory with guessed Content-Type header

    fname always uses '/' as directory separator and isn't allowed to

        if os.path.exists(path):
            break
    try:
        os.stat(path)
        ct = pycompat.sysbytes(
            mimetypes.guess_type(pycompat.fsdecode(path))[0] or r"text/plain")
        with open(path, 'rb') as fh:
            data = fh.read()

        res.headers['Content-Type'] = ct
        res.setbodybytes(data)

        raise ErrorResponse(HTTP_SERVER_ERROR, 'illegal filename')
    except OSError as err:
        if err.errno == errno.ENOENT:
            raise ErrorResponse(HTTP_NOT_FOUND)
        else:
            raise ErrorResponse(HTTP_SERVER_ERROR,
                                encoding.strtolocal(err.strerror))

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 += 1
        if stripecount and count >= stripecount:
            parity = 1 - parity
            count = 0

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
            encoding.environ.get("EMAIL") or "")

def cspvalues(ui):
    """Obtain the Content-Security-Policy header and nonce value.

    Returns a 2-tuple of the CSP header value and the nonce value.

43076:2372284d9457

    Can be overridden by extensions to provide more complex authorization
    schemes.
    """
    return userlist == ['*'] or username in userlist


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.'''

    if allow_read and (not ismember(hgweb.repo.ui, user, 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

    # Allow LFS uploading via PUT requests
    if op == 'upload':
        if req.method != 'PUT':

        raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized')

    allow = hgweb.configlist('web', 'allow-push')
    if not (allow and ismember(hgweb.repo.ui, user, allow)):
        raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized')


# 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, push or upload. Hooks should either
# raise an ErrorResponse exception, or just return.

        if headers is None:
            headers = []
        self.headers = headers
        self.message = message


class continuereader(object):
    """File object wrapper to handle HTTP 100-continue.

    This is used by servers so they automatically handle Expect: 100-continue
    request headers. On first read of the request body, the 100 Continue
    response is sent. This should trigger the client into actually sending
    the request body.
    """

    def __init__(self, f, write):
        self.f = f
        self._write = write
        self.continued = False

    def __getattr__(self, attr):
        if attr in ('close', 'readline', 'readlines', '__iter__'):
            return getattr(self.f, attr)
        raise AttributeError


def _statusmessage(code):
    responses = httpserver.basehttprequesthandler.responses
    return pycompat.bytesurl(
        responses.get(code, (r'Error', r'Unknown error'))[0]
    )


def statusmessage(code, message=None):
    return '%d %s' % (code, message or _statusmessage(code))


def get_stat(spath, fn):
    """stat fn if it exists, spath otherwise"""
    cl_path = os.path.join(spath, fn)
    if os.path.exists(cl_path):
        return os.stat(cl_path)
    else:
        return os.stat(spath)


def get_mtime(spath):
    return get_stat(spath, "00changelog.i")[stat.ST_MTIME]


def ispathsafe(path):
    """Determine if a path is safe to use for filesystem access."""
    parts = path.split('/')
    for part in parts:
        if (
            part in ('', pycompat.oscurdir, pycompat.ospardir)
            or pycompat.ossep in part
            or pycompat.osaltsep is not None
            and pycompat.osaltsep in part
        ):
            return False

    return True


def staticfile(directory, fname, res):
    """return a file inside directory with guessed Content-Type header

    fname always uses '/' as directory separator and isn't allowed to

        if os.path.exists(path):
            break
    try:
        os.stat(path)
        ct = pycompat.sysbytes(
            mimetypes.guess_type(pycompat.fsdecode(path))[0] or r"text/plain"
        )
        with open(path, 'rb') as fh:
            data = fh.read()

        res.headers['Content-Type'] = ct
        res.setbodybytes(data)

        raise ErrorResponse(HTTP_SERVER_ERROR, 'illegal filename')
    except OSError as err:
        if err.errno == errno.ENOENT:
            raise ErrorResponse(HTTP_NOT_FOUND)
        else:
            raise ErrorResponse(
                HTTP_SERVER_ERROR, encoding.strtolocal(err.strerror)
            )


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 += 1
        if stripecount and count >= stripecount:
            parity = 1 - parity
            count = 0


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 encoding.environ.get("EMAIL")
        or ""
    )


def cspvalues(ui):
    """Obtain the Content-Security-Policy header and nonce value.

    Returns a 2-tuple of the CSP header value and the nonce value.