middleware: add a maintenance middleware (#63939)
This commit is contained in:
parent
2bb5c296c0
commit
3e149997e4
|
@ -268,6 +268,7 @@ if 'MIDDLEWARE' not in globals():
|
|||
MIDDLEWARE = (
|
||||
'hobo.middleware.VersionMiddleware', # /__version__
|
||||
'hobo.middleware.cors.CORSMiddleware',
|
||||
'hobo.middleware.maintenance.MaintenanceMiddleware',
|
||||
) + MIDDLEWARE
|
||||
|
||||
if PROJECT_NAME != 'wcs':
|
||||
|
|
|
@ -0,0 +1,50 @@
|
|||
# hobo - portal to configure and deploy applications
|
||||
# Copyright (C) 2015-2022 Entr'ouvert
|
||||
#
|
||||
# This program 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.
|
||||
#
|
||||
# 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 Affero 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/>.
|
||||
|
||||
import dns.exception
|
||||
import dns.resolver
|
||||
from django.conf import settings
|
||||
from django.http import HttpResponse
|
||||
from django.utils.translation import ugettext as _
|
||||
|
||||
|
||||
def pass_through(remote_addr):
|
||||
pass_through_ips = getattr(settings, 'MAINTENANCE_PASS_THROUGH_IPS', [])
|
||||
if remote_addr in pass_through_ips:
|
||||
return True
|
||||
pass_through_ddns = getattr(settings, 'MAINTENANCE_PASS_THROUGH_DDNS', None)
|
||||
if pass_through_ddns:
|
||||
domain = '.'.join(reversed(remote_addr.split('.'))) + '.' + pass_through_ddns
|
||||
try:
|
||||
answers = dns.resolver.query(domain, 'A', lifetime=1)
|
||||
return any(answer.address for answer in answers)
|
||||
except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.exception.DNSException):
|
||||
return False
|
||||
return False
|
||||
|
||||
|
||||
class MaintenanceMiddleware:
|
||||
def __init__(self, get_response):
|
||||
self.get_response = get_response
|
||||
|
||||
def __call__(self, request):
|
||||
maintenance_mode = getattr(settings, 'MAINTENANCE_MODE', None)
|
||||
if maintenance_mode:
|
||||
remote_addr = request.META.get('REMOTE_ADDR')
|
||||
if not (remote_addr and pass_through(remote_addr)):
|
||||
maintenance_msg = _('The site is under maintenance')
|
||||
return HttpResponse('<h1>%s</h1>' % maintenance_msg, status=503)
|
||||
return self.get_response(request)
|
|
@ -12,6 +12,7 @@ TEMPLATES[0]['OPTIONS']['debug'] = True
|
|||
MIDDLEWARE = MIDDLEWARE + (
|
||||
'hobo.middleware.RobotsTxtMiddleware',
|
||||
'hobo.provisionning.middleware.ProvisionningMiddleware',
|
||||
'hobo.middleware.maintenance.MaintenanceMiddleware',
|
||||
)
|
||||
|
||||
HOBO_MANAGER_HOMEPAGE_URL_VAR = 'portal_agent_url'
|
||||
|
|
|
@ -0,0 +1,26 @@
|
|||
import mock
|
||||
from test_manager import login
|
||||
|
||||
|
||||
def test_maintenance_middleware(app, admin_user, db, monkeypatch, settings):
|
||||
app = login(app)
|
||||
resp = app.get('/')
|
||||
assert resp.status_code == 200
|
||||
|
||||
settings.MAINTENANCE_MODE = True
|
||||
resp = app.get('/', status=503)
|
||||
assert 'The site is under maintenance' in resp.text
|
||||
|
||||
settings.MAINTENANCE_PASS_THROUGH_IPS = ['127.0.0.1']
|
||||
resp = app.get('/')
|
||||
assert resp.status_code == 200
|
||||
|
||||
settings.MAINTENANCE_PASS_THROUGH_IPS = []
|
||||
resp = app.get('/', status=503)
|
||||
|
||||
settings.MAINTENANCE_PASS_THROUGH_DDNS = 'ddns.foo.bar'
|
||||
with mock.patch('dns.resolver.resolve', return_value=[mock.Mock(address='127.0.0.2')]):
|
||||
resp = app.get('/')
|
||||
assert resp.status_code == 200
|
||||
with mock.patch('dns.resolver.resolve', return_value=[]):
|
||||
resp = app.get('/', status=503)
|
Loading…
Reference in New Issue