chore: создание каталога sandbox для хранения проверочных скриптов

README.md

@@ -7,10 +7,9 @@ user show (<username> | -a)
 user modify <username> [-a <protocol>:<target> [<protocol>:<target> ...]] [-d <protocol>:<target> [<protocol>:<target> ...]]
 user delete (<username> | -a)
 
-user outbound add <username> -p <protocol> -t <target> [--<param> <value> ...]
-user outbound show <username> (-a | -p <protocol> | -t <target>)
-user outbound modify <username> -p <protocol> -t <target> [--<param> <value> ...]
-user outbound delete <username> (-a | -p <protocol> | -t <target>)
+user outbound set <username> -protocol [--target <target_id>] [--<param> <value> ...]
+user outbound show <username> (-a | -p <protocol> [--target <target_id>])
+user outbound delete <username> (-a | -p <protocol> | --target <target_id>)
 
 user profile build (<username> | -a)
 user profile show (<username> | -a)

src/xray_manager/core/__init__.py

@@ -1,5 +1,14 @@
 from .connection import Connection, ConnectionFactory, ConnectionKey
+from .outbound import (
+    OUTBOUND_FIELDS,
+    Outbound,
+    OutboundFactory,
+    OutboundSpec,
+    ShadowsocksOutbound,
+    VlessOutbound,
+)
 from .profile import Profile, ProfileFactory, ProfileStorage
+from .target import Target, TargetStorage
 from .user import User, UserFactory
 from .xray_config import XrayConfig
 from .xray_manager_config import XrayManagerConfig

src/xray_manager/core/outbound.py

@@ -8,9 +8,11 @@ from dataclasses import dataclass, make_dataclass
 from typing import TYPE_CHECKING, TypeVar
 from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
 
+from ..utils import private_to_public
 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
 
@@ -18,7 +20,7 @@ if TYPE_CHECKING:
 @dataclass(frozen=True)
 class OutboundSpec:
     protocol: str
-    target: str
+    target_id: str
 
 
 O = TypeVar("O", bound="Outbound")
@@ -50,22 +52,19 @@ class OutboundFactory:
         return list(cls._registry)
 
     @classmethod
-    def from_link(
-        cls, xray_manager_config: XrayManagerConfig, link: str
-    ) -> Outbound:
+    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]
-        targets = xray_manager_config.targets
-        return outbound_cls.from_link(targets, link)
+        return outbound_cls.from_link(target_storage, link)
 
     @classmethod
     def from_inbound(
         cls,
-        xray_manager_config: XrayManagerConfig,
+        target_storage: TargetStorage,
         host: str,
         client: dict,
         inbound: dict,
@@ -76,9 +75,8 @@ class OutboundFactory:
             raise ValueError(f"Unsupported protocol: {protocol}")
 
         outbound_cls = cls._registry[protocol]
-        targets = xray_manager_config.targets
         username, outbound = outbound_cls.from_inbound(
-            targets, host, client, inbound
+            target_storage, host, client, inbound
         )
         return username, outbound
 
@@ -86,26 +84,18 @@ class OutboundFactory:
     def from_spec(
         cls,
         xray_config: XrayConfig,
-        xray_manager_config: XrayManagerConfig,
+        target_storage: TargetStorage,
         spec: OutboundSpec,
     ) -> Outbound:
         if spec.protocol not in cls._registry:
             raise ValueError(f"Unsupported protocol: {spec.protocol}")
 
-        if spec.target not in xray_manager_config.get_targets_list():
-            raise ValueError(f"Target {spec.target} not found")
+        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]
 
-        targets = xray_manager_config.targets
-        target_pretty_name = targets.get(spec.target, {}).get(
-            "pretty_name", spec.target
-        )
-
-        return outbound_cls.from_scratch(
-            spec.target, target_pretty_name, xray_config.host, inbound
-        )
+        return outbound_cls.from_scratch(target, xray_config.host, inbound)
 
 
 class Outbound(ABC):
