diff --git a/app/db/crud.py b/app/db/crud.py index 4117289..642c35b 100644 --- a/app/db/crud.py +++ b/app/db/crud.py @@ -1,6 +1,6 @@ import json import secrets -from datetime import datetime +from datetime import datetime, timedelta from enum import Enum from typing import List, Optional, Tuple, Union @@ -173,32 +173,6 @@ def update_host(db: Session, db_host: InboundHost, host: InboundHostModify): return db_host -def update_hosts( - db: Session, inbound_tag: str, modified_hosts: List[InboundHostModify] -): - inbound = get_or_create_inbound(db, inbound_tag) - inbound.hosts = [ - InboundHost( - remark=host.remark, - address=host.address, - port=host.port, - path=host.path, - sni=host.sni, - host=host.host, - inbound=inbound, - security=host.security, - alpn=host.alpn, - fingerprint=host.fingerprint, - allowinsecure=host.allowinsecure, - is_disabled=host.is_disabled, - ) - for host in modified_hosts - ] - db.commit() - db.refresh(inbound) - return inbound.hosts - - def get_user(db: Session, username: str): return db.query(User).filter(User.username == username).first() @@ -470,7 +444,7 @@ def set_owner(db: Session, dbuser: User, admin: Admin): def start_user_expire(db: Session, dbuser: User): - expire = int(datetime.utcnow().timestamp()) + dbuser.on_hold_expire_duration + expire = datetime.utcnow() + timedelta(seconds=dbuser.on_hold_expire_duration) dbuser.expire = expire db.commit() db.refresh(dbuser) diff --git a/app/db/migrations/versions/20faa9f18c0a_init.py b/app/db/migrations/versions/20faa9f18c0a_init.py index 462abeb..7684108 100644 --- a/app/db/migrations/versions/20faa9f18c0a_init.py +++ b/app/db/migrations/versions/20faa9f18c0a_init.py @@ -111,7 +111,7 @@ def upgrade() -> None: sa.Column('data_limit_reset_strategy', sa.Enum('no_reset', 'day', 'week', 'month', 'year', name='userdatalimitresetstrategy'), nullable=False), sa.Column('ip_limit', sa.Integer(), nullable=False), sa.Column('settings', sa.String(), nullable=True), - sa.Column('expire', sa.Integer(), nullable=True), + sa.Column('expire', sa.DateTime(), nullable=True), sa.Column('admin_id', sa.Integer(), nullable=True), sa.Column('sub_updated_at', sa.DateTime(), nullable=True), sa.Column('sub_last_user_agent', sa.String(length=512), nullable=True), diff --git a/app/models/user.py b/app/models/user.py index 7a79847..ac58ef4 100644 --- a/app/models/user.py +++ b/app/models/user.py @@ -1,4 +1,3 @@ -import re import secrets from datetime import datetime from enum import Enum @@ -16,7 +15,7 @@ from pydantic import ( from config import XRAY_SUBSCRIPTION_URL_PREFIX -USERNAME_REGEXP = re.compile(r"^\w{3,32}$") +USERNAME_REGEXP = r"^\w{3,32}$" class ReminderType(str, Enum): @@ -52,9 +51,8 @@ class UserDataLimitResetStrategy(str, Enum): class UserBase(BaseModel): - # proxies: Dict[ProxyTypes, ProxySettings] = {} id: Optional[int] = None - username: Annotated[str, StringConstraints(to_lower=True)] + username: Annotated[str, StringConstraints(to_lower=True, pattern=USERNAME_REGEXP)] expire: Union[datetime, None, Literal[0]] = Field(None) key: str = Field(default_factory=lambda: secrets.token_hex(16)) data_limit: Optional[int] = Field( @@ -63,29 +61,13 @@ class UserBase(BaseModel): data_limit_reset_strategy: UserDataLimitResetStrategy = ( UserDataLimitResetStrategy.no_reset ) - note: Optional[str] = Field(None) + note: Optional[Annotated[str, Field(max_length=500)]] = None sub_updated_at: Optional[datetime] = Field(None) sub_last_user_agent: Optional[str] = Field(None) online_at: Optional[datetime] = Field(None) on_hold_expire_duration: Optional[int] = Field(None) on_hold_timeout: Optional[datetime] = Field(None) - @field_validator("username") - @classmethod - def validate_username(cls, v: str): - if not USERNAME_REGEXP.match(v): - raise ValueError( - "Username should be 3 to 32 characters containing A-Z, a-z, 0-9 and underscores in between." - ) - return v - - @field_validator("note") - @classmethod - def validate_note(cls, v: str): - if v and len(v) > 500: - raise ValueError("User's note can be a maximum of 500 character") - return v - model_config = ConfigDict(from_attributes=True) @@ -136,10 +118,8 @@ class UserCreate(User): @field_validator("expire") @classmethod - def validate_note(cls, v: Union[datetime, None, Literal[0]]): - if isinstance(v, datetime) and ( - v.tzinfo is not None - ): + def validate_expire(cls, v: Union[datetime, None, Literal[0]]): + if isinstance(v, datetime) and (v.tzinfo is not None): raise ValueError( "Expire date should be offset naive, and preferably in utc timezone." ) @@ -178,7 +158,7 @@ class UserModify(User): @field_validator("expire") @classmethod - def validate_note(cls, v: Union[datetime, None, Literal[0]]): + def validate_expire(cls, v: Union[datetime, None, Literal[0]]): if isinstance(v, datetime) and v.tzinfo is not None: raise ValueError( "Expire date should be offset naive, and preferably in utc timezone." diff --git a/app/tasks/review_users.py b/app/tasks/review_users.py index 1399f0e..fb7a176 100644 --- a/app/tasks/review_users.py +++ b/app/tasks/review_users.py @@ -57,7 +57,7 @@ async def review_users(): for user in get_users(db, status=UserStatus.active): limited = user.data_limit and user.used_traffic >= user.data_limit - expired = user.expire and user.expire <= now.timestamp() + expired = user.expire and user.expire <= now if limited: status = UserStatus.limited elif expired: @@ -81,17 +81,15 @@ async def review_users(): for user in get_users(db, status=UserStatus.on_hold): if user.edit_at: - base_time = datetime.timestamp(user.edit_at) + base_time = user.edit_at else: - base_time = datetime.timestamp(user.created_at) + base_time = user.created_at # Check if the user is online After or at 'base_time' - if user.online_at and base_time <= datetime.timestamp(user.online_at): + if user.online_at and base_time <= user.online_at: status = UserStatus.active - elif user.on_hold_timeout and ( - datetime.timestamp(user.on_hold_timeout) <= (now.timestamp()) - ): + elif user.on_hold_timeout and (user.on_hold_timeout <= now): # If the user didn't connect within the timeout period, change status to "Active" status = UserStatus.active diff --git a/app/templates/clash/default.yml b/app/templates/clash/default.yml deleted file mode 100644 index bb7126d..0000000 --- a/app/templates/clash/default.yml +++ /dev/null @@ -1,13 +0,0 @@ -mode: Global -port: 7890 - -{{ conf | except("proxy-groups", "port", "mode") | yaml }} - -proxy-groups: -- name: '♻️ Automatic' - type: 'url-test' - url: 'http://www.gstatic.com/generate_204' - interval: 300 - proxies: - {{ proxy_remarks | yaml | indent(2) }} -{{ conf.get("proxy-groups", []) | yaml }} diff --git a/app/templates/singbox/default.json b/app/templates/singbox/default.json deleted file mode 100644 index 5a84a7a..0000000 --- a/app/templates/singbox/default.json +++ /dev/null @@ -1,100 +0,0 @@ -{ - "log": { - "level": "warn", - "timestamp": true - }, - "dns": { - "servers": [ - { - "tag": "dns_proxy", - "address": "1.1.1.1", - "address_resolver": "dns_direct", - "strategy": "prefer_ipv4", - "detour": "Internet" - }, - { - "tag": "dns_direct", - "address": "local", - "strategy": "prefer_ipv4", - "detour": "direct" - }, - { - "tag": "dns_block", - "address": "rcode://success" - } - ], - "rules": [ - { - "geosite": "private", - "server": "dns_direct" - }, - { - "outbound": "any", - "server": "dns_direct" - } - ], - "fakeip": {}, - "independent_cache": true - }, - "inbounds": [ - { - "type": "tun", - "mtu": 1400, - "inet4_address": "172.19.0.1/30", - "inet6_address": "fdfe:dcba:9876::1/126", - "auto_route": true, - "strict_route": true, - "endpoint_independent_nat": true, - "stack": "mixed", - "sniff": true - } - ], - "outbounds": [ - { - "type": "selector", - "tag": "Internet", - "outbounds": null - }, - { - "type": "urltest", - "tag": "Best Latency", - "outbounds": null, - "url": "https://www.gstatic.com/generate_204", - "interval": "1m0s" - }, - { - "type": "direct", - "tag": "direct" - }, - { - "type": "block", - "tag": "block" - }, - { - "type": "dns", - "tag": "dns-out" - } - ], - "route": { - "geoip": { - "download_url": "https://github.com/soffchen/sing-geoip/releases/latest/download/geoip.db", - "download_detour": "direct" - }, - "geosite": { - "download_url": "https://github.com/soffchen/sing-geosite/releases/latest/download/geosite.db", - "download_detour": "direct" - }, - "rules": [ - { - "protocol": "dns", - "outbound": "dns-out" - }, - { - "geoip": "private", - "outbound": "direct" - } - ], - "auto_detect_interface": true - }, - "experimental": {} -} diff --git a/app/templates/subscription/index.html b/app/templates/subscription/index.html index f26f825..3ec7039 100644 --- a/app/templates/subscription/index.html +++ b/app/templates/subscription/index.html @@ -85,8 +85,8 @@ {% if not user.expire %} ∞ {% else %} - {% set current_timestamp = now().timestamp() %} - {% set remaining_days = ((user.expire - current_timestamp) // (24 * 3600)) %} + {% set current_datetime = now() %} + {% set remaining_days = ((user.expire - current_datetime).days) %} {{ user.expire | datetime }} ({{ remaining_days | int }} days remaining) {% endif %}

diff --git a/app/utils/share.py b/app/utils/share.py index 3290a77..59a170d 100644 --- a/app/utils/share.py +++ b/app/utils/share.py @@ -2,30 +2,23 @@ import base64 import json import random import secrets -import urllib.parse as urlparse from datetime import datetime as dt, timedelta from typing import TYPE_CHECKING, Literal, Union, List from uuid import UUID -import yaml from jdatetime import date as jd +from v2share import V2Data, SingBoxConfig, ClashConfig, ClashMetaConfig +from app.db import GetDB, crud from app.models.proxy import ( - FormatVariables, - InboundBase, - InboundHost, InboundHostSecurity, ) -from app.templates import render_template +from app.utils.keygen import gen_uuid, gen_password from app.utils.system import get_public_ip, readable_size -from app.utils.keygen import generate_settings -from app.db import GetDB, crud if TYPE_CHECKING: from app.models.user import UserResponse -from config import CLASH_SUBSCRIPTION_TEMPLATE, SINGBOX_SUBSCRIPTION_TEMPLATE - SERVER_IP = get_public_ip() STATUS_EMOJIS = { @@ -37,734 +30,18 @@ STATUS_EMOJIS = { } -class V2rayShareLink(str): - @classmethod - def vmess( - cls, - remark: str, - address: str, - port: int, - id: Union[str, UUID], - host="", - net="tcp", - path="", - type="", - tls="none", - sni="", - fp="", - alpn="", - pbk="", - sid="", - spx="", - ais="", - ): - payload = { - "add": address, - "aid": "0", - "host": host, - "id": str(id), - "net": net, - "path": path, - "port": port, - "ps": remark, - "scy": "auto", - "tls": tls, - "type": type, - "v": "2", - } - - if tls == "tls": - payload["sni"] = sni - payload["fp"] = fp - payload["alpn"] = alpn - if ais: - payload["allowInsecure"] = 1 - elif tls == "reality": - payload["sni"] = sni - payload["fp"] = fp - payload["pbk"] = pbk - payload["sid"] = sid - payload["spx"] = spx - - return ( - "vmess://" - + base64.b64encode( - json.dumps(payload, sort_keys=True).encode("utf-8") - ).decode() - ) - - @classmethod - def vless( - cls, - remark: str, - address: str, - port: int, - id: Union[str, UUID], - net="ws", - path="", - host="", - type="", - flow="", - tls="none", - sni="", - fp="", - alpn="", - pbk="", - sid="", - spx="", - ais="", - ): - payload = {"security": tls, "type": net, "host": host, "headerType": type} - if flow and (net in ("tcp", "kcp") and type != "http"): - payload["flow"] = flow - - if net == "grpc": - payload["serviceName"] = path - else: - payload["path"] = path - - if tls == "tls": - payload["sni"] = sni - payload["fp"] = fp - payload["alpn"] = alpn - if ais: - payload["allowInsecure"] = 1 - elif tls == "reality": - payload["sni"] = sni - payload["fp"] = fp - payload["pbk"] = pbk - payload["sid"] = sid - payload["spx"] = spx - - return ( - "vless://" - + f"{id}@{address}:{port}?" - + urlparse.urlencode(payload) - + f"#{( urlparse.quote(remark))}" - ) - - @classmethod - def trojan( - cls, - remark: str, - address: str, - port: int, - password: str, - net="tcp", - path="", - host="", - type="", - flow="", - tls="none", - sni="", - fp="", - alpn="", - pbk="", - sid="", - spx="", - ais="", - ): - payload = {"security": tls, "type": net, "host": host, "headerType": type} - if flow and (net in ("tcp", "kcp") and type != "http"): - payload["flow"] = flow - - if net == "grpc": - payload["serviceName"] = path - else: - payload["path"] = path - - if tls == "tls": - payload["sni"] = sni - payload["fp"] = fp - payload["alpn"] = alpn - if ais: - payload["allowInsecure"] = 1 - elif tls == "reality": - payload["sni"] = sni - payload["fp"] = fp - payload["pbk"] = pbk - payload["sid"] = sid - payload["spx"] = spx - - return ( - "trojan://" - + f"{urlparse.quote(password, safe=':')}@{address}:{port}?" - + urlparse.urlencode(payload) - + f"#{urlparse.quote(remark)}" - ) - - @classmethod - def shadowsocks( - cls, remark: str, address: str, port: int, password: str, method: str - ): - return ( - "ss://" - + base64.b64encode(f"{method}:{password}".encode()).decode() - + f"@{address}:{port}#{urlparse.quote(remark)}" - ) - - -class ClashConfiguration(object): - def __init__(self): - self.data = { - "proxies": [], - "proxy-groups": [], - # Some clients rely on "rules" option and will fail without it. - "rules": [], - } - self.proxy_remarks = [] - - def render(self): - return yaml.dump( - yaml.load( - render_template( - CLASH_SUBSCRIPTION_TEMPLATE, - {"conf": self.data, "proxy_remarks": self.proxy_remarks}, - ), - Loader=yaml.SafeLoader, - ), - sort_keys=False, - allow_unicode=True, - ) - - def __str__(self) -> str: - return self.render() - - def __repr__(self) -> str: - return self.render() - - def _remark_validation(self, remark): - if not remark in self.proxy_remarks: - return remark - c = 2 - while True: - new = f"{remark} ({c})" - if not new in self.proxy_remarks: - return new - c += 1 - - def make_node( - self, - name: str, - type: str, - server: str, - port: int, - network: str, - tls: bool, - sni: str, - host: str, - path: str, - udp: bool = True, - alpn: str = "", - ais: bool = "", - ): - if type == "shadowsocks": - type = "ss" - - remark = self._remark_validation(name) - node = { - "name": remark, - "type": type, - "server": server, - "port": port, - "network": network, - f"{network}-opts": {}, - "udp": udp, - } - - if type == "ss": # shadowsocks - return node - - if tls: - node["tls"] = True - if type == "trojan": - node["sni"] = sni - else: - node["servername"] = sni - if alpn: - node["alpn"] = alpn.split(",") - if ais: - node["skip-cert-verify"] = ais - - net_opts = node[f"{network}-opts"] - - if network == "ws": - if path: - net_opts["path"] = path - if host: - net_opts["headers"] = {"Host": host} - - if network == "grpc": - if path: - net_opts["grpc-service-name"] = path - - if network == "h2": - if path: - net_opts["path"] = path - if host: - net_opts["host"] = [host] - - if network == "http" or network == "tcp": - if path: - net_opts["method"] = "GET" - net_opts["path"] = [path] - if host: - net_opts["method"] = "GET" - net_opts["headers"] = {"Host": host} - - return node - - def add(self, remark: str, address: str, inbound: dict, settings: dict): - node = self.make_node( - name=remark, - type=inbound["protocol"], - server=address, - port=inbound["port"], - network=inbound["network"], - tls=(inbound["tls"] == "tls"), - sni=inbound["sni"], - host=inbound["host"], - path=inbound["path"], - udp=True, - alpn=inbound.get("alpn", ""), - ais=inbound.get("ais", ""), - ) - - if inbound["protocol"] == "vmess": - node["uuid"] = settings["id"] - node["alterId"] = 0 - node["cipher"] = "auto" - self.data["proxies"].append(node) - self.proxy_remarks.append(remark) - - if inbound["protocol"] == "trojan": - node["password"] = settings["password"] - self.data["proxies"].append(node) - self.proxy_remarks.append(remark) - - if inbound["protocol"] == "shadowsocks": - node["password"] = settings["password"] - node["cipher"] = settings["method"] - self.data["proxies"].append(node) - self.proxy_remarks.append(remark) - - -class ClashMetaConfiguration(ClashConfiguration): - def make_node( - self, - name: str, - type: str, - server: str, - port: int, - network: str, - tls: bool, - sni: str, - host: str, - path: str, - udp: bool = True, - alpn: str = "", - fp: str = "", - pbk: str = "", - sid: str = "", - ais: bool = "", - ): - node = super().make_node( - name=name, - type=type, - server=server, - port=port, - network=network, - tls=tls, - sni=sni, - host=host, - path=path, - # headers="", - udp=udp, - alpn=alpn, - ais=ais, - ) - if fp: - node["client-fingerprint"] = fp - if pbk: - node["reality-opts"] = {"public-key": pbk, "short-id": sid} - - return node - - def add(self, remark: str, address: str, inbound: dict, settings: dict): - node = self.make_node( - name=remark, - type=inbound["protocol"], - server=address, - port=inbound["port"], - network=inbound["network"], - tls=(inbound["tls"] in ("tls", "reality")), - sni=inbound["sni"], - host=inbound["host"], - path=inbound["path"], - udp=True, - alpn=inbound.get("alpn", ""), - fp=inbound.get("fp", ""), - pbk=inbound.get("pbk", ""), - sid=inbound.get("sid", ""), - ais=inbound.get("ais", ""), - ) - - if inbound["protocol"] == "vmess": - node["uuid"] = settings["id"] - node["alterId"] = 0 - node["cipher"] = "auto" - self.data["proxies"].append(node) - self.proxy_remarks.append(remark) - - if inbound["protocol"] == "vless": - node["uuid"] = settings["id"] - - if ( - inbound["network"] in ("tcp", "kcp") - and inbound["header_type"] != "http" - ): - node["flow"] = settings.get("flow", "") - - self.data["proxies"].append(node) - self.proxy_remarks.append(remark) - - if inbound["protocol"] == "trojan": - node["password"] = settings["password"] - - if ( - inbound["network"] in ("tcp", "kcp") - and inbound["header_type"] != "http" - ): - node["flow"] = settings.get("flow", "") - - self.data["proxies"].append(node) - self.proxy_remarks.append(remark) - - if inbound["protocol"] == "shadowsocks": - node["password"] = settings["password"] - node["cipher"] = settings["method"] - self.data["proxies"].append(node) - self.proxy_remarks.append(remark) - - -def get_v2ray_link(remark: str, address: str, inbound: dict, settings: dict): - if inbound["protocol"] == "vmess": - return V2rayShareLink.vmess( - remark=remark, - address=address, - port=inbound["port"], - id=settings["id"], - net=inbound["network"], - tls=inbound["tls"], - sni=inbound.get("sni", ""), - fp=inbound.get("fp", ""), - alpn=inbound.get("alpn", ""), - pbk=inbound.get("pbk", ""), - sid=inbound.get("sid", ""), - spx=inbound.get("spx", ""), - host=inbound["host"], - path=inbound["path"], - type=inbound["header_type"], - ais=inbound.get("ais", ""), - ) - - if inbound["protocol"] == "vless": - return V2rayShareLink.vless( - remark=remark, - address=address, - port=inbound["port"], - id=settings["id"], - flow=settings.get("flow", ""), - net=inbound["network"], - tls=inbound["tls"], - sni=inbound.get("sni", ""), - fp=inbound.get("fp", ""), - alpn=inbound.get("alpn", ""), - pbk=inbound.get("pbk", ""), - sid=inbound.get("sid", ""), - spx=inbound.get("spx", ""), - host=inbound["host"], - path=inbound["path"], - type=inbound["header_type"], - ais=inbound.get("ais", ""), - ) - - if inbound["protocol"] == "trojan": - return V2rayShareLink.trojan( - remark=remark, - address=address, - port=inbound["port"], - password=settings["password"], - flow=settings.get("flow", ""), - net=inbound["network"], - tls=inbound["tls"], - sni=inbound.get("sni", ""), - fp=inbound.get("fp", ""), - alpn=inbound.get("alpn", ""), - pbk=inbound.get("pbk", ""), - sid=inbound.get("sid", ""), - spx=inbound.get("spx", ""), - host=inbound["host"], - path=inbound["path"], - type=inbound["header_type"], - ais=inbound.get("ais", ""), - ) - - if inbound["protocol"] == "shadowsocks": - return V2rayShareLink.shadowsocks( - remark=remark, - address=address, - port=inbound["port"], - password=settings["password"], - method=settings["method"], - ) - - -class OutlineConfiguration: - def __init__(self): - self.config = {} - - def add_directly(self, data: dict): - self.config.update(data) - - def render(self): - return json.dumps(self.config, indent=0) - - def make_outbound( - self, remark: str, address: str, port: int, password: str, method: str - ): - config = { - "method": method, - "password": password, - "server": address, - "server_port": port, - "tag": remark, - } - return config - - def add(self, remark: str, address: str, inbound: dict, settings: dict): - if inbound["protocol"] != "shadowsocks": - return - - outbound = self.make_outbound( - remark=remark, - address=address, - port=inbound["port"], - password=settings["password"], - method=settings["method"], - ) - self.add_directly(outbound) - - -class SingBoxConfiguration(str): - def __init__(self): - template = render_template(SINGBOX_SUBSCRIPTION_TEMPLATE) - self.config = json.loads(template) - - def add_outbound(self, outbound_data): - self.config["outbounds"].append(outbound_data) - - def render(self): - urltest_types = ["vmess", "vless", "trojan", "shadowsocks"] - urltest_tags = [ - outbound["tag"] - for outbound in self.config["outbounds"] - if outbound["type"] in urltest_types - ] - selector_types = ["vmess", "vless", "trojan", "shadowsocks", "urltest"] - selector_tags = [ - outbound["tag"] - for outbound in self.config["outbounds"] - if outbound["type"] in selector_types - ] - - for outbound in self.config["outbounds"]: - if outbound.get("type") == "urltest": - outbound["outbounds"] = urltest_tags - - for outbound in self.config["outbounds"]: - if outbound.get("type") == "selector": - outbound["outbounds"] = selector_tags - - return json.dumps(self.config, indent=4) - - @staticmethod - def tls_config( - sni=None, fp=None, tls=None, pbk=None, sid=None, alpn=None, ais=None - ): - config = {} - if tls in ["tls", "reality"]: - config["enabled"] = True - - if sni is not None: - config["server_name"] = sni - - if tls == "tls" and ais: - config["insecure"] = ais - - if tls == "reality": - config["reality"] = {"enabled": True} - if pbk: - config["reality"]["public_key"] = pbk - if sid: - config["reality"]["short_id"] = sid - - if fp: - config["utls"] = {"enabled": bool(fp), "fingerprint": fp} - - if alpn: - config["alpn"] = [alpn] if not isinstance(alpn, list) else alpn - - return config - - @staticmethod - def transport_config( - transport_type="", - host="", - path="", - method="", - headers="", - idle_timeout="15s", - ping_timeout="15s", - max_early_data=None, - early_data_header_name=None, - permit_without_stream=False, - ): - transport_config = {} - - if transport_type: - transport_config["type"] = transport_type - - if transport_type == "http": - if host: - transport_config["host"] = host - if path: - transport_config["path"] = path - if method: - transport_config["method"] = method - if headers: - transport_config["headers"] = headers - if idle_timeout: - transport_config["idle_timeout"] = idle_timeout - if ping_timeout: - transport_config["ping_timeout"] = ping_timeout - - elif transport_type == "ws": - if path: - transport_config["path"] = path - if headers: - transport_config["headers"] = headers - if max_early_data is not None: - transport_config["max_early_data"] = max_early_data - if early_data_header_name: - transport_config["early_data_header_name"] = early_data_header_name - - elif transport_type == "ws": - if path: - transport_config["service_name"] = path - if idle_timeout: - transport_config["idle_timeout"] = idle_timeout - if ping_timeout: - transport_config["ping_timeout"] = ping_timeout - if permit_without_stream: - transport_config["permit_without_stream"] = permit_without_stream - - return transport_config - - def make_outbound( - self, - type: str, - remark: str, - address: str, - port: int, - net="ws", - path="", - host="", - flow="", - tls="", - sni="", - fp="", - alpn="", - pbk="", - sid="", - headers="", - ais="", - ): - config = { - "type": type, - "tag": remark, - "server": address, - "server_port": port, - } - if net in ("tcp", "kcp") or headers != "http": - if flow: - config["flow"] = flow - - if net in ["http", "ws", "quic", "grpc", "h2"]: - if net == "h2": - net = "http" - alpn = "h2" - config["transport"] = self.transport_config( - transport_type=net, host=host, path=path, headers=headers - ) - else: - config["network"]: net - - if tls in ("tls", "reality"): - config["tls"] = self.tls_config( - sni=sni, fp=fp, tls=tls, pbk=pbk, sid=sid, alpn=alpn, ais=ais - ) - - return config - - def add(self, remark: str, address: str, inbound: dict, settings: dict): - outbound = self.make_outbound( - remark=remark, - type=inbound["protocol"], - address=address, - port=inbound["port"], - net=inbound["network"], - tls=(inbound["tls"]), - flow=settings.get("flow", ""), - sni=inbound["sni"], - host=inbound["host"], - path=inbound["path"], - alpn=inbound.get("alpn", ""), - fp=inbound.get("fp", ""), - pbk=inbound.get("pbk", ""), - sid=inbound.get("sid", ""), - headers=inbound["header_type"], - ais=inbound.get("ais", ""), - ) - - if inbound["protocol"] == "vmess": - outbound["uuid"] = settings["id"] - - elif inbound["protocol"] == "vless": - outbound["uuid"] = settings["id"] - - elif inbound["protocol"] == "trojan": - outbound["password"] = settings["password"] - - elif inbound["protocol"] == "shadowsocks": - outbound["password"] = settings["password"] - outbound["method"] = settings["method"] - - self.add_outbound(outbound) - - def generate_v2ray_links(inbounds: list, key: str, extra_data: dict) -> list: format_variables = setup_format_variables(extra_data) return process_inbounds_and_tags(inbounds, key, format_variables, mode="v2ray") def generate_clash_subscription( - inbounds: dict, key: str, extra_data: dict, is_meta: bool = False + inbounds: list, key: str, extra_data: dict, is_meta: bool = False ) -> str: if is_meta is True: - conf = ClashMetaConfiguration() + conf = ClashMetaConfig() else: - conf = ClashConfiguration() + conf = ClashConfig() format_variables = setup_format_variables(extra_data) return process_inbounds_and_tags( @@ -772,8 +49,8 @@ def generate_clash_subscription( ) -def generate_singbox_subscription(inbounds: dict, key: str, extra_data: dict) -> str: - conf = SingBoxConfiguration() +def generate_singbox_subscription(inbounds: list, key: str, extra_data: dict) -> str: + conf = SingBoxConfig() format_variables = setup_format_variables(extra_data) return process_inbounds_and_tags( @@ -785,18 +62,9 @@ def generate_v2ray_subscription(links: list) -> str: return base64.b64encode("\n".join(links).encode()).decode() -def generate_outline_subscription(inbounds: dict, key: str, extra_data: dict) -> str: - conf = OutlineConfiguration() - - format_variables = setup_format_variables(extra_data) - return process_inbounds_and_tags( - inbounds, key, format_variables, mode="outline", conf=conf - ) - - def generate_subscription( user: "UserResponse", - config_format: Literal["v2ray", "clash-meta", "clash", "sing-box", "outline"], + config_format: Literal["v2ray", "clash-meta", "clash", "sing-box"], as_base64: bool, ) -> str: kwargs = { @@ -814,8 +82,6 @@ def generate_subscription( config = generate_clash_subscription(**kwargs) elif config_format == "sing-box": config = generate_singbox_subscription(**kwargs) - elif config_format == "outline": - config = generate_outline_subscription(**kwargs) else: raise ValueError(f'Unsupported format "{config_format}"') @@ -854,13 +120,12 @@ from app.models.user import UserStatus def setup_format_variables(extra_data: dict) -> dict: user_status = extra_data.get("status") - expire_timestamp = extra_data.get("expire") + expire_datetime = extra_data.get("expire") on_hold_expire_duration = extra_data.get("on_hold_expire_duration") if user_status != UserStatus.on_hold: - if expire_timestamp is not None and expire_timestamp >= 0: - seconds_left = expire_timestamp - int(dt.utcnow().timestamp()) - expire_datetime = dt.fromtimestamp(expire_timestamp) + if expire_datetime is not None: + seconds_left = (expire_datetime - dt.utcnow()).total_seconds() expire_date = expire_datetime.date() jalali_expire_date = jd.fromgregorian( year=expire_date.year, month=expire_date.month, day=expire_date.day @@ -929,17 +194,15 @@ def process_inbounds_and_tags( inb.protocol, inb.tag, ) - settings = generate_settings(ukey, protocol) format_variables.update({"PROTOCOL": protocol.name}) if not inbound: continue format_variables.update({"TRANSPORT": inbound["network"]}) - host_inbound = inbound.copy() with GetDB() as db: hosts = crud.get_inbound_hosts(db, inb_id) - # for host in xray.hosts.get(tag, []): + for host in hosts: host_snis = host.sni.split(",") if host.sni else [] sni_list = host_snis or inbound["sni"] @@ -960,41 +223,31 @@ def process_inbounds_and_tags( if host.security == InboundHostSecurity.inbound_default else host.security.value ) - host_inbound.update( - { - "port": host.port or inbound["port"], - "sni": sni, - "host": req_host, - "tls": inbound["tls"] if host_tls is None else host_tls, - "alpn": host.alpn.value or inbound.get("alpn", ""), - "path": ( - inbound.get("path", "") - if host.path is None - else host.path.format_map(format_variables) - ), - "fp": host.fingerprint.value or inbound.get("fp", ""), - "ais": host.allowinsecure or "", - } + data = V2Data( + protocol.value, + host.remark.format_map(format_variables), + host.address.format_map(format_variables), + host.port or inbound["port"], + sni=sni, + host=req_host, + tls=host_tls or inbound["tls"], + alpn=host.alpn.value or inbound.get("alpn", ""), + path=( + host.path.format_map(format_variables) + if host.path + else inbound.get("path", "") + ), + fingerprint=host.fingerprint.value, + allow_insecure=host.allowinsecure, + uuid=UUID(gen_uuid(ukey)), + password=gen_password(ukey), ) - if mode == "v2ray": - results.append( - get_v2ray_link( - remark=host.remark.format_map(format_variables), - address=host.address.format_map(format_variables), - inbound=host_inbound, - settings=settings, - ) - ) - elif mode in ["clash", "sing-box", "outline"]: - conf.add( - remark=host.remark.format_map(format_variables), - address=host.address.format_map(format_variables), - inbound=host_inbound, - settings=settings, - ) - - if mode in ["clash", "sing-box", "outline"]: + results.append(data.to_link()) + elif mode in ["clash", "sing-box"]: + conf.add_proxies([data]) + + if mode in ["clash", "sing-box"]: return conf.render() return results diff --git a/requirements.txt b/requirements.txt index e76800f..f4b7dd0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -35,3 +35,4 @@ python-multipart==0.0.7 xxhash==3.4.1 grpcio==1.60.0 psycopg==3.1.18 +v2share==0.1.0a5 \ No newline at end of file