udpforwarding.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461
  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. ) UdpForWardingService {
  32. return &udpForWardingService{
  33. Service: service,
  34. udpForWardingRepository: udpForWardingRepository,
  35. required: required,
  36. parser: parser,
  37. crawler: crawler,
  38. globalRep: globalRep,
  39. hostRep: hostRep,
  40. wafformatter: wafformatter,
  41. cdn: cdn,
  42. }
  43. }
  44. type udpForWardingService struct {
  45. *Service
  46. udpForWardingRepository repository.UdpForWardingRepository
  47. required RequiredService
  48. parser ParserService
  49. crawler CrawlerService
  50. globalRep repository.GlobalLimitRepository
  51. hostRep repository.HostRepository
  52. wafformatter WafFormatterService
  53. cdn CdnService
  54. }
  55. func (s *udpForWardingService) GetUdpForWarding(ctx context.Context,req v1.GetForwardingRequest) (v1.UdpForwardingDataRequest, error) {
  56. var udpForWarding model.UdpForWarding
  57. var backend model.UdpForwardingRule
  58. var err error
  59. g, gCtx := errgroup.WithContext(ctx)
  60. g.Go(func() error {
  61. res, e := s.udpForWardingRepository.GetUdpForWarding(gCtx, int64(req.Id))
  62. if e != nil {
  63. return fmt.Errorf("GetUdpForWarding failed: %w", e)
  64. }
  65. if res != nil {
  66. udpForWarding = *res
  67. }
  68. return nil
  69. })
  70. g.Go(func() error {
  71. res, e := s.udpForWardingRepository.GetUdpForwardingIpsByID(gCtx, req.Id)
  72. if e != nil {
  73. return fmt.Errorf("GetUdpForWardingByID failed: %w", e)
  74. }
  75. if res != nil {
  76. backend = *res
  77. }
  78. return nil
  79. })
  80. if err = g.Wait(); err != nil {
  81. return v1.UdpForwardingDataRequest{}, err
  82. }
  83. return v1.UdpForwardingDataRequest{
  84. Id: udpForWarding.Id,
  85. Port: udpForWarding.Port,
  86. BackendList: backend.BackendList,
  87. AllowIpList: backend.AllowIpList,
  88. DenyIpList: backend.DenyIpList,
  89. AccessRule: backend.AccessRule,
  90. Comment: udpForWarding.Comment,
  91. }, nil
  92. }
  93. func (s *udpForWardingService) buildUdpForwardingModel(req *v1.UdpForwardingDataRequest, ruleId int, require RequireResponse) *model.UdpForWarding {
  94. return &model.UdpForWarding{
  95. HostId: require.HostId,
  96. CdnWebId: ruleId,
  97. Port: req.Port,
  98. Comment: req.Comment,
  99. }
  100. }
  101. func (s *udpForWardingService) buildUdpRuleModel(reqData *v1.UdpForwardingDataRequest, require RequireResponse, localDbId int, cdnOriginIds map[string]int64) *model.UdpForwardingRule {
  102. return &model.UdpForwardingRule{
  103. Uid: require.Uid,
  104. HostId: require.HostId,
  105. UdpId: localDbId, // 关联到本地数据库的主记录 ID
  106. CdnOriginIds: cdnOriginIds,
  107. BackendList: reqData.BackendList,
  108. AllowIpList: reqData.AllowIpList,
  109. DenyIpList: reqData.DenyIpList,
  110. AccessRule: reqData.AccessRule,
  111. }
  112. }
  113. func (s *udpForWardingService) prepareWafData(ctx context.Context, req *v1.UdpForwardingRequest) (RequireResponse, v1.Website, error) {
  114. require, err := s.wafformatter.Require(ctx, v1.GlobalRequire{
  115. HostId: req.HostId,
  116. Uid: req.Uid,
  117. Comment: req.UdpForwardingData.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: "udp",
  130. Host: v,
  131. Port: req.UdpForwardingData.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: "udpProxy",
  141. Name: require.Tag,
  142. Description: req.UdpForwardingData.Comment,
  143. UdpJSON: 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 *udpForWardingService) AddUdpForwarding(ctx context.Context, req *v1.UdpForwardingRequest) 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. udpId, 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.UdpForwardingData.BackendList {
  166. id, err := s.wafformatter.AddOrigin(ctx, v1.WebJson{
  167. ApiType: "udp",
  168. BackendList: v,
  169. Comment: req.UdpForwardingData.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, udpId, v)
  179. if err != nil {
  180. return err
  181. }
  182. }
  183. udpModel := s.buildUdpForwardingModel(&req.UdpForwardingData, int(udpId), require)
  184. id, err := s.udpForWardingRepository.AddUdpForwarding(ctx, udpModel)
  185. if err != nil {
  186. return err
  187. }
  188. udpRuleModel := s.buildUdpRuleModel(&req.UdpForwardingData, require, id, cdnOriginIds)
  189. if _, err = s.udpForWardingRepository.AddUdpForwardingIps(ctx, *udpRuleModel); err != nil {
  190. return err
  191. }
  192. // 异步任务:将IP添加到白名单
  193. var ips []string
  194. if req.UdpForwardingData.BackendList != nil {
  195. for _, v := range req.UdpForwardingData.BackendList {
  196. ip, _, err := net.SplitHostPort(v)
  197. if err != nil {
  198. return err
  199. }
  200. ips = append(ips, ip)
  201. }
  202. go s.wafformatter.PublishIpWhitelistTask(ips, "add","")
  203. }
  204. var accessRuleIps []string
  205. if req.UdpForwardingData.AllowIpList != nil {
  206. for _, v := range require.GatewayIps {
  207. for _, ip := range req.UdpForwardingData.AllowIpList {
  208. if net.ParseIP(ip) != nil {
  209. accessRuleIps = append(accessRuleIps, ip)
  210. }
  211. }
  212. go s.wafformatter.PublishIpWhitelistTask(accessRuleIps, "add",v)
  213. }
  214. }
  215. return nil
  216. }
  217. func (s *udpForWardingService) EditUdpForwarding(ctx context.Context, req *v1.UdpForwardingRequest) error {
  218. require, formData, err := s.prepareWafData(ctx, req)
  219. if err != nil {
  220. return err
  221. }
  222. oldData, err := s.udpForWardingRepository.GetUdpForWarding(ctx, int64(req.UdpForwardingData.Id))
  223. if err != nil {
  224. return err
  225. }
  226. //修改网站端口
  227. if oldData.Port != req.UdpForwardingData.Port {
  228. err = s.cdn.EditServerType(ctx, v1.EditWebsite{
  229. Id: int64(oldData.CdnWebId),
  230. TypeJSON: formData.TcpJSON,
  231. }, "tcp")
  232. if err != nil {
  233. return err
  234. }
  235. }
  236. //修改网站名字
  237. if oldData.Comment != req.UdpForwardingData.Comment {
  238. err = s.cdn.EditServerBasic(ctx, int64(oldData.CdnWebId), require.Tag)
  239. if err != nil {
  240. return err
  241. }
  242. }
  243. // 异步任务:将IP添加到白名单
  244. ipData, err := s.udpForWardingRepository.GetUdpForwardingIpsByID(ctx, req.UdpForwardingData.Id)
  245. if err != nil {
  246. return err
  247. }
  248. addedIps, removedIps, addedAllowIps, removedAllowIps, err := s.wafformatter.WashEditWafIp(ctx,req.UdpForwardingData.BackendList,req.UdpForwardingData.AllowIpList,ipData.BackendList,ipData.AllowIpList)
  249. if err != nil {
  250. return err
  251. }
  252. if len(addedIps) > 0 {
  253. go s.wafformatter.PublishIpWhitelistTask(addedIps, "add","")
  254. }
  255. if len(removedIps) > 0 {
  256. go s.wafformatter.PublishIpWhitelistTask(removedIps, "del","")
  257. }
  258. if len(addedAllowIps) > 0 {
  259. for _, v := range require.GatewayIps {
  260. go s.wafformatter.PublishIpWhitelistTask(addedAllowIps, "add",v)
  261. }
  262. }
  263. if len(removedAllowIps) > 0 {
  264. for _, v := range require.GatewayIps {
  265. go s.wafformatter.PublishIpWhitelistTask(removedAllowIps, "del",v)
  266. }
  267. }
  268. //修改源站
  269. addOrigins, delOrigins := s.wafformatter.findIpDifferences(ipData.BackendList, req.UdpForwardingData.BackendList)
  270. addedIds := make(map[string]int64)
  271. for _, v := range addOrigins {
  272. id, err := s.wafformatter.AddOrigin(ctx,v1.WebJson{
  273. ApiType: "udp",
  274. BackendList: v,
  275. Comment: req.UdpForwardingData.Comment,
  276. })
  277. if err != nil {
  278. return err
  279. }
  280. addedIds[v] = id
  281. }
  282. for _, v := range addedIds {
  283. err = s.cdn.AddServerOrigin(ctx, int64(oldData.CdnWebId), v)
  284. if err != nil {
  285. return err
  286. }
  287. }
  288. maps.Copy(ipData.CdnOriginIds, addedIds)
  289. for k, v := range ipData.CdnOriginIds {
  290. for _, ip := range delOrigins {
  291. if k == ip {
  292. err = s.cdn.DelServerOrigin(ctx, int64(oldData.CdnWebId), v)
  293. if err != nil {
  294. return err
  295. }
  296. delete(ipData.CdnOriginIds, k)
  297. }
  298. }
  299. }
  300. udpModel := s.buildUdpForwardingModel(&req.UdpForwardingData, oldData.CdnWebId, require)
  301. udpModel.Id = req.UdpForwardingData.Id
  302. if err = s.udpForWardingRepository.EditUdpForwarding(ctx, udpModel); err != nil {
  303. return err
  304. }
  305. udpRuleModel := s.buildUdpRuleModel(&req.UdpForwardingData, require, req.UdpForwardingData.Id, ipData.CdnOriginIds)
  306. if err = s.udpForWardingRepository.EditUdpForwardingIps(ctx, *udpRuleModel); err != nil {
  307. return err
  308. }
  309. return nil
  310. }
  311. func (s *udpForWardingService) DeleteUdpForwarding(ctx context.Context, Ids []int) error {
  312. for _, id := range Ids {
  313. oldData, err := s.udpForWardingRepository.GetUdpForWarding(ctx, int64(id))
  314. if err != nil {
  315. return err
  316. }
  317. err = s.cdn.DelServer(ctx, int64(oldData.CdnWebId))
  318. if err != nil {
  319. return err
  320. }
  321. // 异步任务:将IP添加到白名单
  322. ipData, err := s.udpForWardingRepository.GetUdpForwardingIpsByID(ctx, id)
  323. if err != nil {
  324. return err
  325. }
  326. var ips []string
  327. ips, err = s.wafformatter.WashDeleteWafIp(ctx, ipData.BackendList, ipData.AllowIpList)
  328. if err != nil {
  329. return err
  330. }
  331. if len(ips) > 0 {
  332. go s.wafformatter.PublishIpWhitelistTask(ips, "del","")
  333. }
  334. if err = s.udpForWardingRepository.DeleteUdpForwarding(ctx, int64(id)); err != nil {
  335. return err
  336. }
  337. if err = s.udpForWardingRepository.DeleteUdpForwardingIpsById(ctx, id); err != nil {
  338. return err
  339. }
  340. }
  341. return nil
  342. }
  343. func (s *udpForWardingService) GetUdpForwardingWafUdpAllIps(ctx context.Context, req v1.GetForwardingRequest) ([]v1.UdpForwardingDataRequest, error) {
  344. type CombinedResult struct {
  345. Id int
  346. Forwarding *model.UdpForWarding
  347. BackendRule *model.UdpForwardingRule
  348. Err error // 如果此ID的处理出错,则携带错误
  349. }
  350. g,gCtx := errgroup.WithContext(ctx)
  351. ids, err := s.udpForWardingRepository.GetUdpForwardingWafUdpAllIds(gCtx, req.HostId)
  352. if err != nil {
  353. return nil, fmt.Errorf("GetUdpForwardingWafUdpAllIds failed: %w", err)
  354. }
  355. if len(ids) == 0 {
  356. return nil, nil
  357. }
  358. resChan := make(chan CombinedResult, len(ids))
  359. for _, idVal := range ids {
  360. currentID := idVal
  361. g.Go(func() error {
  362. var wf *model.UdpForWarding
  363. var bk *model.UdpForwardingRule
  364. var localErr error
  365. wf, localErr = s.udpForWardingRepository.GetUdpForWarding(gCtx, int64(currentID))
  366. if localErr != nil {
  367. resChan <- CombinedResult{Id: currentID, Err: localErr}
  368. return localErr
  369. }
  370. bk, localErr = s.udpForWardingRepository.GetUdpForwardingIpsByID(gCtx, currentID)
  371. if localErr != nil {
  372. resChan <- CombinedResult{Id: currentID, Err: localErr}
  373. return localErr
  374. }
  375. resChan <- CombinedResult{Id: currentID, Forwarding: wf, BackendRule: bk}
  376. return nil
  377. })
  378. }
  379. groupErr := g.Wait()
  380. close(resChan)
  381. if groupErr != nil {
  382. return nil, groupErr
  383. }
  384. res := make([]v1.UdpForwardingDataRequest, 0, len(ids))
  385. for r := range resChan {
  386. if r.Err != nil {
  387. return nil, fmt.Errorf("received error from goroutine for ID %d: %w", r.Id, r.Err)
  388. }
  389. if r.Forwarding == nil {
  390. return nil, fmt.Errorf("received nil forwarding from goroutine for ID %d", r.Id)
  391. }
  392. dataReq := v1.UdpForwardingDataRequest{
  393. Id: r.Forwarding.Id,
  394. Port: r.Forwarding.Port,
  395. Comment: r.Forwarding.Comment,
  396. }
  397. if r.BackendRule != nil {
  398. dataReq.BackendList = r.BackendRule.BackendList
  399. dataReq.AllowIpList = r.BackendRule.AllowIpList
  400. dataReq.DenyIpList = r.BackendRule.DenyIpList
  401. dataReq.AccessRule = r.BackendRule.AccessRule
  402. }
  403. res = append(res, dataReq)
  404. }
  405. sort.Slice(res, func(i, j int) bool {
  406. return res[i].Id > res[j].Id
  407. })
  408. return res, nil
  409. }