HY2XS_flamy/apps/service/peer_admission.go

package service

import (
	"sync"
	"time"
)

// Лимит устройств выдерживает ПАРАЛЛЕЛЬНЫЕ запросы авторизации.
//
// Что было. Проверка выглядела так:
//
//	onlineUsers, err := hysteria2Online()
//	if device, exist := onlineUsers[authID]; exist && maxDevices <= device {
//	    return deny
//	}
//	return allow
//
// Между чтением `/online` и ответом «allow» нет ничего, что удержало бы место,
// поэтому при одновременных подключениях выходило:
//
//	A: GET /online -> 2      B: GET /online -> 2
//	                 max = 3
//	A: 2 < 3 -> allow        B: 2 < 3 -> allow
//	                 стало 4
//
// Объявленный в панели «Лимит устройств: 3» превышался ровно тем способом,
// от которого лимит и должен защищать.
//
// Обычный мьютекс вокруг `/online` проблему не решает, и это главное, что
// нужно понимать про этот файл. Ответив «allow», админка не создаёт
// подключение — его только начинает устанавливать Hysteria, и в статистику
// клиент попадает позже. Следующий `/online`, даже строго после первого,
// продолжает показывать прежнее число. Сериализация запросов лишь сузила бы
// окно, оставив дефект на месте.
//
// Поэтому админка ведёт собственный учёт уже выданных, но ещё не проявившихся
// разрешений. HY2XS — один процесс на одном сервере с Hysteria, поэтому учёт
// process-local: ни Redis, ни таблицы в базе, ни распределённых блокировок для
// этого не нужно.
//
// Чего этот механизм НЕ обещает. Без обратного вызова от Hysteria
// «соединение установлено / не установлено» математически точной системы
// резервирования не построить. Он закрывает конкретный и реальный TOCTOU —
// параллельные HTTP-auth одного процесса, — и делает это fail-closed.
//
// Второй TOCTOU: ПЕРЕУПОРЯДОЧИВАНИЕ снимков `/online`.
//
// Учёта выданных разрешений самого по себе оказалось недостаточно, и это
// отдельный дефект, а не оттенок первого. Сетевой запрос выполнялся вне
// мьютекса, поэтому снимки приходили в критическую секцию в произвольном
// порядке — более старый мог обогнать более новый:
//
//	1. A получает разрешение при `/online = 0`; pending = [A], lastOnline = 0
//	2. B читает `/online = 0` и задерживается на обратном пути
//	3. A действительно подключается, Hysteria показывает `/online = 1`
//	4. C читает `/online = 1` и входит в резервацию ПЕРВЫМ:
//	   разрешение A признано проявившимся, lastOnline = 1, C получает отказ
//	5. B входит со своим устаревшим `online = 0`
//	6. `online > lastOnline` ложно, после чего lastOnline откатывается в 0
//	7. `0 + pending(0) < 1` -> B получает ALLOW
//
// При `maxDevices = 1` подключений становится два. Это НЕ data race: вся
// работа с памятью защищена мьютексом, поэтому детектор гонок здесь молчит
// принципиально, и поймать дефект может только семантическая проверка.
//
// Лечится это не глобальным мьютексом вокруг сети — он сериализовал бы
// подключения всех пиров через один HTTP-обмен, — а замком на ОДИН authID:
// см. lockPeerAdmission. Конкурируют между собой только авторизации одного и
// того же пира, а их упорядоченность и есть требуемое свойство: снимок,
// прочитанный под замком, не может оказаться старше уже обработанного.
//
// Оба механизма нужны одновременно и закрывают разные половины:
//
//	замок по authID  — снимки не переупорядочиваются;
//	учёт разрешений  — снимок не успевает измениться к следующему запросу.

// pendingAdmissionTTL — срок жизни выданного разрешения, которое ещё не
// проявилось в `/online`.
//
// Величина внутренняя и пользовательской настройкой не является намеренно: это
// не политика доступа, а компенсация задержки между ответом авторизации и
// появлением клиента в статистике Hysteria. Настройка, смысл которой оператор
// не может оценить, порождает только неверные значения.
//
// Тридцать секунд — с большим запасом относительно установления QUIC-сессии и
// при этом заметно меньше, чем интервал, на котором оператор вообще заметил бы
// занятый слот. Если клиент авторизовался, но так и не подключился, резервация
// исчезает сама.
const pendingAdmissionTTL = 30 * time.Second

// admissionState — учёт по одному пиру.
type admissionState struct {
	// lastOnline — число устройств, показанное Traffic Stats API в прошлый
	// раз. Нужно, чтобы отличить рост (клиент подключился, резервация
	// проявилась) от неизменного значения.
	lastOnline int64
	// pending — сроки годности выданных, но ещё не проявившихся разрешений.
	// Хранится по одному значению на разрешение, а не счётчиком: иначе
	// протухать они могли бы только все разом.
	pending []time.Time
}

