Issue precise subscription tokens without revocation bypass

app/app_factory.py

@@ -12,6 +12,7 @@
 from app.nats import is_multi_worker, require_nats_if_multiworker
 from app.nats.message import MessageTopic
 from app.nats.router import router
+from app.node.errors import NodeRevocationError
 from app.settings import handle_settings_message
 from app.subscription.client_templates import handle_client_template_message
 from app.utils.logger import get_logger
@@ -35,6 +36,15 @@ async def database_operational_error_handler(request: Request, exc: DBAPIError):
     )
 
 
+async def node_revocation_error_handler(request: Request, exc: NodeRevocationError):
+    logger.warning("Node revocation unavailable while handling %s %s: %s", request.method, request.url.path, exc)
+    return JSONResponse(
+        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+        content={"detail": "User removal was not confirmed by all runtime nodes. Retry when nodes are available."},
+        headers={"Retry-After": "1"},
+    )
+
+
 def _use_route_names_as_operation_ids(app: FastAPI) -> None:
     def _simplify_operation_ids(routes):
         for route in routes:
@@ -275,6 +285,7 @@ def validation_exception_handler(request: Request, exc: RequestValidationError):
         )
 
     app.add_exception_handler(DBAPIError, database_operational_error_handler)
+    app.add_exception_handler(NodeRevocationError, node_revocation_error_handler)
 
     from app.operation.permissions import LimitExceeded, PermissionDenied
 

app/db/crud/user.py

@@ -953,7 +953,7 @@ async def _delete_user_dependencies(db: AsyncSession, user_ids: list[int]):
     await db.execute(users_groups_association.delete().where(users_groups_association.c.user_id.in_(user_ids)))
 
 
-async def remove_user(db: AsyncSession, db_user: User) -> User:
+async def remove_user(db: AsyncSession, db_user: User, *, commit: bool = True) -> User:
     """
     Removes a user from the database.
 
@@ -967,11 +967,14 @@ async def remove_user(db: AsyncSession, db_user: User) -> User:
     await release_users_allocations(db, [db_user])
     await _delete_user_dependencies(db, [db_user.id])
     await db.execute(delete(User).where(User.id == db_user.id))
-    await db.commit()
+    if commit:
+        await db.commit()
+    else:
+        await db.flush()
     return db_user
 
 
-async def remove_users(db: AsyncSession, db_users: list[User]):
+async def remove_users(db: AsyncSession, db_users: list[User], *, commit: bool = True):
     """
     Removes multiple users from the database.
 
@@ -987,7 +990,10 @@ async def remove_users(db: AsyncSession, db_users: list[User]):
     await release_users_allocations(db, db_users)
     await _delete_user_dependencies(db, user_ids)
     await db.execute(delete(User).where(User.id.in_(user_ids)))
