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 WireGuardConfig,
20)
21from v2share.base import BaseConfig
22from v2share.data import MuxCoolSettings as V2MuxCoolSettings
23from v2share.data import MuxSettings as V2MuxSettings
24from v2share.data import SingBoxMuxSettings as V2SingBoxMuxSettings
25from v2share.data import SplitHttpSettings as V2SplitHttpSettings
26from v2share.data import XMuxSettings as V2XMuxSettings
27from v2share.data import XrayNoise
28from v2share.links import LinksConfig
29
30from app.config.env import (
31 XRAY_SUBSCRIPTION_TEMPLATE,
32 SINGBOX_SUBSCRIPTION_TEMPLATE,
33 CLASH_SUBSCRIPTION_TEMPLATE,
34 SUBSCRIPTION_PAGE_TEMPLATE,
35)
36from app.db import GetDB
37from app.db.crud import get_hosts_for_user
38from app.models.proxy import (
39 InboundHostSecurity,
40 SplitHttpSettings,
41 MuxSettings,
42)
43from app.models.settings import SubscriptionSettings
44from app.models.user import UserResponse, UserExpireStrategy
45from app.templates import render_template
46from app.utils.keygen import gen_uuid, gen_password, generate_curve25519_pbk
47from app.utils.system import get_public_ip, readable_size
48
49SERVER_IP = get_public_ip()
50
51ACTIVITY_EMOJIS = {
52 True: "✅",
53 False: "❌",
54}
55
56subscription_handlers: dict[str, Type[BaseConfig]] = {
57 "links": LinksConfig,
58 "xray": XrayConfig,
59 "clash-meta": ClashMetaConfig,
60 "clash": ClashConfig,
61 "sing-box": SingBoxConfig,
62 "wireguard": WireGuardConfig,
63}
64
65handlers_templates = {
66 LinksConfig: None,
67 WireGuardConfig: None,
68 XrayConfig: XRAY_SUBSCRIPTION_TEMPLATE
69 or resources.files("app.templates") / "xray.json",
70 ClashConfig: CLASH_SUBSCRIPTION_TEMPLATE,
71 ClashMetaConfig: CLASH_SUBSCRIPTION_TEMPLATE,
72 SingBoxConfig: SINGBOX_SUBSCRIPTION_TEMPLATE
73 or resources.files("app.templates") / "sing-box.json",
74}
75
76
77def generate_subscription_template(
78 db_user, subscription_settings: SubscriptionSettings
79):
80 links = generate_subscription(
81 user=db_user,
82 config_format="links",
83 use_placeholder=not db_user.is_active
84 and subscription_settings.placeholder_if_disabled,
85 placeholder_remark=subscription_settings.placeholder_remark,
86 shuffle=subscription_settings.shuffle_configs,
87 ).split()
88 return render_template(
89 SUBSCRIPTION_PAGE_TEMPLATE,
90 {"user": UserResponse.model_validate(db_user), "links": links},
91 )
92
93
94def generate_subscription(
95 user: "UserResponse",
96 config_format: Literal["links", "xray", "clash-meta", "clash", "sing-box"],
97 as_base64: bool = False,
98 use_placeholder: bool = False,
99 placeholder_remark: str = "disabled",
100 shuffle: bool = False,
101) -> str:
102 extra_data = UserResponse.model_validate(user).model_dump(
103 exclude={"subscription_url", "services", "inbounds"}
104 )
105 format_variables = setup_format_variables(extra_data)
106
107 if config_format not in subscription_handlers.keys():
108 raise ValueError(f'Unsupported format "{config_format}"')
109
110 subscription_handler_class = subscription_handlers[config_format]
111 if template_path := handlers_templates[subscription_handler_class]:
112 subscription_handler = subscription_handler_class(
113 template_path=template_path
114 )
115 else:
116 subscription_handler = subscription_handler_class()
117
118 if use_placeholder:
119 placeholder_config = V2Data(
120 "vmess",
121 placeholder_remark.format_map(format_variables),
122 "127.0.0.1",
123 80,
124 )
125 configs = [placeholder_config]
126
127 else:
128 configs = generate_user_configs(
129 user.inbounds,
130 user.key,
131 user.id,
132 format_variables,
133 chaining_support=subscription_handler.chaining_support,
134 )
135
136 subscription_handler.add_proxies(configs)
137 config = subscription_handler.render(sort=True, shuffle=shuffle)
138
139 return (
140 config if not as_base64 else base64.b64encode(config.encode()).decode()
141 )
142
143
144def format_time_left(seconds_left: int) -> str:
145 if not seconds_left or seconds_left <= 0:
146 return "∞"
147
148 minutes, seconds = divmod(seconds_left, 60)
149 hours, minutes = divmod(minutes, 60)
150 days, hours = divmod(hours, 24)
151 months, days = divmod(days, 30)
152
153 result = []
154 if months:
155 result.append(f"{months}m")
156 if days:
157 result.append(f"{days}d")
158 if hours and (days < 7):
159 result.append(f"{hours}h")
160 if minutes and not (months or days):
161 result.append(f"{minutes}m")
162 if seconds and not (months or days):
163 result.append(f"{seconds}s")
164 return " ".join(result)
165
166
167def calculate_client_address(interface_address: str, user_id: int) -> str:
168 try:
169 interface = ipaddress.ip_interface(interface_address)
170 address = interface.ip
171 network = interface.network
172 except ValueError:
173 return ""
174 user_address = network[user_id]
175 if user_address == network[0]:
176 user_address = network.broadcast_address - 1
177 if user_address == address:
178 user_address = network.broadcast_address - 2
179 return user_address.compressed + "/" + str(user_address.max_prefixlen)
180
181
182def setup_format_variables(extra_data: dict) -> dict:
183 expire_strategy = extra_data.get("expire_strategy")
184 expire_datetime = extra_data.get("expire_date")
185 usage_duration = extra_data.get("usage_duration")
186
187 if expire_strategy != UserExpireStrategy.START_ON_FIRST_USE:
188 if expire_datetime is not None:
189 seconds_left = (expire_datetime - dt.utcnow()).total_seconds()
190 expire_date = expire_datetime.date()
191 jalali_expire_date = jd.fromgregorian(
192 year=expire_date.year,
193 month=expire_date.month,
194 day=expire_date.day,
195 ).strftime("%Y-%m-%d")
196 days_left = (expire_datetime - dt.utcnow()).days + 1
197 time_left = format_time_left(seconds_left)
198 else:
199 days_left, time_left, expire_date, jalali_expire_date = "∞" * 4
200 else:
201 if usage_duration is not None and usage_duration >= 0:
202 days_left = timedelta(seconds=usage_duration).days
203 time_left = format_time_left(usage_duration)
204 expire_date = "-"
205 jalali_expire_date = "-"
206 else:
207 days_left, time_left, expire_date, jalali_expire_date = "∞" * 4
208
209 if extra_data.get("data_limit"):
210 data_limit = readable_size(extra_data["data_limit"])
211 data_left = extra_data["data_limit"] - extra_data["used_traffic"]
212 if data_left < 0:
213 data_left = 0
214 data_left = readable_size(data_left)
215 else:
216 data_limit, data_left = "∞" * 2
217
218 status_emoji = ACTIVITY_EMOJIS.get(extra_data.get("is_active")) or ""
219
220 format_variables = defaultdict(
221 lambda: "<missing>",
222 {
223 "SERVER_IP": SERVER_IP,
224 "USERNAME": extra_data.get("username", "{USERNAME}"),
225 "DATA_USAGE": readable_size(extra_data.get("used_traffic")),
226 "DATA_LIMIT": data_limit,
227 "DATA_LEFT": data_left,
228 "DAYS_LEFT": days_left,
229 "EXPIRE_DATE": expire_date,
230 "JALALI_EXPIRE_DATE": jalali_expire_date,
231 "TIME_LEFT": time_left,
232 "STATUS_EMOJI": status_emoji,
233 },
234 )
235
236 return format_variables
237
238
239def generate_user_configs(
240 inbounds: list,
241 key: str,
242 user_id: int,
243 format_variables: dict,
244 chaining_support: bool,
245) -> Union[List, str]:
246 salt = secrets.token_hex(8)
247 configs = []
248
249 with GetDB() as db:
250 hosts = get_hosts_for_user(db, user_id)
251
252 for host in hosts:
253 chained_hosts = [c.chained_host for c in host.chain]
254 if chained_hosts and not chaining_support:
255 continue
256 data = create_config(
257 host, key, format_variables, salt, user_id, chained_hosts
258 )
259 configs.append(data)
260
261 return configs
262
263
264def create_config(
265 host, key, format_variables, salt, user_id, next_hosts: list | None = None
266):
267 if next_hosts is None:
268 next_hosts = []
269
270 if host.inbound:
271 inbound, protocol = (
272 json.loads(host.inbound.config),
273 host.inbound.protocol,
274 )
275 network = inbound.get("network")
276 auth_uuid, auth_password = UUID(gen_uuid(key)), gen_password(key)
277 else:
278 inbound, protocol, network = {}, host.host_protocol, host.host_network
279 auth_uuid, auth_password = (
280 UUID(host.uuid) if host.uuid else None
281 ), host.password
282
283 format_variables.update(
284 {
285 "PROTOCOL": (
286 host.inbound.protocol.value
287 if host.inbound
288 else host.host_protocol
289 )
290 }
291 )
292 format_variables.update({"TRANSPORT": network or "<missing>"})
293
294 host_snis = host.sni.split(",") if host.sni else []
295 sni_list = host_snis or inbound.get("sni", [])
296 if sni_list:
297 sni = random.choice(sni_list).replace("*", salt)
298 else:
299 sni = ""
300
301 host_hosts = host.host.split(",") if host.host else []
302 req_host_list = host_hosts or inbound.get("host", [])
303 if req_host_list:
304 req_host = random.choice(req_host_list).replace("*", salt)
305 else:
306 req_host = ""
307
308 host_tls = (
309 None
310 if host.security == InboundHostSecurity.inbound_default
311 else host.security.value
312 )
313 splithttp_settings = (
314 SplitHttpSettings.model_validate(host.splithttp_settings)
315 if host.splithttp_settings
316 else None
317 )
318 mux_settings = (
319 MuxSettings.model_validate(host.mux_settings)
320 if host.mux_settings
321 else None
322 )
323
324 data = V2Data(
325 host.inbound.protocol.value if host.inbound else host.host_protocol,
326 host.remark.format_map(format_variables),
327 host.address.format_map(format_variables),
328 host.port or inbound.get("port", 0),
329 transport_type=network,
330 sni=sni,
331 host=req_host,
332 tls=host_tls or inbound.get("tls"),
333 header_type=host.header_type or inbound.get("header_type"),
334 alpn=host.alpn if host.alpn != "none" else None,
335 path=(
336 host.path.format_map(format_variables)
337 if host.path
338 else inbound.get("path")
339 ),
340 fingerprint=host.fingerprint.value or inbound.get("fp"),
341 reality_pbk=inbound.get("pbk"),
342 reality_sid=inbound.get("sid"),
343 client_address=calculate_client_address(
344 inbound.get("address"), user_id
345 ),
346 flow=host.flow or inbound.get("flow"),
347 dns_servers=(host.dns_servers.split(",") if host.dns_servers else []),
348 mtu=host.mtu,
349 allowed_ips=(
350 list(map(str.strip, host.allowed_ips.split(",")))
351 if host.allowed_ips
352 else None
353 ),
354 allow_insecure=host.allowinsecure,
355 uuid=auth_uuid,
356 password=auth_password,
357 ed25519=generate_curve25519_pbk(key),
358 early_data=host.early_data,
359 splithttp_settings=(
360 V2SplitHttpSettings(
361 mode=splithttp_settings.mode,
362 no_grpc_header=splithttp_settings.no_grpc_header,
363 padding_bytes=splithttp_settings.padding_bytes,
364 xmux=(
365 V2XMuxSettings(**splithttp_settings.xmux.model_dump())
366 if splithttp_settings.xmux
367 else None
368 ),
369 )
370 if splithttp_settings
371 else None
372 ),
373 mux_settings=(
374 V2MuxSettings(
375 protocol=mux_settings.protocol,
376 sing_box_mux_settings=(
377 V2SingBoxMuxSettings(
378 **mux_settings.sing_box_mux_settings.model_dump()
379 )
380 if mux_settings.sing_box_mux_settings
381 else None
382 ),
383 mux_cool_settings=(
384 V2MuxCoolSettings(
385 **mux_settings.mux_cool_settings.model_dump()
386 )
387 if mux_settings.mux_cool_settings
388 else None
389 ),
390 )
391 if mux_settings
392 else None
393 ),
394 http_headers=host.http_headers,
395 shadowsocks_method=host.shadowsocks_method or "chacha20-ietf-poly1305",
396 shadowtls_version=host.shadowtls_version
397 or inbound.get("shadowtls_version"),
398 weight=host.weight,
399 xray_noises=(
400 [XrayNoise(**noise) for noise in host.udp_noises]
401 if host.udp_noises
402 else None
403 ),
404 next=(
405 create_config(
406 next_hosts[0],
407 key,
408 format_variables,
409 salt,
410 user_id,
411 next_hosts[1:],
412 )
413 if next_hosts
414 else None
415 ),
416 )
417 if host.fragment:
418 data.fragment = True
419 data.fragment_packets = host.fragment["packets"]
420 data.fragment_length = host.fragment["length"]
421 data.fragment_interval = host.fragment["interval"]
422
423 return data
424
425
426def encode_title(text: str) -> str:
427 return f"base64:{base64.b64encode(text.encode()).decode()}"