manager.go 22 KB

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