159 lines
5.3 KiB
Python
159 lines
5.3 KiB
Python
# Copyright (C) 2010-2024 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
|
|
from rhodecode.lib.middleware.utils import get_path_info
|
|
|
|
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",
|
|
)
|
|
)
|
|
|
|
_SKIP_PATTERN = frozenset(("/_admin/gists/",))
|
|
|
|
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)
|
|
path_info = get_path_info(environ)
|
|
token_expected = path_info not in self._PATHS_WITHOUT_TOKEN
|
|
allowed = True
|
|
for pattern in self._SKIP_PATTERN:
|
|
if path_info.startswith(pattern):
|
|
allowed = False
|
|
break
|
|
|
|
if (
|
|
environ["REQUEST_METHOD"] == "POST"
|
|
and token_expected
|
|
and allowed
|
|
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"], 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
|
|
path_info = get_path_info(environ)
|
|
if path_info not in self._skip_urls and origin and not _equivalent_urls(origin, self._expected_origin):
|
|
log.warning("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)
|