Compare commits

..

1 Commits

Author SHA1 Message Date
Raphael Michel
4142a2b690 Orders API: Fix race condition in voucher redemption (Z#23230391)
The old code relied on the `Voucher.redeemed` value obtained *before*
the lock was taken, not afterwards.

The change in services/orders.py is functionally pointless, but it makes
the pattern of "fill availability only after lock" clearer and might
avoid introducing similar bugs in the future.
2026-04-08 15:07:34 +02:00
16 changed files with 46 additions and 536 deletions

View File

@@ -769,11 +769,7 @@ class PaymentDetailsField(serializers.Field):
pp = value.payment_provider
if not pp:
return {}
try:
return pp.api_payment_details(value)
except Exception:
logger.exception("Failed to retrieve payment_details")
return {}
return pp.api_payment_details(value)
class OrderPaymentSerializer(I18nAwareModelSerializer):
@@ -1416,6 +1412,7 @@ class OrderCreateSerializer(I18nAwareModelSerializer):
qa = QuotaAvailability()
qa.queue(*[q for q, d in quota_diff_for_locking.items() if d > 0])
qa.compute()
v_avail = {}
# These are not technically correct as diff use due to the time offset applied above, so let's prevent accidental
# use further down
@@ -1445,11 +1442,13 @@ class OrderCreateSerializer(I18nAwareModelSerializer):
voucher_usage[v] += 1
if voucher_usage[v] > 0:
redeemed_in_carts = CartPosition.objects.filter(
Q(voucher=pos_data['voucher']) & Q(event=self.context['event']) & Q(expires__gte=now_dt)
).exclude(pk__in=[cp.pk for cp in delete_cps])
v_avail = v.max_usages - v.redeemed - redeemed_in_carts.count()
if v_avail < voucher_usage[v]:
if v not in v_avail:
v.refresh_from_db(fields=['redeemed'])
redeemed_in_carts = CartPosition.objects.filter(
Q(voucher=v) & Q(event=self.context['event']) & Q(expires__gte=now_dt)
).exclude(pk__in=[cp.pk for cp in delete_cps])
v_avail[v] = v.max_usages - v.redeemed - redeemed_in_carts.count()
if v_avail[v] < voucher_usage[v]:
errs[i]['voucher'] = [
'The voucher has already been used the maximum number of times.'
]

View File

@@ -19,10 +19,7 @@
# 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 logging
import smtplib
import socket
from itertools import groupby
from smtplib import SMTPResponseException
from typing import TypeVar
@@ -240,80 +237,3 @@ def base_renderers(sender, **kwargs):
def get_email_context(**kwargs):
return PlaceholderContext(**kwargs).render_all()
def create_connection(address, timeout=socket.getdefaulttimeout(),
source_address=None, *, all_errors=False):
# Taken from the python stdlib, extended with a check for local ips
host, port = address
exceptions = []
for res in socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM):
af, socktype, proto, canonname, sa = res
if not getattr(settings, "MAIL_CUSTOM_SMTP_ALLOW_PRIVATE_NETWORKS", False):
ip_addr = ipaddress.ip_address(sa[0])
if ip_addr.is_multicast:
raise socket.error(f"Request to multicast address {sa[0]} blocked")
if ip_addr.is_loopback or ip_addr.is_link_local:
raise socket.error(f"Request to local address {sa[0]} blocked")
if ip_addr.is_private:
raise socket.error(f"Request to private address {sa[0]} blocked")
sock = None
try:
sock = socket.socket(af, socktype, proto)
if timeout is not socket.getdefaulttimeout():
sock.settimeout(timeout)
if source_address:
sock.bind(source_address)
sock.connect(sa)
# Break explicitly a reference cycle
exceptions.clear()
return sock
except socket.error as exc:
if not all_errors:
exceptions.clear() # raise only the last error
exceptions.append(exc)
if sock is not None:
sock.close()
if len(exceptions):
try:
if not all_errors:
raise exceptions[0]
raise ExceptionGroup("create_connection failed", exceptions)
finally:
# Break explicitly a reference cycle
exceptions.clear()
else:
raise socket.error("getaddrinfo returns an empty list")
class CheckPrivateNetworkMixin:
# _get_socket taken 1:1 from smtplib, just with a call to our own create_connection
def _get_socket(self, host, port, timeout):
# This makes it simpler for SMTP_SSL to use the SMTP connect code
# and just alter the socket connection bit.
if timeout is not None and not timeout:
raise ValueError('Non-blocking socket (timeout=0) is not supported')
if self.debuglevel > 0:
self._print_debug('connect: to', (host, port), self.source_address)
return create_connection((host, port), timeout, self.source_address)
class SMTP(CheckPrivateNetworkMixin, smtplib.SMTP):
pass
# SMTP used here instead of mixin, because smtp.SMTP_SSL._get_socket calls super()._get_socket and then wraps this socket
# super()._get_socket needs to be our version from the mixin
class SMTP_SSL(smtplib.SMTP_SSL, SMTP): # noqa: N801
pass
class CheckPrivateNetworkSmtpBackend(EmailBackend):
@property
def connection_class(self):
return SMTP_SSL if self.use_ssl else SMTP

