DNS防拦截的常用技术实现与实战指南

// 使用Node.js实现DoH客户端
const https = require('https');
const dnsPacket = require('dns-packet');
 
class DoHClient {
  constructor(dohServer = 'https://cloudflare-dns.com/dns-query') {
    this.dohServer = dohServer;
  }
 
  async query(domain, type = 'A') {
    const dnsQuery = dnsPacket.encode({
      type: 'query',
      id: Math.floor(Math.random() * 65536),
      flags: dnsPacket.RECURSION_DESIRED,
      questions: [{
        type: type,
        name: domain
      }]
    });
 
    return new Promise((resolve, reject) => {
      const options = {
        hostname: new URL(this.dohServer).hostname,
        path: '/dns-query?dns=' + dnsQuery.toString('base64url'),
        method: 'GET',
        headers: {
          'Accept': 'application/dns-message',
          'User-Agent': 'DoH-Client/1.0'
        }
      };
 
      const req = https.request(options, (res) => {
        let data = [];
        res.on('data', chunk => data.push(chunk));
        res.on('end', () => {
          try {
            const response = dnsPacket.decode(Buffer.concat(data));
            resolve(response);
          } catch (err) {
            reject(err);
          }
        });
      });
 
      req.on('error', reject);
      req.end();
    });
  }
}
 
// 使用示例
const client = new DoHClient();
client.query('example.com').then(response => {
  console.log('DNS响应:', response);
}).catch(err => {
  console.error('查询失败:', err);
});

import asyncio
import ssl
import dns.message
import dns.query
import dns.resolver
 
class DoTResolver:
    def __init__(self, server='1.1.1.1', port=853):
        self.server = server
        self.port = port
        self.ssl_context = ssl.create_default_context()
        self.ssl_context.check_hostname = False
        self.ssl_context.verify_mode = ssl.CERT_NONE
    
    async def query(self, domain, record_type='A'):
        """执行DoT查询"""
        query = dns.message.make_query(domain, record_type)
        
        try:
            response = await dns.query.tls(
                query, 
                self.server, 
                timeout=5.0,
                ssl_context=self.ssl_context
            )
            return response
        except Exception as e:
            print(f"DoT查询失败: {e}")
            return None
 
# 使用示例
async def main():
    resolver = DoTResolver('8.8.8.8')
    result = await resolver.query('google.com')
    if result:
        for answer in result.answer:
            print(f"解析结果: {answer}")
 
if __name__ == "__main__":
    asyncio.run(main())

package main
 
import (
    "fmt"
    "log"
    "github.com/jedisct1/go-dnsstamps"
    "github.com/jedisct1/dnscrypt-proxy/dnscrypt"
)
 
type DNSCryptClient struct {
    resolver *dnscrypt.Resolver
}
 
func NewDNSCryptClient(stamp string) (*DNSCryptClient, error) {
    serverStamp, err := dnsstamps.NewServerStampFromString(stamp)
    if err != nil {
        return nil, err
    }
    
    resolver := &dnscrypt.Resolver{
        ServerStamp: serverStamp,
        Timeout:     5,
    }
    
    return &DNSCryptClient{resolver: resolver}, nil
}
 
func (c *DNSCryptClient) Query(domain string, qtype uint16) ([]string, error) {
    response, err := c.resolver.Query(domain, qtype)
    if err != nil {
        return nil, err
    }
    
    var results []string
    for _, answer := range response.Answer {
        results = append(results, answer.String())
    }
    
    return results, nil
}
 
func main() {
    // 使用Cloudflare的DNSCrypt服务
    stamp := "sdns://AQcAAAAAAAAABzEuMC4wLjEgIOgT4WJpMCV8rbO4yV7JcVvL6zmz0JEuI5c2b4dYK1x4Mi5kbnNjcnlwdC1jZXJ0LmNsb3VkZmxhcmUuY29t"
    
    client, err := NewDNSCryptClient(stamp)
    if err != nil {
        log.Fatal(err)
    }
    
    results, err := client.Query("example.com", 1) // A记录
    if err != nil {
        log.Fatal(err)
    }
    
    fmt.Printf("DNSCrypt查询结果: %v\n", results)
}

