process_exchange.go

/*
File:    process_exchange.go
Version: 2.42.0
Updated: 11-May-2026 14:31 CEST

Description:
  Upstream Exchange and Payload Transformation for sdproxy.
  Executes the final phase of the core processing pipeline natively:
    - QNAME Label Bounds Constraint Inspection
    - SingleFlight Connection Coalescing
    - Answer Sorting & Payload Optimization
    - Target Name Policy Inspection
    - Response IP Filtering and Rebinding Defenses
    - Buffer capacity & Telemetry execution

  Extracted from process.go to promote advanced modularity.

Changes:
  2.42.0 - [REFACTOR] Instantiated the `RcodeStr` helper to gracefully construct 
           logging parameters organically across the network dialect strings.
  2.41.0 - [REFACTOR] Adopted the centralized `RecordBlockEvent` telemetry function, 
           ensuring cached upstream payloads accurately record analytics naturally.
  2.40.0 - [SECURITY/FIX] Clarified the `QNAME Label Bounds` telemetry reason text. 
           If an oversized, natively spoofed alias target (e.g., `forcesafesearch...`) 
           violates upstream constraints on behalf of a standard user query, the 
           resulting block log natively appends the Target context organically. 
           Prevents administrative confusion over seemingly legal requests.
  2.39.0 - [SECURITY] Incorporated the `QNAME Label Bounds` evaluation explicitly 
           into the upstream pipeline. Identifies and safely terminates malformed 
           or massive domain strings dynamically before executing exterior resolvers, 
           while robustly accommodating grouped exemptions natively (`ignore_qname_labels`).
*/

package main

import (
	"fmt"
	"log"
	"net/netip"

	"github.com/miekg/dns"
)

