<commit_before>package dnsclient

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"net"
	"net/http"
	"strconv"
	"sync"
	"time"

	"github.com/HouzuoGuo/laitos/daemon/dnsd"
	"github.com/HouzuoGuo/laitos/daemon/httpd/middleware"
	"github.com/HouzuoGuo/laitos/lalog"
	"github.com/HouzuoGuo/laitos/misc"
	"github.com/HouzuoGuo/laitos/tcpoverdns"
	"github.com/HouzuoGuo/laitos/toolbox"
)

// ProxiedConnection handles an individual proxy connection to transport
// data between local transmission control and the one on the remote DNS proxy
// server.
type ProxiedConnection struct {
	client               *Client
	out, in              net.Conn
	tc                   *tcpoverdns.TransmissionControl
	outputSegmentBacklog []tcpoverdns.Segment
	mutex                *sync.Mutex
	context              context.Context
	logger               lalog.Logger
}

// Start configures and then starts the transmission control on local side, and
// spawns a background goroutine to transport segments back and forth using
// DNS queries.
// The function returns when the local transmission control transitions to the
// established state, or an error.
func (conn *ProxiedConnection) Start() error {
	conn.logger.Info("Start", "", nil, "start transporting data over DNS")
	// Absorb outgoing segments into the outgoing backlog.
	conn.tc.OutputSegmentCallback = func(seg tcpoverdns.Segment) {
		if seg.Flags.Has(tcpoverdns.FlagKeepAlive) {
			// Add random data to the segment to prevent caching.
			seg.Data = misc.RandomBytes(4)
		}
		// Replace the latest keep-alive or ack-only segment (if any), and
		// de-duplicate adjacent identical segments. These measures not only
		// speed up the exchanges but also ensure that peers can communicate
		// properly even if their timing characteristics differ.
		conn.mutex.Lock()
		defer conn.mutex.Unlock()
		var latest tcpoverdns.Segment
		if len(conn.outputSegmentBacklog) > 0 {
			latest = conn.outputSegmentBacklog[len(conn.outputSegmentBacklog)-1]
		}
		if latest.Flags.Has(tcpoverdns.FlagAckOnly) || latest.Flags.Has(tcpoverdns.FlagKeepAlive) {
			// Substitute the ack-only or keep-alive segment with the latest.
			// These segments may contain random data that are not useful.
			if conn.client.Debug {
				conn.logger.Info("Start", "", nil, "callback is removing duplicated ack/keepalive segment: %+v", conn.outputSegmentBacklog[len(conn.outputSegmentBacklog)-1])
			}
			conn.outputSegmentBacklog[len(conn.outputSegmentBacklog)-1] = seg
		} else if latest.Equals(seg) {
			// De-duplicate adjacent identical segments.
			if conn.client.Debug {
				conn.logger.Info("Start", "", nil, "callback is removing duplicated segment: %+v", seg)
			}
		} else {
			conn.outputSegmentBacklog = append(conn.outputSegmentBacklog, seg)
		}
		conn.logger.Info("Start", "", nil, "queued segment for outbound over DNS: %v", seg)
	}
	conn.tc.Start(conn.context)
	// Start transporting segments back and forth.
	go func() {
		var dialFun func(ctx context.Context, network, address string) (net.Conn, error)
		if conn.client.DNSResolverAddr != "" {
			// Use the custom specified recursive resolver instead of the system
			// default.
			dialFun = func(ctx context.Context, network, address string) (net.Conn, error) {
				if conn.client.Debug {
					conn.logger.Info("Start", "", nil, "dialing resolver %q %s:%d", network, conn.client.DNSResolverAddr, conn.client.DNSResovlerPort)
				}
				return net.Dial(network, fmt.Sprintf("%s:%d", conn.client.DNSResolverAddr, conn.client.DNSResovlerPort))
			}
		}
		resolver := &net.Resolver{
			PreferGo: true,
			Dial:     dialFun,
		}
		defer func() {
			// Linger briefly, then send the last segment. The brief waiting
			// time allows the TC to transition to the closed state.
			time.Sleep(5 * time.Second)
			conn.mutex.Lock()
			var finalSeg tcpoverdns.Segment
			if len(conn.outputSegmentBacklog) > 0 {
				finalSeg = conn.outputSegmentBacklog[len(conn.outputSegmentBacklog)-1]
			}
			conn.mutex.Unlock()
			if finalSeg.Flags != 0 {
				timeoutCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
				defer cancel()
				if _, err := resolver.LookupCNAME(timeoutCtx, finalSeg.DNSName(fmt.Sprintf("%c", dnsd.ProxyPrefix), conn.client.DNSHostName)); err != nil {
					conn.logger.Warning("Start", "", err, "failed to send the final segment")
				}
			}
			conn.logger.Info("Start", "", nil, "DNS data transport finished, the final segment was: %v", finalSeg)
		}()
		countHostNameLabels := dnsd.CountNameLabels(conn.client.DNSHostName)
		for {
			if conn.tc.State() == tcpoverdns.StateClosed {
				return
			}
			var incomingSeg, outgoingSeg, nextInBacklog tcpoverdns.Segment
			var cname string
			var err error
			// Pop a segment.
			conn.mutex.Lock()
			if len(conn.outputSegmentBacklog) == 0 {
				conn.mutex.Unlock()
				// Wait for a segment.
				goto busyWaitInterval
			}
			outgoingSeg = conn.outputSegmentBacklog[0]
			conn.outputSegmentBacklog = conn.outputSegmentBacklog[1:]
			conn.mutex.Unlock()
			// Turn the segment into a DNS query and send the query out
			// (data.data.data.example.com).
			if conn.client.Debug {
				conn.client.logger.Info("Start", fmt.Sprint(conn.tc.ID), nil, "sending output segment over DNS query: %+v", outgoingSeg)
			}
			cname, err = resolver.LookupCNAME(conn.context, outgoingSeg.DNSName(fmt.Sprintf("%c", dnsd.ProxyPrefix), conn.client.DNSHostName))
			if err != nil {
				conn.client.logger.Warning("Start", fmt.Sprint(conn.tc.ID), err, "failed to send output segment %v", outgoingSeg)
				conn.tc.IncreaseTimingInterval()
				goto busyWaitInterval
			}<commit_msg>tcpoverdns: log output segment sent instead of the segment queued<commit_after>package dnsclient

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"net"
	"net/http"
	"strconv"
	"sync"
	"time"

	"github.com/HouzuoGuo/laitos/daemon/dnsd"
	"github.com/HouzuoGuo/laitos/daemon/httpd/middleware"
	"github.com/HouzuoGuo/laitos/lalog"
	"github.com/HouzuoGuo/laitos/misc"
	"github.com/HouzuoGuo/laitos/tcpoverdns"
	"github.com/HouzuoGuo/laitos/toolbox"
)

// ProxiedConnection handles an individual proxy connection to transport
// data between local transmission control and the one on the remote DNS proxy
// server.
type ProxiedConnection struct {
	client               *Client
	out, in              net.Conn
	tc                   *tcpoverdns.TransmissionControl
	outputSegmentBacklog []tcpoverdns.Segment
	mutex                *sync.Mutex
	context              context.Context
	logger               lalog.Logger
}

// Start configures and then starts the transmission control on local side, and
// spawns a background goroutine to transport segments back and forth using
// DNS queries.
// The function returns when the local transmission control transitions to the
// established state, or an error.
func (conn *ProxiedConnection) Start() error {
	conn.logger.Info("Start", "", nil, "start transporting data over DNS")
	// Absorb outgoing segments into the outgoing backlog.
	conn.tc.OutputSegmentCallback = func(seg tcpoverdns.Segment) {
		if seg.Flags.Has(tcpoverdns.FlagKeepAlive) {
			// Add random data to the segment to prevent caching.
			seg.Data = misc.RandomBytes(4)
		}
		// Replace the latest keep-alive or ack-only segment (if any), and
		// de-duplicate adjacent identical segments. These measures not only
		// speed up the exchanges but also ensure that peers can communicate
		// properly even if their timing characteristics differ.
		conn.mutex.Lock()
		defer conn.mutex.Unlock()
		var latest tcpoverdns.Segment
		if len(conn.outputSegmentBacklog) > 0 {
			latest = conn.outputSegmentBacklog[len(conn.outputSegmentBacklog)-1]
		}
		if latest.Flags.Has(tcpoverdns.FlagAckOnly) || latest.Flags.Has(tcpoverdns.FlagKeepAlive) {
			// Substitute the ack-only or keep-alive segment with the latest.
			// These segments may contain random data that are not useful.
			if conn.client.Debug {
				conn.logger.Info("Start", "", nil, "callback is removing duplicated ack/keepalive segment: %+v", conn.outputSegmentBacklog[len(conn.outputSegmentBacklog)-1])
			}
			conn.outputSegmentBacklog[len(conn.outputSegmentBacklog)-1] = seg
		} else if latest.Equals(seg) {
			// De-duplicate adjacent identical segments.
			if conn.client.Debug {
				conn.logger.Info("Start", "", nil, "callback is removing duplicated segment: %+v", seg)
			}
		} else {
			if conn.client.Debug {
				conn.logger.Info("Start", "", nil, "queued segment for outbound over DNS: %v", seg)
			}
			conn.outputSegmentBacklog = append(conn.outputSegmentBacklog, seg)
		}
	}
	conn.tc.Start(conn.context)
	// Start transporting segments back and forth.
	go func() {
		var dialFun func(ctx context.Context, network, address string) (net.Conn, error)
		if conn.client.DNSResolverAddr != "" {
			// Use the custom specified recursive resolver instead of the system
			// default.
			dialFun = func(ctx context.Context, network, address string) (net.Conn, error) {
				if conn.client.Debug {
					conn.logger.Info("Start", "", nil, "dialing resolver %q %s:%d", network, conn.client.DNSResolverAddr, conn.client.DNSResovlerPort)
				}
				return net.Dial(network, fmt.Sprintf("%s:%d", conn.client.DNSResolverAddr, conn.client.DNSResovlerPort))
			}
		}
		resolver := &net.Resolver{
			PreferGo: true,
			Dial:     dialFun,
		}
		defer func() {
			// Linger briefly, then send the last segment. The brief waiting
			// time allows the TC to transition to the closed state.
			time.Sleep(5 * time.Second)
			conn.mutex.Lock()
			var finalSeg tcpoverdns.Segment
			if len(conn.outputSegmentBacklog) > 0 {
				finalSeg = conn.outputSegmentBacklog[len(conn.outputSegmentBacklog)-1]
			}
			conn.mutex.Unlock()
			if finalSeg.Flags != 0 {
				timeoutCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
				defer cancel()
				if _, err := resolver.LookupCNAME(timeoutCtx, finalSeg.DNSName(fmt.Sprintf("%c", dnsd.ProxyPrefix), conn.client.DNSHostName)); err != nil {
					conn.logger.Warning("Start", "", err, "failed to send the final segment")
				}
			}
			conn.logger.Info("Start", "", nil, "DNS data transport finished, the final segment was: %v", finalSeg)
		}()
		countHostNameLabels := dnsd.CountNameLabels(conn.client.DNSHostName)
		for {
			if conn.tc.State() == tcpoverdns.StateClosed {
				return
			}
			var incomingSeg, outgoingSeg, nextInBacklog tcpoverdns.Segment
			var cname string
			var err error
			// Pop a segment.
			conn.mutex.Lock()
			if len(conn.outputSegmentBacklog) == 0 {
				conn.mutex.Unlock()
				// Wait for a segment.
				goto busyWaitInterval
			}
			outgoingSeg = conn.outputSegmentBacklog[0]
			conn.outputSegmentBacklog = conn.outputSegmentBacklog[1:]
			conn.mutex.Unlock()
			// Turn the segment into a DNS query and send the query out
			// (data.data.data.example.com).
			conn.client.logger.Info("Start", fmt.Sprint(conn.tc.ID), nil, "sending output segment over DNS query: %+v", outgoingSeg)
			cname, err = resolver.LookupCNAME(conn.context, outgoingSeg.DNSName(fmt.Sprintf("%c", dnsd.ProxyPrefix), conn.client.DNSHostName))
			if err != nil {
				conn.client.logger.Warning("Start", fmt.Sprint(conn.tc.ID), err, "failed to send output segment %v", outgoingSeg)
				conn.tc.IncreaseTimingInterval()
				goto busyWaitInterval
			}
{'fraction_non_alphanumeric': 0.11678253982223004, 'fraction_numerical': 0.002464137991727537, 'mean_word_length': 3.6026731470230864, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1261058', 'n_tokens_mistral': 4058, 'n_tokens_neox': 3617, 'n_words': 949}