diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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 %}</p>
diff --git 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
@@ -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