# -*- coding: utf-8 -*-# vim: tabstop=4 shiftwidth=4 softtabstop=4## Copyright (C) 2015-2023 GEM Foundation## OpenQuake is free software: you can redistribute it and/or modify it# under the terms of the GNU Affero General Public License as published# by the Free Software Foundation, either version 3 of the License, or# (at your option) any later version.## OpenQuake 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 Affero General Public License for more details.## You should have received a copy of the GNU Affero General Public License# along with OpenQuake. If not, see <http://www.gnu.org/licenses/>.importosimportgetpassimportrequestsimportloggingfromtimeimportsleepfromdjango.confimportsettingsfromdjango.appsimportappsfromdjango.contrib.authimportget_user_modelfromopenquake.engineimport__version__asoqversion
[docs]defis_superuser(request):""" Without authentication (settings.LOCKDOW is false) every user is considered a superuser, otherwise look at the attribute `request.user.is_superuser`. """ifnotsettings.LOCKDOWN:returnTruereturnrequest.user.is_superuserifhasattr(request,'user')elseFalse
[docs]defget_user(request):""" Returns the users from `request` if authentication is enabled, otherwise returns the default user (from settings, or as reported by the OS). """ifsettings.LOCKDOWNandhasattr(request,'user'):ifrequest.user.is_authenticated:user=request.user.usernameelse:# This may happen with crafted requestsuser=''else:user=getattr(settings,'DEFAULT_USER',getpass.getuser())returnuser
[docs]defget_valid_users(request):"""" Returns a list of `users` based on groups membership. Returns a list made of a single user when it is not member of any group. """ifsettings.LOCKDOWN:User=get_user_model()users=[get_user(request)]ifsettings.LOCKDOWNandhasattr(request,'user'):ifrequest.user.is_authenticated:groups=request.user.groups.all()ifgroups:users=list(User.objects.filter(groups__in=groups).values_list('username',flat=True))else:# This may happen with crafted requestsusers=[]returnusers
[docs]defget_acl_on(request):""" Returns `True` if ACL should be honorated, returns otherwise `False`. """acl_on=settings.ACL_ONifis_superuser(request):# ACL is always disabled for superusersacl_on=Falsereturnacl_on
[docs]defuser_has_permission(request,owner):""" Returns `True` if user coming from the request has the permission to view a resource, returns `false` otherwise. """returnowneringet_valid_users(request)ornotget_acl_on(request)
[docs]defoq_server_context_processor(request):""" A custom context processor which allows injection of additional context variables. """# NOTE: defining env variable at runtime, instead of defining it when the# engine imports variable from the server moduleos.environ['OQ_APPLICATION_MODE']=settings.APPLICATION_MODEcontext={}try:announcement_model=apps.get_model(app_label='announcements',model_name='Announcement')exceptLookupError:announcements=Noneelse:announcements=announcement_model.objects.filter(show=True)webui_host=request.get_host()context['oq_engine_server_url']=('//'+(webui_hostifwebui_hostelserequest.META.get('HTTP_HOST','localhost:8800'))+settings.WEBUI_PATHPREFIX)# this context var is also evaluated by the STANDALONE_APPS to identify# the running environment. Keep it as it iscontext['oq_engine_version']=oqversioncontext['disable_version_warning']=settings.DISABLE_VERSION_WARNINGcontext['server_name']=settings.SERVER_NAMEcontext['tools_only']=settings.APPLICATION_MODE=='TOOLS_ONLY'context['announcements']=announcementsreturncontext
[docs]defcheck_webserver_running(url="http://localhost:8800",max_retries=30):""" Returns True if a given URL is responding within a given timeout. """retry=0response=''success=Falsewhileresponse!=requests.codes.okandretry<max_retries:try:response=requests.head(url,allow_redirects=True).status_codesuccess=TrueexceptException:sleep(1)retry+=1ifnotsuccess:logging.warning('Unable to connect to %s within %s retries'%(url,max_retries))returnsuccess