371 lines
10 KiB
Python
371 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
import uuid
|
|
from abc import ABC, abstractmethod
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass, make_dataclass
|
|
from typing import TYPE_CHECKING, TypeVar
|
|
from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
|
|
|
|
from .outbound_fields import OUTBOUND_FIELDS
|
|
|
|
if TYPE_CHECKING:
|
|
from .target import Target, TargetStorage
|
|
from .xray_config import XrayConfig
|
|
from .xray_manager_config import XrayManagerConfig
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OutboundSpec:
|
|
protocol: str
|
|
target_id: str
|
|
|
|
|
|
O = TypeVar("O", bound="Outbound")
|
|
T = TypeVar("T")
|
|
|
|
|
|
class OutboundFactory:
|
|
_registry: dict[str, type[Outbound]] = {}
|
|
_scheme_registry: dict[str, type[Outbound]] = {}
|
|
|
|
@classmethod
|
|
def register(cls, outbound_cls: type[O]) -> type[O]:
|
|
protocol = outbound_cls.PROTOCOL
|
|
scheme = outbound_cls.LINK_SCHEME
|
|
|
|
if protocol in cls._registry:
|
|
raise RuntimeError(f"Protocol already registered: {protocol}")
|
|
|
|
if scheme in cls._scheme_registry:
|
|
raise RuntimeError(f"Scheme already registered: {scheme}")
|
|
|
|
cls._registry[protocol] = outbound_cls
|
|
cls._scheme_registry[scheme] = outbound_cls
|
|
|
|
return outbound_cls
|
|
|
|
@classmethod
|
|
def get_protocols(cls) -> list[str]:
|
|
return list(cls._registry)
|
|
|
|
@classmethod
|
|
def from_link(cls, target_storage: TargetStorage, link: str) -> Outbound:
|
|
scheme = link.split("://", 1)[0]
|
|
|
|
if scheme not in cls._scheme_registry:
|
|
raise ValueError(f"Unsupported link scheme: {scheme}")
|
|
|
|
outbound_cls = cls._scheme_registry[scheme]
|
|
return outbound_cls.from_link(target_storage, link)
|
|
|
|
@classmethod
|
|
def from_inbound(
|
|
cls,
|
|
target_storage: TargetStorage,
|
|
host: str,
|
|
client: dict,
|
|
inbound: dict,
|
|
):
|
|
protocol = inbound.get("protocol")
|
|
|
|
if protocol not in cls._registry:
|
|
raise ValueError(f"Unsupported protocol: {protocol}")
|
|
|
|
outbound_cls = cls._registry[protocol]
|
|
username, outbound = outbound_cls.from_inbound(
|
|
target_storage, host, client, inbound
|
|
)
|
|
return username, outbound
|
|
|
|
@classmethod
|
|
def from_spec(
|
|
cls,
|
|
xray_config: XrayConfig,
|
|
target_storage: TargetStorage,
|
|
spec: OutboundSpec,
|
|
) -> Outbound:
|
|
if spec.protocol not in cls._registry:
|
|
raise ValueError(f"Unsupported protocol: {spec.protocol}")
|
|
|
|
target = target_storage.load_by_id(spec.target_id)
|
|
|
|
inbound = xray_config.find_managed_inbound_by_protocol(spec.protocol)
|
|
outbound_cls = cls._registry[spec.protocol]
|
|
|
|
return outbound_cls.from_scratch(target, xray_config.host, inbound)
|
|
|
|
|
|
class Outbound(ABC):
|
|
PROTOCOL: str
|
|
LINK_SCHEME: str
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def from_link(
|
|
cls, target_storage: TargetStorage, link: str
|
|
) -> Outbound: ...
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def from_inbound(
|
|
cls,
|
|
target_storage: TargetStorage,
|
|
host: str,
|
|
client: dict,
|
|
inbound: dict,
|
|
) -> tuple[str, Outbound]: ...
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def from_scratch(
|
|
cls, target: Target, host: str, inbound: dict
|
|
) -> Outbound: ...
|
|
|
|
@abstractmethod
|
|
def to_link(self) -> str: ...
|
|
|
|
@abstractmethod
|
|
def matches_inbound(self, inbound: dict) -> bool: ...
|
|
|
|
@abstractmethod
|
|
def add_to_inbound(self, inbound: dict, username: str) -> None: ...
|
|
|
|
@abstractmethod
|
|
def delete_from_inbound(self, inbound: dict, username: str) -> None: ...
|
|
|
|
@property
|
|
@abstractmethod
|
|
def protocol(self) -> str: ...
|
|
|
|
@staticmethod
|
|
def split_client_email(email: str) -> tuple[str, str]:
|
|
username, target = email.rsplit("-", 1)
|
|
return username, target
|
|
|
|
@staticmethod
|
|
def find_target_by_pretty_name(
|
|
targets: dict[str, dict[str, str]], pretty_name: str
|
|
) -> str | None:
|
|
search_name = pretty_name.strip()
|
|
for target, target_data in targets.items():
|
|
if target_data.get("pretty_name", "").strip() == search_name:
|
|
return target
|
|
return None
|
|
|
|
@abstractmethod
|
|
def __hash__(self) -> int: ...
|
|
|
|
@abstractmethod
|
|
def __eq__(self, other: object) -> bool: ...
|
|
|
|
|
|
_ShadowsocksFields = make_dataclass(
|
|
"_ShadowsocksFields", fields=OUTBOUND_FIELDS["shadowsocks"]
|
|
)
|
|
|
|
|
|
@OutboundFactory.register
|
|
class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
|
|
PROTOCOL = "shadowsocks"
|
|
LINK_SCHEME = "ss"
|
|
|
|
@classmethod
|
|
def from_link(
|
|
cls, target_storage: TargetStorage, link: str
|
|
) -> ShadowsocksOutbound:
|
|
import base64
|
|
|
|
prefix_b64, rest = link[5:].split("@", 1)
|
|
host_port, quoted_target_pretty_name = rest.split("#", 1)
|
|
host, port_str = host_port.split(":")
|
|
port = int(port_str)
|
|
|
|
prefix_bytes = base64.urlsafe_b64decode(prefix_b64)
|
|
prefix_str = prefix_bytes.decode()
|
|
method, server_password, client_password = prefix_str.split(":")
|
|
|
|
target_pretty_name = unquote(quoted_target_pretty_name)
|
|
target = target_storage.load_by_pretty(target_pretty_name)
|
|
|
|
return cls(
|
|
host=host,
|
|
port=port,
|
|
target=target,
|
|
method=method,
|
|
server_password=server_password,
|
|
client_password=client_password,
|
|
)
|
|
|
|
@classmethod
|
|
def from_inbound(
|
|
cls,
|
|
target_storage: TargetStorage,
|
|
host: str,
|
|
client: dict,
|
|
inbound: dict,
|
|
) -> tuple[str, ShadowsocksOutbound]:
|
|
port = inbound["port"]
|
|
method = inbound["settings"]["method"]
|
|
server_password = inbound["settings"]["password"]
|
|
client_password = client["password"]
|
|
email = client["email"]
|
|
|
|
username, target_id = cls.split_client_email(email)
|
|
|
|
target = target_storage.load_by_id(target_id)
|
|
|
|
outbound = cls(
|
|
host=host,
|
|
port=port,
|
|
target=target,
|
|
method=method,
|
|
server_password=server_password,
|
|
client_password=client_password,
|
|
)
|
|
return username, outbound
|
|
|
|
@classmethod
|
|
def from_scratch(
|
|
cls, target: Target, host: str, inbound: dict
|
|
) -> ShadowsocksOutbound:
|
|
port = inbound["port"]
|
|
method = inbound["settings"]["method"]
|
|
server_password = inbound["settings"]["password"]
|
|
client_password = cls.generate_password(method)
|
|
|
|
return cls(
|
|
host=host,
|
|
port=port,
|
|
target=target,
|
|
method=method,
|
|
server_password=server_password,
|
|
client_password=client_password,
|
|
)
|
|
|
|
def to_link(self) -> str:
|
|
import base64
|
|
|
|
prefix = f"{self.method}:{self.server_password}:{self.client_password}"
|
|
prefix_b64 = base64.urlsafe_b64encode(prefix.encode()).decode()
|
|
tag = quote(self.target.pretty)
|
|
link = f"ss://{prefix_b64}@{self.host}:{self.port}#{tag}"
|
|
return link
|
|
|
|
@staticmethod
|
|
def generate_password(method: str) -> str:
|
|
import base64
|
|
import secrets
|
|
|
|
key_lengths = {
|
|
"2022-blake3-aes-128-gcm": 16,
|
|
"2022-blake3-aes-256-gcm": 32,
|
|
"2022-blake3-chacha20-poly1305": 32,
|
|
}
|
|
|
|
try:
|
|
length = key_lengths[method]
|
|
except KeyError:
|
|
raise ValueError(f"Unsupported shadowsocks method: {method}")
|
|
|
|
key = secrets.token_bytes(length)
|
|
|
|
return base64.b64encode(key).decode()
|
|
|
|
def matches_inbound(self, inbound: dict):
|
|
if inbound.get("protocol") != self.PROTOCOL:
|
|
return False
|
|
|
|
if inbound.get("port") != self.port:
|
|
return False
|
|
|
|
settings = inbound.get("settings", {})
|
|
|
|
if settings.get("method") != self.method:
|
|
return False
|
|
|
|
if settings.get("password") != self.server_password:
|
|
return False
|
|
|
|
return True
|
|
|
|
def add_to_inbound(self, inbound: dict, username: str) -> None:
|
|
client = {
|
|
"email": f"{username}-{self.target}",
|
|
"password": self.client_password,
|
|
}
|
|
|
|
clients = inbound["settings"].setdefault("clients", [])
|
|
clients.append(client)
|
|
|
|
clients_by_user = defaultdict(list)
|
|
for client in clients:
|
|
username, _ = self.split_client_email(client["email"])
|
|
clients_by_user[username].append(client)
|
|
|
|
priority = {"default": 0, "managed": 1, "relay": 2}
|
|
|
|
def sort_key(c):
|
|
_, target = self.split_client_email(c["email"])
|
|
return (priority.get(target, 100), c["email"])
|
|
|
|
for user_clients in clients_by_user.values():
|
|
user_clients.sort(key=sort_key)
|
|
|
|
sorted_clients = []
|
|
for username in sorted(clients_by_user):
|
|
sorted_clients.extend(clients_by_user[username])
|
|
|
|
inbound["settings"]["clients"] = sorted_clients
|
|
|
|
def delete_from_inbound(self, inbound: dict, username: str) -> None:
|
|
target_email = f"{username}-{self.target}"
|
|
settings = inbound["settings"]
|
|
clients = settings["clients"]
|
|
|
|
filtered_clients = []
|
|
for client in clients:
|
|
if client["email"] != target_email:
|
|
filtered_clients.append(client)
|
|
|
|
inbound["settings"]["clients"] = filtered_clients
|
|
|
|
@property
|
|
def protocol(self) -> str:
|
|
return self.PROTOCOL
|
|
|
|
def __hash__(self) -> int:
|
|
return hash(
|
|
(
|
|
self.host,
|
|
self.port,
|
|
self.target,
|
|
self.method,
|
|
self.server_password,
|
|
self.client_password,
|
|
)
|
|
)
|
|
|
|
def __eq__(self, other: object) -> bool:
|
|
if not isinstance(other, ShadowsocksOutbound):
|
|
return NotImplemented
|
|
|
|
return (
|
|
self.host == other.host
|
|
and self.port == other.port
|
|
and self.target == other.target
|
|
and self.method == other.method
|
|
and self.server_password == other.server_password
|
|
and self.client_password == other.client_password
|
|
)
|
|
|
|
|
|
_VlessFields = make_dataclass("_VlessFieldss", fields=OUTBOUND_FIELDS["vless"])
|
|
|
|
|
|
@OutboundFactory.register
|
|
class VlessOutbound(_VlessFields, Outbound):
|
|
PROTOCOL = "vless"
|
|
LINK_SCHEME = "vless"
|