class SmartDNSResolver {
  constructor(servers = []) {
    this.servers = servers;
    this.latencyMap = new Map();
    this.failureCount = new Map();
  }
 
  async measureLatency(server) {
    const start = Date.now();
    try {
      await this.pingServer(server);
      return Date.now() - start;
    } catch (err) {
      return Infinity;
    }
  }
 
  async pingServer(server) {
    // 实现服务器ping逻辑
    return new Promise((resolve, reject) => {
      setTimeout(() => resolve(true), 100); // 模拟ping操作
    });
  }
 
  async selectBestServer() {
    const healthyServers = this.servers.filter(server => {
      const failures = this.failureCount.get(server) || 0;
      return failures < 3; // 失败次数少于3次
    });
 
    if (healthyServers.length === 0) {
      throw new Error('没有可用的DNS服务器');
    }
 
    // 测量延迟并排序
    const latencyPromises = healthyServers.map(async server => {
      const latency = await this.measureLatency(server);
      return { server, latency };
    });
 
    const results = await Promise.all(latencyPromises);
    results.sort((a, b) => a.latency - b.latency);
 
    return results[0].server;
  }
 
  async resolveWithFallback(domain) {
    const maxRetries = 3;
    
    for (let i = 0; i < maxRetries; i++) {
      try {
        const server = await this.selectBestServer();
        const result = await this.queryServer(server, domain);
        
        // 重置失败计数
        this.failureCount.set(server, 0);
        return result;
        
      } catch (err) {
        console.error(`第${i + 1}次解析失败:`, err.message);
        
        if (i === maxRetries - 1) {
          throw new Error('所有DNS服务器都无法解析该域名');
        }
        
        // 等待一段时间后重试
        await new Promise(resolve => setTimeout(resolve, 1000 * (i + 1)));
      }
    }
  }
 
  async queryServer(server, domain) {
    // 实现具体的DNS查询逻辑
    // 这里可以集成DoH、DoT或传统DNS查询
    return { domain, ip: '8.8.8.8', server };
  }
}
 
// 使用示例
const resolver = new SmartDNSResolver([
  'https://cloudflare-dns.com/dns-query',
  'https://dns.google/dns-query',
  'https://dns.quad9.net/dns-query'
]);
 
resolver.resolveWithFallback('example.com')
  .then(result => console.log('解析成功:', result))
  .catch(err => console.error('解析失败:', err));

import geoip2.database
import dns.resolver
from typing import List, Dict, Optional
 
class GeoDNSResolver:
    def __init__(self, geo_db_path: str = 'GeoLite2-City.mmdb'):
        self.geo_reader = geoip2.database.Reader(geo_db_path)
        self.dns_servers = {
            'asia': ['8.8.8.8', '1.1.1.1'],
            'europe': ['8.8.8.8', '208.67.222.222'],
            'america': ['8.8.8.8', '9.9.9.9']
        }
    
    def get_user_location(self, user_ip: str) -> str:
        """根据用户IP确定地理位置"""
        try:
            response = self.geo_reader.city(user_ip)
            country = response.country.iso_code
            
            # 简化的地理区域划分
            if country in ['CN', 'JP', 'KR', 'SG']:
                return 'asia'
            elif country in ['US', 'CA', 'BR']:
                return 'america'
            elif country in ['GB', 'DE', 'FR', 'IT']:
                return 'europe'
            else:
                return 'global'
        except Exception:
            return 'global'
    
    def get_optimal_dns_servers(self, user_ip: str) -> List[str]:
        """获取最优的DNS服务器列表"""
        region = self.get_user_location(user_ip)
        return self.dns_servers.get(region, self.dns_servers['america'])
    
    def resolve_with_geo_optimization(self, domain: str, user_ip: str) -> Optional[Dict]:
        """基于地理位置优化的DNS解析"""
        servers = self.get_optimal_dns_servers(user_ip)
        
        for server in servers:
            try:
                resolver = dns.resolver.Resolver()
                resolver.nameservers = [server]
                
                response = resolver.resolve(domain, 'A')
                return {
                    'domain': domain,
                    'ips': [str(rdata) for rdata in response],
                    'server': server,
                    'region': self.get_user_location(user_ip)
                }
            except Exception as e:
                print(f"服务器 {server} 解析失败: {e}")
                continue
        
        return None
 
