Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.web.framework.middleware.remoteuser
"""
Middleware for handling $REMOTE_USER if use_remote_user is enabled.
"""
import logging
import socket
from galaxy.util import safe_str_cmp
log = logging.getLogger(__name__)
errorpage = """
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html lang="en">
<head>
<title>Galaxy</title>
<style type="text/css">
body {
min-width: 500px;
text-align: center;
}
.errormessage {
font: 75%% verdana, "Bitstream Vera Sans", geneva, arial, helvetica, helve, sans-serif;
padding: 10px;
margin: 100px auto;
min-height: 32px;
max-width: 500px;
border: 1px solid #AA6666;
background-color: #FFCCCC;
text-align: left;
}
</style>
</head>
<body>
<div class="errormessage">
<h4>%s</h4>
<p>%s</p>
</div>
</body>
</html>
"""
[docs]class RemoteUser:
[docs] def __init__(
self,
app,
maildomain=None,
display_servers=None,
admin_users=None,
single_user=None,
remote_user_header=None,
remote_user_secret_header=None,
normalize_remote_user_email=False,
):
self.app = app
self.maildomain = maildomain
self.display_servers = display_servers or []
self.admin_users = admin_users or []
self.remote_user_header = remote_user_header or "HTTP_REMOTE_USER"
self.single_user = single_user
self.config_secret_header = remote_user_secret_header
self.normalize_remote_user_email = normalize_remote_user_email
def __call__(self, environ, start_response):
# Allow display servers
if self.display_servers and "REMOTE_ADDR" in environ:
try:
host = socket.gethostbyaddr(environ["REMOTE_ADDR"])[0]
except (OSError, socket.herror, socket.gaierror, socket.timeout):
# in the event of a lookup failure, deny access
host = None
if host in self.display_servers:
environ[self.remote_user_header] = "remote_display_server@%s" % (self.maildomain or "example.org")
return self.app(environ, start_response)
if self.single_user:
assert self.remote_user_header not in environ
environ[self.remote_user_header] = self.single_user
if environ.get(self.remote_user_header, "").startswith("(null)"):
# Throw away garbage headers.
# Apache sets REMOTE_USER to the string '(null)' when using the
# Rewrite* method for passing REMOTE_USER and a user is not authenticated.
# Any other possible values need to go here as well.
log.debug(
"Discarding invalid remote user header %s:%s.",
self.remote_user_header,
environ.get(self.remote_user_header, None),
)
environ.pop(self.remote_user_header)
if self.remote_user_header in environ:
# process remote user with configuration options.
if self.normalize_remote_user_email:
environ[self.remote_user_header] = environ[self.remote_user_header].lower()
if self.maildomain and "@" not in environ[self.remote_user_header]:
environ[self.remote_user_header] = f"{environ[self.remote_user_header]}@{self.maildomain}"
path_info = environ.get("PATH_INFO", "")
# The API handles its own authentication via keys
# Check for API key before checking for header
if path_info.startswith("/api/"):
return self.app(environ, start_response)
# If the secret header is enabled, we expect upstream to send along some key
# in HTTP_GX_SECRET, so we'll need to compare that here to the correct value
#
# This is not an ideal location for this function. The reason being
# that because this check is done BEFORE the REMOTE_USER check, it is
# possible to attack the GX_SECRET key without having correct
# credentials. However, that's why it's not "ideal", but it is "good
# enough". The only users able to exploit this are ones with access to
# the local system (unless Galaxy is listening on 0.0.0.0....). It
# seems improbable that an attacker with access to the server hosting
# Galaxy would not have access to Galaxy itself, and be attempting to
# attack the system
if self.config_secret_header is not None:
if environ.get("HTTP_GX_SECRET") is None:
title = "Access to Galaxy is denied"
message = """
Galaxy is configured to authenticate users via an external
method (such as HTTP authentication in Apache), but
no shared secret key was provided by the
upstream (proxy) server.</p>
<p>Please contact your local Galaxy administrator. The
variable <code>remote_user_secret</code> and
<code>GX_SECRET</code> header must be set before you may
access Galaxy.
"""
return self.error(start_response, title, message)
if not safe_str_cmp(environ.get("HTTP_GX_SECRET", ""), self.config_secret_header):
title = "Access to Galaxy is denied"
message = """
Galaxy is configured to authenticate users via an external
method (such as HTTP authentication in Apache), but an
incorrect shared secret key was provided by the
upstream (proxy) server.</p>
<p>Please contact your local Galaxy administrator. The
variable <code>remote_user_secret</code> and
<code>GX_SECRET</code> header must be set before you may
access Galaxy.
"""
return self.error(start_response, title, message)
if environ.get(self.remote_user_header, None):
if not environ[self.remote_user_header].count("@"):
if self.maildomain is not None:
environ[self.remote_user_header] += f"@{self.maildomain}"
else:
title = "Access to Galaxy is denied"
message = """
Galaxy is configured to authenticate users via an external
method (such as HTTP authentication in Apache), but only a
username (not an email address) was provided by the
upstream (proxy) server. Since Galaxy usernames are email
addresses, a default mail domain must be set.</p>
<p>Please contact your local Galaxy administrator. The
variable <code>remote_user_maildomain</code> must be set
before you may access Galaxy.
"""
return self.error(start_response, title, message)
user_accessible_paths = (
"/users",
"/user/api_key",
"/user/edit_username",
"/user/dbkeys",
"/user/logout",
"/user/toolbox_filters",
"/user/set_default_permissions",
)
admin_accessible_paths = (
"/user/create",
"/user/logout",
"/user/manage_user_info",
"/user/edit_info",
"/userskeys/all_users",
"/userskeys/admin_api_keys",
)
if not path_info.startswith("/user"):
# shortcut the following allowlist for non-user-controller
# requests.
pass
elif environ[self.remote_user_header] in self.admin_users and any(
path_info.startswith(prefix) for prefix in admin_accessible_paths
):
# If the user is an admin user, and any of the admin accessible paths match..., allow them to execute that action.
pass
elif any(path_info.startswith(prefix) for prefix in user_accessible_paths):
# If the user is allowed to access the path, pass
pass
elif path_info == "/user" or path_info == "/user/":
pass # We do allow access to the root user preferences page.
elif path_info.startswith("/user"):
# Any other endpoint in the user controller is off limits
title = "Access to Galaxy user controls is disabled"
message = """
User controls are disabled when Galaxy is configured
for external authentication.
"""
return self.error(start_response, title, message)
return self.app(environ, start_response)
else:
log.debug(f"Unable to identify user. {self.remote_user_header} not found")
for k, v in environ.items():
log.debug("%s = %s", k, v)
title = "Access to Galaxy is denied"
message = """
Galaxy is configured to authenticate users via an external
method (such as HTTP authentication in Apache), but a username
was not provided by the upstream (proxy) server. This is
generally due to a misconfiguration in the upstream server.</p>
<p>Please contact your local Galaxy administrator.
"""
return self.error(start_response, title, message)
[docs] def error(self, start_response, title="Access denied", message="Please contact your local Galaxy administrator."):
start_response("403 Forbidden", [("Content-type", "text/html")])
return [errorpage % (title, message)]