clash: only support rule-set now

apps/api/urls.py

@@ -12,6 +12,16 @@
         views.ClashProxyProviderView.as_view(),
         name="proxy_providers",
     ),
+    path(
+        "subscribe/clash/direct_domain_rule_set/",
+        views.ClashDirectDomainRuleSetView.as_view(),
+        name="direct_domain_rule_set",
+    ),
+    path(
+        "subscribe/clash/direct_ip_rule_set/",
+        views.ClashDirectIPRuleSetView.as_view(),
+        name="direct_domain_rule_set",
+    ),
     path("shop/", views.purchase, name="purchase"),
     path("change/theme/", views.change_theme, name="change_theme"),
     path("checkin/", views.UserCheckInView.as_view(), name="checkin"),

apps/api/views.py

@@ -5,6 +5,7 @@
 from django.conf import settings
 from django.contrib.auth.decorators import login_required, permission_required
 from django.http import HttpResponse, HttpResponseBadRequest, JsonResponse
+from django.shortcuts import render
 from django.utils.decorators import method_decorator
 from django.views import View
 from django.views.decorators.csrf import csrf_exempt
@@ -22,6 +23,7 @@
     get_client_ip,
     get_current_datetime,
     handle_json_request,
+    is_ip_address,
     traffic_format,
 )
 
@@ -63,66 +65,113 @@ def post(self, request):
         return JsonResponse(data)
 
 
-class SubscribeView(View):
-    def get(self, request):
-        user = None
+class UserNodeBaseView(View):
+    def get_user_and_nodes(self, request):
         if uid := request.GET.get("uid"):
-            # check if uid is valid
             try:
                 uuid.UUID(uid)
             except ValueError:
-                return HttpResponseBadRequest("invalid uid")
+                return None, HttpResponseBadRequest("invalid uid")
         else:
-            return HttpResponseBadRequest("uid is required")
+            return None, HttpResponseBadRequest("uid is required")
+
         user = User.objects.filter(uid=uid).first()
         if not user:
-            return HttpResponseBadRequest("user not found")
-        node_list = m.ProxyNode.get_user_active_nodes(user)
-
-        if protocol := request.GET.get("protocol"):
-            if protocol in m.ProxyNode.NODE_TYPE_SET:
-                node_list = node_list.filter(node_type=protocol)
+            return None, HttpResponseBadRequest("user not found")
 
+        node_list = m.ProxyNode.get_user_active_nodes(user)
         if len(node_list) == 0:
-            return HttpResponseBadRequest("no active nodes for you")
-
-        sub_client = request.GET.get("client", UserSubManager.CLIENT_CLASH)
-        sub_info = UserSubManager(user, sub_client, node_list).get_sub_info()
-        return HttpResponse(
-            sub_info,
-            content_type="text/plain; charset=utf-8",
-            headers=user.get_sub_info_header(
-                for_android=sub_client != UserSubManager.CLIENT_SHADOWROCKET
-            ),
-        )
+            return None, HttpResponseBadRequest("no active nodes for you")
+
+        return user, node_list
 
 
-class ClashProxyProviderView(View):
+class SubscribeView(UserNodeBaseView):
     def get(self, request):
-        user = None
-        if uid := request.GET.get("uid"):
-            # check if uid is valid
+        user, response_or_nodes = self.get_user_and_nodes(request)
+        if response_or_nodes is not HttpResponse:
+            node_list = response_or_nodes
+
+            if protocol := request.GET.get("protocol"):
+                if protocol in m.ProxyNode.NODE_TYPE_SET:
+                    node_list = node_list.filter(node_type=protocol)
+
+            sub_client = request.GET.get("client")
             try:
-                uuid.UUID(uid)
-            except ValueError:
-                return HttpResponseBadRequest("invalid uid")
+                sub_info = UserSubManager(user, node_list, sub_client).get_sub_info()
+            except ValueError as e:
+                return HttpResponseBadRequest(str(e))
+            return HttpResponse(
+                sub_info,
+                content_type="text/plain; charset=utf-8",
+                headers=user.get_sub_info_header(
+                    for_android=sub_client != UserSubManager.CLIENT_SHADOWROCKET
+                ),
+            )
         else:
-            return HttpResponseBadRequest("uid is required")
-        user = User.objects.filter(uid=uid).first()
-        if not user:
-            return HttpResponseBadRequest("user not found")
-        node_list = m.ProxyNode.get_user_active_nodes(user)
-        if len(node_list) == 0:
-            return HttpResponseBadRequest("no active nodes for you")
+            return response_or_nodes
 
-        providers = UserSubManager(
-            user, request.GET.get("sub_type"), node_list
-        ).get_clash_proxy_providers()
 
-        return HttpResponse(
-            providers,
-            content_type="text/plain; charset=utf-8",
-        )
+class ClashProxyProviderView(UserNodeBaseView):
+    def get(self, request):
+        user, response_or_nodes = self.get_user_and_nodes(request)
+        if response_or_nodes is not HttpResponse:
+            node_list = response_or_nodes
+            providers = UserSubManager(user, node_list).get_clash_proxy_providers()
+            return HttpResponse(
+                providers,
+                content_type="text/plain; charset=utf-8",
+            )
+        else:
+            return response_or_nodes
+
+
+class ClashDirectRuleSetBaseView(UserNodeBaseView):
+    def get_rule_set(self, node_list, is_ip: bool):
+        rule_set = set()
+        for node in node_list:
+            if node.enable_relay:
+                for rule in node.get_enabled_relay_rules():
+                    if is_ip == is_ip_address(rule.relay_host):
+                        rule_set.add(rule.relay_host)
+            if node.enable_direct:
+                if is_ip == is_ip_address(node.server):
+                    rule_set.add(node.server)
+        return sorted(rule_set)
+
+
+class ClashDirectDomainRuleSetView(ClashDirectRuleSetBaseView):
+    def get(self, request):
+        _, response_or_nodes = self.get_user_and_nodes(request)
+        if response_or_nodes is not HttpResponse:
+            node_list = response_or_nodes
+            domain_list = self.get_rule_set(node_list, is_ip=False)
+            context = {"domain_list": domain_list}
+            return render(
+                request,
+                "clash/direct_domain.yaml",
+                context=context,
+                content_type="text/plain; charset=utf-8",
+            )
+        else:
+            return response_or_nodes
+
+
+class ClashDirectIPRuleSetView(ClashDirectRuleSetBaseView):
+    def get(self, request):
+        _, response_or_nodes = self.get_user_and_nodes(request)
+        if response_or_nodes is not HttpResponse:
+            node_list = response_or_nodes
+            ip_list = self.get_rule_set(node_list, is_ip=True)
+            context = {"ip_list": ip_list}
+            return render(
+                request,
+                "clash/direct_ip.yaml",
+                context=context,
+                content_type="text/plain; charset=utf-8",
+            )
+        else:
+            return response_or_nodes
 
 
 class UserRefChartView(View):

apps/sspanel/models.py

@@ -255,6 +255,22 @@ def clash_proxy_provider_endpoint(self):
             + f"/api/subscribe/clash/proxy_providers/?{urlencode(params)}"
         )
 
+    @property
+    def direct_ip_rule_set_endpoint(self):
+        params = {"uid": self.uid}
+        return (
+            settings.SITE_HOST
+            + f"/api/subscribe/clash/direct_ip_rule_set/?{urlencode(params)}"
+        )
+
+    @property
+    def direct_domain_rule_set_endpoint(self):
+        params = {"uid": self.uid}
+        return (
+            settings.SITE_HOST
+            + f"/api/subscribe/clash/direct_domain_rule_set/?{urlencode(params)}"
+        )
+
     def update_proxy_config_from_dict(self, data):
         clean_fields = ["proxy_password"]
         for k, v in data.items():

apps/sub.py

@@ -3,39 +3,47 @@
 from django.conf import settings
 from django.template.loader import render_to_string
 
-from apps.utils import get_clash_direct_rule
+from apps.sspanel.models import User
 
 
 class UserSubManager:
     """统一管理用户的订阅"""
 
     CLIENT_SHADOWROCKET = "shadowrocket"
     CLIENT_CLASH = "clash"
