s1d3sw1ped/steamcache2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 20 Wiki Activity

Remove plans/ directory (P0/P1/P2 work complete)

Browse Source
This commit is contained in:
Justin Harms
2026-05-27 02:12:21 -05:00
parent 0c1840d223
commit 0dbb2e02ed
33 changed files with 1906 additions and 990 deletions
Download Patch File Download Diff File Expand all files Collapse all files
steamcache/steamcache.go
+49 -23
View File
@@ -269,6 +269,7 @@ func (sc *SteamCache) streamCachedResponse(w http.ResponseWriter, r *http.Reques
Err(err).
Msg("Failed to read status line from cached response")
sc.metrics.IncrementErrors()
sc.metrics.IncrementServiceError("cache_corrupt")
http.Error(w, "Internal server error", http.StatusInternalServerError)
return
}
@@ -282,6 +283,7 @@ func (sc *SteamCache) streamCachedResponse(w http.ResponseWriter, r *http.Reques
Err(err).
Msg("Failed to parse status code from cached response")
sc.metrics.IncrementErrors()
sc.metrics.IncrementServiceError("cache_corrupt")
http.Error(w, "Internal server error", http.StatusInternalServerError)
return
}
@@ -297,6 +299,7 @@ func (sc *SteamCache) streamCachedResponse(w http.ResponseWriter, r *http.Reques
Err(err).
Msg("Failed to read headers from cached response")
sc.metrics.IncrementErrors()
sc.metrics.IncrementServiceError("cache_corrupt")
http.Error(w, "Internal server error", http.StatusInternalServerError)
return
}
@@ -742,7 +745,7 @@ func (sc *SteamCache) removeCoalescedRequest(cacheKey string) {
}
// isTrustedProxy reports whether ipStr matches any CIDR or IP in trustedProxies list.
// Used for P1-02 safe client IP extraction (rightmost untrusted wins).
// Used for safe client IP extraction from X-Forwarded-For (rightmost untrusted proxy wins).
func isTrustedProxy(ipStr string, trustedProxies []string) bool {
ip := net.ParseIP(strings.TrimSpace(ipStr))
if ip == nil {
@@ -767,7 +770,7 @@ func isTrustedProxy(ipStr string, trustedProxies []string) bool {
}
// getClientIP extracts the client IP address from the request.
// P1-02: if trustedProxies empty (default), ALWAYS use RemoteAddr only (spoof-proof).
// If trustedProxies is empty (the safe default), always use RemoteAddr only (prevents spoofing).
// When list non-empty, use rightmost-untrusted from XFF+Remote chain (proper proxy extraction, not naive first XFF).
// X-Real-IP is ignored for simplicity/safety (XFF is the standard multi-hop header).
// Security: prevents clients spoofing XFF to bypass per-client rate limits.
@@ -779,7 +782,7 @@ func getClientIP(r *http.Request, trustedProxies []string) string {
}
if len(trustedProxies) == 0 {
// Conservative safe default: never trust forwarded headers (P1-02)
// Conservative safe default: never trust forwarded headers (spoof prevention)
return remoteIP
}
@@ -882,7 +885,7 @@ type SteamCache struct {
clientRequestsMu sync.RWMutex
maxRequestsPerClient int64
// P1 config (plumbed)
// Hardening config fields (plumbed)
maxObjectSize int64
trustedProxies []string
@@ -898,12 +901,12 @@ type SteamCache struct {
}
// New creates a new SteamCache instance.
// Since P0-01, it returns an error (instead of panicking) on invalid memorySize or diskSize strings from units.FromHumanSize.
// Since P1, also validates maxObjectSize (P1-01) and accepts trustedProxies (P1-02).
// Empty maxObjectSize or nil trustedProxies are normalized to safe defaults ("0", []) *before* parsing.
// Callers (including cmd/root.go and all tests) must check the returned error.
// Migration note (P1): the 2 new positional params on New() are breaking for direct importers.
// Prefer NewWithOptions (or config file) for forward compatibility. See README "Migration / Breaking Changes (P1)".
// Returns an error (instead of panicking) on invalid memorySize or diskSize strings.
// Also validates maxObjectSize and accepts trustedProxies for X-Forwarded-For handling.
// Empty maxObjectSize or nil trustedProxies are normalized to safe defaults before parsing.
// Callers must check the returned error.
// The two new positional parameters are a breaking change for direct importers of the simple constructor.
// Prefer NewWithOptions (or config file) for forward compatibility. See README migration notes.
func New(address string, memorySize string, diskSize string, diskPath, upstream, memoryGC, diskGC string, maxConcurrentRequests int64, maxRequestsPerClient int64, maxObjectSize string, trustedProxies []string) (*SteamCache, error) {
memorysize, err := units.FromHumanSize(memorySize)
if err != nil {
@@ -915,7 +918,7 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
return nil, fmt.Errorf("invalid disk size: %w", err)
}
// P1 defaults *before* parse (fixes zero-value Options / NewWithOptions("") callers)
// Apply safe defaults before parsing user-provided sizes (handles zero-value Options)
if maxObjectSize == "" {
maxObjectSize = "0"
}
@@ -1046,7 +1049,7 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
maxRequestsPerClient: maxRequestsPerClient,
clientLimiterCleanupStop: make(chan struct{}),
// P1 plumbed
// Hardening config plumbed
maxObjectSize: maxObjBytes,
trustedProxies: trustedProxies,
@@ -1160,7 +1163,7 @@ func (sc *SteamCache) Shutdown() {
}
}
sc.wg.Wait()
// Brief reap window after stopping workers (helps T2 delta checks see low goroutine counts immediately; workers have already exited their loops).
// Brief reap window after stopping workers (helps goroutine delta checks see low counts quickly).
time.Sleep(10 * time.Millisecond)
})
}
@@ -1178,8 +1181,8 @@ func (sc *SteamCache) GetMetrics() *metrics.Stats {
return sc.metrics.GetStats()
}
// Minimal Options + NewWithOptions for T3 (small, delegates to positional New).
// NewWithOptions propagates the P0-01 error return (see New godoc).
// Minimal Options + NewWithOptions usage (delegates to the main positional constructor).
// NewWithOptions propagates the error return from New (see New godoc).
type Options struct {
Address string
MemorySize string
@@ -1191,7 +1194,7 @@ type Options struct {
MaxConcurrentRequests int64
MaxRequestsPerClient int64
// P1: new config plumbed for hardening (smallest extension)
// New config fields for hardening (max object size + trusted proxies)
MaxObjectSize string
TrustedProxies []string
}
@@ -1213,13 +1216,14 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Keep-Alive", "timeout=300, max=1000")
// Apply global concurrency limit first
// C4 (smallest): propagate r.Context for cancellation (review item)
// Propagate request context for cancellation support
if err := sc.requestSemaphore.Acquire(r.Context(), 1); err != nil {
// Capacity rejections are counted in Errors + RateLimited but intentionally *before* TotalRequests.
// This preserves original hit-rate / processed-traffic semantics for accepted requests only.
// (All other 5xx occur after Total inc.)
sc.metrics.IncrementRateLimited()
sc.metrics.IncrementErrors()
sc.metrics.IncrementServiceError("rate_limit")
logger.Logger.Warn().Str("client_ip", clientIP).Msg("Server at capacity, rejecting request")
http.Error(w, "Server busy, please try again later", http.StatusServiceUnavailable)
return
@@ -1232,7 +1236,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Apply per-client rate limiting
clientLimiter := sc.getOrCreateClientLimiter(clientIP)
// C4 (smallest): per-client too
// Per-client request limiting (context aware)
if err := clientLimiter.semaphore.Acquire(r.Context(), 1); err != nil {
logger.Logger.Warn().
Str("client_ip", clientIP).
@@ -1282,6 +1286,18 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
fmt.Fprintf(w, "cache_coalesced %d\n", stats.CacheCoalesced)
fmt.Fprintf(w, "errors %d\n", stats.Errors)
fmt.Fprintf(w, "rate_limited %d\n", stats.RateLimited)
fmt.Fprintf(w, "upstream_errors %d\n", stats.UpstreamErrors)
fmt.Fprintf(w, "cache_write_failures %d\n", stats.CacheWriteFailures)
fmt.Fprintf(w, "memory_cache_hits %d\n", stats.MemoryCacheHits)
fmt.Fprintf(w, "disk_cache_hits %d\n", stats.DiskCacheHits)
fmt.Fprintf(w, "promotions %d\n", stats.Promotions)
fmt.Fprintf(w, "evictions %d\n", stats.Evictions)
for svc, cnt := range stats.ServiceErrors {
fmt.Fprintf(w, "service_errors{service=%q} %d\n", svc, cnt)
}
for svc, cnt := range stats.ServiceRequests {
fmt.Fprintf(w, "service_requests{service=%q} %d\n", svc, cnt)
}
fmt.Fprintf(w, "hit_rate %.4f\n", stats.HitRate)
fmt.Fprintf(w, "avg_response_time_ms %.2f\n", float64(stats.AvgResponseTime.Nanoseconds())/1e6)
fmt.Fprintf(w, "total_bytes_served %d\n", stats.TotalBytesServed)
@@ -1547,6 +1563,8 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
sc.metrics.IncrementErrors()
sc.metrics.IncrementUpstreamErrors()
sc.metrics.IncrementServiceError("upstream")
http.Error(w, "Failed to fetch the requested URL", http.StatusInternalServerError)
return
}
@@ -1560,6 +1578,8 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
sc.metrics.IncrementErrors()
sc.metrics.IncrementUpstreamErrors()
sc.metrics.IncrementServiceError("upstream")
http.Error(w, "Failed to fetch the requested URL", http.StatusInternalServerError)
return
}
@@ -1584,14 +1604,14 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
expectedSize := resp.ContentLength
// Reject only truly invalid content lengths (zero or negative)
// P1-01: when limit set, treat unknown/lying-CL as potential oversize (413) instead of 502.
// When max object size limit is set, treat unknown or lying Content-Length as potential oversize (return 413).
if expectedSize <= 0 {
if sc.maxObjectSize > 0 {
logger.Logger.Warn().
Str("url", req.URL.String()).
Int64("content_length", expectedSize).
Int64("max_object_size", sc.maxObjectSize).
Msg("Chunked/unknown CL with limit set - treating as potential oversize (P1-01)")
Msg("Chunked/unknown Content-Length with size limit set - treating as potential oversize")
if isNew {
coalescedReq.complete(nil, fmt.Errorf("chunked response with size limit"))
}
@@ -1610,7 +1630,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Content length is valid - no size restrictions to keep logs clean
// P1-01: bounded response size to prevent OOM (cap approach chosen for minimal VFS impact vs full streaming tee).
// Bounded response size to prevent OOM (capped reader chosen for minimal VFS impact).
// Large objects still served if <= limit; >limit returns 413 without caching or unbounded ReadAll.
// Coalesced paths also protected (leader enforces before buffering).
// Security: mitigates DoS via huge malicious upstream responses/manifests.
@@ -1619,7 +1639,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Str("url", req.URL.String()).
Int64("content_length", expectedSize).
Int64("max_object_size", sc.maxObjectSize).
Msg("Response exceeds max_object_size limit - rejecting to prevent OOM (P1-01)")
Msg("Response exceeds configured max object size limit - rejecting to prevent OOM")
if isNew {
coalescedReq.complete(nil, fmt.Errorf("response too large: %d > %d", expectedSize, sc.maxObjectSize))
}
@@ -1633,7 +1653,7 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
validationPassed := true
// Read the entire response body into memory to avoid consuming it twice
// P1-01: LimitReader caps even if CL lied small (protects against chunked/lying-CL OOM).
// LimitReader caps the body even if the client lied about Content-Length.
readLimit := resp.ContentLength
if sc.maxObjectSize > 0 && (readLimit <= 0 || readLimit > sc.maxObjectSize) {
readLimit = sc.maxObjectSize
@@ -1707,6 +1727,8 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Str("url", urlPath).
Err(err).
Msg("Failed to serialize cache file")
sc.metrics.IncrementCacheWriteFailures()
sc.metrics.IncrementServiceError("serialize")
} else {
// Store the serialized cache data
cacheWriter, err := sc.vfs.Create(cachePath, int64(len(cacheData)))
@@ -1724,6 +1746,8 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Int("written", bytesWritten).
Err(cacheErr).
Msg("Cache write failed or incomplete - removing corrupted entry")
sc.metrics.IncrementCacheWriteFailures()
sc.metrics.IncrementServiceError("cache_write")
sc.vfs.Delete(cachePath)
} else {
// Track successful cache write
@@ -1741,6 +1765,8 @@ func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Str("url", urlPath).
Err(err).
Msg("Failed to create cache file")
sc.metrics.IncrementCacheWriteFailures()
sc.metrics.IncrementServiceError("cache_create")
}
}