feat(mtproto): enforce per-client quota & expiry via mtg-multi limits

Map each mtproto client's totalGB and expiryTime onto mtg-multi's new
[secret-limits] (quota/expires): emit them into the generated config and
hot-apply through PUT /secrets so live connections survive. Quota is
written as an exact "<n>B" byte count that round-trips through both the
config and API parsers without the precision loss of a base-2 unit.

The sidecar's quota counter is not pruned when a secret is dropped, so a
panel-side traffic reset re-pushes the client's secret and then calls
POST /secrets/{name}/reset-quota (wired into every reset path) so a
renewed client is not immediately re-blocked.

Resolve the mtg-multi binary from the fork's latest release tag in
DockerInit.sh and release.yml instead of a hardcoded version pin, so the
panel no longer needs a manual bump per fork release.
This commit is contained in:
MHSanaei
2026-07-08 15:30:56 +02:00
parent 61e12e4c29
commit 328d920e98
8 changed files with 251 additions and 28 deletions
+9 -4
View File
@@ -158,8 +158,11 @@ jobs:
mv xray xray-linux-${{ matrix.platform }}
# mtg-multi (MTProto sidecar) ships prebuilt release binaries whose
# platform labels match our matrix, so download and unpack the matching
# archive. Only the platforms the fork publishes are packaged.
MTG_MULTI_VER="v1.14.0"
# archive. Only the platforms the fork publishes are packaged. The tag
# is resolved from the fork's latest release so it never needs bumping
# here; the token only lifts the API rate limit for a public read.
MTG_MULTI_VER=$(curl -sfL -H "Authorization: Bearer ${{ secrets.GITHUB_TOKEN }}" "https://api.github.com/repos/mhsanaei/mtg-multi/releases/latest" | sed -n 's/.*"tag_name": *"\([^"]*\)".*/\1/p' | head -n 1)
if [ -z "$MTG_MULTI_VER" ]; then echo "could not resolve the latest mtg-multi release tag"; exit 1; fi
case "${{ matrix.platform }}" in
amd64|arm64|armv7|armv6|386)
MTG_PKG="mtg-multi-${MTG_MULTI_VER#v}-linux-${{ matrix.platform }}"
@@ -284,8 +287,10 @@ jobs:
Rename-Item xray.exe xray-windows-amd64.exe
# mtg-multi (MTProto sidecar) publishes a prebuilt Windows binary, so
# download and unpack it instead of compiling.
$MTG_MULTI_VER = "v1.14.0"
# download and unpack it instead of compiling. The tag tracks the
# fork's latest release so it never needs bumping here.
$MTG_MULTI_VER = (Invoke-RestMethod -Uri "https://api.github.com/repos/mhsanaei/mtg-multi/releases/latest" -Headers @{ Authorization = "Bearer ${{ secrets.GITHUB_TOKEN }}"; "User-Agent" = "x-ui-release" }).tag_name
if (-not $MTG_MULTI_VER) { throw "could not resolve the latest mtg-multi release tag" }
$MTG_PKG = "mtg-multi-$($MTG_MULTI_VER.TrimStart('v'))-windows-amd64"
curl.exe -sfLRO "https://github.com/mhsanaei/mtg-multi/releases/download/$MTG_MULTI_VER/$MTG_PKG.zip"
Expand-Archive -Path "$MTG_PKG.zip" -DestinationPath "mtg-tmp" -Force
+7 -4
View File
@@ -15,10 +15,13 @@ file locations when it can answer in one hop.
(`github.com/mhsanaei/mtg-multi`, a multi-secret fork built from source;
`internal/mtproto/`) — outside Xray, one process per inbound serving each
client's FakeTLS secret via the fork's `[secrets]` section (plus per-client
ad-tags via `[secret-ad-tags]`). Client and ad-tag edits are hot-applied
through the fork's management API (`PUT /secrets`, bearer-token guarded) so
connections survive; the manager falls back to a process restart on older
binaries.
ad-tags via `[secret-ad-tags]` and per-client data quota / expiry via
`[secret-limits]`, mapped from the client's `totalGB`/`expiryTime`). Client,
ad-tag and quota/expiry edits are hot-applied through the fork's management API
(`PUT /secrets`, bearer-token guarded) so connections survive; the manager
falls back to a process restart on older binaries. A client's panel-side
traffic reset also calls `POST /secrets/{name}/reset-quota` so a renewed client
is not re-blocked by the sidecar's quota counter.
- Storage: SQLite by default (`/etc/x-ui/x-ui.db` on Linux; the executable dir on
Windows), PostgreSQL optional (`XUI_DB_TYPE` / `XUI_DB_DSN`). The CGo SQLite
driver (`mattn/go-sqlite3`) needs a C compiler — `CGO_ENABLED=0` builds fail.
+5 -1
View File
@@ -25,7 +25,11 @@ case $1 in
FNAME="amd64"
;;
esac
MTG_MULTI_VER="v1.14.0"
MTG_MULTI_VER=$(curl -sfL "https://api.github.com/repos/mhsanaei/mtg-multi/releases/latest" | sed -n 's/.*"tag_name": *"\([^"]*\)".*/\1/p' | head -n 1)
if [ -z "$MTG_MULTI_VER" ]; then
echo "DockerInit: could not resolve the latest mtg-multi release tag" >&2
exit 1
fi
mkdir -p build/bin
cd build/bin
curl -sfLRO "https://github.com/XTLS/Xray-core/releases/download/v26.6.27/Xray-linux-${ARCH}.zip"
+102 -14
View File
@@ -9,6 +9,7 @@ import (
"fmt"
"net"
"net/http"
"net/url"
"os"
"slices"
"strconv"
@@ -26,9 +27,11 @@ import (
// client's own advertising-tag override, emitted into the [secret-ad-tags]
// section; empty falls back to the instance-level tag.
type SecretEntry struct {
Name string
Secret string
AdTag string
Name string
Secret string
AdTag string
QuotaBytes int64
ExpiresUnix int64
}
// Instance is the desired runtime configuration of one mtproto inbound. A single
@@ -100,12 +103,12 @@ func (inst Instance) structuralFingerprint() string {
// secretsFingerprint identifies the reloadable secret config regardless of
// client order, so a reordered clients array in the stored settings does not
// read as a change. It moves whenever a client is added, removed, disabled,
// re-keyed, or re-tagged — all of which mtg applies in place without dropping
// connections.
// re-keyed, re-tagged, or re-limited (quota/expiry) — all of which mtg applies
// in place without dropping connections.
func (inst Instance) secretsFingerprint() string {
pairs := make([]string, 0, len(inst.Secrets))
for _, e := range inst.Secrets {
pairs = append(pairs, e.Name+"="+e.Secret+";tag="+e.AdTag)
pairs = append(pairs, fmt.Sprintf("%s=%s;tag=%s;q=%d;exp=%d", e.Name, e.Secret, e.AdTag, e.QuotaBytes, e.ExpiresUnix))
}
slices.Sort(pairs)
return strings.Join(pairs, "|")
@@ -182,10 +185,12 @@ func InstanceFromInbound(ib *model.Inbound) (Instance, bool) {
PublicIPv4 string `json:"publicIpv4"`
PublicIPv6 string `json:"publicIpv6"`
Clients []struct {
Email string `json:"email"`
Secret string `json:"secret"`
AdTag string `json:"adTag"`
Enable bool `json:"enable"`
Email string `json:"email"`
Secret string `json:"secret"`
AdTag string `json:"adTag"`
Enable bool `json:"enable"`
TotalGB int64 `json:"totalGB"`
ExpiryTime int64 `json:"expiryTime"`
} `json:"clients"`
}
if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
@@ -196,7 +201,14 @@ func InstanceFromInbound(ib *model.Inbound) (Instance, bool) {
if !c.Enable || c.Secret == "" || c.Email == "" {
continue
}
secrets = append(secrets, SecretEntry{Name: c.Email, Secret: c.Secret, AdTag: usableAdTag(c.AdTag)})
entry := SecretEntry{Name: c.Email, Secret: c.Secret, AdTag: usableAdTag(c.AdTag)}
if c.TotalGB > 0 {
entry.QuotaBytes = c.TotalGB
}
if c.ExpiryTime > 0 {
entry.ExpiresUnix = c.ExpiryTime / 1000
}
secrets = append(secrets, entry)
}
if len(secrets) == 0 {
return Instance{}, false
@@ -449,6 +461,53 @@ func (m *Manager) CollectTraffic() ([]Traffic, []string) {
return out, online
}
func (m *Manager) HasRunning() bool {
m.mu.Lock()
defer m.mu.Unlock()
for _, cur := range m.procs {
if cur.proc != nil && cur.proc.IsRunning() {
return true
}
}
return false
}
func (m *Manager) ResetQuota(email string) {
if email == "" {
return
}
type target struct {
port int
token string
}
m.mu.Lock()
targets := make([]target, 0, len(m.procs))
for _, cur := range m.procs {
if cur.proc != nil && cur.proc.IsRunning() {
targets = append(targets, target{cur.apiPort, cur.apiToken})
}
}
m.mu.Unlock()
for _, t := range targets {
resetQuota(t.port, t.token, email)
}
}
func resetQuota(port int, token, email string) {
client := http.Client{Timeout: 3 * time.Second}
endpoint := fmt.Sprintf("http://127.0.0.1:%d/secrets/%s/reset-quota", port, url.PathEscape(email))
req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, endpoint, nil)
if err != nil {
return
}
authorize(req, token)
resp, err := client.Do(req)
if err != nil {
return
}
_ = resp.Body.Close()
}
// FreeLocalPort asks the OS for an unused loopback TCP port. It is used both
// for mtg's /stats API endpoint and to allocate the per-inbound SOCKS egress
// bridge port persisted into mtproto inbound settings.
@@ -525,6 +584,18 @@ func renderConfig(inst Instance, apiPort int, apiToken string) string {
}
fmt.Fprintf(&b, "%q = %q\n", e.Name, e.AdTag)
}
for _, e := range inst.Secrets {
if e.QuotaBytes <= 0 && e.ExpiresUnix <= 0 {
continue
}
fmt.Fprintf(&b, "\n[secret-limits.%q]\n", e.Name)
if e.QuotaBytes > 0 {
fmt.Fprintf(&b, "quota = %q\n", quotaString(e.QuotaBytes))
}
if e.ExpiresUnix > 0 {
fmt.Fprintf(&b, "expires = %q\n", expiresString(e.ExpiresUnix))
}
}
b.WriteString("\n[secrets]\n")
for _, e := range inst.Secrets {
fmt.Fprintf(&b, "%q = %q\n", e.Name, e.Secret)
@@ -549,8 +620,10 @@ type statsUser struct {
}
type secretPutEntry struct {
Secret string `json:"secret"`
AdTag string `json:"ad_tag,omitempty"`
Secret string `json:"secret"`
AdTag string `json:"ad_tag,omitempty"`
Quota string `json:"quota,omitempty"`
Expires string `json:"expires,omitempty"`
}
type secretsPutBody struct {
@@ -560,11 +633,26 @@ type secretsPutBody struct {
func secretsPayload(inst Instance) secretsPutBody {
secrets := make(map[string]secretPutEntry, len(inst.Secrets))
for _, e := range inst.Secrets {
secrets[e.Name] = secretPutEntry{Secret: e.Secret, AdTag: e.AdTag}
entry := secretPutEntry{Secret: e.Secret, AdTag: e.AdTag}
if e.QuotaBytes > 0 {
entry.Quota = quotaString(e.QuotaBytes)
}
if e.ExpiresUnix > 0 {
entry.Expires = expiresString(e.ExpiresUnix)
}
secrets[e.Name] = entry
}
return secretsPutBody{Secrets: secrets}
}
func quotaString(bytes int64) string {
return strconv.FormatInt(bytes, 10) + "B"
}
func expiresString(unix int64) string {
return time.Unix(unix, 0).UTC().Format(time.RFC3339)
}
// newAPIToken mints the bearer token one mtg process and its manager share for
// the lifetime of that process. The management API can replace the whole
// secret set, so even though it only listens on loopback it must not be open
+58 -2
View File
@@ -22,7 +22,7 @@ func TestInstanceFromInbound(t *testing.T) {
`"routeThroughXray":true,"routeXrayPort":50000,` +
`"adTag":" 0123456789abcdef0123456789abcdef ",` +
`"clients":[` +
`{"email":"alice","secret":"` + aliceSecret + `","adTag":"fedcba9876543210fedcba9876543210","enable":true},` +
`{"email":"alice","secret":"` + aliceSecret + `","adTag":"fedcba9876543210fedcba9876543210","enable":true,"totalGB":1073741824,"expiryTime":1893456000000},` +
`{"email":"bob","secret":"","enable":true},` +
`{"email":"carol","secret":"eeaa","enable":false}]}`,
}
@@ -42,6 +42,12 @@ func TestInstanceFromInbound(t *testing.T) {
if inst.Secrets[0].AdTag != "fedcba9876543210fedcba9876543210" {
t.Fatalf("the client ad-tag must be parsed, got %q", inst.Secrets[0].AdTag)
}
if inst.Secrets[0].QuotaBytes != 1073741824 {
t.Fatalf("totalGB must map to the byte quota, got %d", inst.Secrets[0].QuotaBytes)
}
if inst.Secrets[0].ExpiresUnix != 1893456000 {
t.Fatalf("expiryTime (ms) must map to a unix-second deadline, got %d", inst.Secrets[0].ExpiresUnix)
}
if inst.Port != 8443 || inst.Id != 3 {
t.Fatalf("bad instance %+v", inst)
}
@@ -186,6 +192,54 @@ func TestRenderConfigXrayEgress(t *testing.T) {
}
}
func TestRenderConfigSecretLimits(t *testing.T) {
cfg := renderConfig(Instance{
Secrets: []SecretEntry{
{Name: "alice", Secret: "ee11", QuotaBytes: 1073741824, ExpiresUnix: 1893456000},
{Name: "bob", Secret: "ee22", QuotaBytes: 500},
{Name: "carol", Secret: "ee33"},
},
Listen: "0.0.0.0", Port: 443,
}, 6000, "")
for _, want := range []string{
"[secret-limits.\"alice\"]",
`quota = "1073741824B"`,
`expires = "2030-01-01T00:00:00Z"`,
"[secret-limits.\"bob\"]",
`quota = "500B"`,
} {
if !strings.Contains(cfg, want) {
t.Fatalf("limits config missing %q:\n%s", want, cfg)
}
}
if strings.Contains(cfg, "[secret-limits.\"carol\"]") {
t.Fatalf("a client without a limit must not get a table:\n%s", cfg)
}
if strings.LastIndex(cfg, "[secrets]") < strings.Index(cfg, "[secret-limits.\"alice\"]") {
t.Fatalf("[secret-limits] must precede the final [secrets] section:\n%s", cfg)
}
if strings.Contains(cfg, "usage-state-file") {
t.Fatalf("no usage-state-file must be emitted; quota is an in-memory guard:\n%s", cfg)
}
}
func TestSecretsPayloadLimits(t *testing.T) {
body := secretsPayload(Instance{Secrets: []SecretEntry{
{Name: "alice", Secret: "ee11", QuotaBytes: 1073741824, ExpiresUnix: 1893456000},
{Name: "bob", Secret: "ee22"},
}})
if got := body.Secrets["alice"].Quota; got != "1073741824B" {
t.Fatalf("quota must be sent as an exact byte count, got %q", got)
}
if got := body.Secrets["alice"].Expires; got != "2030-01-01T00:00:00Z" {
t.Fatalf("expiry must be sent as RFC3339, got %q", got)
}
if body.Secrets["bob"].Quota != "" || body.Secrets["bob"].Expires != "" {
t.Fatalf("an unlimited client must carry no quota/expiry: %+v", body.Secrets["bob"])
}
}
func TestFingerprintSplit(t *testing.T) {
base := Instance{Secrets: []SecretEntry{{Name: "a", Secret: "ee"}}, Listen: "0.0.0.0", Port: 443}
@@ -224,6 +278,8 @@ func TestFingerprintSplit(t *testing.T) {
"clientAdTag": func(i *Instance) {
i.Secrets = []SecretEntry{{Name: "a", Secret: "ee", AdTag: "0123456789abcdef0123456789abcdef"}}
},
"quota": func(i *Instance) { i.Secrets = []SecretEntry{{Name: "a", Secret: "ee", QuotaBytes: 1 << 30}} },
"expiry": func(i *Instance) { i.Secrets = []SecretEntry{{Name: "a", Secret: "ee", ExpiresUnix: 1893456000}} },
} {
t.Run("secrets/"+name, func(t *testing.T) {
changed := base
@@ -241,7 +297,7 @@ func TestFingerprintSplit(t *testing.T) {
t.Run("orderInsensitive", func(t *testing.T) {
forward := Instance{Secrets: []SecretEntry{{Name: "alice", Secret: "ee11"}, {Name: "bob", Secret: "ee22"}}}
reversed := Instance{Secrets: []SecretEntry{{Name: "bob", Secret: "ee22"}, {Name: "alice", Secret: "ee11"}}}
if got, want := forward.secretsFingerprint(), "alice=ee11;tag=|bob=ee22;tag="; got != want {
if got, want := forward.secretsFingerprint(), "alice=ee11;tag=;q=0;exp=0|bob=ee22;tag=;q=0;exp=0"; got != want {
t.Fatalf("secrets fingerprint must join sorted pairs: got %q, want %q", got, want)
}
if forward.secretsFingerprint() != reversed.secretsFingerprint() {
+5 -1
View File
@@ -122,9 +122,13 @@ func (s *ClientService) BulkResetTraffic(inboundSvc *InboundService, emails []st
}
func (s *ClientService) ResetAllClientTraffics(inboundSvc *InboundService, id int) error {
return submitTrafficWrite(func() error {
err := submitTrafficWrite(func() error {
return s.resetAllClientTrafficsLocked(id)
})
if err == nil {
inboundSvc.resetAllMtprotoQuotas()
}
return err
}
func (s *ClientService) resetAllClientTrafficsLocked(id int) error {
+52
View File
@@ -103,3 +103,55 @@ func (s *InboundService) applyLocalMtproto(inboundId int) {
logger.Debug("mtproto: immediate client apply failed for inbound", inboundId, ":", err)
}
}
func (s *InboundService) resetMtprotoClientQuota(email string) {
mgr := mtproto.GetManager()
if !mgr.HasRunning() {
return
}
id, ok := s.localMtprotoInboundIdForEmail(email)
if !ok {
return
}
s.applyLocalMtproto(id)
mgr.ResetQuota(email)
}
func (s *InboundService) resetAllMtprotoQuotas() {
mgr := mtproto.GetManager()
if !mgr.HasRunning() {
return
}
desired, err := s.DesiredMtprotoInstances()
if err != nil {
return
}
mgr.Reconcile(desired)
for _, inst := range desired {
for _, sec := range inst.Secrets {
mgr.ResetQuota(sec.Name)
}
}
}
func (s *InboundService) localMtprotoInboundIdForEmail(email string) (int, bool) {
db := database.GetDB()
var inbounds []*model.Inbound
if err := db.Model(model.Inbound{}).
Where("protocol = ? AND node_id IS NULL", model.MTProto).
Find(&inbounds).Error; err != nil {
return 0, false
}
for _, ib := range inbounds {
inst, ok := mtproto.InstanceFromInbound(ib)
if !ok {
continue
}
for _, sec := range inst.Secrets {
if sec.Name == email {
return ib.Id, true
}
}
}
return 0, false
}
+13 -2
View File
@@ -509,7 +509,7 @@ func (s *InboundService) delClientStatsByEmails(tx *gorm.DB, emails []string) er
}
func (s *InboundService) ResetClientTrafficByEmail(clientEmail string) error {
return submitTrafficWrite(func() error {
err := submitTrafficWrite(func() error {
return database.GetDB().Transaction(func(tx *gorm.DB) error {
if err := adjustGroupBaselinesForRemovedTraffic(tx, []string{clientEmail}); err != nil {
return err
@@ -525,6 +525,10 @@ func (s *InboundService) ResetClientTrafficByEmail(clientEmail string) error {
return tx.Where("email = ?", clientEmail).Delete(&model.NodeClientTraffic{}).Error
})
})
if err == nil {
s.resetMtprotoClientQuota(clientEmail)
}
return err
}
func (s *InboundService) ResetClientTraffic(id int, clientEmail string) (needRestart bool, err error) {
@@ -533,6 +537,9 @@ func (s *InboundService) ResetClientTraffic(id int, clientEmail string) (needRes
needRestart, inner = s.resetClientTrafficLocked(id, clientEmail)
return inner
})
if err == nil {
s.resetMtprotoClientQuota(clientEmail)
}
return
}
@@ -646,9 +653,13 @@ func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (b
}
func (s *InboundService) ResetAllTraffics() error {
return submitTrafficWrite(func() error {
err := submitTrafficWrite(func() error {
return s.resetAllTrafficsLocked()
})
if err == nil {
s.resetAllMtprotoQuotas()
}
return err
}
func (s *InboundService) resetAllTrafficsLocked() error {