0001-middleware-add-a-maintenance-middleware-63939.patch
debian/debian_config_common.py | ||
---|---|---|
268 | 268 |
MIDDLEWARE = ( |
269 | 269 |
'hobo.middleware.VersionMiddleware', # /__version__ |
270 | 270 |
'hobo.middleware.cors.CORSMiddleware', |
271 |
'hobo.middleware.maintenance.MaintenanceMiddleware', |
|
271 | 272 |
) + MIDDLEWARE |
272 | 273 | |
273 | 274 |
if PROJECT_NAME != 'wcs': |
hobo/middleware/maintenance.py | ||
---|---|---|
1 |
# hobo - portal to configure and deploy applications |
|
2 |
# Copyright (C) 2015-2022 Entr'ouvert |
|
3 |
# |
|
4 |
# This program is free software: you can redistribute it and/or modify it |
|
5 |
# under the terms of the GNU Affero General Public License as published |
|
6 |
# by the Free Software Foundation, either version 3 of the License, or |
|
7 |
# (at your option) any later version. |
|
8 |
# |
|
9 |
# This program is distributed in the hope that it will be useful, |
|
10 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 |
# GNU Affero General Public License for more details. |
|
13 |
# |
|
14 |
# You should have received a copy of the GNU Affero General Public License |
|
15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | ||
17 |
import dns.exception |
|
18 |
import dns.resolver |
|
19 |
from django.conf import settings |
|
20 |
from django.http import HttpResponse |
|
21 |
from django.utils.translation import ugettext as _ |
|
22 | ||
23 | ||
24 |
def pass_through(remote_addr): |
|
25 |
pass_through_ips = getattr(settings, 'MAINTENANCE_PASS_THROUGH_IPS', []) |
|
26 |
if remote_addr in pass_through_ips: |
|
27 |
return True |
|
28 |
pass_through_ddns = getattr(settings, 'MAINTENANCE_PASS_THROUGH_DDNS', None) |
|
29 |
if pass_through_ddns: |
|
30 |
domain = '.'.join(reversed(remote_addr.split('.'))) + '.' + pass_through_ddns |
|
31 |
try: |
|
32 |
answers = dns.resolver.query(domain, 'A', lifetime=1) |
|
33 |
return any(answer.address for answer in answers) |
|
34 |
except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.exception.DNSException): |
|
35 |
return False |
|
36 |
return False |
|
37 | ||
38 | ||
39 |
class MaintenanceMiddleware: |
|
40 |
def __init__(self, get_response): |
|
41 |
self.get_response = get_response |
|
42 | ||
43 |
def __call__(self, request): |
|
44 |
maintenance_mode = getattr(settings, 'MAINTENANCE_MODE', None) |
|
45 |
if maintenance_mode: |
|
46 |
remote_addr = request.META.get('REMOTE_ADDR') |
|
47 |
if not (remote_addr and pass_through(remote_addr)): |
|
48 |
maintenance_msg = _('The site is under maintenance') |
|
49 |
return HttpResponse('<h1>%s</h1>' % maintenance_msg, status=503) |
|
50 |
return self.get_response(request) |
tests/settings.py | ||
---|---|---|
12 | 12 |
MIDDLEWARE = MIDDLEWARE + ( |
13 | 13 |
'hobo.middleware.RobotsTxtMiddleware', |
14 | 14 |
'hobo.provisionning.middleware.ProvisionningMiddleware', |
15 |
'hobo.middleware.maintenance.MaintenanceMiddleware', |
|
15 | 16 |
) |
16 | 17 | |
17 | 18 |
HOBO_MANAGER_HOMEPAGE_URL_VAR = 'portal_agent_url' |
tests/test_maintenance.py | ||
---|---|---|
1 |
import mock |
|
2 |
from test_manager import login |
|
3 | ||
4 | ||
5 |
def test_maintenance_middleware(app, admin_user, db, monkeypatch, settings): |
|
6 |
app = login(app) |
|
7 |
resp = app.get('/') |
|
8 |
assert resp.status_code == 200 |
|
9 | ||
10 |
settings.MAINTENANCE_MODE = True |
|
11 |
resp = app.get('/', status=503) |
|
12 |
assert 'The site is under maintenance' in resp.text |
|
13 | ||
14 |
settings.MAINTENANCE_PASS_THROUGH_IPS = ['127.0.0.1'] |
|
15 |
resp = app.get('/') |
|
16 |
assert resp.status_code == 200 |
|
17 | ||
18 |
settings.MAINTENANCE_PASS_THROUGH_IPS = [] |
|
19 |
resp = app.get('/', status=503) |
|
20 | ||
21 |
settings.MAINTENANCE_PASS_THROUGH_DDNS = 'ddns.foo.bar' |
|
22 |
with mock.patch('dns.resolver.resolve', return_value=[mock.Mock(address='127.0.0.2')]): |
|
23 |
resp = app.get('/') |
|
24 |
assert resp.status_code == 200 |
|
25 |
with mock.patch('dns.resolver.resolve', return_value=[]): |
|
26 |
resp = app.get('/', status=503) |
|
0 |
- |