1import base64
2import ipaddress
3import json
4import random
5import secrets
6from collections import defaultdict
7from datetime import datetime as dt, timedelta
8from importlib import resources
9from typing import Literal, Union, List, Type
10from uuid import UUID
11
12from jdatetime import date as jd
13from v2share import (
14 V2Data,
15 SingBoxConfig,
16 ClashConfig,
17 ClashMetaConfig,
18 XrayConfig,
19)
20from v2share.base import BaseConfig
21from v2share.links import LinksConfig
22
23from app.config.env import (
24 XRAY_SUBSCRIPTION_TEMPLATE,
25 SINGBOX_SUBSCRIPTION_TEMPLATE,
26 CLASH_SUBSCRIPTION_TEMPLATE,
27 SUBSCRIPTION_PAGE_TEMPLATE,
28)
29from app.db import GetDB, crud
30from app.models.proxy import (
31 InboundHostSecurity,
32)
33from app.models.settings import SubscriptionSettings
34from app.models.user import UserResponse, UserExpireStrategy
35from app.templates import render_template
36from app.utils.keygen import gen_uuid, gen_password, generate_curve25519_pbk
37from app.utils.system import get_public_ip, readable_size
38
39SERVER_IP = get_public_ip()
40
41ACTIVITY_EMOJIS = {
42 True: "✅",
43 False: "❌",
44}
45
46subscription_handlers: dict[str, Type[BaseConfig]] = {
47 "links": LinksConfig,
48 "xray": XrayConfig,
49 "clash-meta": ClashMetaConfig,
50 "clash": ClashConfig,
51 "sing-box": SingBoxConfig,
52}
53
54handlers_templates = {
55 LinksConfig: None,
56 XrayConfig: XRAY_SUBSCRIPTION_TEMPLATE
57 or resources.files("app.templates") / "xray.json",
58 ClashConfig: CLASH_SUBSCRIPTION_TEMPLATE,
59 ClashMetaConfig: CLASH_SUBSCRIPTION_TEMPLATE,
60 SingBoxConfig: SINGBOX_SUBSCRIPTION_TEMPLATE
61 or resources.files("app.templates") / "sing-box.json",
62}
63
64
65def generate_subscription_template(
66 db_user, subscription_settings: SubscriptionSettings
67):
68 links = generate_subscription(
69 user=db_user,
70 config_format="links",
71 use_placeholder=not db_user.is_active
72 and subscription_settings.placeholder_if_disabled,
73 placeholder_remark=subscription_settings.placeholder_remark,
74 shuffle=subscription_settings.shuffle_configs,
75 ).split()
76 return render_template(
77 SUBSCRIPTION_PAGE_TEMPLATE,
78 {"user": UserResponse.model_validate(db_user), "links": links},
79 )
80
81
82def generate_subscription(
83 user: "UserResponse",
84 config_format: Literal["links", "xray", "clash-meta", "clash", "sing-box"],
85 as_base64: bool = False,
86 use_placeholder: bool = False,
87 placeholder_remark: str = "disabled",
88 shuffle: bool = False,
89) -> str:
90 extra_data = UserResponse.model_validate(user).model_dump(
91 exclude={"subscription_url", "services", "inbounds"}
92 )
93 format_variables = setup_format_variables(extra_data)
94
95 if config_format not in subscription_handlers.keys():
96 raise ValueError(f'Unsupported format "{config_format}"')
97
98 subscription_handler_class = subscription_handlers[config_format]
99 if template_path := handlers_templates[subscription_handler_class]:
100 subscription_handler = subscription_handler_class(
101 template_path=template_path
102 )
103 else:
104 subscription_handler = subscription_handler_class()
105
106 if use_placeholder:
107 placeholder_config = V2Data(
108 "vmess",
109 placeholder_remark.format_map(format_variables),
110 "127.0.0.1",
111 80,
112 )
113 configs = [placeholder_config]
114
115 else:
116 configs = generate_user_configs(
117 user.inbounds,
118 user.key,
119 user.id,
120 format_variables,
121 )
122
123 subscription_handler.add_proxies(configs)
124 config = subscription_handler.render(sort=True, shuffle=shuffle)
125
126 return (
127 config if not as_base64 else base64.b64encode(config.encode()).decode()
128 )
129
130
131def format_time_left(seconds_left: int) -> str:
132 if not seconds_left or seconds_left <= 0:
133 return "∞"
134
135 minutes, seconds = divmod(seconds_left, 60)
136 hours, minutes = divmod(minutes, 60)
137 days, hours = divmod(hours, 24)
138 months, days = divmod(days, 30)
139
140 result = []
141 if months:
142 result.append(f"{months}m")
143 if days:
144 result.append(f"{days}d")
145 if hours and (days < 7):
146 result.append(f"{hours}h")
147 if minutes and not (months or days):
148 result.append(f"{minutes}m")
149 if seconds and not (months or days):
150 result.append(f"{seconds}s")
151 return " ".join(result)
152
153
154def setup_format_variables(extra_data: dict) -> dict:
155 expire_strategy = extra_data.get("expire_strategy")
156 expire_datetime = extra_data.get("expire_date")
157 usage_duration = extra_data.get("usage_duration")
158
159 if expire_strategy != UserExpireStrategy.START_ON_FIRST_USE:
160 if expire_datetime is not None:
161 seconds_left = (expire_datetime - dt.utcnow()).total_seconds()
162 expire_date = expire_datetime.date()
163 jalali_expire_date = jd.fromgregorian(
164 year=expire_date.year,
165 month=expire_date.month,
166 day=expire_date.day,
167 ).strftime("%Y-%m-%d")
168 days_left = (expire_datetime - dt.utcnow()).days + 1
169 time_left = format_time_left(seconds_left)
170 else:
171 days_left, time_left, expire_date, jalali_expire_date = "∞" * 4
172 else:
173 if usage_duration is not None and usage_duration >= 0:
174 days_left = timedelta(seconds=usage_duration).days
175 time_left = format_time_left(usage_duration)
176 expire_date = "-"
177 jalali_expire_date = "-"
178 else:
179 days_left, time_left, expire_date, jalali_expire_date = "∞" * 4
180
181 if extra_data.get("data_limit"):
182 data_limit = readable_size(extra_data["data_limit"])
183 data_left = extra_data["data_limit"] - extra_data["used_traffic"]
184 if data_left < 0:
185 data_left = 0
186 data_left = readable_size(data_left)
187 else:
188 data_limit, data_left = "∞" * 2
189
190 status_emoji = ACTIVITY_EMOJIS.get(extra_data.get("is_active")) or ""
191
192 format_variables = defaultdict(
193 lambda: "<missing>",
194 {
195 "SERVER_IP": SERVER_IP,
196 "USERNAME": extra_data.get("username", "{USERNAME}"),
197 "DATA_USAGE": readable_size(extra_data.get("used_traffic")),
198 "DATA_LIMIT": data_limit,
199 "DATA_LEFT": data_left,
200 "DAYS_LEFT": days_left,
201 "EXPIRE_DATE": expire_date,
202 "JALALI_EXPIRE_DATE": jalali_expire_date,
203 "TIME_LEFT": time_left,
204 "STATUS_EMOJI": status_emoji,
205 },
206 )
207
208 return format_variables
209
210
211def generate_user_configs(
212 inbounds: list,
213 key: str,
214 user_id: int,
215 format_variables: dict,
216) -> Union[List, str]:
217
218 salt = secrets.token_hex(8)
219 configs = []
220
221 for inb in inbounds:
222 inb_id, inbound, protocol, tag = (
223 inb.id,
224 json.loads(inb.config),
225 inb.protocol,
226 inb.tag,
227 )
228
229 format_variables.update({"PROTOCOL": protocol.name})
230 if not inbound:
231 continue
232
233 format_variables.update(
234 {"TRANSPORT": inbound.get("network", "<missing>")}
235 )
236 with GetDB() as db:
237 hosts = crud.get_inbound_hosts(db, inb_id)
238
239 for host in hosts:
240 if host.is_disabled:
241 continue
242 host_snis = host.sni.split(",") if host.sni else []
243 sni_list = host_snis or inbound.get("sni", [])
244 if sni_list:
245 sni = random.choice(sni_list).replace("*", salt)
246 else:
247 sni = ""
248
249 host_hosts = host.host.split(",") if host.host else []
250 req_host_list = host_hosts or inbound.get("host", [])
251 if req_host_list:
252 req_host = random.choice(req_host_list).replace("*", salt)
253 else:
254 req_host = ""
255
256 host_tls = (
257 None
258 if host.security == InboundHostSecurity.inbound_default
259 else host.security.value
260 )
261 data = V2Data(
262 protocol.value,
263 host.remark.format_map(format_variables),
264 host.address.format_map(format_variables),
265 host.port or inbound["port"],
266 transport_type=inbound.get("network"),
267 sni=sni,
268 host=req_host,
269 tls=host_tls or inbound.get("tls"),
270 header_type=inbound.get("header_type"),
271 alpn=host.alpn if host.alpn != "none" else None,
272 path=(
273 host.path.format_map(format_variables)
274 if host.path
275 else inbound.get("path")
276 ),
277 fingerprint=host.fingerprint.value or inbound.get("fp"),
278 reality_pbk=inbound.get("pbk"),
279 reality_sid=inbound.get("sid"),
280 client_address=(
281 ipaddress.ip_network(inbound["address"], strict=False)[
282 user_id
283 ].compressed
284 + "/32"
285 if inbound.get("address")
286 else None
287 ),
288 flow=inbound.get("flow"),
289 allow_insecure=host.allowinsecure,
290 uuid=UUID(gen_uuid(key)),
291 password=gen_password(key),
292 ed25519=generate_curve25519_pbk(key),
293 enable_mux=host.mux,
294 http_headers={
295 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.3"
296 },
297 shadowsocks_method="chacha20-ietf-poly1305",
298 weight=host.weight,
299 )
300 if host.fragment:
301 data.fragment = True
302 data.fragment_packets = host.fragment["packets"]
303 data.fragment_length = host.fragment["length"]
304 data.fragment_interval = host.fragment["interval"]
305 configs.append(data)
306
307 return configs
308
309
310def encode_title(text: str) -> str:
311 return f"base64:{base64.b64encode(text.encode()).decode()}"