@@ -115,14 +105,14 @@ class Outbound(ABC):
     @classmethod
     @abstractmethod
     def from_link(
-        cls, targets: dict[str, dict[str, str]], link: str
+        cls, target_storage: TargetStorage, link: str
     ) -> Outbound: ...
 
     @classmethod
     @abstractmethod
     def from_inbound(
         cls,
-        targets: dict[str, dict[str, str]],
+        target_storage: TargetStorage,
         host: str,
         client: dict,
         inbound: dict,
@@ -131,7 +121,7 @@ class Outbound(ABC):
     @classmethod
     @abstractmethod
     def from_scratch(
-        cls, target: str, target_pretty_name: str, host: str, inbound: dict
+        cls, target: Target, host: str, inbound: dict
     ) -> Outbound: ...
 
     @abstractmethod
@@ -155,16 +145,6 @@ class Outbound(ABC):
         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: ...
 
@@ -184,7 +164,7 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
 
     @classmethod
     def from_link(
-        cls, targets: dict[str, dict[str, str]], link: str
+        cls, target_storage: TargetStorage, link: str
     ) -> ShadowsocksOutbound:
         import base64
 
@@ -198,15 +178,12 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
         method, server_password, client_password = prefix_str.split(":")
 
         target_pretty_name = unquote(quoted_target_pretty_name)
-        target = cls.find_target_by_pretty_name(targets, target_pretty_name)
-        if target is None:
-            target = "unknown"
+        target = target_storage.load_by_pretty(target_pretty_name)
 
         return cls(
             host=host,
             port=port,
             target=target,
-            target_pretty_name=target_pretty_name,
             method=method,
             server_password=server_password,
             client_password=client_password,
@@ -215,7 +192,7 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
     @classmethod
     def from_inbound(
         cls,
-        targets: dict[str, dict[str, str]],
+        target_storage: TargetStorage,
         host: str,
         client: dict,
         inbound: dict,
@@ -225,15 +202,15 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
         server_password = inbound["settings"]["password"]
         client_password = client["password"]
         email = client["email"]
-        username, target = cls.split_client_email(email)
 
-        target_pretty_name = targets.get(target, {}).get("pretty_name", target)
+        username, target_id = cls.split_client_email(email)
+
+        target = target_storage.load_by_id(target_id)
 
         outbound = cls(
             host=host,
             port=port,
             target=target,
-            target_pretty_name=target_pretty_name,
             method=method,
             server_password=server_password,
             client_password=client_password,
@@ -242,7 +219,7 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
 
     @classmethod
     def from_scratch(
-        cls, target: str, target_pretty_name: str, host: str, inbound: dict
+        cls, target: Target, host: str, inbound: dict
     ) -> ShadowsocksOutbound:
         port = inbound["port"]
         method = inbound["settings"]["method"]
@@ -253,7 +230,6 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
             host=host,
             port=port,
             target=target,
-            target_pretty_name=target_pretty_name,
             method=method,
             server_password=server_password,
             client_password=client_password,
@@ -264,7 +240,7 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
 
         prefix = f"{self.method}:{self.server_password}:{self.client_password}"
         prefix_b64 = base64.urlsafe_b64encode(prefix.encode()).decode()
-        tag = quote(self.target_pretty_name)
+        tag = quote(self.target.pretty)
         link = f"ss://{prefix_b64}@{self.host}:{self.port}#{tag}"
         return link
 
@@ -307,7 +283,7 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
 
     def add_to_inbound(self, inbound: dict, username: str) -> None:
         client = {
-            "email": f"{username}-{self.target}",
+            "email": f"{username}-{self.target.id}",
             "password": self.client_password,
         }
 
@@ -322,8 +298,8 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
         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"])
+            _, target_id = self.split_client_email(c["email"])
+            return (priority.get(target_id, 100), c["email"])
 
         for user_clients in clients_by_user.values():
             user_clients.sort(key=sort_key)
@@ -335,7 +311,7 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
         inbound["settings"]["clients"] = sorted_clients
 
     def delete_from_inbound(self, inbound: dict, username: str) -> None:
-        target_email = f"{username}-{self.target}"
+        target_email = f"{username}-{self.target.id}"
         settings = inbound["settings"]
         clients = settings["clients"]
 
@@ -376,10 +352,299 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
         )
 
 
-_VlessFields = make_dataclass("_VlessFieldss", fields=OUTBOUND_FIELDS["vless"])
+_VlessFields = make_dataclass("_VlessFields", fields=OUTBOUND_FIELDS["vless"])
 
 
 @OutboundFactory.register
 class VlessOutbound(_VlessFields, Outbound):
     PROTOCOL = "vless"
     LINK_SCHEME = "vless"
+
+    @classmethod
+    def from_link(
+        cls, target_storage: TargetStorage, link: str
+    ) -> VlessOutbound:
+        parsed = urlparse(link)
+        query = parse_qs(parsed.query)
+
+        def require(value: T | None, name: str) -> T:
+            if value is None:
+                raise ValueError(f"Missing required field: {name}")
+            return value
+
+        def require_str(value: str | None, name: str) -> str:
+            value = require(value, name)
+            if value == "":
+                raise ValueError(f"Empty value for required field: {name}")
+            return unquote(value)
+
+        def get_required(key: str) -> str:
+            values = query.get(key)
+            if not values:
+                raise ValueError(f"Missing required query param: {key}")
+            raw = values[0]
+            if raw is None or raw == "":
+                raise ValueError(f"Empty value for required query param: {key}")
+            return unquote(raw)
+
+        return cls(
+            host=require_str(parsed.hostname, "host"),
+            port=require(parsed.port, "port"),
+            target=target_storage.load_by_pretty(
+                require_str(parsed.fragment, "target_pretty_name")
+            ),
+            id=require_str(parsed.username, "id"),
+            security=get_required("security"),
+            encryption=get_required("encryption"),
+            public_key=get_required("pbk"),
+            network_type=get_required("type"),
+            sni=get_required("sni"),
+            short_id=get_required("sid"),
+            flow=get_required("flow"),
+            header_type=get_required("headerType"),
+            finger_print=get_required("fp"),
+        )
+
+    @classmethod
+    def from_inbound(
+        cls,
+        target_storage: TargetStorage,
+        host: str,
+        client: dict,
+        inbound: dict,
+    ) -> tuple[str, VlessOutbound]:
+        settings = inbound["settings"]
+        stream = inbound["streamSettings"]
+        reality = stream["realitySettings"]
+
+        port = inbound["port"]
+        id = client["id"]
+        flow = client["flow"]
+        encryption = settings["decryption"]
+        network_type = stream["network"]
+        security = stream["security"]
+        sni = reality["serverNames"][0]
+        public_key = private_to_public(reality["privateKey"])
+
+        clients = settings["clients"]
+        short_ids = reality["shortIds"]
+
+        try:
+            index = clients.index(client)
+        except ValueError:
+            raise RuntimeError(f"Client not found in inbound: {client}")
+
+        try:
+            short_id = short_ids[index]
+        except IndexError:
+            raise RuntimeError(f"No shortId for client at index {index}")
+
+        email = client["email"]
+        username, target_id = cls.split_client_email(email)
+
+        connection = cls(
+            host=host,
+            port=port,
+            target=target_storage.load_by_id(target_id),
+            id=id,
+            security=security,
+            encryption=encryption,
+            public_key=public_key,
+            network_type=network_type,
+            sni=sni,
+            short_id=short_id,
+            flow=flow,
+        )
+        return username, connection
+
+    @classmethod
+    def from_scratch(
+        cls, target: Target, host: str, inbound: dict
+    ) -> VlessOutbound:
+        settings = inbound["settings"]
+        stream = inbound["streamSettings"]
+        reality = stream["realitySettings"]
+
+        port = inbound["port"]
+        encryption = settings["decryption"]
+        network_type = stream["network"]
+        security = stream["security"]
+        sni = reality["serverNames"][0]
+        public_key = private_to_public(reality["privateKey"])
+
+        id = cls.generate_id()
+        short_id = cls.generate_short_id()
+
+        return cls(
+            host=host,
+            port=port,
+            target=target,
+            id=id,
+            security=security,
+            encryption=encryption,
+            public_key=public_key,
+            network_type=network_type,
+            sni=sni,
+            short_id=short_id,
+        )
+
+    def to_link(self) -> str:
+        netloc = f"{self.id}@{self.host}:{self.port}"
+        params = {
+            "security": self.security,
+            "encryption": self.encryption,
+            "pbk": self.public_key,
+            "headerType": self.header_type,
+            "fp": self.finger_print,
+            "type": self.network_type,
+            "flow": self.flow,
+            "sni": self.sni,
+            "sid": self.short_id,
+        }
+        query = urlencode(params, quote_via=quote)
+        fragment = quote(self.target.pretty)
+        return f"vless://{netloc}?{query}#{fragment}"
+
+    @staticmethod
+    def generate_id() -> str:
+        return str(uuid.uuid4())
+
+    @staticmethod
+    def generate_short_id() -> str:
+        return secrets.token_hex(8)
+
+    def matches_inbound(self, inbound: dict) -> bool:
+        if inbound.get("protocol") != self.PROTOCOL:
+            return False
+
+        if inbound.get("port") != self.port:
+            return False
+
+        settings = inbound.get("settings", {})
+        if settings.get("decryption") != self.encryption:
+            return False
+
+        stream_settings = inbound.get("streamSettings", {})
+        if stream_settings.get("network") != self.network_type:
+            return False
+        if stream_settings.get("security") != self.security:
+            return False
+
+        reality_settings = stream_settings.get("realitySettings", {})
+        if self.sni not in reality_settings.get("serverNames", []):
+            return False
+
+        inbound_private_key = reality_settings.get("privateKey")
+        if not inbound_private_key:
+            raise ValueError("Reality Settings do not have private key")
+        inbound_public_key = private_to_public(inbound_private_key)
+        if inbound_public_key != self.public_key:
+            return False
+
+        return True
+
+    def add_to_inbound(self, inbound: dict, username: str) -> None:
+        client = {
+            "email": f"{username}-{self.target.id}",
+            "id": self.id,
+            "flow": self.flow,
+        }
+
+        clients = inbound["settings"].setdefault("clients", [])
+        reality = inbound["streamSettings"].setdefault("realitySettings", {})
+        short_ids = reality.setdefault("shortIds", [])
+
+        clients.append(client)
+        short_ids.append(self.short_id)
+
+        clients_and_sids_by_user = defaultdict(list)
+        for client, sid in zip(clients, short_ids):
+            username, _ = self.split_client_email(client["email"])
+            clients_and_sids_by_user[username].append((client, sid))
+
+        priority = {"default": 0, "managed": 1, "relay": 2}
+
+        def sort_key(pair):
+            client, _ = pair
+            _, target_id = self.split_client_email(client["email"])
+            return (priority.get(target_id, 100), client["email"])
+
+        for pair in clients_and_sids_by_user.values():
+            pair.sort(key=sort_key)
+
+        sorted_clients = []
+        sorted_short_ids = []
+        for username in sorted(clients_and_sids_by_user):
+            for client, sid in clients_and_sids_by_user[username]:
+                sorted_clients.append(client)
+                sorted_short_ids.append(sid)
+
+        inbound["settings"]["clients"] = sorted_clients
+        reality["shortIds"] = sorted_short_ids
+
+    def delete_from_inbound(self, inbound: dict, username: str) -> None:
+        target_email = f"{username}-{self.target.id}"
+
+        clients = inbound["settings"].get("clients", [])
+        reality = inbound["streamSettings"].get("realitySettings", {})
+        short_ids = reality.get("shortIds", [])
+
+        if not clients or not short_ids:
+            return
+
+        filtered_pairs = [
+            (client, sid)
+            for client, sid in zip(clients, short_ids)
+            if client["email"] != target_email
+        ]
+
+        inbound["settings"]["clients"] = [
+            client for client, _ in filtered_pairs
+        ]
+
+        inbound["streamSettings"]["realitySettings"]["shortIds"] = [
+            sid for _, sid in filtered_pairs
+        ]
+
+    @property
+    def protocol(self) -> str:
+        return self.PROTOCOL
+
+    def __hash__(self) -> int:
+        return hash(
+            (
+                self.host,
+                self.port,
+                self.target,
+                self.id,
+                self.security,
+                self.encryption,
+                self.public_key,
+                self.network_type,
+                self.sni,
+                self.short_id,
+                self.flow,
+                self.header_type,
+                self.finger_print,
+            )
+        )
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, VlessOutbound):
+            return NotImplemented
+
+        return (
+            self.host == other.host
+            and self.port == other.port
+            and self.target == other.target
+            and self.id == other.id
+            and self.security == other.security
+            and self.encryption == other.encryption
+            and self.public_key == other.public_key
+            and self.network_type == other.network_type
+            and self.sni == other.sni
+            and self.short_id == other.short_id
+            and self.flow == other.flow
+            and self.header_type == other.header_type
+            and self.finger_print == other.finger_print
+        )

src/xray_manager/core/outbound_fields.py

@@ -1,11 +1,8 @@
 from dataclasses import field
 
-_BASE_FIELDS = [
-    ("host", str),
-    ("port", str),
-    ("target", str),
-    ("target_pretty_name", str),
-]
+from .target import Target
+
+_BASE_FIELDS = [("host", str), ("port", str), ("target", Target)]
 
 
 OUTBOUND_FIELDS = {

src/xray_manager/core/target.py

@@ -6,6 +6,7 @@ from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from .xray_manager_config import XrayManagerConfig
 
+
 @dataclass(frozen=True)
 class Target:
     id: str
@@ -24,25 +25,25 @@ class TargetStorage:
     def load_by_id(self, target_id: str) -> Target:
         if target_id not in self.target_dict:
             raise ValueError(f"Target not found: {target_id}")
-        
+
         target_data = self.target_dict[target_id]
 
         if "route" not in target_data:
             raise ValueError(
                 f"Missing required 'route' field for target: {target_id}"
             )
-        
+
         return Target(
             id=target_id,
             pretty=target_data.get("pretty_name", target_id),
-            route=target_data["route"]
+            route=target_data["route"],
         )
 
     def load_by_pretty(self, target_pretty_name: str) -> Target:
         for target_id, target_data in self.target_dict.items():
             if target_data.get("pretty_name", target_id) == target_pretty_name:
                 return self.load_by_id(target_id)
-        
+
         raise ValueError(f"Target pretty name not found: {target_pretty_name}")
 
     def load_all_targets(self) -> list[Target]:
@@ -57,5 +58,5 @@ class TargetStorage:
     def delete_target(self, target: Target):
         if target not in self.load_all_targets():
             raise ValueError(f"Target not found: {target.id}")
-        
-        self.xrmc.delete_target(target.id)
+
+        self.xrmc.delete_target(target.id)

src/xray_manager/core/xray_manager_config.py

@@ -73,13 +73,6 @@ class XrayManagerConfig:
             for username, folder_name in self.profile_folder_mapping.items()
         }
 
-    @property
-    def targets(self) -> dict[str, dict[str, str]]:
-        return self.config.get("target", {})
-
-    def get_targets_list(self) -> list[str]:
-        return list(self.targets.keys())
-
     def set_profile_folder(self, username: str, folder_name: str):
         self.config["profiles"]["folder_mapping"][username] = folder_name
         self._save_json()
@@ -89,11 +82,11 @@ class XrayManagerConfig:
         self._save_json()
 
     def set_target(
-            self, target_id: str, target_pretty_name: str, target_route: str
+        self, target_id: str, target_pretty_name: str, target_route: str
     ):
         self.config["target"][target_id] = {
             "pretty_name": target_pretty_name,
-            "route": target_route
+            "route": target_route,
         }
         self.config["target"][target_id]["pretty_name"] = target_pretty_name
         self.config["target"][target_id]["route"] = target_route
@@ -101,4 +94,4 @@ class XrayManagerConfig:
 
     def delete_target(self, target_id: str):
         self.config["target"].pop(target_id, None)
-        self._save_json()
+        self._save_json()

tests/mock_data/etc/xray-manager/config.json

@@ -27,5 +27,16 @@
             "route": "out-us2",
             "pretty_name": "🇺🇸 USA #2"
         }
+    },
+    "override": {
+        "test": {
+            "vless": {
+                "port": 443,
+                "finger_print": "qq",
+                "by_target_id": {
+                    "us2": "chrome"
+                }
+            }
+        }
     }
 }