# 使用示例
if __name__ == "__main__":
    resolver = GeoDNSResolver()
    result = resolver.resolve_with_geo_optimization('google.com', '8.8.8.8')
    if result:
        print(f"地理优化解析结果: {result}")

// TRAE IDE会自动识别dns-packet库的方法
const dnsPacket = require('dns-packet');
 
// 智能提示：dnsPacket.encode() 的参数结构
const query = dnsPacket.encode({
  type: 'query',
  id: 12345,
  flags: dnsPacket.RECURSION_DESIRED, // IDE会提示可用的flag常量
  questions: [{
    type: 'A', // IDE会提示所有支持的记录类型
    name: 'example.com'
  }]
});

// 使用TRAE IDE的网络监控功能
const https = require('https');
 
// IDE会自动捕获并显示HTTPS请求详情
const req = https.request(options, (res) => {
  console.log('状态码:', res.statusCode);
  console.log('响应头:', res.headers);
  
  // TRAE IDE会在调试面板显示完整的请求-响应流程
  res.on('data', (chunk) => {
    // 数据处理逻辑
  });
});

{
  "trae-ide": {
    "multi-language": {
      "dns-implementation": {
        "javascript": "DoH客户端实现",
        "python": "DoT解析器实现", 
        "go": "DNSCrypt客户端实现"
      }
    }
  }
}

# 在TRAE IDE终端中测试DNS解析
dig @1.1.1.1 example.com +tls-port=853
 
# 测试DoH服务
curl -H 'Accept: application/dns-json' \
     'https://cloudflare-dns.com/dns-query?name=example.com&type=A'
 
# 监控DNS查询延迟
ping -c 5 1.1.1.1

# docker-compose.yml - 企业DNS安全网关
dns-security-gateway:
  image: coredns/coredns
  ports:
    - "53:53/udp"
    - "853:853/tcp"  # DoT端口
    - "443:443/tcp"  # DoH端口
  volumes:
    - ./Corefile:/etc/coredns/Corefile
    - ./certs:/etc/coredns/certs
  environment:
    - COREDNS_POD_NAMESPACE=dns-security
  networks:
    - dns-network
 
networks:
  dns-network:
    driver: bridge

# Corefile - CoreDNS配置
.:53 {
    forward . 1.1.1.1 8.8.8.8 {
        max_concurrent 1000
        expire 10s
        health_check 5s
    }
    
    # 启用DNS over TLS
    tls://.:853 {
        tls cert.pem key.pem
        forward . 1.1.1.1 8.8.8.8
    }
    
    # 启用DNS over HTTPS
    https://.:443 {
        tls cert.pem key.pem
        forward . 1.1.1.1 8.8.8.8
    }
    
    cache 3600
    log
    errors
}

import Foundation
import Network
 
class SecureDNSResolver {
    private let monitor = NWPathMonitor()
    private let queue = DispatchQueue(label: "dns-monitor")
    
    func setupSecureDNS() {
        // 配置DoH服务器
        let dohServers = [
            "https://cloudflare-dns.com/dns-query",
            "https://dns.google/dns-query"
        ]
        
        // 监控网络状态变化
        monitor.pathUpdateHandler = { [weak self] path in
            if path.status == .satisfied {
                self?.selectBestDNSServer(from: dohServers)
            }
        }
        
        monitor.start(queue: queue)
    }
    
    private func selectBestDNSServer(from servers: [String]) {
        var bestServer: String?
        var minLatency = Double.greatestFiniteMagnitude
        
        let group = DispatchGroup()
        
        for server in servers {
            group.enter()
            measureLatency(for: server) { latency in
                if latency < minLatency {
                    minLatency = latency
                    bestServer = server
                }
                group.leave()
            }
        }
        
        group.notify(queue: .main) {
            if let server = bestServer {
                self.configureDNS(with: server)
            }
        }
    }
    
