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