tests/sandbox/shadowsocks_outbound.py

@@ -3,7 +3,7 @@ from xray_manager.core.outbound import (
     OutboundSpec,
     ShadowsocksOutbound,
 )
-from xray_manager.core.xray_config import XrayConfig
+from xray_manager.core.target import Target, TargetStorage
 from xray_manager.core.xray_manager_config import XrayManagerConfig
 
 inbound = {
@@ -32,16 +32,13 @@ inbound = {
 if __name__ == "__main__":
     DEV_CONFIG_PATH = "tests/mock_data/etc/xray-manager/config.json"
     xray_manager_config = XrayManagerConfig(DEV_CONFIG_PATH)
+    target_storage = TargetStorage(xray_manager_config)
 
-    target = "us1"
-    target_pretty_name = xray_manager_config.targets.get(target, {}).get(
-        "pretty_name", target
-    )
+    target = target_storage.load_by_id("default")
 
-    outbound = ShadowsocksOutbound.from_scratch(
-        target, target_pretty_name, "127.0.0.1", inbound
-    )
+    outbound = ShadowsocksOutbound.from_scratch(target, "127.0.0.1", inbound)
 
     link = outbound.to_link()
-    o = ShadowsocksOutbound.from_link(xray_manager_config.targets, link)
+    print(link)
+    o = ShadowsocksOutbound.from_link(target_storage, link)
     print(o)

tests/sandbox/target.py

@@ -9,5 +9,5 @@ if __name__ == "__main__":
     storage = TargetStorage(xray_manager_config)
 
     target = Target("de1", "pretty", "out-de1")
-    
-    storage.delete_target(target)
+
+    storage.delete_target(target)

tests/sandbox/vless_outbound.py

@@ -0,0 +1,95 @@
+from xray_manager.core import (
+    OutboundFactory,
+    ProfileStorage,
+    Target,
+    TargetStorage,
+    VlessOutbound,
+    XrayManagerConfig,
+)
+
+inbound = {
+    "tag": "in-vless",
+    "listen": "127.0.0.1",
+    "port": 443,
+    "protocol": "vless",
+    "settings": {
+        "clients": [
+            {
+                "email": "test-fr1",
+                "id": "0eb7a1a8-c24f-4eb2-9f89-8023a10ed070",
+                "flow": "xtls-rprx-vision",
+            },
+            {
+                "email": "test2-us2",
+                "id": "7b6da4ce-bb10-4127-a0c6-243618c62edd",
+                "flow": "xtls-rprx-vision",
+            },
+        ],
+        "decryption": "none",
+    },
+    "streamSettings": {
+        "network": "tcp",
+        "security": "reality",
+        "realitySettings": {
+            "show": False,
+            "dest": "google.com:443",
+            "xver": 0,
+            "serverNames": ["google.com"],
+            "privateKey": "yHJQq57MNRi-0UQm7ymiy2DI17iO_Ovu2HF5EV1ViGg",
+            "minClientVer": "",
+            "maxClientVer": "",
+            "maxTimeDiff": 0,
+            "shortIds": ["91e7c34ea2c07723", "0419a64d06467571"],
+        },
+    },
+}
+
+
+def test_from_scratch(target_storage: TargetStorage) -> VlessOutbound:
+    target = target_storage.load_by_id("fr1")
+    outbound = VlessOutbound.from_scratch(target, "localhost", inbound)
+    print("Test from scratch")
+    print(outbound)
+    return outbound
+
+
+def test_to_link(outbound: VlessOutbound) -> str:
+    link = outbound.to_link()
+    print("Test to link")
+    print(link)
+    return link
+
+
+def test_from_link(target_storage: TargetStorage, link: str):
+    outbound = VlessOutbound.from_link(target_storage, link)
+    print("Test from link")
+    print(outbound)
+
+
+def test_from_inbound(target_storage: TargetStorage):
+    client = {
+        "email": "test-fr1",
+        "id": "0eb7a1a8-c24f-4eb2-9f89-8023a10ed070",
+        "flow": "xtls-rprx-vision",
+    }
+
+    username, outbound = VlessOutbound.from_inbound(
+        target_storage, "localhost", client, inbound
+    )
+
+    print("Test from inbound")
+    print(username)
+    print(outbound)
+
+
+if __name__ == "__main__":
+    DEV_CONFIG_PATH = "tests/mock_data/etc/xray-manager/config.json"
+    xray_manager_config = XrayManagerConfig(DEV_CONFIG_PATH)
+    profile_storage = ProfileStorage(xray_manager_config)
+    target_storage = TargetStorage(xray_manager_config)
+
+    outbound = test_from_scratch(target_storage)
+    link = test_to_link(outbound)
+    test_from_link(target_storage, link)
+
+    test_from_inbound(target_storage)