-    CLIENT_CLASH_PREMIUM = "clash_premium"
     CLIENT_CLASH_PROXY_PROVIDER = "clash_proxy_provider"
 
     CLIENT_SET = {
         CLIENT_SHADOWROCKET,
         CLIENT_CLASH,
-        CLIENT_CLASH_PREMIUM,
         CLIENT_CLASH_PROXY_PROVIDER,
     }
 
-    def __init__(self, user, sub_client, node_list):
+    def __init__(self, user, node_list, sub_client=CLIENT_CLASH):
         self.user = user
-        if sub_client not in self.CLIENT_SET:
-            sub_client = self.CLIENT_CLASH_PROXY_PROVIDER
-        self.sub_client = sub_client
+        if sub_client in self.CLIENT_SET:
+            self.sub_client = sub_client
+        elif not sub_client or "clash" in sub_client:
+            self.sub_client = self.CLIENT_CLASH
+        else:
+            raise ValueError(f"sub_client {sub_client} not support")
+
         self.node_list = node_list
 
     def _get_clash_sub_yaml(self):
+        user: User = self.user
+        user_proxy_provider_url = user.clash_proxy_provider_endpoint
+        direct_ip_rule_set_url = user.direct_ip_rule_set_endpoint
+        direct_domain_rule_set_url = user.direct_domain_rule_set_endpoint
+
         return render_to_string(
             "clash/main.yaml",
             {
                 "sub_client": self.sub_client,
                 "provider_name": settings.SITE_TITLE,
-                "proxy_provider_url": self.user.clash_proxy_provider_endpoint,
-                "direct_rules": self.get_clash_direct_rules(),
+                "proxy_provider_url": user_proxy_provider_url,
+                "direct_ip_rule_set_url": direct_ip_rule_set_url,
+                "direct_domain_rule_set_url": direct_domain_rule_set_url,
             },
         )
 
@@ -60,12 +68,14 @@ def _get_shadowrocket_sub_links(self):
         return sub_links
 
     def get_sub_info(self):
-        if self.sub_client in [self.CLIENT_CLASH, self.CLIENT_CLASH_PREMIUM]:
+        if self.sub_client == self.CLIENT_CLASH:
             return self._get_clash_sub_yaml()
         elif self.sub_client == self.CLIENT_SHADOWROCKET:
             return self._get_shadowrocket_sub_links()
-        else:
+        elif self.sub_client == self.CLIENT_CLASH_PROXY_PROVIDER:
             return self.get_clash_proxy_providers()
+        else:
+            raise ValueError(f"sub_client {self.sub_client} not support")
 
     def get_clash_proxy_providers(self):
         """todo support multi provider group"""
@@ -95,13 +105,3 @@ def get_clash_proxy_providers(self):
             "clash/providers.yaml",
             {"nodes": sorted(node_configs, key=lambda x: x["name"])},
         )
-
-    def get_clash_direct_rules(self):
-        rules = set()
-        for node in self.node_list:
-            if node.enable_relay:
-                for rule in node.get_enabled_relay_rules():
-                    rules.add(get_clash_direct_rule(rule.relay_host))
-            if node.enable_direct:
-                rules.add(get_clash_direct_rule(node.server))
-        return sorted(list(rules))

apps/utils.py

@@ -96,14 +96,9 @@ def get_client_ip(request):
         return request.META.get("REMOTE_ADDR")
 
 
-def get_clash_direct_rule(addr):
+def is_ip_address(addr):
     ip_pattern = r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b"
-
-    # 检查输入字符串是否匹配 IP 地址正则表达式
-    if re.match(ip_pattern, addr):
-        return f"IP-CIDR,{addr}/32,DIRECT"
-    else:
-        return f"DOMAIN,{addr},DIRECT"
+    return bool(re.match(ip_pattern, addr))
 
 
 class BytesToGigabytesField(forms.CharField):

templates/clash/direct_domain.yaml

@@ -0,0 +1,5 @@
+payload:
+    - "clash.razord.top"
+    - "yacd.haishan.me"
+    {% autoescape off %}{% for doamin in domain_list %}
+    - '{{ doamin }}'{% endfor %}{% endautoescape %}

templates/clash/direct_ip.yaml

@@ -0,0 +1,3 @@
+payload:
+{% autoescape off %}{% for ip in ip_list %}
+- '{{ ip }}/32'{% endfor %}{% endautoescape %}