##// END OF EJS Templates
api: use consistent way to extract users, repos, repo groups and user groups by id or name....
api: use consistent way to extract users, repos, repo groups and user groups by id or name. - makes usage of Number vs String to differenciate if we pick objec ID or it's name this will allow easy fetching of objects by either id or it's name, including numeric string name - fixes #5230

File last commit:

r1271:47a44c03 default
r1530:1efcb4ee default
Show More
csrf.py
151 lines | 4.9 KiB | text/x-python | PythonLexer
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2017 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import logging
import textwrap
import routes.middleware
import urlobject
import webob
import webob.exc
import rhodecode.lib.auth
log = logging.getLogger(__name__)
class CSRFDetector(object):
"""
Middleware for preventing CSRF.
It checks that all requests are either GET or POST.
For POST requests, it logs the requests that do not have a CSRF token.
Eventually it will raise an error.
It special cases some endpoints as they do not really require a token.
Note: this middleware is only intended for testing.
"""
_PUT_DELETE_MESSAGE = textwrap.dedent('''
Do not call in tests app.delete or app.put, use instead
app.post(..., params={'_method': 'delete'}.
The reason is twofold. The first is because that's how the browser is
calling rhodecode and the second is because it allow us to detect
potential CSRF.''').strip()
_PATHS_WITHOUT_TOKEN = frozenset((
# The password is the token.
'/_admin/login',
# Captcha may be enabled.
'/_admin/password_reset',
# Captcha may be enabled.
'/_admin/password_reset_confirmation',
# Captcha may be enabled.
'/_admin/register',
# No change in state with this controller.
'/error/document',
))
def __init__(self, app):
self._app = app
def __call__(self, environ, start_response):
if environ['REQUEST_METHOD'].upper() not in ('GET', 'POST'):
raise Exception(self._PUT_DELETE_MESSAGE)
if (environ['REQUEST_METHOD'] == 'POST' and
environ['PATH_INFO'] not in self._PATHS_WITHOUT_TOKEN and
routes.middleware.is_form_post(environ)):
body = environ['wsgi.input']
if body.seekable():
pos = body.tell()
content = body.read()
body.seek(pos)
elif hasattr(body, 'peek'):
content = body.peek()
else:
raise Exception("Cannot check if the request has a CSRF token")
if rhodecode.lib.auth.csrf_token_key not in content:
raise Exception(
'%s to %s does not have a csrf_token %r' %
(environ['REQUEST_METHOD'], environ['PATH_INFO'], content))
return self._app(environ, start_response)
def _get_scheme_host_port(url):
url = urlobject.URLObject(url)
if '://' not in url:
return None, url, None
scheme = url.scheme or 'http'
port = url.port
if not port:
if scheme == 'http':
port = 80
elif scheme == 'https':
port = 443
host = url.netloc.without_port()
return scheme, host, port
def _equivalent_urls(url1, url2):
"""Check if both urls are equivalent."""
return _get_scheme_host_port(url1) == _get_scheme_host_port(url2)
class OriginChecker(object):
"""
Check whether the request has a valid Origin header.
See https://wiki.mozilla.org/Security/Origin for details.
"""
def __init__(self, app, expected_origin, skip_urls=None):
"""
:param expected_origin: the value we expect to see for the Origin
header.
:param skip_urls: list of urls for which we do not need to check the
Origin header.
"""
self._app = app
self._expected_origin = expected_origin
self._skip_urls = frozenset(skip_urls or [])
def __call__(self, environ, start_response):
origin_header = environ.get('HTTP_ORIGIN', '')
origin = origin_header.split(' ', 1)[0]
if origin == 'null':
origin = None
if (environ['PATH_INFO'] not in self._skip_urls and origin and
not _equivalent_urls(origin, self._expected_origin)):
log.warn(
'Invalid Origin header detected: got %s, expected %s',
origin_header, self._expected_origin)
return webob.exc.HTTPForbidden('Origin header mismatch')(
environ, start_response)
else:
return self._app(environ, start_response)