package zeroconf

import (

"context"

"fmt"

"log"

"math/rand"

"net"

"reflect"

"strings"

"time"

"github.com/enbility/zeroconf/v3/api"

"github.com/miekg/dns"

)

// IPType specifies the IP traffic the client listens for.

// This does not guarantee that only mDNS entries of this sepcific

// type passes. E.g. typical mDNS packets distributed via IPv4, often contain

// both DNS A and AAAA entries.

type IPType uint8

// Options for IPType.

const (

IPv4 IPType = 0x01

IPv6 IPType = 0x02

IPv4AndIPv6 = IPv4 | IPv6 // default option

)

var initialQueryInterval = 4 * time.Second

// Client structure encapsulates both IPv4/IPv6 UDP connections.

type Client struct {

ipv4conn api.PacketConn

ipv6conn api.PacketConn

ipv4Mgr *InterfaceManager

ipv6Mgr *InterfaceManager

provider api.InterfaceProvider

}

type clientOpts struct {

listenOn IPType

ifaces []net.Interface

connFactory api.ConnectionFactory

provider api.InterfaceProvider

}

// ClientOption fills the option struct to configure intefaces, etc.

type ClientOption func(*clientOpts)

// SelectIPTraffic selects the type of IP packets (IPv4, IPv6, or both) this

// instance listens for.

// This does not guarantee that only mDNS entries of this sepcific

// type passes. E.g. typical mDNS packets distributed via IPv4, may contain

// both DNS A and AAAA entries.

func SelectIPTraffic(t IPType) ClientOption {

return func(o *clientOpts) {

o.listenOn = t

}

}

// SelectIfaces selects the interfaces to query for mDNS records

func SelectIfaces(ifaces []net.Interface) ClientOption {

return func(o *clientOpts) {

o.ifaces = ifaces

}

}

// WithClientConnFactory sets a custom connection factory for the client.

// This is primarily useful for testing with mock connections.

func WithClientConnFactory(factory api.ConnectionFactory) ClientOption {

return func(o *clientOpts) {

o.connFactory = factory

}

}

// WithClientInterfaceProvider sets a custom interface provider for the client.

// This is primarily useful for testing with mock interface lists.

func WithClientInterfaceProvider(provider api.InterfaceProvider) ClientOption {

return func(o *clientOpts) {

o.provider = provider

}

}

// Browse for all services of a given type in a given domain.

// Received entries are sent on the entries channel.

// It blocks until the context is canceled (or an error occurs).

func Browse(ctx context.Context, service, domain string, entries, removed chan<- *ServiceEntry, opts ...ClientOption) error {

cl, err := newClient(applyOpts(opts...))

if err != nil {

return err

}

params := defaultParams(service)

if domain != "" {

params.Domain = domain

}

params.Entries = entries

params.Removed = removed

params.isBrowsing = true

return cl.run(ctx, params)

}

// Lookup a specific service by its name and type in a given domain.

// Received entries are sent on the entries channel.

// It blocks until the context is canceled (or an error occurs).

func Lookup(ctx context.Context, instance, service, domain string, entries chan<- *ServiceEntry, opts ...ClientOption) error {

cl, err := newClient(applyOpts(opts...))

if err != nil {

return err

}

params := defaultParams(service)

params.Instance = instance

if domain != "" {

params.Domain = domain

}

params.Entries = entries

return cl.run(ctx, params)

}

func applyOpts(options ...ClientOption) clientOpts {

// Apply default configuration and load supplied options.

var conf = clientOpts{

listenOn: IPv4AndIPv6,

}

for _, o := range options {

if o != nil {

o(&conf)

}

}

return conf

}

func (c *Client) run(ctx context.Context, params *lookupParams) error {

// Run immediate sync on startup to catch any interfaces that changed

// between client creation and run()

c.syncInterfaces()

ctx, cancel := context.WithCancel(ctx)

// Start interface sync in background

syncDone := make(chan struct{})

go func() {

defer close(syncDone)

c.runInterfaceSync(ctx)

}()

done := make(chan struct{})

go func() {

defer close(done)

c.mainloop(ctx, params)

}()

// If previous probe was ok, it should be fine now. In case of an error later on,

// the entries' queue is closed.

err := c.periodicQuery(ctx, params)

cancel()

<-done

<-syncDone

return err

}

// defaultParams returns a default set of QueryParams.

func defaultParams(service string) *lookupParams {

return newLookupParams("", service, "local", false, make(chan *ServiceEntry), make(chan *ServiceEntry))

}

