Compare commits

..

6 Commits

Author SHA1 Message Date
Mira Weller
9da8c1f7b2 reauth flow token
add a special token to always allow completing a form submission, even if the reauthentication time has expired
2026-03-26 12:41:19 +01:00
Mira Weller
6b340682b2 ui changes 2026-03-20 12:56:57 +01:00
Mira Weller
0dc436067f always perform 2fa activation as dedicated step 2026-03-20 12:56:38 +01:00
Mira Weller
db66c91108 generate emergency tokens during 2fa activation 2026-03-20 12:55:58 +01:00
Mira Weller
3a1db55e8b remove broken blackberry link 2026-03-20 12:06:56 +01:00
Mira Weller
57da5cbae2 improve 2fa type selection 2026-03-20 11:44:10 +01:00
41 changed files with 268 additions and 451 deletions

View File

@@ -24,7 +24,7 @@ jobs:
name: Packaging
strategy:
matrix:
python-version: ["3.13"]
python-version: ["3.11"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}

View File

@@ -24,10 +24,10 @@ jobs:
name: Check gettext syntax
steps:
- uses: actions/checkout@v4
- name: Set up Python 3.13
- name: Set up Python 3.11
uses: actions/setup-python@v5
with:
python-version: 3.13
python-version: 3.11
- uses: actions/cache@v4
with:
path: ~/.cache/pip
@@ -49,10 +49,10 @@ jobs:
name: Spellcheck
steps:
- uses: actions/checkout@v4
- name: Set up Python 3.13
- name: Set up Python 3.11
uses: actions/setup-python@v5
with:
python-version: 3.13
python-version: 3.11
- uses: actions/cache@v4
with:
path: ~/.cache/pip

View File

@@ -24,10 +24,10 @@ jobs:
runs-on: ubuntu-22.04
steps:
- uses: actions/checkout@v4
- name: Set up Python 3.13
- name: Set up Python 3.11
uses: actions/setup-python@v5
with:
python-version: 3.13
python-version: 3.11
- uses: actions/cache@v4
with:
path: ~/.cache/pip
@@ -44,10 +44,10 @@ jobs:
runs-on: ubuntu-22.04
steps:
- uses: actions/checkout@v4
- name: Set up Python 3.13
- name: Set up Python 3.11
uses: actions/setup-python@v5
with:
python-version: 3.13
python-version: 3.11
- uses: actions/cache@v4
with:
path: ~/.cache/pip
@@ -64,10 +64,10 @@ jobs:
runs-on: ubuntu-22.04
steps:
- uses: actions/checkout@v4
- name: Set up Python 3.13
- name: Set up Python 3.11
uses: actions/setup-python@v5
with:
python-version: 3.13
python-version: 3.11
- name: Install Dependencies
run: pip3 install licenseheaders
- name: Run licenseheaders

View File

@@ -23,15 +23,13 @@ jobs:
name: Tests
strategy:
matrix:
python-version: ["3.11", "3.13", "3.14"]
python-version: ["3.10", "3.11", "3.13"]
database: [sqlite, postgres]
exclude:
- database: sqlite
python-version: "3.10"
- database: sqlite
python-version: "3.11"
- database: sqlite
python-version: "3.12"
services:
postgres:
image: postgres:15
@@ -83,4 +81,4 @@ jobs:
file: src/coverage.xml
token: ${{ secrets.CODECOV_TOKEN }}
fail_ci_if_error: false
if: matrix.database == 'postgres' && matrix.python-version == '3.13'
if: matrix.database == 'postgres' && matrix.python-version == '3.11'

View File

