tcpforwarding.go 14 KB

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