● Notice //GitHub (@shehzade) is temporarily suspended pending an automated false-positive review. All tools are hosted on this Code Mirror and available on request (contact); GitHub links will return once reinstated.
PUBLIC MIRRORGo v1.0

Multi-protocol network egress assessment and DLP verification framework featuring automated L4 sweeps, corporate proxy traversal, and covert protocol smuggling across 20+ channels.

$ git clone https://shehzade.io/repos/contraband.git Copy
Source ZIP (1.93 MB)
Code 174 Commits 1
contraband / pkg / client / traffic / shaper.go
← Back to directory
326 lines · 7.5 KB
Raw
1package traffic
2
3import (
4 "context"
5 "math"
6 "sync"
7 "sync/atomic"
8 "time"
9
10 "github.com/shehzade/contraband/pkg/common"
11)
12
13// RateLimiter defines the interface for token bucket bandwidth and packet pacing.
14type RateLimiter interface {
15 // WaitN blocks until n tokens (bytes) are available or context is cancelled.
16 WaitN(ctx context.Context, n int) error
17
18 // AllowN returns true if n tokens are immediately available and consumes them.
19 AllowN(n int) bool
20
21 // ReserveN reserves n tokens and returns wait duration before consumption is valid.
22 ReserveN(n int) (time.Duration, bool)
23
24 // Refund restores unused tokens (e.g. on short read or partial write).
25 Refund(n int)
26
27 // SetRate dynamically updates bandwidth rate in bytes per second.
28 SetRate(rateBytesPerSec float64)
29
30 // SetBurst dynamically updates burst capacity in bytes.
31 SetBurst(burstBytes int64)
32
33 // SetPacketGap sets minimum microsecond inter-packet gap.
34 SetPacketGap(gap time.Duration)
35
36 // Rate returns current rate in bytes per second (0.0 = unlimited).
37 Rate() float64
38
39 // Burst returns maximum burst capacity in bytes.
40 Burst() int64
41
42 // PacketGap returns minimum microsecond inter-packet gap.
43 PacketGap() time.Duration
44
45 // IsUnlimited returns true if bandwidth limiting is bypassed.
46 IsUnlimited() bool
47}
48
49// TokenBucket implements RateLimiter with microsecond precision and burst control.
50type TokenBucket struct {
51 mu sync.Mutex
52 rate float64 // Bytes per second (0.0 = unlimited)
53 capacity float64 // Burst capacity in bytes
54 tokens float64 // Current available tokens
55 lastRefill time.Time // Monotonic timestamp of last refill
56 packetGap time.Duration // Microsecond inter-packet gap
57 lastPacketTime time.Time // Monotonic timestamp of last write
58 isUnlimited atomic.Bool // Fast-path bypass flag
59}
60
61// NewTokenBucket creates a token bucket with given rate and burst capacity.
62// If rateBytesPerSec <= 0, bucket operates in unlimited mode.
63func NewTokenBucket(rateBytesPerSec float64, burstBytes int64, packetGap time.Duration) *TokenBucket {
64 now := time.Now()
65 if rateBytesPerSec <= 0.0 || math.IsNaN(rateBytesPerSec) || math.IsInf(rateBytesPerSec, 0) {
66 tb := &TokenBucket{
67 rate: 0.0,
68 capacity: float64(burstBytes),
69 tokens: float64(burstBytes),
70 lastRefill: now,
71 packetGap: packetGap,
72 lastPacketTime: now,
73 }
74 tb.isUnlimited.Store(true)
75 return tb
76 }
77
78 // Default burst capacity: 200ms worth of tokens or at least 64KB, max 4MB
79 capFloat := float64(burstBytes)
80 if capFloat <= 0 {
81 capFloat = math.Max(64*1024, math.Min(rateBytesPerSec*0.2, 4*1024*1024))
82 }
83
84 tb := &TokenBucket{
85 rate: rateBytesPerSec,
86 capacity: capFloat,
87 tokens: capFloat, // start full
88 lastRefill: now,
89 packetGap: packetGap,
90 lastPacketTime: now,
91 }
92 tb.isUnlimited.Store(false)
93 return tb
94}
95
96// NewTokenBucketFromString initializes a TokenBucket using human rate string ("100KB/s", "unlimited").
97func NewTokenBucketFromString(rateStr string, burstBytes int64, packetGap time.Duration) (*TokenBucket, error) {
98 rate, err := common.ParseBandwidthRate(rateStr)
99 if err != nil {
100 return nil, err
101 }
102 return NewTokenBucket(rate, burstBytes, packetGap), nil
103}
104
105// WaitN blocks until n tokens are consumed or ctx is cancelled.
106func (tb *TokenBucket) WaitN(ctx context.Context, n int) error {
107 if n <= 0 || tb.isUnlimited.Load() {
108 return nil
109 }
110
111 for {
112 // Fast check for context cancellation
113 select {
114 case <-ctx.Done():
115 return ctx.Err()
116 default:
117 }
118
119 if tb.isUnlimited.Load() {
120 return nil
121 }
122
123 tb.mu.Lock()
124 tb.refill()
125
126 if tb.isUnlimited.Load() || tb.rate <= 0.0 {
127 tb.mu.Unlock()
128 return nil
129 }
130
131 // Enforce microsecond packet gap if configured
132 var gapWait time.Duration
133 now := time.Now()
134 if tb.packetGap > 0 {
135 elapsedGap := now.Sub(tb.lastPacketTime)
136 if elapsedGap < tb.packetGap {
137 gapWait = tb.packetGap - elapsedGap
138 }
139 }
140
141 needed := float64(n)
142 if needed > tb.capacity {
143 tb.capacity = needed * 2
144 }
145 if tb.tokens >= needed {
146 // Tokens available immediately
147 tb.tokens -= needed
148 tb.lastPacketTime = now.Add(gapWait)
149 tb.mu.Unlock()
150
151 if gapWait > 0 {
152 return sleepContext(ctx, gapWait)
153 }
154 return nil
155 }
156
157 // Calculate wait time for remaining deficit
158 deficit := needed - tb.tokens
159 waitSec := deficit / tb.rate
160 waitDur := time.Duration(waitSec * float64(time.Second))
161 if gapWait > waitDur {
162 waitDur = gapWait
163 }
164 tb.mu.Unlock()
165
166 // Sleep for required deficit duration
167 if err := sleepContext(ctx, waitDur); err != nil {
168 return err
169 }
170 }
171}
172
173// AllowN performs non-blocking token check and deduction.
174func (tb *TokenBucket) AllowN(n int) bool {
175 if n <= 0 || tb.isUnlimited.Load() {
176 return true
177 }
178
179 tb.mu.Lock()
180 defer tb.mu.Unlock()
181
182 if tb.isUnlimited.Load() || tb.rate <= 0.0 {
183 return true
184 }
185
186 tb.refill()
187 needed := float64(n)
188 if tb.tokens >= needed {
189 tb.tokens -= needed
190 tb.lastPacketTime = time.Now()
191 return true
192 }
193 return false
194}
195
196// ReserveN calculates how long caller must wait before consuming n tokens.
197func (tb *TokenBucket) ReserveN(n int) (time.Duration, bool) {
198 if n <= 0 || tb.isUnlimited.Load() {
199 return 0, true
200 }
201
202 tb.mu.Lock()
203 defer tb.mu.Unlock()
204
205 if tb.isUnlimited.Load() || tb.rate <= 0.0 {
206 return 0, true
207 }
208
209 tb.refill()
210 needed := float64(n)
211 if needed > tb.capacity {
212 tb.capacity = needed * 2
213 }
214 if tb.tokens >= needed {
215 tb.tokens -= needed
216 return 0, true
217 }
218
219 deficit := needed - tb.tokens
220 waitSec := deficit / tb.rate
221 waitDur := time.Duration(waitSec * float64(time.Second))
222 tb.tokens = 0 // consume all currently available
223 return waitDur, false
224}
225
226// Refund returns unconsumed tokens back to the bucket (clamped to capacity).
227func (tb *TokenBucket) Refund(n int) {
228 if n <= 0 || tb.isUnlimited.Load() {
229 return
230 }
231
232 tb.mu.Lock()
233 defer tb.mu.Unlock()
234
235 if tb.isUnlimited.Load() || tb.rate <= 0.0 {
236 return
237 }
238
239 tb.refill()
240 tb.tokens = math.Min(tb.capacity, tb.tokens+float64(n))
241}
242
243func (tb *TokenBucket) SetRate(rateBytesPerSec float64) {
244 tb.mu.Lock()
245 defer tb.mu.Unlock()
246 tb.refill()
247 if rateBytesPerSec <= 0.0 || math.IsNaN(rateBytesPerSec) || math.IsInf(rateBytesPerSec, 0) {
248 tb.rate = 0.0
249 tb.isUnlimited.Store(true)
250 } else {
251 tb.rate = rateBytesPerSec
252 tb.isUnlimited.Store(false)
253 }
254}
255
256func (tb *TokenBucket) SetBurst(burstBytes int64) {
257 tb.mu.Lock()
258 defer tb.mu.Unlock()
259 if burstBytes <= 0 {
260 burstBytes = 64 * 1024
261 }
262 tb.capacity = float64(burstBytes)
263 if tb.tokens > tb.capacity {
264 tb.tokens = tb.capacity
265 }
266}
267
268func (tb *TokenBucket) SetPacketGap(gap time.Duration) {
269 tb.mu.Lock()
270 defer tb.mu.Unlock()
271 tb.packetGap = gap
272}
273
274func (tb *TokenBucket) Rate() float64 {
275 tb.mu.Lock()
276 defer tb.mu.Unlock()
277 return tb.rate
278}
279
280func (tb *TokenBucket) Burst() int64 {
281 tb.mu.Lock()
282 defer tb.mu.Unlock()
283 return int64(tb.capacity)
284}
285
286func (tb *TokenBucket) PacketGap() time.Duration {
287 tb.mu.Lock()
288 defer tb.mu.Unlock()
289 return tb.packetGap
290}
291
292func (tb *TokenBucket) IsUnlimited() bool {
293 return tb.isUnlimited.Load()
294}
295
296func (tb *TokenBucket) refill() {
297 now := time.Now()
298 elapsed := now.Sub(tb.lastRefill).Seconds()
299 if elapsed > 0 {
300 tb.tokens = math.Min(tb.capacity, tb.tokens+elapsed*tb.rate)
301 tb.lastRefill = now
302 }
303}
304
305// sleepContext executes a non-allocating timer sleep with immediate context cancellation.
306func sleepContext(ctx context.Context, d time.Duration) error {
307 if d <= 0 {
308 return nil
309 }
310 timer := time.NewTimer(d)
311 defer func() {
312 if !timer.Stop() {
313 select {
314 case <-timer.C:
315 default:
316 }
317 }
318 }()
319
320 select {
321 case <-ctx.Done():
322 return ctx.Err()
323 case <-timer.C:
324 return nil
325 }
326}