@@ -3,7 +3,7 @@ name = "pretix"
dynamic = ["version"]
description = "Reinventing presales, one ticket at a time"
readme = "README.rst"
requires-python = ">=3.11"
requires-python = ">=3.10"
license = {file = "LICENSE"}
keywords = ["tickets", "web", "shop", "ecommerce"]
authors = [
@@ -19,11 +19,10 @@ classifiers = [
"Topic :: Internet :: WWW/HTTP :: Dynamic Content",
"Environment :: Web Environment",
"License :: OSI Approved :: GNU Affero General Public License v3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Framework :: Django :: 5.2",
"Framework :: Django :: 4.2",
]
dependencies = [
@@ -37,7 +36,7 @@ dependencies = [
"css-inline==0.20.*",
"defusedcsv>=1.1.0",
"dnspython==2.*",
"Django[argon2]==5.2.*",
"Django[argon2]==4.2.*,>=4.2.26",
"django-bootstrap3==26.1",
"django-compressor==4.6.0",
"django-countries==8.2.*",
@@ -60,7 +59,7 @@ dependencies = [
"dnspython==2.8.*",
"drf_ujson2==1.7.*",
"geoip2==5.*",
"importlib_metadata==9.*", # Polyfill, we can probably drop this once we require Python 3.10+
"importlib_metadata==8.*", # Polyfill, we can probably drop this once we require Python 3.10+
"isoweek",
"jsonschema",
"kombu==5.6.*",
@@ -93,7 +92,7 @@ dependencies = [
"redis==7.1.*",
"reportlab==4.4.*",
"requests==2.32.*",
"sentry-sdk==2.56.*",
"sentry-sdk==2.54.*",
"sepaxml==2.7.*",
"stripe==7.9.*",
"text-unidecode==1.*",

View File

@@ -45,6 +45,7 @@ import pycountry
from django import forms
from django.conf import settings
from django.contrib import messages
from django.contrib.gis.geoip2 import GeoIP2
from django.core.exceptions import ValidationError
from django.core.files.uploadedfile import SimpleUploadedFile
from django.core.validators import (
@@ -101,7 +102,6 @@ from pretix.helpers.countries import (
from pretix.helpers.escapejson import escapejson_attr
from pretix.helpers.http import get_client_ip
from pretix.helpers.i18n import get_format_without_seconds
from pretix.helpers.security import get_geoip
from pretix.presale.signals import question_form_fields
logger = logging.getLogger(__name__)
@@ -393,7 +393,7 @@ class WrappedPhoneNumberPrefixWidget(PhoneNumberPrefixWidget):
def guess_country_from_request(request, event):
if settings.HAS_GEOIP:
g = get_geoip()
g = GeoIP2()
try:
res = g.country(get_client_ip(request))
if res['country_code'] and len(res['country_code']) == 2:

View File

@@ -36,9 +36,8 @@ from django.core.management.commands.makemigrations import Command as Parent
from ._migrations import monkeypatch_migrations
monkeypatch_migrations()
class Command(Parent):
def handle(self, *args, **kwargs):
monkeypatch_migrations()
return super().handle(*args, **kwargs)
pass

View File

@@ -64,7 +64,7 @@ class Command(BaseCommand):
if not periodic_task.receivers or periodic_task.sender_receivers_cache.get(self) is NO_RECEIVERS:
return
for receiver in periodic_task._live_receivers(self)[0]:
for receiver in periodic_task._live_receivers(self):
name = f'{receiver.__module__}.{receiver.__name__}'
if options['list_tasks']:
print(name)

View File

@@ -41,20 +41,16 @@ class Migration(migrations.Migration):
name='datetime',
field=models.DateTimeField(),
),
migrations.AddIndex(
'logentry',
models.Index(fields=('datetime', 'id'), name="pretixbase__datetim_b1fe5a_idx"),
migrations.AlterIndexTogether(
name='logentry',
index_together={('datetime', 'id')},
),
migrations.AddIndex(
'order',
models.Index(fields=["datetime", "id"], name="pretixbase__datetim_66aff0_idx"),
migrations.AlterIndexTogether(
name='order',
index_together={('datetime', 'id'), ('last_modified', 'id')},
),
migrations.AddIndex(
'order',
models.Index(fields=["last_modified", "id"], name="pretixbase__last_mo_4ebf8b_idx"),
),
migrations.AddIndex(
'transaction',
models.Index(fields=('datetime', 'id'), name="pretixbase__datetim_b20405_idx"),
migrations.AlterIndexTogether(
name='transaction',
index_together={('datetime', 'id')},
),
]

View File

@@ -61,10 +61,7 @@ class Migration(migrations.Migration):
options={
'ordering': ('identifier', 'type', 'organizer'),
'unique_together': {('identifier', 'type', 'organizer')},
'indexes': [
models.Index(fields=('identifier', 'type', 'organizer'), name='reusable_medium_organizer_index'),
models.Index(fields=('updated', 'id'), name="pretixbase__updated_093277_idx")
],
'index_together': {('identifier', 'type', 'organizer'), ('updated', 'id')},
},
bases=(models.Model, pretix.base.models.base.LoggingMixin),
),

View File

@@ -9,6 +9,31 @@ class Migration(migrations.Migration):
]
operations = [
migrations.RenameIndex(
model_name="logentry",
new_name="pretixbase__datetim_b1fe5a_idx",
old_fields=("datetime", "id"),
),
migrations.RenameIndex(
model_name="order",
new_name="pretixbase__datetim_66aff0_idx",
old_fields=("datetime", "id"),
),
migrations.RenameIndex(
model_name="order",
new_name="pretixbase__last_mo_4ebf8b_idx",
old_fields=("last_modified", "id"),
),
migrations.RenameIndex(
model_name="reusablemedium",
new_name="pretixbase__updated_093277_idx",
old_fields=("updated", "id"),
),
migrations.RenameIndex(
model_name="transaction",
new_name="pretixbase__datetim_b20405_idx",
old_fields=("datetime", "id"),
),
migrations.AlterField(
model_name="attendeeprofile",
name="id",

View File

@@ -1,6 +1,6 @@
# Generated by Django 4.2.10 on 2024-04-02 15:16
from django.db import migrations, models
from django.db import migrations
class Migration(migrations.Migration):
@@ -10,8 +10,8 @@ class Migration(migrations.Migration):
]
operations = [
migrations.RemoveIndex(
"reusablemedium",
'reusable_medium_organizer_index',
migrations.AlterIndexTogether(
name="reusablemedium",
index_together=set(),
),
]

View File

@@ -88,7 +88,7 @@ class LogEntry(models.Model):
class Meta:
ordering = ('-datetime', '-id')
indexes = [models.Index(fields=["datetime", "id"], name="pretixbase__datetim_b1fe5a_idx")]
indexes = [models.Index(fields=["datetime", "id"])]
def display(self):
from pretix.base.logentrytype_registry import log_entry_types

View File

@@ -122,7 +122,7 @@ class ReusableMedium(LoggedModel):
class Meta:
unique_together = (("identifier", "type", "organizer"),)
indexes = [
models.Index(fields=("updated", "id"), name="pretixbase__updated_093277_idx"),
models.Index(fields=("updated", "id")),
]
ordering = "identifier", "type", "organizer"

View File

@@ -336,8 +336,8 @@ class Order(LockModel, LoggedModel):
verbose_name_plural = _("Orders")
ordering = ("-datetime", "-pk")
indexes = [
models.Index(fields=["datetime", "id"], name="pretixbase__datetim_66aff0_idx"),
models.Index(fields=["last_modified", "id"], name="pretixbase__last_mo_4ebf8b_idx"),
models.Index(fields=["datetime", "id"]),
models.Index(fields=["last_modified", "id"]),
]
constraints = [
models.UniqueConstraint(fields=["organizer", "code"], name="order_organizer_code_uniq"),
@@ -3080,7 +3080,7 @@ class Transaction(models.Model):
class Meta:
ordering = 'datetime', 'pk'
indexes = [
models.Index(fields=['datetime', 'id'], name="pretixbase__datetim_b20405_idx")
models.Index(fields=['datetime', 'id'])
]
def save(self, *args, **kwargs):

View File

@@ -100,7 +100,7 @@ def primary_font_kwargs():
choices = [('Open Sans', 'Open Sans')]
choices += sorted([
(a, FontSelect.FontOption(title=a, data=v)) for a, v in get_fonts(pdf_support_required=False).items()
(a, {"title": a, "data": v}) for a, v in get_fonts(pdf_support_required=False).items()
], key=lambda a: a[0])
return {
'choices': choices,
@@ -4148,14 +4148,6 @@ def validate_event_settings(event, settings_dict):
)
]}
)
if (
settings_dict.get('invoice_address_from_vat_id') and
settings_dict.get('invoice_address_from_country') and
settings_dict.get('invoice_address_from_country') not in VAT_ID_COUNTRIES
):
raise ValidationError({
'invoice_address_from_vat_id': _('VAT-ID is not supported for "{}".').format(settings_dict.get('invoice_address_from_country'))
})
payment_term_last = settings_dict.get('payment_term_last')
if payment_term_last and event.presale_end:

View File

@@ -32,7 +32,6 @@
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under the License.
import logging
import warnings
from typing import Any, Callable, Generic, List, Tuple, TypeVar
@@ -49,8 +48,6 @@ from .plugins import (
PLUGIN_LEVEL_ORGANIZER,
)
logger = logging.getLogger(__name__)
app_cache = {}
T = TypeVar('T')
@@ -63,25 +60,23 @@ def _populate_app_cache():
def get_defining_app(o):
# If sentry packed this in a wrapper, unpack that
module = getattr(o, "__module__", None)
if module and "sentry" in module:
if "sentry" in o.__module__:
o = o.__wrapped__
if hasattr(o, "__mocked_app"):
return o.__mocked_app
# Find the Django application this belongs to
searchpath = module or getattr(o.__class__, "__module__", None) or ""
searchpath = o.__module__
# Core modules are always active
if searchpath and any(searchpath.startswith(cm) for cm in settings.CORE_MODULES):
if any(searchpath.startswith(cm) for cm in settings.CORE_MODULES):
return 'CORE'
if not app_cache:
_populate_app_cache()
app = None
while searchpath:
while True:
app = app_cache.get(searchpath)
if "." not in searchpath or app:
break
@@ -162,7 +157,7 @@ class PluginSignal(Generic[T], django.dispatch.Signal):
if not app_cache:
_populate_app_cache()
for receiver in self._live_receivers(sender)[0]:
for receiver in self._sorted_receivers(sender):
if self._is_receiver_active(sender, receiver):
response = receiver(signal=self, sender=sender, **named)
responses.append((receiver, response))
@@ -184,7 +179,7 @@ class PluginSignal(Generic[T], django.dispatch.Signal):
if not app_cache:
_populate_app_cache()
for receiver in self._live_receivers(sender)[0]:
for receiver in self._sorted_receivers(sender):
if self._is_receiver_active(sender, receiver):
named[chain_kwarg_name] = response
response = receiver(signal=self, sender=sender, **named)
@@ -209,7 +204,7 @@ class PluginSignal(Generic[T], django.dispatch.Signal):
if not app_cache:
_populate_app_cache()
for receiver in self._live_receivers(sender)[0]:
for receiver in self._sorted_receivers(sender):
if self._is_receiver_active(sender, receiver):
try:
response = receiver(signal=self, sender=sender, **named)
@@ -219,35 +214,17 @@ class PluginSignal(Generic[T], django.dispatch.Signal):
responses.append((receiver, response))
return responses
def asend(self, sender: T, **named):
raise NotImplementedError() # NOQA
def asend_robust(self, sender: T, **named):
raise NotImplementedError() # NOQA
def _live_receivers(self, sender):
orig_list, orig_async_list = super()._live_receivers(sender)
if orig_async_list:
logger.error('Async receivers are not supported.')
raise NotImplementedError
def _getattr_fallback_to_class(obj, key):
return getattr(obj, key, getattr(obj.__class__, key))
def _is_core_module(receiver):
m = _getattr_fallback_to_class(receiver, "__module__")
return any(m.startswith(c) for c in settings.CORE_MODULES)
def _sorted_receivers(self, sender):
orig_list = self._live_receivers(sender)
sorted_list = sorted(
orig_list,
key=lambda receiver: (
0 if _is_core_module(receiver) else 1,
_getattr_fallback_to_class(receiver, "__module__"),
_getattr_fallback_to_class(receiver, "__name__"),
0 if any(receiver.__module__.startswith(m) for m in settings.CORE_MODULES) else 1,
receiver.__module__,
receiver.__name__,
)
)
return sorted_list, []
return sorted_list
class EventPluginSignal(PluginSignal[Event]):
@@ -323,41 +300,23 @@ class GlobalSignal(django.dispatch.Signal):
if not self.receivers or self.sender_receivers_cache.get(sender) is NO_RECEIVERS:
return response
for receiver in self._live_receivers(sender)[0]:
for receiver in self._live_receivers(sender):
named[chain_kwarg_name] = response
response = receiver(signal=self, sender=sender, **named)
return response
def asend(self, sender: T, **named):
raise NotImplementedError() # NOQA
def asend_robust(self, sender: T, **named):
raise NotImplementedError() # NOQA
def _live_receivers(self, sender):
# Ensure consistent sorting of receivers
orig_list, orig_async_list = super()._live_receivers(sender)
if orig_async_list:
logger.error('Async receivers are not supported.')
raise NotImplementedError
def _getattr_fallback_to_class(obj, key):
return getattr(obj, key, getattr(obj.__class__, key))
def _is_core_module(receiver):
m = _getattr_fallback_to_class(receiver, "__module__")
return any(m.startswith(c) for c in settings.CORE_MODULES)
orig_list = super()._live_receivers(sender)
sorted_list = sorted(
orig_list,
key=lambda receiver: (
0 if _is_core_module(receiver) else 1,
_getattr_fallback_to_class(receiver, "__module__"),
_getattr_fallback_to_class(receiver, "__name__"),
0 if any(receiver.__module__.startswith(m) for m in settings.CORE_MODULES) else 1,
receiver.__module__,
receiver.__name__,
)
)
return sorted_list, []
return sorted_list
class DeprecatedSignal(GlobalSignal):

View File

@@ -34,7 +34,6 @@
import datetime
import os
from dataclasses import dataclass
from django import forms
from django.conf import settings
@@ -421,11 +420,6 @@ class SplitDateTimeField(forms.SplitDateTimeField):
class FontSelect(forms.RadioSelect):
option_template_name = 'pretixcontrol/font_option.html'
@dataclass
class FontOption:
title: str
data: str
class ItemMultipleChoiceField(SafeModelMultipleChoiceField):
def label_from_instance(self, obj):

View File

@@ -63,7 +63,7 @@ from pretix.base.forms import (
from pretix.base.models import Event, Organizer, TaxRule, Team
from pretix.base.models.event import EventFooterLink, EventMetaValue, SubEvent
from pretix.base.models.organizer import TeamQuerySet
from pretix.base.models.tax import TAX_CODE_LISTS, VAT_ID_COUNTRIES
from pretix.base.models.tax import TAX_CODE_LISTS
from pretix.base.reldate import RelativeDateField, RelativeDateTimeField
from pretix.base.services.placeholders import FormPlaceholderMixin
from pretix.base.settings import (
@@ -73,8 +73,8 @@ from pretix.base.settings import (
)
from pretix.base.validators import multimail_validate
from pretix.control.forms import (
FontSelect, MultipleLanguagesWidget, SalesChannelCheckboxSelectMultiple,
SlugWidget, SplitDateTimeField, SplitDateTimePickerWidget,
MultipleLanguagesWidget, SalesChannelCheckboxSelectMultiple, SlugWidget,
SplitDateTimeField, SplitDateTimePickerWidget,
)
from pretix.control.forms.widgets import Select2
from pretix.helpers.countries import CachedCountries
@@ -531,13 +531,6 @@ class EventUpdateForm(I18nModelForm):
class EventSettingsValidationMixin:
def clean_invoice_address_from_vat_id(self):
value = self.cleaned_data.get('invoice_address_from_vat_id')
country = self.cleaned_data.get('invoice_address_from_country')
if value and country and country not in VAT_ID_COUNTRIES:
return None
return value
def clean(self):
data = super().clean()
settings_dict = self.obj.settings.freeze()
@@ -729,7 +722,7 @@ class EventSettingsForm(EventSettingsValidationMixin, FormPlaceholderMixin, Sett
del self.fields['event_list_filters']
del self.fields['event_calendar_future_only']
self.fields['primary_font'].choices = [('Open Sans', 'Open Sans')] + sorted([
(a, FontSelect.FontOption(title=a, data=v)) for a, v in get_fonts(self.event, pdf_support_required=False).items()
(a, {"title": a, "data": v}) for a, v in get_fonts(self.event, pdf_support_required=False).items()
], key=lambda a: a[0])
# create "virtual" fields for better UX when editing <name>_asked and <name>_required fields

View File

@@ -363,7 +363,7 @@ def get_global_navigation(request):
'icon': 'dashboard',
},
]
if request.user.is_in_any_teams or request.user.is_staff:
if request.user.is_in_any_teams:
nav += [
{
'label': _('Events'),

View File

@@ -6,9 +6,38 @@
<h1>{% trans "Add a two-factor authentication device" %}</h1>
<form action="" method="post" class="form-horizontal">
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
{% bootstrap_form_errors form %}
{% bootstrap_field form.name layout='horizontal' %}
{% bootstrap_field form.devicetype layout='horizontal' %}
<div class="form-group">
<label class="col-md-3 control-label">{% trans "Device type" %}</label>
<div class="col-md-9">
<div class="big-radio radio">
<label>
<input type="radio" value="totp" name="{{ form.devicetype.html_name }}" {% if form.devicetype.value == "totp" %}checked{% endif %}>
<strong>{% trans "Smartphone with the Authenticator application" %}</strong><br>
<div class="help-block">
{% blocktrans trimmed %}
Use your smartphone with any Time-based One-Time-Password app like freeOTP, Google Authenticator or Proton Authenticator.
{% endblocktrans %}
</div>
</label>
</div>
<div class="big-radio radio">
<label>
<input type="radio" value="webauthn" name="{{ form.devicetype.html_name }}" {% if form.devicetype.value == "webauthn" %}checked{% endif %}>
<strong>{% trans "WebAuthn-compatible hardware token" %}</strong><br>
<div class="help-block">
{% blocktrans trimmed %}
Use a hardware token like the Yubikey, or biometric authentication on iOS, macOS and Android.
{% endblocktrans %}
</div>
</label>
</div>
</div>
</div>
<div class="form-group submit-group">
<button type="submit" class="btn btn-primary btn-save">
{% trans "Continue" %}

View File

@@ -28,11 +28,6 @@
{% trans "iOS (iTunes)" %}
</a>
</li>
<li>
<a href="https://m.google.com/authenticator">
{% trans "Blackberry (Link via Google)" %}
</a>
</li>
</ul>
</li>
<li>
@@ -74,14 +69,11 @@
{% trans "Enter the displayed code here:" %}
<form class="form form-inline" method="post" action="">
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
<input type="number" name="token" class="form-control" required="required">
<button class="btn btn-primary" type="submit">
{% trans "Continue" %}
</button><br>
<label>
<input type="checkbox" name="activate" checked="checked" value="on">
{% trans "Require second factor for future logins" %}
</label>
</form>
</li>
</ol>

View File

@@ -12,13 +12,9 @@
</p>
<form class="form form-inline" method="post" action="" id="webauthn-form">
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
<input type="hidden" id="webauthn-response" name="token" class="form-control" required="required">
<p>
<label>
<input type="checkbox" name="activate" checked="checked" value="on">
{% trans "Require second factor for future logins" %}
</label>
</p>
<button class="btn btn-primary sr-only" type="submit"></button>
</form>

View File

@@ -6,6 +6,7 @@
<h1>{% trans "Delete a two-factor authentication device" %}</h1>
<form action="" method="post" class="form-horizontal">
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
<p>{% blocktrans trimmed with device=device.name %}
Are you sure you want to delete the authentication device "{{ device }}"?
{% endblocktrans %}</p>

View File

@@ -6,6 +6,7 @@
<h1>{% trans "Disable two-factor authentication" %}</h1>
<form action="" method="post" class="form-horizontal">
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
<p>
{% trans "Do you really want to disable two-factor authentication?" %}
</p>

View File

@@ -1,23 +1,58 @@
{% extends "pretixcontrol/base.html" %}
{% load i18n %}
{% load bootstrap3 %}
{% load icon %}
{% block title %}{% trans "Enable two-factor authentication" %}{% endblock %}
{% block content %}
<h1>{% trans "Enable two-factor authentication" %}</h1>
<form action="" method="post" class="form-horizontal">
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
<p>
{% trans "Do you really want to enable two-factor authentication?" %}
</p>
<p>
{% trans "You will no longer be able to log in to pretix without one of your configured devices." %}
{% trans "Please make sure to print out or copy the emergency tokens and store them in a safe place." %}
</p>
{% if new_emergency_tokens %}
<div class="panel panel-default">
<div class="panel-heading">
<h3 class="panel-title">{% trans "Your emergency codes" %}</h3>
</div>
<div class="panel-body">
<p>
{% blocktrans trimmed %}
If you lose access to your devices, you can use one of your emergency tokens to log in.
We recommend to store them in a safe place, e.g. printed out or in a password manager.
Every token can be used at most once.
{% endblocktrans %}
</p>
<ul>
{% for code in new_emergency_tokens %}
<li>{{ code }}</li>
{% endfor %}
</ul>
<p>
<label>
<input type="checkbox" required>
{% trans "I stored my emergency tokens in a safe place." %}
</label>
</p>
</div>
</div>
{% else %}
<p>
{% icon "info-circle" %}
{% blocktrans trimmed with generation_date_time=static_tokens_device.created_at %}
You generated your emergency tokens on {{ generation_date_time }}.
{% endblocktrans %}
</p>
{% endif %}
<div class="form-group submit-group">
<a href="{% url "control:user.settings.2fa" %}" class="btn btn-default btn-cancel">
{% trans "Cancel" %}
</a>
<button type="submit" class="btn btn-danger btn-save">
<button type="submit" class="btn btn-primary btn-save">
{% trans "Enable" %}
</button>
</div>

View File

@@ -6,6 +6,7 @@
<h1>{% trans "Leave teams that require two-factor authentication" %}</h1>
<form action="" method="post" class="form-horizontal">
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
<p>
<strong>{% trans "Do you really want to leave the following teams?" %}</strong>
</p>

View File

@@ -1,5 +1,6 @@
{% extends "pretixcontrol/base.html" %}
{% load i18n %}
{% load icon %}
{% load bootstrap3 %}
{% block title %}{% trans "Two-factor authentication" %}{% endblock %}
{% block content %}
@@ -120,7 +121,7 @@
Delete
</a>
{% if d.devicetype == "totp" %}
<span class="fa fa-mobile"></span>
<span class="fa fa-mobile fa-lg"></span>
{% elif d.devicetype == "webauthn" %}
<span class="fa fa-usb"></span>
{% elif d.devicetype == "u2f" %}
@@ -152,19 +153,30 @@
</p>
{% if static_tokens_device %}
<p>
{% icon "info-circle" %}
{% blocktrans trimmed with generation_date_time=static_tokens_device.created_at %}
You generated your emergency tokens on {{ generation_date_time }}.
{% endblocktrans %}
</p>
{% else %}
<a href="{% url "control:user.settings.2fa.regenemergency" %}" class="btn btn-default">
<span class="fa fa-refresh"></span>
{% trans "Generate new emergency tokens" %}
</a>
{% elif user.require_2fa %}
<p>
{% trans "You don't have any emergency tokens yet." %}
{% icon "warning" %}
<strong>{% trans "You don't have any emergency tokens yet." %}</strong>
</p>
<a href="{% url "control:user.settings.2fa.regenemergency" %}" class="btn btn-default">
<span class="fa fa-refresh"></span>
{% trans "Generate emergency tokens" %}
</a>
{% else %}
<p class="help-block">
{% icon "info-circle" %}
{% trans "Emergency tokens will be generated when you enable two-factor authentication." %}
</p>
{% endif %}
<a href="{% url "control:user.settings.2fa.regenemergency" %}" class="btn btn-default">
<span class="fa fa-refresh"></span>
{% trans "Generate new emergency tokens" %}
</a>
</div>
</div>
{% endblock %}

View File

@@ -6,6 +6,7 @@
<h1>{% trans "Regenerate emergency codes" %}</h1>
<form action="" method="post" class="form-horizontal">
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
<p>
{% trans "Do you really want to regenerate your emergency codes?" %}
</p>

View File

@@ -8,6 +8,7 @@
{% trans "Change login email address" %}
</h1>
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
{% bootstrap_form_errors form %}
<p class="text-muted">
{% trans "This changes the email address used to login to your account, as well as where we send email notifications." %}

View File

@@ -9,6 +9,7 @@
</h1>
<br>
{% csrf_token %}
<input type="hidden" name="flow_token" value="{{ flow_token }}">
{% bootstrap_form_errors form %}
{% bootstrap_field form.email %}
{% bootstrap_field form.old_pw %}

View File

@@ -641,7 +641,7 @@ def user_index(request):
ctx = {
'widgets': rearrange(widgets),
'can_create_event': request.user.teams.with_organizer_permission("organizer.events:create").exists() or request.user.is_staff,
'can_create_event': request.user.teams.with_organizer_permission("organizer.events:create").exists(),
'upcoming': widgets_for_event_qs(
request,
annotated_event_query(request, lazy=True).filter(

View File

@@ -89,13 +89,31 @@ logger = logging.getLogger(__name__)
class RecentAuthenticationRequiredMixin:
max_time = 900
max_form_time = 900
@method_decorator(never_cache)
def dispatch(self, request, *args, **kwargs):
tdelta = time.time() - request.session.get('pretix_auth_login_time', 0)
if tdelta > self.max_time:
auth_is_recent = time.time() - request.session.get('pretix_auth_login_time', 0) < self.max_time
allowed_by_token = (
request.session.pop('pretix_reauthed_flow_token', None) == request.POST.get('flow_token', '')
and request.session.pop('pretix_reauthed_flow_allowed_url', None) == request.get_full_path()
and time.time() - request.session.pop('pretix_reauthed_flow_start_time', 0) < self.max_form_time
)
if auth_is_recent or allowed_by_token:
return super().dispatch(request, *args, **kwargs)
else:
return redirect(reverse('control:user.reauth') + '?next=' + quote(request.get_full_path()))
return super().dispatch(request, *args, **kwargs)
def get_flow_token(self):
self.request.session['pretix_reauthed_flow_allowed_url'] = self.request.get_full_path()
self.request.session['pretix_reauthed_flow_token'] = get_random_string(22)
self.request.session['pretix_reauthed_flow_start_time'] = time.time()
return self.request.session['pretix_reauthed_flow_token']
def get_context_data(self, **kwargs):
ctx = super().get_context_data()
ctx['flow_token'] = self.get_flow_token()
return ctx
class ReauthView(TemplateView):
@@ -283,6 +301,7 @@ class UserHistoryView(ListView):
class User2FAMainView(RecentAuthenticationRequiredMixin, TemplateView):
max_time = 7200
template_name = 'pretixcontrol/user/2fa_main.html'
def get_context_data(self, **kwargs):
@@ -465,25 +484,15 @@ class User2FADeviceConfirmWebAuthnView(RecentAuthenticationRequiredMixin, Templa
notices = [
_('A new two-factor authentication device has been added to your account.')
]
activate = request.POST.get('activate', '')
if activate == 'on' and not self.request.user.require_2fa:
self.request.user.require_2fa = True
self.request.user.save()
self.request.user.log_action('pretix.user.settings.2fa.enabled', user=self.request.user)
notices.append(
_('Two-factor authentication has been enabled.')
)
self.request.user.send_security_notice(notices)
self.request.user.update_session_token()
update_session_auth_hash(self.request, self.request.user)
note = ''
if not self.request.user.require_2fa:
note = ' ' + str(_('Please note that you still need to enable two-factor authentication for your '
'account using the buttons below to make a second factor required for logging '
'into your account.'))
messages.success(request, str(_('The device has been verified and can now be used.')) + note)
return redirect(reverse('control:user.settings.2fa'))
messages.success(request, str(_('The device has been verified and can now be used.')))
if self.request.user.require_2fa:
return redirect(reverse('control:user.settings.2fa'))
else:
return redirect(reverse('control:user.settings.2fa.enable'))
except Exception:
messages.error(request, _('The registration could not be completed. Please try again.'))
logger.exception('WebAuthn registration failed')
@@ -494,6 +503,7 @@ class User2FADeviceConfirmWebAuthnView(RecentAuthenticationRequiredMixin, Templa
class User2FADeviceConfirmTOTPView(RecentAuthenticationRequiredMixin, TemplateView):
template_name = 'pretixcontrol/user/2fa_confirm_totp.html'
max_form_time = 7200 # this should have effectively no timeout, as the user might need to download the 2fa app first
@cached_property
def device(self):
@@ -514,7 +524,6 @@ class User2FADeviceConfirmTOTPView(RecentAuthenticationRequiredMixin, TemplateVi
def post(self, request, *args, **kwargs):
token = request.POST.get('token', '')
activate = request.POST.get('activate', '')
if self.device.verify_token(token):
self.device.confirmed = True
self.device.save()
@@ -526,24 +535,15 @@ class User2FADeviceConfirmTOTPView(RecentAuthenticationRequiredMixin, TemplateVi
notices = [
_('A new two-factor authentication device has been added to your account.')
]
if activate == 'on' and not self.request.user.require_2fa:
self.request.user.require_2fa = True
self.request.user.save()
self.request.user.log_action('pretix.user.settings.2fa.enabled', user=self.request.user)
notices.append(
_('Two-factor authentication has been enabled.')
)
self.request.user.send_security_notice(notices)
self.request.user.update_session_token()
update_session_auth_hash(self.request, self.request.user)
note = ''
if not self.request.user.require_2fa:
note = ' ' + str(_('Please note that you still need to enable two-factor authentication for your '
'account using the buttons below to make a second factor required for logging '
'into your account.'))
messages.success(request, str(_('The device has been verified and can now be used.')) + note)
return redirect(reverse('control:user.settings.2fa'))
messages.success(request, str(_('The device has been verified and can now be used.')))
if self.request.user.require_2fa:
return redirect(reverse('control:user.settings.2fa'))
else:
return redirect(reverse('control:user.settings.2fa.enable'))
else:
messages.error(request, _('The code you entered was not valid. If this problem persists, please check '
'that the date and time of your phone are configured correctly.'))
@@ -576,6 +576,7 @@ class User2FALeaveTeamsView(RecentAuthenticationRequiredMixin, TemplateView):
class User2FAEnableView(RecentAuthenticationRequiredMixin, TemplateView):
template_name = 'pretixcontrol/user/2fa_enable.html'
max_form_time = 7200 # this should have effectively no timeout, as the user might take some time to print out their emergency codes, and they would become invalid in case of a timeout
def dispatch(self, request, *args, **kwargs):
if not any(dt.objects.filter(user=self.request.user, confirmed=True) for dt in REAL_DEVICE_TYPES):
@@ -584,14 +585,39 @@ class User2FAEnableView(RecentAuthenticationRequiredMixin, TemplateView):
return redirect(reverse('control:user.settings.2fa'))
return super().dispatch(request, *args, **kwargs)
def get(self, request, *args, **kwargs):
new_tokens = None
try:
static_tokens_device = StaticDevice.objects.get(user=self.request.user, name='emergency')
except StaticDevice.MultipleObjectsReturned:
static_tokens_device = StaticDevice.objects.filter(
user=self.request.user, name='emergency'
).first()
except StaticDevice.DoesNotExist:
static_tokens_device = None
new_tokens = [get_random_string(length=12, allowed_chars='1234567890') for _ in range(10)]
request.session['pretix_2fa_new_emergency_tokens'] = new_tokens
return super().get(request, *args, new_emergency_tokens=new_tokens, static_tokens_device=static_tokens_device, **kwargs)
def post(self, request, *args, **kwargs):
notices = [
_('Two-factor authentication has been enabled.')
]
if 'pretix_2fa_new_emergency_tokens' in request.session:
d = StaticDevice.objects.create(user=self.request.user, name='emergency')
for code in request.session['pretix_2fa_new_emergency_tokens']:
d.token_set.create(token=code)
self.request.user.log_action('pretix.user.settings.2fa.regenemergency', user=self.request.user)
notices += [
_('Your two-factor emergency codes have been regenerated.')
]
del request.session['pretix_2fa_new_emergency_tokens']
self.request.user.require_2fa = True
self.request.user.save()
self.request.user.log_action('pretix.user.settings.2fa.enabled', user=self.request.user)
messages.success(request, _('Two-factor authentication is now enabled for your account.'))
self.request.user.send_security_notice([
_('Two-factor authentication has been enabled.')
])
self.request.user.send_security_notice(notices)
self.request.user.update_session_token()
update_session_auth_hash(self.request, self.request.user)
return redirect(reverse('control:user.settings.2fa'))

View File

@@ -19,26 +19,12 @@
# You should have received a copy of the GNU Affero General Public License along with this program. If not, see
# <https://www.gnu.org/licenses/>.
#
import ipaddress
import socket
import sys
import types
from datetime import datetime
from http import cookies
from django.conf import settings
from PIL import Image
from requests.adapters import HTTPAdapter
from urllib3.connection import HTTPConnection, HTTPSConnection
from urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool
from urllib3.exceptions import (
ConnectTimeoutError, HTTPError, LocationParseError, NameResolutionError,
NewConnectionError,
)
from urllib3.util.connection import (
_TYPE_SOCKET_OPTIONS, _set_socket_options, allowed_gai_family,
)
from urllib3.util.timeout import _DEFAULT_TIMEOUT
def monkeypatch_vobject_performance():
@@ -103,123 +89,6 @@ def monkeypatch_requests_timeout():
HTTPAdapter.send = httpadapter_send
def monkeypatch_urllib3_ssrf_protection():
"""
pretix allows HTTP requests to untrusted URLs, e.g. through webhooks or external API URLs. This is dangerous since
it can allow access to private networks that should not be reachable by users ("server-side request forgery", SSRF).
Validating URLs at submission is not sufficient, since with DNS rebinding an attacker can make a domain name pass
validation and then resolve to a private IP address on actual execution. Unfortunately, there seems no clean solution
to this in Python land, so we monkeypatch urllib3's connection management to check the IP address to be external
*after* the DNS resolution.
This does not work when a global http(s) proxy is used, but in that scenario the proxy can perform the validation.
"""
if settings.ALLOW_HTTP_TO_PRIVATE_NETWORKS:
# Settings are not supposed to change during runtime, so we can optimize performance and complexity by skipping
# this if not needed.
return
def create_connection(
address: tuple[str, int],
timeout=_DEFAULT_TIMEOUT,
source_address: tuple[str, int] | None = None,
socket_options: _TYPE_SOCKET_OPTIONS | None = None,
) -> socket.socket:
# This is copied from urllib3.util.connection v2.3.0
host, port = address
if host.startswith("["):
host = host.strip("[]")
err = None
# Using the value from allowed_gai_family() in the context of getaddrinfo lets
# us select whether to work with IPv4 DNS records, IPv6 records, or both.
# The original create_connection function always returns all records.
family = allowed_gai_family()
try:
host.encode("idna")
except UnicodeError:
raise LocationParseError(f"'{host}', label empty or too long") from None
for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM):
af, socktype, proto, canonname, sa = res
if not settings.ALLOW_HTTP_TO_PRIVATE_NETWORKS:
ip_addr = ipaddress.ip_address(sa[0])
if ip_addr.is_multicast:
raise HTTPError(f"Request to multicast address {sa[0]} blocked")
if ip_addr.is_loopback or ip_addr.is_link_local:
raise HTTPError(f"Request to local address {sa[0]} blocked")
if ip_addr.is_private:
raise HTTPError(f"Request to private address {sa[0]} blocked")
sock = None
try:
sock = socket.socket(af, socktype, proto)
# If provided, set socket level options before connecting.
_set_socket_options(sock, socket_options)
if timeout is not _DEFAULT_TIMEOUT:
sock.settimeout(timeout)
if source_address:
sock.bind(source_address)
sock.connect(sa)
# Break explicitly a reference cycle
err = None
return sock
except OSError as _:
err = _
if sock is not None:
sock.close()
if err is not None:
try:
raise err
finally:
# Break explicitly a reference cycle
err = None
else:
raise OSError("getaddrinfo returns an empty list")
class ProtectionMixin:
def _new_conn(self) -> socket.socket:
# This is 1:1 the version from urllib3.connection.HTTPConnection._new_conn v2.3.0
# just with a call to our own create_connection
try:
sock = create_connection(
(self._dns_host, self.port),
self.timeout,
source_address=self.source_address,
socket_options=self.socket_options,
)
except socket.gaierror as e:
raise NameResolutionError(self.host, self, e) from e
except socket.timeout as e:
raise ConnectTimeoutError(
self,
f"Connection to {self.host} timed out. (connect timeout={self.timeout})",
) from e
except OSError as e:
raise NewConnectionError(
self, f"Failed to establish a new connection: {e}"
) from e
sys.audit("http.client.connect", self, self.host, self.port)
return sock
class ProtectedHTTPConnection(ProtectionMixin, HTTPConnection):
pass
class ProtectedHTTPSConnection(ProtectionMixin, HTTPSConnection):
pass
HTTPConnectionPool.ConnectionCls = ProtectedHTTPConnection
HTTPSConnectionPool.ConnectionCls = ProtectedHTTPSConnection
def monkeypatch_cookie_morsel():
# See https://code.djangoproject.com/ticket/34613
cookies.Morsel._flags.add("partitioned")
@@ -230,5 +99,4 @@ def monkeypatch_all_at_ready():
monkeypatch_vobject_performance()
monkeypatch_pillow_safer()
monkeypatch_requests_timeout()
monkeypatch_urllib3_ssrf_protection()
monkeypatch_cookie_morsel()

View File

@@ -25,7 +25,7 @@ import time
from django.conf import settings
from django.contrib.auth import login as auth_login
from django.contrib.gis import geoip2
from django.contrib.gis.geoip2 import GeoIP2
from django.core.cache import cache
from django.utils.timezone import now
from django.utils.translation import gettext_lazy as _
@@ -63,20 +63,14 @@ def get_user_agent_hash(request):
_geoip = None
def get_geoip() -> geoip2.GeoIP2:
# See https://code.djangoproject.com/ticket/36988#ticket
def _get_country(request):
global _geoip
geoip2.SUPPORTED_DATABASE_TYPES.add("Geoacumen-Country")
if not _geoip:
_geoip = geoip2.GeoIP2()
return _geoip
_geoip = GeoIP2()
def _get_country(request):
try:
res = get_geoip().country(get_client_ip(request))
res = _geoip.country(get_client_ip(request))
except AddressNotFoundError:
return None
return res['country_code']

View File

@@ -576,7 +576,7 @@ def filter_subevents_with_plugins(subevents, sales_channel=None):
if not app_cache:
_populate_app_cache()
for receiver in filter_subevents._live_receivers(None)[0]:
for receiver in filter_subevents._live_receivers(None):
app = get_defining_app(receiver)
event_state = {}

View File

@@ -49,7 +49,7 @@ from django.views.decorators.cache import cache_page
from django.views.decorators.gzip import gzip_page
from django.views.decorators.http import condition
from django.views.i18n import (
JavaScriptCatalog, builtin_template_path, get_formats,
JavaScriptCatalog, get_formats, js_catalog_template,
)
from lxml import html
@@ -170,8 +170,7 @@ def generate_widget_js(version, lang):
'September', 'October', 'November', 'December'
)
catalog = dict((k, v) for k, v in catalog.items() if k.startswith('widget\u0004') or k in str_wl)
with builtin_template_path("i18n_catalog.js").open(encoding="utf-8") as fh:
template = Engine().from_string(fh.read())
template = Engine().from_string(js_catalog_template)
context = Context({
'catalog_str': indent(json.dumps(
catalog, sort_keys=True, indent=2)) if catalog else None,

View File

@@ -208,7 +208,6 @@ CSRF_TRUSTED_ORIGINS = [urlparse(SITE_URL).scheme + '://' + urlparse(SITE_URL).h
TRUST_X_FORWARDED_FOR = config.getboolean('pretix', 'trust_x_forwarded_for', fallback=False)
USE_X_FORWARDED_HOST = config.getboolean('pretix', 'trust_x_forwarded_host', fallback=False)
ALLOW_HTTP_TO_PRIVATE_NETWORKS = config.getboolean('pretix', 'allow_http_to_private_networks', fallback=False)
REQUEST_ID_HEADER = config.get('pretix', 'request_id_header', fallback=False)
@@ -535,7 +534,6 @@ X_FRAME_OPTIONS = 'DENY'
# URL settings
ROOT_URLCONF = 'pretix.multidomain.maindomain_urlconf'
FORMS_URLFIELD_ASSUME_HTTPS = True # transitional for django 6.0
WSGI_APPLICATION = 'pretix.wsgi.application'

View File

@@ -23,7 +23,9 @@ filterwarnings =
error
ignore:.*invalid escape sequence.*:
ignore:The 'warn' method is deprecated:DeprecationWarning
ignore::django.utils.deprecation.RemovedInDjango60Warning:
ignore::django.utils.deprecation.RemovedInDjango51Warning:django.core.files.storage
ignore:.*index_together.*:django.utils.deprecation.RemovedInDjango51Warning:
ignore:.*get_storage_class.*:django.utils.deprecation.RemovedInDjango51Warning:compressor
ignore:.*This signal will soon be only available for plugins that declare to be organizer-level.*:DeprecationWarning:
ignore::DeprecationWarning:mt940
ignore::DeprecationWarning:cbor2

View File

@@ -19,13 +19,14 @@
# You should have received a copy of the GNU Affero General Public License along with this program. If not, see
# <https://www.gnu.org/licenses/>.
#
from datetime import datetime, timedelta, timezone
from datetime import datetime, timedelta
import aiohttp
import pytest
import pytest_asyncio
from django.utils.timezone import now
from django_scopes import scopes_disabled
from pytz import UTC
from pretix.base.models import (
Device, Event, Item, Organizer, Quota, SeatingPlan,
@@ -52,7 +53,7 @@ def organizer():
def event(organizer):
e = Event.objects.create(
organizer=organizer, name='Dummy', slug='dummy',
date_from=datetime(2017, 12, 27, 10, 0, 0, tzinfo=timezone.utc),
date_from=datetime(2017, 12, 27, 10, 0, 0, tzinfo=UTC),
presale_end=now() + timedelta(days=300),
plugins='pretix.plugins.banktransfer,pretix.plugins.ticketoutputpdf',
is_public=True, live=True
@@ -108,8 +109,8 @@ def customer(event, membership_type):
def membership(event, membership_type, customer):
return customer.memberships.create(
membership_type=membership_type,
date_start=datetime(2017, 1, 1, 0, 0, tzinfo=timezone.utc),
date_end=datetime(2099, 1, 1, 0, 0, tzinfo=timezone.utc),
date_start=datetime(2017, 1, 1, 0, 0, tzinfo=UTC),
date_end=datetime(2099, 1, 1, 0, 0, tzinfo=UTC),
)

View File

@@ -1,93 +0,0 @@
#
# This file is part of pretix (Community Edition).
#
# Copyright (C) 2014-2020 Raphael Michel and contributors
# Copyright (C) 2020-today pretix GmbH and contributors
#
# 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 in version 3 of the License.
#
# ADDITIONAL TERMS APPLY: Pursuant to Section 7 of the GNU Affero General Public License, additional terms are
# applicable granting you additional permissions and placing additional restrictions on your usage of this software.
# Please refer to the pretix LICENSE file to obtain the full terms applicable to this work. If you did not receive
# this file, see <https://pretix.eu/about/en/license>.
#
# 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
# <https://www.gnu.org/licenses/>.
#
from socket import AF_INET, SOCK_STREAM
from unittest import mock
import pytest
import requests
from django.test import override_settings
from dns.inet import AF_INET6
from urllib3.exceptions import HTTPError
def test_local_blocked():
with pytest.raises(HTTPError, match="Request to local address.*"):
requests.get("http://localhost", timeout=0.1)
with pytest.raises(HTTPError, match="Request to local address.*"):
requests.get("https://localhost", timeout=0.1)
def test_private_ip_blocked():
with pytest.raises(HTTPError, match="Request to private address.*"):
requests.get("http://10.0.0.1", timeout=0.1)
with pytest.raises(HTTPError, match="Request to private address.*"):
requests.get("https://10.0.0.1", timeout=0.1)
@pytest.mark.django_db
@pytest.mark.parametrize("res", [
[(AF_INET, SOCK_STREAM, 6, '', ('10.0.0.3', 443))],
[(AF_INET, SOCK_STREAM, 6, '', ('0.0.0.0', 443))],
[(AF_INET, SOCK_STREAM, 6, '', ('127.1.1.1', 443))],
[(AF_INET, SOCK_STREAM, 6, '', ('192.168.5.3', 443))],
[(AF_INET, SOCK_STREAM, 6, '', ('224.0.0.1', 443))],
[(AF_INET6, SOCK_STREAM, 6, '', ('::1', 443, 0, 0))],
[(AF_INET6, SOCK_STREAM, 6, '', ('fe80::1', 443, 0, 0))],
[(AF_INET6, SOCK_STREAM, 6, '', ('ff00::1', 443, 0, 0))],
[(AF_INET6, SOCK_STREAM, 6, '', ('fc00::1', 443, 0, 0))],
])
def test_dns_resolving_to_local_blocked(res):
with mock.patch('socket.getaddrinfo') as mock_addr:
mock_addr.return_value = res
with pytest.raises(HTTPError, match="Request to (multicast|private|local) address.*"):
requests.get("https://example.org", timeout=0.1)
with pytest.raises(HTTPError, match="Request to (multicast|private|local) address.*"):
requests.get("http://example.org", timeout=0.1)
def test_dns_remote_allowed():
class SocketOk(Exception):
pass
def side_effect(*args, **kwargs):
raise SocketOk
with mock.patch('socket.getaddrinfo') as mock_addr, mock.patch('socket.socket') as mock_socket:
mock_addr.return_value = [(AF_INET, SOCK_STREAM, 6, '', ('8.8.8.8', 443))]
mock_socket.side_effect = side_effect
with pytest.raises(SocketOk):
requests.get("https://example.org", timeout=0.1)
@override_settings(ALLOW_HTTP_TO_PRIVATE_NETWORKS=True)
def test_local_is_allowed():
class SocketOk(Exception):
pass
def side_effect(*args, **kwargs):
raise SocketOk
with mock.patch('socket.getaddrinfo') as mock_addr, mock.patch('socket.socket') as mock_socket:
mock_addr.return_value = [(AF_INET, SOCK_STREAM, 6, '', ('10.0.0.1', 443))]
mock_socket.side_effect = side_effect
with pytest.raises(SocketOk):
requests.get("https://example.org", timeout=0.1)