mirror of
https://github.com/ScuroNeko/mtg.git
synced 2026-08-31 14:44:02 +03:00
Move cert noise calibration into doppelganger scout
Instead of a separate cert_probe.go that duplicates the scout's TLS connection logic, measure the cert chain size directly from the same HTTPS connections the scout already makes. Changes: - Extend ScoutConnResult with payloadLen field - Add Write interception to ScoutConn for handshake boundary detection - Scout.learn() now computes cert size (sum of ApplicationData between CCS and first client Write) alongside inter-record durations - Ganger aggregates cert sizes across raids and exposes NoiseParams() via atomic pointer for lock-free reads from proxy goroutines - Proxy reads NoiseParams from Ganger on each handshake instead of probing at startup - Remove cert_probe.go, disk cache, and related config options (noise-cache-path, noise-cache-ttl, noise-probe-count) Falls back to legacy 2500-4700 range until the first scout raid completes (typically within 1-2 seconds of startup).
This commit is contained in:
@@ -2,7 +2,9 @@ package doppel
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/9seconds/mtg/v2/essentials"
|
||||
@@ -12,8 +14,22 @@ const (
|
||||
DoppelGangerMaxDurations = 4096
|
||||
DoppelGangerScoutRaidEach = 6 * time.Hour
|
||||
DoppelGangerScoutRepeats = 10
|
||||
|
||||
MinCertSizesToCalculate = 3
|
||||
)
|
||||
|
||||
// NoiseParams holds the measured cert chain size for FakeTLS noise calibration.
|
||||
// If Mean is 0, the caller should use a legacy fallback.
|
||||
type NoiseParams struct {
|
||||
Mean int
|
||||
Jitter int
|
||||
}
|
||||
|
||||
type scoutRaidResult struct {
|
||||
durations []time.Duration
|
||||
certSizes []int
|
||||
}
|
||||
|
||||
type gangerConnRequest struct {
|
||||
ret chan<- Conn
|
||||
payload essentials.Conn
|
||||
@@ -33,6 +49,9 @@ type Ganger struct {
|
||||
|
||||
stats *Stats
|
||||
durations []time.Duration
|
||||
certSizes []int
|
||||
|
||||
noiseParams atomic.Pointer[NoiseParams]
|
||||
|
||||
connRequests chan gangerConnRequest
|
||||
}
|
||||
@@ -48,6 +67,16 @@ func (g *Ganger) Run() {
|
||||
})
|
||||
}
|
||||
|
||||
// NoiseParams returns the current cert-size-based noise parameters.
|
||||
// Returns zero-value NoiseParams if not yet measured (caller should use fallback).
|
||||
func (g *Ganger) NoiseParams() NoiseParams {
|
||||
if p := g.noiseParams.Load(); p != nil {
|
||||
return *p
|
||||
}
|
||||
|
||||
return NoiseParams{}
|
||||
}
|
||||
|
||||
func (g *Ganger) NewConn(conn essentials.Conn) (Conn, error) {
|
||||
rvChan := make(chan Conn)
|
||||
req := gangerConnRequest{
|
||||
@@ -81,7 +110,7 @@ func (g *Ganger) run() {
|
||||
}
|
||||
}()
|
||||
|
||||
scoutCollectedChan := make(chan []time.Duration)
|
||||
scoutCollectedChan := make(chan scoutRaidResult)
|
||||
currentScoutCollectedChan := scoutCollectedChan
|
||||
|
||||
updatedStatsChan := make(chan *Stats)
|
||||
@@ -94,18 +123,29 @@ func (g *Ganger) run() {
|
||||
select {
|
||||
case <-g.ctx.Done():
|
||||
return
|
||||
case durations := <-currentScoutCollectedChan:
|
||||
g.durations = append(g.durations, durations...)
|
||||
case result := <-currentScoutCollectedChan:
|
||||
g.durations = append(g.durations, result.durations...)
|
||||
|
||||
if len(g.durations) > DoppelGangerMaxDurations {
|
||||
copy(g.durations, g.durations[len(g.durations)-DoppelGangerMaxDurations:])
|
||||
g.durations = g.durations[:DoppelGangerMaxDurations]
|
||||
}
|
||||
|
||||
// Update cert sizes and recompute noise params.
|
||||
g.certSizes = append(g.certSizes, result.certSizes...)
|
||||
if len(g.certSizes) > DoppelGangerMaxDurations {
|
||||
g.certSizes = g.certSizes[len(g.certSizes)-DoppelGangerMaxDurations:]
|
||||
}
|
||||
|
||||
if len(g.certSizes) >= MinCertSizesToCalculate {
|
||||
g.updateNoiseParams()
|
||||
}
|
||||
|
||||
if len(g.durations) < MinDurationsToCalculate {
|
||||
continue
|
||||
}
|
||||
|
||||
durations := g.durations
|
||||
currentScoutCollectedChan = nil
|
||||
g.wg.Go(func() {
|
||||
select {
|
||||
@@ -129,8 +169,45 @@ func (g *Ganger) run() {
|
||||
}
|
||||
}
|
||||
|
||||
func (g *Ganger) runScoutRaid(rvChan chan<- []time.Duration) {
|
||||
durations := []time.Duration{}
|
||||
func (g *Ganger) updateNoiseParams() {
|
||||
if len(g.certSizes) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
sum := 0
|
||||
for _, s := range g.certSizes {
|
||||
sum += s
|
||||
}
|
||||
|
||||
mean := sum / len(g.certSizes)
|
||||
|
||||
maxDev := 0
|
||||
for _, s := range g.certSizes {
|
||||
d := s - mean
|
||||
if d < 0 {
|
||||
d = -d
|
||||
}
|
||||
|
||||
if d > maxDev {
|
||||
maxDev = d
|
||||
}
|
||||
}
|
||||
|
||||
if maxDev < 100 {
|
||||
maxDev = 100
|
||||
}
|
||||
|
||||
np := &NoiseParams{Mean: mean, Jitter: maxDev}
|
||||
g.noiseParams.Store(np)
|
||||
|
||||
g.logger.Info(fmt.Sprintf(
|
||||
"updated noise params: mean=%d jitter=%d samples=%d",
|
||||
mean, maxDev, len(g.certSizes),
|
||||
))
|
||||
}
|
||||
|
||||
func (g *Ganger) runScoutRaid(rvChan chan<- scoutRaidResult) {
|
||||
var result scoutRaidResult
|
||||
|
||||
for range g.scoutRaidRepeats {
|
||||
learned, err := g.scout.Learn(g.ctx)
|
||||
@@ -138,13 +215,18 @@ func (g *Ganger) runScoutRaid(rvChan chan<- []time.Duration) {
|
||||
g.logger.WarningError("cannot learn", err)
|
||||
continue
|
||||
}
|
||||
durations = append(durations, learned...)
|
||||
|
||||
result.durations = append(result.durations, learned.Durations...)
|
||||
|
||||
if learned.CertSize > 0 {
|
||||
result.certSizes = append(result.certSizes, learned.CertSize)
|
||||
}
|
||||
}
|
||||
|
||||
select {
|
||||
case <-g.ctx.Done():
|
||||
return
|
||||
case rvChan <- durations:
|
||||
case rvChan <- result:
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -12,36 +12,46 @@ import (
|
||||
"github.com/9seconds/mtg/v2/mtglib/internal/tls"
|
||||
)
|
||||
|
||||
// ScoutResult holds measurements from a single scout HTTP request.
|
||||
type ScoutResult struct {
|
||||
Durations []time.Duration
|
||||
CertSize int // total ApplicationData bytes during TLS handshake; 0 if unknown
|
||||
}
|
||||
|
||||
type Scout struct {
|
||||
network Network
|
||||
urls []string
|
||||
}
|
||||
|
||||
func (s Scout) Learn(ctx context.Context) ([]time.Duration, error) {
|
||||
var durations []time.Duration
|
||||
func (s Scout) Learn(ctx context.Context) (ScoutResult, error) {
|
||||
var combined ScoutResult
|
||||
|
||||
for _, url := range s.urls {
|
||||
learned, err := s.learn(ctx, url)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return ScoutResult{}, err
|
||||
}
|
||||
|
||||
durations = append(durations, learned...)
|
||||
combined.Durations = append(combined.Durations, learned.Durations...)
|
||||
|
||||
if learned.CertSize > 0 && combined.CertSize == 0 {
|
||||
combined.CertSize = learned.CertSize
|
||||
}
|
||||
}
|
||||
|
||||
return durations, nil
|
||||
return combined, nil
|
||||
}
|
||||
|
||||
func (s Scout) learn(ctx context.Context, url string) ([]time.Duration, error) {
|
||||
func (s Scout) learn(ctx context.Context, url string) (ScoutResult, error) {
|
||||
client, results := s.makeClient()
|
||||
|
||||
if !strings.HasPrefix(url, "https://") {
|
||||
return nil, fmt.Errorf("url %s must be https", url)
|
||||
return ScoutResult{}, fmt.Errorf("url %s must be https", url)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return ScoutResult{}, err
|
||||
}
|
||||
|
||||
resp, err := client.Do(req)
|
||||
@@ -52,10 +62,12 @@ func (s Scout) learn(ctx context.Context, url string) ([]time.Duration, error) {
|
||||
}
|
||||
|
||||
if err != nil || len(results.data) == 0 {
|
||||
return nil, err
|
||||
return ScoutResult{}, err
|
||||
}
|
||||
|
||||
durations := []time.Duration{}
|
||||
var result ScoutResult
|
||||
|
||||
// Compute inter-record durations (existing logic).
|
||||
lastTimestamp := time.Time{}
|
||||
|
||||
for i, v := range results.data {
|
||||
@@ -71,11 +83,34 @@ func (s Scout) learn(ctx context.Context, url string) ([]time.Duration, error) {
|
||||
}
|
||||
}
|
||||
|
||||
durations = append(durations, v.timestamp.Sub(lastTimestamp))
|
||||
result.Durations = append(result.Durations, v.timestamp.Sub(lastTimestamp))
|
||||
lastTimestamp = v.timestamp
|
||||
}
|
||||
|
||||
return durations, nil
|
||||
// Compute cert size: sum of ApplicationData payload between CCS and
|
||||
// the first client Write (which marks the end of server handshake).
|
||||
seenCCS := false
|
||||
boundary := results.writeIndex
|
||||
if boundary < 0 {
|
||||
boundary = len(results.data)
|
||||
}
|
||||
|
||||
for i, v := range results.data {
|
||||
if i >= boundary {
|
||||
break
|
||||
}
|
||||
|
||||
if v.recordType == tls.TypeChangeCipherSpec {
|
||||
seenCCS = true
|
||||
continue
|
||||
}
|
||||
|
||||
if seenCCS && v.recordType == tls.TypeApplicationData {
|
||||
result.CertSize += v.payloadLen
|
||||
}
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s Scout) makeClient() (*http.Client, *ScoutConnCollected) {
|
||||
|
||||
@@ -14,9 +14,10 @@ type ScoutConn struct {
|
||||
|
||||
results *ScoutConnCollected
|
||||
rawBuf *bytes.Buffer
|
||||
seenCCS bool
|
||||
}
|
||||
|
||||
func (s ScoutConn) Read(p []byte) (int, error) {
|
||||
func (s *ScoutConn) Read(p []byte) (int, error) {
|
||||
buf := &bytes.Buffer{}
|
||||
|
||||
for {
|
||||
@@ -31,7 +32,11 @@ func (s ScoutConn) Read(p []byte) (int, error) {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
s.results.Add(recordType)
|
||||
if recordType == tls.TypeChangeCipherSpec {
|
||||
s.seenCCS = true
|
||||
}
|
||||
|
||||
s.results.Add(recordType, int(length))
|
||||
s.rawBuf.Write([]byte{recordType})
|
||||
s.rawBuf.Write(tls.TLSVersion[:])
|
||||
|
||||
@@ -45,11 +50,19 @@ func (s ScoutConn) Read(p []byte) (int, error) {
|
||||
}
|
||||
}
|
||||
|
||||
func NewScoutConn(conn essentials.Conn, results *ScoutConnCollected) ScoutConn {
|
||||
func (s *ScoutConn) Write(p []byte) (int, error) {
|
||||
if s.seenCCS {
|
||||
s.results.MarkWrite()
|
||||
}
|
||||
|
||||
return s.Conn.Write(p)
|
||||
}
|
||||
|
||||
func NewScoutConn(conn essentials.Conn, results *ScoutConnCollected) *ScoutConn {
|
||||
rawBuf := &bytes.Buffer{}
|
||||
rawBuf.Grow(tls.MaxRecordSize)
|
||||
|
||||
return ScoutConn{
|
||||
return &ScoutConn{
|
||||
Conn: tls.New(conn, false, false),
|
||||
results: results,
|
||||
rawBuf: rawBuf,
|
||||
|
||||
@@ -9,21 +9,32 @@ const (
|
||||
type ScoutConnResult struct {
|
||||
timestamp time.Time
|
||||
recordType byte
|
||||
payloadLen int
|
||||
}
|
||||
|
||||
type ScoutConnCollected struct {
|
||||
data []ScoutConnResult
|
||||
data []ScoutConnResult
|
||||
writeIndex int // index at which client first wrote post-handshake data; -1 if not set
|
||||
}
|
||||
|
||||
func (s *ScoutConnCollected) Add(record byte) {
|
||||
func (s *ScoutConnCollected) Add(record byte, payloadLen int) {
|
||||
s.data = append(s.data, ScoutConnResult{
|
||||
timestamp: time.Now(),
|
||||
recordType: record,
|
||||
payloadLen: payloadLen,
|
||||
})
|
||||
}
|
||||
|
||||
// MarkWrite records the current data length as the handshake boundary.
|
||||
func (s *ScoutConnCollected) MarkWrite() {
|
||||
if s.writeIndex < 0 {
|
||||
s.writeIndex = len(s.data)
|
||||
}
|
||||
}
|
||||
|
||||
func NewScoutConnCollected() *ScoutConnCollected {
|
||||
return &ScoutConnCollected{
|
||||
data: make([]ScoutConnResult, 0, ScoutConnCollectedPreallocSize),
|
||||
data: make([]ScoutConnResult, 0, ScoutConnCollectedPreallocSize),
|
||||
writeIndex: -1,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,7 +14,7 @@ type ScoutConnCollectedTestSuite struct {
|
||||
|
||||
func (suite *ScoutConnCollectedTestSuite) TestAddSingle() {
|
||||
collected := NewScoutConnCollected()
|
||||
collected.Add(tls.TypeApplicationData)
|
||||
collected.Add(tls.TypeApplicationData, 100)
|
||||
|
||||
suite.Len(collected.data, 1)
|
||||
suite.Equal(byte(tls.TypeApplicationData), collected.data[0].recordType)
|
||||
@@ -23,13 +23,13 @@ func (suite *ScoutConnCollectedTestSuite) TestAddSingle() {
|
||||
func (suite *ScoutConnCollectedTestSuite) TestAddTimestampsAreMonotonic() {
|
||||
collected := NewScoutConnCollected()
|
||||
|
||||
collected.Add(tls.TypeApplicationData)
|
||||
collected.Add(tls.TypeApplicationData, 100)
|
||||
|
||||
time.Sleep(time.Microsecond)
|
||||
collected.Add(tls.TypeApplicationData)
|
||||
collected.Add(tls.TypeApplicationData, 100)
|
||||
|
||||
time.Sleep(time.Microsecond)
|
||||
collected.Add(tls.TypeApplicationData)
|
||||
collected.Add(tls.TypeApplicationData, 100)
|
||||
|
||||
for i := 1; i < len(collected.data); i++ {
|
||||
suite.True(collected.data[i].timestamp.After(collected.data[i-1].timestamp))
|
||||
|
||||
@@ -22,9 +22,9 @@ func (suite *ScoutTestSuite) SetupSuite() {
|
||||
}
|
||||
|
||||
func (suite *ScoutTestSuite) TestCollectResults() {
|
||||
durations, err := suite.scout.Learn(suite.ctx)
|
||||
result, err := suite.scout.Learn(suite.ctx)
|
||||
suite.NoError(err)
|
||||
suite.Less(3, len(durations))
|
||||
suite.Less(3, len(result.Durations))
|
||||
}
|
||||
|
||||
func (suite *ScoutTestSuite) TestCollectNothing() {
|
||||
|
||||
Reference in New Issue
Block a user