mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-02 11:15:03 +03:00
fix(outbound): store subscription refreshes without trusting the row read
The refresh rewrote the whole row it read before a download whose length the provider controls, so an admin's delete, disable or edit made in the meantime was undone; Update had the mirror problem. Both now write only their own columns, and a refresh of a row that was deleted or re-pointed stores nothing. Around the same path: - the preview fetches without creating (and leaking) a scratch row, and Create keeps enabled=false instead of falling back to the column default; - fetch errors name the host but not the subscription URL, which is a credential and ends up in the log and in lastError; - each fetch closes its transport's idle connections; - one document loads at most 2000 links, and the cut is reported; - wireguard outbounds from a subscription always run on the userspace TUN; - editing a subscription schedules an Xray reload like delete and move do.
This commit is contained in:
@@ -565,6 +565,8 @@ func (a *XraySettingController) updateOutboundSub(c *gin.Context) {
|
||||
jsonMsg(c, "Failed to update outbound subscription", err)
|
||||
return
|
||||
}
|
||||
// enabled and prepend decide which outbounds are merged and where.
|
||||
a.XrayService.SetToNeedRestart()
|
||||
jsonObj(c, "", nil)
|
||||
}
|
||||
|
||||
@@ -630,19 +632,7 @@ func (a *XraySettingController) parseOutboundSubURL(c *gin.Context) {
|
||||
allowPrivate := c.PostForm("allowPrivate") == "true"
|
||||
allowInsecure := c.PostForm("allowInsecure") == "true"
|
||||
userAgent := c.PostForm("userAgent")
|
||||
// Use a throw-away service instance; it only needs the settingService for proxy.
|
||||
svc := service.OutboundSubscriptionService{}
|
||||
// We don't have a direct "fetch once" that returns without storing, so we
|
||||
// temporarily create a disabled row, refresh it, then delete. Cleaner would
|
||||
// be to expose a pure ParseURL on the service, but this keeps the surface small.
|
||||
tmp, err := svc.Create("preview", rawURL, "", userAgent, false, 600, allowPrivate, false, allowInsecure)
|
||||
if err != nil {
|
||||
jsonMsg(c, "Failed to preview subscription", err)
|
||||
return
|
||||
}
|
||||
obs, err := svc.Refresh(tmp.Id)
|
||||
// best-effort cleanup
|
||||
_ = svc.Delete(tmp.Id)
|
||||
obs, err := a.OutboundSubscriptionService.Preview(rawURL, userAgent, allowPrivate, allowInsecure)
|
||||
if err != nil {
|
||||
jsonMsg(c, "Failed to fetch/parse subscription", err)
|
||||
return
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -67,3 +68,32 @@ func TestWarpIntervalReportsClockPersistenceFailure(t *testing.T) {
|
||||
t.Fatalf("response omitted clock persistence error: %s", w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// enabled and prepend decide which subscription outbounds the core runs, so an edit
|
||||
// must schedule a reload like the sibling handlers do.
|
||||
func TestUpdateOutboundSubSchedulesXrayReload(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
dbtest.InitDB(t, filepath.Join(t.TempDir(), "x-ui.db"))
|
||||
sub := &model.OutboundSubscription{Url: "https://provider.example/sub", TagPrefix: "p-", Enabled: true}
|
||||
if err := database.GetDB().Create(sub).Error; err != nil {
|
||||
t.Fatalf("seed subscription: %v", err)
|
||||
}
|
||||
engine := gin.New()
|
||||
engine.Use(func(c *gin.Context) {
|
||||
c.Set("I18n", func(_ locale.I18nType, key string, _ ...string) string { return key })
|
||||
c.Next()
|
||||
})
|
||||
a := NewXraySettingController(engine.Group("/panel/api"))
|
||||
a.XrayService.IsNeedRestartAndSetFalse()
|
||||
form := url.Values{"url": {"https://1.1.1.1/sub"}, "enabled": {"false"}}
|
||||
req := httptest.NewRequest(http.MethodPost, "/panel/api/xray/outbound-subs/"+strconv.Itoa(sub.Id), strings.NewReader(form.Encode()))
|
||||
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
|
||||
rec := httptest.NewRecorder()
|
||||
engine.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), `"success":true`) {
|
||||
t.Fatalf("update = %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if !a.XrayService.IsNeedRestartAndSetFalse() {
|
||||
t.Fatalf("disabling an outbound subscription did not schedule an Xray reload")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strconv"
|
||||
@@ -171,7 +172,7 @@ func (s *OutboundSubscriptionService) nextDefaultSubPrefix(excludeId int) (strin
|
||||
func (s *OutboundSubscriptionService) Create(remark, rawURL, tagPrefix, userAgent string, enabled bool, updateInterval int, allowPrivate, prepend, allowInsecure bool) (*model.OutboundSubscription, error) {
|
||||
cleanURL, err := SanitizePublicHTTPURL(rawURL, allowPrivate)
|
||||
if err != nil {
|
||||
return nil, common.NewError("invalid subscription URL:", err)
|
||||
return nil, common.NewError("invalid subscription URL:", withoutRequestURL(err))
|
||||
}
|
||||
if cleanURL == "" {
|
||||
return nil, common.NewError("subscription URL is required")
|
||||
@@ -206,6 +207,12 @@ func (s *OutboundSubscriptionService) Create(remark, rawURL, tagPrefix, userAgen
|
||||
if err := database.GetDB().Create(sub).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// GORM leaves a zero value out of the INSERT, and the column defaults to enabled.
|
||||
if !enabled {
|
||||
if err := database.GetDB().Model(sub).Update("enabled", false).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return sub, nil
|
||||
}
|
||||
|
||||
@@ -217,7 +224,7 @@ func (s *OutboundSubscriptionService) Update(id int, remark, rawURL, tagPrefix,
|
||||
}
|
||||
cleanURL, err := SanitizePublicHTTPURL(rawURL, allowPrivate)
|
||||
if err != nil {
|
||||
return common.NewError("invalid subscription URL:", err)
|
||||
return common.NewError("invalid subscription URL:", withoutRequestURL(err))
|
||||
}
|
||||
if cleanURL == "" {
|
||||
return common.NewError("subscription URL is required")
|
||||
@@ -232,16 +239,19 @@ func (s *OutboundSubscriptionService) Update(id int, remark, rawURL, tagPrefix,
|
||||
return err
|
||||
}
|
||||
}
|
||||
sub.Remark = strings.TrimSpace(remark)
|
||||
sub.Url = cleanURL
|
||||
sub.Enabled = enabled
|
||||
sub.AllowPrivate = allowPrivate
|
||||
sub.AllowInsecure = allowInsecure
|
||||
sub.UserAgent = strings.TrimSpace(userAgent)
|
||||
sub.Prepend = prepend
|
||||
sub.TagPrefix = prefix
|
||||
sub.UpdateInterval = updateInterval
|
||||
return database.GetDB().Save(sub).Error
|
||||
// Only the editable columns: a whole-row save would put back fetch results that a
|
||||
// refresh stored meanwhile, or re-create a row deleted meanwhile.
|
||||
return database.GetDB().Model(&model.OutboundSubscription{}).Where("id = ?", sub.Id).Updates(map[string]any{
|
||||
"remark": strings.TrimSpace(remark),
|
||||
"url": cleanURL,
|
||||
"enabled": enabled,
|
||||
"allow_private": allowPrivate,
|
||||
"allow_insecure": allowInsecure,
|
||||
"user_agent": strings.TrimSpace(userAgent),
|
||||
"prepend": prepend,
|
||||
"tag_prefix": prefix,
|
||||
"update_interval": updateInterval,
|
||||
}).Error
|
||||
}
|
||||
|
||||
// Delete removes a subscription.
|
||||
@@ -332,22 +342,75 @@ func (s *OutboundSubscriptionService) subscriptionFetchClient(timeout time.Durat
|
||||
return client
|
||||
}
|
||||
|
||||
// fetchAndStore does the actual network + parse + stability + persist work.
|
||||
// subscriptionFetch is what one download of a subscription produces, before it is stored.
|
||||
type subscriptionFetch struct {
|
||||
url string
|
||||
outbounds []any
|
||||
identities string
|
||||
lastError string
|
||||
}
|
||||
|
||||
// fetchAndStore downloads a subscription and stores the result on its row.
|
||||
func (s *OutboundSubscriptionService) fetchAndStore(sub *model.OutboundSubscription) ([]any, error) {
|
||||
readURL := sub.Url
|
||||
fetched, err := s.fetchOutbounds(sub)
|
||||
if err != nil {
|
||||
s.recordError(sub, err)
|
||||
return nil, err
|
||||
}
|
||||
obsJSON, _ := json.Marshal(fetched.outbounds)
|
||||
// Only the fetch's own columns, and only while the row still points where it was read:
|
||||
// the download takes as long as the provider likes, and the admin may edit or delete meanwhile.
|
||||
res := database.GetDB().Model(&model.OutboundSubscription{}).
|
||||
Where("id = ? AND url = ?", sub.Id, readURL).
|
||||
Updates(map[string]any{
|
||||
"url": fetched.url,
|
||||
"last_fetched_outbounds": string(obsJSON),
|
||||
"link_identities": fetched.identities,
|
||||
"last_updated": time.Now().Unix(),
|
||||
"last_error": fetched.lastError,
|
||||
})
|
||||
if res.Error != nil {
|
||||
return nil, res.Error
|
||||
}
|
||||
if res.RowsAffected == 0 {
|
||||
return nil, common.NewError("subscription was deleted or pointed at another URL during the fetch")
|
||||
}
|
||||
return fetched.outbounds, nil
|
||||
}
|
||||
|
||||
// Preview downloads and parses a subscription URL without storing anything.
|
||||
func (s *OutboundSubscriptionService) Preview(rawURL, userAgent string, allowPrivate, allowInsecure bool) ([]any, error) {
|
||||
prefix, err := s.nextDefaultSubPrefix(0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
fetched, err := s.fetchOutbounds(&model.OutboundSubscription{
|
||||
Url: rawURL, UserAgent: userAgent, AllowPrivate: allowPrivate, AllowInsecure: allowInsecure, TagPrefix: prefix,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return fetched.outbounds, nil
|
||||
}
|
||||
|
||||
// fetchOutbounds does the network, parse, filter and tag-stability work for sub.
|
||||
func (s *OutboundSubscriptionService) fetchOutbounds(sub *model.OutboundSubscription) (*subscriptionFetch, error) {
|
||||
// Re-sanitize on every fetch (handles legacy rows + defense in depth against
|
||||
// any direct DB tampering). Private targets are blocked unless this
|
||||
// subscription was explicitly created with AllowPrivate.
|
||||
cleanURL, err := SanitizePublicHTTPURL(sub.Url, sub.AllowPrivate)
|
||||
if err != nil {
|
||||
s.recordError(sub, err)
|
||||
return nil, err
|
||||
return nil, withoutRequestURL(err)
|
||||
}
|
||||
if cleanURL == "" {
|
||||
return nil, common.NewError("subscription has no valid URL")
|
||||
}
|
||||
sub.Url = cleanURL // persist the cleaned version
|
||||
|
||||
client := s.subscriptionFetchClient(30*time.Second, sub.AllowInsecure)
|
||||
// The transport lives for this fetch only; left alone, the provider decides how long
|
||||
// the connections it parks stay open.
|
||||
defer client.CloseIdleConnections()
|
||||
// Re-validate every redirect hop: the initial host is checked above, but a
|
||||
// redirect could still point at a private/internal address (SSRF). Cap the
|
||||
// redirect chain as well.
|
||||
@@ -364,10 +427,9 @@ func (s *OutboundSubscriptionService) fetchAndStore(sub *model.OutboundSubscript
|
||||
}
|
||||
|
||||
reqCtx := netsafe.ContextWithAllowPrivate(context.Background(), sub.AllowPrivate)
|
||||
req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, sub.Url, nil)
|
||||
req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, cleanURL, nil)
|
||||
if err != nil {
|
||||
s.recordError(sub, err)
|
||||
return nil, err
|
||||
return nil, withoutRequestURL(err)
|
||||
}
|
||||
userAgent := strings.TrimSpace(sub.UserAgent)
|
||||
if userAgent == "" {
|
||||
@@ -381,25 +443,20 @@ func (s *OutboundSubscriptionService) fetchAndStore(sub *model.OutboundSubscript
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
s.recordError(sub, err)
|
||||
return nil, err
|
||||
return nil, withoutRequestURL(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
err := fmt.Errorf("http %d", resp.StatusCode)
|
||||
s.recordError(sub, err)
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("http %d", resp.StatusCode)
|
||||
}
|
||||
body, err := readBoundedOutboundSubscriptionBody(resp.Body)
|
||||
if err != nil {
|
||||
s.recordError(sub, err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
parsed, identities, err := link.ParseSubscriptionBody(body)
|
||||
if err != nil {
|
||||
s.recordError(sub, err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -436,6 +493,7 @@ func (s *OutboundSubscriptionService) fetchAndStore(sub *model.OutboundSubscript
|
||||
dropped = append(dropped, fmt.Sprintf("%.64s: %s", ob["tag"], reason))
|
||||
continue
|
||||
}
|
||||
forceUserspaceWireguard(map[string]any(ob))
|
||||
if _, rejected := filterOutboundsRejectedByCore(fmt.Sprintf("outbound sub %d", sub.Id), []any{map[string]any(ob)}); len(rejected) > 0 {
|
||||
dropped = append(dropped, rejected...)
|
||||
continue
|
||||
@@ -462,22 +520,45 @@ func (s *OutboundSubscriptionService) fetchAndStore(sub *model.OutboundSubscript
|
||||
kept[i] = map[string]any(keptLinks[i])
|
||||
}
|
||||
|
||||
// Persist the outbounds (as compact JSON array)
|
||||
obsJSON, _ := json.Marshal(kept)
|
||||
|
||||
sub.LastFetchedOutbounds = string(obsJSON)
|
||||
sub.LinkIdentities = string(identJSON)
|
||||
sub.LastUpdated = time.Now().Unix()
|
||||
sub.LastError = ""
|
||||
fetched := &subscriptionFetch{url: cleanURL, outbounds: kept, identities: string(identJSON)}
|
||||
if len(dropped) > 0 {
|
||||
sub.LastError = fmt.Sprintf("dropped %d outbound(s): %s", len(dropped), dropped[0])
|
||||
fetched.lastError = fmt.Sprintf("dropped %d outbound(s): %s", len(dropped), dropped[0])
|
||||
}
|
||||
|
||||
if err := database.GetDB().Save(sub).Error; err != nil {
|
||||
return nil, err
|
||||
if len(parsed) >= link.MaxSubscriptionOutbounds {
|
||||
fetched.lastError = strings.TrimPrefix(fetched.lastError+"; ", "; ") +
|
||||
fmt.Sprintf("only the first %d links of the document are loaded", link.MaxSubscriptionOutbounds)
|
||||
}
|
||||
return fetched, nil
|
||||
}
|
||||
|
||||
return kept, nil
|
||||
// withoutRequestURL keeps only the host of the URL net/url and net/http quote in their
|
||||
// errors: a subscription URL is a credential, and these errors are logged and stored.
|
||||
func withoutRequestURL(err error) error {
|
||||
var urlErr *url.Error
|
||||
if !errors.As(err, &urlErr) {
|
||||
return err
|
||||
}
|
||||
if u, parseErr := url.Parse(urlErr.URL); parseErr == nil && u.Host != "" {
|
||||
return fmt.Errorf("%s %s: %w", urlErr.Op, u.Host, urlErr.Err)
|
||||
}
|
||||
return fmt.Errorf("%s: %w", urlErr.Op, urlErr.Err)
|
||||
}
|
||||
|
||||
// forceUserspaceWireguard keeps a provider's wireguard fields off the host: the kernel
|
||||
// TUN path would install its addresses and routes on the panel's machine.
|
||||
func forceUserspaceWireguard(outbound any) {
|
||||
ob, ok := outbound.(map[string]any)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if protocol, _ := ob["protocol"].(string); !strings.EqualFold(protocol, "wireguard") {
|
||||
return
|
||||
}
|
||||
if settings, ok := ob["settings"].(map[string]any); ok {
|
||||
settings["noKernelTun"] = true
|
||||
} else {
|
||||
ob["settings"] = map[string]any{"noKernelTun": true}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *OutboundSubscriptionService) recordError(sub *model.OutboundSubscription, err error) {
|
||||
@@ -587,6 +668,9 @@ func (s *OutboundSubscriptionService) activeOutboundsSplit() (prepend []any, app
|
||||
}
|
||||
return reason != ""
|
||||
})
|
||||
for _, ob := range arr {
|
||||
forceUserspaceWireguard(ob)
|
||||
}
|
||||
arr, _ = filterOutboundsRejectedByCore(fmt.Sprintf("outbound sub %d", sub.Id), arr)
|
||||
if sub.Prepend {
|
||||
prepend = append(prepend, arr...)
|
||||
|
||||
@@ -6,12 +6,16 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
@@ -504,3 +508,190 @@ func TestNamesInternalHost(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The provider decides when its answer arrives, so an admin's delete or disable in the
|
||||
// meantime must survive the refresh that read the row before the download.
|
||||
func TestOutboundSubscriptionRefreshKeepsAdminChangesMadeDuringTheFetch(t *testing.T) {
|
||||
setupSettingTestDB(t)
|
||||
const body = "vless://00000000-0000-4000-8000-000000000000@8.8.8.8:443?security=tls&type=tcp#B"
|
||||
var during func(id int)
|
||||
var subID int
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
during(subID)
|
||||
_, _ = w.Write([]byte(body))
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
svc := &OutboundSubscriptionService{}
|
||||
create := func() {
|
||||
t.Helper()
|
||||
sub := &model.OutboundSubscription{Url: server.URL, AllowPrivate: true, TagPrefix: "p-", Enabled: true}
|
||||
if err := database.GetDB().Create(sub).Error; err != nil {
|
||||
t.Fatalf("seed subscription: %v", err)
|
||||
}
|
||||
subID = sub.Id
|
||||
}
|
||||
|
||||
create()
|
||||
during = func(id int) { _ = svc.Delete(id) }
|
||||
if _, err := svc.Refresh(subID); err == nil {
|
||||
t.Fatalf("Refresh of a subscription deleted mid-fetch succeeded")
|
||||
}
|
||||
var count int64
|
||||
database.GetDB().Model(&model.OutboundSubscription{}).Where("id = ?", subID).Count(&count)
|
||||
if count != 0 {
|
||||
t.Fatalf("subscription deleted during the fetch was re-created")
|
||||
}
|
||||
|
||||
create()
|
||||
during = func(id int) {
|
||||
database.GetDB().Model(&model.OutboundSubscription{}).Where("id = ?", id).Update("enabled", false)
|
||||
}
|
||||
if _, err := svc.Refresh(subID); err != nil {
|
||||
t.Fatalf("Refresh: %v", err)
|
||||
}
|
||||
got, err := svc.Get(subID)
|
||||
if err != nil || got.Enabled || !strings.Contains(got.LastFetchedOutbounds, "8.8.8.8") {
|
||||
t.Fatalf("after disable mid-fetch: enabled=%v outbounds=%q (%v), want disabled with the fetch stored", got.Enabled, got.LastFetchedOutbounds, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A subscription URL is the credential for the provider account; fetch errors are
|
||||
// logged and stored, so they may name the host but never the path or query.
|
||||
func TestOutboundSubscriptionFetchErrorsKeepTheURLSecret(t *testing.T) {
|
||||
setupSettingTestDB(t)
|
||||
closed := httptest.NewServer(http.NotFoundHandler())
|
||||
closed.Close()
|
||||
const token = "s3cr3t-token"
|
||||
sub := &model.OutboundSubscription{Url: closed.URL + "/sub/" + token + "?key=" + token, AllowPrivate: true, TagPrefix: "p-"}
|
||||
if err := database.GetDB().Create(sub).Error; err != nil {
|
||||
t.Fatalf("seed subscription: %v", err)
|
||||
}
|
||||
_, err := (&OutboundSubscriptionService{}).Refresh(sub.Id)
|
||||
if err == nil || strings.Contains(err.Error(), token) {
|
||||
t.Fatalf("Refresh error = %v, want a failure that does not quote the URL", err)
|
||||
}
|
||||
stored, _ := (&OutboundSubscriptionService{}).Get(sub.Id)
|
||||
if stored.LastError == "" || strings.Contains(stored.LastError, token) {
|
||||
t.Fatalf("lastError = %q, want a failure that does not quote the URL", stored.LastError)
|
||||
}
|
||||
_, err = (&OutboundSubscriptionService{}).Create("r", "https://provider.example/sub/"+token+"%zz", "", "", true, 600, false, false, false)
|
||||
if err == nil || strings.Contains(err.Error(), token) {
|
||||
t.Fatalf("Create error = %v, want a parse failure that does not quote the URL", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOutboundSubscriptionPreviewStoresNothing(t *testing.T) {
|
||||
setupSettingTestDB(t)
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte("vless://00000000-0000-4000-8000-000000000000@8.8.8.8:443?security=tls&type=tcp#B"))
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
obs, err := (&OutboundSubscriptionService{}).Preview(server.URL, "", true, false)
|
||||
if err != nil || len(obs) != 1 {
|
||||
t.Fatalf("Preview = %v, %v; want the one outbound", obs, err)
|
||||
}
|
||||
var count int64
|
||||
database.GetDB().Model(&model.OutboundSubscription{}).Count(&count)
|
||||
if count != 0 {
|
||||
t.Fatalf("Preview left %d subscription row(s) behind", count)
|
||||
}
|
||||
}
|
||||
|
||||
// Each fetch builds its own transport; whatever it leaves parked would otherwise live
|
||||
// for as long as the provider keeps the connection open.
|
||||
func TestOutboundSubscriptionRefreshLeavesNoIdleConnection(t *testing.T) {
|
||||
setupSettingTestDB(t)
|
||||
var open atomic.Int32
|
||||
server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte("vless://00000000-0000-4000-8000-000000000000@8.8.8.8:443?security=tls&type=tcp#B"))
|
||||
}))
|
||||
server.Config.ConnState = func(_ net.Conn, state http.ConnState) {
|
||||
switch state {
|
||||
case http.StateNew:
|
||||
open.Add(1)
|
||||
case http.StateClosed, http.StateHijacked:
|
||||
open.Add(-1)
|
||||
}
|
||||
}
|
||||
server.Start()
|
||||
t.Cleanup(server.Close)
|
||||
sub := &model.OutboundSubscription{Url: server.URL, AllowPrivate: true, TagPrefix: "p-"}
|
||||
if err := database.GetDB().Create(sub).Error; err != nil {
|
||||
t.Fatalf("seed subscription: %v", err)
|
||||
}
|
||||
for range 3 {
|
||||
if _, err := (&OutboundSubscriptionService{}).Refresh(sub.Id); err != nil {
|
||||
t.Fatalf("Refresh: %v", err)
|
||||
}
|
||||
}
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for open.Load() != 0 && time.Now().Before(deadline) {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
if n := open.Load(); n != 0 {
|
||||
t.Fatalf("%d connection(s) to the provider still open after the refreshes", n)
|
||||
}
|
||||
}
|
||||
|
||||
// One document must not turn into an unbounded number of outbounds: each is held,
|
||||
// validated and merged into the core's config.
|
||||
func TestOutboundSubscriptionRefreshCapsTheNumberOfLinks(t *testing.T) {
|
||||
setupSettingTestDB(t)
|
||||
subID := serveOutboundSubscription(t, "p-", func(int) string {
|
||||
return strings.Repeat("ss://YWVzLTEyOC1nY206cA@8.8.8.8:8388#s\n", link.MaxSubscriptionOutbounds+5)
|
||||
})
|
||||
obs, err := (&OutboundSubscriptionService{}).Refresh(subID)
|
||||
if err != nil {
|
||||
t.Fatalf("Refresh: %v", err)
|
||||
}
|
||||
if len(obs) != link.MaxSubscriptionOutbounds {
|
||||
t.Fatalf("stored %d outbounds, want the cap of %d", len(obs), link.MaxSubscriptionOutbounds)
|
||||
}
|
||||
sub, _ := (&OutboundSubscriptionService{}).Get(subID)
|
||||
if !strings.Contains(sub.LastError, strconv.Itoa(link.MaxSubscriptionOutbounds)) {
|
||||
t.Fatalf("lastError = %q, want the cut reported", sub.LastError)
|
||||
}
|
||||
}
|
||||
|
||||
// A provider's wireguard fields must not configure the host: the kernel TUN path
|
||||
// installs its addresses and routes, so subscription outbounds stay in userspace.
|
||||
func TestOutboundSubscriptionWireguardStaysOffKernelTun(t *testing.T) {
|
||||
setupSettingTestDB(t)
|
||||
subID := serveOutboundSubscription(t, "p-", func(int) string {
|
||||
return "wireguard://JOSLNr3Kk8BViwrrTJ2M19gRBbGttnRfApY3nQ3fdT8%3D@8.8.8.8:51820?publickey=biSnuvUOX2pmtzf249FK8UGSq2PQslGSdoj4UqH4yew%3D&address=10.0.0.2%2F32#wg"
|
||||
})
|
||||
obs, err := (&OutboundSubscriptionService{}).Refresh(subID)
|
||||
if err != nil || len(obs) != 1 {
|
||||
t.Fatalf("Refresh = %v, %v", obs, err)
|
||||
}
|
||||
stored := `[{"tag":"p-old","protocol":"wireguard","settings":{"secretKey":"JOSLNr3Kk8BViwrrTJ2M19gRBbGttnRfApY3nQ3fdT8=","address":["10.0.0.2/32"],"peers":[{"publicKey":"biSnuvUOX2pmtzf249FK8UGSq2PQslGSdoj4UqH4yew=","endpoint":"8.8.8.8:51820"}]}}]`
|
||||
old := &model.OutboundSubscription{Url: "https://provider.example/sub", TagPrefix: "q-", Enabled: true, LastFetchedOutbounds: stored}
|
||||
if err := database.GetDB().Create(old).Error; err != nil {
|
||||
t.Fatalf("seed subscription: %v", err)
|
||||
}
|
||||
if err := database.GetDB().Model(&model.OutboundSubscription{}).Where("id = ?", subID).Update("enabled", true).Error; err != nil {
|
||||
t.Fatalf("enable: %v", err)
|
||||
}
|
||||
active, err := (&OutboundSubscriptionService{}).AllActiveOutbounds()
|
||||
if err != nil || len(active) != 2 {
|
||||
t.Fatalf("AllActiveOutbounds = %v, %v", active, err)
|
||||
}
|
||||
for _, ob := range append(obs, active...) {
|
||||
settings, _ := ob.(map[string]any)["settings"].(map[string]any)
|
||||
if settings["noKernelTun"] != true {
|
||||
t.Fatalf("subscription wireguard outbound without noKernelTun: %v", ob)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestOutboundSubscriptionCreateKeepsItDisabled(t *testing.T) {
|
||||
setupSettingTestDB(t)
|
||||
sub, err := (&OutboundSubscriptionService{}).Create("r", "https://1.1.1.1/sub", "", "", false, 600, false, false, false)
|
||||
if err != nil {
|
||||
t.Fatalf("Create: %v", err)
|
||||
}
|
||||
stored, err := (&OutboundSubscriptionService{}).Get(sub.Id)
|
||||
if err != nil || stored.Enabled {
|
||||
t.Fatalf("stored enabled = %v (%v), want the subscription created disabled", stored.Enabled, err)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user