Dispatcharr/dispatcharr/utils.py
SergeantPanda 84fbbfa3c4
Some checks are pending
CI Pipeline / prepare (push) Waiting to run
CI Pipeline / docker (amd64, ubuntu-24.04) (push) Blocked by required conditions
CI Pipeline / docker (arm64, ubuntu-24.04-arm) (push) Blocked by required conditions
CI Pipeline / create-manifest (push) Blocked by required conditions
Build and Push Multi-Arch Docker Image / build-and-push (push) Waiting to run
refactor: Clean up get_client_ip() function in utils.py
2026-05-09 09:48:53 -05:00

78 lines
2.7 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, NETWORK_ACCESS_KEY
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):
try:
network_access = CoreSettings.objects.get(key=NETWORK_ACCESS_KEY).value
except CoreSettings.DoesNotExist:
network_access = {}
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