udpforwarding.go 13 KB

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