View File

@@ -727,8 +727,6 @@ def _check_positions(event: Event, now_dt: datetime, time_machine_now_dt: dateti
_check_date(event, time_machine_now_dt)
products_seen = Counter()
q_avail = Counter()
v_avail = Counter()
v_usages = Counter()
v_budget = {}
deleted_positions = set()
@@ -793,6 +791,9 @@ def _check_positions(event: Event, now_dt: datetime, time_machine_now_dt: dateti
shared_lock_objects=[event]
)
q_avail = Counter()
v_avail = Counter()
# Check maximum order size
limit = min(int(event.settings.max_items_per_order), settings.PRETIX_MAX_ORDER_SIZE)
if sum(1 for cp in sorted_positions if not cp.addon_to) > limit:

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 getattr(settings, "ALLOW_HTTP_TO_PRIVATE_NETWORKS", False):
# 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 getattr(settings, "ALLOW_HTTP_TO_PRIVATE_NETWORKS", False):
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

@@ -8,7 +8,7 @@ msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2026-03-30 11:22+0000\n"
"PO-Revision-Date: 2026-04-08 18:00+0000\n"
"PO-Revision-Date: 2026-03-23 21:00+0000\n"
"Last-Translator: Hijiri Umemoto <hijiri@umemoto.org>\n"
"Language-Team: Japanese <https://translate.pretix.eu/projects/pretix/pretix/"
"ja/>\n"
@@ -12939,7 +12939,7 @@ msgstr "企業名を必須にするには、請求先住所を必須にする必
#: pretix/base/settings.py:4157
#, python-brace-format
msgid "VAT-ID is not supported for \"{}\"."
msgstr "VAT-IDは「{}」に対してサポートされていません。"
msgstr ""
#: pretix/base/settings.py:4164
msgid "The last payment date cannot be before the end of presale."
@@ -26796,6 +26796,8 @@ msgid "Add a two-factor authentication device"
msgstr "2要素認証デバイスを追加してください"
#: pretix/control/templates/pretixcontrol/user/2fa_add.html:19
#, fuzzy
#| msgid "Smartphone with the Authenticator application"
msgid "Smartphone with Authenticator app"
msgstr "Authenticatorアプリを搭載したスマートフォン"
@@ -26804,20 +26806,18 @@ msgid ""
"Use your smartphone with any Time-based One-Time-Password app like freeOTP, "
"Google Authenticator or Proton Authenticator."
msgstr ""
"freeOTP、Google Authenticator、Proton Authenticator などの時間ベースの"
"ワンタイムパスワードアプリをスマートフォンでご利用ください。"
#: pretix/control/templates/pretixcontrol/user/2fa_add.html:30
#, fuzzy
#| msgid "WebAuthn-compatible hardware token (e.g. Yubikey)"
msgid "WebAuthn-compatible hardware token"
msgstr "WebAuthn対応のハードウェアトークン"
msgstr "WebAuthn対応のハードウェアトークンYubikey"
#: pretix/control/templates/pretixcontrol/user/2fa_add.html:32
msgid ""
"Use a hardware token like the Yubikey, or other biometric authentication "
"like fingerprint or face recognition."
msgstr ""
"Yubikey などのハードウェアトークンや、指紋や顔認識などの生体認証を使用してく"
"ださい。"
#: pretix/control/templates/pretixcontrol/user/2fa_confirm_totp.html:8
msgid "To set up this device, please follow the following steps:"

View File

@@ -8,7 +8,7 @@ msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2026-03-30 11:22+0000\n"
"PO-Revision-Date: 2026-04-08 18:00+0000\n"
"PO-Revision-Date: 2026-04-01 17:00+0000\n"
"Last-Translator: Ruud Hendrickx <ruud@leckxicon.eu>\n"
"Language-Team: Dutch (Belgium) <https://translate.pretix.eu/projects/pretix/"
"pretix/nl_BE/>\n"
@@ -31346,7 +31346,7 @@ msgstr "We zullen u een e-mail sturen zodra we uw betaling ontvangen hebben."
#: pretix/plugins/banktransfer/templates/pretixplugins/banktransfer/refund_export.html:7
#: pretix/plugins/banktransfer/templates/pretixplugins/banktransfer/sepa_export.html:7
msgid "Export bank transfer refunds"
msgstr "Terugbetalingen per bankoverschrijving exporteren"
msgstr ""
#: pretix/plugins/banktransfer/templates/pretixplugins/banktransfer/refund_export.html:9
#, python-format
@@ -31354,8 +31354,6 @@ msgid ""
"<strong>%(num_new)s</strong> Bank transfer refunds have been placed and are "
"not yet part of an export."
msgstr ""
"<strong>%(num_new)s</strong> terugbetalingen per bankoverschrijving zijn "
"aangemaakt en nog niet geëxporteerd."
#: pretix/plugins/banktransfer/templates/pretixplugins/banktransfer/refund_export.html:15
msgid "In test mode, your exports will only contain test mode orders."
@@ -31368,8 +31366,6 @@ msgid ""
"If you want, you can now also create these exports for multiple events "
"combined."
msgstr ""
"Als u dat wilt, kunt u deze exportbestanden nu ook voor meerdere evenementen "
"tegelijk aanmaken."
#: pretix/plugins/banktransfer/templates/pretixplugins/banktransfer/refund_export.html:22
msgid "Go to organizer-level exports"
@@ -31381,7 +31377,7 @@ msgstr "Nieuw exportbestand aanmaken"
#: pretix/plugins/banktransfer/templates/pretixplugins/banktransfer/refund_export.html:38
msgid "Aggregate transactions to the same bank account"
msgstr "Overschrijvingen naar hetzelfde rekeningnummer samenvoegen"
msgstr ""
#: pretix/plugins/banktransfer/templates/pretixplugins/banktransfer/refund_export.html:43
msgid ""

View File

@@ -83,7 +83,7 @@ class AuthenticationForm(forms.Form):
self.request = request
self.customer_cache = None
super().__init__(*args, **kwargs)
self.fields['password'].help_text = "<a target='_blank' href='{}'>{}</a>".format(
self.fields['password'].help_text = "<a href='{}'>{}</a>".format(
build_absolute_uri(False, 'presale:organizer.customer.resetpw', kwargs={
'organizer': request.organizer.slug,
}),

View File

@@ -681,6 +681,8 @@ class EventIndex(EventViewMixin, EventListMixin, CartMixin, TemplateView):
context = {}
context['list_type'] = self.request.GET.get("style", self.request.event.settings.event_list_type)
if context['list_type'] not in ("calendar", "week") and self.request.event.subevents.filter(date_from__gt=time_machine_now()).count() > 50:
if self.request.event.settings.event_list_type not in ("calendar", "week"):
self.request.event.settings.event_list_type = "calendar"
context['list_type'] = "calendar"
if context['list_type'] == "calendar":

View File

@@ -66,27 +66,22 @@ class WaitingView(EventViewMixin, FormView):
if customer else None
),
)
groups = {}
choices = []
for i in items:
if not i.allow_waitinglist:
continue
category_name = str(i.category.name) if i.category else ''
group = groups.setdefault(category_name, [])
if i.has_variations:
for v in i.available_variations:
if v.cached_availability[0] == Quota.AVAILABILITY_OK:
continue
group.append((f'{i.pk}-{v.pk}', f'{i.name} {v.value}'))
choices.append((f'{i.pk}-{v.pk}', f'{i.name} {v.value}'))
else:
if i.cached_availability[0] == Quota.AVAILABILITY_OK:
continue
group.append((f'{i.pk}', f'{i.name}'))
# Remove categories where all items were available (no waiting list choices)
return [(cat, choices) for cat, choices in groups.items() if choices]
choices.append((f'{i.pk}', f'{i.name}'))
return choices
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()

View File

@@ -530,10 +530,12 @@ class WidgetAPIProductList(EventListMixin, View):
]
if hasattr(self.request, 'event') and data['list_type'] not in ("calendar", "week"):
# only allow list-view of more than 50 subevents if ordering is by date as this can be done in the database
# only allow list-view of more than 50 subevents if ordering is by data as this can be done in the database
# ordering by name is currently not supported in database due to I18NField-JSON
ordering = self.request.event.settings.get('frontpage_subevent_ordering', default='date_ascending', as_type=str)
if ordering not in ("date_ascending", "date_descending") and self.request.event.subevents.filter(date_from__gt=now()).count() > 50:
if self.request.event.settings.event_list_type not in ("calendar", "week"):
self.request.event.settings.event_list_type = "calendar"
data['list_type'] = list_type = 'calendar'
if hasattr(self.request, 'event'):