    private func measureLatency(for server: String, completion: @escaping (Double) -> Void) {
        let startTime = CFAbsoluteTimeGetCurrent()
        
        // 执行DNS查询并测量延迟
        let task = URLSession.shared.dataTask(with: URL(string: server)!) { _, _, _ in
            let latency = CFAbsoluteTimeGetCurrent() - startTime
            completion(latency)
        }
        
        task.resume()
    }
    
    private func configureDNS(with server: String) {
        // 应用DNS配置到应用网络层
        print("使用最优DNS服务器: \(server)")
    }
}

const EventEmitter = require('events');
 
class DNSMonitor extends EventEmitter {
  constructor(config = {}) {
    super();
    this.config = {
      checkInterval: 30000, // 30秒检查一次
      timeout: 5000,
      failureThreshold: 3,
      ...config
    };
    this.stats = new Map();
    this.isMonitoring = false;
  }
 
  startMonitoring() {
    if (this.isMonitoring) return;
    
    this.isMonitoring = true;
    this.monitorLoop();
  }
 
  stopMonitoring() {
    this.isMonitoring = false;
  }
 
  async monitorLoop() {
    while (this.isMonitoring) {
      await this.checkDNSServers();
      await this.sleep(this.config.checkInterval);
    }
  }
 
  async checkDNSServers() {
    const servers = [
      { name: 'Cloudflare DoH', url: 'https://cloudflare-dns.com/dns-query' },
      { name: 'Google DoH', url: 'https://dns.google/dns-query' },
      { name: 'Quad9 DoH', url: 'https://dns.quad9.net/dns-query' }
    ];
 
    for (const server of servers) {
      try {
        const startTime = Date.now();
        await this.testDNSQuery(server.url);
        const latency = Date.now() - startTime;
        
        this.updateStats(server.name, {
          latency,
          status: 'healthy',
          lastCheck: new Date(),
          consecutiveFailures: 0
        });
 
        this.emit('dnsHealthy', server.name, latency);
        
      } catch (error) {
        this.updateStats(server.name, {
          status: 'unhealthy',
          lastCheck: new Date(),
          error: error.message,
          consecutiveFailures: (this.getStats(server.name).consecutiveFailures || 0) + 1
        });
 
        if (this.getStats(server.name).consecutiveFailures >= this.config.failureThreshold) {
          this.emit('dnsUnhealthy', server.name, error.message);
        }
      }
    }
  }
 
  async testDNSQuery(dohUrl) {
    // 实现DNS查询测试逻辑
    return new Promise((resolve, reject) => {
      setTimeout(() => resolve(true), 100); // 模拟测试
    });
  }
 
  updateStats(server, data) {
    this.stats.set(server, { ...this.getStats(server), ...data });
  }
 
  getStats(server) {
    return this.stats.get(server) || {};
  }
 
  getAllStats() {
    return Object.fromEntries(this.stats);
  }
 
  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}
 
// 使用示例
const monitor = new DNSMonitor();
 
monitor.on('dnsHealthy', (server, latency) => {
  console.log(`✅ ${server} 正常，延迟: ${latency}ms`);
});
 
monitor.on('dnsUnhealthy', (server, error) => {
  console.error(`❌ ${server} 异常: ${error}`);
  // 发送告警通知
  sendAlert(`DNS服务器异常: ${server} - ${error}`);
});
 
function sendAlert(message) {
  // 实现告警发送逻辑
  console.log(`🚨 告警: ${message}`);
}
 
// 启动监控
monitor.startMonitoring();
 
// 每5分钟输出统计报告
setInterval(() => {
  const stats = monitor.getAllStats();
  console.log('DNS监控统计:', JSON.stringify(stats, null, 2));
}, 5 * 60 * 1000);

import dns.resolver
import dns.dnssec
import dns.name
import dns.message
import dns.query
 