-    await db.commit()
+    if commit:
+        await db.commit()
+    else:
+        await db.flush()
 
 
 async def modify_user(
@@ -1424,20 +1430,10 @@ async def get_users_subscription_agent_stats(
     return rows
 
 
-async def autodelete_expired_users(
+async def get_autodelete_expired_users(
     db: AsyncSession, include_limited_users: bool = False
-) -> list[UserNotificationResponse]:
-    """
-    Deletes expired (optionally also limited) users whose auto-delete time has passed.
-
-    Args:
-        db (AsyncSession): Database session
-        include_limited_users (bool, optional): Whether to delete limited users as well.
-            Defaults to False.
-
-    Returns:
-        list[UserNotificationResponse]: List of deleted users.
-    """
+) -> tuple[list[User], list[UserNotificationResponse]]:
+    """Return auto-delete targets and their node-removal snapshots without deleting them."""
     target_status = [UserStatus.expired] if not include_limited_users else [UserStatus.expired, UserStatus.limited]
 
     auto_delete = func.coalesce(User.auto_delete_in_days, literal(user_cleanup_settings.autodelete_days))
@@ -1449,6 +1445,7 @@ async def autodelete_expired_users(
         )
         .where(
             auto_delete >= 0,  # Negative values prevent auto-deletion
+            auto_delete <= 36500,  # Keep persisted legacy values within datetime arithmetic bounds
             User.status.in_(target_status),
         )
         .options(joinedload(User.admin))
@@ -1461,12 +1458,29 @@ async def autodelete_expired_users(
     ]
 
     result: list[UserNotificationResponse] = []
-    if expired_users:
-        for user in expired_users:
-            await load_user_attrs(user)
-            result.append(UserNotificationResponse.model_validate(user))
+    for user in expired_users:
+        await load_user_attrs(user)
+        result.append(UserNotificationResponse.model_validate(user))
+    return expired_users, result
+
+
+async def autodelete_expired_users(
+    db: AsyncSession, include_limited_users: bool = False, *, commit: bool = True
+) -> list[UserNotificationResponse]:
+    """
+    Delete expired (optionally also limited) users whose auto-delete time has passed.
 
-        await remove_users(db, expired_users)
+    Args:
+        db (AsyncSession): Database session
+        include_limited_users (bool, optional): Whether to delete limited users as well.
+            Defaults to False.
+
+    Returns:
+        list[UserNotificationResponse]: List of deleted users.
+    """
+    expired_users, result = await get_autodelete_expired_users(db, include_limited_users)
+    if expired_users:
+        await remove_users(db, expired_users, commit=commit)
 
     return result
 

app/db/migrations/versions/9e0d7a1c4b52_preserve_subscription_revocation_microseconds.py

@@ -0,0 +1,56 @@
+"""preserve subscription token timestamp microseconds
+
+Revision ID: 9e0d7a1c4b52
+Revises: d12f6a8b9c30
+Create Date: 2026-08-09
+
+"""
+
+from alembic import op
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision = "9e0d7a1c4b52"
+down_revision = "d12f6a8b9c30"
+branch_labels = None
+depends_on = None
+
+
+def _is_mysql_family() -> bool:
+    return op.get_bind().dialect.name in {"mysql", "mariadb"}
+
+
+def upgrade() -> None:
+    if _is_mysql_family():
+        op.alter_column(
+            "users",
+            "created_at",
+            existing_type=mysql.DATETIME(fsp=0),
+            type_=mysql.DATETIME(fsp=6),
+            existing_nullable=False,
+        )
+        op.alter_column(
+            "users",
+            "sub_revoked_at",
+            existing_type=mysql.DATETIME(fsp=0),
+            type_=mysql.DATETIME(fsp=6),
+            existing_nullable=True,
+        )
+
+
+def downgrade() -> None:
+    if _is_mysql_family():
+        op.alter_column(
+            "users",
+            "sub_revoked_at",
+            existing_type=mysql.DATETIME(fsp=6),
+            type_=mysql.DATETIME(fsp=0),
+            existing_nullable=True,
+        )
+        op.alter_column(
+            "users",
+            "created_at",
+            existing_type=mysql.DATETIME(fsp=6),
+            type_=mysql.DATETIME(fsp=0),
+            existing_nullable=False,
+        )

app/db/migrations/versions/a8c2d491e705_bind_xray_client_inbounds_to_loopback.py

@@ -0,0 +1,89 @@
+"""Bind generated Xray client proxy inbounds to loopback.
+
+Revision ID: a8c2d491e705
+Revises: fb32155473c1
+Create Date: 2026-08-09
+"""
+
+import re
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "a8c2d491e705"
+down_revision = "fb32155473c1"
+branch_labels = None
+depends_on = None
+
+
+client_templates = sa.table(
+    "client_templates",
+    sa.column("id", sa.Integer()),
+    sa.column("template_type", sa.String()),
+    sa.column("content", sa.Text()),
+    sa.column("is_system", sa.Boolean()),
+)
+
+EXPOSED_LISTENER = re.compile(r'("listen"\s*:\s*)"0\.0\.0\.0"')
+INBOUNDS_ARRAY = re.compile(r'"inbounds"\s*:\s*\[')
+
+
+def _json_array_end(content: str, start: int) -> int | None:
+    depth = 0
+    in_string = False
+    escaped = False
+    for index in range(start, len(content)):
+        character = content[index]
+        if in_string:
+            if escaped:
+                escaped = False
+            elif character == "\\":
+                escaped = True
+            elif character == '"':
+                in_string = False
+            continue
+        if character == '"':
+            in_string = True
+        elif character == "[":
+            depth += 1
+        elif character == "]":
+            depth -= 1
+            if depth == 0:
+                return index + 1
+    return None
+
+
+def _bind_client_listeners_to_loopback(content: str) -> str:
+    """Rewrite only exposed Xray client listener values, preserving the template."""
+    match = INBOUNDS_ARRAY.search(content)
+    if not match:
+        return content
+    start = match.end() - 1
+    end = _json_array_end(content, start)
+    if end is None:
+        return content
+    inbounds = EXPOSED_LISTENER.sub(r'\1"127.0.0.1"', content[start:end])
+    return content[:start] + inbounds + content[end:]
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    rows = connection.execute(
+        sa.select(client_templates.c.id, client_templates.c.content).where(
+            client_templates.c.template_type == "xray_subscription",
+            client_templates.c.is_system.is_(True),
+        )
+    ).mappings()
+    for row in rows:
+        content = row["content"]
+        updated_content = _bind_client_listeners_to_loopback(content)
+        if updated_content == content:
+            continue
+        connection.execute(
+            client_templates.update().where(client_templates.c.id == row["id"]).values(content=updated_content)
+        )
+
+
+def downgrade() -> None:
+    # Do not reintroduce an unauthenticated network listener on downgrade.
+    pass

app/db/migrations/versions/d12f6a8b9c30_add_bridge_sync_namespaces.py

@@ -0,0 +1,48 @@
+"""Add stable Bridge node and user sync namespaces.
+
+Revision ID: d12f6a8b9c30
+Revises: a8c2d491e705
+Create Date: 2026-08-10
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "d12f6a8b9c30"
+down_revision = "a8c2d491e705"
+branch_labels = None
+depends_on = None
+
+
+def _backfill_legacy_namespace(connection, table: str, column: str) -> None:
+    # Preserve the namespace already used by running Bridge/core processes and
+    # NATS KV. Newly created ORM rows use UUID defaults, so a later reused
+    # numeric database id cannot inherit this incarnation's state or stats.
+    table_ref = sa.table(table, sa.column("id"), sa.column(column))
+    connection.execute(
+        sa.update(table_ref).values({column: sa.cast(table_ref.c.id, sa.String(36))})
+    )
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    op.add_column("nodes", sa.Column("bridge_id", sa.String(length=36), nullable=True))
+    op.add_column("users", sa.Column("sync_id", sa.String(length=36), nullable=True))
+    _backfill_legacy_namespace(connection, "nodes", "bridge_id")
+    _backfill_legacy_namespace(connection, "users", "sync_id")
+
+    with op.batch_alter_table("nodes") as batch_op:
+        batch_op.alter_column("bridge_id", existing_type=sa.String(length=36), nullable=False)
+        batch_op.create_unique_constraint("uq_nodes_bridge_id", ["bridge_id"])
+    with op.batch_alter_table("users") as batch_op:
+        batch_op.alter_column("sync_id", existing_type=sa.String(length=36), nullable=False)
+        batch_op.create_unique_constraint("uq_users_sync_id", ["sync_id"])
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("users") as batch_op:
+        batch_op.drop_constraint("uq_users_sync_id", type_="unique")
+        batch_op.drop_column("sync_id")
+    with op.batch_alter_table("nodes") as batch_op:
+        batch_op.drop_constraint("uq_nodes_bridge_id", type_="unique")
+        batch_op.drop_column("bridge_id")