var deviceAdmissions = struct {
	sync.Mutex
	byAuthID map[string]*admissionState
}{byAuthID: map[string]*admissionState{}}

// admissionGate — замок одного authID вместе со счётчиком тех, кому он сейчас
// нужен.
//
// Счётчик существует ради удаления записи. Без него карта замков росла бы по
// одной записи на каждый когда-либо авторизовавшийся authId и не уменьшалась
// бы никогда — то есть та же утечка, от которой в учёте разрешений защищает
// forgetIfIdle, только этажом выше.
type admissionGate struct {
	mu sync.Mutex
	// waiting — сколько вызывающих держат замок или ждут его. Пока значение
	// больше нуля, запись обязана оставаться в карте: удалив её, второй
	// вызывающий создал бы НОВЫЙ замок и разошёлся бы с первым.
	waiting int
}

var admissionGates = struct {
	sync.Mutex
	byAuthID map[string]*admissionGate
}{byAuthID: map[string]*admissionGate{}}

// lockPeerAdmission сериализует последовательность «прочитать `/online` ->
// занять место» для ОДНОГО пира и возвращает функцию освобождения.
//
// Замок именно по authID, а не один на процесс, и это существенно. Внутри него
// выполняется сетевой запрос к Traffic Stats API, поэтому общий замок означал
// бы, что все подключения всех пиров выстраиваются в очередь за одним
// HTTP-обменом. Здесь же конкурируют только авторизации одного пира — то есть
// ровно те, для которых порядок и решается.
//
// Время удержания ограничено сверху таймаутом самого обращения к Traffic Stats
// API (proxy.Hysteria2Api ставит контекст на 3 секунды), поэтому «застрявшая»
// Hysteria не превращает замок в бессрочный.
//
// Карта замков и карта учёта разрешений намеренно раздельны: первая описывает,
// кто сейчас проходит авторизацию, вторая — что уже выдано. Совмещение их в
// одной структуре означало бы удерживать мьютекс учёта на время сетевого
// запроса.
func lockPeerAdmission(authID string) func() {
	admissionGates.Lock()
	gate := admissionGates.byAuthID[authID]
	if gate == nil {
		gate = &admissionGate{}
		admissionGates.byAuthID[authID] = gate
	}
	gate.waiting++
	admissionGates.Unlock()

	gate.mu.Lock()

	return func() {
		gate.mu.Unlock()

		admissionGates.Lock()
		defer admissionGates.Unlock()
		gate.waiting--
		if gate.waiting == 0 {
			delete(admissionGates.byAuthID, authID)
		}
	}
}

// reserveDeviceSlot решает, есть ли для нового подключения свободное место, и
// занимает его.
//
// Возвращает true, если подключение можно разрешить.
//
// Сетевой запрос к `/online` выполняется ВНЕ этого мьютекса — вызывающий
// передаёт сюда уже полученное число. Под блокировкой остаются только
// несколько операций с map: держать её на время HTTP-обмена значило бы
// сериализовать все подключения всех пиров через один сетевой запрос.
//
// Упорядоченность снимков обеспечивает не этот мьютекс, а замок по authID:
// вызывающий обязан удерживать lockPeerAdmission от чтения `/online` и до
// возврата отсюда. Без него сюда попадал бы снимок старше уже обработанного, и
// строка `state.lastOnline = online` откатывала бы учёт назад — см. описание
// второго TOCTOU в начале файла.
func reserveDeviceSlot(authID string, online int64, maxDevices int64, now time.Time) bool {
	deviceAdmissions.Lock()
	defer deviceAdmissions.Unlock()

	state := deviceAdmissions.byAuthID[authID]
	if state == nil {
		state = &admissionState{}
		deviceAdmissions.byAuthID[authID] = state
	}

	// 1. Протухшие разрешения освобождают место: клиент, который авторизовался
	//    и не подключился, не должен занимать слот вечно.
	state.dropExpired(now)

	// 2. Рост числа онлайн-устройств означает, что ровно столько выданных
	//    разрешений уже превратились в подключения. Не сняв их, админка
	//    посчитала бы одно и то же устройство дважды — сначала как резервацию,
	//    потом как реальное подключение, — и лимит стал бы вдвое строже
	//    объявленного.
	if online > state.lastOnline {
		state.dropMaterialized(online - state.lastOnline)
	}
	state.lastOnline = online

	// 3. Решение принимается по сумме: подтверждённые подключения плюс ещё не
	//    проявившиеся разрешения.
	if online+int64(len(state.pending)) >= maxDevices {
		// Место не занято и запись может оказаться ненужной: убираем её, чтобы
		// карта не росла от одних отказов.
		state.forgetIfIdle(authID)
		return false
	}

	state.pending = append(state.pending, now.Add(pendingAdmissionTTL))
	return true
}