class DNSSECValidator:
    def __init__(self):
        self.resolver = dns.resolver.Resolver()
        self.resolver.use_edns(0, dns.flags.DO, 4096)
    
    def validate_domain(self, domain: str) -> dict:
        """验证域名的DNSSEC签名"""
        try:
            # 获取DNSKEY记录
            dnskey_answer = self.resolver.resolve(domain, 'DNSKEY')
            
            # 获取DS记录（父区域授权）
            ds_answer = self.resolver.resolve(domain, 'DS')
            
            # 获取RRSIG记录
            rrsig_answer = self.resolver.resolve(domain, 'RRSIG')
            
            # 执行DNSSEC验证
            is_valid = self.perform_dnssec_validation(domain, dnskey_answer, ds_answer)
            
            return {
                'domain': domain,
                'dnssec_enabled': True,
                'validation_status': 'valid' if is_valid else 'invalid',
                'dnskey_count': len(dnskey_answer),
                'ds_count': len(ds_answer),
                'rrsig_count': len(rrsig_answer)
            }
            
        except dns.resolver.NXDOMAIN:
            return {
                'domain': domain,
                'dnssec_enabled': False,
                'error': 'Domain not found'
            }
        except dns.resolver.NoAnswer:
            return {
                'domain': domain,
                'dnssec_enabled': False,
                'error': 'No DNSSEC records found'
            }
        except Exception as e:
            return {
                'domain': domain,
                'dnssec_enabled': False,
                'error': str(e)
            }
    
    def perform_dnssec_validation(self, domain, dnskey_answer, ds_answer):
        """执行DNSSEC签名验证"""
        try:
            # 这里应该实现完整的DNSSEC验证逻辑
            # 包括密钥验证、签名验证等
            # 简化处理，实际应用中需要完整的验证流程
            return len(dnskey_answer) > 0 and len(ds_answer) > 0
        except Exception:
            return False
 
# 使用示例
validator = DNSSECValidator()
result = validator.validate_domain('example.com')
print(f"DNSSEC验证结果: {result}")

class PrivacyCompliantDNSResolver {
  constructor(options = {}) {
    this.options = {
      logQueries: false, // 默认不记录查询日志
      anonymizeClientIP: true,
      retentionPeriod: 24 * 60 * 60 * 1000, // 24小时
      ...options
    };
    
    this.queryLog = [];
    this.privacyFilter = new PrivacyFilter();
  }
 
  async resolve(domain, options = {}) {
    // 应用隐私过滤
    const filteredDomain = this.privacyFilter.filterDomain(domain);
    
    // 匿名化客户端IP
    const clientIP = options.clientIP ? 
      this.anonymizeIP(options.clientIP) : null;
    
    try {
      const result = await this.performDNSQuery(filteredDomain, options);
      
      // 只有在明确同意的情况下才记录日志
      if (this.options.logQueries && options.consented) {
        this.logQuery({
          domain: filteredDomain,
          clientIP: clientIP,
          timestamp: new Date(),
          result: this.sanitizeResult(result)
        });
      }
      
      return result;
    } catch (error) {
      this.handleError(error, { domain: filteredDomain, clientIP });
      throw error;
    }
  }
 
  anonymizeIP(ip) {
    // 保留前两个IP段，其余置零
    const parts = ip.split('.');
    if (parts.length === 4) {
      return `${parts[0]}.${parts[1]}.0.0`;
    }
    return '0.0.0.0';
  }
 
  sanitizeResult(result) {
    // 移除敏感信息
    const sanitized = { ...result };
    delete sanitized.serverDetails;
    delete sanitized.debugInfo;
    return sanitized;
  }
 
  logQuery(logEntry) {
    this.queryLog.push(logEntry);
    
    // 定期清理过期日志
    this.cleanupOldLogs();
  }
 
  cleanupOldLogs() {
    const cutoffTime = Date.now() - this.options.retentionPeriod;
    this.queryLog = this.queryLog.filter(
      entry => entry.timestamp.getTime() > cutoffTime
    );
  }
 
  async performDNSQuery(domain, options) {
    // 实现DNS查询逻辑
    return { ip: '8.8.8.8', domain };
  }
 
  handleError(error, context) {
    // 错误处理逻辑，确保不泄露敏感信息
    console.error('DNS查询错误:', error.message);
  }
 
