Dispatcharr/dispatcharr/utils.py
SergeantPanda 2b62928b42 feat(settings): implement Redis caching for CoreSettings and enhance network access handling
This update introduces Redis caching for grouped settings in the CoreSettings model, improving performance by reducing database queries. A new method, `get_network_access_settings`, is added to streamline access to network settings. Additionally, the `setting_flag_enabled` function is introduced to handle boolean settings more robustly. The cache is invalidated on save and delete operations, ensuring data consistency. Tests have been added to validate the caching behavior and the new settings functionality.
2026-07-18 17:33:02 +00:00

75 lines
2.6 KiB
Python

# dispatcharr/utils.py
import json
import ipaddress
from django.http import JsonResponse
from django.core.exceptions import ValidationError
from core.models import CoreSettings
def json_error_response(message, status=400):
"""Return a standardized error JSON response."""
return JsonResponse({"success": False, "error": message}, status=status)
def json_success_response(data=None, status=200):
"""Return a standardized success JSON response."""
response = {"success": True}
if data is not None:
response.update(data)
return JsonResponse(response, status=status)
def validate_logo_file(file):
"""Validate uploaded logo file size and MIME type."""
valid_mime_types = ["image/jpeg", "image/png", "image/gif", "image/webp", "image/svg+xml"]
if file.content_type not in valid_mime_types:
raise ValidationError("Unsupported file type. Allowed types: JPEG, PNG, GIF, WebP, SVG.")
if file.size > 5 * 1024 * 1024: # 5MB
raise ValidationError("File too large. Max 5MB.")
def get_client_ip(request):
return request.META.get("HTTP_X_REAL_IP") or request.META.get("REMOTE_ADDR")
def network_access_allowed(request, settings_key, user=None):
network_access = CoreSettings.get_network_access_settings()
local_cidrs = ["127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "::1/128", "fc00::/7", "fe80::/10"]
# Set defaults based on endpoint type
if settings_key == "M3U_EPG":
# M3U/EPG endpoints: local IPv4 and IPv6 only by default
default_cidrs = local_cidrs
else:
# Other endpoints: allow all by default
default_cidrs = ["0.0.0.0/0", "::/0"]
cidrs = (
network_access[settings_key].split(",")
if settings_key in network_access
else default_cidrs
)
network_allowed = False
client_ip = ipaddress.ip_address(get_client_ip(request))
for cidr in cidrs:
network = ipaddress.ip_network(cidr)
if client_ip in network:
network_allowed = True
break
if not network_allowed:
return False
if user is not None:
user_networks = (getattr(user, 'custom_properties', None) or {}).get('allowed_networks', {})
raw = user_networks.get(settings_key, '')
if raw:
for cidr in (c.strip() for c in raw.split(',') if c.strip()):
try:
if client_ip in ipaddress.ip_network(cidr, strict=False):
return True
except ValueError:
continue
return False
return True