1import base64
2import json
3import random
4import secrets
5import urllib.parse as urlparse
6from datetime import datetime as dt, timedelta
7from typing import TYPE_CHECKING, Literal, Union, List
8from uuid import UUID
9
10import yaml
11from jdatetime import date as jd
12
13from app.models.proxy import FormatVariables, InboundBase, InboundHost, InboundHostSecurity
14from app.templates import render_template
15from app.utils.system import get_public_ip, readable_size
16from app.utils.keygen import generate_settings
17from app.db import GetDB, crud
18
19if TYPE_CHECKING:
20 from app.models.user import UserResponse
21
22from config import CLASH_SUBSCRIPTION_TEMPLATE, SINGBOX_SUBSCRIPTION_TEMPLATE
23
24SERVER_IP = get_public_ip()
25
26STATUS_EMOJIS = {
27 "active": "✅",
28 "expired": "⌛️",
29 "limited": "🪫",
30 "disabled": "❌",
31 "connect_to_start": "🔌",
32}
33
34class V2rayShareLink(str):
35 @classmethod
36 def vmess(
37 cls,
38 remark: str,
39 address: str,
40 port: int,
41 id: Union[str, UUID],
42 host="",
43 net="tcp",
44 path="",
45 type="",
46 tls="none",
47 sni="",
48 fp="",
49 alpn="",
50 pbk="",
51 sid="",
52 spx="",
53 ais="",
54 ):
55 payload = {
56 "add": address,
57 "aid": "0",
58 "host": host,
59 "id": str(id),
60 "net": net,
61 "path": path,
62 "port": port,
63 "ps": remark,
64 "scy": "auto",
65 "tls": tls,
66 "type": type,
67 "v": "2",
68 }
69
70 if tls == "tls":
71 payload["sni"] = sni
72 payload["fp"] = fp
73 payload["alpn"] = alpn
74 if ais:
75 payload["allowInsecure"] = 1
76 elif tls == "reality":
77 payload["sni"] = sni
78 payload["fp"] = fp
79 payload["pbk"] = pbk
80 payload["sid"] = sid
81 payload["spx"] = spx
82
83 return (
84 "vmess://"
85 + base64.b64encode(
86 json.dumps(payload, sort_keys=True).encode("utf-8")
87 ).decode()
88 )
89
90 @classmethod
91 def vless(
92 cls,
93 remark: str,
94 address: str,
95 port: int,
96 id: Union[str, UUID],
97 net="ws",
98 path="",
99 host="",
100 type="",
101 flow="",
102 tls="none",
103 sni="",
104 fp="",
105 alpn="",
106 pbk="",
107 sid="",
108 spx="",
109 ais="",
110 ):
111 payload = {"security": tls, "type": net,
112 "host": host, "headerType": type}
113 if flow and (net in ("tcp", "kcp") and type != "http"):
114 payload["flow"] = flow
115
116 if net == "grpc":
117 payload["serviceName"] = path
118 else:
119 payload["path"] = path
120
121 if tls == "tls":
122 payload["sni"] = sni
123 payload["fp"] = fp
124 payload["alpn"] = alpn
125 if ais:
126 payload["allowInsecure"] = 1
127 elif tls == "reality":
128 payload["sni"] = sni
129 payload["fp"] = fp
130 payload["pbk"] = pbk
131 payload["sid"] = sid
132 payload["spx"] = spx
133
134 return (
135 "vless://"
136 + f"{id}@{address}:{port}?"
137 + urlparse.urlencode(payload)
138 + f"#{( urlparse.quote(remark))}"
139 )
140
141 @classmethod
142 def trojan(
143 cls,
144 remark: str,
145 address: str,
146 port: int,
147 password: str,
148 net="tcp",
149 path="",
150 host="",
151 type="",
152 flow="",
153 tls="none",
154 sni="",
155 fp="",
156 alpn="",
157 pbk="",
158 sid="",
159 spx="",
160 ais="",
161 ):
162 payload = {"security": tls, "type": net,
163 "host": host, "headerType": type}
164 if flow and (net in ("tcp", "kcp") and type != "http"):
165 payload["flow"] = flow
166
167 if net == "grpc":
168 payload["serviceName"] = path
169 else:
170 payload["path"] = path
171
172 if tls == "tls":
173 payload["sni"] = sni
174 payload["fp"] = fp
175 payload["alpn"] = alpn
176 if ais:
177 payload["allowInsecure"] = 1
178 elif tls == "reality":
179 payload["sni"] = sni
180 payload["fp"] = fp
181 payload["pbk"] = pbk
182 payload["sid"] = sid
183 payload["spx"] = spx
184
185 return (
186 "trojan://"
187 + f"{urlparse.quote(password, safe=':')}@{address}:{port}?"
188 + urlparse.urlencode(payload)
189 + f"#{urlparse.quote(remark)}"
190 )
191
192 @classmethod
193 def shadowsocks(
194 cls, remark: str, address: str, port: int, password: str, method: str
195 ):
196 return (
197 "ss://"
198 + base64.b64encode(f"{method}:{password}".encode()).decode()
199 + f"@{address}:{port}#{urlparse.quote(remark)}"
200 )
201
202
203class ClashConfiguration(object):
204 def __init__(self):
205 self.data = {
206 "proxies": [],
207 "proxy-groups": [],
208 # Some clients rely on "rules" option and will fail without it.
209 "rules": [],
210 }
211 self.proxy_remarks = []
212
213 def render(self):
214 return yaml.dump(
215 yaml.load(
216 render_template(
217 CLASH_SUBSCRIPTION_TEMPLATE,
218 {"conf": self.data, "proxy_remarks": self.proxy_remarks},
219 ),
220 Loader=yaml.SafeLoader,
221 ),
222 sort_keys=False,
223 allow_unicode=True,
224 )
225
226 def __str__(self) -> str:
227 return self.render()
228
229 def __repr__(self) -> str:
230 return self.render()
231
232 def _remark_validation(self, remark):
233 if not remark in self.proxy_remarks:
234 return remark
235 c = 2
236 while True:
237 new = f"{remark} ({c})"
238 if not new in self.proxy_remarks:
239 return new
240 c += 1
241
242 def make_node(
243 self,
244 name: str,
245 type: str,
246 server: str,
247 port: int,
248 network: str,
249 tls: bool,
250 sni: str,
251 host: str,
252 path: str,
253 udp: bool = True,
254 alpn: str = "",
255 ais: bool = "",
256 ):
257 if type == "shadowsocks":
258 type = "ss"
259
260 remark = self._remark_validation(name)
261 node = {
262 "name": remark,
263 "type": type,
264 "server": server,
265 "port": port,
266 "network": network,
267 f"{network}-opts": {},
268 "udp": udp,
269 }
270
271 if type == "ss": # shadowsocks
272 return node
273
274 if tls:
275 node["tls"] = True
276 if type == "trojan":
277 node["sni"] = sni
278 else:
279 node["servername"] = sni
280 if alpn:
281 node["alpn"] = alpn.split(",")
282 if ais:
283 node["skip-cert-verify"] = ais
284
285 net_opts = node[f"{network}-opts"]
286
287 if network == "ws":
288 if path:
289 net_opts["path"] = path
290 if host:
291 net_opts["headers"] = {"Host": host}
292
293 if network == "grpc":
294 if path:
295 net_opts["grpc-service-name"] = path
296
297 if network == "h2":
298 if path:
299 net_opts["path"] = path
300 if host:
301 net_opts["host"] = [host]
302
303 if network == "http" or network == "tcp":
304 if path:
305 net_opts["method"] = "GET"
306 net_opts["path"] = [path]
307 if host:
308 net_opts["method"] = "GET"
309 net_opts["headers"] = {"Host": host}
310
311 return node
312
313 def add(self, remark: str, address: str, inbound: dict, settings: dict):
314 node = self.make_node(
315 name=remark,
316 type=inbound["protocol"],
317 server=address,
318 port=inbound["port"],
319 network=inbound["network"],
320 tls=(inbound["tls"] == "tls"),
321 sni=inbound["sni"],
322 host=inbound["host"],
323 path=inbound["path"],
324 udp=True,
325 alpn=inbound.get("alpn", ""),
326 ais=inbound.get("ais", ""),
327 )
328
329 if inbound["protocol"] == "vmess":
330 node["uuid"] = settings["id"]
331 node["alterId"] = 0
332 node["cipher"] = "auto"
333 self.data["proxies"].append(node)
334 self.proxy_remarks.append(remark)
335
336 if inbound["protocol"] == "trojan":
337 node["password"] = settings["password"]
338 self.data["proxies"].append(node)
339 self.proxy_remarks.append(remark)
340
341 if inbound["protocol"] == "shadowsocks":
342 node["password"] = settings["password"]
343 node["cipher"] = settings["method"]
344 self.data["proxies"].append(node)
345 self.proxy_remarks.append(remark)
346
347
348class ClashMetaConfiguration(ClashConfiguration):
349 def make_node(
350 self,
351 name: str,
352 type: str,
353 server: str,
354 port: int,
355 network: str,
356 tls: bool,
357 sni: str,
358 host: str,
359 path: str,
360 udp: bool = True,
361 alpn: str = "",
362 fp: str = "",
363 pbk: str = "",
364 sid: str = "",
365 ais: bool = "",
366 ):
367 node = super().make_node(
368 name=name,
369 type=type,
370 server=server,
371 port=port,
372 network=network,
373 tls=tls,
374 sni=sni,
375 host=host,
376 path=path,
377 #headers="",
378 udp=udp,
379 alpn=alpn,
380 ais=ais,
381 )
382 if fp:
383 node["client-fingerprint"] = fp
384 if pbk:
385 node["reality-opts"] = {"public-key": pbk, "short-id": sid}
386
387 return node
388
389 def add(self, remark: str, address: str, inbound: dict, settings: dict):
390 node = self.make_node(
391 name=remark,
392 type=inbound["protocol"],
393 server=address,
394 port=inbound["port"],
395 network=inbound["network"],
396 tls=(inbound["tls"] in ("tls", "reality")),
397 sni=inbound["sni"],
398 host=inbound["host"],
399 path=inbound["path"],
400 udp=True,
401 alpn=inbound.get("alpn", ""),
402 fp=inbound.get("fp", ""),
403 pbk=inbound.get("pbk", ""),
404 sid=inbound.get("sid", ""),
405 ais=inbound.get("ais", ""),
406 )
407
408 if inbound["protocol"] == "vmess":
409 node["uuid"] = settings["id"]
410 node["alterId"] = 0
411 node["cipher"] = "auto"
412 self.data["proxies"].append(node)
413 self.proxy_remarks.append(remark)
414
415 if inbound["protocol"] == "vless":
416 node["uuid"] = settings["id"]
417
418 if (
419 inbound["network"] in ("tcp", "kcp")
420 and inbound["header_type"] != "http"
421 ):
422 node["flow"] = settings.get("flow", "")
423
424 self.data["proxies"].append(node)
425 self.proxy_remarks.append(remark)
426
427 if inbound["protocol"] == "trojan":
428 node["password"] = settings["password"]
429
430 if (
431 inbound["network"] in ("tcp", "kcp")
432 and inbound["header_type"] != "http"
433 ):
434 node["flow"] = settings.get("flow", "")
435
436 self.data["proxies"].append(node)
437 self.proxy_remarks.append(remark)
438
439 if inbound["protocol"] == "shadowsocks":
440 node["password"] = settings["password"]
441 node["cipher"] = settings["method"]
442 self.data["proxies"].append(node)
443 self.proxy_remarks.append(remark)
444
445
446def get_v2ray_link(remark: str, address: str, inbound: dict, settings: dict):
447 if inbound["protocol"] == "vmess":
448 return V2rayShareLink.vmess(
449 remark=remark,
450 address=address,
451 port=inbound["port"],
452 id=settings["id"],
453 net=inbound["network"],
454 tls=inbound["tls"],
455 sni=inbound.get("sni", ""),
456 fp=inbound.get("fp", ""),
457 alpn=inbound.get("alpn", ""),
458 pbk=inbound.get("pbk", ""),
459 sid=inbound.get("sid", ""),
460 spx=inbound.get("spx", ""),
461 host=inbound["host"],
462 path=inbound["path"],
463 type=inbound["header_type"],
464 ais=inbound.get("ais", ""),
465 )
466
467 if inbound["protocol"] == "vless":
468 return V2rayShareLink.vless(
469 remark=remark,
470 address=address,
471 port=inbound["port"],
472 id=settings["id"],
473 flow=settings.get("flow", ""),
474 net=inbound["network"],
475 tls=inbound["tls"],
476 sni=inbound.get("sni", ""),
477 fp=inbound.get("fp", ""),
478 alpn=inbound.get("alpn", ""),
479 pbk=inbound.get("pbk", ""),
480 sid=inbound.get("sid", ""),
481 spx=inbound.get("spx", ""),
482 host=inbound["host"],
483 path=inbound["path"],
484 type=inbound["header_type"],
485 ais=inbound.get("ais", ""),
486 )
487
488 if inbound["protocol"] == "trojan":
489 return V2rayShareLink.trojan(
490 remark=remark,
491 address=address,
492 port=inbound["port"],
493 password=settings["password"],
494 flow=settings.get("flow", ""),
495 net=inbound["network"],
496 tls=inbound["tls"],
497 sni=inbound.get("sni", ""),
498 fp=inbound.get("fp", ""),
499 alpn=inbound.get("alpn", ""),
500 pbk=inbound.get("pbk", ""),
501 sid=inbound.get("sid", ""),
502 spx=inbound.get("spx", ""),
503 host=inbound["host"],
504 path=inbound["path"],
505 type=inbound["header_type"],
506 ais=inbound.get("ais", ""),
507 )
508
509 if inbound["protocol"] == "shadowsocks":
510 return V2rayShareLink.shadowsocks(
511 remark=remark,
512 address=address,
513 port=inbound["port"],
514 password=settings["password"],
515 method=settings["method"],
516 )
517
518
519class OutlineConfiguration:
520 def __init__(self):
521 self.config = {}
522
523 def add_directly(self, data: dict):
524 self.config.update(data)
525
526 def render(self):
527 return json.dumps(self.config, indent=0)
528
529 def make_outbound(
530 self, remark: str, address: str, port: int, password: str, method: str
531 ):
532 config = {
533 "method": method,
534 "password": password,
535 "server": address,
536 "server_port": port,
537 "tag": remark,
538 }
539 return config
540
541 def add(self, remark: str, address: str, inbound: dict, settings: dict):
542 if inbound["protocol"] != "shadowsocks":
543 return
544
545 outbound = self.make_outbound(
546 remark=remark,
547 address=address,
548 port=inbound["port"],
549 password=settings["password"],
550 method=settings["method"],
551 )
552 self.add_directly(outbound)
553
554
555class SingBoxConfiguration(str):
556 def __init__(self):
557 template = render_template(SINGBOX_SUBSCRIPTION_TEMPLATE)
558 self.config = json.loads(template)
559
560 def add_outbound(self, outbound_data):
561 self.config["outbounds"].append(outbound_data)
562
563 def render(self):
564 urltest_types = ["vmess", "vless", "trojan", "shadowsocks"]
565 urltest_tags = [
566 outbound["tag"]
567 for outbound in self.config["outbounds"]
568 if outbound["type"] in urltest_types
569 ]
570 selector_types = ["vmess", "vless", "trojan", "shadowsocks", "urltest"]
571 selector_tags = [
572 outbound["tag"]
573 for outbound in self.config["outbounds"]
574 if outbound["type"] in selector_types
575 ]
576
577 for outbound in self.config["outbounds"]:
578 if outbound.get("type") == "urltest":
579 outbound["outbounds"] = urltest_tags
580
581 for outbound in self.config["outbounds"]:
582 if outbound.get("type") == "selector":
583 outbound["outbounds"] = selector_tags
584
585 return json.dumps(self.config, indent=4)
586
587 @staticmethod
588 def tls_config(
589 sni=None, fp=None, tls=None, pbk=None, sid=None, alpn=None, ais=None
590 ):
591 config = {}
592 if tls in ["tls", "reality"]:
593 config["enabled"] = True
594
595 if sni is not None:
596 config["server_name"] = sni
597
598 if tls == "tls" and ais:
599 config["insecure"] = ais
600
601 if tls == "reality":
602 config["reality"] = {"enabled": True}
603 if pbk:
604 config["reality"]["public_key"] = pbk
605 if sid:
606 config["reality"]["short_id"] = sid
607
608 if fp:
609 config["utls"] = {"enabled": bool(fp), "fingerprint": fp}
610
611 if alpn:
612 config["alpn"] = [alpn] if not isinstance(alpn, list) else alpn
613
614 return config
615
616 @staticmethod
617 def transport_config(
618 transport_type="",
619 host="",
620 path="",
621 method="",
622 headers="",
623 idle_timeout="15s",
624 ping_timeout="15s",
625 max_early_data=None,
626 early_data_header_name=None,
627 permit_without_stream=False,
628 ):
629 transport_config = {}
630
631 if transport_type:
632 transport_config["type"] = transport_type
633
634 if transport_type == "http":
635 if host:
636 transport_config["host"] = host
637 if path:
638 transport_config["path"] = path
639 if method:
640 transport_config["method"] = method
641 if headers:
642 transport_config["headers"] = headers
643 if idle_timeout:
644 transport_config["idle_timeout"] = idle_timeout
645 if ping_timeout:
646 transport_config["ping_timeout"] = ping_timeout
647
648 elif transport_type == "ws":
649 if path:
650 transport_config["path"] = path
651 if headers:
652 transport_config["headers"] = headers
653 if max_early_data is not None:
654 transport_config["max_early_data"] = max_early_data
655 if early_data_header_name:
656 transport_config["early_data_header_name"] = early_data_header_name
657
658 elif transport_type == "ws":
659 if path:
660 transport_config["service_name"] = path
661 if idle_timeout:
662 transport_config["idle_timeout"] = idle_timeout
663 if ping_timeout:
664 transport_config["ping_timeout"] = ping_timeout
665 if permit_without_stream:
666 transport_config["permit_without_stream"] = permit_without_stream
667
668 return transport_config
669
670 def make_outbound(
671 self,
672 type: str,
673 remark: str,
674 address: str,
675 port: int,
676 net="ws",
677 path="",
678 host="",
679 flow="",
680 tls="",
681 sni="",
682 fp="",
683 alpn="",
684 pbk="",
685 sid="",
686 headers="",
687 ais="",
688 ):
689 config = {
690 "type": type,
691 "tag": remark,
692 "server": address,
693 "server_port": port,
694 }
695 if net in ("tcp", "kcp") or headers != "http":
696 if flow:
697 config["flow"] = flow
698
699 if net in ["http", "ws", "quic", "grpc", "h2"]:
700 if net == "h2":
701 net = "http"
702 alpn = "h2"
703 config["transport"] = self.transport_config(
704 transport_type=net, host=host, path=path, headers=headers
705 )
706 else:
707 config["network"]: net
708
709 if tls in ("tls", "reality"):
710 config["tls"] = self.tls_config(
711 sni=sni, fp=fp, tls=tls, pbk=pbk, sid=sid, alpn=alpn, ais=ais
712 )
713
714 return config
715
716 def add(self, remark: str, address: str, inbound: dict, settings: dict):
717 outbound = self.make_outbound(
718 remark=remark,
719 type=inbound["protocol"],
720 address=address,
721 port=inbound["port"],
722 net=inbound["network"],
723 tls=(inbound["tls"]),
724 flow=settings.get("flow", ""),
725 sni=inbound["sni"],
726 host=inbound["host"],
727 path=inbound["path"],
728 alpn=inbound.get("alpn", ""),
729 fp=inbound.get("fp", ""),
730 pbk=inbound.get("pbk", ""),
731 sid=inbound.get("sid", ""),
732 headers=inbound["header_type"],
733 ais=inbound.get("ais", ""),
734 )
735
736 if inbound["protocol"] == "vmess":
737 outbound["uuid"] = settings["id"]
738
739 elif inbound["protocol"] == "vless":
740 outbound["uuid"] = settings["id"]
741
742 elif inbound["protocol"] == "trojan":
743 outbound["password"] = settings["password"]
744
745 elif inbound["protocol"] == "shadowsocks":
746 outbound["password"] = settings["password"]
747 outbound["method"] = settings["method"]
748
749 self.add_outbound(outbound)
750
751
752def generate_v2ray_links(inbounds: list, key: str, extra_data: dict) -> list:
753 format_variables = setup_format_variables(extra_data)
754 return process_inbounds_and_tags(inbounds, key, format_variables, mode="v2ray")
755
756
757def generate_clash_subscription(
758 inbounds: dict, key: str, extra_data: dict, is_meta: bool = False
759) -> str:
760 if is_meta is True:
761 conf = ClashMetaConfiguration()
762 else:
763 conf = ClashConfiguration()
764
765 format_variables = setup_format_variables(extra_data)
766 return process_inbounds_and_tags(
767 inbounds, key, format_variables, mode="clash", conf=conf
768 )
769
770
771def generate_singbox_subscription(
772 inbounds: dict, key: str, extra_data: dict
773) -> str:
774 conf = SingBoxConfiguration()
775
776 format_variables = setup_format_variables(extra_data)
777 return process_inbounds_and_tags(
778 inbounds, key, format_variables, mode="sing-box", conf=conf
779 )
780
781
782def generate_v2ray_subscription(links: list) -> str:
783 return base64.b64encode("\n".join(links).encode()).decode()
784
785
786def generate_outline_subscription(
787 inbounds: dict, key: str, extra_data: dict
788) -> str:
789 conf = OutlineConfiguration()
790
791 format_variables = setup_format_variables(extra_data)
792 return process_inbounds_and_tags(
793 inbounds, key, format_variables, mode="outline", conf=conf
794 )
795
796
797def generate_subscription(
798 user: "UserResponse",
799 config_format: Literal["v2ray", "clash-meta", "clash", "sing-box", "outline"],
800 as_base64: bool,
801) -> str:
802 kwargs = {
803 # "proxies": user.proxies,
804 "inbounds": user.inbounds,
805 "key": user.key,
806 "extra_data": user.__dict__,
807 }
808
809 if config_format == "v2ray":
810 config = "\n".join(generate_v2ray_links(**kwargs))
811 elif config_format == "clash-meta":
812 config = generate_clash_subscription(**kwargs, is_meta=True)
813 elif config_format == "clash":
814 config = generate_clash_subscription(**kwargs)
815 elif config_format == "sing-box":
816 config = generate_singbox_subscription(**kwargs)
817 elif config_format == "outline":
818 config = generate_outline_subscription(**kwargs)
819 else:
820 raise ValueError(f'Unsupported format "{config_format}"')
821
822 if as_base64:
823 config = base64.b64encode(config.encode()).decode()
824
825 return config
826
827
828def format_time_left(seconds_left: int) -> str:
829 if not seconds_left or seconds_left <= 0:
830 return "∞"
831
832 minutes, seconds = divmod(seconds_left, 60)
833 hours, minutes = divmod(minutes, 60)
834 days, hours = divmod(hours, 24)
835 months, days = divmod(days, 30)
836
837 result = []
838 if months:
839 result.append(f"{months}m")
840 if days:
841 result.append(f"{days}d")
842 if hours and (days < 7):
843 result.append(f"{hours}h")
844 if minutes and not (months or days):
845 result.append(f"{minutes}m")
846 if seconds and not (months or days):
847 result.append(f"{seconds}s")
848 return " ".join(result)
849
850from app.models.user import UserStatus
851
852def setup_format_variables(extra_data: dict) -> dict:
853
854
855 user_status = extra_data.get("status")
856 expire_timestamp = extra_data.get("expire")
857 on_hold_expire_duration = extra_data.get("on_hold_expire_duration")
858
859 if user_status != UserStatus.on_hold:
860 if expire_timestamp is not None and expire_timestamp >= 0:
861 seconds_left = expire_timestamp - int(dt.utcnow().timestamp())
862 expire_datetime = dt.fromtimestamp(expire_timestamp)
863 expire_date = expire_datetime.date()
864 jalali_expire_date = jd.fromgregorian(
865 year=expire_date.year, month=expire_date.month, day=expire_date.day
866 ).strftime("%Y-%m-%d")
867 days_left = (expire_datetime - dt.utcnow()).days + 1
868 time_left = format_time_left(seconds_left)
869 else:
870 days_left = "∞"
871 time_left = "∞"
872 expire_date = "∞"
873 jalali_expire_date = "∞"
874 else:
875 if on_hold_expire_duration is not None and on_hold_expire_duration >= 0:
876 days_left = timedelta(seconds=on_hold_expire_duration).days
877 time_left = format_time_left(on_hold_expire_duration)
878 expire_date = "-"
879 jalali_expire_date = "-"
880 else:
881 days_left = "∞"
882 time_left = "∞"
883 expire_date = "∞"
884 jalali_expire_date = "∞"
885
886 if extra_data.get("data_limit"):
887 data_limit = readable_size(extra_data["data_limit"])
888 data_left = extra_data["data_limit"] - extra_data["used_traffic"]
889 if data_left < 0:
890 data_left = 0
891 data_left = readable_size(data_left)
892 else:
893 data_limit = "∞"
894 data_left = "∞"
895
896 status_emoji = STATUS_EMOJIS.get(extra_data.get("status")) or ""
897
898 format_variables = {
899 "SERVER_IP": SERVER_IP,
900 "USERNAME": extra_data.get("username", "{USERNAME}"),
901 "DATA_USAGE": readable_size(extra_data.get("used_traffic")),
902 "DATA_LIMIT": data_limit,
903 "DATA_LEFT": data_left,
904 "DAYS_LEFT": days_left,
905 "EXPIRE_DATE": expire_date,
906 "JALALI_EXPIRE_DATE": jalali_expire_date,
907 "TIME_LEFT": time_left,
908 "STATUS_EMOJI": status_emoji,
909 }
910
911 return format_variables
912
913
914def process_inbounds_and_tags(
915 inbounds: list,
916 ukey: str,
917 format_variables: dict,
918 mode: str = "v2ray",
919 conf=None,
920) -> Union[List, str]:
921 salt = secrets.token_hex(8)
922 results = []
923
924 for inb in inbounds:
925 inb_id, inbound, protocol, tag = inb.id, json.loads(inb.config), inb.protocol, inb.tag
926 settings = generate_settings(ukey, protocol)
927
928 format_variables.update({"PROTOCOL": protocol.name})
929 if not inbound:
930 continue
931
932 format_variables.update({"TRANSPORT": inbound["network"]})
933 host_inbound = inbound.copy()
934 with GetDB() as db:
935 hosts = crud.get_inbound_hosts(db, inb_id)
936 # for host in xray.hosts.get(tag, []):
937 for host in hosts:
938 host_snis = host.sni.split(",") if host.sni else []
939 sni_list = host_snis or inbound["sni"]
940 if sni_list:
941 sni = random.choice(sni_list).replace("*", salt)
942 else:
943 sni = ""
944
945 host_hosts = host.host.split(",") if host.host else []
946 req_host_list = host_hosts or inbound["host"]
947 if req_host_list:
948 req_host = random.choice(req_host_list).replace("*", salt)
949 else:
950 req_host = ""
951
952 host_tls = None if host.security == InboundHostSecurity.inbound_default else host.security.value
953 host_inbound.update(
954 {
955 "port": host.port or inbound["port"],
956 "sni": sni,
957 "host": req_host,
958 "tls": inbound["tls"] if host_tls is None else host_tls,
959 "alpn": host.alpn.value or inbound.get("alpn", ""),
960 "path": inbound.get("path", "")
961 if host.path is None
962 else host.path.format_map(format_variables),
963 "fp": host.fingerprint.value or inbound.get("fp", ""),
964 "ais": host.allowinsecure or ""
965 }
966 )
967
968 if mode == "v2ray":
969 results.append(
970 get_v2ray_link(
971 remark=host.remark.format_map(format_variables),
972 address=host.address.format_map(
973 format_variables),
974 inbound=host_inbound,
975 settings=settings,
976 )
977 )
978 elif mode in ["clash", "sing-box", "outline"]:
979 conf.add(
980 remark=host.remark.format_map(format_variables),
981 address=host.address.format_map(format_variables),
982 inbound=host_inbound,
983 settings=settings,
984 )
985
986 if mode in ["clash", "sing-box", "outline"]:
987 return conf.render()
988 return results
989
990
991def encode_title(text: str) -> str:
992 return f"base64:{base64.b64encode(text.encode()).decode()}"