package waf import ( "context" "encoding/json" "fmt" v1 "github.com/go-nunu/nunu-layout-advanced/api/v1" "github.com/go-nunu/nunu-layout-advanced/internal/model" "github.com/go-nunu/nunu-layout-advanced/internal/repository/api/waf" "github.com/go-nunu/nunu-layout-advanced/internal/service" "github.com/go-nunu/nunu-layout-advanced/internal/service/api/flexCdn" "github.com/go-nunu/nunu-layout-advanced/pkg/rabbitmq" "golang.org/x/sync/errgroup" "maps" "net" "sort" ) type WebForwardingService interface { GetWebForwarding(ctx context.Context, req v1.GetForwardingRequest) (v1.WebForwardingDataRequest, error) GetWebForwardingWafWebAllIps(ctx context.Context, req v1.GetForwardingRequest) ([]v1.WebForwardingDataRequest, error) AddWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) (int, error) EditWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) error DeleteWebForwarding(ctx context.Context, req v1.DeleteWebForwardingRequest) error } func NewWebForwardingService( service *service.Service, required service.RequiredService, webForwardingRepository waf.WebForwardingRepository, crawler service.CrawlerService, parser service.ParserService, wafformatter WafFormatterService, aoDun service.AoDunService, mq *rabbitmq.RabbitMQ, gatewayIp GatewayipService, globalLimitRep waf.GlobalLimitRepository, cdn flexCdn.CdnService, proxy flexCdn.ProxyService, sslCert flexCdn.SslCertService, websocket flexCdn.WebsocketService, cc CcService, ccIpList CcIpListService, aidedWeb AidedWebService, ) WebForwardingService { return &webForwardingService{ Service: service, webForwardingRepository: webForwardingRepository, required: required, parser: parser, crawler: crawler, wafformatter: wafformatter, aoDun: aoDun, mq: mq, gatewayIp: gatewayIp, cdn: cdn, globalLimitRep: globalLimitRep, proxy: proxy, sslCert: sslCert, websocket: websocket, cc: cc, ccIpList: ccIpList, aidedWeb: aidedWeb, } } type webForwardingService struct { *service.Service webForwardingRepository waf.WebForwardingRepository required service.RequiredService parser service.ParserService crawler service.CrawlerService wafformatter WafFormatterService aoDun service.AoDunService mq *rabbitmq.RabbitMQ gatewayIp GatewayipService cdn flexCdn.CdnService globalLimitRep waf.GlobalLimitRepository proxy flexCdn.ProxyService sslCert flexCdn.SslCertService websocket flexCdn.WebsocketService cc CcService ccIpList CcIpListService aidedWeb AidedWebService } func (s *webForwardingService) GetWebForwarding(ctx context.Context, req v1.GetForwardingRequest) (v1.WebForwardingDataRequest, error) { var webForwarding model.WebForwarding var backend model.WebForwardingRule g, gCtx := errgroup.WithContext(ctx) g.Go(func() error { res, e := s.webForwardingRepository.GetWebForwarding(gCtx, int64(req.Id)) if e != nil { // 直接返回错误，errgroup 会捕获它 return fmt.Errorf("GetWebForwarding failed: %w", e) } if res != nil { webForwarding = *res } return nil }) g.Go(func() error { res, e := s.webForwardingRepository.GetWebForwardingIpsByID(ctx, req.Id) if e != nil { return fmt.Errorf("GetWebForwardingByID failed: %w", e) } if res != nil { backend = *res } return nil }) if err := g.Wait(); err != nil { return v1.WebForwardingDataRequest{}, err } return v1.WebForwardingDataRequest{ Id: webForwarding.Id, Port: webForwarding.Port, Domain: webForwarding.Domain, IsHttps: webForwarding.IsHttps, Comment: webForwarding.Comment, BackendList: backend.BackendList, HttpsKey: webForwarding.HttpsKey, HttpsCert: webForwarding.HttpsCert, Proxy: webForwarding.Proxy, CcConfig: v1.CcConfigRequest{ IsOn: webForwarding.Cc, ThresholdMethod: webForwarding.ThresholdMethod, Level: webForwarding.Level, Limit5s: webForwarding.Limit5s, Limit60s: webForwarding.Limit60s, Limit300s: webForwarding.Limit300s, }, }, nil } // AddWebForwarding 添加Web转发配置 // 该函数负责创建完整的Web转发配置，包括： // 1. 数据验证和预处理 // 2. SSL证书管理 // 3. CDN网站创建和配置 // 4. 源站服务器添加 // 5. 各种功能开启（WebSocket、Proxy、日志、CC防护等） // 6. 数据库记录保存 // 7. 白名单任务发布 func (s *webForwardingService) AddWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) (int, error) { // 1. 数据准备和验证 require, formData, err := s.aidedWeb.PrepareWafData(ctx, req) if err != nil { return 0, err } if err := s.aidedWeb.ValidateAddRequest(ctx, req, require); err != nil { return 0, err } // 2. 处理SSL证书 if err := s.aidedWeb.ProcessSSLCertificate(ctx, req, require, formData); err != nil { return 0, err } // 3. 创建CDN网站 webId, err := s.aidedWeb.CreateCdnWebsite(ctx, req, require, formData) if err != nil { return 0, err } // 4. 配置WebSocket if err := s.aidedWeb.ConfigureWebsocket(ctx, webId); err != nil { return 0, err } // 5. 添加源站到网站 cdnOriginIds, err := s.aidedWeb.AddOriginsToWebsite(ctx, req, webId) if err != nil { return 0, err } // 6. 配置各种功能 if err := s.aidedWeb.ConfigureProxyProtocol(ctx, req, webId); err != nil { return 0, err } if err := s.aidedWeb.EditLog(ctx, webId); err != nil { return 0, err } if err := s.aidedWeb.ConfigureCCProtection(ctx, req, webId); err != nil { return 0, err } if err := s.aidedWeb.ConfigureWafFirewall(ctx, webId, require.GroupId); err != nil { return 0, err } // 7. 保存到数据库 id, err := s.aidedWeb.SaveToDatabase(ctx, req, require, webId, cdnOriginIds) if err != nil { return 0, err } // 8. 处理异步任务 s.aidedWeb.ProcessAsyncTasks(ctx, req, require) return id, nil } func (s *webForwardingService) EditWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) error { err := s.wafformatter.validateWafDomainCount(ctx, v1.GlobalRequire{ HostId: req.HostId, Domain: req.WebForwardingData.Domain, Comment: req.WebForwardingData.Comment, Uid: req.Uid, }) if err != nil { return err } oldData, err := s.webForwardingRepository.GetWebForwarding(ctx, int64(req.WebForwardingData.Id)) if err != nil { return err } req.WebForwardingData.SslCertId = int64(oldData.SslCertId) req.WebForwardingData.SslPolicyId = int64(oldData.SslPolicyId) require, formData, err := s.aidedWeb.PrepareWafData(ctx, req) if err != nil { return err } // 验证端口重复 protocol := s.aidedWeb.GetProtocolType(req.WebForwardingData.IsHttps) err = s.wafformatter.VerifyPort(ctx, protocol, int64(req.WebForwardingData.Id), req.WebForwardingData.Port, int64(require.HostId), req.WebForwardingData.Domain) if err != nil { return err } //修改网站端口 if oldData.Port != req.WebForwardingData.Port || oldData.IsHttps != req.WebForwardingData.IsHttps || oldData.HttpsCert != req.WebForwardingData.HttpsCert || oldData.HttpsKey != req.WebForwardingData.HttpsKey { var apiType string var closeType string // 修改证书 if oldData.HttpsCert != req.WebForwardingData.HttpsCert || oldData.HttpsKey != req.WebForwardingData.HttpsKey { err = s.sslCert.EditSSLCert(ctx, v1.SSL{ Name: req.WebForwardingData.Domain, CertId: oldData.SslCertId, CertData: req.WebForwardingData.HttpsCert, KeyData: req.WebForwardingData.HttpsKey, CdnUserId: require.CdnUid, Domain: req.WebForwardingData.Domain, Description: req.WebForwardingData.Comment, }) if err != nil { return err } } // 切换协议 var typeConfig v1.TypeJSON var closeConfig v1.TypeJSON if s.aidedWeb.IsHttpsProtocol(req.WebForwardingData.IsHttps) { typeConfig = formData.HttpsJSON closeConfig = formData.HttpJSON apiType = s.aidedWeb.GetProtocolType(req.WebForwardingData.IsHttps) closeType = s.aidedWeb.GetProtocolType(0) // HTTP } else { typeConfig = formData.HttpJSON closeConfig = formData.HttpsJSON apiType = s.aidedWeb.GetProtocolType(req.WebForwardingData.IsHttps) closeType = s.aidedWeb.GetProtocolType(1) // HTTPS } typeJson,err := json.Marshal(typeConfig) if err != nil { return err } closeJson,err := json.Marshal(closeConfig) if err != nil { return err } // 切换协议 err = s.cdn.EditServerType(ctx, v1.EditWebsite{ Id: int64(oldData.CdnWebId), TypeJSON: typeJson, }, apiType) if err != nil { return err } err = s.cdn.EditServerType(ctx, v1.EditWebsite{ Id: int64(oldData.CdnWebId), TypeJSON: closeJson, }, closeType) if err != nil { return err } } //修改网站域名 if oldData.Domain != req.WebForwardingData.Domain { type serverName struct { Name string `json:"name" form:"name"` Type string `json:"type" form:"type"` } var serverData []serverName var serverJson []byte serverData = append(serverData, serverName{ Name: req.WebForwardingData.Domain, Type: "full", }) serverJson, err = json.Marshal(serverData) if err != nil { return err } err = s.cdn.EditServerName(ctx, v1.EditServerNames{ ServerId: int64(oldData.CdnWebId), ServerNamesJSON: serverJson, }) if err != nil { return err } } //修改网站名字 if oldData.Comment != req.WebForwardingData.Comment { nodeId,err := s.globalLimitRep.GetNodeId(ctx, oldData.CdnWebId) if err != nil { return err } err = s.cdn.EditServerBasic(ctx, int64(oldData.CdnWebId), require.Tag,nodeId) if err != nil { return err } } //修改Proxy if oldData.Proxy != req.WebForwardingData.Proxy { err = s.proxy.EditProxy(ctx, int64(oldData.CdnWebId), v1.ProxyProtocolJSON{ IsOn: req.WebForwardingData.Proxy, Version: 1, }) if err != nil { return err } } // 修改CC配置 err = s.cc.EditCcConfig(ctx, int64(oldData.CdnWebId), v1.CcConfigRequest{ IsOn: req.WebForwardingData.CcConfig.IsOn, Level: req.WebForwardingData.CcConfig.Level, Limit5s: req.WebForwardingData.CcConfig.Limit5s, Limit60s: req.WebForwardingData.CcConfig.Limit60s, Limit300s: req.WebForwardingData.CcConfig.Limit300s, ThresholdMethod: req.WebForwardingData.CcConfig.ThresholdMethod, }) if err != nil { return err } // 将域名添加到白名单 webData, err := s.webForwardingRepository.GetWebForwarding(ctx, int64(req.WebForwardingData.Id)) if err != nil { return err } // 异步任务：将域名添加到白名单 if webData.Domain != req.WebForwardingData.Domain { firstIp, err := s.gatewayIp.GetGatewayipByHostIdFirst(ctx, int64(req.HostId), int64(req.Uid)) if err != nil { return err } doMain, err := s.wafformatter.ConvertToWildcardDomain(ctx, req.WebForwardingData.Domain) if err != nil { return err } oldDomain, err := s.wafformatter.ConvertToWildcardDomain(ctx, webData.Domain) if err != nil { return err } if len(require.GatewayIps) == 0 { return fmt.Errorf("网关组不存在") } // 查找域名数量,如果数量小于2，删除旧域名 count, err := s.webForwardingRepository.GetDomainCount(ctx, req.HostId, webData.Domain) if err != nil { return err } if count < 2 { go s.wafformatter.PublishDomainWhitelistTask(oldDomain, firstIp, "del") } go s.wafformatter.PublishDomainWhitelistTask(doMain, firstIp, "add") } // IP过白 ipData, err := s.webForwardingRepository.GetWebForwardingIpsByID(ctx, req.WebForwardingData.Id) if err != nil { return err } var oldIps []string var newIps []string for _, v := range ipData.BackendList { ip, _, err := net.SplitHostPort(v.Addr) if err != nil { return err } oldIps = append(oldIps, ip) } for _, v := range req.WebForwardingData.BackendList { ip, _, err := net.SplitHostPort(v.Addr) if err != nil { return err } newIps = append(newIps, ip) } addedIps, removedIps := s.wafformatter.findIpDifferences(oldIps, newIps) if len(addedIps) > 0 { go s.wafformatter.PublishIpWhitelistTask(addedIps, "add", "", "white") } // IP过白 if len(removedIps) > 0 { // 1. 一次性获取所有相关IP的数量 ipsToDelist, err := s.wafformatter.WashDelIps(ctx, removedIps) if err != nil { return err } // 4. 如果有需要处理的IP，则批量发布一次任务 if len(ipsToDelist) > 0 { go s.wafformatter.PublishIpWhitelistTask(ipsToDelist, "del", "0", "white") } } //修改源站 addOrigins, delOrigins := s.aidedWeb.FindDifferenceList(ipData.BackendList, req.WebForwardingData.BackendList) addedIds := make(map[string]int64) for _, v := range addOrigins { apiType := s.aidedWeb.GetProtocolType(v.IsHttps) id, err := s.wafformatter.AddOrigin(ctx, v1.WebJson{ ApiType: apiType, BackendList: v.Addr, Host: v.CustomHost, Comment: req.WebForwardingData.Comment, }) if err != nil { return err } addedIds[v.Addr] = id } for _, v := range addedIds { err = s.cdn.AddServerOrigin(ctx, int64(oldData.CdnWebId), v) if err != nil { return err } } for k, v := range ipData.CdnOriginIds { for _, ip := range delOrigins { if k == ip.Addr { err = s.cdn.DelServerOrigin(ctx, int64(oldData.CdnWebId), v) if err != nil { return err } delete(ipData.CdnOriginIds, k) } } } maps.Copy(ipData.CdnOriginIds, addedIds) webModel := s.aidedWeb.BuildWebForwardingModel(&req.WebForwardingData, req.WebForwardingData.CdnWebId, require) webModel.Id = req.WebForwardingData.Id if err = s.webForwardingRepository.EditWebForwarding(ctx, webModel); err != nil { return err } webRuleModel := s.aidedWeb.BuildWebRuleModel(&req.WebForwardingData, require, req.WebForwardingData.Id, ipData.CdnOriginIds) if err = s.webForwardingRepository.EditWebForwardingIps(ctx, *webRuleModel); err != nil { return err } return nil } func (s *webForwardingService) DeleteWebForwarding(ctx context.Context, req v1.DeleteWebForwardingRequest) error { for _, Id := range req.Ids { oldData, err := s.webForwardingRepository.GetWebForwarding(ctx, int64(Id)) if err != nil { return err } if oldData.HostId != req.HostId { return fmt.Errorf("用户权限不足") } err = s.cdn.DelServer(ctx, int64(oldData.CdnWebId)) if err != nil { return err } // 异步任务：将域名添加到白名单 firstIp, err := s.gatewayIp.GetGatewayipByHostIdFirst(ctx, int64(oldData.HostId), int64(req.Uid)) if err != nil { return err } if oldData.Domain != "" { doMain, err := s.wafformatter.ConvertToWildcardDomain(ctx, oldData.Domain) if err != nil { return err } go s.wafformatter.PublishDomainWhitelistTask(doMain, firstIp, "del") } // IP过白 ipData, err := s.webForwardingRepository.GetWebForwardingIpsByID(ctx, Id) if err != nil { return err } var ips []string if ipData != nil && len(ipData.BackendList) > 0 { for _, v := range ipData.BackendList { ip, _, err := net.SplitHostPort(v.Addr) if err != nil { return err } ips = append(ips, ip) } } if len(ips) > 0 { ipsToDelist, err := s.wafformatter.WashDelIps(ctx, ips) if err != nil { return err } // 4. 如果有需要处理的IP，则批量发布一次任务 if len(ipsToDelist) > 0 { go s.wafformatter.PublishIpWhitelistTask(ipsToDelist, "del", "0", "white") } } // 删除ssl if oldData.SslCertId != 0 { err = s.cdn.DelSSLCert(ctx, int64(oldData.SslCertId)) if err != nil { return err } err = s.sslCert.EditSslPolicy(ctx, int64(oldData.SslPolicyId), []int64{int64(oldData.SslCertId)}, "del") if err != nil { return err } } if err = s.webForwardingRepository.DeleteWebForwarding(ctx, int64(Id)); err != nil { return err } if err = s.webForwardingRepository.DeleteWebForwardingIpsById(ctx, Id); err != nil { return err } } return nil } func (s *webForwardingService) GetWebForwardingWafWebAllIps(ctx context.Context, req v1.GetForwardingRequest) ([]v1.WebForwardingDataRequest, error) { type CombinedResult struct { Id int Forwarding *model.WebForwarding BackendRule *model.WebForwardingRule Err error // 如果此ID的处理出错，则携带错误 } g, gCtx := errgroup.WithContext(ctx) ids, err := s.webForwardingRepository.GetWebForwardingWafWebAllIds(gCtx, req.HostId) if err != nil { return nil, fmt.Errorf("GetWebForwardingWafWebAllIds failed: %w", err) } if len(ids) == 0 { return nil, nil // 没有ID，直接返回空切片 } // 创建一个通道来接收每个ID的处理结果 // 通道缓冲区大小设为ID数量，这样发送者不会因为接收者慢而阻塞（在所有goroutine都启动后） resultsChan := make(chan CombinedResult, len(ids)) for _, idVal := range ids { currentID := idVal // 捕获循环变量 g.Go(func() error { var wf *model.WebForwarding var bk *model.WebForwardingRule var localErr error // 1. 获取 WebForwarding 信息 wf, localErr = s.webForwardingRepository.GetWebForwarding(gCtx, int64(currentID)) if localErr != nil { // 发送错误到通道，并由 errgroup 捕获 // errgroup 会处理第一个非nil错误，并取消其他 goroutine resultsChan <- CombinedResult{Id: currentID, Err: fmt.Errorf("GetWebForwarding for id %d failed: %w", currentID, localErr)} return localErr // 返回错误给 errgroup } if wf == nil { // 正常情况下，如果没错误，wf不应为nil，但防御性检查 localErr = fmt.Errorf("GetWebForwarding for id %d returned nil data without error", currentID) resultsChan <- CombinedResult{Id: currentID, Err: localErr} return localErr } // 2. 获取 Backend IP 信息 // 注意：这里我们允许 GetWebForwardingIpsByID 可能返回 nil 数据（例如没有规则）而不是错误 // 如果它也可能返回错误，则处理方式与上面类似 bk, localErr = s.webForwardingRepository.GetWebForwardingIpsByID(gCtx, currentID) if localErr != nil { // 如果获取IP信息失败是一个致命错误，则也应返回错误 // 如果允许部分成功（比如有WebForwarding但没有IP信息），则可以不将此视为errgroup的错误 // 这里假设它也是一个需要errgroup捕获的错误 resultsChan <- CombinedResult{Id: currentID, Forwarding: wf, Err: fmt.Errorf("GetWebForwardingIpsByID for id %d failed: %w", currentID, localErr)} return localErr // 返回错误给 errgroup } // bk 可能是 nil 如果没有错误且没有规则，这取决于业务逻辑 // 发送成功的结果到通道 resultsChan <- CombinedResult{Id: currentID, Forwarding: wf, BackendRule: bk} return nil // 此goroutine成功 }) } // 等待所有goroutine完成 groupErr := g.Wait() // 关闭通道，表示所有发送者都已完成 // 这一步很重要，这样下面的 range 循环才能正常结束 close(resultsChan) // 如果 errgroup 捕获到任何错误，优先返回该错误 if groupErr != nil { // 虽然errgroup已经出错了，但通道中可能已经有一些结果（来自出错前成功或出错的goroutine） // 我们需要排空通道以避免goroutine泄漏（如果它们在发送时阻塞） // 但由于我们优先返回groupErr，这些结果将被丢弃。 // 在这种设计下，通常任何一个子任务失败都会导致整个操作失败。 return nil, groupErr } // 如果没有错误，收集所有成功的结果 finalResults := make([]v1.WebForwardingDataRequest, 0, len(ids)) for res := range resultsChan { // 再次检查通道中的错误，尽管 errgroup 应该已经捕获了 // 但这是一种更细致的错误处理，以防万一有goroutine在errgroup.Wait()前发送了错误但未被errgroup捕获 // （理论上，如果goroutine返回了错误，errgroup会处理） // 主要目的是处理 res.forwarding 为 nil 的情况 (如果上面允许不返回错误) if res.Err != nil { // 如果到这里还有错误，说明逻辑可能有问题，或者我们决定忽略某些类型的子错误 // 在此示例中，因为 g.Wait() 没有错误，所以这里的 res.err 应该是nil // 如果不是，那么可能是goroutine在return nil前发送了带有错误的res。 // 严格来说，如果errgroup没有错误，这里res.err也应该是nil // 但以防万一，我们可以记录日志 return nil, fmt.Errorf("received error from goroutine for ID %d: %w", res.Id, res.Err) } if res.Forwarding == nil { return nil, fmt.Errorf("received nil forwarding from goroutine for ID %d", res.Id) } dataReq := v1.WebForwardingDataRequest{ Id: res.Forwarding.Id, Port: res.Forwarding.Port, Domain: res.Forwarding.Domain, IsHttps: res.Forwarding.IsHttps, Comment: res.Forwarding.Comment, HttpsKey: res.Forwarding.HttpsKey, HttpsCert: res.Forwarding.HttpsCert, Proxy: res.Forwarding.Proxy, CcConfig: v1.CcConfigRequest{ IsOn: res.Forwarding.Cc, ThresholdMethod: res.Forwarding.ThresholdMethod, Level: res.Forwarding.Level, Limit5s: res.Forwarding.Limit5s, Limit60s: res.Forwarding.Limit60s, Limit300s: res.Forwarding.Limit300s, }, } if res.BackendRule != nil { // 只有当 BackendRule 存在时才填充相关字段 dataReq.BackendList = res.BackendRule.BackendList } finalResults = append(finalResults, dataReq) } sort.Slice(finalResults, func(i, j int) bool { return finalResults[i].Id > finalResults[j].Id }) return finalResults, nil }