package hysteria2 import ( "context" "crypto/rand" "encoding/hex" "encoding/json" "errors" "fmt" "io" "net/http" "net/netip" "regexp" "strings" "sync" "time" "github.com/sagernet/sing-box/log" E "github.com/sagernet/sing/common/exceptions" "github.com/go-chi/chi/v5" "github.com/go-chi/render" ) const ( sessionTTL = time.Minute realmNamePattern = `^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$` maxRequestBodyBytes = 4 << 10 maxAddresses = 8 nonceHexLength = 32 obfsHexLength = 64 eventChannelSize = 16 maxPendingAttempts = 16 connectResponseTimeout = 10 * time.Second ) var realmPattern = regexp.MustCompile(realmNamePattern) type contextKey int const ( contextKeyUser contextKey = iota contextKeySession ) type realmUser struct { name string maxRealms int } type realmSession struct { id string realmID string username string addresses []string expires time.Time events chan realmEvent timer *time.Timer done chan struct{} closed bool pending map[string]chan punchResponsePayload } type realmEvent struct { kind string data any } type punchEvent struct { Addresses []string `json:"addresses"` Nonce string `json:"nonce"` Obfs string `json:"obfs"` } type punchResponsePayload struct { addresses []string } type server struct { access sync.Mutex realms map[string]*realmSession sessions map[string]*realmSession userCounts map[string]int logger log.ContextLogger tokenMap map[string]*realmUser } func newServer(logger log.ContextLogger, tokenMap map[string]*realmUser) *server { return &server{ realms: make(map[string]*realmSession), sessions: make(map[string]*realmSession), userCounts: make(map[string]int), logger: logger, tokenMap: tokenMap, } } func validateRealmID(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") if !realmPattern.MatchString(id) { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": "invalid realm name"}) return } next.ServeHTTP(w, r) }) } func (s *server) authBearer(name string, key contextKey, lookup func(r *http.Request, token string) (any, bool)) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { header := r.Header.Get("Authorization") bearer, token, found := strings.Cut(header, " ") if bearer != "Bearer" || !found { render.Status(r, http.StatusUnauthorized) render.JSON(w, r, render.M{"error": "invalid_token", "message": "invalid " + name + " token"}) return } value, authenticated := lookup(r, token) if !authenticated { render.Status(r, http.StatusUnauthorized) render.JSON(w, r, render.M{"error": "invalid_token", "message": "invalid " + name + " token"}) return } next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), key, value))) }) } } func (s *server) authUser(next http.Handler) http.Handler { return s.authBearer("realm", contextKeyUser, func(_ *http.Request, token string) (any, bool) { user, authenticated := s.tokenMap[token] return user, authenticated })(next) } func (s *server) authSession(next http.Handler) http.Handler { return s.authBearer("session", contextKeySession, func(r *http.Request, token string) (any, bool) { sess := s.getSessionByToken(token) if sess == nil || sess.realmID != chi.URLParam(r, "id") { return nil, false } return sess, true })(next) } func (s *server) getSessionByToken(token string) *realmSession { s.access.Lock() defer s.access.Unlock() sess := s.sessions[token] if sess == nil || sess.closed || time.Now().After(sess.expires) { return nil } return sess } func (s *server) removeSessionLocked(sess *realmSession) { if sess.closed { return } sess.closed = true close(sess.done) if s.realms[sess.realmID] == sess { delete(s.realms, sess.realmID) } if _, found := s.sessions[sess.id]; found { s.userCounts[sess.username]-- if s.userCounts[sess.username] <= 0 { delete(s.userCounts, sess.username) } } delete(s.sessions, sess.id) sess.timer.Stop() close(sess.events) for nonce, ch := range sess.pending { close(ch) delete(sess.pending, nonce) } } func (s *server) removeSession(sess *realmSession) { s.access.Lock() defer s.access.Unlock() s.removeSessionLocked(sess) } func (s *server) removeExpiredSession(sess *realmSession) bool { s.access.Lock() defer s.access.Unlock() if sess.closed || !time.Now().After(sess.expires) { return false } s.removeSessionLocked(sess) return true } func (s *server) closeAll() { s.access.Lock() defer s.access.Unlock() for _, sess := range s.sessions { s.removeSessionLocked(sess) } } func (s *server) registerPending(sess *realmSession, nonce string) (chan punchResponsePayload, bool) { s.access.Lock() defer s.access.Unlock() if sess.closed || len(sess.pending) >= maxPendingAttempts { return nil, false } if _, exists := sess.pending[nonce]; exists { return nil, false } ch := make(chan punchResponsePayload, 1) sess.pending[nonce] = ch return ch, true } func (s *server) deliverPending(sess *realmSession, nonce string, payload punchResponsePayload) bool { s.access.Lock() defer s.access.Unlock() if sess.closed { return false } ch, found := sess.pending[nonce] if !found { return false } delete(sess.pending, nonce) select { case ch <- payload: default: } return true } func (s *server) cancelPending(sess *realmSession, nonce string) { s.access.Lock() defer s.access.Unlock() delete(sess.pending, nonce) } func (s *server) sendEvent(sess *realmSession, ev realmEvent) bool { s.access.Lock() defer s.access.Unlock() if sess.closed { return false } select { case sess.events <- ev: return true default: return false } } func (s *server) handleRegister(w http.ResponseWriter, r *http.Request) { user := r.Context().Value(contextKeyUser).(*realmUser) id := chi.URLParam(r, "id") var req struct { Addresses []string `json:"addresses"` } err := render.DecodeJSON(r.Body, &req) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": "invalid json"}) return } err = validateAddresses(req.Addresses) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": err.Error()}) return } s.access.Lock() if _, exists := s.realms[id]; exists { s.access.Unlock() render.Status(r, http.StatusConflict) render.JSON(w, r, render.M{"error": "realm_taken", "message": "realm already registered"}) return } if user.maxRealms > 0 && s.userCounts[user.name] >= user.maxRealms { s.access.Unlock() render.Status(r, http.StatusTooManyRequests) render.JSON(w, r, render.M{"error": "realm_limit_reached", "message": "per-user realm limit reached"}) return } var b [16]byte _, err = rand.Read(b[:]) if err != nil { s.access.Unlock() render.Status(r, http.StatusInternalServerError) render.JSON(w, r, render.M{"error": "internal", "message": "entropy failure"}) return } sess := &realmSession{ id: hex.EncodeToString(b[:]), realmID: id, username: user.name, addresses: append([]string(nil), req.Addresses...), expires: time.Now().Add(sessionTTL), events: make(chan realmEvent, eventChannelSize), done: make(chan struct{}), pending: make(map[string]chan punchResponsePayload), } s.realms[id] = sess s.sessions[sess.id] = sess s.userCounts[user.name]++ sess.timer = time.AfterFunc(sessionTTL, func() { if s.removeExpiredSession(sess) { s.logger.Debug("[", sess.username, "] session expired realm=", sess.realmID) } }) s.access.Unlock() s.logger.InfoContext(r.Context(), "[", user.name, "] registered realm=", id) render.JSON(w, r, render.M{ "session_id": sess.id, "ttl": int(sessionTTL.Seconds()), }) } func (s *server) handleDeregister(w http.ResponseWriter, r *http.Request) { sess := r.Context().Value(contextKeySession).(*realmSession) s.logger.InfoContext(r.Context(), "[", sess.username, "] deregistered realm=", sess.realmID) s.removeSession(sess) render.NoContent(w, r) } func (s *server) handleHeartbeat(w http.ResponseWriter, r *http.Request) { sess := r.Context().Value(contextKeySession).(*realmSession) var req struct { Addresses []string `json:"addresses"` } err := render.DecodeJSON(r.Body, &req) if err != nil && !errors.Is(err, io.EOF) { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": "invalid json"}) return } if req.Addresses != nil { err = validateAddresses(req.Addresses) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": err.Error()}) return } } s.access.Lock() sess.expires = time.Now().Add(sessionTTL) if req.Addresses != nil { sess.addresses = append([]string(nil), req.Addresses...) } sess.timer.Reset(sessionTTL) s.access.Unlock() s.logger.DebugContext(r.Context(), "[", sess.username, "] heartbeat realm=", sess.realmID) s.sendEvent(sess, realmEvent{kind: "heartbeat_ack", data: render.M{"ttl": int(sessionTTL.Seconds())}}) render.JSON(w, r, render.M{"ttl": int(sessionTTL.Seconds())}) } func (s *server) handleEvents(w http.ResponseWriter, r *http.Request) { sess := r.Context().Value(contextKeySession).(*realmSession) flusher, supportsFlusher := w.(http.Flusher) if !supportsFlusher { render.Status(r, http.StatusInternalServerError) render.JSON(w, r, render.M{"error": "internal", "message": "streaming unsupported"}) return } w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") w.WriteHeader(http.StatusOK) flusher.Flush() ctx := r.Context() for { select { case <-ctx.Done(): return case ev, open := <-sess.events: if !open { return } data, _ := json.Marshal(ev.data) fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.kind, data) flusher.Flush() } } } func (s *server) handleConnect(w http.ResponseWriter, r *http.Request) { user := r.Context().Value(contextKeyUser).(*realmUser) id := chi.URLParam(r, "id") var req struct { Addresses []string `json:"addresses"` Nonce string `json:"nonce"` Obfs string `json:"obfs"` } err := render.DecodeJSON(r.Body, &req) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": "invalid json"}) return } err = validateAddresses(req.Addresses) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": err.Error()}) return } err = validateHexField("nonce", req.Nonce, nonceHexLength) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": err.Error()}) return } err = validateHexField("obfs", req.Obfs, obfsHexLength) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": err.Error()}) return } s.access.Lock() // Any authenticated realm user may connect to a registered realm. The user name // is for logging and per-user registration quota, not an ownership boundary here. sess := s.realms[id] if sess == nil || sess.closed || time.Now().After(sess.expires) { s.access.Unlock() render.Status(r, http.StatusNotFound) render.JSON(w, r, render.M{"error": "realm_not_found", "message": "realm not registered"}) return } serverAddresses := append([]string(nil), sess.addresses...) s.access.Unlock() respCh, ready := s.registerPending(sess, req.Nonce) if !ready { render.Status(r, http.StatusServiceUnavailable) render.JSON(w, r, render.M{"error": "rate_limited", "message": "too many in-flight connect attempts"}) return } defer s.cancelPending(sess, req.Nonce) if !s.sendEvent(sess, realmEvent{kind: "punch", data: punchEvent{Addresses: req.Addresses, Nonce: req.Nonce, Obfs: req.Obfs}}) { render.Status(r, http.StatusServiceUnavailable) render.JSON(w, r, render.M{"error": "rate_limited", "message": "server event buffer full"}) return } s.logger.DebugContext(r.Context(), "[", user.name, "] connect realm=", id) timer := time.NewTimer(connectResponseTimeout) defer timer.Stop() select { case payload, open := <-respCh: if !open { render.Status(r, http.StatusNotFound) render.JSON(w, r, render.M{"error": "realm_not_found", "message": "realm not registered"}) return } if len(payload.addresses) > 0 { serverAddresses = payload.addresses } case <-timer.C: case <-sess.done: render.Status(r, http.StatusNotFound) render.JSON(w, r, render.M{"error": "realm_not_found", "message": "realm not registered"}) return case <-r.Context().Done(): return } render.JSON(w, r, render.M{ "addresses": serverAddresses, "nonce": req.Nonce, "obfs": req.Obfs, }) } func (s *server) handleConnectResponse(w http.ResponseWriter, r *http.Request) { sess := r.Context().Value(contextKeySession).(*realmSession) nonce := chi.URLParam(r, "nonce") err := validateHexField("nonce", nonce, nonceHexLength) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": err.Error()}) return } var req struct { Addresses []string `json:"addresses"` } err = render.DecodeJSON(r.Body, &req) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": "invalid json"}) return } err = validateAddresses(req.Addresses) if err != nil { render.Status(r, http.StatusBadRequest) render.JSON(w, r, render.M{"error": "bad_request", "message": err.Error()}) return } delivered := s.deliverPending(sess, nonce, punchResponsePayload{addresses: append([]string(nil), req.Addresses...)}) if !delivered { render.Status(r, http.StatusNotFound) render.JSON(w, r, render.M{"error": "attempt_not_found", "message": "no pending attempt for nonce"}) return } s.logger.DebugContext(r.Context(), "[", sess.username, "] connect-response realm=", sess.realmID) render.NoContent(w, r) } func validateAddresses(addresses []string) error { if len(addresses) == 0 { return E.New("at least one address required") } if len(addresses) > maxAddresses { return E.New("too many addresses (max ", maxAddresses, ")") } for _, address := range addresses { _, err := netip.ParseAddrPort(address) if err != nil { return E.New("invalid address: ", address) } } return nil } func validateHexField(name, value string, length int) error { if len(value) != length { return E.New(name, " must be ", length, " hex characters") } _, err := hex.DecodeString(value) if err != nil { return E.New(name, " must be valid hex") } return nil }