// NewClient creates a new mDNS client with the given options.

// This is the low-level constructor. For most use cases, prefer Browse() or Lookup().

func NewClient(opts ...ClientOption) (*Client, error) {

return newClient(applyOpts(opts...))

}

// newClient is the internal constructor that takes pre-applied options.

func newClient(opts clientOpts) (*Client, error) {

// Get interface provider (use default if not injected for testing)

provider := opts.provider

if provider == nil {

provider = NewInterfaceProvider()

}

ifaces := opts.ifaces

var requested []string

// Determine mode based on whether interfaces were explicitly provided

if len(ifaces) > 0 {

// Explicit mode: extract names for the manager

requested = make([]string, len(ifaces))

for i, iface := range ifaces {

requested[i] = iface.Name

}

} else {

// Dynamic mode: get current interfaces

ifaces = provider.MulticastInterfaces()

}

factory := opts.connFactory

if factory == nil {

factory = NewConnectionFactory()

}

// Create SEPARATE managers for IPv4 and IPv6.

// This ensures IPv6 failures don't affect IPv4 (and vice versa).

ipv4Mgr := NewInterfaceManager(ifaces, requested)

ipv6Mgr := NewInterfaceManager(ifaces, requested)

// IPv4 interfaces

var ipv4conn api.PacketConn

var err4 error

if (opts.listenOn & IPv4) > 0 {

ipv4conn, err4 = factory.CreateIPv4Conn(ifaces)

if err4 != nil {

log.Printf("[zeroconf] no suitable IPv4 interface: %s", err4.Error())

}

}

// IPv6 interfaces

var ipv6conn api.PacketConn

var err6 error

if (opts.listenOn & IPv6) > 0 {

ipv6conn, err6 = factory.CreateIPv6Conn(ifaces)

if err6 != nil {

log.Printf("[zeroconf] no suitable IPv6 interface: %s", err6.Error())

}

}

if err4 != nil && err6 != nil {

return nil, fmt.Errorf("no supported interface")

}

return &Client{

ipv4conn: ipv4conn,

ipv6conn: ipv6conn,

ipv4Mgr: ipv4Mgr,

ipv6Mgr: ipv6Mgr,

provider: provider,

}, nil

}

var cleanupFreq = 5 * time.Second

// Start listeners and waits for the shutdown signal from exit channel

