ShadowNet VPN

services/database/users.py, migrations.py

def norm_email(email: str | None) -> str | None:    """Канонический вид e-mail: без краевых пробелов, в нижнем регистре.    Уникальность в БД держит функциональный индекс idx_users_email_lower, поэтому    и хранить, и искать надо в одном регистре. До этой нормализации точечный поиск    (email = $1) расходился с индексом: аккаунт, сохранённый как Ivan@Mail.ru, не    находился по ivan@mail.ru — вход и сброс пароля молча не работали.    """    return email.strip().lower() if email else email...    await pool.execute(        "CREATE INDEX IF NOT EXISTS idx_users_email_canon "        "ON users (LOWER(BTRIM(email))) WHERE email IS NOT NULL"    )

core/internal/tun/netop.go, cleanup_windows.go

var (    // errNotFound — такого маршрута или адреса нет (ERROR_NOT_FOUND,    // ERROR_FILE_NOT_FOUND). Для снятия это не отказ, а цель, достигнутая    // раньше нас.    errNotFound = errors.New("такого нет")    // errNoInterface — интерфейса с таким именем или номером в системе нет.    errNoInterface = errors.New("интерфейса нет"))// absent отвечает, значит ли отказ «снимать нечего»: ни объекта, ни// интерфейса, на котором он стоял.func absent(err error) bool {    return errors.Is(err, errNotFound) || errors.Is(err, errNoInterface)}...for _, r := range before {    // Исчезнувший между вопросом и снятием — цель достигнута (absent).    if err := do(ctx, routeOp{prefix: r.prefix, nextHop: r.nextHop, iface: r.iface}); err != nil && !absent(err) {        errs = append(errs, err)    }}

services/scheduler.py

# Пока касса молчит — напоминать не чаще, чем раз в 6 ч: сбой, который не починили за# ночь, утром должен быть виден снова, а лента из одинаковых сообщений раз в 15 минут# приучает их не читать (так и вышло со старым правилом analytics_bot).PAYMENT_SILENCE_REMIND_HOURS = 6async def watch_payment_silence(bot: Bot):    ...    now = datetime.now(timezone.utc)    silent = s["streak_alert"] or s["day_silence"]    if silent:        if state is None:            if await send_alert(bot, _payment_silence_text(s, now, None, links)):                await db.set_setting(_PAYMENT_SILENCE_KEY, json.dumps(                    {"since": now.isoformat(), "sent_at": now.isoformat()}))        elif now - sent_at >= timedelta(hours=PAYMENT_SILENCE_REMIND_HOURS):            if await send_alert(bot, _payment_silence_text(s, now, since, links)):                await db.set_setting(_PAYMENT_SILENCE_KEY, json.dumps(                    {"since": since.isoformat(), "sent_at": now.isoformat()}))        return    if state is not None:        ...        if await send_alert(bot, text):            await db.delete_setting(_PAYMENT_SILENCE_KEY)

leakwatch/matcher.py

"""Сопоставление чужих конфигов с нашим инвентарём.Уровни находок:  L1 — наш IP (нода в публичном обороте);  L2 — наш домен/SNI;  L3 — reality public key (доказательство, что утёк именно наш inbound);  L4 — UUID нашего клиента (видно, кто именно слил);  L5 — упоминание наших доменов в тексте (ссылка на подписку, README и т.п.)."""...def lookup_client(uuid: str, inv: dict) -> dict:    """Кто в панели стоит за UUID из конфига: {user_id, panel_id, status} или {}."""    client = inv["uuid_index"].get(inventory.hash_uuid(uuid)) if uuid else None    if not client:        return {}    return client if isinstance(client, dict) else {"user_id": client}...        if e.pbk and e.pbk in inv["pbks"]:            matched_here.append(Finding(                "L3", "high", e.pbk, "reality public key нашего inbound",                source, path, e.host, e.remark,            ))