diff --git a/.env.example b/.env.example index 110eb97de..0397474d8 100644 --- a/.env.example +++ b/.env.example @@ -119,11 +119,6 @@ UVICORN_PORT = 8000 # JOB_CHECK_NODE_LIMITS_INTERVAL = 60 # JOB_CLEANUP_SUBSCRIPTION_UPDATES_INTERVAL = 600 -## WireGuard peer allocation and subscription output. -# WIREGUARD_ENABLED = True -# WIREGUARD_GLOBAL_POOL = "10.0.0.0/8" -# WIREGUARD_RESERVED = "10.0.0.0/31" - ## Developer and test flags. # DOCS = False # DEBUG = False diff --git a/Makefile b/Makefile index 3963e51b1..25ad8543a 100644 --- a/Makefile +++ b/Makefile @@ -68,12 +68,12 @@ check-bun: check-nodejs # Install frontend dependencies (Node.js packages) .PHONY: install-front install-front: check-bun - @cd dashboard && bun install + @cd dashboard && bun install # Run database migrations using Alembic .PHONY: run-migration run-migration: - @uv run alembic upgrade head + @uv run alembic upgrade head .PHONY: check-migrations check-migrations: diff --git a/app/core/manager.py b/app/core/manager.py index 40524c91a..33007d576 100644 --- a/app/core/manager.py +++ b/app/core/manager.py @@ -205,12 +205,23 @@ async def initialize(self, db): core_configs, _ = await get_core_configs(db, CoreListQuery()) cores: dict[int, AbstractCore] = {} for config in core_configs: - core_config = self.validate_core( - config.config, - config.exclude_inbound_tags, - config.fallbacks_inbound_tags, - config.type, - ) + try: + core_config = self.validate_core( + config.config, + config.exclude_inbound_tags, + config.fallbacks_inbound_tags, + config.type, + ) + except Exception as exc: + # Broken DB cores must not take down the process (restart loop). + self._logger.error( + "Skipping broken core id=%s name=%r type=%s: %s", + config.id, + getattr(config, "name", None), + config.type, + exc, + ) + continue cores[config.id] = core_config async with self._lock: @@ -233,12 +244,21 @@ async def update_inbounds(self): async def _update_core_local(self, db_core_config: CoreConfig, core_config: AbstractCore | None = None): if core_config is None: - core_config = self.validate_core( - db_core_config.config, - db_core_config.exclude_inbound_tags, - db_core_config.fallbacks_inbound_tags, - db_core_config.type, - ) + try: + core_config = self.validate_core( + db_core_config.config, + db_core_config.exclude_inbound_tags, + db_core_config.fallbacks_inbound_tags, + db_core_config.type, + ) + except Exception as exc: + self._logger.error( + "Skipping broken core id=%s type=%s: %s", + getattr(db_core_config, "id", None), + getattr(db_core_config, "type", None), + exc, + ) + return async with self._lock: self._cores.update({db_core_config.id: core_config}) diff --git a/app/db/crud/bulk.py b/app/db/crud/bulk.py index e3ea2a55e..a5be01663 100644 --- a/app/db/crud/bulk.py +++ b/app/db/crud/bulk.py @@ -4,7 +4,6 @@ from sqlalchemy import and_, case, cast, delete, func, or_, select, text, update from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.orm import selectinload from app.db.models import ( Admin, @@ -268,21 +267,6 @@ async def count_bulk_group_scope(db: AsyncSession, bulk_model: BulkGroup) -> int return (await db.execute(select(func.count(User.id)).where(final_filter))).scalar_one_or_none() or 0 -async def get_bulk_wireguard_peer_ip_users( - db: AsyncSession, - bulk_model: BulkUserFilter, - admin_id: int | None = None, -) -> list[User]: - final_filter = _create_final_filter(bulk_model) - if admin_id is not None: - final_filter = and_(final_filter, User.admin_id == admin_id) - - result = await db.execute( - select(User).options(selectinload(User.groups).selectinload(Group.inbounds)).where(final_filter) - ) - return list(result.unique().scalars().all()) - - def _create_final_filter(bulk_model: BulkUserFilter): """Create a comprehensive SQLAlchemy filter condition from a bulk model.""" other_conditions = [] diff --git a/app/db/crud/user.py b/app/db/crud/user.py index 6c565f463..753f4f7af 100644 --- a/app/db/crud/user.py +++ b/app/db/crud/user.py @@ -48,6 +48,13 @@ from app.models.validators import MAX_ON_HOLD_EXPIRE_DURATION_SECONDS from config import user_cleanup_settings +from .wireguard import ( + release_allocations_by_user_ids, + release_users_allocations, + sync_user_allocations, + sync_users_allocations, + tags_from_groups, +) from .general import ( _build_trunc_expression, attach_timezone_to_period_start, @@ -263,33 +270,6 @@ async def get_users_with_proxy_settings( return list(result.scalars().all()) -async def get_all_wireguard_peer_ips_raw( - db: AsyncSession, - *, - exclude_user_id: int | None = None, -) -> dict[int, dict]: - """ - Retrieve only id and proxy_settings for all users (lightweight variant). - - Returns a dict mapping user_id -> {'proxy_settings': ...} for IP pool operations. - This avoids loading full ORM objects, related collections, and unnecessary columns. - - Args: - db: Database session - exclude_user_id: User ID to exclude from results - - Returns: - Dict mapping user_id to dict containing proxy_settings - """ - stmt = select(User.id, User.proxy_settings) - if exclude_user_id is not None: - stmt = stmt.where(User.id != exclude_user_id) - - result = await db.execute(stmt) - rows = result.all() - return {row[0]: {"proxy_settings": row[1]} for row in rows} - - def _build_user_sort_clause(sort_option: UserSortOption): field_map = { UserSortField.username: User.username, @@ -557,6 +537,7 @@ async def remove_expired_users( for start in range(0, len(user_ids), 1000): chunk = user_ids[start : start + 1000] + await release_allocations_by_user_ids(db, chunk) await _delete_user_dependencies(db, chunk) await db.execute(delete(User).where(User.id.in_(chunk))) await db.commit() @@ -896,6 +877,7 @@ async def create_user( db.add(db_user) await db.flush() + await sync_user_allocations(db, db_user, accessible_tags=await tags_from_groups(groups)) if new_user.next_plan: db_user.next_plan = NextPlan(user_id=db_user.id, **new_user.next_plan.model_dump()) @@ -930,6 +912,8 @@ async def create_users_bulk( db.add_all(db_users) await db.flush() + group_tags = await tags_from_groups(groups) + await sync_users_allocations(db, db_users, tags_by_user={user.id: group_tags for user in db_users}) next_plans: list[NextPlan] = [] for db_user, new_user in zip(db_users, new_users): @@ -967,6 +951,7 @@ async def remove_user(db: AsyncSession, db_user: User) -> User: Returns: User: Removed user object. """ + 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() @@ -986,6 +971,7 @@ async def remove_users(db: AsyncSession, db_users: list[User]): user_ids = list({user.id for user in db_users}) + 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() @@ -1015,8 +1001,9 @@ async def modify_user( if modify.proxy_settings is not None: db_user.proxy_settings = modify.proxy_settings.dict() - if modify.group_ids: + if modify.group_ids is not None: db_user.groups = groups or await get_groups_by_ids(db, modify.group_ids, load_users=False, load_inbounds=True) + await sync_user_allocations(db, db_user, accessible_tags=await tags_from_groups(db_user.groups)) if modify.status is not None: if modify.status == UserStatus.active and db_user.status == UserStatus.disabled: @@ -1210,6 +1197,7 @@ async def reset_user_by_next(db: AsyncSession, db_user: User, *, clean_chart_dat await db_user.next_plan.awaitable_attrs.user_template await db_user.next_plan.user_template.awaitable_attrs.groups db_user.groups = db_user.next_plan.user_template.groups + await sync_user_allocations(db, db_user, accessible_tags=await tags_from_groups(db_user.groups)) db_user.data_limit = db_user.next_plan.user_template.data_limit + ( 0 if not db_user.next_plan.add_remaining_traffic else remaining_traffic ) diff --git a/app/db/crud/wireguard.py b/app/db/crud/wireguard.py new file mode 100644 index 000000000..25deba88c --- /dev/null +++ b/app/db/crud/wireguard.py @@ -0,0 +1,582 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network, ip_interface, ip_network +from typing import Iterable + +from sqlalchemy import and_, delete, insert, select +from sqlalchemy.dialects.mysql import insert as mysql_insert +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import ( + CoreConfig, + CoreType, + Group, + ProxyInbound, + User, + WireGuardSubnet, + inbounds_groups_association, + users_groups_association, +) +from app.utils.crypto import generate_wireguard_keypair, get_wireguard_public_key +from app.utils.logger import get_logger + +logger = get_logger("wireguard-subnets") + +FREE_IPS_LIMIT = 10000 +# ponytail: free-list capped to bound JSON size; holes above the kept set stay +# reclaimable on reconcile (upgrade: sparse bitmap / run-length encoding). +FREE_OFFSETS_CAP = FREE_IPS_LIMIT + +IpNetwork = IPv4Network | IPv6Network +IpAddress = IPv4Address | IPv6Address + +# --- pure helpers ----------------------------------------------------------- + + +def core_config_dict(core: CoreConfig) -> dict: + cfg = core.config or {} + if isinstance(cfg, str): + cfg = json.loads(cfg) + return cfg + + +def wg_core_subnets(config: dict) -> list[IpNetwork]: + """Unique client subnets of a WG core (IPv4 and/or IPv6), from interface addresses.""" + seen: dict[str, IpNetwork] = {} + for cidr in (config or {}).get("address") or []: + try: + net = ip_interface(str(cidr).strip()).network + except ValueError: + continue + seen.setdefault(str(net), net) + return list(seen.values()) + + +def render_peer_ip(subnet: IpNetwork, offset: int) -> str: + host = type(subnet.network_address)(int(subnet.network_address) + offset) + return f"{host}/{'32' if subnet.version == 4 else '128'}" + + +def peer_host(entry: str) -> tuple[int, int] | None: + """(version, host_int) of a stored peer_ips entry; None if invalid.""" + try: + net = ip_network(str(entry).strip(), strict=False) + except ValueError: + return None + return net.version, int(net.network_address) + + +def _host_address(version: int, host_int: int) -> IpAddress: + return IPv4Address(host_int) if version == 4 else IPv6Address(host_int) + + +@dataclass(frozen=True) +class WgNamespace: + """One allocation namespace: an exact client subnet shared by every WG core that uses it.""" + + key: str # canonical str(subnet), e.g. "10.0.0.0/24" or "fd00::/64" + subnet: IpNetwork + tags: frozenset[str] + reserved: frozenset[int] # network, last, and server interface offsets + + +def wg_namespaces(cores: Iterable[CoreConfig]) -> dict[str, WgNamespace]: + by_key: dict[str, list[tuple[dict, IpNetwork]]] = {} + for core in cores: + cfg = core_config_dict(core) + for subnet in wg_core_subnets(cfg): + by_key.setdefault(str(subnet), []).append((cfg, subnet)) + + namespaces: dict[str, WgNamespace] = {} + for key, entries in by_key.items(): + subnet = entries[0][1] + base = int(subnet.network_address) + reserved = {0, subnet.num_addresses - 1} + tags: set[str] = set() + for cfg, _ in entries: + tag = str(cfg.get("interface_name") or "").strip() + if tag: + tags.add(tag) + for cidr in cfg.get("address") or []: + try: + iface = ip_interface(str(cidr).strip()) + except ValueError: + continue + if iface.version == subnet.version and iface.ip in subnet: + reserved.add(int(iface.ip) - base) + namespaces[key] = WgNamespace(key=key, subnet=subnet, tags=frozenset(tags), reserved=frozenset(reserved)) + return _collapse_overlapping_namespaces(namespaces) + + +def _collapse_overlapping_namespaces(namespaces: dict[str, WgNamespace]) -> dict[str, WgNamespace]: + """Keep the largest subnet when CIDRs overlap; merge tags/reserved from smaller ones into it.""" + if len(namespaces) < 2: + return namespaces + + # largest first (smaller prefixlen), then stable by key + ordered = sorted(namespaces.values(), key=lambda ns: (ns.subnet.prefixlen, ns.key)) + kept: dict[str, WgNamespace] = {} + for ns in ordered: + keeper = None + for other in kept.values(): + if other.subnet.version != ns.subnet.version: + continue + if other.subnet.overlaps(ns.subnet): + keeper = other + break + if keeper is None: + kept[ns.key] = ns + continue + + logger.warning( + "WireGuard subnet %s overlaps %s; skipping smaller/overlapping pool, merging into %s", + ns.key, + keeper.key, + keeper.key, + ) + base = int(keeper.subnet.network_address) + reserved = set(keeper.reserved) + for offset in ns.reserved: + host = _host_address(ns.subnet.version, int(ns.subnet.network_address) + offset) + if host in keeper.subnet: + reserved.add(int(host) - base) + kept[keeper.key] = WgNamespace( + key=keeper.key, + subnet=keeper.subnet, + tags=frozenset(keeper.tags | ns.tags), + reserved=frozenset(reserved), + ) + return kept + + +def wg_core_tags(cores: Iterable[CoreConfig]) -> set[str]: + """Interface names (= inbound tags) of all WG cores, subnet or not.""" + tags: set[str] = set() + for core in cores: + tag = str(core_config_dict(core).get("interface_name") or "").strip() + if tag: + tags.add(tag) + return tags + + +def offset_valid(ns: WgNamespace, offset: int) -> bool: + return 0 < offset < ns.subnet.num_addresses - 1 and offset not in ns.reserved + + +def match_namespace(namespaces: dict[str, WgNamespace], version: int, host_int: int) -> WgNamespace | None: + addr = _host_address(version, host_int) + for ns in namespaces.values(): + if ns.subnet.version == version and addr in ns.subnet: + return ns + return None + + +def pick_peer_ip_for_inbound(inbound_addresses: Iterable[str], peer_ips: Iterable[str]) -> list[str]: + """The user's peer IPs that belong to this inbound's subnet(s).""" + subnets = wg_core_subnets({"address": list(inbound_addresses or [])}) + if not subnets: + return [] + picked = [] + for entry in peer_ips or []: + host = peer_host(entry) + if host is None: + continue + addr = _host_address(*host) + if any(addr in subnet for subnet in subnets): + picked.append(str(entry)) + return picked + + +def _usable_capacity(ns: WgNamespace) -> int: + last_host = ns.subnet.num_addresses - 1 + in_range_reserved = sum(1 for offset in ns.reserved if 0 < offset < last_host) + return max(0, last_host - 1 - in_range_reserved) + + +def _take_offset(ns: WgNamespace, row: WireGuardSubnet) -> int: + """Pop the lowest valid free offset, or advance the high-water mark. Mutates the row.""" + free = list(row.free_offsets or []) + valid = [offset for offset in free if offset_valid(ns, offset)] + if valid: + offset = min(valid) + free.remove(offset) + row.free_offsets = free + return offset + offset = row.next_offset + while offset in ns.reserved: + offset += 1 + if offset >= ns.subnet.num_addresses - 1: + raise ValueError(f"WireGuard subnet {ns.subnet} has no free addresses") + row.next_offset = offset + 1 + return offset + + +def _trim_free(offsets: list[int]) -> list[int]: + if len(offsets) <= FREE_OFFSETS_CAP: + return offsets + return sorted(offsets)[:FREE_OFFSETS_CAP] + + +def _give_back(row: WireGuardSubnet, offsets: Iterable[int]) -> None: + current = set(row.free_offsets or []) + add = [offset for offset in offsets if offset > 0 and offset < row.next_offset and offset not in current] + if add: + row.free_offsets = _trim_free(list(row.free_offsets or []) + add) + + +async def tags_from_groups(groups: Iterable) -> set[str]: + """Inbound tags of enabled groups (for freshly-created users whose association rows aren't queryable yet).""" + tags: set[str] = set() + for group in groups or []: + if getattr(group, "is_disabled", False): + continue + inbounds = group.__dict__.get("inbounds") + if inbounds is None: + inbounds = await group.awaitable_attrs.inbounds + tags.update(inbound.tag for inbound in inbounds) + return tags + + +def _ensure_wireguard_keys(db_user: User) -> bool: + """Fill missing WG keys in proxy_settings. Returns True if the user was changed.""" + proxy_settings = dict(db_user.proxy_settings or {}) + wg = dict(proxy_settings.get("wireguard") or {}) + private_key = wg.get("private_key") + if private_key and wg.get("public_key"): + return False + if private_key: + wg["public_key"] = get_wireguard_public_key(private_key) + else: + wg["private_key"], wg["public_key"] = generate_wireguard_keypair() + proxy_settings["wireguard"] = wg + db_user.proxy_settings = proxy_settings + return True + + +def _user_peer_ips(db_user_settings: dict | None) -> list[str]: + return list(((db_user_settings or {}).get("wireguard") or {}).get("peer_ips") or []) + + +def _set_user_peer_ips(db_user: User, peer_ips: list[str]) -> None: + proxy_settings = dict(db_user.proxy_settings or {}) + wg = dict(proxy_settings.get("wireguard") or {}) + wg["peer_ips"] = peer_ips + proxy_settings["wireguard"] = wg + db_user.proxy_settings = proxy_settings + + +def _peer_sort_key(entry: str) -> tuple[int, int]: + host = peer_host(entry) + return host if host is not None else (99, 0) + + +# --- queries ---------------------------------------------------------------- + + +async def get_wg_cores(db: AsyncSession) -> list[CoreConfig]: + result = await db.execute(select(CoreConfig).where(CoreConfig.type == CoreType.wg)) + return list(result.scalars().all()) + + +async def get_users_accessible_tags(db: AsyncSession, user_ids: list[int]) -> dict[int, set[str]]: + """user_id -> inbound tags reachable through enabled groups. One joined SELECT.""" + if not user_ids: + return {} + stmt = ( + select(users_groups_association.c.user_id, ProxyInbound.tag) + .join( + Group, + and_(Group.id == users_groups_association.c.groups_id, Group.is_disabled.is_(False)), + ) + .join(inbounds_groups_association, inbounds_groups_association.c.group_id == Group.id) + .join(ProxyInbound, ProxyInbound.id == inbounds_groups_association.c.inbound_id) + .where(users_groups_association.c.user_id.in_(user_ids)) + ) + tags_by_user: dict[int, set[str]] = {} + for user_id, tag in (await db.execute(stmt)).all(): + tags_by_user.setdefault(user_id, set()).add(tag) + return tags_by_user + + +async def _lock_subnet_rows(db: AsyncSession, keys: Iterable[str]) -> dict[str, WireGuardSubnet]: + """Get-or-create the pool rows and lock them for this transaction (sorted for deadlock safety).""" + keys = sorted(set(keys)) + if not keys: + return {} + dialect = db.bind.dialect.name + values = [{"network": key, "next_offset": 1, "free_offsets": []} for key in keys] + if dialect == "postgresql": + stmt = pg_insert(WireGuardSubnet).on_conflict_do_nothing(index_elements=["network"]) + elif dialect == "mysql": + stmt = mysql_insert(WireGuardSubnet).on_duplicate_key_update(network=WireGuardSubnet.network) + else: # SQLite + stmt = insert(WireGuardSubnet).prefix_with("OR IGNORE") + await db.execute(stmt, values) + + stmt = ( + select(WireGuardSubnet) + .where(WireGuardSubnet.network.in_(keys)) + .order_by(WireGuardSubnet.network) + .execution_options(populate_existing=True) + ) + if dialect != "sqlite": # SQLite serializes writes; FOR UPDATE is a no-op there anyway + stmt = stmt.with_for_update() + rows = (await db.execute(stmt)).scalars().all() + return {row.network: row for row in rows} + + +# --- sync (the single trigger) ---------------------------------------------- + + +async def sync_users_allocations( + db: AsyncSession, + users: list[User], + *, + tags_by_user: dict[int, set[str]] | None = None, +) -> list[User]: + """Reconcile each user's wireguard peer_ips and keys with their current group access. + + Allocates from the subnet pool rows for newly-accessible subnets, releases entries for + subnets the user can no longer reach, and fills missing WG keys. All pool updates and + user updates happen in the caller's transaction. Flushes; the caller commits. + Returns the users that changed. + """ + if not users: + return [] + namespaces = wg_namespaces(await get_wg_cores(db)) + if tags_by_user is None: + tags_by_user = await get_users_accessible_tags(db, [user.id for user in users]) + + touched_keys: set[str] = set() + for user in users: + tags = tags_by_user.get(user.id, set()) + touched_keys.update(ns.key for ns in namespaces.values() if ns.tags & tags) + for entry in _user_peer_ips(user.proxy_settings): + host = peer_host(entry) + if host is not None and (ns := match_namespace(namespaces, *host)): + touched_keys.add(ns.key) + rows = await _lock_subnet_rows(db, touched_keys) + + changed: list[User] = [] + for user in users: + tags = tags_by_user.get(user.id, set()) + targets = {ns.key for ns in namespaces.values() if ns.tags & tags} + old_ips = _user_peer_ips(user.proxy_settings) + + kept: dict[str, int] = {} + for entry in old_ips: + host = peer_host(entry) + if host is None: + continue + ns = match_namespace(namespaces, *host) + if ns is None: + continue # foreign/legacy entry: drop, nothing to give back + offset = host[1] - int(ns.subnet.network_address) + if ns.key in targets and ns.key not in kept and offset_valid(ns, offset): + kept[ns.key] = offset + else: + _give_back(rows[ns.key], [offset]) + + for key in sorted(targets - set(kept)): + kept[key] = _take_offset(namespaces[key], rows[key]) + + new_ips = sorted( + (render_peer_ip(namespaces[key].subnet, offset) for key, offset in kept.items()), + key=_peer_sort_key, + ) + user_changed = False + if new_ips != old_ips: + _set_user_peer_ips(user, new_ips) + user_changed = True + if targets and _ensure_wireguard_keys(user): + user_changed = True + if user_changed: + changed.append(user) + + if changed or rows: + await db.flush() + return changed + + +async def sync_user_allocations( + db: AsyncSession, + db_user: User, + *, + accessible_tags: set[str] | None = None, +) -> bool: + tags_by_user = {db_user.id: accessible_tags} if accessible_tags is not None else None + return bool(await sync_users_allocations(db, [db_user], tags_by_user=tags_by_user)) + + +async def _release_proxy_settings(db: AsyncSession, settings_list: Iterable[dict | None]) -> None: + namespaces = wg_namespaces(await get_wg_cores(db)) + releases: dict[str, list[int]] = {} + for proxy_settings in settings_list: + for entry in _user_peer_ips(proxy_settings): + host = peer_host(entry) + if host is not None and (ns := match_namespace(namespaces, *host)): + releases.setdefault(ns.key, []).append(host[1] - int(ns.subnet.network_address)) + if not releases: + return + rows = await _lock_subnet_rows(db, releases) + for key, offsets in releases.items(): + _give_back(rows[key], offsets) + await db.flush() + + +async def release_users_allocations(db: AsyncSession, users: list[User]) -> None: + """Give the users' peer IPs back to the pool (called right before user deletion).""" + if not users: + return + await _release_proxy_settings(db, [user.proxy_settings for user in users]) + + +async def release_allocations_by_user_ids(db: AsyncSession, user_ids: list[int]) -> None: + """Same as release_users_allocations for delete paths that only hold user ids.""" + if not user_ids: + return + settings_list = (await db.execute(select(User.proxy_settings).where(User.id.in_(user_ids)))).scalars().all() + await _release_proxy_settings(db, settings_list) + + +# --- full rebuild (core lifecycle & repair) --------------------------------- + + +async def reconcile_wireguard_subnets(db: AsyncSession) -> list[int]: + """Rebuild pool rows and fix user peer_ips from ground truth after a WG core change. + + Handles subnet resize/move, server-IP moves, interface renames, duplicate IPs and + orphaned namespaces. Flushes; the caller commits. Returns changed user ids. + """ + namespaces = wg_namespaces(await get_wg_cores(db)) + + # Lock live namespaces up front so sync cannot race the in-memory alloc pass. + # Also lock any orphan pool rows we are about to drop. + if namespaces: + orphan_stmt = select(WireGuardSubnet.network).where(WireGuardSubnet.network.not_in(list(namespaces))) + else: + orphan_stmt = select(WireGuardSubnet.network) + orphan_keys = (await db.execute(orphan_stmt)).scalars().all() + await _lock_subnet_rows(db, list(namespaces) + list(orphan_keys)) + + # ponytail: full lightweight user scan — core saves are rare admin operations + user_rows = (await db.execute(select(User.id, User.proxy_settings))).all() + all_tags = {tag for ns in namespaces.values() for tag in ns.tags} + eligible_tags = await get_users_accessible_tags(db, [row[0] for row in user_rows]) if all_tags else {} + + used: dict[str, set[int]] = {key: set() for key in namespaces} + desired: dict[int, list[str]] = {} + need_alloc: dict[str, list[int]] = {} # key -> user_ids (in scan order) + for user_id, proxy_settings in user_rows: + tags = eligible_tags.get(user_id, set()) + targets = {ns.key for ns in namespaces.values() if ns.tags & tags} + old_ips = _user_peer_ips(proxy_settings) + kept: dict[str, int] = {} + for entry in old_ips: + host = peer_host(entry) + if host is None: + continue + ns = match_namespace(namespaces, *host) + if ns is None: + continue + offset = host[1] - int(ns.subnet.network_address) + # first holder keeps a duplicated IP; later holders get reallocated + if ns.key in targets and ns.key not in kept and offset_valid(ns, offset) and offset not in used[ns.key]: + kept[ns.key] = offset + used[ns.key].add(offset) + for key in sorted(targets - set(kept)): + need_alloc.setdefault(key, []).append(user_id) + desired[user_id] = [render_peer_ip(namespaces[key].subnet, offset) for key, offset in sorted(kept.items())] + + # in-memory allocation under the row locks taken above + granted: dict[int, dict[str, int]] = {} # user_id -> {key: offset} + for key, user_ids in need_alloc.items(): + ns = namespaces[key] + cursor = 1 + for user_id in user_ids: + while cursor in ns.reserved or cursor in used[key]: + cursor += 1 + if cursor >= ns.subnet.num_addresses - 1: + logger.warning(f"WireGuard subnet {ns.subnet} exhausted; user {user_id} left without an IP") + continue + used[key].add(cursor) + granted.setdefault(user_id, {})[key] = cursor + cursor += 1 + + changed_ids: list[int] = [] + settings_by_id = {row[0]: row[1] for row in user_rows} + for user_id, ips in desired.items(): + for key, offset in sorted(granted.get(user_id, {}).items()): + ips.append(render_peer_ip(namespaces[key].subnet, offset)) + ips.sort(key=_peer_sort_key) + if ips != _user_peer_ips(settings_by_id[user_id]): + changed_ids.append(user_id) + + if changed_ids: + users = (await db.execute(select(User).where(User.id.in_(changed_ids)))).scalars().all() + for user in users: + _set_user_peer_ips(user, desired[user.id]) + if desired[user.id]: + _ensure_wireguard_keys(user) + + # rebuild pool rows from the used sets; drop rows for dead namespaces + if used: + await db.execute(delete(WireGuardSubnet).where(WireGuardSubnet.network.not_in(list(used)))) + rows = await _lock_subnet_rows(db, used) + for key, offsets in used.items(): + ns = namespaces[key] + row = rows[key] + row.next_offset = max(offsets) + 1 if offsets else 1 + row.free_offsets = _trim_free( + [offset for offset in range(1, row.next_offset) if offset not in offsets and offset_valid(ns, offset)] + ) + else: + await db.execute(delete(WireGuardSubnet)) + await db.flush() + return changed_ids + + +# --- visibility ------------------------------------------------------------- + + +async def get_subnet_usage(db: AsyncSession, *, free_limit: int = FREE_IPS_LIMIT) -> list[dict]: + """Per-subnet address usage: capacity, used/free counts, and the first free IPs.""" + namespaces = wg_namespaces(await get_wg_cores(db)) + pool_rows = (await db.execute(select(WireGuardSubnet))).scalars().all() + rows_by_key = {row.network: row for row in pool_rows} + + usage: list[dict] = [] + for key in sorted(namespaces): + ns = namespaces[key] + row = rows_by_key.get(key) + next_offset = row.next_offset if row else 1 + free_list = sorted(offset for offset in (row.free_offsets if row else []) or [] if offset_valid(ns, offset)) + reserved_below = sum(1 for offset in ns.reserved if 0 < offset < next_offset) + used = max(0, (next_offset - 1) - reserved_below - len(free_list)) + capacity = _usable_capacity(ns) + + free_ips = [ + str(_host_address(ns.subnet.version, int(ns.subnet.network_address) + offset)) + for offset in free_list[:free_limit] + ] + offset = next_offset + last_host = ns.subnet.num_addresses - 1 + while len(free_ips) < free_limit and offset < last_host: + if offset not in ns.reserved: + free_ips.append(str(_host_address(ns.subnet.version, int(ns.subnet.network_address) + offset))) + offset += 1 + + usage.append( + { + "subnet": key, + "interface_tags": sorted(ns.tags), + "capacity": capacity, + "used": used, + "free": max(0, capacity - used), + "free_ips": free_ips, + } + ) + return usage diff --git a/app/db/migrations/versions/3c1a7e5b9d20_add_wireguard_subnets_pool_table.py b/app/db/migrations/versions/3c1a7e5b9d20_add_wireguard_subnets_pool_table.py new file mode 100644 index 000000000..20dc282d4 --- /dev/null +++ b/app/db/migrations/versions/3c1a7e5b9d20_add_wireguard_subnets_pool_table.py @@ -0,0 +1,261 @@ +"""add wireguard_subnets pool table + +Creates the per-subnet WireGuard allocation pool (free-list + high-water offset) +keyed by exact canonical network CIDR (IPv4 and IPv6), and backfills it from WG +core configs and users' existing peer_ips, preserving every IP that fits its +core's subnet. Users whose IPs don't fit get a fresh one. + +Revision ID: 3c1a7e5b9d20 +Revises: f976bfcf4738 +Create Date: 2026-07-19 00:00:00.000000 + +""" + +import json +from ipaddress import ip_interface, ip_network + +import sqlalchemy as sa +from alembic import op + +from app.db.compiles_types import SqliteCompatibleBigInteger + +revision = "3c1a7e5b9d20" +down_revision = "f976bfcf4738" +branch_labels = None +depends_on = None + + +def _load_json(value): + if isinstance(value, str): + try: + return json.loads(value) + except json.JSONDecodeError: + return {} + return value or {} + + +def _core_subnets(config): + seen = {} + for cidr in config.get("address") or []: + try: + net = ip_interface(str(cidr).strip()).network + except ValueError: + continue + seen.setdefault(str(net), net) + return list(seen.values()) + + +def _peer_host(entry): + try: + net = ip_network(str(entry).strip(), strict=False) + except ValueError: + return None + return net.version, int(net.network_address) + + +def _build_namespaces(core_rows): + """canonical network str -> {subnet, reserved offsets, tags}""" + by_key = {} + for config in core_rows: + for subnet in _core_subnets(config): + by_key.setdefault(str(subnet), []).append(config) + + namespaces = {} + for key, configs in by_key.items(): + subnet = ip_network(key) + base = int(subnet.network_address) + reserved = {0, subnet.num_addresses - 1} + tags = set() + for config in configs: + tag = str(config.get("interface_name") or "").strip() + if tag: + tags.add(tag) + for cidr in config.get("address") or []: + try: + iface = ip_interface(str(cidr).strip()) + except ValueError: + continue + if iface.version == subnet.version and iface.ip in subnet: + reserved.add(int(iface.ip) - base) + namespaces[key] = {"subnet": subnet, "reserved": reserved, "tags": tags} + return _collapse_overlapping(namespaces) + + +def _collapse_overlapping(namespaces): + """Keep the largest subnet when CIDRs overlap; merge tags/reserved from smaller ones into it.""" + if len(namespaces) < 2: + return namespaces + + ordered = sorted(namespaces.items(), key=lambda item: (item[1]["subnet"].prefixlen, item[0])) + kept = {} + for key, ns in ordered: + keeper_key = None + for kkey, kns in kept.items(): + if kns["subnet"].version != ns["subnet"].version: + continue + if kns["subnet"].overlaps(ns["subnet"]): + keeper_key = kkey + break + if keeper_key is None: + kept[key] = { + "subnet": ns["subnet"], + "reserved": set(ns["reserved"]), + "tags": set(ns["tags"]), + } + continue + + print( + f"WARNING: WireGuard subnet {key} overlaps {keeper_key}; " + f"skipping pool row for smaller/overlapping {key}, merging tags into {keeper_key}" + ) + keeper = kept[keeper_key] + base = int(keeper["subnet"].network_address) + for offset in ns["reserved"]: + host_int = int(ns["subnet"].network_address) + offset + host = type(keeper["subnet"].network_address)(host_int) + if host in keeper["subnet"]: + keeper["reserved"].add(host_int - base) + keeper["tags"] |= ns["tags"] + return kept + + +def _render_peer_ip(subnet, offset): + host = type(subnet.network_address)(int(subnet.network_address) + offset) + return f"{host}/{'32' if subnet.version == 4 else '128'}" + + +def upgrade() -> None: + op.create_table( + "wireguard_subnets", + sa.Column("id", SqliteCompatibleBigInteger(), autoincrement=True, nullable=False), + sa.Column("network", sa.String(length=64), nullable=False), + sa.Column("next_offset", SqliteCompatibleBigInteger(), nullable=False, server_default="1"), + sa.Column("free_offsets", sa.JSON(none_as_null=True), nullable=False), + sa.PrimaryKeyConstraint("id", name=op.f("pk_wireguard_subnets")), + sa.UniqueConstraint("network", name=op.f("uq_wireguard_subnets_network")), + ) + + bind = op.get_bind() + + core_configs = sa.table( + "core_configs", sa.column("id", sa.Integer), sa.column("type", sa.String), sa.column("config", sa.JSON) + ) + users = sa.table("users", sa.column("id", sa.Integer), sa.column("proxy_settings", sa.JSON)) + users_groups = sa.table( + "users_groups_association", sa.column("user_id", sa.Integer), sa.column("groups_id", sa.Integer) + ) + groups = sa.table("groups", sa.column("id", sa.Integer), sa.column("is_disabled", sa.Boolean)) + inbounds_groups = sa.table( + "inbounds_groups_association", sa.column("inbound_id", sa.Integer), sa.column("group_id", sa.Integer) + ) + inbounds = sa.table("inbounds", sa.column("id", sa.Integer), sa.column("tag", sa.String)) + pool = sa.table( + "wireguard_subnets", + sa.column("network", sa.String), + sa.column("next_offset", sa.BigInteger), + sa.column("free_offsets", sa.JSON), + ) + + core_rows = [ + _load_json(config) + for (config,) in bind.execute( + sa.select(core_configs.c.config).where(sa.cast(core_configs.c.type, sa.String) == "wg") + ).fetchall() + ] + namespaces = _build_namespaces(core_rows) + if not namespaces: + return + + # user_id -> accessible inbound tags (through enabled groups) + tag_rows = bind.execute( + sa.select(users_groups.c.user_id, inbounds.c.tag) + .select_from( + users_groups.join( + groups, sa.and_(groups.c.id == users_groups.c.groups_id, groups.c.is_disabled.is_(False)) + ) + .join(inbounds_groups, inbounds_groups.c.group_id == groups.c.id) + .join(inbounds, inbounds.c.id == inbounds_groups.c.inbound_id) + ) + ).fetchall() + tags_by_user = {} + for user_id, tag in tag_rows: + tags_by_user.setdefault(user_id, set()).add(tag) + + def match_namespace(version, host_int): + for key, ns in namespaces.items(): + subnet = ns["subnet"] + if subnet.version != version: + continue + if int(subnet.network_address) <= host_int <= int(subnet.broadcast_address): + return key + return None + + used = {key: set() for key in namespaces} + updates = [] + user_rows = bind.execute(sa.select(users.c.id, users.c.proxy_settings)).fetchall() + parsed_users = [] + for user_id, proxy_settings in user_rows: + proxy_settings = _load_json(proxy_settings) + old_ips = list((proxy_settings.get("wireguard") or {}).get("peer_ips") or []) + tags = tags_by_user.get(user_id, set()) + targets = {key for key, ns in namespaces.items() if ns["tags"] & tags} + kept = {} + for entry in old_ips: + host = _peer_host(entry) + key = match_namespace(*host) if host is not None else None + if key is None: + continue + ns = namespaces[key] + offset = host[1] - int(ns["subnet"].network_address) + if ( + key in targets + and key not in kept + and 0 < offset < ns["subnet"].num_addresses - 1 + and offset not in ns["reserved"] + and offset not in used[key] + ): + kept[key] = offset + used[key].add(offset) + parsed_users.append((user_id, proxy_settings, old_ips, targets, kept)) + + # in-memory allocation for eligible users missing an IP (scan order = user id order) + cursors = {key: 1 for key in namespaces} + for user_id, proxy_settings, old_ips, targets, kept in parsed_users: + for key in sorted(targets - set(kept)): + ns = namespaces[key] + cursor = cursors[key] + while cursor in ns["reserved"] or cursor in used[key]: + cursor += 1 + cursors[key] = cursor + 1 + if cursor >= ns["subnet"].num_addresses - 1: + continue # subnet exhausted; user is left without an IP for it + used[key].add(cursor) + kept[key] = cursor + + new_ips = [_render_peer_ip(namespaces[key]["subnet"], offset) for key, offset in sorted(kept.items())] + if new_ips != old_ips: + wg = dict(proxy_settings.get("wireguard") or {}) + wg["peer_ips"] = new_ips + proxy_settings = dict(proxy_settings) + proxy_settings["wireguard"] = wg + updates.append({"_id": user_id, "proxy_settings": proxy_settings}) + + if updates: + bind.execute(users.update().where(users.c.id == sa.bindparam("_id")), updates) + + pool_rows = [] + for key, ns in namespaces.items(): + offsets = used[key] + next_offset = max(offsets) + 1 if offsets else 1 + free = [ + offset + for offset in range(1, next_offset) + if offset not in offsets and offset not in ns["reserved"] and offset < ns["subnet"].num_addresses - 1 + ] + pool_rows.append({"network": key, "next_offset": next_offset, "free_offsets": free}) + bind.execute(pool.insert(), pool_rows) + + +def downgrade() -> None: + # peer_ips live in users.proxy_settings either way; only the pool state is dropped + op.drop_table("wireguard_subnets") diff --git a/app/db/migrations/versions/a1b2c3d4e5f6_add_wireguard_settings_to_proxy_settings.py b/app/db/migrations/versions/a1b2c3d4e5f6_add_wireguard_settings_to_proxy_settings.py index f81d364e8..9074dee92 100644 --- a/app/db/migrations/versions/a1b2c3d4e5f6_add_wireguard_settings_to_proxy_settings.py +++ b/app/db/migrations/versions/a1b2c3d4e5f6_add_wireguard_settings_to_proxy_settings.py @@ -11,8 +11,6 @@ from alembic import op from app.utils.crypto import generate_wireguard_keypair, get_wireguard_public_key -from app.utils.ip_pool import WireGuardPeerIPAllocator, collect_used_peer_networks_from_proxy_settings_rows -from config import wireguard_settings revision = "a1b2c3d4e5f6" down_revision = "6b7a1e8c2d14" @@ -30,12 +28,6 @@ def upgrade() -> None: ) users = bind.execute(sa.select(users_table.c.id, users_table.c.proxy_settings)).fetchall() - user_rows = [{"id": user_id, "proxy_settings": proxy_settings} for user_id, proxy_settings in users] - allocator = ( - WireGuardPeerIPAllocator(collect_used_peer_networks_from_proxy_settings_rows(user_rows)) - if wireguard_settings.enabled - else None - ) updates = [] for user_id, proxy_settings in users: @@ -56,11 +48,6 @@ def upgrade() -> None: elif not wg.get("public_key"): wg["public_key"] = get_wireguard_public_key(wg["private_key"]) changed = True - if allocator is not None and not wg.get("peer_ips"): - peer_ip = allocator.allocate() - if peer_ip: - wg["peer_ips"] = [peer_ip] - changed = True if changed: proxy_settings["wireguard"] = wg updates.append({"_id": user_id, "proxy_settings": proxy_settings}) diff --git a/app/db/models.py b/app/db/models.py index e277a2886..f17b295af 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -837,6 +837,21 @@ class CoreConfig(Base, CreatedAtUTCMixin): fallbacks_inbound_tags: Mapped[Optional[set[str]]] = mapped_column(StringArray(2048), default_factory=set) +class WireGuardSubnet(Base, IdMixin): + """Allocation state of one WireGuard client subnet: a free-list plus a high-water offset. + + One row per exact client network (identical CIDRs across cores share the row). Allocating + pops the free-list or takes next_offset; releasing pushes the offset back. Users' assigned + IPs stay in their proxy_settings; this row only guarantees no offset is handed out twice. + Resizing changes the network key — reconcile rebuilds the row and keeps peer IPs that still fit. + """ + + __tablename__ = "wireguard_subnets" + network: Mapped[str] = mapped_column(String(64), unique=True) # canonical CIDR, e.g. 10.0.0.0/24 or fd00::/64 + next_offset: Mapped[int] = mapped_column(SqliteCompatibleBigInteger, default=1) + free_offsets: Mapped[list] = mapped_column(JSON(True), default_factory=list) + + class ClientTemplate(Base): __tablename__ = "client_templates" __table_args__ = ( diff --git a/app/models/host.py b/app/models/host.py index c0e17eb47..b03216773 100644 --- a/app/models/host.py +++ b/app/models/host.py @@ -268,12 +268,8 @@ class XMuxSettings(BaseModel): max_concurrency: str | None = Field(None, pattern=r"^\d{1,16}(-\d{1,16})?$", alias="maxConcurrency") max_connections: str | None = Field(None, pattern=r"^\d{1,16}(-\d{1,16})?$", alias="maxConnections") c_max_reuse_times: str | None = Field(None, pattern=r"^\d{1,16}(-\d{1,16})?$", alias="cMaxReuseTimes") - h_max_reusable_secs: str | None = Field( - None, pattern=r"^\d{1,16}(-\d{1,16})?$", alias="hMaxReusableSecs" - ) - h_max_request_times: str | None = Field( - None, pattern=r"^\d{1,16}(-\d{1,16})?$", alias="hMaxRequestTimes" - ) + h_max_reusable_secs: str | None = Field(None, pattern=r"^\d{1,16}(-\d{1,16})?$", alias="hMaxReusableSecs") + h_max_request_times: str | None = Field(None, pattern=r"^\d{1,16}(-\d{1,16})?$", alias="hMaxRequestTimes") h_keep_alive_period: int | None = Field(None, alias="hKeepAlivePeriod") @field_validator( diff --git a/app/models/reality_scan.py b/app/models/reality_scan.py index a1d918a36..22468eb3b 100644 --- a/app/models/reality_scan.py +++ b/app/models/reality_scan.py @@ -3,7 +3,9 @@ class RealityScanRequest(BaseModel): target: str = Field(min_length=1, max_length=253, description="host or host:port to probe (port defaults to 443)") - timeout: float | None = Field(default=None, ge=1, le=20, description="Per-probe timeout in seconds (1-20, default 10)") + timeout: float | None = Field( + default=None, ge=1, le=20, description="Per-probe timeout in seconds (1-20, default 10)" + ) class RealityScanResult(BaseModel): diff --git a/app/models/system.py b/app/models/system.py index 63a2c5040..4a0807d6a 100644 --- a/app/models/system.py +++ b/app/models/system.py @@ -43,3 +43,12 @@ class WorkerHealth(BaseModel): class WorkersHealth(BaseModel): scheduler: WorkerHealth node: WorkerHealth + + +class WireGuardSubnetUsage(BaseModel): + subnet: str + interface_tags: list[str] + capacity: int + used: int + free: int + free_ips: list[str] diff --git a/app/models/user.py b/app/models/user.py index 7a49adc9a..c84ef6a51 100644 --- a/app/models/user.py +++ b/app/models/user.py @@ -90,6 +90,7 @@ def group_ids_validator(cls, v): class UserModify(UserWithValidator): status: UserStatus | None = Field(default=None) proxy_settings: ProxyTable | None = Field(default=None) + group_ids: list[int] | None = Field(default=None) @field_validator("status", mode="before", check_fields=False) def validate_status(cls, status, values): @@ -443,13 +444,6 @@ class BulkUsersProxy(BulkUserFilter): method: ShadowsocksMethods | None = Field(default=None) -class BulkWireGuardPeerIPs(BulkUserFilter): - """Re-seat WireGuard peer IPs (same scoping as BulkUser: users, admins, group_ids, status).""" - - confirm: bool = False - replace_all: bool = False - - class BulkOperationDryRunResponse(BaseModel): """Preview for bulk user/group operations (no DB writes).""" @@ -457,15 +451,6 @@ class BulkOperationDryRunResponse(BaseModel): affected_users: int -class WireGuardPeerIPsReallocateResponse(BaseModel): - wireguard_inbound_tags: int - candidates: int - updated: int - dry_run: bool - sample_usernames: list[str] - affected_users: int - - class UsernameGenerationStrategy(str, Enum): random = "random" sequence = "sequence" diff --git a/app/operation/core.py b/app/operation/core.py index 58bdbccf9..6ee3edeef 100644 --- a/app/operation/core.py +++ b/app/operation/core.py @@ -13,6 +13,13 @@ remove_cores, ) from app.db.crud.host import get_hosts +from app.db.crud.user import get_users_by_ids +from app.db.crud.wireguard import ( + core_config_dict, + get_wg_cores, + reconcile_wireguard_subnets, + wg_core_subnets, +) from app.models.admin import AdminDetails from app.models.core import ( BulkCoreSelection, @@ -23,8 +30,10 @@ CoreSimpleListQuery, CoreSimple, CoresSimpleResponse, + CoreType, RemoveCoresResponse, ) +from app.node.sync import sync_users from app.models.reality_scan import RealityScanRequest, RealityScanResult from app.operation import BaseOperation from app.utils.logger import get_logger @@ -38,6 +47,37 @@ async def _refresh_hosts_from_db(self, db: AsyncSession) -> None: db_hosts = await get_hosts(db=db) await host_manager.add_hosts(db, db_hosts) + async def _validate_wireguard_subnet(self, db: AsyncSession, config: dict, *, exclude_core_id: int | None) -> None: + """WG cores need at least one client subnet (v4 and/or v6). Overlapping subnets + are rejected; identical CIDRs may be shared across cores (one allocation namespace).""" + subnets = wg_core_subnets(config) + if not subnets: + await self.raise_error(message="WireGuard core needs an IPv4 or IPv6 interface address", code=400, db=db) + for other in await get_wg_cores(db): + if other.id == exclude_core_id: + continue + for subnet in subnets: + for other_subnet in wg_core_subnets(core_config_dict(other)): + if other_subnet == subnet: + continue + if subnet.overlaps(other_subnet): + await self.raise_error( + message=( + f"WireGuard subnet {subnet} overlaps {other_subnet} " + f"of core '{other.name}' — use the identical CIDR or a disjoint subnet" + ), + code=400, + db=db, + ) + + async def _reconcile_wireguard(self, db: AsyncSession) -> None: + """Fix pool rows and user peer IPs after a WG core change, then resync changed users.""" + changed_ids = await reconcile_wireguard_subnets(db) + await db.commit() + if changed_ids: + users = await get_users_by_ids(db, changed_ids, load_admin_role=True) + await sync_users(users) + async def scan_reality_target(self, request: RealityScanRequest) -> RealityScanResult: try: result = await scan_reality_target(target=request.target, timeout=request.timeout) @@ -49,6 +89,8 @@ async def scan_reality_target(self, request: RealityScanRequest) -> RealityScanR return RealityScanResult.model_validate(result) async def create_core(self, db: AsyncSession, new_core: CoreCreate, admin: AdminDetails) -> CoreResponse: + if new_core.type == CoreType.wg: + await self._validate_wireguard_subnet(db, new_core.config, exclude_core_id=None) try: validated_core = core_manager.validate_core( new_core.config, @@ -66,6 +108,8 @@ async def create_core(self, db: AsyncSession, new_core: CoreCreate, admin: Admin core = CoreResponse.model_validate(db_core) asyncio.create_task(notification.create_core(core, admin.username)) + if new_core.type == CoreType.wg: + await self._reconcile_wireguard(db) await self._refresh_hosts_from_db(db) return core @@ -86,6 +130,9 @@ async def modify_core( self, db: AsyncSession, core_id: int, modified_core: CoreCreate, admin: AdminDetails ) -> CoreResponse: db_core = await self.get_validated_core_config(db, core_id) + was_wg = db_core.type == CoreType.wg + if modified_core.type == CoreType.wg: + await self._validate_wireguard_subnet(db, modified_core.config, exclude_core_id=db_core.id) try: validated_core = core_manager.validate_core( modified_core.config, @@ -104,6 +151,8 @@ async def modify_core( core = CoreResponse.model_validate(db_core) asyncio.create_task(notification.modify_core(core, admin.username)) + if was_wg or modified_core.type == CoreType.wg: + await self._reconcile_wireguard(db) await self._refresh_hosts_from_db(db) return core @@ -113,6 +162,7 @@ async def delete_core(self, db: AsyncSession, core_id: int, admin: AdminDetails) return await self.raise_error(message="Cannot delete default core config", code=403) db_core = await self.get_validated_core_config(db, core_id) + was_wg = db_core.type == CoreType.wg await remove_core_config(db, db_core) await core_manager.remove_core(db_core.id) @@ -121,6 +171,8 @@ async def delete_core(self, db: AsyncSession, core_id: int, admin: AdminDetails) logger.info(f'core config "{db_core.name}" deleted by admin "{admin.username}"') + if was_wg: + await self._reconcile_wireguard(db) await self._refresh_hosts_from_db(db) async def bulk_remove_cores( @@ -141,6 +193,7 @@ async def bulk_remove_cores( core_ids = [c.id for c in db_cores_list] core_names = [c.name for c in db_cores_list] + any_wg = any(c.type == CoreType.wg for c in db_cores_list) # Batch delete using CRUD function await remove_cores(db, core_ids) @@ -151,6 +204,8 @@ async def bulk_remove_cores( asyncio.create_task(notification.remove_core(core_id, admin.username)) logger.info(f'core config "{core_name}" deleted by admin "{admin.username}"') + if any_wg: + await self._reconcile_wireguard(db) await self._refresh_hosts_from_db(db) return RemoveCoresResponse(cores=core_names, count=len(db_cores_list)) diff --git a/app/operation/group.py b/app/operation/group.py index 173657a8c..394e5537f 100644 --- a/app/operation/group.py +++ b/app/operation/group.py @@ -14,7 +14,7 @@ remove_groups, ) from app.db.crud.user import get_users -from app.db.models import Admin, UserStatus +from app.db.models import Admin from app.models.group import ( BulkGroupsActionResponse, BulkGroup, @@ -31,10 +31,10 @@ RemoveGroupsResponse, ) from app.models.user import BulkOperationDryRunResponse, UserListQuery +from app.db.crud.wireguard import sync_users_allocations from app.node.sync import sync_users from app.operation import BaseOperation, OperatorType from app.operation.permissions import apply_group_access -from app.utils.wireguard import bulk_reallocate_wireguard_peer_ips from app.utils.logger import get_logger logger = get_logger("group-operation") @@ -50,6 +50,12 @@ async def _get_group_with_access(self, db: AsyncSession, group_id: int, admin: A db_group = await self.get_validated_group(db, group_id) return db_group + async def _sync_users_allocations(self, db: AsyncSession, users) -> None: + try: + await sync_users_allocations(db, users) + except ValueError as exc: # WireGuard subnet exhausted + await self.raise_error(message=str(exc), code=400, db=db) + async def create_group(self, db: AsyncSession, new_group: GroupCreate, admin: Admin) -> Group: await self.check_inbound_tags(new_group.inbound_tags) db_group = await create_group(db, new_group) @@ -86,10 +92,11 @@ async def modify_group(self, db: AsyncSession, group_id: int, modified_group: Gr users = await get_users( db, - query=UserListQuery(group_ids=[db_group.id], status=[UserStatus.active, UserStatus.on_hold]), + query=UserListQuery(group_ids=[db_group.id]), load_admin_role=True, ) - await bulk_reallocate_wireguard_peer_ips(db, users, dry_run=False, replace_all=False) + await self._sync_users_allocations(db, users) + await db.commit() await sync_users(users) group = GroupResponse.model_validate(db_group) @@ -108,6 +115,8 @@ async def remove_group(self, db: AsyncSession, group_id: int, admin: Admin) -> N await remove_group(db, db_group) users = await get_users(db, query=UserListQuery(username=username_list), load_admin_role=True) + await self._sync_users_allocations(db, users) + await db.commit() await sync_users(users) logger.info(f'Group "{db_group.name}" deleted by admin "{admin.username}"') @@ -121,7 +130,8 @@ async def bulk_add_groups(self, db: AsyncSession, bulk_model: BulkGroup): return BulkOperationDryRunResponse(affected_users=n) users, users_count = await add_groups_to_users(db, bulk_model) - await bulk_reallocate_wireguard_peer_ips(db, users, dry_run=False, replace_all=False) + await self._sync_users_allocations(db, users) + await db.commit() await sync_users(users) if self.operator_type in (OperatorType.API, OperatorType.WEB): @@ -135,6 +145,8 @@ async def bulk_remove_groups(self, db: AsyncSession, bulk_model: BulkGroup): return BulkOperationDryRunResponse(affected_users=n) users, users_count = await remove_groups_from_users(db, bulk_model) + await self._sync_users_allocations(db, users) + await db.commit() await sync_users(users) if self.operator_type in (OperatorType.API, OperatorType.WEB): @@ -169,6 +181,8 @@ async def bulk_remove_groups_by_id( users = await get_users( db, query=UserListQuery(username=list(all_affected_usernames)), load_admin_role=True ) + await self._sync_users_allocations(db, users) + await db.commit() await sync_users(users) for name, group_id in zip(group_names, group_ids): @@ -212,12 +226,11 @@ async def bulk_set_groups_disabled( if groups_to_update: users = await get_users( db, - query=UserListQuery( - group_ids=[group.id for group in groups_to_update], - status=[UserStatus.active, UserStatus.on_hold], - ), + query=UserListQuery(group_ids=[group.id for group in groups_to_update]), load_admin_role=True, ) + await self._sync_users_allocations(db, users) + await db.commit() await sync_users(users) for db_group in groups_to_update: diff --git a/app/operation/user.py b/app/operation/user.py index d6752bc45..443d9b430 100644 --- a/app/operation/user.py +++ b/app/operation/user.py @@ -16,7 +16,6 @@ count_bulk_datalimit_targets, count_bulk_expire_targets, count_bulk_proxy_targets, - get_bulk_wireguard_peer_ip_users, reset_all_users_data_usage, update_users_datalimit, update_users_expire, @@ -74,7 +73,6 @@ BulkUsersProxy, BulkUsersSelection, BulkUsersSetOwner, - BulkWireGuardPeerIPs, CreateUserFromTemplate, ExpiredUsersQuery, ModifyUserByTemplate, @@ -95,7 +93,6 @@ UserSubscriptionUpdateList, UsersUsageQuery, UserUsageQuery, - WireGuardPeerIPsReallocateResponse, ) from app.node.sync import remove_user as sync_remove_user, sync_user, sync_users from app.operation import BaseOperation, OperatorType @@ -113,16 +110,8 @@ from app.utils.jwt import create_subscription_token from app.utils.logger import get_logger from app.utils.system import readable_duration, readable_size -from app.utils.wireguard import ( - build_wireguard_peer_ip_allocator, - bulk_reallocate_wireguard_peer_ips as run_bulk_reallocate_wireguard_peer_ips, - ensure_unique_wireguard_public_key, - get_wireguard_tags_from_groups, - prepare_wireguard_keys_only, - prepare_wireguard_proxy_settings, - prepare_wireguard_proxy_settings_with_allocator, -) -from config import subscription_env_settings, usage_settings, wireguard_settings +from app.utils.wireguard import ensure_unique_wireguard_public_key, prepare_wireguard_keys +from config import subscription_env_settings, usage_settings def _has_permission(admin: AdminDetails, resource: str, action: str) -> bool: @@ -379,26 +368,14 @@ async def _persist_bulk_users( next_plan=user_to_create.next_plan, ) - wireguard_tags = await get_wireguard_tags_from_groups(groups) - use_shared_allocator = bool(wireguard_tags) and wireguard_settings.enabled - - if use_shared_allocator: - allocator = await build_wireguard_peer_ip_allocator(db) - for user_to_create in users_to_create: - try: - user_to_create.proxy_settings = prepare_wireguard_proxy_settings_with_allocator( - user_to_create.proxy_settings, - allocator, - ) - except ValueError as exc: - await self.raise_error(message=str(exc), code=400, db=db) - else: - for user_to_create in users_to_create: - user_to_create.proxy_settings = await self._prepare_user_proxy_settings( - db, - groups, - user_to_create.proxy_settings, - ) + for user_to_create in users_to_create: + # peer IPs are never taken from input; the subnet pool assigns them at creation + user_to_create.proxy_settings.wireguard.peer_ips = [] + user_to_create.proxy_settings = await self._prepare_user_proxy_settings( + db, + groups, + user_to_create.proxy_settings, + ) duplicate_key = _duplicate_wireguard_public_key_usernames(users_to_create) if duplicate_key is not None: @@ -416,7 +393,10 @@ async def _persist_bulk_users( except ValueError as exc: await self.raise_error(message=str(exc), code=400, db=db) - db_users = await create_users_bulk(db, users_to_create, groups, db_admin, commit=commit) + try: + db_users = await create_users_bulk(db, users_to_create, groups, db_admin, commit=commit) + except ValueError as exc: # WireGuard subnet exhausted + await self.raise_error(message=str(exc), code=400, db=db) if not commit: for user in db_users: await load_user_attrs(user, load_admin_role=True) @@ -451,23 +431,14 @@ async def _prepare_user_proxy_settings( proxy_settings: ProxyTable, *, exclude_user_id: int | None = None, - skip_peer_ip_validation: bool = False, ) -> ProxyTable: try: - if skip_peer_ip_validation: - return await prepare_wireguard_keys_only( - db, - proxy_settings, - groups, - exclude_user_id=exclude_user_id, - ) - else: - return await prepare_wireguard_proxy_settings( - db, - proxy_settings, - groups, - exclude_user_id=exclude_user_id, - ) + return await prepare_wireguard_keys( + db, + proxy_settings, + groups, + exclude_user_id=exclude_user_id, + ) except ValueError as exc: await self.raise_error(message=str(exc), code=400, db=db) @@ -481,7 +452,6 @@ async def _prepare_revoked_proxy_settings(self, db: AsyncSession, db_user: User) groups, ProxyTable.model_validate(build_revoked_proxy_settings(db_user)), exclude_user_id=db_user.id, - skip_peer_ip_validation=True, ) async def _get_validated_template_with_access( @@ -702,12 +672,16 @@ async def create_user( all_groups = await self.validate_all_groups(db, new_user) db_admin = await get_admin(db, admin.username, load_users=False, load_usage_logs=False) + # peer IPs are never taken from input; the subnet pool assigns them at creation + new_user.proxy_settings.wireguard.peer_ips = [] new_user.proxy_settings = await self._prepare_user_proxy_settings(db, all_groups, new_user.proxy_settings) try: db_user = await create_user(db, new_user, all_groups, db_admin) except IntegrityError: await self.raise_error(message="User already exists", code=409, db=db) + except ValueError as exc: # WireGuard subnet exhausted + await self.raise_error(message=str(exc), code=400, db=db) user = await self.update_user(db_user) @@ -813,7 +787,7 @@ async def _prepare_modified_user( ) validated_groups = None - if modified_user.group_ids: + if modified_user.group_ids is not None: validated_groups = await self.validate_all_groups(db, modified_user) if modified_user.next_plan is not None and modified_user.next_plan.user_template_id is not None: @@ -828,16 +802,14 @@ async def _prepare_modified_user( else ProxyTable.model_validate(current_proxy_settings_data) ) - old_peer_ips = set(current_proxy_settings.wireguard.peer_ips or []) - new_peer_ips = set(proxy_settings_to_prepare.wireguard.peer_ips or []) - peer_ips_changed = old_peer_ips != new_peer_ips + # peer IPs are pool-managed: whatever the client sent, keep the stored allocation + proxy_settings_to_prepare.wireguard.peer_ips = list(current_proxy_settings.wireguard.peer_ips or []) prepared_proxy_settings = await self._prepare_user_proxy_settings( db, effective_groups, proxy_settings_to_prepare, exclude_user_id=db_user.id, - skip_peer_ip_validation=not peer_ips_changed, ) if modified_user.proxy_settings is not None or prepared_proxy_settings.dict() != current_proxy_settings_data: modified_user.proxy_settings = prepared_proxy_settings @@ -856,7 +828,10 @@ async def _apply_modified_user( old_status = db_user.status self._apply_explicit_null_hwid_limit(db_user, modified_user) - db_user = await crud_modify_user(db, db_user, modified_user, groups=validated_groups) + try: + db_user = await crud_modify_user(db, db_user, modified_user, groups=validated_groups) + except ValueError as exc: # WireGuard subnet exhausted + await self.raise_error(message=str(exc), code=400, db=db) user = await self.update_user(db_user) logger.info(f'User "{user.username}" with id "{db_user.id}" modified by admin "{admin.username}"') @@ -1944,23 +1919,6 @@ async def bulk_modify_proxy_settings(self, db: AsyncSession, bulk_model: BulkUse return {"detail": f"operation has been successfuly done on {users_count} users"} return users_count - async def bulk_reallocate_wireguard_peer_ips( - self, db: AsyncSession, body: BulkWireGuardPeerIPs, admin: AdminDetails - ) -> WireGuardPeerIPsReallocateResponse: - users = await get_bulk_wireguard_peer_ip_users( - db, - body, - admin_id=get_scope_admin_id(admin, "users", "update"), - ) - - out = await run_bulk_reallocate_wireguard_peer_ips( - db, - users, - dry_run=body.dry_run, - replace_all=body.replace_all, - ) - return WireGuardPeerIPsReallocateResponse(**out) - async def _get_users_sub_update_list( self, db: AsyncSession, db_user: User, offset: int = 0, limit: int = 10 ) -> UserSubscriptionUpdateList: diff --git a/app/routers/system.py b/app/routers/system.py index 52e273077..d987417e7 100644 --- a/app/routers/system.py +++ b/app/routers/system.py @@ -8,11 +8,13 @@ from app.db import AsyncSession, get_db from app.models.admin import AdminDetails from app.models.settings import Telegram +from app.db.crud.wireguard import get_subnet_usage from app.models.system import ( InboundSummary, SystemResourceStats, SystemStats, SystemUsersStats, + WireGuardSubnetUsage, WorkerHealth, WorkersHealth, ) @@ -78,6 +80,15 @@ async def get_inbound_details(_: AdminDetails = Depends(require_permission("syst return await system_operator.get_inbound_details() +@router.get("/wireguard/subnets", response_model=list[WireGuardSubnetUsage]) +async def get_wireguard_subnets( + db: AsyncSession = Depends(get_db), + _: AdminDetails = Depends(require_permission("cores", "read")), +): + """Per-subnet WireGuard address usage: capacity, used/free counts and the first free IPs.""" + return await get_subnet_usage(db) + + async def _measure_worker_health(request_coro) -> WorkerHealth: start = time.monotonic() try: diff --git a/app/routers/user.py b/app/routers/user.py index a89147b27..269aafc80 100644 --- a/app/routers/user.py +++ b/app/routers/user.py @@ -1,6 +1,6 @@ from typing import Annotated -from fastapi import APIRouter, Depends, HTTPException, Request, status +from fastapi import APIRouter, Depends, Request, status from app.db import AsyncSession, get_db from app.models.admin import AdminDetails @@ -19,7 +19,6 @@ BulkUsersApplyTemplate, BulkUsersSelection, BulkUsersSetOwner, - BulkWireGuardPeerIPs, CreateUserFromTemplate, ExpiredUsersQuery, ModifyUserByTemplate, @@ -36,7 +35,6 @@ UsersUsageQuery, UserSubscriptionUpdateChart, UserSubscriptionUpdateList, - WireGuardPeerIPsReallocateResponse, ) from app.operation import OperatorType from app.operation.node import NodeOperation @@ -837,22 +835,3 @@ async def bulk_modify_users_proxy_settings( _: AdminDetails = Depends(require_scope_all("users", "update")), ): return await user_operator.bulk_modify_proxy_settings(db, bulk_model) - - -@router.post( - "s/bulk/wireguard/reallocate-peer-ips", - response_model=WireGuardPeerIPsReallocateResponse, - summary="Bulk reallocate WireGuard peer IPs", - description="Same scoping as other bulk user actions (users, admins, group_ids, optional status filter). non-owner admins only affect their own users.", -) -async def bulk_reallocate_wireguard_peer_ips( - body: BulkWireGuardPeerIPs, - db: AsyncSession = Depends(get_db), - admin: AdminDetails = Depends(require_scope_all("users", "update")), -): - if not body.dry_run and not body.confirm: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="Set confirm=true to apply changes, or use dry_run=true to preview.", - ) - return await user_operator.bulk_reallocate_wireguard_peer_ips(db, body, admin) diff --git a/app/subscription/share.py b/app/subscription/share.py index a49c58cc3..7f8313f8f 100644 --- a/app/subscription/share.py +++ b/app/subscription/share.py @@ -7,6 +7,7 @@ from jdatetime import date as jd from app.core.hosts import host_manager +from app.db.crud.wireguard import pick_peer_ip_for_inbound from app.db.models import UserStatus from app.models.status_emojis import STATUS_EMOJIS from app.models.subscription import SubscriptionInboundData @@ -14,8 +15,6 @@ from app.settings import subscription_settings from app.subscription.client_templates import subscription_client_templates, subscription_xray_templates from app.utils.system import readable_size -from config import wireguard_settings - from . import ( ClashConfiguration, ClashMetaConfiguration, @@ -295,6 +294,10 @@ async def process_host( if user_id is not None: settings["_user_id"] = user_id + # Each WG interface only gets the user's peer IP from its own subnet. + if inbound.protocol == "wireguard": + settings["peer_ips"] = pick_peer_ip_for_inbound(inbound.wireguard_local_address, settings.get("peer_ips") or []) + # Update format variables format_variables.update({"PROTOCOL": inbound.protocol}) format_variables.update({"TRANSPORT": inbound.network}) @@ -436,9 +439,6 @@ def _resolve_host_xray_template_content(inbound: SubscriptionInboundData) -> str return xray_template_overrides.get(template_id) for host_data in hosts: - if host_data.protocol == "wireguard" and not wireguard_settings.enabled: - continue - result = await process_host(host_data, format_variables, user.inbounds, proxy_settings, custom_variables) if not result: continue diff --git a/app/utils/ip_pool.py b/app/utils/ip_pool.py deleted file mode 100644 index 6822a7abf..000000000 --- a/app/utils/ip_pool.py +++ /dev/null @@ -1,335 +0,0 @@ -from __future__ import annotations - -from ipaddress import IPv4Network, IPv6Network, ip_address, ip_network - -from sqlalchemy import cast, func, select -from sqlalchemy.dialects.postgresql import JSONB -from sqlalchemy.ext.asyncio import AsyncSession - -from app.db.crud.user import get_all_wireguard_peer_ips_raw -from app.db.models import User - -from .wireguard_pool import WIREGUARD_GLOBAL_POOL, WIREGUARD_RESERVED - -# Backward-compatible names -GLOBAL_IP_POOL = WIREGUARD_GLOBAL_POOL -SERVER_RESERVED = WIREGUARD_RESERVED - - -def peer_ipv4_network_in_global_pool(net: IPv4Network | IPv6Network) -> bool: - if net.version != 4: - return False - return net.subnet_of(WIREGUARD_GLOBAL_POOL) - - -def peer_ips_outside_global_pool(peer_ips: list[str]) -> bool: - """True if any IPv4 peer CIDR is not contained in the configured global pool.""" - for peer_ip in peer_ips: - try: - candidate = ip_network(peer_ip, strict=False) - except ValueError: - continue - if candidate.version == 4 and not peer_ipv4_network_in_global_pool(candidate): - return True - return False - - -def validate_peer_ips_within_global_pool(peer_ips: list[str]) -> None: - """Require every IPv4 peer network to lie inside WIREGUARD_GLOBAL_POOL (IPv6 entries are not checked).""" - for peer_ip in peer_ips: - try: - candidate = ip_network(peer_ip, strict=False) - except ValueError: - raise ValueError(f"invalid IP/network format: '{peer_ip}'") - if candidate.version == 4 and not peer_ipv4_network_in_global_pool(candidate): - raise ValueError(f"peer IP '{peer_ip}' is outside WIREGUARD_GLOBAL_POOL ({WIREGUARD_GLOBAL_POOL})") - - -async def get_global_used_networks( - db: AsyncSession, - *, - exclude_user_id: int | None = None, -) -> set[IPv4Network | IPv6Network]: - """Get all currently used peer networks from the database. - - Uses dialect-specific optimized queries where available. - Falls back to lightweight column-only query for all databases. - """ - dialect = db.bind.dialect.name - - if dialect == "postgresql": - return await _get_global_used_networks_postgresql(db, exclude_user_id=exclude_user_id) - else: - # MySQL and SQLite: use lightweight query fetching only id and proxy_settings - return await _get_global_used_networks_lightweight(db, exclude_user_id=exclude_user_id) - - -async def _get_global_used_networks_postgresql( - db: AsyncSession, - *, - exclude_user_id: int | None = None, -) -> set[IPv4Network | IPv6Network]: - """PostgreSQL-optimized query using JSONB operators for native JSON extraction.""" - - # Cast the JSON column to JSONB so SQLAlchemy exposes JSONB-specific operators. - # Subscript access on a plain JSON column returns a BinaryExpression that lacks - # JSONB methods (.astext, jsonb_array_elements_text, etc.). - jsonb_col = cast(User.proxy_settings, JSONB) - peer_ips_path = jsonb_col["wireguard"]["peer_ips"] - - # jsonb_array_elements_text unnests the array into individual text rows. - # The IS NOT NULL guard skips users whose wireguard.peer_ips key is absent/null. - stmt = select(func.jsonb_array_elements_text(peer_ips_path).label("peer_ip")).where(peer_ips_path.isnot(None)) - - if exclude_user_id is not None: - stmt = stmt.where(User.id != exclude_user_id) - - result = await db.execute(stmt) - rows = result.all() - - used_ips: set[IPv4Network | IPv6Network] = set() - for row in rows: - peer_ip = row[0] - if peer_ip: - try: - used_ips.add(ip_network(peer_ip, strict=False)) - except ValueError: - continue - - return used_ips - - -async def _get_global_used_networks_lightweight( - db: AsyncSession, - *, - exclude_user_id: int | None = None, -) -> set[IPv4Network | IPv6Network]: - """Lightweight query fetching only id and proxy_settings columns.""" - user_data = await get_all_wireguard_peer_ips_raw(db, exclude_user_id=exclude_user_id) - - used_ips: set[IPv4Network | IPv6Network] = set() - for user_id, data in user_data.items(): - proxy_settings = data.get("proxy_settings") or {} - if isinstance(proxy_settings, str): - import json - - proxy_settings = json.loads(proxy_settings) - - wireguard_settings = proxy_settings.get("wireguard") or {} - peer_ips = wireguard_settings.get("peer_ips") or [] - for peer_ip in peer_ips: - try: - used_ips.add(ip_network(peer_ip, strict=False)) - except ValueError: - continue - - return used_ips - - -def collect_used_peer_networks_from_proxy_settings_rows( - rows: list[dict], - *, - exclude_user_id: int | None = None, -) -> set[IPv4Network | IPv6Network]: - """Sync helper for migrations: build used peer networks from user proxy_settings dicts.""" - used: set[IPv4Network | IPv6Network] = set() - for row in rows: - uid = row.get("id") - if exclude_user_id is not None and uid == exclude_user_id: - continue - ps = row.get("proxy_settings") or {} - if isinstance(ps, str): - import json - - ps = json.loads(ps) - wg = ps.get("wireguard") or {} - for peer_ip in wg.get("peer_ips") or []: - try: - used.add(ip_network(peer_ip, strict=False)) - except ValueError: - continue - return used - - -def allocate_one_from_pool_sync(used_networks: set[IPv4Network | IPv6Network]) -> str | None: - """Pick first free IPv4 /32 in the global pool (sync; for migrations). - - Uses bitset-style integer lookup for O(1) per candidate instead of O(U) per candidate. - """ - pool = WIREGUARD_GLOBAL_POOL - start = int(pool.network_address) - end = int(pool.broadcast_address) - - # Pre-build set of all blocked integer addresses (reserved + used) for O(1) lookup - blocked: set[int] = set() - - # Add all reserved addresses - for net in WIREGUARD_RESERVED: - for addr in net: - blocked.add(int(addr)) - - # Add all used addresses (only IPv4) - for net in used_networks: - if net.version == 4: - for addr in net: - blocked.add(int(addr)) - - # Scan for first free address, skipping network and broadcast - for raw_candidate in range(start + 1, end): - if raw_candidate not in blocked: - return f"{ip_address(raw_candidate)}/32" - - return None - - -class WireGuardPeerIPAllocator: - """Stateful IPv4 /32 allocator for bulk operations.""" - - def __init__(self, used_networks: set[IPv4Network | IPv6Network]): - self._pool = WIREGUARD_GLOBAL_POOL - self._end = int(self._pool.broadcast_address) - self._next = int(self._pool.network_address) + 1 - self._blocked: set[int] = set() - - for net in WIREGUARD_RESERVED: - for addr in net: - self._blocked.add(int(addr)) - - for net in used_networks: - if net.version == 4: - for addr in net: - self._blocked.add(int(addr)) - - def allocate(self) -> str | None: - while self._next < self._end: - raw_candidate = self._next - self._next += 1 - if raw_candidate in self._blocked: - continue - self._blocked.add(raw_candidate) - return f"{ip_address(raw_candidate)}/32" - return None - - def is_reserved(self, peer_ip: str) -> bool: - """Whether the given IP/network falls inside the WireGuard reserved ranges.""" - try: - candidate = ip_network(peer_ip, strict=False) - except ValueError: - return False - candidate_ip = ip_address(candidate.network_address) - return any(candidate_ip in net for net in WIREGUARD_RESERVED) - - def conflicts(self, peer_ip: str) -> bool: - """Whether the given IP/network overlaps already-blocked addresses (used or reserved).""" - try: - candidate = ip_network(peer_ip, strict=False) - except ValueError: - return False - if candidate.version != 4: - return False - for addr in candidate: - if int(addr) in self._blocked: - return True - return False - - def reserve(self, peer_ip: str) -> None: - """Mark every address in the given IPv4 network as blocked so future allocations skip it.""" - try: - candidate = ip_network(peer_ip, strict=False) - except ValueError: - return - if candidate.version != 4: - return - for addr in candidate: - self._blocked.add(int(addr)) - - -async def allocate_from_global_pool( - db: AsyncSession, - *, - exclude_user_id: int | None = None, -) -> str | None: - used_ips = await get_global_used_networks(db, exclude_user_id=exclude_user_id) - return allocate_one_from_pool_sync(used_ips) - - -async def allocate_and_validate_peer_ips( - db: AsyncSession, - peer_ips: list[str], - *, - exclude_user_id: int | None = None, -) -> list[str]: - """ - Allocate peer IPs from global pool if not provided, or validate provided peer IPs. - - Fetches the used networks from database once and reuses for both allocation and validation. - This avoids double scanning the entire user table. - - Args: - db: Database session - peer_ips: List of peer IPs to use. If empty, allocates from global pool. - exclude_user_id: User ID to exclude from used networks check (for updates) - - Returns: - List of allocated or validated peer IPs - - Raises: - ValueError: If peer IPs are invalid, in use, reserved, or if allocation fails - """ - # Single DB fetch for both allocation and validation - used_networks = await get_global_used_networks(db, exclude_user_id=exclude_user_id) - - if not peer_ips: - # Allocation path: find first free IP from pool - candidate = allocate_one_from_pool_sync(used_networks) - if candidate is None: - raise ValueError("unable to allocate wireguard peer IP") - return [candidate] - - # Validation path: check provided IPs against used networks - validate_peer_ips_within_global_pool(peer_ips) - - for peer_ip in peer_ips: - try: - candidate = ip_network(peer_ip, strict=False) - except ValueError: - raise ValueError(f"invalid IP/network format: '{peer_ip}'") - - if any(candidate.overlaps(used_ip) for used_ip in used_networks): - raise ValueError(f"peer IP/network '{peer_ip}' is already in use by an existing user's peer network") - - candidate_ip = ip_address(candidate.network_address) - if any(candidate_ip in net for net in WIREGUARD_RESERVED): - raise ValueError(f"peer IP '{peer_ip}' is reserved") - - return peer_ips - - -async def validate_peer_ips_globally( - db: AsyncSession, - peer_ips: list[str], - *, - exclude_user_id: int | None = None, -) -> None: - """ - Validate that supplied peer IPs/networks don't overlap with existing user's peer networks. - - Raises ValueError if any supplied IP/network overlaps with an existing user's peer networks. - - Note: For new code, consider using allocate_and_validate_peer_ips which is more efficient - when allocation and validation need to happen together. - """ - used_networks = await get_global_used_networks(db, exclude_user_id=exclude_user_id) - - for peer_ip in peer_ips: - try: - candidate = ip_network(peer_ip, strict=False) - except ValueError: - raise ValueError(f"invalid IP/network format: '{peer_ip}'") - - if any(candidate.overlaps(used_ip) for used_ip in used_networks): - raise ValueError(f"peer IP/network '{peer_ip}' is already in use by an existing user's peer network") - - candidate_ip = ip_address(candidate.network_address) - if any(candidate_ip in net for net in WIREGUARD_RESERVED): - raise ValueError(f"peer IP '{peer_ip}' is reserved") diff --git a/app/utils/reality_scan.py b/app/utils/reality_scan.py index 44d52c51c..8eefd7c2e 100644 --- a/app/utils/reality_scan.py +++ b/app/utils/reality_scan.py @@ -24,7 +24,9 @@ DNS_TIMEOUT = 5.0 MAX_CONCURRENT_SCANS = 4 -_scan_semaphores: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Semaphore]" = weakref.WeakKeyDictionary() +_scan_semaphores: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Semaphore]" = ( + weakref.WeakKeyDictionary() +) _scan_executor: "ThreadPoolExecutor | None" = None @@ -152,7 +154,9 @@ def _select_public_ip(host: str, infos: list) -> str: saw_blocked = True if saw_blocked: - raise RealityScanError("Target resolves only to private or reserved addresses; only public hosts can be scanned.") + raise RealityScanError( + "Target resolves only to private or reserved addresses; only public hosts can be scanned." + ) raise RealityScanError(f"Could not resolve host to a usable address: {host}") @@ -177,7 +181,7 @@ async def _resolve_public_ip_async(host: str, timeout: float) -> str: loop = asyncio.get_running_loop() try: infos = await asyncio.wait_for(loop.getaddrinfo(host, None, type=socket.SOCK_STREAM), timeout=timeout) - except (asyncio.TimeoutError, TimeoutError): + except asyncio.TimeoutError, TimeoutError: raise RealityScanError(f"DNS lookup for {host} timed out.") except socket.gaierror: raise RealityScanError(f"Could not resolve host: {host}") @@ -221,7 +225,9 @@ def _parse_certificate(der: bytes | None) -> dict: return out out["cert_subject"] = _name_common_name(cert.subject) or (cert.subject.rfc4514_string() or None) - out["cert_issuer"] = _name_organization(cert.issuer) or _name_common_name(cert.issuer) or (cert.issuer.rfc4514_string() or None) + out["cert_issuer"] = ( + _name_organization(cert.issuer) or _name_common_name(cert.issuer) or (cert.issuer.rfc4514_string() or None) + ) try: out["not_after"] = cert.not_valid_after_utc.isoformat() @@ -310,7 +316,9 @@ def _drive_handshake(tls: ssl.SSLSocket, deadline: float) -> None: raise TimeoutError("TLS handshake timed out") -def _tls_wrap_with_deadline(ctx: ssl.SSLContext, ip: str, port: int, server_hostname: str | None, timeout: float) -> tuple[ssl.SSLSocket, float]: +def _tls_wrap_with_deadline( + ctx: ssl.SSLContext, ip: str, port: int, server_hostname: str | None, timeout: float +) -> tuple[ssl.SSLSocket, float]: started = time.monotonic() deadline = started + timeout sock = socket.create_connection((ip, port), timeout=timeout) @@ -365,7 +373,9 @@ def _tls_probe(ip: str, port: int, sni: str | None, timeout: float) -> dict: alpn: str | None = None der: bytes | None = None - def _handshake(ctx: ssl.SSLContext, server_hostname: str | None) -> tuple[str | None, str | None, bytes | None, float]: + def _handshake( + ctx: ssl.SSLContext, server_hostname: str | None + ) -> tuple[str | None, str | None, bytes | None, float]: tls, latency = _tls_wrap_with_deadline(ctx, ip, port, server_hostname, timeout) with tls: return tls.version(), tls.selected_alpn_protocol(), tls.getpeercert(binary_form=True), latency @@ -401,7 +411,7 @@ def _handshake(ctx: ssl.SSLContext, server_hostname: str | None) -> tuple[str | result["reason"] = f"Certificate did not validate: {getattr(exc, 'verify_message', None) or exc}" except (ssl.SSLError, socket.timeout, TimeoutError, OSError, UnicodeError) as exc: result["reason"] = f"Certificate re-validation failed: {exc}" - except (socket.timeout, TimeoutError): + except socket.timeout, TimeoutError: result["reason"] = "Connection timed out." return result except ssl.SSLError as exc: @@ -503,7 +513,7 @@ def _recv_exact(sock: socket.socket, count: int, deadline: float) -> bytes | Non sock.settimeout(remaining) try: chunk = sock.recv(count - len(buf)) - except (socket.timeout, TimeoutError): + except socket.timeout, TimeoutError: return None if not chunk: return None @@ -677,7 +687,9 @@ def _scan_sync(host: str, ip: str, port: int, sni: str | None, timeout: float) - result["feasible"] = base_ok and result["x25519"] is True if base_ok and result["x25519"] is not True and not result["reason"]: if result["x25519"] is False: - result["reason"] = f"Key exchange is {result['curve'] or 'not X25519'}; REALITY needs X25519 or X25519MLKEM768." + result["reason"] = ( + f"Key exchange is {result['curve'] or 'not X25519'}; REALITY needs X25519 or X25519MLKEM768." + ) else: result["reason"] = "Could not confirm an X25519 key exchange." return result diff --git a/app/utils/wireguard.py b/app/utils/wireguard.py index 263876037..c6d9ed1e6 100644 --- a/app/utils/wireguard.py +++ b/app/utils/wireguard.py @@ -1,81 +1,26 @@ from __future__ import annotations -import json -from ipaddress import ip_network from typing import Iterable from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from app.core.manager import core_manager -from app.db.crud.user import get_all_wireguard_peer_ips_raw -from app.db.models import CoreConfig, CoreType, User +from app.db.crud.wireguard import get_wg_cores, tags_from_groups, wg_core_tags +from app.db.models import User from app.models.proxy import ProxyTable -from app.node.sync import sync_users from app.utils.crypto import generate_wireguard_keypair, get_wireguard_public_key -from app.utils.ip_pool import ( - WireGuardPeerIPAllocator, - allocate_and_validate_peer_ips, - collect_used_peer_networks_from_proxy_settings_rows, - get_global_used_networks, - peer_ips_outside_global_pool, - validate_peer_ips_within_global_pool, -) -from config import wireguard_settings -def _normalized_peer_networks(peer_ips: Iterable[str]) -> list[str]: - networks: list[str] = [] - for peer_ip in peer_ips: - try: - networks.append(str(ip_network(peer_ip, strict=False))) - except ValueError: - continue - return networks - - -def _peer_network_owners_from_rows(rows: Iterable[dict]) -> dict[str, set[int]]: - owners: dict[str, set[int]] = {} - for row in rows: - uid = row.get("id") - if uid is None: - continue - proxy_settings = row.get("proxy_settings") or {} - if isinstance(proxy_settings, str): - proxy_settings = json.loads(proxy_settings) - wireguard_settings = proxy_settings.get("wireguard") or {} - for network in _normalized_peer_networks(wireguard_settings.get("peer_ips") or []): - owners.setdefault(network, set()).add(uid) - return owners - - -def _wireguard_public_key_from_proxy_settings(proxy_settings) -> str | None: - if not proxy_settings: - return None - if isinstance(proxy_settings, str): - proxy_settings = json.loads(proxy_settings) - wireguard_settings = proxy_settings.get("wireguard") or {} - public_key = wireguard_settings.get("public_key") - if not public_key: - private_key = wireguard_settings.get("private_key") - if private_key: - try: - public_key = get_wireguard_public_key(private_key) - except ValueError: - return None - return str(public_key).strip() if public_key else None - - -def _wireguard_public_key_owners_from_rows(rows: Iterable[dict]) -> dict[str, set[int]]: - owners: dict[str, set[int]] = {} - for row in rows: - uid = row.get("id") - if uid is None: - continue - public_key = _wireguard_public_key_from_proxy_settings(row.get("proxy_settings")) - if public_key: - owners.setdefault(public_key, set()).add(uid) - return owners +async def wireguard_public_key_in_use( + db: AsyncSession, + public_key: str, + *, + exclude_user_id: int | None = None, +) -> bool: + stmt = select(User.id).where(User.proxy_settings["wireguard"]["public_key"].as_string() == public_key).limit(1) + if exclude_user_id is not None: + stmt = stmt.where(User.id != exclude_user_id) + return (await db.execute(stmt)).first() is not None async def ensure_unique_wireguard_public_key( @@ -87,178 +32,27 @@ async def ensure_unique_wireguard_public_key( public_key = proxy_settings.wireguard.public_key if not public_key: return - - rows = [ - {"id": user_id, **data} - for user_id, data in (await get_all_wireguard_peer_ips_raw(db, exclude_user_id=exclude_user_id)).items() - ] - owners = _wireguard_public_key_owners_from_rows(rows) - if public_key in owners: + if await wireguard_public_key_in_use(db, public_key, exclude_user_id=exclude_user_id): raise ValueError("wireguard public_key is already assigned to another user") -async def get_wireguard_tags(tags: Iterable[str]) -> list[str]: - """Get WireGuard inbound tags from a list of tags (requires core manager; unused by global pool path).""" - inbounds_by_tag = await core_manager.get_inbounds_by_tag() - wireguard_tags: list[str] = [] - seen: set[str] = set() - for tag in tags: - if tag in seen: - continue - if inbounds_by_tag.get(tag, {}).get("protocol") == "wireguard": - seen.add(tag) - wireguard_tags.append(tag) - return wireguard_tags - - -async def get_wireguard_tags_from_groups(groups: Iterable) -> list[str]: - """Get WireGuard inbound tags from a list of groups.""" - tags: list[str] = [] - for group in groups: - if getattr(group, "is_disabled", False): - continue - if hasattr(group, "awaitable_attrs"): - await group.awaitable_attrs.inbounds - tags.extend(inbound.tag for inbound in group.inbounds) - return await get_wireguard_tags(tags) - - -async def get_wireguard_inbound_tags_from_db(db: AsyncSession) -> set[str]: - """Inbound tags (interface names) for all WireGuard cores.""" - rows = (await db.execute(select(CoreConfig).where(CoreConfig.type == CoreType.wg))).scalars().all() - tags: set[str] = set() - for row in rows: - cfg = row.config or {} - if isinstance(cfg, str): - cfg = json.loads(cfg) - name = (cfg or {}).get("interface_name") - if name: - tags.add(str(name).strip()) - return tags - - -async def user_in_wireguard_group(user: User, wg_tags: set[str]) -> bool: - groups = user.__dict__.get("groups") - if groups is None: - groups = await user.awaitable_attrs.groups - - for group in groups: - if group.is_disabled: - continue - inbounds = group.__dict__.get("inbounds") - if inbounds is None: - inbounds = await group.awaitable_attrs.inbounds - for inbound in inbounds: - if inbound.tag in wg_tags: - return True - return False - - -async def prepare_wireguard_proxy_settings( - db: AsyncSession, - proxy_settings: ProxyTable, - groups: Iterable, - *, - exclude_user_id: int | None = None, -) -> ProxyTable: - """Prepare WireGuard proxy settings with key generation and global pool IP allocation.""" - wireguard_tags = await get_wireguard_tags_from_groups(groups) - if not wireguard_tags: - return proxy_settings - - if not wireguard_settings.enabled: - return proxy_settings - - await ensure_unique_wireguard_public_key(db, proxy_settings, exclude_user_id=exclude_user_id) - - if proxy_settings.wireguard.public_key and not proxy_settings.wireguard.private_key: - raise ValueError("wireguard private_key is required when user is assigned to a WireGuard interface") - - if not proxy_settings.wireguard.private_key: - private_key, public_key = generate_wireguard_keypair() - proxy_settings.wireguard.private_key = private_key - proxy_settings.wireguard.public_key = public_key - elif not proxy_settings.wireguard.public_key: - proxy_settings.wireguard.public_key = get_wireguard_public_key(proxy_settings.wireguard.private_key) - - if not wireguard_settings.enabled: - return proxy_settings - - peer_ips = list(proxy_settings.wireguard.peer_ips or []) - - # Use merged allocate+validate function to avoid double DB scan - peer_ips = await allocate_and_validate_peer_ips(db, peer_ips, exclude_user_id=exclude_user_id) - - proxy_settings.wireguard.peer_ips = peer_ips - return proxy_settings - - -async def build_wireguard_peer_ip_allocator( - db: AsyncSession, - *, - exclude_user_id: int | None = None, -) -> "WireGuardPeerIPAllocator": - """Build a stateful peer-IP allocator pre-loaded with all currently used peer networks. - - Used by bulk-creation flows where many users need IPs allocated within a single - transaction; reusing one allocator avoids the duplicate-allocation bug that occurs - when each user independently re-reads the database before any of the new users have - been committed. - """ - used_networks = await get_global_used_networks(db, exclude_user_id=exclude_user_id) - return WireGuardPeerIPAllocator(used_networks) +async def user_has_wireguard_access(db: AsyncSession, groups: Iterable) -> bool: + wg_tags = wg_core_tags(await get_wg_cores(db)) + return bool(wg_tags and wg_tags & await tags_from_groups(groups)) -def prepare_wireguard_proxy_settings_with_allocator( - proxy_settings: ProxyTable, - allocator: "WireGuardPeerIPAllocator", -) -> ProxyTable: - """Prepare WireGuard settings for a single user against a shared allocator. - - Caller is responsible for confirming the user belongs to a WireGuard group and that - `wireguard_settings.enabled` is true. Validates any user-supplied peer_ips against - the allocator's current blocked set, then either reserves them or allocates a fresh - IP. The allocator is mutated to reflect the new reservation. - """ - prepare_wireguard_keys_for_member(proxy_settings) - - peer_ips = list(proxy_settings.wireguard.peer_ips or []) - - if peer_ips: - validate_peer_ips_within_global_pool(peer_ips) - for peer_ip in peer_ips: - if allocator.is_reserved(peer_ip): - raise ValueError(f"peer IP '{peer_ip}' is reserved") - if allocator.conflicts(peer_ip): - raise ValueError(f"peer IP/network '{peer_ip}' is already in use by an existing user's peer network") - allocator.reserve(peer_ip) - proxy_settings.wireguard.peer_ips = peer_ips - return proxy_settings - - candidate = allocator.allocate() - if candidate is None: - raise ValueError("unable to allocate wireguard peer IP") - proxy_settings.wireguard.peer_ips = [candidate] - return proxy_settings - - -async def prepare_wireguard_keys_only( +async def prepare_wireguard_keys( db: AsyncSession, proxy_settings: ProxyTable, groups: Iterable, *, exclude_user_id: int | None = None, ) -> ProxyTable: - """Generate WireGuard keys without validation or IP allocation. + """Ensure WG keys for a user assigned to a WireGuard interface. - Used when peer_ips haven't changed during user modification. - Avoids expensive database scans for unchanged peer networks. + Peer IPs are managed by the subnet pool (app/db/crud/wireguard.py), never here. """ - wireguard_tags = await get_wireguard_tags_from_groups(groups) - if not wireguard_tags: - return proxy_settings - - if not wireguard_settings.enabled: + if not await user_has_wireguard_access(db, groups): return proxy_settings await ensure_unique_wireguard_public_key(db, proxy_settings, exclude_user_id=exclude_user_id) @@ -274,171 +68,3 @@ async def prepare_wireguard_keys_only( proxy_settings.wireguard.public_key = get_wireguard_public_key(proxy_settings.wireguard.private_key) return proxy_settings - - -def prepare_wireguard_keys_for_member(proxy_settings: ProxyTable) -> ProxyTable: - """Generate WireGuard keys for a user already known to belong to a WireGuard group.""" - if proxy_settings.wireguard.public_key and not proxy_settings.wireguard.private_key: - raise ValueError("wireguard private_key is required when user is assigned to a WireGuard interface") - - if not proxy_settings.wireguard.private_key: - private_key, public_key = generate_wireguard_keypair() - proxy_settings.wireguard.private_key = private_key - proxy_settings.wireguard.public_key = public_key - elif not proxy_settings.wireguard.public_key: - proxy_settings.wireguard.public_key = get_wireguard_public_key(proxy_settings.wireguard.private_key) - - return proxy_settings - - -async def bulk_reallocate_wireguard_peer_ips( - db: AsyncSession, - target_users: Iterable[User], - *, - dry_run: bool, - replace_all: bool, -) -> dict: - """ - Re-seat peer_ips for users in WireGuard groups when IPs are outside the current global pool - or duplicated, or when replace_all is True. Preserves WireGuard keys. Syncs each updated user to nodes. - - ``target_users`` should be the users allowed by bulk scope (group/admin/user filters). - """ - if not wireguard_settings.enabled: - return { - "wireguard_inbound_tags": 0, - "candidates": 0, - "updated": 0, - "dry_run": dry_run, - "sample_usernames": [], - "affected_users": 0, - } - - wg_tags = await get_wireguard_inbound_tags_from_db(db) - if not wg_tags: - return { - "wireguard_inbound_tags": 0, - "candidates": 0, - "updated": 0, - "dry_run": dry_run, - "sample_usernames": [], - "affected_users": 0, - } - - users = list(target_users) - eligible_users: list[tuple[User, list[str]]] = [] - to_touch: list[User] = [] - reallocate_peer_ip_user_ids: set[int] = set() - sample: list[str] = [] - - for user in users: - if not await user_in_wireguard_group(user, wg_tags): - continue - proxy_settings = ProxyTable.model_validate(user.proxy_settings or {}) - peer_ips = list(proxy_settings.wireguard.peer_ips or []) - eligible_users.append((user, peer_ips)) - - eligible_user_ids = {user.id for user, _ in eligible_users} - all_peer_ip_rows = [{"id": user_id, **data} for user_id, data in (await get_all_wireguard_peer_ips_raw(db)).items()] - peer_network_owners = _peer_network_owners_from_rows(all_peer_ip_rows) - public_key_owners = _wireguard_public_key_owners_from_rows(all_peer_ip_rows) - duplicated_user_ids: set[int] = set() - for owner_ids in peer_network_owners.values(): - if len(owner_ids) > 1: - target_owner_ids = owner_ids & eligible_user_ids - if not target_owner_ids: - continue - if owner_ids <= eligible_user_ids: - duplicated_user_ids.update(sorted(owner_ids)[1:]) - else: - duplicated_user_ids.update(target_owner_ids) - - duplicated_public_key_user_ids: set[int] = set() - for owner_ids in public_key_owners.values(): - if len(owner_ids) > 1: - target_owner_ids = owner_ids & eligible_user_ids - if not target_owner_ids: - continue - if owner_ids <= eligible_user_ids: - duplicated_public_key_user_ids.update(sorted(owner_ids)[1:]) - else: - duplicated_public_key_user_ids.update(target_owner_ids) - - for user, peer_ips in eligible_users: - needs_peer_ip = False - needs_rekey = user.id in duplicated_public_key_user_ids - if replace_all: - needs_peer_ip = True - elif not peer_ips: - needs_peer_ip = True - elif peer_ips_outside_global_pool(peer_ips): - needs_peer_ip = True - elif user.id in duplicated_user_ids: - needs_peer_ip = True - - if not needs_peer_ip and not needs_rekey: - continue - if needs_peer_ip: - reallocate_peer_ip_user_ids.add(user.id) - to_touch.append(user) - if len(sample) < 20: - sample.append(user.username) - - if dry_run: - n = len(to_touch) - return { - "wireguard_inbound_tags": len(wg_tags), - "candidates": n, - "updated": 0, - "dry_run": True, - "sample_usernames": sample, - "affected_users": n, - } - - if not to_touch: - return { - "wireguard_inbound_tags": len(wg_tags), - "candidates": 0, - "updated": 0, - "dry_run": False, - "sample_usernames": sample, - "affected_users": 0, - } - - peer_ip_rows = [row for row in all_peer_ip_rows if row.get("id") not in reallocate_peer_ip_user_ids] - used_networks = collect_used_peer_networks_from_proxy_settings_rows(peer_ip_rows) - - updated = 0 - allocator = WireGuardPeerIPAllocator(used_networks) - updated_users: list[User] = [] - for user in to_touch: - proxy_settings = ProxyTable.model_validate(user.proxy_settings or {}) - try: - prepared = prepare_wireguard_keys_for_member(proxy_settings) - except ValueError: - continue - if user.id in duplicated_public_key_user_ids: - private_key, public_key = generate_wireguard_keypair() - prepared.wireguard.private_key = private_key - prepared.wireguard.public_key = public_key - if user.id in reallocate_peer_ip_user_ids: - peer_ip = allocator.allocate() - if peer_ip is None: - continue - prepared.wireguard.peer_ips = [peer_ip] - user.proxy_settings = prepared.dict() - updated_users.append(user) - updated += 1 - - if updated_users: - await db.commit() - await sync_users(updated_users) - - return { - "wireguard_inbound_tags": len(wg_tags), - "candidates": len(to_touch), - "updated": updated, - "dry_run": False, - "sample_usernames": sample, - "affected_users": updated, - } diff --git a/app/utils/wireguard_pool.py b/app/utils/wireguard_pool.py deleted file mode 100644 index 22df7a96b..000000000 --- a/app/utils/wireguard_pool.py +++ /dev/null @@ -1,36 +0,0 @@ -from __future__ import annotations - -from ipaddress import IPv4Network, ip_network - -from config import wireguard_settings - - -def _parse_global_pool(raw: str) -> IPv4Network: - try: - n = ip_network(raw.strip(), strict=False) - except ValueError as exc: - raise ValueError(f"Invalid WIREGUARD_GLOBAL_POOL: {raw!r}") from exc - if n.version != 4: - raise ValueError("WIREGUARD_GLOBAL_POOL must be an IPv4 CIDR (e.g. 10.0.0.0/8)") - return n - - -def _parse_reserved_networks(raw: str) -> frozenset[IPv4Network]: - """Comma-separated IPv4 CIDR subnets whose addresses are never auto-assigned from the pool.""" - out: set[IPv4Network] = set() - for part in raw.split(","): - part = part.strip() - if not part: - continue - try: - net = ip_network(part, strict=False) - except ValueError as exc: - raise ValueError(f"Invalid CIDR in WIREGUARD_RESERVED: {part!r}") from exc - if net.version != 4: - raise ValueError(f"WIREGUARD_RESERVED must be IPv4 CIDR subnets (e.g. 10.0.0.0/31): {part!r}") - out.add(net) - return frozenset(out) - - -WIREGUARD_GLOBAL_POOL: IPv4Network = _parse_global_pool(wireguard_settings.global_pool) -WIREGUARD_RESERVED: frozenset[IPv4Network] = _parse_reserved_networks(wireguard_settings.reserved) diff --git a/config.py b/config.py index 5400c9206..95164b813 100644 --- a/config.py +++ b/config.py @@ -88,9 +88,7 @@ class NatsSettings(EnvSettings): node_log_subject: str = Field(default="pasarguard.node.logs", validation_alias="NATS_NODE_LOG_SUBJECT") node_rpc_timeout: float = Field(default=30.0, validation_alias="NATS_NODE_RPC_TIMEOUT") scheduler_rpc_timeout: float = Field(default=5.0, validation_alias="NATS_SCHEDULER_RPC_TIMEOUT") - node_command_max_payload_bytes: int = Field( - default=900000, validation_alias="NATS_NODE_COMMAND_MAX_PAYLOAD_BYTES" - ) + node_command_max_payload_bytes: int = Field(default=900000, validation_alias="NATS_NODE_COMMAND_MAX_PAYLOAD_BYTES") node_update_users_batch_size: int = Field(default=100, validation_alias="NATS_NODE_UPDATE_USERS_BATCH_SIZE") core_pubsub_channel: str = Field(default="core_hosts_updates", validation_alias="CORE_PUBSUB_CHANNEL") host_pubsub_channel: str = Field(default="host_manager_updates", validation_alias="HOST_PUBSUB_CHANNEL") @@ -203,12 +201,6 @@ class FeatureSettings(EnvSettings): stop_nodes_on_shutdown: bool = Field(default=True, validation_alias="STOP_NODES_ON_SHUTDOWN") -class WireGuardSettings(EnvSettings): - enabled: bool = Field(default=True, validation_alias="WIREGUARD_ENABLED") - global_pool: str = Field(default="10.0.0.0/8", validation_alias="WIREGUARD_GLOBAL_POOL") - reserved: str = Field(default="10.0.0.0/31", validation_alias="WIREGUARD_RESERVED") - - database_settings = DatabaseSettings() server_settings = ServerSettings() dashboard_settings = DashboardSettings() @@ -224,7 +216,6 @@ class WireGuardSettings(EnvSettings): usage_settings = UsageSettings() job_settings = JobSettings() feature_settings = FeatureSettings() -wireguard_settings = WireGuardSettings() if not database_settings.is_postgresql: usage_settings.enable_recording_nodes_stats = False diff --git a/dashboard/public/statics/locales/en.json b/dashboard/public/statics/locales/en.json index 8806cfb85..b0bd8ccb2 100644 --- a/dashboard/public/statics/locales/en.json +++ b/dashboard/public/statics/locales/en.json @@ -118,6 +118,18 @@ "2h": "2 hours" } }, + "wireguard": { + "title": "WireGuard", + "description": "Peer IP pools for WireGuard cores — usage, free addresses, and interface tags", + "empty": "No WireGuard subnets yet. Create a WireGuard core with an interface address to start allocating peer IPs.", + "used": "Used", + "free": "Free", + "capacity": "Capacity", + "usage": "Usage", + "freeIps": "Free IPs", + "noTags": "No interface tags", + "refresh": "Refresh" + }, "reconnectinfo": "Refresh all nodes connections to resolve connectivity issues", "reconnectAll": "Reconnect All Nodes", "reconnectingAll": "Reconnecting...", @@ -1784,9 +1796,7 @@ "proxySettings.hysteriaAuth": "Hysteria Auth", "proxySettings.wireguardPrivateKey": "WireGuard Private key", "proxySettings.wireguardPublicKey": "WireGuard Public key", - "proxySettings.wireguardPeerIps": "WireGuard Peer IPs", "proxySettings.generateWireGuardKeyPair": "WireGuard keypair", - "proxySettings.peerIpsHint": "Leave empty to auto-assign from the global WireGuard peer pool. For manual entries, enter one CIDR per line, and keep each value within that pool.", "proxySettings.wireguardGenerated": "WireGuard keypair generated", "proxySettings.desc": "Configure protocol-specific settings for this user.", "selectNode": "Select Node", @@ -3273,11 +3283,6 @@ "createUsersDesc": "Create multiple users at once from a template", "groups": "Groups", "proxySettings": "Proxy Settings", - "wireguardPeerIps": "WireGuard IPs", - "wireguardPeerIpsDesc": "Reallocate peer IPs from the global pool", - "replaceAllPeerIps": "Replace all IPs", - "replaceAllPeerIpsHint": "When enabled, every affected user gets a new peer IP from the pool. When disabled, only invalid or missing peer IPs are updated.", - "replaceInvalidPeerIpsOnly": "Invalid or missing IPs only", "preview": "Preview", "previewing": "Previewing…", "previewTitle": "Preview", diff --git a/dashboard/public/statics/locales/fa.json b/dashboard/public/statics/locales/fa.json index efbfbbbab..9ee0b7eed 100644 --- a/dashboard/public/statics/locales/fa.json +++ b/dashboard/public/statics/locales/fa.json @@ -1621,9 +1621,7 @@ "userDialog.proxySettings.hysteriaAuth": "احراز هویت Hysteria", "userDialog.proxySettings.wireguardPrivateKey": "کلید خصوصی وایرگارد", "userDialog.proxySettings.wireguardPublicKey": "کلید عمومی وایرگارد", - "userDialog.proxySettings.wireguardPeerIps": "IPهای همتای وایرگارد", "userDialog.proxySettings.generateWireGuardKeyPair": "جفت‌کلید وایرگارد", - "userDialog.proxySettings.peerIpsHint": "برای تخصیص خودکار از استخر سراسری آی‌پی وایرگارد خالی بگذارید. در ورود دستی، هر CIDR را در یک خط جدا وارد کنید و همه مقادیر باید داخل همان محدوده باشند.", "userDialog.proxySettings.wireguardGenerated": "جفت‌کلید وایرگارد تولید شد", "userDialog.proxySettings.desc": "تنظیمات اختصاصی پروتکل برای این کاربر را پیکربندی کنید.", "userDialog.expireDate": "تاریخ انقضا", @@ -1979,6 +1977,18 @@ "2h": "۲ ساعت" } }, + "wireguard": { + "title": "WireGuard", + "description": "استخر IP همتا برای هسته‌های WireGuard — مصرف، آدرس‌های آزاد و تگ اینترفیس", + "empty": "هنوز زیرشبکه WireGuard وجود ندارد. یک هسته WireGuard با آدرس اینترفیس بسازید تا تخصیص IP شروع شود.", + "used": "مصرف‌شده", + "free": "آزاد", + "capacity": "ظرفیت", + "usage": "مصرف", + "freeIps": "IPهای آزاد", + "noTags": "بدون تگ اینترفیس", + "refresh": "بازنشانی" + }, "reconnectinfo": "به‌روزرسانی همه اتصالات گره‌ها برای رفع مشکلات اتصال", "reconnectAll": "اتصال مجدد همه گره‌ها", "reconnectingAll": "در حال اتصال مجدد...", @@ -3214,11 +3224,6 @@ "createUsersDesc": "ایجاد چندین کاربر به صورت یکجا از یک الگو", "groups": "گروه‌ها", "proxySettings": "تنظیمات پروکسی", - "wireguardPeerIps": "آی‌پی‌های وایرگارد", - "wireguardPeerIpsDesc": "تخصیص مجدد آی‌پی همتای از محدودهٔ سراسری", - "replaceAllPeerIps": "جایگزینی همهٔ آی‌پی‌ها", - "replaceAllPeerIpsHint": "در صورت فعال بودن، هر کاربر تحت تأثیر آی‌پی همتای جدید از استخر دریافت می‌کند. در غیر این صورت فقط آی‌پی‌های نامعتبر یا ناموجود به‌روزرسانی می‌شوند.", - "replaceInvalidPeerIpsOnly": "فقط نامعتبر یا ناموجود", "preview": "پیش‌نمایش", "previewing": "در حال پیش‌نمایش…", "previewTitle": "پیش‌نمایش", diff --git a/dashboard/public/statics/locales/ru.json b/dashboard/public/statics/locales/ru.json index 06de0f1e3..b763d99da 100644 --- a/dashboard/public/statics/locales/ru.json +++ b/dashboard/public/statics/locales/ru.json @@ -145,6 +145,18 @@ "2h": "2 часа" } }, + "wireguard": { + "title": "WireGuard", + "description": "Пулы peer IP для ядер WireGuard — использование, свободные адреса и теги интерфейсов", + "empty": "Подсетей WireGuard пока нет. Создайте ядро WireGuard с адресом интерфейса, чтобы начать выдачу peer IP.", + "used": "Занято", + "free": "Свободно", + "capacity": "Ёмкость", + "usage": "Использование", + "freeIps": "Свободные IP", + "noTags": "Нет тегов интерфейса", + "refresh": "Обновить" + }, "reconnectinfo": "Обновить все соединения узлов для устранения проблем с подключением", "reconnectAll": "Переподключить все узлы", "reconnectingAll": "Переподключение...", @@ -3013,9 +3025,7 @@ "proxySettings.hysteriaAuth": "Аутентификация Hysteria", "proxySettings.wireguardPrivateKey": "Приватный ключ WireGuard", "proxySettings.wireguardPublicKey": "Публичный ключ WireGuard", - "proxySettings.wireguardPeerIps": "Peer IP-адреса WireGuard", "proxySettings.generateWireGuardKeyPair": "Пара ключей WireGuard", - "proxySettings.peerIpsHint": "Оставьте пустым для автоназначения из глобального пула peer-адресов WireGuard. При ручном вводе указывайте по одному CIDR в строке, и каждое значение должно входить в этот пул.", "proxySettings.wireguardGenerated": "Пара ключей WireGuard сгенерирована", "proxySettings.desc": "Настройте параметры протокола для этого пользователя.", "selectNode": "Выберите узел", @@ -3188,11 +3198,6 @@ "createUsersDesc": "Создайте несколько пользователей одновременно из шаблона", "groups": "Группы", "proxySettings": "Настройки прокси", - "wireguardPeerIps": "WireGuard IPs", - "wireguardPeerIpsDesc": "Переназначить peer IPs из глобального пула", - "replaceAllPeerIps": "Заменить все IPs", - "replaceAllPeerIpsHint": "Если включено: у каждого затронутого пользователя новый peer IPs из пула. Если выключено: обновляются только неверные или отсутствующие peer IPs.", - "replaceInvalidPeerIpsOnly": "Только неверные или отсутствующие", "preview": "Предпросмотр", "previewing": "Идёт предпросмотр…", "previewTitle": "Предпросмотр", diff --git a/dashboard/public/statics/locales/zh.json b/dashboard/public/statics/locales/zh.json index d0900cdde..8a2759038 100644 --- a/dashboard/public/statics/locales/zh.json +++ b/dashboard/public/statics/locales/zh.json @@ -114,6 +114,18 @@ "2h": "2小时" } }, + "wireguard": { + "title": "WireGuard", + "description": "WireGuard 核心的对端 IP 池 — 用量、空闲地址与接口标签", + "empty": "尚无 WireGuard 子网。请创建带接口地址的 WireGuard 核心以开始分配对端 IP。", + "used": "已用", + "free": "空闲", + "capacity": "容量", + "usage": "使用率", + "freeIps": "空闲 IP", + "noTags": "无接口标签", + "refresh": "刷新" + }, "reconnectinfo": "刷新所有节点连接以解决连接问题", "reconnectAll": "重新连接所有节点", "reconnectingAll": "重新连接中...", @@ -1752,9 +1764,7 @@ "userDialog.proxySettings.hysteriaAuth": "Hysteria 认证", "userDialog.proxySettings.wireguardPrivateKey": "WireGuard 私钥", "userDialog.proxySettings.wireguardPublicKey": "WireGuard 公钥", - "userDialog.proxySettings.wireguardPeerIps": "WireGuard 对端 IP", "userDialog.proxySettings.generateWireGuardKeyPair": "WireGuard 密钥对", - "userDialog.proxySettings.peerIpsHint": "留空则从全局 WireGuard 对端地址池自动分配；手动填写时每行输入一个 CIDR，且每个值都必须在该地址池范围内。", "userDialog.proxySettings.wireguardGenerated": "WireGuard 密钥对已生成", "userDialog.proxySettings.desc": "为此用户配置协议专属设置。", "userDialog.expireDate": "过期日期", @@ -3238,11 +3248,6 @@ "createUsersDesc": "从模板一次性创建多个用户", "groups": "用户组", "proxySettings": "代理设置", - "wireguardPeerIps": "WireGuard IPs", - "wireguardPeerIpsDesc": "从全局地址池重新分配对端 IPs", - "replaceAllPeerIps": "替换全部 IPs", - "replaceAllPeerIpsHint": "启用时：为每个受影响的用户从池中分配新的对端 IPs。禁用时：仅更新无效或缺失的对端 IPs。", - "replaceInvalidPeerIpsOnly": "仅无效或缺失", "preview": "预览", "previewing": "正在预览…", "previewTitle": "预览", diff --git a/dashboard/src/app/router.tsx b/dashboard/src/app/router.tsx index c958fd752..d10fafb43 100644 --- a/dashboard/src/app/router.tsx +++ b/dashboard/src/app/router.tsx @@ -22,12 +22,12 @@ const BulkDataPage = lazyWithChunkRecovery(() => import('../pages/_dashboard.bul const BulkExpirePage = lazyWithChunkRecovery(() => import('../pages/_dashboard.bulk.expire')) const BulkGroupsPage = lazyWithChunkRecovery(() => import('../pages/_dashboard.bulk.groups')) const BulkProxyPage = lazyWithChunkRecovery(() => import('../pages/_dashboard.bulk.proxy')) -const BulkWireguardPage = lazyWithChunkRecovery(() => import('../pages/_dashboard.bulk.wireguard')) const Groups = lazyWithChunkRecovery(() => import('../pages/_dashboard.groups')) const Hosts = lazyWithChunkRecovery(() => import('../pages/_dashboard.hosts')) const Nodes = lazyWithChunkRecovery(() => import('../pages/_dashboard.nodes')) const NodesPage = lazyWithChunkRecovery(() => import('../pages/_dashboard.nodes._index')) const NodeLogs = lazyWithChunkRecovery(() => import('../pages/_dashboard.nodes.logs')) +const NodeWireGuard = lazyWithChunkRecovery(() => import('../pages/_dashboard.nodes.wireguard')) const Settings = lazyWithChunkRecovery(() => import('../pages/_dashboard.settings')) const CleanupSettings = lazyWithChunkRecovery(() => import('../pages/_dashboard.settings.cleanup')) const GeneralSettings = lazyWithChunkRecovery(() => import('../pages/_dashboard.settings.general')) @@ -168,6 +168,14 @@ export const router = createHashRouter([ ), }, + { + path: '/nodes/wireguard', + element: ( + }> + + + ), + }, ], }, { @@ -372,14 +380,6 @@ export const router = createHashRouter([ ), }, - { - path: '/bulk/wireguard', - element: ( - }> - - - ), - }, ], }, { diff --git a/dashboard/src/components/layout/sidebar.tsx b/dashboard/src/components/layout/sidebar.tsx index a44307abb..5e5e882e5 100644 --- a/dashboard/src/components/layout/sidebar.tsx +++ b/dashboard/src/components/layout/sidebar.tsx @@ -94,6 +94,11 @@ export function AppSidebar({ ...props }: React.ComponentProps) { icon: Cpu, matchPrefix: true, }, + { + title: 'nodes.wireguard.title', + url: '/nodes/wireguard', + icon: Network, + }, ] : []), ...(canReadNodeLogs @@ -316,11 +321,6 @@ export function AppSidebar({ ...props }: React.ComponentProps) { url: '/bulk/proxy', icon: Lock, }, - { - title: 'bulk.wireguardPeerIps', - url: '/bulk/wireguard', - icon: Network, - }, ] : []), ], diff --git a/dashboard/src/components/layout/tabbed-route-suspense-fallback.tsx b/dashboard/src/components/layout/tabbed-route-suspense-fallback.tsx index 51551cd4d..69c727cc9 100644 --- a/dashboard/src/components/layout/tabbed-route-suspense-fallback.tsx +++ b/dashboard/src/components/layout/tabbed-route-suspense-fallback.tsx @@ -34,6 +34,7 @@ type TabDef = { id: string; labelKey: string; icon: LucideIcon; url: string } const NODES_TABS: TabDef[] = [ { id: 'nodes.title', labelKey: 'nodes.title', icon: Share2, url: '/nodes' }, { id: 'core', labelKey: 'core', icon: Cpu, url: '/nodes/cores' }, + { id: 'nodes.wireguard.title', labelKey: 'nodes.wireguard.title', icon: Network, url: '/nodes/wireguard' }, { id: 'nodes.logs.title', labelKey: 'nodes.logs.title', icon: Logs, url: '/nodes/logs' }, ] @@ -55,7 +56,6 @@ const BULK_SUDO_TABS: TabDef[] = [ { id: 'expire', labelKey: 'bulk.expireDate', icon: Calendar, url: '/bulk/expire' }, { id: 'data', labelKey: 'bulk.dataLimit', icon: ArrowUpDown, url: '/bulk/data' }, { id: 'proxy', labelKey: 'bulk.proxySettings', icon: Lock, url: '/bulk/proxy' }, - { id: 'wireguard', labelKey: 'bulk.wireguardPeerIps', icon: Network, url: '/bulk/wireguard' }, ] const BULK_NON_SUDO_TABS: TabDef[] = [{ id: 'create', labelKey: 'bulk.createUsers', icon: UserPlus, url: '/bulk' }] @@ -67,6 +67,7 @@ const TEMPLATES_TABS: TabDef[] = [ function nodesActiveTabId(pathname: string): string { if (pathname.startsWith('/nodes/cores')) return 'core' + if (pathname.startsWith('/nodes/wireguard')) return 'nodes.wireguard.title' if (pathname.startsWith('/nodes/logs')) return 'nodes.logs.title' return 'nodes.title' } @@ -75,6 +76,9 @@ function nodesHeader(pathname: string): { title: string; description: string } { if (pathname.startsWith('/nodes/cores')) { return { title: 'settings.cores.title', description: 'settings.cores.description' } } + if (pathname.startsWith('/nodes/wireguard')) { + return { title: 'nodes.wireguard.title', description: 'nodes.wireguard.description' } + } if (pathname.startsWith('/nodes/logs')) { return { title: 'nodes.logs.title', description: 'nodes.logs.description' } } @@ -102,7 +106,6 @@ function bulkHeader(pathname: string): { title: string; description: string } { '/bulk/expire': { title: 'bulk.expireDate', description: 'bulk.expireDateDesc' }, '/bulk/data': { title: 'bulk.dataLimit', description: 'bulk.dataLimitDesc' }, '/bulk/proxy': { title: 'bulk.proxySettings', description: 'bulk.proxySettingsDesc' }, - '/bulk/wireguard': { title: 'bulk.wireguardPeerIps', description: 'bulk.wireguardPeerIpsDesc' }, } return pathToHeader[pathname] ?? pathToHeader['/bulk']! } diff --git a/dashboard/src/features/bulk/components/bulk-flow.tsx b/dashboard/src/features/bulk/components/bulk-flow.tsx index 27f26f18d..02ffc1743 100644 --- a/dashboard/src/features/bulk/components/bulk-flow.tsx +++ b/dashboard/src/features/bulk/components/bulk-flow.tsx @@ -10,7 +10,6 @@ import { useBulkModifyUsersExpire, useBulkAddGroupsToUsers, useBulkRemoveUsersFromGroups, - useBulkReallocateWireguardPeerIps, ShadowsocksMethods, UserStatus, } from '@/service/api' @@ -40,7 +39,7 @@ import { endOfDay, startOfDay } from 'date-fns' const PAGE_SIZE = 50 -type BulkOperationType = 'proxy' | 'data' | 'expire' | 'groups' | 'wireguard' +type BulkOperationType = 'proxy' | 'data' | 'expire' | 'groups' type ExpiryUnit = TimeUnit interface BulkFlowProps { @@ -68,8 +67,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { const [groupsOperation, setGroupsOperation] = useState<'add' | 'remove'>('add') - const [replaceAllPeerIps, setReplaceAllPeerIps] = useState(false) - const [selectedGroups, setSelectedGroups] = useState([]) const [selectedUsers, setSelectedUsers] = useState([]) const [selectedAdmins, setSelectedAdmins] = useState([]) @@ -138,7 +135,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { const expireMutation = useBulkModifyUsersExpire() const addGroupsMutation = useBulkAddGroupsToUsers() const removeGroupsMutation = useBulkRemoveUsersFromGroups() - const wireguardPeerIpsMutation = useBulkReallocateWireguardPeerIps() const nextStep = () => { if (currentStep < 3) setCurrentStep((currentStep + 1) as 1 | 2 | 3) @@ -151,9 +147,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { const canProceedToNext = () => { switch (currentStep) { case 1: - if (operationType === 'wireguard') { - return true - } if (operationType === 'proxy') { return selectedMethod } @@ -178,8 +171,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { case 'groups': // Allow proceeding even if no targets selected - will apply to all users return true - case 'wireguard': - return true default: return false } @@ -209,22 +200,13 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { (operationType === 'proxy' && proxyMutation.isPending) || (operationType === 'data' && dataMutation.isPending) || (operationType === 'expire' && expireMutation.isPending) || - (operationType === 'groups' && (groupsOperation === 'add' ? addGroupsMutation.isPending : removeGroupsMutation.isPending)) || - (operationType === 'wireguard' && wireguardPeerIpsMutation.isPending) + (operationType === 'groups' && (groupsOperation === 'add' ? addGroupsMutation.isPending : removeGroupsMutation.isPending)) const bulkPreviewDescription = (response: unknown) => { if (!response || typeof response !== 'object') return '' const r = response as Record const count = typeof r.affected_users === 'number' ? r.affected_users : undefined if (count === undefined) return '' - const inbounds = typeof r.wireguard_inbound_tags === 'number' ? r.wireguard_inbound_tags : undefined - if (inbounds !== undefined && inbounds > 0) { - return t('bulk.previewToastWithInbounds', { - count, - inbounds, - defaultValue: '{{count}} would be affected · {{inbounds}} inbounds (dry run)', - }) - } return t('bulk.previewToast', { count, defaultValue: '{{count}} would be affected (dry run)', @@ -272,14 +254,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { admins: selectedAdmins.length ? selectedAdmins : [], dry_run: false, } - case 'wireguard': - return { - ...basePayload, - ...statusPayload, - confirm: true, - dry_run: false, - replace_all: replaceAllPeerIps, - } default: return basePayload } @@ -295,8 +269,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { return expireMutation case 'groups': return groupsOperation === 'add' ? addGroupsMutation : removeGroupsMutation - case 'wireguard': - return wireguardPeerIpsMutation default: return proxyMutation } @@ -308,29 +280,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { { data: payload as any }, { onSuccess: response => { - if (response && typeof response === 'object' && 'wireguard_inbound_tags' in response && 'dry_run' in response && (response as { dry_run?: boolean }).dry_run === false) { - const r = response as { affected_users?: number; updated?: number; wireguard_inbound_tags: number } - const n = typeof r.affected_users === 'number' ? r.affected_users : (r.updated ?? 0) - toast.success(t('operationSuccess', { defaultValue: 'Done' }), { - description: t('bulk.applySuccessWithInbounds', { - count: n, - inbounds: r.wireguard_inbound_tags, - defaultValue: '{{count}} updated · {{inbounds}} inbounds', - }), - }) - setCurrentStep(1) - setReplaceAllPeerIps(false) - setSelectedGroups([]) - setSelectedUsers([]) - setSelectedAdmins([]) - setSelectedHasGroups([]) - setSelectedStatuses([]) - setExpiredAfter(undefined) - setExpiredBefore(undefined) - setShowConfirmDialog(false) - return - } - const detail = typeof response === 'object' && response && 'detail' in response ? response.detail : undefined let description = '' if (detail) { @@ -411,14 +360,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { admins: selectedAdmins.length ? selectedAdmins : [], dry_run: true, } - case 'wireguard': - return { - ...basePayload, - ...statusPayload, - dry_run: true, - confirm: false, - replace_all: replaceAllPeerIps, - } } })() @@ -432,8 +373,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { return expireMutation case 'groups': return groupsOperation === 'add' ? addGroupsMutation : removeGroupsMutation - case 'wireguard': - return wireguardPeerIpsMutation default: return proxyMutation } @@ -462,7 +401,7 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { // For groups operation, groups are the operation target, not user targets // So isApplyToAll should only check users, admins, and hasGroups const totalTargets = selectedUsers.length + selectedAdmins.length + (operationType === 'groups' ? selectedHasGroups.length : selectedGroups.length) - const hasStatusFilter = (operationType === 'data' || operationType === 'expire' || operationType === 'wireguard') && selectedStatuses.length > 0 + const hasStatusFilter = (operationType === 'data' || operationType === 'expire') && selectedStatuses.length > 0 const statusTargetCount = hasStatusFilter ? selectedStatuses.length : 0 const hasExpireDateFilter = (operationType === 'data' || operationType === 'expire') && Boolean(expiredAfter || expiredBefore) const expireDateFilterCount = hasExpireDateFilter ? Number(Boolean(expiredAfter)) + Number(Boolean(expiredBefore)) : 0 @@ -769,22 +708,6 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { )} - - {operationType === 'wireguard' && ( -
- setReplaceAllPeerIps(v === true)} className="mt-0.5" /> -
- -