  // 数据导出功能（符合GDPR等法规）
  exportUserData(clientIdentifier) {
    return this.queryLog.filter(entry => 
      entry.clientIP === clientIdentifier || 
      entry.domain.includes(clientIdentifier)
    );
  }
 
  // 数据删除功能（符合GDPR"被遗忘权"）
  deleteUserData(clientIdentifier) {
    this.queryLog = this.queryLog.filter(entry => 
      entry.clientIP !== clientIdentifier && 
      !entry.domain.includes(clientIdentifier)
    );
  }
}
 
class PrivacyFilter {
  filterDomain(domain) {
    // 过滤掉可能包含个人信息的域名
    const sensitivePatterns = [
      /[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/, // 邮箱地址
      /\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b/, // IP地址
      /\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b/ // 信用卡号
    ];
    
    let filtered = domain;
    for (const pattern of sensitivePatterns) {
      filtered = filtered.replace(pattern, '[REDACTED]');
    }
    
    return filtered;
  }
}
 
// 使用示例
const resolver = new PrivacyCompliantDNSResolver({
  logQueries: true,
  anonymizeClientIP: true
});
 
// 用户明确同意的情况下记录查询
resolver.resolve('example.com', {
  clientIP: '192.168.1.100',
  consented: true
}).then(result => {
  console.log('隐私合规的DNS查询结果:', result);
});

class LRUCache {
  constructor(maxSize = 1000, ttl = 300000) { // 默认5分钟TTL
    this.maxSize = maxSize;
    this.ttl = ttl;
    this.cache = new Map();
    this.accessOrder = [];
  }
 
  get(key) {
    const entry = this.cache.get(key);
    if (!entry) return null;
 
    // 检查是否过期
    if (Date.now() - entry.timestamp > this.ttl) {
      this.delete(key);
      return null;
    }
 
    // 更新访问顺序
    this.updateAccessOrder(key);
    return entry.value;
  }
 
  set(key, value) {
    // 如果已存在，更新值和时间戳
    if (this.cache.has(key)) {
      this.cache.set(key, {
        value,
        timestamp: Date.now()
      });
      this.updateAccessOrder(key);
      return;
    }
 
    // 检查容量限制
    if (this.cache.size >= this.maxSize) {
      this.evictLRU();
    }
 
    this.cache.set(key, {
      value,
      timestamp: Date.now()
    });
    this.accessOrder.push(key);
  }
 
  updateAccessOrder(key) {
    const index = this.accessOrder.indexOf(key);
    if (index > -1) {
      this.accessOrder.splice(index, 1);
      this.accessOrder.push(key);
    }
  }
 
  evictLRU() {
    const lruKey = this.accessOrder.shift();
    if (lruKey) {
      this.cache.delete(lruKey);
    }
  }
 
  delete(key) {
    this.cache.delete(key);
    const index = this.accessOrder.indexOf(key);
    if (index > -1) {
      this.accessOrder.splice(index, 1);
    }
  }
 
  clear() {
    this.cache.clear();
    this.accessOrder = [];
  }
 
  getStats() {
    return {
      size: this.cache.size,
      maxSize: this.maxSize,
      hitRate: this.hitRate,
      missRate: this.missRate
    };
  }
}
 
class CachedDNSResolver {
  constructor(options = {}) {
    this.cache = new LRUCache(
      options.cacheSize || 1000,
      options.cacheTTL || 300000
    );
    this.resolver = new SecureDNSResolver(options);
  }
 
  async resolve(domain, options = {}) {
    const cacheKey = `${domain}:${options.type || 'A'}`;
    
    // 先检查缓存
    const cached = this.cache.get(cacheKey);
    if (cached) {
      console.log(`缓存命中: ${domain}`);
      return cached;
    }
 
    console.log(`缓存未命中，执行DNS查询: ${domain}`);
    
    // 缓存未命中，执行实际查询
    const result = await this.resolver.resolve(domain, options);
    
    // 缓存结果（仅缓存成功的查询）
    if (result && !result.error) {
      this.cache.set(cacheKey, result);
    }
    
    return result;
  }
 
  clearCache() {
    this.cache.clear();
  }
 
  getCacheStats() {
    return this.cache.getStats();
  }
}