// Функции «вернуть занятое место» здесь нет намеренно. После выдачи разрешения
// в Hysteria2Auth не остаётся ни одного шага, способного отказать, а
// controller.Hysteria2Auth на успешном ответе только обновляет
// last_connection_at и неудачу этого обновления считает несущественной.
// Освобождение, у которого нет вызывающего, было бы вторым способом менять
// состояние трекера — и первым кандидатом разойтись с reserveDeviceSlot.
// Разрешение, за которым не последовало подключения, снимает TTL.

func (s *admissionState) dropExpired(now time.Time) {
	kept := s.pending[:0]
	for _, deadline := range s.pending {
		if deadline.After(now) {
			kept = append(kept, deadline)
		}
	}
	s.pending = kept
}

// dropMaterialized снимает count самых старых разрешений: раньше выдано —
// раньше подключилось.
func (s *admissionState) dropMaterialized(count int64) {
	if count >= int64(len(s.pending)) {
		s.pending = s.pending[:0]
		return
	}
	s.pending = s.pending[count:]
}

// forgetIfIdle убирает запись, о которой больше нечего помнить.
//
// Без этого карта росла бы по одной записи на каждый когда-либо
// авторизовавшийся authId и не уменьшалась бы никогда — включая записи
// давно удалённых пиров.
func (s *admissionState) forgetIfIdle(authID string) {
	if len(s.pending) == 0 && s.lastOnline == 0 {
		delete(deviceAdmissions.byAuthID, authID)
	}
}

// sweepDeviceAdmissions убирает записи о пирах, за которыми ничего не числится.
//
// Зачем это нужно отдельно от forgetIfIdle. Тот срабатывает ТОЛЬКО на ветке
// отказа: после успешной выдачи разрешения запись остаётся с непустым pending,
// а когда разрешение протухает, снять запись уже некому — следующего обращения
// к этому authId может не быть никогда. Так в карте оставались пиры, удалённые
// из панели, и старые authId, переписанные импортом: за время жизни процесса
// она только росла.
//
// Решение принимается по ФАКТИЧЕСКОЙ картине подключений, а не по хранимому
// lastOnline: последний обновляется только на пути авторизации, поэтому у
// отключившегося пира он остаётся прежним сколь угодно долго.
//
// Удаление записи, у которой нет ни одного действующего разрешения и нет
// подключений, не меняет ни одного будущего решения. Следующая резервация
// начнёт с чистой записи и придёт к тому же ответу: dropMaterialized на пустом
// списке — no-op, а `state.lastOnline` в любом случае перезаписывается
// пришедшим значением до сравнения с лимитом.
//
// Замок authID здесь не берётся намеренно: удерживать его на всём обходе карты
// значило бы останавливать авторизацию каждые 30 секунд. Гонка с параллельной
// авторизацией безопасна — она либо уже добавила разрешение (тогда pending не
// пуст и запись остаётся), либо ещё не дошла до учёта (тогда она создаст
// запись заново, и это ровно та же чистая запись).
func sweepDeviceAdmissions(online map[string]int64, now time.Time) {
	deviceAdmissions.Lock()
	defer deviceAdmissions.Unlock()

	for authID, state := range deviceAdmissions.byAuthID {
		state.dropExpired(now)
		if len(state.pending) > 0 {
			continue
		}
		if online[authID] > 0 {
			continue
		}
		delete(deviceAdmissions.byAuthID, authID)
	}
}

// resetDeviceAdmissions очищает учёт. Существует ради тестов: состояние здесь
// принадлежит процессу, и без сброса тесты видели бы резервации друг друга.
func resetDeviceAdmissions() {
	deviceAdmissions.Lock()
	deviceAdmissions.byAuthID = map[string]*admissionState{}
	deviceAdmissions.Unlock()

	// Замки сбрасывать НЕЛЬЗЯ: удерживаемый кем-то замок, потерянный из карты,
	// перестал бы исключать второго вызывающего. Вместо сброса тесты проверяют,
	// что после завершения работы карта пуста сама.
}

// heldPeerAdmissionGates — число живых замков. Существует ради тестов: утечка
// здесь выглядит именно как незакрытая запись в карте.
func heldPeerAdmissionGates() int {
	admissionGates.Lock()
	defer admissionGates.Unlock()
	return len(admissionGates.byAuthID)
}