- {t('bulk.replaceAllPeerIpsHint', { - defaultValue: 'When enabled, every affected user gets a new peer IP from the pool. When disabled, only invalid or missing peer IPs are updated.', - })} -

-
-
- )} )} @@ -804,7 +727,7 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { )} - {(operationType === 'data' || operationType === 'expire' || operationType === 'wireguard') && ( + {(operationType === 'data' || operationType === 'expire') && (
@@ -977,19 +900,9 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { {operationType === 'data' && t('bulk.dataLimit')} {operationType === 'expire' && t('bulk.expireDate')} {operationType === 'groups' && t('bulk.groups')} - {operationType === 'wireguard' && t('bulk.wireguardPeerIps')}
- {operationType === 'wireguard' && ( -
- {t('bulk.settings', { defaultValue: 'Settings' })}: - - {replaceAllPeerIps ? t('bulk.replaceAllPeerIps', { defaultValue: 'Replace all IPs' }) : t('bulk.replaceInvalidPeerIpsOnly', { defaultValue: 'Invalid or missing IPs only' })} - -
- )} - {operationType === 'proxy' && (
{t('bulk.settings', { defaultValue: 'Settings' })}: @@ -1017,7 +930,7 @@ export default function BulkFlow({ operationType }: BulkFlowProps) {
)} - {(operationType === 'data' || operationType === 'expire' || operationType === 'wireguard') && selectedStatuses.length > 0 && ( + {(operationType === 'data' || operationType === 'expire') && selectedStatuses.length > 0 && (
{t('status', { defaultValue: 'Status' })}: {selectedStatuses.map(status => t(`status.${status}`, { defaultValue: status.replace(/_/g, ' ') })).join(', ')} @@ -1157,11 +1070,9 @@ export default function BulkFlow({ operationType }: BulkFlowProps) { {t('cancel', { defaultValue: 'Cancel' })} - {proxyMutation.isPending || dataMutation.isPending || expireMutation.isPending || addGroupsMutation.isPending || removeGroupsMutation.isPending || wireguardPeerIpsMutation.isPending + {proxyMutation.isPending || dataMutation.isPending || expireMutation.isPending || addGroupsMutation.isPending || removeGroupsMutation.isPending ? t('applying', { defaultValue: 'Applying...' }) : t('confirm', { defaultValue: 'Confirm' })} diff --git a/dashboard/src/features/nodes/components/wireguard-subnets-list.tsx b/dashboard/src/features/nodes/components/wireguard-subnets-list.tsx new file mode 100644 index 000000000..69c82c347 --- /dev/null +++ b/dashboard/src/features/nodes/components/wireguard-subnets-list.tsx @@ -0,0 +1,127 @@ +import { Badge } from '@/components/ui/badge' +import { Button } from '@/components/ui/button' +import { Card } from '@/components/ui/card' +import { Progress } from '@/components/ui/progress' +import { Skeleton } from '@/components/ui/skeleton' +import { useGetWireguardSubnets, type WireGuardSubnetUsage } from '@/service/api' +import { cn } from '@/lib/utils' +import { ChevronDown, RefreshCw } from 'lucide-react' +import { useMemo, useState } from 'react' +import { useTranslation } from 'react-i18next' + +function usagePercent(row: WireGuardSubnetUsage): number { + if (row.capacity <= 0) return 0 + return Math.min(100, Math.round((row.used / row.capacity) * 100)) +} + +function SubnetCard({ row }: { row: WireGuardSubnetUsage }) { + const { t } = useTranslation() + const [showFree, setShowFree] = useState(false) + const percent = usagePercent(row) + const previewFree = row.free_ips.slice(0, 12) + const hasMoreFree = row.free_ips.length > previewFree.length + + return ( + +
+
+
+
{row.subnet}
+
+ {row.interface_tags.length === 0 ? ( + {t('nodes.wireguard.noTags')} + ) : ( + row.interface_tags.map(tag => ( + + {tag} + + )) + )} +
+
+
+
+ {t('nodes.wireguard.used')}: {row.used} +
+
+ {t('nodes.wireguard.free')}: {row.free} +
+
+ {t('nodes.wireguard.capacity')}: {row.capacity} +
+
+
+ +
+
+ {t('nodes.wireguard.usage')} + {percent}% +
+ = 90 ? 'bg-destructive' : percent >= 70 ? 'bg-amber-500' : undefined)} + /> +
+ + {row.free_ips.length > 0 && ( +
+ + {showFree && ( +
+ {previewFree.map(ip => ( + + {ip} + + ))} + {hasMoreFree && …} +
+ )} +
+ )} +
+ + ) +} + +export default function WireGuardSubnetsList() { + const { t } = useTranslation() + const { data, isLoading, isFetching, refetch } = useGetWireguardSubnets() + const rows = useMemo(() => data ?? [], [data]) + + return ( +
+
+ +
+ + {isLoading ? ( +
+ {Array.from({ length: 4 }).map((_, i) => ( + + ))} +
+ ) : rows.length === 0 ? ( + {t('nodes.wireguard.empty')} + ) : ( +
+ {rows.map(row => ( + + ))} +
+ )} +
+ ) +} diff --git a/dashboard/src/features/users/dialogs/user-modal.tsx b/dashboard/src/features/users/dialogs/user-modal.tsx index ea8120029..cab52c3b7 100644 --- a/dashboard/src/features/users/dialogs/user-modal.tsx +++ b/dashboard/src/features/users/dialogs/user-modal.tsx @@ -1231,7 +1231,6 @@ function UserModal({ isDialogOpen, onOpenChange, form, editingUser, editingUserI wireguard: { private_key: keyPair.privateKey, public_key: keyPair.publicKey, - peer_ips: form.getValues('proxy_settings.wireguard.peer_ips') ?? [], }, } form.setValue('proxy_settings', newSettings as any, { shouldDirty: true, shouldValidate: true }) @@ -1258,10 +1257,6 @@ function UserModal({ isDialogOpen, onOpenChange, form, editingUser, editingUserI [form, handleFieldChange], ) - const parseWireGuardPeerIps = React.useCallback((value: string) => { - return value.split('\n') - }, []) - const hasMeaningfulProxyValue = React.useCallback((value: unknown): boolean => { if (Array.isArray(value)) { return value.some(item => hasMeaningfulProxyValue(item)) @@ -1285,18 +1280,7 @@ function UserModal({ isDialogOpen, onOpenChange, form, editingUser, editingUserI const cleanedProtocolSettings = Object.entries(settings as Record).reduce( (protocolAcc, [key, value]) => { if (Array.isArray(value)) { - const cleanedList = value - .flatMap(item => { - if (typeof item !== 'string') { - return [item] - } - if (protocol === 'wireguard' && key === 'peer_ips') { - return item.split(',') - } - return [item] - }) - .map(item => (typeof item === 'string' ? item.trim() : item)) - .filter(item => hasMeaningfulProxyValue(item)) + const cleanedList = value.map(item => (typeof item === 'string' ? item.trim() : item)).filter(item => hasMeaningfulProxyValue(item)) if (cleanedList.length > 0) { protocolAcc[key] = cleanedList @@ -2135,34 +2119,6 @@ function UserModal({ isDialogOpen, onOpenChange, form, editingUser, editingUserI )} /> - ( - - {t('userDialog.proxySettings.wireguardPeerIps', { defaultValue: 'WireGuard Peer IPs' })} - -