func (c *Client) mainloop(ctx context.Context, params *lookupParams) {

// start listening for responses

msgCh := make(chan *dns.Msg, 32)

if c.ipv4conn != nil {

go c.recv(ctx, c.ipv4conn, msgCh)

}

if c.ipv6conn != nil {

go c.recv(ctx, c.ipv6conn, msgCh)

}

// Iterate through channels from listeners goroutines

var entries map[string]*ServiceEntry

sentEntries := make(map[string]*ServiceEntry)

ticker := time.NewTicker(cleanupFreq)

defer ticker.Stop()

for {

var now time.Time

select {

case <-ctx.Done():

// Context expired. Notify subscriber that we are done here.

params.done()

c.shutdown()

return

case t := <-ticker.C:

for k, e := range sentEntries {

if t.After(e.Expiry) {

params.Removed <- e

delete(sentEntries, k)

}

}

continue

case msg := <-msgCh:

now = time.Now()

entries = make(map[string]*ServiceEntry)

sections := append(msg.Answer, msg.Ns...)

sections = append(sections, msg.Extra...)

for _, answer := range sections {

switch rr := answer.(type) {

case *dns.PTR:

if params.ServiceName() != rr.Hdr.Name {

continue

}

if params.ServiceInstanceName() != "" && params.ServiceInstanceName() != rr.Ptr {

continue

}

if _, ok := entries[rr.Ptr]; !ok {

entries[rr.Ptr] = newServiceEntry(

trimDot(strings.ReplaceAll(rr.Ptr, rr.Hdr.Name, "")),

params.Service,

params.Domain)

}

entries[rr.Ptr].Expiry = now.Add(time.Duration(rr.Hdr.Ttl) * time.Second)

case *dns.SRV:

if params.ServiceInstanceName() != "" && params.ServiceInstanceName() != rr.Hdr.Name {

continue

} else if !strings.HasSuffix(rr.Hdr.Name, params.ServiceName()) {

continue

}

if _, ok := entries[rr.Hdr.Name]; !ok {

entries[rr.Hdr.Name] = newServiceEntry(

trimDot(strings.Replace(rr.Hdr.Name, params.ServiceName(), "", 1)),

params.Service,

params.Domain)

}

entries[rr.Hdr.Name].HostName = rr.Target

entries[rr.Hdr.Name].Port = int(rr.Port)

entries[rr.Hdr.Name].Expiry = now.Add(time.Duration(rr.Hdr.Ttl) * time.Second)

case *dns.TXT:

if params.ServiceInstanceName() != "" && params.ServiceInstanceName() != rr.Hdr.Name {

continue

} else if !strings.HasSuffix(rr.Hdr.Name, params.ServiceName()) {

continue

}

if _, ok := entries[rr.Hdr.Name]; !ok {

entries[rr.Hdr.Name] = newServiceEntry(

trimDot(strings.Replace(rr.Hdr.Name, params.ServiceName(), "", 1)),

params.Service,

params.Domain)

}

entries[rr.Hdr.Name].Text = rr.Txt

entries[rr.Hdr.Name].Expiry = now.Add(time.Duration(rr.Hdr.Ttl) * time.Second)

}

}

// Associate IPs in a second round as other fields should be filled by now.

for _, answer := range sections {

switch rr := answer.(type) {

case *dns.A:

for k, e := range entries {

if e.HostName == rr.Hdr.Name {

entries[k].AddrIPv4 = append(entries[k].AddrIPv4, rr.A)

}

}

case *dns.AAAA:

for k, e := range entries {

if e.HostName == rr.Hdr.Name {

entries[k].AddrIPv6 = append(entries[k].AddrIPv6, rr.AAAA)

}

}

}

}

}

if len(entries) > 0 {

for k, e := range entries {

if !e.Expiry.After(now) {

delete(entries, k)

if se, ok := sentEntries[k]; ok {

params.Removed <- se

delete(sentEntries, k)

}

continue

}

if se, ok := sentEntries[k]; ok {

// only resend if a value differes from the previously sent item

if e.HostName == se.HostName &&

e.Port == se.Port &&

reflect.DeepEqual(e.Text, se.Text) &&

e.Expiry.After(now) == se.Expiry.After(now) &&

reflect.DeepEqual(e.AddrIPv4, se.AddrIPv4) &&

reflect.DeepEqual(e.AddrIPv6, se.AddrIPv6) {

sentEntries[k].Expiry = e.Expiry

continue

}

}

// If this is an DNS-SD query do not throw PTR away.

// It is expected to have only PTR for enumeration

if params.ServiceTypeName() != params.ServiceName() {

// Require at least one resolved IP address for ServiceEntry

// TODO: wait some more time as chances are high both will arrive.

if len(e.AddrIPv4) == 0 && len(e.AddrIPv6) == 0 {

continue

}

}

// Submit entry to subscriber and cache it.

// This is also a point to possibly stop probing actively for a

// service entry.

params.Entries <- e

sentEntries[k] = e

if !params.isBrowsing {

params.disableProbing()

}

}

}

}

}

// Shutdown client will close currently open connections and channel implicitly.

func (c *Client) shutdown() {

if c.ipv4conn != nil {

err := c.ipv4conn.Close()

if err != nil {

log.Printf("[zeroconf] unable to cleanly close client IPv4 connection: %s", err.Error())

}

}

if c.ipv6conn != nil {

err := c.ipv6conn.Close()

if err != nil {

log.Printf("[zeroconf] unable to cleanly close client IPv6 connection: %s", err.Error())

}

}

}

// Data receiving routine reads from connection, unpacks packets into dns.Msg

// structures and sends them to a given msgCh channel

func (c *Client) recv(ctx context.Context, conn api.PacketConn, msgCh chan *dns.Msg) {

if conn == nil {

return

}

buf := make([]byte, 65536)

var fatalErr error

for {

// Handles the following cases:

// - ReadFrom aborts with error due to closed UDP connection -> causes ctx cancel

// - ReadFrom aborts otherwise.

if ctx.Err() != nil || fatalErr != nil {

return

}

n, _, _, err := conn.ReadFrom(buf)

if err != nil {

fatalErr = err

continue

}

msg := new(dns.Msg)

if err := msg.Unpack(buf[:n]); err != nil {

// log.Printf("[WARN] mdns: Failed to unpack packet: %v", err)

continue

}

select {

case msgCh <- msg:

// Submit decoded DNS message and continue.

case <-ctx.Done():

// Abort.

return

}

}

}

// periodicQuery sens multiple probes until a valid response is received by

