mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-18 09:06:06 +00:00
89 lines
2.9 KiB
Python
89 lines
2.9 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):
|
|
x_forwarded_for = request.META.get("HTTP_X_REAL_IP")
|
|
if x_forwarded_for:
|
|
# X-Forwarded-For can be a comma-separated list of IPs
|
|
ip = x_forwarded_for.split(",")[0].strip()
|
|
else:
|
|
ip = request.META.get("REMOTE_ADDR")
|
|
return ip
|
|
|
|
|
|
def user_xc_ip_allowed(request, user):
|
|
"""Check request IP against per-user XC allowed ranges. Empty = allow all (0.0.0.0/0)."""
|
|
allowed_ips = getattr(user, 'xc_allowed_ips', '') or ''
|
|
if not allowed_ips.strip():
|
|
return True
|
|
|
|
cidrs = [c.strip() for c in allowed_ips.split(',') if c.strip()]
|
|
if not cidrs:
|
|
return True
|
|
|
|
client_ip = ipaddress.ip_address(get_client_ip(request))
|
|
for cidr in cidrs:
|
|
try:
|
|
if client_ip in ipaddress.ip_network(cidr, strict=False):
|
|
return True
|
|
except ValueError:
|
|
continue
|
|
return False
|
|
|
|
|
|
def network_access_allowed(request, settings_key):
|
|
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
|
|
|
|
return network_allowed
|