mirrors/dae
1
0
Fork 0
mirror of https://github.com/daeuniverse/dae.git synced 2025-07-04 15:27:55 +07:00
Code Issues Packages Projects Releases Wiki Activity

feat: support dns.ipversion_prefer

Browse Source
This commit is contained in:
mzz2017
2023-04-07 23:06:04 +08:00
parent 127a000058
commit 38cc66d1d3
7 changed files with 203 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
cmd/run.go
View File

@ -177,7 +177,11 @@ loop:
// New control plane.
obj := c.EjectBpf()
dnsCache := c.CloneDnsCache()
var dnsCache map[string]*control.DnsCache
if conf.Dns.IpVersionPrefer == newConf.Dns.IpVersionPrefer {
// Only keep dns cache when ip version preference not change.
dnsCache = c.CloneDnsCache()
}
log.Warnln("[Reload] Load new control plane")
newC, err := newControlPlane(log, obj, dnsCache, newConf, externGeoDataDirs)
if err != nil {

17
component/dns/dns.go
View File

@ -142,22 +142,7 @@ func (s *Dns) InitUpstreams() {
wg.Wait()
}
func (s *Dns) RequestSelect(msg *dnsmessage.Message) (upstreamIndex consts.DnsRequestOutboundIndex, upstream *Upstream, err error) {
if msg.Response {
return 0, nil, fmt.Errorf("DNS request expected but DNS response received")
}
// Prepare routing.
var qname string
var qtype dnsmessage.Type
if len(msg.Questions) == 0 {
qname = ""
qtype = 0
} else {
q := msg.Questions[0]
qname = q.Name.String()
qtype = q.Type
}
func (s *Dns) RequestSelect(qname string, qtype dnsmessage.Type) (upstreamIndex consts.DnsRequestOutboundIndex, upstream *Upstream, err error) {
// Route.
upstreamIndex, err = s.reqMatcher.Match(qname, qtype)
if err != nil {

5
config/config.go
View File

@ -76,8 +76,9 @@ type DnsRouting struct {
}
type KeyableString string
type Dns struct {
Upstream []KeyableString `mapstructure:"upstream"`
Routing DnsRouting `mapstructure:"routing"`
IpVersionPrefer int `mapstructure:"ipversion_prefer"`
Upstream []KeyableString `mapstructure:"upstream"`
Routing DnsRouting `mapstructure:"routing"`
}
type Routing struct {

5
config/desc.go
View File

@ -41,7 +41,7 @@ var GlobalDesc = Desc{
"udp_check_dns": "This DNS will be used to check UDP connectivity of nodes. And if dns_upstream below contains tcp, it also be used to check TCP DNS connectivity of nodes.\nThis DNS should have both IPv4 and IPv6 if you have double stack in local.",
"check_interval": "Interval of connectivity check for TCP and UDP",
"check_tolerance": "Group will switch node only when new_latency <= old_latency - tolerance.",
"lan_interface": "The LAN interface to bind. Use it if you only want to proxy LAN instead of localhost.",
"lan_interface": "The LAN interface to bind. Use it if you want to proxy LAN.",
"wan_interface": "The WAN interface to bind. Use it if you want to proxy localhost. Use \"auto\" to auto detect.",
"allow_insecure": "Allow insecure TLS certificates. It is not recommended to turn it on unless you have to.",
"dial_mode": `Optional values of dial_mode are:
@ -54,7 +54,8 @@ var GlobalDesc = Desc{
}
var DnsDesc = Desc{
"upstream": "Value can be scheme://host:port, where the scheme can be tcp/udp/tcp+udp.\nIf host is a domain and has both IPv4 and IPv6 record, dae will automatically choose IPv4 or IPv6 to use according to group policy (such as min latency policy).\nPlease make sure DNS traffic will go through and be forwarded by dae, which is REQUIRED for domain routing.\nIf dial_mode is \"ip\", the upstream DNS answer SHOULD NOT be polluted, so domestic public DNS is not recommended.",
"ipversion_prefer": "For example, if ipversion_prefer is 4 and the domain name has both type A and type AAAA records, the dae will only respond to type A queries and response empty answer to type AAAA queries.",
"upstream": "Value can be scheme://host:port, where the scheme can be tcp/udp/tcp+udp.\nIf host is a domain and has both IPv4 and IPv6 record, dae will automatically choose IPv4 or IPv6 to use according to group policy (such as min latency policy).\nPlease make sure DNS traffic will go through and be forwarded by dae, which is REQUIRED for domain routing.\nIf dial_mode is \"ip\", the upstream DNS answer SHOULD NOT be polluted, so domestic public DNS is not recommended.",
"request": `DNS requests will follow this routing.
Built-in outbound: asis.
Available functions: qname, qtype`,

17
control/control_plane.go
View File

@ -379,15 +379,14 @@ func NewControlPlane(
}, nil
},
BestDialerChooser: plane.chooseBestDnsDialer,
IpVersionPrefer: dnsConfig.IpVersionPrefer,
}); err != nil {
return nil, err
}
// Refresh domain routing cache with new routing.
if dnsCache != nil {
if dnsCache != nil && len(dnsCache) > 0 {
for cacheKey, cache := range dnsCache {
if time.Now().After(cache.Deadline) {
continue
}
// Also refresh out-dated routing because kernel map items have no expiration.
lastDot := strings.LastIndex(cacheKey, ".")
if lastDot == -1 || lastDot == len(cacheKey)-1 {
// Not a valid key.
@ -398,6 +397,16 @@ func NewControlPlane(
typ := cacheKey[lastDot+1:]
_ = plane.dnsController.UpdateDnsCache(host, typ, cache.Answers, cache.Deadline)
}
} else if _bpf != nil {
// Is reloading, and dnsCache == nil.
// Remove all map items.
// Normally, it is due to the change of ip version preference.
var key [4]uint32
var val bpfDomainRouting
iter := core.bpf.DomainRoutingMap.Iterate()
for iter.Next(&key, &val) {
_ = core.bpf.DomainRoutingMap.Delete(&key)
}
}
// Init immediately to avoid DNS leaking in the very beginning because param control_plane_dns_routing will

209
control/dns_control.go
View File

@ -15,6 +15,7 @@ import (
"github.com/daeuniverse/dae/component/dns"
"github.com/daeuniverse/dae/component/outbound"
"github.com/daeuniverse/dae/component/outbound/dialer"
"github.com/mohae/deepcopy"
"github.com/mzz2017/softwind/netproxy"
"github.com/mzz2017/softwind/pool"
"github.com/sirupsen/logrus"
@ -32,6 +33,14 @@ const (
minFirefoxCacheTimeout = 120 * time.Second
)
type IpVersionPrefer int
const (
IpVersionPrefer_No IpVersionPrefer = 0
IpVersionPrefer_4 IpVersionPrefer = 4
IpVersionPrefer_6 IpVersionPrefer = 6
)
var (
SuspectedRushAnswerError = fmt.Errorf("suspected DNS rush-answer")
UnsupportedQuestionTypeError = fmt.Errorf("unsupported question type")
@ -42,10 +51,14 @@ type DnsControllerOption struct {
CacheAccessCallback func(cache *DnsCache) (err error)
NewCache func(fqdn string, answers []dnsmessage.Resource, deadline time.Time) (cache *DnsCache, err error)
BestDialerChooser func(req *udpRequest, upstream *dns.Upstream) (*dialArgument, error)
IpVersionPrefer int
}
type DnsController struct {
routing *dns.Dns
handling sync.Map
routing *dns.Dns
qtypePrefer dnsmessage.Type
log *logrus.Logger
cacheAccessCallback func(cache *DnsCache) (err error)
@ -57,9 +70,29 @@ type DnsController struct {
dnsCache map[string]*DnsCache
}
func parseIpVersionPreference(prefer int) (dnsmessage.Type, error) {
switch prefer := IpVersionPrefer(prefer); prefer {
case IpVersionPrefer_No:
return 0, nil
case IpVersionPrefer_4:
return dnsmessage.TypeA, nil
case IpVersionPrefer_6:
return dnsmessage.TypeAAAA, nil
default:
return 0, fmt.Errorf("unknown preference: %v", prefer)
}
}
func NewDnsController(routing *dns.Dns, option *DnsControllerOption) (c *DnsController, err error) {
// Parse ip version preference.
prefer, err := parseIpVersionPreference(option.IpVersionPrefer)
if err != nil {
return nil, err
}
return &DnsController{
routing: routing,
routing: routing,
qtypePrefer: prefer,
log: option.Log,
cacheAccessCallback: option.CacheAccessCallback,
@ -71,15 +104,28 @@ func NewDnsController(routing *dns.Dns, option *DnsControllerOption) (c *DnsCont
}, nil
}
func (c *DnsController) LookupDnsRespCache(domain string, t dnsmessage.Type) (cache *DnsCache) {
func (c *DnsController) cacheKey(qname string, qtype dnsmessage.Type) string {
// To fqdn.
if !strings.HasSuffix(qname, ".") {
qname = qname + "."
}
return strings.ToLower(qname) + qtype.String()
}
func (c *DnsController) RemoveDnsRespCache(qname string, qtype dnsmessage.Type) {
c.dnsCacheMu.Lock()
key := c.cacheKey(qname, qtype)
_, ok := c.dnsCache[key]
if ok {
delete(c.dnsCache, key)
}
c.dnsCacheMu.Unlock()
}
func (c *DnsController) LookupDnsRespCache(qname string, qtype dnsmessage.Type) (cache *DnsCache) {
now := time.Now()
// To fqdn.
if !strings.HasSuffix(domain, ".") {
domain = domain + "."
}
c.dnsCacheMu.Lock()
cache, ok := c.dnsCache[strings.ToLower(domain)+t.String()]
cache, ok := c.dnsCache[c.cacheKey(qname, qtype)]
c.dnsCacheMu.Unlock()
// We should make sure the remaining TTL is greater than 120s (minFirefoxCacheTimeout), or
// return nil and request a new lookup to refresh the cache.
@ -299,6 +345,85 @@ func (c *DnsController) Handle_(dnsMessage *dnsmessage.Message, req *udpRequest)
)
}
if dnsMessage.Response {
return fmt.Errorf("DNS request expected but DNS response received")
}
// Prepare qname, qtype.
var qname string
var qtype dnsmessage.Type
if len(dnsMessage.Questions) != 0 {
qname = dnsMessage.Questions[0].Name.String()
qtype = dnsMessage.Questions[0].Type
}
// Check ip version preference and qtype.
switch qtype {
case dnsmessage.TypeA, dnsmessage.TypeAAAA:
if c.qtypePrefer == 0 {
return c.handle_(dnsMessage, req, true)
}
default:
return c.handle_(dnsMessage, req, true)
}
// Try to make both A and AAAA lookups.
dnsMessage2 := deepcopy.Copy(dnsMessage).(*dnsmessage.Message)
var qtype2 dnsmessage.Type
switch qtype {
case dnsmessage.TypeA:
qtype2 = dnsmessage.TypeAAAA
case dnsmessage.TypeAAAA:
qtype2 = dnsmessage.TypeA
default:
return fmt.Errorf("unexpected qtype path")
}
dnsMessage2.Questions[0].Type = qtype2
done := make(chan struct{})
go func() {
_ = c.handle_(dnsMessage2, req, false)
done <- struct{}{}
}()
err = c.handle_(dnsMessage, req, false)
<-done
if err != nil {
return err
}
// Join results and consider whether to response.
dnsMessage.Response = false
resp := c.LookupDnsRespCache_(dnsMessage)
if resp == nil {
// resp is not valid.
c.log.WithFields(logrus.Fields{
"qname": qname,
}).Tracef("Reject %v due to resp not valid", qtype.String())
return c.sendReject_(dnsMessage, req)
}
// resp is valid.
cache2 := c.LookupDnsRespCache(qname, qtype2)
if c.qtypePrefer == qtype || cache2 == nil {
return sendPkt(resp, req.realDst, req.realSrc, req.src, req.lConn, req.lanWanFlag)
} else {
return c.sendReject_(dnsMessage, req)
}
}
func (c *DnsController) handle_(
dnsMessage *dnsmessage.Message,
req *udpRequest,
needResp bool,
) (err error) {
// Prepare qname, qtype.
var qname string
var qtype dnsmessage.Type
if len(dnsMessage.Questions) != 0 {
q := dnsMessage.Questions[0]
qname = q.Name.String()
qtype = q.Type
}
//// NOTICE: Rush-answer detector was removed because it does not always work in all districts.
//// Make sure there is additional record OPT in the request to filter DNS rush-answer in the response process.
//// Because rush-answer has no resp OPT. We can distinguish them from multiple responses.
@ -306,37 +431,29 @@ func (c *DnsController) Handle_(dnsMessage *dnsmessage.Message, req *udpRequest)
//_, _ = EnsureAdditionalOpt(dnsMessage, true)
// Route request.
upstreamIndex, upstream, err := c.routing.RequestSelect(dnsMessage)
upstreamIndex, upstream, err := c.routing.RequestSelect(qname, qtype)
if err != nil {
return err
}
if upstreamIndex == consts.DnsRequestOutboundIndex_Reject {
// Reject with empty answer.
dnsMessage.Answers = nil
dnsMessage.RCode = dnsmessage.RCodeSuccess
dnsMessage.Response = true
dnsMessage.RecursionAvailable = true
dnsMessage.Truncated = false
if c.log.IsLevelEnabled(logrus.TraceLevel) {
c.log.WithFields(logrus.Fields{
"question": dnsMessage.Questions,
}).Traceln("Reject with empty answer")
}
data, err := dnsMessage.Pack()
if err != nil {
return fmt.Errorf("pack DNS packet: %w", err)
}
if err = sendPkt(data, req.realDst, req.realSrc, req.src, req.lConn, req.lanWanFlag); err != nil {
return err
}
return nil
c.RemoveDnsRespCache(qname, qtype)
return c.sendReject_(dnsMessage, req)
}
// No parallel for the same lookup.
_mu, _ := c.handling.LoadOrStore(c.cacheKey(qname, qtype), new(sync.Mutex))
mu := _mu.(*sync.Mutex)
mu.Lock()
defer mu.Unlock()
if resp := c.LookupDnsRespCache_(dnsMessage); resp != nil {
// Send cache to client directly.
if err = sendPkt(resp, req.realDst, req.realSrc, req.src, req.lConn, req.lanWanFlag); err != nil {
return fmt.Errorf("failed to write cached DNS resp: %w", err)
if needResp {
if err = sendPkt(resp, req.realDst, req.realSrc, req.src, req.lConn, req.lanWanFlag); err != nil {
return fmt.Errorf("failed to write cached DNS resp: %w", err)
}
}
if c.log.IsLevelEnabled(logrus.DebugLevel) && len(dnsMessage.Questions) > 0 {
q := dnsMessage.Questions[0]
@ -363,10 +480,32 @@ func (c *DnsController) Handle_(dnsMessage *dnsmessage.Message, req *udpRequest)
if err != nil {
return fmt.Errorf("pack DNS packet: %w", err)
}
return c.dialSend(req, data, dnsMessage.ID, upstream, 0)
return c.dialSend(0, req, data, dnsMessage.ID, upstream, needResp)
}
func (c *DnsController) dialSend(req *udpRequest, data []byte, id uint16, upstream *dns.Upstream, invokingDepth int) (err error) {
// sendReject_ send empty answer.
func (c *DnsController) sendReject_(dnsMessage *dnsmessage.Message, req *udpRequest) (err error) {
dnsMessage.Answers = nil
dnsMessage.RCode = dnsmessage.RCodeSuccess
dnsMessage.Response = true
dnsMessage.RecursionAvailable = true
dnsMessage.Truncated = false
if c.log.IsLevelEnabled(logrus.TraceLevel) {
c.log.WithFields(logrus.Fields{
"question": dnsMessage.Questions,
}).Traceln("Reject with empty answer")
}
data, err := dnsMessage.Pack()
if err != nil {
return fmt.Errorf("pack DNS packet: %w", err)
}
if err = sendPkt(data, req.realDst, req.realSrc, req.src, req.lConn, req.lanWanFlag); err != nil {
return err
}
return nil
}
func (c *DnsController) dialSend(invokingDepth int, req *udpRequest, data []byte, id uint16, upstream *dns.Upstream, needResp bool) (err error) {
if invokingDepth >= MaxDnsLookupDepth {
return fmt.Errorf("too deep DNS lookup invoking (depth: %v); there may be infinite loop in your DNS response routing", MaxDnsLookupDepth)
}
@ -577,7 +716,7 @@ func (c *DnsController) dialSend(req *udpRequest, data []byte, id uint16, upstre
"next_upstream": nextUpstream.String(),
}).Traceln("Change DNS upstream and resend")
}
return c.dialSend(req, data, id, nextUpstream, invokingDepth+1)
return c.dialSend(invokingDepth+1, req, data, id, nextUpstream, needResp)
}
if upstreamIndex.IsReserved() && c.log.IsLevelEnabled(logrus.InfoLevel) {
var qname, qtype string
@ -612,8 +751,10 @@ func (c *DnsController) dialSend(req *udpRequest, data []byte, id uint16, upstre
if err != nil {
return err
}
if err = sendPkt(data, req.realDst, req.realSrc, req.src, req.lConn, req.lanWanFlag); err != nil {
return err
if needResp {
if err = sendPkt(data, req.realDst, req.realSrc, req.src, req.lConn, req.lanWanFlag); err != nil {
return err
}
}
return nil
}

4
example.dae
View File

@ -22,7 +22,7 @@ global {
# Group will switch node only when new_latency <= old_latency - tolerance.
check_tolerance: 50ms
# The LAN interface to bind. Use it if you only want to proxy LAN instead of localhost.
# The LAN interface to bind. Use it if you want to proxy LAN.
# Multiple interfaces split by ",".
#lan_interface: docker0
@ -79,6 +79,8 @@ node {
# See https://github.com/daeuniverse/dae/blob/main/docs/dns.md for full examples.
dns {
# For example, if ipversion_prefer is 4 and the domain name has both type A and type AAAA records, the dae will only respond to type A queries and response empty answer to type AAAA queries.
#ipversion_prefer: 4
upstream {
# Value can be scheme://host:port, where the scheme can be tcp/udp/tcp+udp.
# If host is a domain and has both IPv4 and IPv6 record, dae will automatically choose