// the main processing loop or some timeout/cancel fires.

// TODO: move error reporting to shutdown function as periodicQuery is called from

// go routine context.

func (c *Client) periodicQuery(ctx context.Context, params *lookupParams) error {

// Do the first query immediately.

if err := c.query(params); err != nil {

return err

}

const maxInterval = 60 * time.Second

interval := initialQueryInterval

timer := time.NewTimer(interval)

defer timer.Stop()

for {

select {

case <-timer.C:

// Wait for next iteration.

case <-params.stopProbing:

// Chan is closed (or happened in the past).

// Done here. Received a matching mDNS entry.

return nil

case <-ctx.Done():

if params.isBrowsing {

return nil

}

return ctx.Err()

}

if err := c.query(params); err != nil {

return err

}

// Exponential increase of the interval with jitter:

// the new interval will be between 1.5x and 2.5x the old interval, capped at maxInterval.

if interval != maxInterval {

interval += time.Duration(rand.Int63n(interval.Nanoseconds())) + interval/2

if interval > maxInterval {

interval = maxInterval

}

}

timer.Reset(interval)

}

}

// Performs the actual query by service name (browse) or service instance name (lookup),

// start response listeners goroutines and loops over the entries channel.

func (c *Client) query(params *lookupParams) error {

var serviceName, serviceInstanceName string

serviceName = fmt.Sprintf("%s.%s.", trimDot(params.Service), trimDot(params.Domain))

// send the query

m := new(dns.Msg)

if params.Instance != "" { // service instance name lookup

serviceInstanceName = fmt.Sprintf("%s.%s", params.Instance, serviceName)

m.Question = []dns.Question{

{Name: serviceInstanceName, Qtype: dns.TypeSRV, Qclass: dns.ClassINET},

{Name: serviceInstanceName, Qtype: dns.TypeTXT, Qclass: dns.ClassINET},

}

} else if len(params.Subtypes) > 0 { // service subtype browse

m.SetQuestion(params.Subtypes[0], dns.TypePTR)

} else { // service name browse

m.SetQuestion(serviceName, dns.TypePTR)

}

m.RecursionDesired = false

return c.sendQuery(m)

}

// sendQuery packs the dns.Msg and writes to available connections (multicast).

//

// THE CRITICAL FIX: Dynamic iteration using ActiveIndices().

// Gets a fresh snapshot of active indices on each call. The snapshot may become

// stale during iteration (race with syncInterfaces), but this is BENIGN because:

// - Sends to removed indices fail immediately

// - MarkFailed is idempotent (safe to call on already-removed index)

// - New indices are picked up on the next sendQuery call

func (c *Client) sendQuery(msg *dns.Msg) error {

buf, err := msg.Pack()

if err != nil {

return err

}

// IPv4: iterate over CURRENT active indices

if c.ipv4conn != nil {

for _, idx := range c.ipv4Mgr.ActiveIndices() {

if _, err := c.ipv4conn.WriteTo(buf, idx, ipv4Addr); err != nil {

c.ipv4Mgr.MarkFailed(idx, err)

}

}

}

// IPv6: same pattern, separate manager

if c.ipv6conn != nil {

for _, idx := range c.ipv6Mgr.ActiveIndices() {

if _, err := c.ipv6conn.WriteTo(buf, idx, ipv6Addr); err != nil {

c.ipv6Mgr.MarkFailed(idx, err)

}

}

}

return nil

}

// runInterfaceSync periodically polls for interface changes.

func (c *Client) runInterfaceSync(ctx context.Context) {

ticker := time.NewTicker(10 * time.Second)

defer ticker.Stop()

for {

select {

case <-ctx.Done():

return

case <-ticker.C:

c.syncInterfaces()

}

}

}

// syncInterfaces polls for interface changes and recovers interfaces.

func (c *Client) syncInterfaces() {

current := c.provider.MulticastInterfaces()

// Helper to sync a single manager

syncManager := func(mgr *InterfaceManager, conn api.PacketConn, groupIP net.IP) {

if conn == nil || mgr == nil {

return

}

for _, iface := range mgr.Sync(current) {

if err := conn.JoinGroup(&iface, &net.UDPAddr{IP: groupIP}); err != nil {

mgr.SetBackoff(iface.Name)

} else {

mgr.Activate(iface)

}

}

}

syncManager(c.ipv4Mgr, c.ipv4conn, mdnsGroupIPv4)

syncManager(c.ipv6Mgr, c.ipv6conn, mdnsGroupIPv6)

}