73 lines
3.2 KiB
Python
73 lines
3.2 KiB
Python
"""Migration identity map validation, before starting Telegram polling."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from uuid import UUID, uuid5
|
|
from collections.abc import Iterable
|
|
|
|
|
|
def deterministic_user_uuid(user_id: int) -> str:
|
|
return str(uuid5(UUID("2a438583-9a10-4b90-831f-7db0432b5578"), str(user_id)))
|
|
|
|
|
|
def load_legacy_user_ids(path: Path) -> dict[int, str]:
|
|
entries = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(entries, list) or not entries:
|
|
raise ValueError("Legacy Remnawave identity map must be a non-empty list")
|
|
result: dict[int, str] = {}
|
|
seen_uuids: set[str] = set()
|
|
for entry in entries:
|
|
user_id = entry["id"]
|
|
if not isinstance(user_id, int) or isinstance(user_id, bool) or user_id <= 0:
|
|
raise ValueError("Legacy Remnawave user ID must be a positive integer")
|
|
local_uuid = str(UUID(entry["uuid"]))
|
|
if user_id in result or local_uuid in seen_uuids:
|
|
raise ValueError("Duplicate user identity in legacy Remnawave map")
|
|
result[user_id] = local_uuid
|
|
seen_uuids.add(local_uuid)
|
|
return result
|
|
|
|
|
|
def validate_cached_identities(
|
|
legacy: dict[int, str],
|
|
rows: Iterable[tuple[int, str] | tuple[int, str, str]],
|
|
*,
|
|
remote_identities: Iterable[tuple[int, str] | tuple[int, str, str]] | None = None,
|
|
) -> int:
|
|
"""Keep historical deleted cache rows only after a complete remote snapshot.
|
|
|
|
Never turn a deleted account's UUID into an alias for its old numeric ID.
|
|
A reused ID or a UUID assigned to a different existing ID fails closed.
|
|
"""
|
|
snapshot = None if remote_identities is None else list(remote_identities)
|
|
remote = None if snapshot is None else {
|
|
item[0]: str(UUID(item[1])) for item in snapshot
|
|
}
|
|
remote_short_uuids = {} if snapshot is None else {
|
|
item[0]: item[2] for item in snapshot if len(item) > 2
|
|
}
|
|
legacy_ids_by_uuid = {local_uuid: user_id for user_id, local_uuid in legacy.items()}
|
|
remote_ids_by_uuid = {} if remote is None else {
|
|
local_uuid: user_id for user_id, local_uuid in remote.items()
|
|
}
|
|
historical_deleted = 0
|
|
for row in rows:
|
|
user_id, value = row[:2]
|
|
local_uuid = str(UUID(value))
|
|
if len(row) > 2 and user_id in remote_short_uuids and row[2] != remote_short_uuids[user_id]:
|
|
raise ValueError("Remote Remnawave short UUID conflicts with the bot cache")
|
|
expected = legacy.get(user_id, deterministic_user_uuid(user_id))
|
|
if legacy_ids_by_uuid.get(local_uuid, user_id) != user_id:
|
|
raise ValueError("Legacy Remnawave UUID belongs to another cached ID")
|
|
if remote_ids_by_uuid.get(local_uuid, user_id) != user_id:
|
|
raise ValueError("Remote Remnawave UUID belongs to another cached ID")
|
|
if local_uuid == expected:
|
|
if remote is not None and user_id in remote and remote[user_id] != local_uuid:
|
|
raise ValueError("Remote Remnawave identity conflicts with the bot cache")
|
|
continue
|
|
if user_id in legacy or remote is None or user_id in remote:
|
|
raise ValueError("Legacy Remnawave map does not match the existing bot cache")
|
|
historical_deleted += 1
|
|
return historical_deleted
|