feat(mtproto): adopt dolonet/mtg-multi and make MTProto inbounds multi-client

Replace the upstream 9seconds/mtg sidecar with the dolonet/mtg-multi fork so a single MTProto inbound can serve many per-user secrets. Each panel client is now one named FakeTLS secret in the fork's [secrets] section: clients are first-class (attach/detach, limits, expiry, per-client tg:// links) exactly like every other protocol, mirroring the WireGuard multi-client model. Per-client traffic and online status come from the fork's /stats JSON API (its Prometheus output has no per-user label), fed into the existing email-keyed client_traffics accumulator; an optional throttle caps concurrent connections. A one-time seeder converts each legacy single-secret inbound into a one-client inbound.

The fork ships only linux/darwin amd64/arm64 binaries but is pure Go, so provisioning builds it from source for every supported platform (release.yml, DockerInit.sh) while keeping the panel-expected mtg-<os>-<arch> filename and the 'run' verb, so process.go is untouched. Also fixes a pre-existing update.sh gap that never renamed the mtg binary for armv6/armv7 updates.

internal/mtproto/manager.go

@@ -1,7 +1,6 @@
 package mtproto
 
 import (
-	"bufio"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -17,13 +16,23 @@ import (
 	"github.com/mhsanaei/3x-ui/v3/internal/logger"
 )
 
-// Instance is the desired runtime configuration of one mtproto inbound.
-type Instance struct {
-	Id     int
-	Tag    string
-	Listen string
-	Port   int
+// SecretEntry is one named FakeTLS secret served by an mtg-multi process. Name is
+// the client email, used both as the [secrets] key and as the per-user key in the
+// /stats API so traffic can be attributed back to the client.
+type SecretEntry struct {
+	Name   string
 	Secret string
+}
+
+// Instance is the desired runtime configuration of one mtproto inbound. A single
+// mtg-multi process serves every active client's secret through the [secrets]
+// section, so one inbound maps to one process with many named secrets.
+type Instance struct {
+	Id      int
+	Tag     string
+	Listen  string
+	Port    int
+	Secrets []SecretEntry
 
 	// Optional mtg tuning; each is omitted from the generated TOML when
 	// zero-valued so mtg falls back to its own defaults.
@@ -34,6 +43,10 @@ type Instance struct {
 	FrontingPort          int
 	FrontingProxyProtocol bool
 
+	// ThrottleMaxConnections caps concurrent connections across all users with a
+	// fair-share algorithm; zero disables throttling.
+	ThrottleMaxConnections int
+
 	// When RouteThroughXray is set, mtg dials Telegram through the loopback
 	// SOCKS bridge the panel injects into the Xray config at XrayRoutePort, so
 	// the egress obeys the core's routing rules instead of going out directly.
@@ -50,37 +63,47 @@ func (inst Instance) bindTo() string {
 }
 
 // fingerprint changes whenever any value that ends up in the generated TOML
-// changes, so ensureLocked restarts mtg when the operator edits a setting.
+// changes, so ensureLocked restarts mtg when the operator edits a setting or a
+// client is added, removed, disabled, or re-keyed.
 func (inst Instance) fingerprint() string {
-	return strings.Join([]string{
+	parts := []string{
 		inst.bindTo(),
-		inst.Secret,
 		strconv.FormatBool(inst.Debug),
 		strconv.FormatBool(inst.ProxyProtocolListener),
 		inst.PreferIP,
 		inst.FrontingIP,
 		strconv.Itoa(inst.FrontingPort),
 		strconv.FormatBool(inst.FrontingProxyProtocol),
+		strconv.Itoa(inst.ThrottleMaxConnections),
 		strconv.FormatBool(inst.RouteThroughXray),
 		strconv.Itoa(inst.XrayRoutePort),
-	}, "|")
+	}
+	for _, e := range inst.Secrets {
+		parts = append(parts, e.Name+"="+e.Secret)
+	}
+	return strings.Join(parts, "|")
 }
 
-// Traffic is a per-inbound traffic delta scraped from an mtg metrics endpoint.
+// Traffic is a per-client traffic delta scraped from an mtg /stats endpoint. Tag
+// is the owning inbound's tag and Email is the client the bytes belong to.
 type Traffic struct {
-	Tag  string
-	Up   int64
-	Down int64
+	Tag   string
+	Email string
+	Up    int64
+	Down  int64
+}
+
+type clientCounters struct {
+	up   int64
+	down int64
 }
 
 type managed struct {
 	proc        *Process
 	tag         string
 	fingerprint string
-	metricsPort int
-	lastUp      int64
-	lastDown    int64
-	haveLast    bool
+	apiPort     int
+	last        map[string]clientCounters
 }
 
 // Manager owns the set of running mtg processes keyed by inbound id.
@@ -106,49 +129,61 @@ func GetManager() *Manager {
 }
 
 // InstanceFromInbound derives a desired Instance from an mtproto inbound,
-// healing the FakeTLS secret so it always matches the configured domain.
-// Returns false when the inbound is not a usable mtproto inbound.
+// building one named secret per active client. Secrets are healed on save (see
+// normalizeMtprotoSecret) and by the migration, so they are read as-is here to
+// keep the fingerprint stable across reconciles. Returns false when the inbound
+// is not a usable mtproto inbound or has no active client secret to serve.
 func InstanceFromInbound(ib *model.Inbound) (Instance, bool) {
 	if ib == nil || ib.Protocol != model.MTProto {
 		return Instance{}, false
 	}
 	settings := ib.Settings
-	if healed, ok := model.HealMtprotoSecret(settings); ok {
-		settings = healed
-	}
 	var parsed struct {
-		Secret                string `json:"secret"`
-		Debug                 bool   `json:"debug"`
-		ProxyProtocolListener bool   `json:"proxyProtocolListener"`
-		PreferIP              string `json:"preferIp"`
+		ProxyProtocolListener bool `json:"proxyProtocolListener"`
+		Debug                 bool `json:"debug"`
 		DomainFronting        struct {
 			IP            string `json:"ip"`
 			Port          int    `json:"port"`
 			ProxyProtocol bool   `json:"proxyProtocol"`
 		} `json:"domainFronting"`
-		RouteThroughXray bool `json:"routeThroughXray"`
-		RouteXrayPort    int  `json:"routeXrayPort"`
+		PreferIP               string `json:"preferIp"`
+		ThrottleMaxConnections int    `json:"throttleMaxConnections"`
+		RouteThroughXray       bool   `json:"routeThroughXray"`
+		RouteXrayPort          int    `json:"routeXrayPort"`
+		Clients                []struct {
+			Email  string `json:"email"`
+			Secret string `json:"secret"`
+			Enable bool   `json:"enable"`
+		} `json:"clients"`
 	}
 	if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
 		return Instance{}, false
 	}
-	if parsed.Secret == "" {
+	secrets := make([]SecretEntry, 0, len(parsed.Clients))
+	for _, c := range parsed.Clients {
+		if !c.Enable || c.Secret == "" || c.Email == "" {
+			continue
+		}
+		secrets = append(secrets, SecretEntry{Name: c.Email, Secret: c.Secret})
+	}
+	if len(secrets) == 0 {
 		return Instance{}, false
 	}
 	return Instance{
-		Id:                    ib.Id,
-		Tag:                   ib.Tag,
-		Listen:                ib.Listen,
-		Port:                  ib.Port,
-		Secret:                parsed.Secret,
-		Debug:                 parsed.Debug,
-		ProxyProtocolListener: parsed.ProxyProtocolListener,
-		PreferIP:              parsed.PreferIP,
-		FrontingIP:            parsed.DomainFronting.IP,
-		FrontingPort:          parsed.DomainFronting.Port,
-		FrontingProxyProtocol: parsed.DomainFronting.ProxyProtocol,
-		RouteThroughXray:      parsed.RouteThroughXray,
-		XrayRoutePort:         parsed.RouteXrayPort,
+		Id:                     ib.Id,
+		Tag:                    ib.Tag,
+		Listen:                 ib.Listen,
+		Port:                   ib.Port,
+		Secrets:                secrets,
+		Debug:                  parsed.Debug,
+		ProxyProtocolListener:  parsed.ProxyProtocolListener,
+		PreferIP:               parsed.PreferIP,
+		FrontingIP:             parsed.DomainFronting.IP,
+		FrontingPort:           parsed.DomainFronting.Port,
+		FrontingProxyProtocol:  parsed.DomainFronting.ProxyProtocol,
+		ThrottleMaxConnections: parsed.ThrottleMaxConnections,
+		RouteThroughXray:       parsed.RouteThroughXray,
+		XrayRoutePort:          parsed.RouteXrayPort,
 	}, true
 }
 
@@ -185,12 +220,12 @@ func (m *Manager) ensureLocked(inst Instance) error {
 		_ = cur.proc.Stop()
 		delete(m.procs, inst.Id)
 	}
-	metricsPort, err := FreeLocalPort()
+	apiPort, err := FreeLocalPort()
 	if err != nil {
 		return err
 	}
 	cfgPath := configPathForID(inst.Id)
-	if err := writeConfig(cfgPath, inst, metricsPort); err != nil {
+	if err := writeConfig(cfgPath, inst, apiPort); err != nil {
 		return err
 	}
 	proc := newProcess(cfgPath, fmt.Sprintf("inbound %d", inst.Id))
@@ -201,7 +236,8 @@ func (m *Manager) ensureLocked(inst Instance) error {
 		proc:        proc,
 		tag:         inst.Tag,
 		fingerprint: fp,
-		metricsPort: metricsPort,
+		apiPort:     apiPort,
+		last:        map[string]clientCounters{},
 	}
 	logger.Infof("mtproto: started mtg for inbound %d on %s", inst.Id, inst.bindTo())
 	return nil
@@ -255,18 +291,15 @@ func (m *Manager) StopAll() {
 	}
 }
 
-// CollectTraffic scrapes each running mtg metrics endpoint and returns the
-// per-inbound byte deltas since the previous scrape.
-func (m *Manager) CollectTraffic() []Traffic {
-	// Snapshot the state we need under the lock, then release before doing
-	// network I/O so that Ensure/Reconcile/Remove are not blocked.
+// CollectTraffic scrapes each running mtg /stats endpoint and returns the
+// per-client byte deltas since the previous scrape, plus the emails of clients
+// with at least one live connection.
+func (m *Manager) CollectTraffic() ([]Traffic, []string) {
 	type snap struct {
-		id          int
-		metricsPort int
-		tag         string
-		haveLast    bool
-		lastUp      int64
-		lastDown    int64
+		id      int
+		apiPort int
+		tag     string
+		last    map[string]clientCounters
 	}
 	m.mu.Lock()
 	snaps := make([]snap, 0, len(m.procs))
@@ -274,54 +307,57 @@ func (m *Manager) CollectTraffic() []Traffic {
 		if cur.proc == nil || !cur.proc.IsRunning() {
 			continue
 		}
-		snaps = append(snaps, snap{
-			id:          id,
-			metricsPort: cur.metricsPort,
-			tag:         cur.tag,
-			haveLast:    cur.haveLast,
-			lastUp:      cur.lastUp,
-			lastDown:    cur.lastDown,
-		})
+		lastCopy := make(map[string]clientCounters, len(cur.last))
+		for k, v := range cur.last {
+			lastCopy[k] = v
+		}
+		snaps = append(snaps, snap{id: id, apiPort: cur.apiPort, tag: cur.tag, last: lastCopy})
 	}
 	m.mu.Unlock()
 
-	out := make([]Traffic, 0, len(snaps))
+	var out []Traffic
+	var online []string
 	for _, s := range snaps {
-		up, down, ok := scrapeTraffic(s.metricsPort)
+		users, ok := scrapeStats(s.apiPort)
 		if !ok {
 			continue
 		}
-		var du, dd int64
-		if s.haveLast {
-			du = up - s.lastUp
-			dd = down - s.lastDown
+		newLast := make(map[string]clientCounters, len(users))
+		for email, u := range users {
+			up := u.BytesIn
+			down := u.BytesOut
+			newLast[email] = clientCounters{up: up, down: down}
+			if u.Connections > 0 {
+				online = append(online, email)
+			}
+			prev, had := s.last[email]
+			if !had {
+				continue
+			}
+			du := up - prev.up
+			dd := down - prev.down
 			if du < 0 {
 				du = 0
 			}
 			if dd < 0 {
 				dd = 0
 			}
+			if du > 0 || dd > 0 {
+				out = append(out, Traffic{Tag: s.tag, Email: email, Up: du, Down: dd})
+			}
 		}
 
-		// Re-acquire lock to persist the new baseline, but only if the entry
-		// still exists (it may have been removed during the scrape).
 		m.mu.Lock()
 		if cur, ok := m.procs[s.id]; ok {
-			cur.lastUp = up
-			cur.lastDown = down
-			cur.haveLast = true
+			cur.last = newLast
 		}
 		m.mu.Unlock()
-
-		if s.haveLast && (du > 0 || dd > 0) {
-			out = append(out, Traffic{Tag: s.tag, Up: du, Down: dd})
-		}
 	}
-	return out
+	return out, online
 }
 
 // FreeLocalPort asks the OS for an unused loopback TCP port. It is used both
-// for mtg's metrics endpoint and to allocate the per-inbound SOCKS egress
+// for mtg's /stats API endpoint and to allocate the per-inbound SOCKS egress
 // bridge port persisted into mtproto inbound settings.
 func FreeLocalPort() (int, error) {
 	l, err := (&net.ListenConfig{}).Listen(context.Background(), "tcp", "127.0.0.1:0")
@@ -332,13 +368,13 @@ func FreeLocalPort() (int, error) {
 	return l.Addr().(*net.TCPAddr).Port, nil
 }
 
-// renderConfig builds the mtg TOML for an instance. Top-level keys must precede
-// any [section] header in TOML, so the layout is: required keys, then the
-// optional scalar tuning, then [domain-fronting], and finally [stats.prometheus]
-// — which x-ui always emits and scrapes for traffic (see scrapeTraffic).
-func renderConfig(inst Instance, metricsPort int) string {
+// renderConfig builds the mtg-multi TOML for an instance. Top-level keys must
+// precede any [section] header in TOML, and [secrets] must be the final section
+// so trailing keys are not swallowed by another table. The layout is therefore:
+// top-level scalars (incl. api-bind-to), then [domain-fronting], [network] and
+// [throttle], and finally [secrets] with one named secret per active client.
+func renderConfig(inst Instance, apiPort int) string {
 	var b strings.Builder
-	fmt.Fprintf(&b, "secret = %q\n", inst.Secret)
 	fmt.Fprintf(&b, "bind-to = %q\n", inst.bindTo())
 	if inst.Debug {
 		b.WriteString("debug = true\n")
@@ -349,10 +385,11 @@ func renderConfig(inst Instance, metricsPort int) string {
 	if inst.PreferIP != "" {
 		fmt.Fprintf(&b, "prefer-ip = %q\n", inst.PreferIP)
 	}
+	fmt.Fprintf(&b, "api-bind-to = \"127.0.0.1:%d\"\n", apiPort)
 	if inst.FrontingIP != "" || inst.FrontingPort > 0 || inst.FrontingProxyProtocol {
 		b.WriteString("\n[domain-fronting]\n")
 		if inst.FrontingIP != "" {
-			fmt.Fprintf(&b, "ip = %q\n", inst.FrontingIP)
+			fmt.Fprintf(&b, "host = %q\n", inst.FrontingIP)
 		}
 		if inst.FrontingPort > 0 {
 			fmt.Fprintf(&b, "port = %d\n", inst.FrontingPort)
@@ -367,92 +404,51 @@ func renderConfig(inst Instance, metricsPort int) string {
 	if inst.RouteThroughXray && inst.XrayRoutePort > 0 {
 		fmt.Fprintf(&b, "\n[network]\nproxies = [\"socks5://127.0.0.1:%d\"]\n", inst.XrayRoutePort)
 	}
-	fmt.Fprintf(&b, "\n[stats.prometheus]\nenabled = true\nbind-to = \"127.0.0.1:%d\"\nhttp-path = \"/metrics\"\nmetric-prefix = \"mtg\"\n", metricsPort)
+	if inst.ThrottleMaxConnections > 0 {
+		fmt.Fprintf(&b, "\n[throttle]\nmax-connections = %d\n", inst.ThrottleMaxConnections)
+	}
+	b.WriteString("\n[secrets]\n")
+	for _, e := range inst.Secrets {
+		fmt.Fprintf(&b, "%q = %q\n", e.Name, e.Secret)
+	}
 	return b.String()
 }
 
-func writeConfig(path string, inst Instance, metricsPort int) error {
+func writeConfig(path string, inst Instance, apiPort int) error {
 	if err := os.MkdirAll(configDir(), 0o750); err != nil {
 		return err
 	}
-	return os.WriteFile(path, []byte(renderConfig(inst, metricsPort)), 0o640)
+	return os.WriteFile(path, []byte(renderConfig(inst, apiPort)), 0o640)
 }
 
-// scrapeTraffic reads the mtg Prometheus metrics endpoint and sums byte
-// counters by direction. mtg exposes a traffic counter labelled with a
-// direction; "to_telegram" is treated as upload and "to_client" as download.
-// Best-effort: an unreachable endpoint or unrecognised format yields ok=false.
-func scrapeTraffic(port int) (up int64, down int64, ok bool) {
+// statsUser is one entry of the mtg-multi /stats users map. bytes_in is traffic
+// the client sent to the proxy (upload) and bytes_out is what the proxy returned
+// (download).
+type statsUser struct {
+	Connections int64 `json:"connections"`
+	BytesIn     int64 `json:"bytes_in"`
+	BytesOut    int64 `json:"bytes_out"`
+}
+
+// scrapeStats reads the mtg-multi /stats JSON API and returns the per-user
+// cumulative counters. Best-effort: an unreachable endpoint or unparseable body
+// yields ok=false.
+func scrapeStats(port int) (map[string]statsUser, bool) {
 	client := http.Client{Timeout: 3 * time.Second}
-	req, reqErr := http.NewRequestWithContext(context.Background(), http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/metrics", port), nil)
-	if reqErr != nil {
-		return 0, 0, false
+	req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/stats", port), nil)
+	if err != nil {
+		return nil, false
 	}
 	resp, err := client.Do(req)
 	if err != nil {
-		return 0, 0, false
+		return nil, false
 	}
 	defer resp.Body.Close()
-	scanner := bufio.NewScanner(resp.Body)
-	scanner.Buffer(make([]byte, 64*1024), 1024*1024)
-	found := false
-	for scanner.Scan() {
-		line := strings.TrimSpace(scanner.Text())
-		if line == "" || line[0] == '#' || !strings.Contains(line, "traffic") {
-			continue
-		}
-		name, labels, value, perr := parseMetricLine(line)
-		if perr != nil || !strings.HasPrefix(name, "mtg") {
-			continue
-		}
-		switch labels["direction"] {
-		case "to_telegram", "egress", "up":
-			up += int64(value)
-		case "to_client", "ingress", "down":
-			down += int64(value)
-		default:
-			down += int64(value)
-		}
-		found = true
+	var parsed struct {
+		Users map[string]statsUser `json:"users"`
 	}
-	if err := scanner.Err(); err != nil {
-		logger.Debug("mtproto: metrics scan error:", err)
+	if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
+		return nil, false
 	}
-	return up, down, found
-}
-
-func parseMetricLine(line string) (name string, labels map[string]string, value float64, err error) {
-	labels = map[string]string{}
-	var rest string
-	if brace := strings.IndexByte(line, '{'); brace >= 0 {
-		name = line[:brace]
-		end := strings.IndexByte(line, '}')
-		if end < brace {
-			return "", nil, 0, fmt.Errorf("malformed metric line")
-		}
-		for kv := range strings.SplitSeq(line[brace+1:end], ",") {
-			before, after, ok := strings.Cut(kv, "=")
-			if !ok {
-				continue
-			}
-			labels[strings.TrimSpace(before)] = strings.Trim(strings.TrimSpace(after), `"`)
-		}
-		rest = strings.TrimSpace(line[end+1:])
-	} else {
-		fields := strings.Fields(line)
-		if len(fields) < 2 {
-			return "", nil, 0, fmt.Errorf("malformed metric line")
-		}
-		name = fields[0]
-		rest = fields[1]
-	}
-	valFields := strings.Fields(rest)
-	if len(valFields) == 0 {
-		return "", nil, 0, fmt.Errorf("missing metric value")
-	}
-	value, err = strconv.ParseFloat(valFields[0], 64)
-	if err != nil {
-		return "", nil, 0, err
-	}
-	return name, labels, value, nil
+	return parsed.Users, true
 }