// executeUpstreamExchange performs the outbound network dial, validates the payload responses,
// executes modifications, and replies to the client natively.
func executeUpstreamExchange(
	w dns.ResponseWriter, r *dns.Msg, q dns.Question, qNameTrimmed string,
	doBit bool, originalID uint16, cacheKey DNSCacheKey,
	routeName string, routeIdx uint16, routeOriginType string,
	clientID, clientName, clientMAC, clientIP string, clientAddr netip.Addr,
	protocol, sni, path string,
	parentalForcedTTL uint32, parentalReason, parentalCategory, parentalMatchedApex string,
	bypassPolicies bool,
	originalQName, originalQNameTrimmed, spoofedAlias string,
) {

	// ── 1. Upstream selection + pre-validation ────────────────────────────
	group, exists := routeUpstreams[routeName]
	if !exists || len(group.Servers) == 0 {
		group = routeUpstreams["default"]
	}
	if group == nil || len(group.Servers) == 0 {
		log.Printf("[DNS] [%s] %s -> %s %s | ROUTE: %s (%s) | NO UPSTREAMS CONFIGURED",
			protocol, clientID, originalQName, dns.TypeToString[q.Qtype],
			routeName, routeOriginType)
		dns.HandleFailed(w, r)
		return
	}

	// ── 1.5 EDNS0 Client Subnet (ECS) Log Inspection ──────────────────────
	var clientSentECS bool
	if opt := r.IsEdns0(); opt != nil {
		for _, o := range opt.Option {
			if _, ok := o.(*dns.EDNS0_SUBNET); ok {
				clientSentECS = true
				break
			}
		}
	}

	ecsMark := ""
	if group.ECSAction == "add" && clientAddr.IsValid() {
		var m int
		if clientAddr.Is4() {
			m = group.ECSV4Mask
		} else {
			m = group.ECSV6Mask
		}
		ecsMark = fmt.Sprintf(" [ECS: ADD /%d]", m)
	} else if group.ECSAction == "pass" && clientSentECS {
		ecsMark = " [ECS: PASS]"
	}

	// ── 1.8 QNAME Label Bounds Validation ─────────────────────────────────
	if !group.IgnoreQnameLabels {
		labels := countDomainLabels(qNameTrimmed)
		if labels < cfg.Server.QnameMinLabels || labels > cfg.Server.QnameMaxLabels {
			IncrPolicyBlock()
			
			// [SECURITY/FIX] Append the Target evaluation structure transparently 
			// if the bounds violation originated from a CNAME Spoofed Alias natively.
			reason := fmt.Sprintf("QNAME Label Bounds (%d labels)", labels)
			if originalQNameTrimmed != qNameTrimmed {
				reason += fmt.Sprintf(" (Target: %s)", qNameTrimmed)
			}
			
			RecordBlockEvent(clientIP, originalQNameTrimmed, reason)

			dropped := writePolicyAction(w, r, PolicyActionBlock)

			if cacheSynthFlag && !dropped && globalBlockAction != BlockActionDrop {
				synthMsg := buildSynthCacheMsg(q, PolicyActionBlock)
				CacheSetSynth(cacheKey, synthMsg)
				msgPool.Put(synthMsg) // [PERF] Zero-allocation memory recycling
			}

			if logQueries {
				var actionLogStr string
				if globalBlockAction == BlockActionLog {
					actionLogStr = "LOG ONLY"
				} else if globalBlockAction == BlockActionDrop {
					actionLogStr = "DROP"
				} else {
					actionLogStr = getBlockActionLogStr(q.Qtype)
				}

				statusMark := "POLICY BLOCK"
				if dropped {
					statusMark = "POLICY DROP"
				}
				if spoofedAlias != "" {
					statusMark = fmt.Sprintf("SPOOFED ALIAS (%s) | %s", spoofedAlias, statusMark)
				}

				log.Printf("[DNS] [%s] %s -> %s %s | %s (Label Bounds) | %s",
					protocol, clientID, originalQName, dns.TypeToString[q.Qtype], statusMark, actionLogStr)
			}
			return
		}
	}

	// ── 2. Upstream forwarding — coalesced via singleflight ───────────────
	sfClientName := ""
	if hasClientNameUpstream {
		sfClientName = clientName
	}
	// Note: The RouteIdx inherently isolates Cache keys, so specific subnets 
	// utilizing distinct upstream groups will securely partition the cache seamlessly.
	sfKey       := buildSFKey(qNameTrimmed, q.Qtype, routeIdx, doBit, r.CheckingDisabled, sfClientName)
	didUpstream := false
	
	v, sfErr, shared := sfGroup.Do(sfKey, func() (any, error) {
		didUpstream = true
		IncrUpstreamCall()
		
		ctx, cancel := newUpstreamCtx()
		defer cancel()
		
		// The active routing pipeline transmits the underlying subnet to intelligently populate ECS
		msg, addr, err := group.Exchange(ctx, r, clientID, clientName, clientAddr)
		if err != nil || msg == nil {
			// [FEAT] Infinite Serve-Stale Fallback
			// Intercept upstream outages and connection timeouts natively by 
			// probing the cache arrays for expired records as an absolute last resort.
			if cfg.Cache.ServeStaleInfinite {
				fallbackMsg := msgPool.Get().(*dns.Msg)
				*fallbackMsg = dns.Msg{}
				if CacheGetExpired(cacheKey, fallbackMsg) {
					return sfResult{msg: fallbackMsg, addr: "stale-fallback"}, nil
				}
				msgPool.Put(fallbackMsg)
			}
			return sfResult{msg: msg, addr: addr}, err
		}
		isNeg := msg.Rcode == dns.RcodeNameError || (msg.Rcode == dns.RcodeSuccess && len(msg.Answer) == 0)
		if !isNeg || cacheUpstreamNeg {
			CacheSet(cacheKey, msg, routeName)
		}
		return sfResult{msg: msg, addr: addr}, nil
	})
	
	if shared && !didUpstream {
		coalescedTotal.Add(1)
	}

	var finalResp *dns.Msg
	var upstreamUsed string
	if v != nil {
		if res, ok := v.(sfResult); ok {
			upstreamUsed = res.addr
			if res.msg != nil && sfErr == nil {
				if shared {
					finalResp = res.msg.Copy()
				} else {
					finalResp = res.msg
				}
			}
		}
	}

	// ── 3. Error Handle ───────────────────────────────────────────────────
	if finalResp == nil {
		if sfErr != nil && sfErr.Error() == "silent_drop" {
			if logQueries {
				log.Printf("[DNS] [%s] %s -> %s %s | ROUTE: %s (%s) | CONSENSUS DROP | DROP",
					protocol, clientID, originalQName, dns.TypeToString[q.Qtype],
					routeName, routeOriginType)
			}
			return // Natively shed the load without responding
		}

		upstreamLog := ""
		if upstreamUsed != "" {
			upstreamLog = " | UPSTREAM: " + cleanUpstreamHost(upstreamUsed)
		}
		log.Printf("[DNS] [%s] %s -> %s %s | ROUTE: %s (%s)%s | FAILED: %v",
			protocol, clientID, originalQName, dns.TypeToString[q.Qtype],
			routeName, routeOriginType, upstreamLog, sfErr)
		dns.HandleFailed(w, r)
		return
	}

	// ── 4. Answer Sorting ─────────────────────────────────────────────────
	// MUST execute BEFORE any client-specific payload mutations (like TTL Caps
	// or DNSSEC stripping) to prevent corrupting the global cache byte arrays.
	if cfg.Cache.AnswerSort != "none" {
		if applyAnswerSort(finalResp, cfg.Cache.AnswerSort) {
			// [PERFORMANCE] Prevent severe CPU starvation and lock contention. 
			// 'random' sorting shuffles records ephemerally for the client. 
			// If 'shared' is true, this is a coalesced SingleFlight cache-miss.
			// Only the primary worker executes CacheUpdateOrder to prevent parallel goroutines from thrashing.
			if cfg.Cache.AnswerSort != "random" && !shared {
				CacheUpdateOrder(cacheKey, finalResp)
			}
		}
	}

	// ── 5. Target Name Policy Check (Cache Miss) ──────────────────────────
	if checkTargetNames(w, r, finalResp, clientMAC, clientIP, clientAddr, clientName, clientID, protocol, sni, path, bypassPolicies, originalQName) {
		return
	}

	// ── 6. Response transforms ────────────────────────────────────────────
	finalResp = transformResponse(finalResp, q.Qtype, doBit, true)

	// ── 7. IP Filter Policy Check (Cache Miss) ────────────────────────────
	if filterResponseIPs(w, r, finalResp, clientMAC, clientIP, clientAddr, clientName, clientID, protocol, sni, path, bypassPolicies, originalQName, originalQNameTrimmed) {
		return
	}

	// ── 8. DNS Rebinding Protection (Cache Miss) ──────────────────────────
	if filterRebinding(w, r, finalResp, clientIP, clientAddr, clientName, clientID, protocol, originalQName, originalQNameTrimmed) {
		return
	}

	// ── 9. NULL-IP Detection (Pre-Truncation) ─────────────────────────────
	// Evaluate NULL-IPs natively BEFORE UDP Fragmentation defense executes.
	// This guarantees that artificially truncated payloads (where Answer == nil)
	// do not silently evade telemetry trackers.
	isNullIP := responseContainsNullIP(finalResp)
	if isNullIP {
		IncrPolicyBlock()
		RecordBlockEvent(clientIP, originalQNameTrimmed, "Upstream NULL-IP ("+routeName+")")
	}

	// ── 10. UDP Amplification & Fragmentation Defense ──────────────────────
	upstreamSize, clientAdvertised := enforceUDPDefenses(r, finalResp, protocol)

	// ── 11. Final reply ───────────────────────────────────────────────────
	if parentalForcedTTL > 0 {
		CapResponseTTL(finalResp, parentalForcedTTL)
	}
	finalResp.Id = originalID
	
	IncrReturnCode(finalResp.Rcode, isNullIP)
	if finalResp.Rcode == dns.RcodeNameError {
		IncrNXDomain(originalQNameTrimmed)
	}

	if upstreamUsed != "" {
		IncrUpstreamHost(cleanUpstreamHost(upstreamUsed))
	}
	
	w.WriteMsg(finalResp)

	// [MEMORY LIFECYCLE & GC]
	// Note: `finalResp` is inherently generated via a network `Unpack()` allocation 
	// from the upstream dialer (or cloned during a SingleFlight coalescing event).
	// It is intentionally NOT placed back into the `msgPool` to be safely garbage-collected.
	// This natively prevents pool pollution from external, variable-sized network arrays 
	// and guarantees pristine baseline capacities for local synthetic messages.

	if logQueries {
		rcodeStr := RcodeStr(finalResp.Rcode)
		if isNullIP {
			rcodeStr += " (NULL-IP)"
		}

		var status string
		switch {
		case shared && !didUpstream:
			status = "COALESCED | " + rcodeStr
		case isNullIP:
			status = rcodeStr + " (NULL-IP)"
		case finalResp.Truncated:
			status = fmt.Sprintf("TRUNCATED (TC=1, Upstream:%dB, Client:%dB) | %s", upstreamSize, clientAdvertised, rcodeStr)
		}

		if parentalReason == "FREE" {
			if parentalCategory != "" {
				status += fmt.Sprintf(" (PARENTAL FREE: %s, apex: %s)", parentalCategory, parentalMatchedApex)
			} else {
				status += " (PARENTAL FREE)"
			}
		} else if parentalReason == "ALLOW" {
			if parentalCategory != "" {
				status += fmt.Sprintf(" (PARENTAL ALLOW: %s, apex: %s)", parentalCategory, parentalMatchedApex)
			} else {
				status += " (PARENTAL ALLOW)"
			}
		} else if parentalReason == "LOG" {
			if parentalCategory != "" {
				status += fmt.Sprintf(" (PARENTAL LOG: %s, apex: %s)", parentalCategory, parentalMatchedApex)
			} else {
				status += " (PARENTAL LOG)"
			}
		} else if parentalCategory != "" {
			status += fmt.Sprintf(" (CATEGORY: %s, apex: %s)", parentalCategory, parentalMatchedApex)
		}
		
		if upstreamUsed == "stale-fallback" {
			status = "INFINITE STALE FALLBACK | " + status
		}
		
		if spoofedAlias != "" {
			status = fmt.Sprintf("SPOOFED ALIAS (%s) | %s", spoofedAlias, status)
		}

		if ecsMark != "" {
			status += ecsMark
		}
		
		log.Printf("[DNS] [%s] %s -> %s %s | ROUTE: %s (%s) | UPSTREAM: %s (%s) | %s",
			protocol, clientID, originalQName, dns.TypeToString[q.Qtype],
			routeName, routeOriginType, upstreamUsed, routeName, status)
	}
}