manager.go 23 KB

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