manager.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725
  1. package mtproto
  2. import (
  3. "bytes"
  4. "context"
  5. "crypto/rand"
  6. "encoding/hex"
  7. "encoding/json"
  8. "fmt"
  9. "maps"
  10. "net"
  11. "net/http"
  12. "net/url"
  13. "os"
  14. "slices"
  15. "strconv"
  16. "strings"
  17. "sync"
  18. "time"
  19. "github.com/mhsanaei/3x-ui/v3/internal/database/model"
  20. "github.com/mhsanaei/3x-ui/v3/internal/logger"
  21. )
  22. // SecretEntry is one named FakeTLS secret served by an mtg-multi process. Name is
  23. // the client email, used both as the [secrets] key and as the per-user key in the
  24. // /stats API so traffic can be attributed back to the client. AdTag is the
  25. // client's own advertising-tag override, emitted into the [secret-ad-tags]
  26. // section; empty falls back to the instance-level tag.
  27. type SecretEntry struct {
  28. Name string
  29. Secret string
  30. AdTag string
  31. QuotaBytes int64
  32. ExpiresUnix int64
  33. }
  34. // Instance is the desired runtime configuration of one mtproto inbound. A single
  35. // mtg-multi process serves every active client's secret through the [secrets]
  36. // section, so one inbound maps to one process with many named secrets.
  37. type Instance struct {
  38. Id int
  39. Tag string
  40. Listen string
  41. Port int
  42. Secrets []SecretEntry
  43. // Optional mtg tuning; each is omitted from the generated TOML when
  44. // zero-valued so mtg falls back to its own defaults.
  45. Debug bool
  46. ProxyProtocolListener bool
  47. PreferIP string
  48. FrontingIP string
  49. FrontingPort int
  50. FrontingProxyProtocol bool
  51. // ThrottleMaxConnections caps concurrent connections across all users with a
  52. // fair-share algorithm; zero disables throttling.
  53. ThrottleMaxConnections int
  54. // PublicIPv4/PublicIPv6 pin the proxy's reachable address the Telegram
  55. // middle proxy needs when clients carry advertising tags; they are omitted
  56. // when empty so mtg auto-detects, and a change forces a restart.
  57. PublicIPv4 string
  58. PublicIPv6 string
  59. // When RouteThroughXray is set, mtg dials Telegram through the loopback
  60. // SOCKS bridge the panel injects into the Xray config at XrayRoutePort, so
  61. // the egress obeys the core's routing rules instead of going out directly.
  62. RouteThroughXray bool
  63. XrayRoutePort int
  64. }
  65. func (inst Instance) bindTo() string {
  66. listen := inst.Listen
  67. if listen == "" {
  68. listen = "0.0.0.0"
  69. }
  70. return fmt.Sprintf("%s:%d", listen, inst.Port)
  71. }
  72. // structuralFingerprint changes whenever a value outside the [secrets] section
  73. // of the generated TOML changes. Such a change can only be applied by
  74. // restarting mtg, unlike a secrets-only change, which a reload-capable mtg can
  75. // absorb in place.
  76. func (inst Instance) structuralFingerprint() string {
  77. parts := []string{
  78. inst.bindTo(),
  79. strconv.FormatBool(inst.Debug),
  80. strconv.FormatBool(inst.ProxyProtocolListener),
  81. inst.PreferIP,
  82. inst.FrontingIP,
  83. strconv.Itoa(inst.FrontingPort),
  84. strconv.FormatBool(inst.FrontingProxyProtocol),
  85. strconv.Itoa(inst.ThrottleMaxConnections),
  86. strconv.FormatBool(inst.RouteThroughXray),
  87. strconv.Itoa(inst.XrayRoutePort),
  88. inst.PublicIPv4,
  89. inst.PublicIPv6,
  90. }
  91. return strings.Join(parts, "|")
  92. }
  93. // secretsFingerprint identifies the reloadable secret config regardless of
  94. // client order, so a reordered clients array in the stored settings does not
  95. // read as a change. It moves whenever a client is added, removed, disabled,
  96. // re-keyed, re-tagged, or re-limited (quota/expiry) — all of which mtg applies
  97. // in place without dropping connections.
  98. func (inst Instance) secretsFingerprint() string {
  99. pairs := make([]string, 0, len(inst.Secrets))
  100. for _, e := range inst.Secrets {
  101. pairs = append(pairs, fmt.Sprintf("%s=%s;tag=%s;q=%d;exp=%d", e.Name, e.Secret, e.AdTag, e.QuotaBytes, e.ExpiresUnix))
  102. }
  103. slices.Sort(pairs)
  104. return strings.Join(pairs, "|")
  105. }
  106. // Traffic is a per-client traffic delta scraped from an mtg /stats endpoint. Tag
  107. // is the owning inbound's tag and Email is the client the bytes belong to.
  108. type Traffic struct {
  109. Tag string
  110. Email string
  111. Up int64
  112. Down int64
  113. }
  114. type clientCounters struct {
  115. up int64
  116. down int64
  117. }
  118. type managed struct {
  119. proc *Process
  120. tag string
  121. structuralFP string
  122. secretsFP string
  123. apiPort int
  124. apiToken string
  125. last map[string]clientCounters
  126. }
  127. // Manager owns the set of running mtg processes keyed by inbound id.
  128. type Manager struct {
  129. mu sync.Mutex
  130. procs map[int]*managed
  131. // swept records that the one-time startup cleanup of orphaned mtg
  132. // processes (survivors of a previous x-ui run) has already run.
  133. swept bool
  134. }
  135. var (
  136. managerOnce sync.Once
  137. manager *Manager
  138. )
  139. // GetManager returns the process-wide mtg manager singleton.
  140. func GetManager() *Manager {
  141. managerOnce.Do(func() {
  142. manager = &Manager{procs: map[int]*managed{}}
  143. })
  144. return manager
  145. }
  146. // InstanceFromInbound derives a desired Instance from an mtproto inbound,
  147. // building one named secret per active client. Secrets are healed on save (see
  148. // normalizeMtprotoSecret) and by the migration, so they are read as-is here to
  149. // keep the fingerprint stable across reconciles. Returns false when the inbound
  150. // is not a usable mtproto inbound or has no active client secret to serve.
  151. func InstanceFromInbound(ib *model.Inbound) (Instance, bool) {
  152. if ib == nil || ib.Protocol != model.MTProto {
  153. return Instance{}, false
  154. }
  155. settings := ib.Settings
  156. var parsed struct {
  157. ProxyProtocolListener bool `json:"proxyProtocolListener"`
  158. Debug bool `json:"debug"`
  159. DomainFronting struct {
  160. IP string `json:"ip"`
  161. Port int `json:"port"`
  162. ProxyProtocol bool `json:"proxyProtocol"`
  163. } `json:"domainFronting"`
  164. PreferIP string `json:"preferIp"`
  165. ThrottleMaxConnections int `json:"throttleMaxConnections"`
  166. RouteThroughXray bool `json:"routeThroughXray"`
  167. RouteXrayPort int `json:"routeXrayPort"`
  168. PublicIPv4 string `json:"publicIpv4"`
  169. PublicIPv6 string `json:"publicIpv6"`
  170. Clients []struct {
  171. Email string `json:"email"`
  172. Secret string `json:"secret"`
  173. AdTag string `json:"adTag"`
  174. Enable bool `json:"enable"`
  175. TotalGB int64 `json:"totalGB"`
  176. ExpiryTime int64 `json:"expiryTime"`
  177. } `json:"clients"`
  178. }
  179. if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
  180. return Instance{}, false
  181. }
  182. secrets := make([]SecretEntry, 0, len(parsed.Clients))
  183. for _, c := range parsed.Clients {
  184. if !c.Enable || c.Secret == "" || c.Email == "" {
  185. continue
  186. }
  187. entry := SecretEntry{Name: c.Email, Secret: c.Secret, AdTag: usableAdTag(c.AdTag)}
  188. if c.TotalGB > 0 {
  189. entry.QuotaBytes = c.TotalGB
  190. }
  191. if c.ExpiryTime > 0 {
  192. entry.ExpiresUnix = c.ExpiryTime / 1000
  193. }
  194. secrets = append(secrets, entry)
  195. }
  196. if len(secrets) == 0 {
  197. return Instance{}, false
  198. }
  199. return Instance{
  200. Id: ib.Id,
  201. Tag: ib.Tag,
  202. Listen: ib.Listen,
  203. Port: ib.Port,
  204. Secrets: secrets,
  205. Debug: parsed.Debug,
  206. ProxyProtocolListener: parsed.ProxyProtocolListener,
  207. PreferIP: parsed.PreferIP,
  208. FrontingIP: parsed.DomainFronting.IP,
  209. FrontingPort: parsed.DomainFronting.Port,
  210. FrontingProxyProtocol: parsed.DomainFronting.ProxyProtocol,
  211. ThrottleMaxConnections: parsed.ThrottleMaxConnections,
  212. RouteThroughXray: parsed.RouteThroughXray,
  213. XrayRoutePort: parsed.RouteXrayPort,
  214. PublicIPv4: strings.TrimSpace(parsed.PublicIPv4),
  215. PublicIPv6: strings.TrimSpace(parsed.PublicIPv6),
  216. }, true
  217. }
  218. // usableAdTag returns a stored advertising tag only when it is well-formed.
  219. // The save paths validate tags, but settings can arrive from raw API payloads
  220. // or older data, and one malformed tag in a generated config makes mtg reject
  221. // the whole file — taking every client of the inbound down with it.
  222. func usableAdTag(tag string) string {
  223. tag = strings.TrimSpace(tag)
  224. if !model.ValidMtprotoAdTag(tag) {
  225. return ""
  226. }
  227. return tag
  228. }
  229. // Ensure starts the mtg process for an instance, or restarts it when its
  230. // configuration changed. A no-op when the desired process is already running.
  231. func (m *Manager) Ensure(inst Instance) error {
  232. m.mu.Lock()
  233. defer m.mu.Unlock()
  234. m.sweepOrphansLocked()
  235. return m.ensureLocked(inst)
  236. }
  237. // sweepOrphansLocked kills mtg processes left running by a previous x-ui run,
  238. // exactly once per process lifetime and before any of our own mtg are started.
  239. // Because x-ui owns every mtg process, anything alive at this point is an orphan
  240. // that would otherwise keep holding an inbound port with a stale secret.
  241. func (m *Manager) sweepOrphansLocked() {
  242. if m.swept {
  243. return
  244. }
  245. m.swept = true
  246. if n := killStrayMtgProcesses(GetBinaryPath()); n > 0 {
  247. logger.Warningf("mtproto: terminated %d orphaned mtg process(es) from a previous run", n)
  248. }
  249. }
  250. // ensureAction is what ensureLocked must do to move a running mtg process to a
  251. // desired instance: leave it alone, hot-reload only its secrets, or fully
  252. // restart it.
  253. type ensureAction int
  254. const (
  255. ensureNoop ensureAction = iota
  256. ensureReload
  257. ensureRestart
  258. )
  259. // ensureActionFor decides how to apply a desired instance to the currently
  260. // managed process. A structural change (or a dead process) forces a restart; a
  261. // secrets-only change is a candidate for an in-place reload; identical
  262. // fingerprints on a live process need nothing.
  263. func ensureActionFor(running bool, curStructFP, curSecretsFP, newStructFP, newSecretsFP string) ensureAction {
  264. if !running || curStructFP != newStructFP {
  265. return ensureRestart
  266. }
  267. if curSecretsFP != newSecretsFP {
  268. return ensureReload
  269. }
  270. return ensureNoop
  271. }
  272. func (m *Manager) ensureLocked(inst Instance) error {
  273. structFP := inst.structuralFingerprint()
  274. secFP := inst.secretsFingerprint()
  275. if cur, ok := m.procs[inst.Id]; ok {
  276. switch ensureActionFor(cur.proc.IsRunning(), cur.structuralFP, cur.secretsFP, structFP, secFP) {
  277. case ensureNoop:
  278. cur.tag = inst.Tag
  279. return nil
  280. case ensureReload:
  281. if err := writeConfig(configPathForID(inst.Id), inst, cur.apiPort, cur.apiToken); err != nil {
  282. return err
  283. }
  284. if applySecrets(cur.apiPort, cur.apiToken, inst) {
  285. cur.tag = inst.Tag
  286. cur.secretsFP = secFP
  287. logger.Infof("mtproto: applied secret update to inbound %d in place", inst.Id)
  288. return nil
  289. }
  290. logger.Warningf("mtproto: live secret update unavailable for inbound %d, restarting", inst.Id)
  291. fallthrough
  292. case ensureRestart:
  293. _ = cur.proc.Stop()
  294. delete(m.procs, inst.Id)
  295. }
  296. }
  297. apiPort, err := FreeLocalPort()
  298. if err != nil {
  299. return err
  300. }
  301. apiToken, err := newAPIToken()
  302. if err != nil {
  303. return err
  304. }
  305. cfgPath := configPathForID(inst.Id)
  306. if err := writeConfig(cfgPath, inst, apiPort, apiToken); err != nil {
  307. return err
  308. }
  309. proc := newProcess(cfgPath, fmt.Sprintf("inbound %d", inst.Id))
  310. if err := proc.Start(); err != nil {
  311. return err
  312. }
  313. m.procs[inst.Id] = &managed{
  314. proc: proc,
  315. tag: inst.Tag,
  316. structuralFP: structFP,
  317. secretsFP: secFP,
  318. apiPort: apiPort,
  319. apiToken: apiToken,
  320. last: map[string]clientCounters{},
  321. }
  322. logger.Infof("mtproto: started mtg for inbound %d on %s", inst.Id, inst.bindTo())
  323. return nil
  324. }
  325. // Remove stops and forgets the mtg process for an inbound id.
  326. func (m *Manager) Remove(id int) {
  327. m.mu.Lock()
  328. defer m.mu.Unlock()
  329. if cur, ok := m.procs[id]; ok {
  330. _ = cur.proc.Stop()
  331. delete(m.procs, id)
  332. _ = os.Remove(configPathForID(id))
  333. logger.Infof("mtproto: stopped mtg for inbound %d", id)
  334. }
  335. }
  336. // Reconcile drives the running set toward the desired instances: it stops
  337. // processes that are no longer wanted and (re)starts the rest. Used at boot
  338. // and periodically to recover from crashes.
  339. func (m *Manager) Reconcile(desired []Instance) {
  340. m.mu.Lock()
  341. defer m.mu.Unlock()
  342. m.sweepOrphansLocked()
  343. want := make(map[int]struct{}, len(desired))
  344. for _, inst := range desired {
  345. want[inst.Id] = struct{}{}
  346. }
  347. for id, cur := range m.procs {
  348. if _, ok := want[id]; !ok {
  349. _ = cur.proc.Stop()
  350. delete(m.procs, id)
  351. _ = os.Remove(configPathForID(id))
  352. }
  353. }
  354. for _, inst := range desired {
  355. if err := m.ensureLocked(inst); err != nil {
  356. logger.Warningf("mtproto: reconcile failed for inbound %d: %v", inst.Id, err)
  357. }
  358. }
  359. }
  360. // StopAll stops every managed mtg process. Called on panel shutdown.
  361. func (m *Manager) StopAll() {
  362. m.mu.Lock()
  363. defer m.mu.Unlock()
  364. for id, cur := range m.procs {
  365. _ = cur.proc.Stop()
  366. _ = os.Remove(configPathForID(id))
  367. delete(m.procs, id)
  368. }
  369. }
  370. // CollectTraffic scrapes each running mtg /stats endpoint and returns the
  371. // per-client byte deltas since the previous scrape, plus the emails of clients
  372. // with at least one live connection.
  373. func (m *Manager) CollectTraffic() ([]Traffic, []string) {
  374. type snap struct {
  375. id int
  376. apiPort int
  377. apiToken string
  378. tag string
  379. last map[string]clientCounters
  380. }
  381. m.mu.Lock()
  382. snaps := make([]snap, 0, len(m.procs))
  383. for id, cur := range m.procs {
  384. if cur.proc == nil || !cur.proc.IsRunning() {
  385. continue
  386. }
  387. lastCopy := make(map[string]clientCounters, len(cur.last))
  388. maps.Copy(lastCopy, cur.last)
  389. snaps = append(snaps, snap{id: id, apiPort: cur.apiPort, apiToken: cur.apiToken, tag: cur.tag, last: lastCopy})
  390. }
  391. m.mu.Unlock()
  392. var out []Traffic
  393. var online []string
  394. for _, s := range snaps {
  395. users, ok := scrapeStats(s.apiPort, s.apiToken)
  396. if !ok {
  397. continue
  398. }
  399. newLast := make(map[string]clientCounters, len(users))
  400. for email, u := range users {
  401. up := u.BytesIn
  402. down := u.BytesOut
  403. newLast[email] = clientCounters{up: up, down: down}
  404. if u.Connections > 0 {
  405. online = append(online, email)
  406. }
  407. prev, had := s.last[email]
  408. if !had {
  409. continue
  410. }
  411. du := up - prev.up
  412. dd := down - prev.down
  413. if du < 0 {
  414. du = 0
  415. }
  416. if dd < 0 {
  417. dd = 0
  418. }
  419. if du > 0 || dd > 0 {
  420. out = append(out, Traffic{Tag: s.tag, Email: email, Up: du, Down: dd})
  421. }
  422. }
  423. m.mu.Lock()
  424. if cur, ok := m.procs[s.id]; ok {
  425. cur.last = newLast
  426. }
  427. m.mu.Unlock()
  428. }
  429. return out, online
  430. }
  431. func (m *Manager) HasRunning() bool {
  432. m.mu.Lock()
  433. defer m.mu.Unlock()
  434. for _, cur := range m.procs {
  435. if cur.proc != nil && cur.proc.IsRunning() {
  436. return true
  437. }
  438. }
  439. return false
  440. }
  441. func (m *Manager) ResetQuota(email string) {
  442. if email == "" {
  443. return
  444. }
  445. type target struct {
  446. port int
  447. token string
  448. }
  449. m.mu.Lock()
  450. targets := make([]target, 0, len(m.procs))
  451. for _, cur := range m.procs {
  452. if cur.proc != nil && cur.proc.IsRunning() {
  453. targets = append(targets, target{cur.apiPort, cur.apiToken})
  454. }
  455. }
  456. m.mu.Unlock()
  457. for _, t := range targets {
  458. resetQuota(t.port, t.token, email)
  459. }
  460. }
  461. func resetQuota(port int, token, email string) {
  462. client := http.Client{Timeout: 3 * time.Second}
  463. endpoint := fmt.Sprintf("http://127.0.0.1:%d/secrets/%s/reset-quota", port, url.PathEscape(email))
  464. req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, endpoint, nil)
  465. if err != nil {
  466. return
  467. }
  468. authorize(req, token)
  469. resp, err := client.Do(req)
  470. if err != nil {
  471. return
  472. }
  473. _ = resp.Body.Close()
  474. }
  475. // FreeLocalPort asks the OS for an unused loopback TCP port. It is used both
  476. // for mtg's /stats API endpoint and to allocate the per-inbound SOCKS egress
  477. // bridge port persisted into mtproto inbound settings.
  478. func FreeLocalPort() (int, error) {
  479. l, err := (&net.ListenConfig{}).Listen(context.Background(), "tcp", "127.0.0.1:0")
  480. if err != nil {
  481. return 0, err
  482. }
  483. defer l.Close()
  484. return l.Addr().(*net.TCPAddr).Port, nil
  485. }
  486. // renderConfig builds the mtg-multi TOML for an instance. Top-level keys must
  487. // precede any [section] header in TOML, and [secrets] must be the final section
  488. // so trailing keys are not swallowed by another table. The layout is therefore:
  489. // top-level scalars (incl. api-bind-to and api-token), then [domain-fronting],
  490. // [network] and [throttle], then [secret-ad-tags] for clients overriding the
  491. // global advertising tag, and finally [secrets] with one named secret per
  492. // active client.
  493. func renderConfig(inst Instance, apiPort int, apiToken string) string {
  494. var b strings.Builder
  495. fmt.Fprintf(&b, "bind-to = %q\n", inst.bindTo())
  496. if inst.Debug {
  497. b.WriteString("debug = true\n")
  498. }
  499. if inst.ProxyProtocolListener {
  500. b.WriteString("proxy-protocol-listener = true\n")
  501. }
  502. if inst.PreferIP != "" {
  503. fmt.Fprintf(&b, "prefer-ip = %q\n", inst.PreferIP)
  504. }
  505. fmt.Fprintf(&b, "api-bind-to = \"127.0.0.1:%d\"\n", apiPort)
  506. if apiToken != "" {
  507. fmt.Fprintf(&b, "api-token = %q\n", apiToken)
  508. }
  509. if inst.PublicIPv4 != "" {
  510. fmt.Fprintf(&b, "public-ipv4 = %q\n", inst.PublicIPv4)
  511. }
  512. if inst.PublicIPv6 != "" {
  513. fmt.Fprintf(&b, "public-ipv6 = %q\n", inst.PublicIPv6)
  514. }
  515. if inst.FrontingIP != "" || inst.FrontingPort > 0 || inst.FrontingProxyProtocol {
  516. b.WriteString("\n[domain-fronting]\n")
  517. if inst.FrontingIP != "" {
  518. fmt.Fprintf(&b, "host = %q\n", inst.FrontingIP)
  519. }
  520. if inst.FrontingPort > 0 {
  521. fmt.Fprintf(&b, "port = %d\n", inst.FrontingPort)
  522. }
  523. if inst.FrontingProxyProtocol {
  524. b.WriteString("proxy-protocol = true\n")
  525. }
  526. }
  527. // When the inbound opts into Xray routing, mtg reaches Telegram through the
  528. // loopback SOCKS bridge the panel injects into the running Xray config. mtg
  529. // only supports SOCKS5 upstreams, which is exactly what the bridge exposes.
  530. if inst.RouteThroughXray && inst.XrayRoutePort > 0 {
  531. fmt.Fprintf(&b, "\n[network]\nproxies = [\"socks5://127.0.0.1:%d\"]\n", inst.XrayRoutePort)
  532. }
  533. if inst.ThrottleMaxConnections > 0 {
  534. fmt.Fprintf(&b, "\n[throttle]\nmax-connections = %d\n", inst.ThrottleMaxConnections)
  535. }
  536. // Only clients present in [secrets] may appear here: mtg rejects a config
  537. // whose [secret-ad-tags] names an unknown secret, so a disabled client's
  538. // override must vanish together with its secret.
  539. tagged := false
  540. for _, e := range inst.Secrets {
  541. if e.AdTag == "" {
  542. continue
  543. }
  544. if !tagged {
  545. b.WriteString("\n[secret-ad-tags]\n")
  546. tagged = true
  547. }
  548. fmt.Fprintf(&b, "%q = %q\n", e.Name, e.AdTag)
  549. }
  550. for _, e := range inst.Secrets {
  551. if e.QuotaBytes <= 0 && e.ExpiresUnix <= 0 {
  552. continue
  553. }
  554. fmt.Fprintf(&b, "\n[secret-limits.%q]\n", e.Name)
  555. if e.QuotaBytes > 0 {
  556. fmt.Fprintf(&b, "quota = %q\n", quotaString(e.QuotaBytes))
  557. }
  558. if e.ExpiresUnix > 0 {
  559. fmt.Fprintf(&b, "expires = %q\n", expiresString(e.ExpiresUnix))
  560. }
  561. }
  562. b.WriteString("\n[secrets]\n")
  563. for _, e := range inst.Secrets {
  564. fmt.Fprintf(&b, "%q = %q\n", e.Name, e.Secret)
  565. }
  566. return b.String()
  567. }
  568. func writeConfig(path string, inst Instance, apiPort int, apiToken string) error {
  569. if err := os.MkdirAll(configDir(), 0o750); err != nil {
  570. return err
  571. }
  572. return os.WriteFile(path, []byte(renderConfig(inst, apiPort, apiToken)), 0o640)
  573. }
  574. // statsUser is one entry of the mtg-multi /stats users map. bytes_in is traffic
  575. // the client sent to the proxy (upload) and bytes_out is what the proxy returned
  576. // (download).
  577. type statsUser struct {
  578. Connections int64 `json:"connections"`
  579. BytesIn int64 `json:"bytes_in"`
  580. BytesOut int64 `json:"bytes_out"`
  581. }
  582. type secretPutEntry struct {
  583. Secret string `json:"secret"`
  584. AdTag string `json:"ad_tag,omitempty"`
  585. Quota string `json:"quota,omitempty"`
  586. Expires string `json:"expires,omitempty"`
  587. }
  588. type secretsPutBody struct {
  589. Secrets map[string]secretPutEntry `json:"secrets"`
  590. }
  591. func secretsPayload(inst Instance) secretsPutBody {
  592. secrets := make(map[string]secretPutEntry, len(inst.Secrets))
  593. for _, e := range inst.Secrets {
  594. entry := secretPutEntry{Secret: e.Secret, AdTag: e.AdTag}
  595. if e.QuotaBytes > 0 {
  596. entry.Quota = quotaString(e.QuotaBytes)
  597. }
  598. if e.ExpiresUnix > 0 {
  599. entry.Expires = expiresString(e.ExpiresUnix)
  600. }
  601. secrets[e.Name] = entry
  602. }
  603. return secretsPutBody{Secrets: secrets}
  604. }
  605. func quotaString(bytes int64) string {
  606. return strconv.FormatInt(bytes, 10) + "B"
  607. }
  608. func expiresString(unix int64) string {
  609. return time.Unix(unix, 0).UTC().Format(time.RFC3339)
  610. }
  611. // newAPIToken mints the bearer token one mtg process and its manager share for
  612. // the lifetime of that process. The management API can replace the whole
  613. // secret set, so even though it only listens on loopback it must not be open
  614. // to every local process. The token lives in the generated config (mtg reads
  615. // it at startup only — a rewritten token would not apply until a restart,
  616. // which is why the reload path reuses the stored one) and in the manager's
  617. // memory, nowhere else.
  618. func newAPIToken() (string, error) {
  619. buf := make([]byte, 16)
  620. if _, err := rand.Read(buf); err != nil {
  621. return "", err
  622. }
  623. return hex.EncodeToString(buf), nil
  624. }
  625. func authorize(req *http.Request, token string) {
  626. if token != "" {
  627. req.Header.Set("Authorization", "Bearer "+token)
  628. }
  629. }
  630. // applySecrets pushes the desired secret set and advertising tags to a running
  631. // mtg-multi through its management API (PUT /secrets on the same loopback port
  632. // that serves /stats), so a client add, removal, re-key, or ad-tag change is
  633. // applied in place. mtg keeps every connection whose secret is unchanged and
  634. // closes only the removed or re-keyed ones. It returns true only on a 200: an
  635. // older binary without the endpoint (404), a refused connection, or any other
  636. // status yields false, so the caller falls back to a full restart.
  637. func applySecrets(port int, token string, inst Instance) bool {
  638. body, err := json.Marshal(secretsPayload(inst))
  639. if err != nil {
  640. return false
  641. }
  642. client := http.Client{Timeout: 3 * time.Second}
  643. req, err := http.NewRequestWithContext(context.Background(), http.MethodPut, fmt.Sprintf("http://127.0.0.1:%d/secrets", port), bytes.NewReader(body))
  644. if err != nil {
  645. return false
  646. }
  647. req.Header.Set("Content-Type", "application/json")
  648. authorize(req, token)
  649. resp, err := client.Do(req)
  650. if err != nil {
  651. return false
  652. }
  653. defer resp.Body.Close()
  654. return resp.StatusCode == http.StatusOK
  655. }
  656. // scrapeStats reads the mtg-multi /stats JSON API and returns the per-user
  657. // cumulative counters. Best-effort: an unreachable endpoint or unparseable body
  658. // yields ok=false.
  659. func scrapeStats(port int, token string) (map[string]statsUser, bool) {
  660. client := http.Client{Timeout: 3 * time.Second}
  661. req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/stats", port), nil)
  662. if err != nil {
  663. return nil, false
  664. }
  665. authorize(req, token)
  666. resp, err := client.Do(req)
  667. if err != nil {
  668. return nil, false
  669. }
  670. defer resp.Body.Close()
  671. var parsed struct {
  672. Users map[string]statsUser `json:"users"`
  673. }
  674. if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
  675. return nil, false
  676. }
  677. return parsed.Users, true
  678. }