View File

@@ -223,7 +223,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)
@@ -264,8 +263,7 @@ EMAIL_HOST_PASSWORD = config.get('mail', 'password', fallback='')
EMAIL_USE_TLS = config.getboolean('mail', 'tls', fallback=False)
EMAIL_USE_SSL = config.getboolean('mail', 'ssl', fallback=False)
EMAIL_SUBJECT_PREFIX = '[pretix] '
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_CUSTOM_SMTP_BACKEND = 'pretixbase.email.CheckPrivateNetworkSmtpBackend'
EMAIL_BACKEND = EMAIL_CUSTOM_SMTP_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_TIMEOUT = 60
ADMINS = [('Admin', n) for n in config.get('mail', 'admins', fallback='').split(",") if n]

View File

@@ -1835,9 +1835,10 @@
"integrity": "sha512-XpNj6GDQzdfW+r2Wnn7xiSAd7TM3jzkxGXBGTtWKuSXv1xUV+azxAm8jdWZN06QTQk+2N2XB9jRDkvbmQmcRtg=="
},
"node_modules/brace-expansion": {
"version": "1.1.13",
"resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.13.tgz",
"integrity": "sha512-9ZLprWS6EENmhEOpjCYW2c8VkmOvckIJZfkr7rBW6dObmfgJ/L1GpSYW5Hpo9lDz4D1+n0Ckz8rU7FwHDQiG/w==",
"version": "1.1.12",
"resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.12.tgz",
"integrity": "sha512-9T9UjW3r0UW5c1Q7GTwllptXwhvYmEzFhzMfZ9H7FQWt+uZePjZPjBP/W1ZEyZ1twGWom5/56TF4lPcqjnDHcg==",
"license": "MIT",
"optional": true,
"dependencies": {
"balanced-match": "^1.0.0",
@@ -2878,9 +2879,9 @@
"integrity": "sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA=="
},
"node_modules/picomatch": {
"version": "2.3.2",
"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-2.3.2.tgz",
"integrity": "sha512-V7+vQEJ06Z+c5tSye8S+nHUfI51xoXIXjHQ99cQtKUkQqqO1kO/KCJUfZXuB47h/YBlDhah2H3hdUGXn8ie0oA==",
"version": "2.3.1",
"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-2.3.1.tgz",
"integrity": "sha512-JU3teHTNjmE2VCGFzuY8EXzCDVwEqB2a8fsIvwaStHhAWJEeVd1o1QD80CU6+ZdEXXSLbSsuLwJjkCBWqRQUVA==",
"engines": {
"node": ">=8.6"
},
@@ -4935,9 +4936,9 @@
"integrity": "sha512-XpNj6GDQzdfW+r2Wnn7xiSAd7TM3jzkxGXBGTtWKuSXv1xUV+azxAm8jdWZN06QTQk+2N2XB9jRDkvbmQmcRtg=="
},
"brace-expansion": {
"version": "1.1.13",
"resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.13.tgz",
"integrity": "sha512-9ZLprWS6EENmhEOpjCYW2c8VkmOvckIJZfkr7rBW6dObmfgJ/L1GpSYW5Hpo9lDz4D1+n0Ckz8rU7FwHDQiG/w==",
"version": "1.1.12",
"resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.12.tgz",
"integrity": "sha512-9T9UjW3r0UW5c1Q7GTwllptXwhvYmEzFhzMfZ9H7FQWt+uZePjZPjBP/W1ZEyZ1twGWom5/56TF4lPcqjnDHcg==",
"optional": true,
"requires": {
"balanced-match": "^1.0.0",
@@ -5714,9 +5715,9 @@
"integrity": "sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA=="
},
"picomatch": {
"version": "2.3.2",
"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-2.3.2.tgz",
"integrity": "sha512-V7+vQEJ06Z+c5tSye8S+nHUfI51xoXIXjHQ99cQtKUkQqqO1kO/KCJUfZXuB47h/YBlDhah2H3hdUGXn8ie0oA=="
"version": "2.3.1",
"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-2.3.1.tgz",
"integrity": "sha512-JU3teHTNjmE2VCGFzuY8EXzCDVwEqB2a8fsIvwaStHhAWJEeVd1o1QD80CU6+ZdEXXSLbSsuLwJjkCBWqRQUVA=="
},
"pify": {
"version": "4.0.1",

View File

@@ -94,9 +94,6 @@ class DisableMigrations(object):
def __getitem__(self, item):
return None
def setdefault(self, key, default=None):
return
if not os.environ.get("GITHUB_WORKFLOW", ""):
MIGRATION_MODULES = DisableMigrations()

View File

@@ -35,11 +35,8 @@
import datetime
import os
import re
import socket
from contextlib import contextmanager
from decimal import Decimal
from email.mime.text import MIMEText
from unittest import mock
import pytest
from django.conf import settings
@@ -594,117 +591,3 @@ def test_attached_ical_localization(env, order):
assert len(djmail.outbox) == 1
assert len(djmail.outbox[0].attachments) == 1
assert description in djmail.outbox[0].attachments[0][1]
PRIVATE_IPS_RES = [
[(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('10.0.0.3', 443))],
[(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('0.0.0.0', 443))],
[(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('127.1.1.1', 443))],
[(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('192.168.5.3', 443))],
[(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('224.0.0.1', 443))],
[(socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('::1', 443, 0, 0))],
[(socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('fe80::1', 443, 0, 0))],
[(socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('ff00::1', 443, 0, 0))],
[(socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('fc00::1', 443, 0, 0))],
]
@contextmanager
def assert_mail_connection(res, should_connect, use_ssl):
with (
mock.patch('socket.socket') as mock_socket,
mock.patch('socket.getaddrinfo', return_value=res),
mock.patch('smtplib.SMTP.getreply', return_value=(220, "")),
mock.patch('smtplib.SMTP.sendmail'),
mock.patch('ssl.SSLContext.wrap_socket') as mock_ssl
):
yield
if should_connect:
mock_socket.assert_called_once()
mock_socket.return_value.connect.assert_called_once_with(res[0][-1])
if use_ssl:
mock_ssl.assert_called_once()
else:
mock_socket.assert_not_called()
mock_socket.return_value.connect.assert_not_called()
mock_ssl.assert_not_called()
@pytest.mark.parametrize("res", PRIVATE_IPS_RES)
@pytest.mark.parametrize("use_ssl", [
True, False
])
def test_private_smtp_ip(res, use_ssl, settings):
settings.EMAIL_CUSTOM_SMTP_BACKEND = 'pretix.base.email.CheckPrivateNetworkSmtpBackend'
settings.MAIL_CUSTOM_SMTP_ALLOW_PRIVATE_NETWORKS = False
with assert_mail_connection(res=res, should_connect=False, use_ssl=use_ssl), pytest.raises(match="Request to .* blocked"):
connection = djmail.get_connection(backend=settings.EMAIL_CUSTOM_SMTP_BACKEND,
host="localhost",
use_ssl=use_ssl)
connection.open()
settings.MAIL_CUSTOM_SMTP_ALLOW_PRIVATE_NETWORKS = True
with assert_mail_connection(res=res, should_connect=True, use_ssl=use_ssl):
connection = djmail.get_connection(backend=settings.EMAIL_CUSTOM_SMTP_BACKEND,
host="localhost",
use_ssl=use_ssl)
connection.open()
@pytest.mark.parametrize("use_ssl", [
True, False
])
@pytest.mark.parametrize("allow_private", [
True, False
])
def test_public_smtp_ip(use_ssl, allow_private, settings):
settings.EMAIL_CUSTOM_SMTP_BACKEND = 'pretix.base.email.CheckPrivateNetworkSmtpBackend'
settings.MAIL_CUSTOM_SMTP_ALLOW_PRIVATE_NETWORKS = allow_private
with assert_mail_connection(res=[(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('8.8.8.8', 443))], should_connect=True, use_ssl=use_ssl):
connection = djmail.get_connection(backend=settings.EMAIL_CUSTOM_SMTP_BACKEND,
host="localhost",
use_ssl=use_ssl)
connection.open()
@pytest.mark.django_db
@pytest.mark.parametrize("use_ssl", [
True, False
])
@pytest.mark.parametrize("allow_private_networks", [
True, False
])
@pytest.mark.parametrize("res", PRIVATE_IPS_RES)
def test_send_mail_private_ip(res, use_ssl, allow_private_networks, env):
settings.EMAIL_CUSTOM_SMTP_BACKEND = 'pretix.base.email.CheckPrivateNetworkSmtpBackend'
settings.MAIL_CUSTOM_SMTP_ALLOW_PRIVATE_NETWORKS = allow_private_networks
event, user, organizer = env
event.settings.smtp_use_custom = True
event.settings.smtp_host = "example.com"
event.settings.smtp_use_ssl = use_ssl
event.settings.smtp_use_tls = False
def send_mail():
m = OutgoingMail.objects.create(
to=['recipient@example.com'],
subject='Test',
body_plain='Test',
sender='sender@example.com',
event=event
)
assert m.status == OutgoingMail.STATUS_QUEUED
mail_send_task.apply(kwargs={
'outgoing_mail': m.pk,
}, max_retries=0)
m.refresh_from_db()
return m
with assert_mail_connection(res=res, should_connect=allow_private_networks, use_ssl=use_ssl):
m = send_mail()
if allow_private_networks:
assert m.status == OutgoingMail.STATUS_SENT
else:
assert m.status == OutgoingMail.STATUS_FAILED

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)

View File

@@ -1162,65 +1162,6 @@ class WaitingListTest(EventTestMixin, SoupTest):
assert wle.voucher is None
assert wle.locale == 'en'
def test_initial_selection(self):
with scopes_disabled():
cat = ItemCategory.objects.create(event=self.event, name='Tickets')
self.item.category = cat
self.item.save()
item2 = Item.objects.create(
event=self.event, name='VIP ticket',
default_price=Decimal('25.00'),
active=True, category=cat,
)
self.q.items.add(item2)
response = self.client.get(
'/%s/%s/waitinglist/?item=%d' % (
self.orga.slug, self.event.slug, item2.pk
)
)
self.assertEqual(response.status_code, 200)
doc = BeautifulSoup(response.render().content, "lxml")
select = doc.find('select', {'name': 'itemvar'})
optgroup = select.find('optgroup')
self.assertIsNotNone(optgroup, 'Choices should be grouped by category')
self.assertEqual(optgroup['label'], 'Tickets')
selected = select.find_all('option', selected=True)
self.assertEqual(len(selected), 1, 'Exactly one option should be pre-selected')
self.assertEqual(selected[0]['value'], str(item2.pk))
def test_initial_selection_with_variation(self):
with scopes_disabled():
cat = ItemCategory.objects.create(event=self.event, name='Tickets')
self.item.category = cat
self.item.has_variations = True
self.item.save()
var1 = ItemVariation.objects.create(item=self.item, value='Standard')
var2 = ItemVariation.objects.create(item=self.item, value='Premium')
self.q.variations.add(var1, var2)
response = self.client.get(
'/%s/%s/waitinglist/?item=%d&var=%d' % (
self.orga.slug, self.event.slug,
self.item.pk, var2.pk,
)
)
self.assertEqual(response.status_code, 200)
doc = BeautifulSoup(response.render().content, "lxml")
select = doc.find('select', {'name': 'itemvar'})
optgroup = select.find('optgroup')
self.assertIsNotNone(optgroup, 'Choices should be grouped by category')
self.assertEqual(optgroup['label'], 'Tickets')
selected = select.find_all('option', selected=True)
self.assertEqual(len(selected), 1, 'Exactly one option should be pre-selected')
self.assertEqual(selected[0]['value'], '%d-%d' % (self.item.pk, var2.pk))
def test_subevent_valid(self):
with scopes_disabled():
self.event.has_subevents = True