looooots of updates for user-management, initial commit of access control

dispatcharr/urls.py

@@ -7,7 +7,7 @@ from rest_framework import permissions
 from drf_yasg.views import get_schema_view
 from drf_yasg import openapi
 from .routing import websocket_urlpatterns
-from apps.output.views import xc_player_api, xc_get, xc_xmltv
+from apps.output.views import xc_player_api, xc_panel_api, xc_get, xc_xmltv
 from apps.proxy.ts_proxy.views import stream_xc
 
 # Define schema_view for Swagger
@@ -40,15 +40,21 @@ urlpatterns = [
     # Add proxy apps - Move these before the catch-all
     path("proxy/", include(("apps.proxy.urls", "proxy"), namespace="proxy")),
     path("proxy", RedirectView.as_view(url="/proxy/", permanent=True)),
+    # xc
+    re_path("player_api.php", xc_player_api, name="xc_player_api"),
+    re_path("panel_api.php", xc_panel_api, name="xc_panel_api"),
+    re_path("get.php", xc_get, name="xc_get"),
+    re_path("xmltv.php", xc_xmltv, name="xc_xmltv"),
     path(
-        "<slug:username>/<slug:password>/<int:channel_id>",
+        "live/<str:username>/<str:password>/<str:channel_id>",
+        stream_xc,
+        name="xc_live_stream_endpoint",
+    ),
+    path(
+        "<str:username>/<str:password>/<str:channel_id>",
         stream_xc,
         name="xc_stream_endpoint",
     ),
-    # xc
-    re_path("player_api.php", xc_player_api, name="xc_get"),
-    re_path("get.php", xc_get, name="xc_get"),
-    re_path("xmltv.php", xc_xmltv, name="xc_xmltv"),
     # Swagger UI
     path(
         "swagger/",

dispatcharr/utils.py

@@ -1,23 +1,58 @@
 # 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
+
 
 def json_error_response(message, status=400):
     """Return a standardized error JSON response."""
-    return JsonResponse({'success': False, 'error': message}, status=status)
+    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}
+    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']
+    valid_mime_types = ["image/jpeg", "image/png", "image/gif"]
     if file.content_type not in valid_mime_types:
-        raise ValidationError('Unsupported file type. Allowed types: JPEG, PNG, GIF.')
+        raise ValidationError("Unsupported file type. Allowed types: JPEG, PNG, GIF.")
     if file.size > 2 * 1024 * 1024:
-        raise ValidationError('File too large. Max 2MB.')
+        raise ValidationError("File too large. Max 2MB.")
 
+
+def get_client_ip(request):
+    x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
+    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 network_access_allowed(request, settings_key):
+    network_access = json.loads(CoreSettings.objects.get(key=NETWORK_ACCESS).value)
+
+    cidrs = (
+        network_access[settings_key].split(",")
+        if settings_key in network_access
+        else "0.0.0.0/0"
+    )
+
+    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