tcpforwarding.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480
  1. package service
  2. import (
  3. "context"
  4. "encoding/json"
  5. "fmt"
  6. v1 "github.com/go-nunu/nunu-layout-advanced/api/v1"
  7. "github.com/go-nunu/nunu-layout-advanced/internal/model"
  8. "github.com/go-nunu/nunu-layout-advanced/internal/repository"
  9. "golang.org/x/sync/errgroup"
  10. "maps"
  11. "net"
  12. "sort"
  13. )
  14. type TcpforwardingService interface {
  15. GetTcpforwarding(ctx context.Context, req v1.GetForwardingRequest) (v1.TcpForwardingDataRequest, error)
  16. AddTcpForwarding(ctx context.Context, req *v1.TcpForwardingRequest) error
  17. EditTcpForwarding(ctx context.Context, req *v1.TcpForwardingRequest) error
  18. DeleteTcpForwarding(ctx context.Context, req v1.DeleteTcpForwardingRequest) error
  19. GetTcpForwardingAllIpsByHostId(ctx context.Context, req v1.GetForwardingRequest) ([]v1.TcpForwardingDataRequest, error)
  20. }
  21. func NewTcpforwardingService(
  22. service *Service,
  23. tcpforwardingRepository repository.TcpforwardingRepository,
  24. parser ParserService,
  25. required RequiredService,
  26. crawler CrawlerService,
  27. globalRep repository.GlobalLimitRepository,
  28. hostRep repository.HostRepository,
  29. wafformatter WafFormatterService,
  30. cdn CdnService,
  31. proxy ProxyService,
  32. ) TcpforwardingService {
  33. return &tcpforwardingService{
  34. Service: service,
  35. tcpforwardingRepository: tcpforwardingRepository,
  36. parser: parser,
  37. required: required,
  38. crawler: crawler,
  39. globalRep: globalRep,
  40. hostRep: hostRep,
  41. wafformatter: wafformatter,
  42. cdn: cdn,
  43. proxy: proxy,
  44. }
  45. }
  46. type tcpforwardingService struct {
  47. *Service
  48. tcpforwardingRepository repository.TcpforwardingRepository
  49. parser ParserService
  50. required RequiredService
  51. crawler CrawlerService
  52. globalRep repository.GlobalLimitRepository
  53. hostRep repository.HostRepository
  54. wafformatter WafFormatterService
  55. cdn CdnService
  56. proxy ProxyService
  57. }
  58. func (s *tcpforwardingService) GetTcpforwarding(ctx context.Context, req v1.GetForwardingRequest) (v1.TcpForwardingDataRequest, error) {
  59. var tcpForwarding model.Tcpforwarding
  60. var backend model.TcpForwardingRule
  61. var err error
  62. g, gCtx := errgroup.WithContext(ctx)
  63. g.Go(func() error {
  64. res, e := s.tcpforwardingRepository.GetTcpforwarding(gCtx, int64(req.Id))
  65. if e != nil {
  66. return fmt.Errorf("GetTcpforwarding failed: %w", e)
  67. }
  68. if res != nil {
  69. tcpForwarding = *res
  70. }
  71. return nil
  72. })
  73. g.Go(func() error {
  74. res, e := s.tcpforwardingRepository.GetTcpForwardingIpsByID(gCtx, req.Id)
  75. if e != nil {
  76. return fmt.Errorf("GetTcpforwardingIps failed: %w", e)
  77. }
  78. if res != nil {
  79. backend = *res
  80. }
  81. return nil
  82. })
  83. if err = g.Wait(); err != nil {
  84. return v1.TcpForwardingDataRequest{}, err
  85. }
  86. return v1.TcpForwardingDataRequest{
  87. Id: tcpForwarding.Id,
  88. Port: tcpForwarding.Port,
  89. Comment: tcpForwarding.Comment,
  90. Proxy: tcpForwarding.Proxy,
  91. BackendList: backend.BackendList,
  92. }, nil
  93. }
  94. func (s *tcpforwardingService) buildTcpForwardingModel(req *v1.TcpForwardingDataRequest, ruleId int, require RequireResponse) *model.Tcpforwarding {
  95. return &model.Tcpforwarding{
  96. HostId: require.HostId,
  97. CdnWebId: ruleId,
  98. Port: req.Port,
  99. Comment: req.Comment,
  100. Proxy: req.Proxy,
  101. }
  102. }
  103. func (s *tcpforwardingService) buildTcpRuleModel(reqData *v1.TcpForwardingDataRequest, require RequireResponse, localDbId int, cdnOriginIds map[string]int64) *model.TcpForwardingRule {
  104. return &model.TcpForwardingRule{
  105. Uid: require.Uid,
  106. HostId: require.HostId,
  107. TcpId: localDbId, // 关联到本地数据库的主记录 ID
  108. CdnOriginIds: cdnOriginIds,
  109. BackendList: reqData.BackendList,
  110. }
  111. }
  112. func (s *tcpforwardingService) prepareWafData(ctx context.Context, req *v1.TcpForwardingRequest) (RequireResponse, v1.Website, error) {
  113. // 1. 获取必要的全局信息
  114. require, err := s.wafformatter.Require(ctx, v1.GlobalRequire{
  115. HostId: req.HostId,
  116. Uid: req.Uid,
  117. Comment: req.TcpForwardingData.Comment,
  118. })
  119. if err != nil {
  120. return RequireResponse{}, v1.Website{}, err
  121. }
  122. if require.GatewayGroupId == 0 || require.Uid == 0 {
  123. return RequireResponse{}, v1.Website{}, fmt.Errorf("请先配置实例")
  124. }
  125. var jsonData v1.TypeJSON
  126. jsonData.IsOn = true
  127. for _, v := range require.GatewayIps {
  128. jsonData.Listen = append(jsonData.Listen, v1.Listen{
  129. Protocol: "tcp",
  130. Host: v,
  131. Port: req.TcpForwardingData.Port,
  132. })
  133. }
  134. byteData, err := json.Marshal(jsonData)
  135. if err != nil {
  136. return RequireResponse{}, v1.Website{}, err
  137. }
  138. formData := v1.Website{
  139. UserId: int64(require.CdnUid),
  140. Type: "tcpProxy",
  141. Name: require.Tag,
  142. Description: req.TcpForwardingData.Comment,
  143. TcpJSON: byteData,
  144. ServerGroupIds: []int64{int64(require.GroupId)},
  145. UserPlanId: int64(require.RuleId),
  146. NodeClusterId: 1,
  147. }
  148. return require, formData, nil
  149. }
  150. func (s *tcpforwardingService) AddTcpForwarding(ctx context.Context, req *v1.TcpForwardingRequest) error {
  151. require, formData, err := s.prepareWafData(ctx, req)
  152. if err != nil {
  153. return err
  154. }
  155. err = s.wafformatter.validateWafPortCount(ctx, require.HostId)
  156. if err != nil {
  157. return err
  158. }
  159. tcpId, err := s.cdn.CreateWebsite(ctx, formData)
  160. if err != nil {
  161. return err
  162. }
  163. // 添加源站
  164. cdnOriginIds := make(map[string]int64)
  165. for _, v := range req.TcpForwardingData.BackendList{
  166. id, err := s.wafformatter.AddOrigin(ctx, v1.WebJson{
  167. ApiType: "tcp",
  168. BackendList: v,
  169. Comment: req.TcpForwardingData.Comment,
  170. })
  171. if err != nil {
  172. return err
  173. }
  174. cdnOriginIds[v] = id
  175. }
  176. // 添加源站到网站
  177. for _, v := range cdnOriginIds {
  178. err = s.cdn.AddServerOrigin(ctx, tcpId, v)
  179. if err != nil {
  180. return err
  181. }
  182. }
  183. // 开启proxy
  184. if req.TcpForwardingData.Proxy {
  185. err = s.proxy.EditProxy(ctx,tcpId, v1.ProxyProtocolJSON{
  186. IsOn: true,
  187. Version: 1,
  188. })
  189. if err != nil {
  190. return err
  191. }
  192. }
  193. tcpModel := s.buildTcpForwardingModel(&req.TcpForwardingData, int(tcpId), require)
  194. id, err := s.tcpforwardingRepository.AddTcpforwarding(ctx, tcpModel)
  195. if err != nil {
  196. return err
  197. }
  198. TcpRuleModel := s.buildTcpRuleModel(&req.TcpForwardingData, require, id, cdnOriginIds)
  199. if _, err = s.tcpforwardingRepository.AddTcpforwardingIps(ctx, *TcpRuleModel); err != nil {
  200. return err
  201. }
  202. // 异步任务:将源站IP添加到白名单
  203. var ips []string
  204. if req.TcpForwardingData.BackendList != nil {
  205. for _, v := range req.TcpForwardingData.BackendList {
  206. ip, _, err := net.SplitHostPort(v)
  207. if err != nil {
  208. return err
  209. }
  210. ips = append(ips, ip)
  211. }
  212. go s.wafformatter.PublishIpWhitelistTask(ips, "add","","white")
  213. }
  214. return nil
  215. }
  216. func (s *tcpforwardingService) EditTcpForwarding(ctx context.Context, req *v1.TcpForwardingRequest) error {
  217. require, formData, err := s.prepareWafData(ctx, req)
  218. if err != nil {
  219. return err
  220. }
  221. oldData, err := s.tcpforwardingRepository.GetTcpforwarding(ctx, int64(req.TcpForwardingData.Id))
  222. if err != nil {
  223. return err
  224. }
  225. //修改网站端口
  226. if oldData.Port != req.TcpForwardingData.Port {
  227. err = s.cdn.EditServerType(ctx, v1.EditWebsite{
  228. Id: int64(oldData.CdnWebId),
  229. TypeJSON: formData.TcpJSON,
  230. }, "tcp")
  231. if err != nil {
  232. return err
  233. }
  234. }
  235. //修改网站名字
  236. if oldData.Comment != req.TcpForwardingData.Comment {
  237. nodeId, err := s.globalRep.GetNodeId(ctx, oldData.CdnWebId)
  238. err = s.cdn.EditServerBasic(ctx, int64(oldData.CdnWebId), require.Tag, nodeId)
  239. if err != nil {
  240. return err
  241. }
  242. }
  243. //修改Proxy
  244. if oldData.Proxy != req.TcpForwardingData.Proxy {
  245. err = s.proxy.EditProxy(ctx, int64(oldData.CdnWebId), v1.ProxyProtocolJSON{
  246. IsOn: req.TcpForwardingData.Proxy,
  247. Version: 1,
  248. })
  249. if err != nil {
  250. return err
  251. }
  252. }
  253. // 异步任务:将IP添加到白名单
  254. ipData, err := s.tcpforwardingRepository.GetTcpForwardingIpsByID(ctx, req.TcpForwardingData.Id)
  255. if err != nil {
  256. return err
  257. }
  258. addedIps, removedIps, err := s.wafformatter.WashEditWafIp(ctx,req.TcpForwardingData.BackendList, ipData.BackendList)
  259. if err != nil {
  260. return err
  261. }
  262. if len(addedIps) > 0 {
  263. go s.wafformatter.PublishIpWhitelistTask(addedIps, "add","","white")
  264. }
  265. if len(removedIps) > 0 {
  266. ipsToDelist, err := s.wafformatter.WashDelIps(ctx, removedIps)
  267. if err != nil {
  268. return err
  269. }
  270. // 4. 如果有需要处理的IP,则批量发布一次任务
  271. if len(ipsToDelist) > 0 {
  272. go s.wafformatter.PublishIpWhitelistTask(ipsToDelist, "del", "0", "white")
  273. }
  274. }
  275. //修改源站
  276. addOrigins, delOrigins := s.wafformatter.findIpDifferences(ipData.BackendList, req.TcpForwardingData.BackendList)
  277. addedIds := make(map[string]int64)
  278. for _, v := range addOrigins {
  279. id, err := s.wafformatter.AddOrigin(ctx,v1.WebJson{
  280. ApiType: "tcp",
  281. BackendList: v,
  282. Comment: req.TcpForwardingData.Comment,
  283. })
  284. if err != nil {
  285. return err
  286. }
  287. addedIds[v] = id
  288. }
  289. for _, v := range addedIds {
  290. err = s.cdn.AddServerOrigin(ctx, int64(oldData.CdnWebId), v)
  291. if err != nil {
  292. return err
  293. }
  294. }
  295. maps.Copy(ipData.CdnOriginIds, addedIds)
  296. for k, v := range ipData.CdnOriginIds {
  297. for _, ip := range delOrigins {
  298. if k == ip {
  299. err = s.cdn.DelServerOrigin(ctx, int64(oldData.CdnWebId), v)
  300. if err != nil {
  301. return err
  302. }
  303. delete(ipData.CdnOriginIds, k)
  304. }
  305. }
  306. }
  307. tcpModel := s.buildTcpForwardingModel(&req.TcpForwardingData,oldData.CdnWebId, require)
  308. tcpModel.Id = req.TcpForwardingData.Id
  309. if err = s.tcpforwardingRepository.EditTcpforwarding(ctx, tcpModel); err != nil {
  310. return err
  311. }
  312. TcpRuleModel := s.buildTcpRuleModel(&req.TcpForwardingData, require, req.TcpForwardingData.Id, ipData.CdnOriginIds)
  313. if err = s.tcpforwardingRepository.EditTcpforwardingIps(ctx, *TcpRuleModel); err != nil {
  314. return err
  315. }
  316. return nil
  317. }
  318. func (s *tcpforwardingService) DeleteTcpForwarding(ctx context.Context, req v1.DeleteTcpForwardingRequest) error {
  319. for _, Id := range req.Ids {
  320. oldData, err := s.tcpforwardingRepository.GetTcpforwarding(ctx, int64(Id))
  321. if err != nil {
  322. return err
  323. }
  324. err = s.cdn.DelServer(ctx, int64(oldData.CdnWebId))
  325. if err != nil {
  326. return err
  327. }
  328. // 删除白名单
  329. var ips []string
  330. ipData, err := s.tcpforwardingRepository.GetTcpForwardingIpsByID(ctx, Id)
  331. if err != nil {
  332. return err
  333. }
  334. ips, err = s.wafformatter.WashDeleteWafIp(ctx, ipData.BackendList)
  335. if err != nil {
  336. return err
  337. }
  338. if len(ips) > 0 {
  339. ipsToDelist, err := s.wafformatter.WashDelIps(ctx, ips)
  340. if err != nil {
  341. return err
  342. }
  343. // 4. 如果有需要处理的IP,则批量发布一次任务
  344. if len(ipsToDelist) > 0 {
  345. go s.wafformatter.PublishIpWhitelistTask(ipsToDelist, "del", "0", "white")
  346. }
  347. }
  348. if err = s.tcpforwardingRepository.DeleteTcpforwarding(ctx, int64(Id)); err != nil {
  349. return err
  350. }
  351. if err = s.tcpforwardingRepository.DeleteTcpForwardingIpsById(ctx, Id); err != nil {
  352. return err
  353. }
  354. }
  355. return nil
  356. }
  357. func (s *tcpforwardingService) GetTcpForwardingAllIpsByHostId(ctx context.Context, req v1.GetForwardingRequest) ([]v1.TcpForwardingDataRequest, error) {
  358. type CombinedResult struct {
  359. Id int
  360. Forwarding *model.Tcpforwarding
  361. BackendRule *model.TcpForwardingRule
  362. Err error // 如果此ID的处理出错,则携带错误
  363. }
  364. g,gCtx := errgroup.WithContext(ctx)
  365. ids, err := s.tcpforwardingRepository.GetTcpForwardingAllIdsByID(gCtx, req.HostId)
  366. if err != nil {
  367. return nil, fmt.Errorf("GetTcpForwardingAllIds failed: %w", err)
  368. }
  369. if len(ids) == 0 {
  370. return nil, nil
  371. }
  372. resChan := make(chan CombinedResult, len(ids))
  373. g.Go(func() error {
  374. for _, idVal := range ids {
  375. currentID := idVal
  376. g.Go(func() error {
  377. var wf *model.Tcpforwarding
  378. var bk *model.TcpForwardingRule
  379. var localErr error
  380. wf, localErr = s.tcpforwardingRepository.GetTcpforwarding(gCtx, int64(currentID))
  381. if localErr != nil {
  382. resChan <- CombinedResult{Id: currentID, Err: localErr}
  383. return localErr
  384. }
  385. bk, localErr = s.tcpforwardingRepository.GetTcpForwardingIpsByID(gCtx, currentID)
  386. if localErr != nil {
  387. resChan <- CombinedResult{Id: currentID, Err: localErr}
  388. return localErr
  389. }
  390. resChan <- CombinedResult{Id: currentID, Forwarding: wf, BackendRule: bk}
  391. return nil
  392. })
  393. }
  394. return nil
  395. })
  396. groupErr := g.Wait()
  397. close(resChan)
  398. if groupErr != nil {
  399. return nil, groupErr
  400. }
  401. res := make([]v1.TcpForwardingDataRequest, 0, len(ids))
  402. for r := range resChan {
  403. if r.Err != nil {
  404. return nil, fmt.Errorf("received error from goroutine for ID %d: %w", r.Id, r.Err)
  405. }
  406. if r.Forwarding == nil {
  407. return nil,fmt.Errorf("received nil forwarding from goroutine for ID %d", r.Id)
  408. }
  409. dataReq := v1.TcpForwardingDataRequest{
  410. Id: r.Forwarding.Id,
  411. Port: r.Forwarding.Port,
  412. Comment: r.Forwarding.Comment,
  413. Proxy: r.Forwarding.Proxy,
  414. }
  415. if r.BackendRule != nil {
  416. dataReq.BackendList = r.BackendRule.BackendList
  417. }
  418. res = append(res, dataReq)
  419. }
  420. sort.Slice(res, func(i, j int) bool {
  421. return res[i].Id > res[j].Id
  422. })
  423. return res, nil
  424. }