diff --git a/src/pretix/helpers/security.py b/src/pretix/helpers/security.py index beae87312..de605fbaa 100644 --- a/src/pretix/helpers/security.py +++ b/src/pretix/helpers/security.py @@ -23,6 +23,10 @@ import hashlib import time from django.conf import settings +from django.contrib.gis.geoip2 import GeoIP2 +from django.core.cache import cache + +from pretix.helpers.http import get_client_ip class SessionInvalid(Exception): @@ -37,6 +41,19 @@ def get_user_agent_hash(request): return hashlib.sha256(request.headers['User-Agent'].encode()).hexdigest() +_geoip = None + + +def _get_country(request): + global _geoip + + if not _geoip: + _geoip = GeoIP2() + + res = _geoip.country(get_client_ip(request)) + return res['country_code'] + + def assert_session_valid(request): if not settings.PRETIX_LONG_SESSIONS or not request.session.get('pretix_auth_long_session', False): last_used = request.session.get('pretix_auth_last_used', time.time()) @@ -54,5 +71,16 @@ def assert_session_valid(request): else: request.session['pinned_user_agent'] = get_user_agent_hash(request) + if settings.HAS_GEOIP: + client_ip = get_client_ip(request) + hashed_client_ip = hashlib.sha256(client_ip.encode()).hexdigest() + country = cache.get_or_set(f'geoip_country_{hashed_client_ip}', lambda: _get_country(request), timeout=300) + + if 'pinned_country' in request.session: + if request.session.get('pinned_country') != country: + raise SessionInvalid() + else: + request.session['pinned_country'] = country + request.session['pretix_auth_last_used'] = int(time.time()) return True diff --git a/src/tests/control/test_auth.py b/src/tests/control/test_auth.py index 394bedea5..1b1020275 100644 --- a/src/tests/control/test_auth.py +++ b/src/tests/control/test_auth.py @@ -47,6 +47,7 @@ from django_otp.oath import TOTP from django_otp.plugins.otp_totp.models import TOTPDevice from pretix.base.models import U2FDevice, User +from pretix.helpers import security class LoginFormTest(TestCase): @@ -797,6 +798,25 @@ class SessionTimeOutTest(TestCase): response = self.client.get('/control/') self.assertEqual(response.status_code, 302) + @override_settings(HAS_GEOIP=True) + def test_pinned_country(self): + class FakeGeoIp: + def country(self, ip): + if ip == '1.2.3.4': + return {'country_code': 'DE'} + return {'country_code': 'US'} + + security._geoip = FakeGeoIp() + self.client.defaults['REMOTE_ADDR'] = '1.2.3.4' + response = self.client.get('/control/') + self.assertEqual(response.status_code, 200) + + self.client.defaults['REMOTE_ADDR'] = '4.3.2.1' + response = self.client.get('/control/') + self.assertEqual(response.status_code, 302) + + security._geoip = None + @pytest.fixture def user():