Harden bridge transport and compatibility

PasarGuardNodeBridge/__init__.py

@@ -12,10 +12,10 @@
 - Extensible with custom metadata via the `extra` argument
 
 Author: PasarGuard
-Version: 0.9.0
+Version: 0.10.0
 """
 
-__version__ = "0.9.0"
+__version__ = "0.10.0"
 __author__ = "PasarGuard"
 
 
@@ -31,12 +31,20 @@
     InMemoryNodeRegistry,
     InMemoryUserSyncStore,
     LifecycleLease,
+    LifecycleLeaseLostError,
     LifecycleOperation,
     LifecycleStatus,
     NodeConfig,
     NodeLifecycleCoordinatorProtocol,
     NodeLifecycleState,
     NodeRegistryProtocol,
+    RevocationAwareUserSyncStoreProtocol,
+    StartupUserSyncLease,
+    UserRevocationConflictError,
+    UserRevocationResult,
+    UserSyncLease,
+    UserSyncLeaseLostError,
+    UserSyncStoreFullError,
     UserSyncStoreProtocol,
 )
 from PasarGuardNodeBridge.utils import create_proxy, create_user
@@ -53,6 +61,8 @@ def create_node(
     port: int,
     server_ca: str,
     api_key: str,
+    api_port: int | None = None,
+    max_message_size: int | None = None,
     **kwargs,
 ) -> PasarGuardNode:
     """
@@ -67,6 +77,10 @@ def create_node(
         port (int): Port number used to connect to the node.
         server_ca (str): The server's SSL certificate as a string (PEM format).
         api_key (str): API key used for authentication with the node.
+        api_port (int | None): Port for the maintenance JSON API. Defaults to
+            ``port`` for backwards compatibility with shared-port deployments.
+        max_message_size (int | None): Maximum gRPC message size. Ignored for
+            REST nodes.
         **kwargs: Additional optional arguments:
             - name (str): Node instance name for logging. Defaults to "default".
             - extra (dict): Optional dictionary to pass custom metadata or configuration. Defaults to {}.
@@ -112,19 +126,24 @@ def create_node(
           HTTP CONNECT and SOCKS proxy schemes.
     """
 
+    resolved_api_port = port if api_port is None else api_port
+
     if connection is NodeType.grpc:
         return GrpcNode(
             address=address,
             port=port,
+            api_port=resolved_api_port,
             server_ca=server_ca,
             api_key=api_key,
+            max_message_size=max_message_size,
             **kwargs,
         )
 
     elif connection is NodeType.rest:
         return RestNode(
             address=address,
             port=port,
+            api_port=resolved_api_port,
             server_ca=server_ca,
             api_key=api_key,
             **kwargs,
@@ -167,14 +186,22 @@ async def create_node_from_registry(
     "create_node_from_config",
     "InMemoryUserSyncStore",
     "InMemoryNodeRegistry",
+    "UserSyncStoreFullError",
     "UserSyncStoreProtocol",
     "NodeRegistryProtocol",
     "NodeConfig",
     "ClaimedUser",
+    "UserSyncLease",
+    "UserSyncLeaseLostError",
+    "UserRevocationConflictError",
+    "UserRevocationResult",
+    "RevocationAwareUserSyncStoreProtocol",
+    "StartupUserSyncLease",
     "NodeLifecycleState",
     "NodeLifecycleCoordinatorProtocol",
     "LifecycleStatus",
     "LifecycleOperation",
     "LifecycleLease",
+    "LifecycleLeaseLostError",
     "InMemoryNodeLifecycleCoordinator",
 ]

PasarGuardNodeBridge/abstract_node.py

@@ -17,6 +17,7 @@ async def start(
         keep_alive: int = 0,
         exclude_inbounds: list[str] = [],
         timeout: int | None = None,
+        reconcile_user_sync: bool = False,
     ) -> service.BaseInfoResponse | None:
         raise NotImplementedError
 
@@ -58,13 +59,30 @@ async def get_user_online_ip_list(
 
     @abstractmethod
     async def sync_users(
-        self, users: list[service.User], flush_pending: bool = False, timeout: int | None = None
+        self,
+        users: list[service.User],
+        flush_pending: bool = False,
+        timeout: int | None = None,
+        revocation_id: str | None = None,
     ) -> service.Empty | None:
         raise NotImplementedError
 
+    async def reconcile_users(
+        self,
+        users: list[service.User],
+        flush_pending: bool = False,
+        timeout: int | None = None,
+    ) -> service.Empty | None:
+        raise NodeAPIError(501, "This node transport does not support authoritative user reconciliation")
+
     @abstractmethod
     async def sync_users_chunked(
-        self, users: list[service.User], chunk_size: int = 100, flush_pending: bool = False, timeout: int | None = None
+        self,
+        users: list[service.User],
+        chunk_size: int = 100,
+        flush_pending: bool = False,
+        timeout: int | None = None,
+        revocation_id: str | None = None,
     ) -> list[service.User]:
         raise NotImplementedError
 
@@ -114,7 +132,7 @@ async def _check_node_health(self):
         raise NotImplementedError
 
     @abstractmethod
-    async def _sync_batch_users(self, users: list[service.User]) -> list[service.User]:
+    async def _sync_batch_users(self, users: list[service.User], user_sync_epoch: int = 0) -> list[service.User]:
         """Sync a batch of users individually. Returns list of failed users to requeue."""
         raise NotImplementedError
 

PasarGuardNodeBridge/aiohttp_compat.py

@@ -29,7 +29,7 @@ def json(self) -> Any:
         return json.loads(self.text)
 
     def raise_for_status(self) -> None:
-        if 400 <= self.status_code:
+        if 300 <= self.status_code:
             raise BufferedStatusError(self)
 
 
@@ -90,10 +90,15 @@ async def _get_session(self) -> aiohttp.ClientSession:
         return self._session
 
     def request(self, *args, **kwargs):
+        # aiohttp follows redirects by default and preserves custom headers such
+        # as x-api-key across origins. Node API calls must stay pinned to the
+        # configured origin.
+        kwargs["allow_redirects"] = False
         return _LazyRequestContext(self, args, kwargs)
 
     async def get(self, *args, **kwargs) -> aiohttp.ClientResponse:
         session = await self._get_session()
+        kwargs["allow_redirects"] = False
         return await session.get(*args, **kwargs)
 
     async def close(self) -> None:

PasarGuardNodeBridge/common/service.proto

@@ -11,6 +11,8 @@ message BaseInfoResponse {
   bool started = 1;
   string core_version = 2;
   string node_version = 3;
+  bool user_sync_epoch_supported = 4;
+  uint64 user_sync_epoch = 5;
 }
 
 enum BackendType {
@@ -24,6 +26,7 @@ message Backend {
   repeated User users = 3;
   uint64 keep_alive = 4;
   repeated string exclude_inbounds = 5;
+  uint64 user_sync_epoch = 6;
 }
 
 // log
@@ -150,16 +153,19 @@ message User {
   string email = 1;
   Proxy proxies = 2;
   repeated string inbounds = 3;
+  uint64 user_sync_epoch = 4;
 }
 
 message Users {
   repeated User users = 1;
+  uint64 user_sync_epoch = 2;
 }
 
 message UsersChunk {
   repeated User users = 1;
   uint64 index = 2;
   bool last = 3;
+  uint64 user_sync_epoch = 4;
 }
 
 // Routing (mirrors xray app/router/command, node-friendly shapes)