Files
traefikoidc/singleton_resources_test.go
T
lukaszraczylo 546ceb949c security: remediate audit findings (ranks 1–16 + 22 Lows) + yaegi load validation (#144)
* fix(security): encrypt session cookies + fail closed on invalid config

Batch 1 of security audit remediation (ranks 1, 2, 6).

- session.go: derive independent HMAC + AES-256 keys via stdlib HKDF-SHA256
  and build the gorilla cookie store with both, so session cookies are now
  encrypted, not merely signed. The single-key store previously left OIDC
  access/refresh/ID tokens recoverable from raw cookie bytes. Cookie format
  changes, so existing sessions are invalidated on deploy (one-time re-login).
- main.go: call config.Validate() at construction and error out on failure,
  instead of silently substituting a public hardcoded encryption key for
  empty/short keys (which allowed session forgery). The yaegi analyzer
  passes via .traefik.yml testData.
- settings.go: isValidSecureURL permits plaintext HTTP for loopback hosts
  only (RFC 8252); remote providers must still use HTTPS.
- tests: complete configs that did not satisfy Validate(); add regression
  tests in security_audit_fixes_test.go.

Configs below documented minimums (rateLimit < 10, key < 32 chars) are now
rejected at startup (fail closed).

* fix(security): validate discovered OIDC endpoints + pin introspection host

Batch 2 of security audit remediation (ranks 3, 4).

- url_helpers.go: add validateDiscoveredEndpoint, an SSRF screen for endpoints
  taken from the provider discovery document (jwks_uri, token, authorization,
  revocation, end_session, introspection, registration). Blocks link-local
  (cloud metadata 169.254.169.254), multicast, unspecified and private
  addresses (unless allowPrivateIPAddresses); blocks loopback unless the
  configured providerURL is itself loopback (dev/test). Cross-domain JWKS
  hosts (e.g. Google) stay allowed. Add sameHost helper.
- main.go: updateMetadataEndpoints screens every discovered endpoint and
  blanks any that fail (fail closed downstream). The introspection endpoint
  carries the client secret via HTTP Basic, so it is additionally pinned to
  the providerURL host to stop a poisoned discovery document exfiltrating the
  secret to an attacker-controlled host.
- tests: regression tests for the SSRF guard and the host pin.

* fix(security): close open redirects + anchor excluded-URL matching

Batch 3 of security audit remediation (ranks 5, 14, 15).

- auth_flow.go: run the stored incoming path through normalizeLogoutPath
  before using it as the post-login redirect, so //evil.com and /\evil.com
  payloads become host-relative (open-redirect, rank 5).
- url_helpers.go: excluded-URL matching is anchored at a natural boundary
  (exact, sub-path "/", or file extension "."), so excluding "/public" no
  longer also bypasses auth on "/publicsecret"; "/favicon" still matches
  "/favicon.ico" (rank 14).
- internal/utils: X-Forwarded-Host is sanitized (first value only; reject
  CRLF/whitespace/multi-value) before building redirect URLs (rank 15).
- helpers.go: the logout redirect used when there is no provider end-session
  endpoint is host-relative, never an absolute URL derived from the
  client-controllable request host (logout open-redirect, rank 15).
- tests: update two logout cases that asserted the old absolute redirect;
  add regression tests.

* fix(security): reject unverified Azure tokens; fix transport TLS reuse

Batch 4 of security audit remediation (ranks 7, 11).

- token_validation_rs.go: an Azure nonce-bearing access token that cannot be
  cryptographically verified no longer returns "authenticated" when there is
  no ID token to corroborate it; it refreshes (if possible) or forces
  re-authentication instead of failing open (rank 7).
- http_client_pool.go: the at-limit transport-reuse path now takes the write
  lock before mutating refCount (fixes a data race) and only reuses a
  transport whose TLS settings (CA pool + InsecureSkipVerify) match the
  caller's, never one with a different trust store; if none matches it returns
  nil so the caller falls back to a verifying default transport (rank 11).
- tests: add a transport-pool TLS-isolation regression test.

* fix(security): stop logging templated header values (token leak)

Batch 5 of security audit remediation (rank 16).

middleware.go: templated downstream headers commonly carry the access token
(e.g. "Authorization: Bearer {{.AccessToken}}"). The debug log line printed
the full header value, leaking credentials into logs. Log the header name and
byte length instead.

* fix(security): cache-key collision, cache-config divergence, fleet cleanup

Batch 6 of security audit remediation (ranks 9, 10, 12).

- token_manager.go: detectTokenType keys its cache on a SHA-256 hash of the
  full token instead of the first 32 chars (which are only the base64url JWT
  header). Distinct tokens sharing alg+kid no longer collide and get
  mis-classified (rank 10).
- cache_manager.go: the process-global cache manager is initialized once and
  shared across plugin instances; it now logs a loud warning when a later
  instance requests a different explicit Redis backend that is silently
  ignored, surfacing the cross-instance state-isolation hazard (rank 9).
- singleton_resources.go / main.go / utilities.go: track a process-global live
  instance count; the shared singleton-token-cleanup task is stopped only when
  the LAST instance shuts down, so one instance's Close() (e.g. a config reload)
  no longer kills cleanup for surviving instances (rank 12).
- tests: update TestDetectTokenTypeCaching for the new key; add regression tests.

* fix(security): bound introspection cache + cookie lifetime to config

Batch 7 of security audit remediation (ranks 8, 13).

- token_introspection.go: when requireTokenIntrospection is enabled, cap the
  positive introspection-result cache at 30s (instead of 5m) so a token
  revoked at the provider stops passing within ~30s, matching the operator's
  near-real-time revocation expectation (rank 8).
- session.go: bind the cookie store's MaxAge to the configured sessionMaxAge,
  so the cookie codec's cryptographic timestamp validity is no longer fixed at
  gorilla's 30-day default; a stolen cookie is valid only for the configured
  session lifetime (rank 13).
- tests: add a cookie-lifetime regression test.

* fix(security): low-severity hardening (cache, DoS caps, PKCE, throttle)

Batch 8 of security audit remediation — low severity
(ranks 24, 25, 27, 29, 31, 36, 37, 41, 45, 46, 49).

- universal_cache.go: updateLocalCache updates an existing key in place instead
  of orphaning its LRU element and double-counting currentSize/currentMemory
  (rank 36 — the only production-reachable bug in this batch).
- jwk.go / metadata_cache.go / token_introspection.go: bound response bodies
  with io.LimitReader (1 MiB) to prevent memory exhaustion from a hostile or
  buggy provider (ranks 24, 25).
- jwk.go: skip JWKs not usable for signature verification (use != sig, or
  key_ops without "verify") when building the key set (rank 49).
- auth_flow.go: fail closed at the callback when PKCE is enabled but the code
  verifier is missing, instead of silently dropping it (rank 27).
- utilities.go / main.go: match allowedUserDomains case-insensitively (rank 31).
- bearer_auth.go: a single success no longer wipes an active per-IP penalty;
  the counter resets only when no penalty is in effect (rank 29).
- main.go: handle (not discard) the NewSessionManager error (rank 37).
- error_recovery.go: take a write lock in isServiceDegraded (it deletes from a
  map); compare retryable-error substrings case-insensitively (ranks 45, 46).
- singleton_resources.go: bind the generic-cache cleanup goroutine to the
  resource-manager shutdown channel so it cannot outlive its owner (rank 41).
- tests: update the bearer throttle test to the corrected penalty semantics.

* fix(security): header sanitization, issuer pinning, fail-closed paths

Batch 9 of security audit remediation (ranks 18, 19, 20, 21, 22, 30, 33, 34).

- middleware.go / bearer_auth.go: sanitize claim-derived values on the cookie
  auth path before injecting them into downstream headers. Drop group/role and
  identifier values containing control chars, bidi-override runes, or the
  , ; = delimiters (a comma would inject phantom entries into X-User-Groups);
  reject control/bidi/over-length in rendered templated header output (but
  permit , ; = in free-form values such as a bearer token). The bearer path
  already sanitized; the cookie path did not (ranks 33, 34).
- main.go / metadata_cache.go: pin the discovered issuer to the configured
  provider host (sameHost) and refuse/never-cache a mismatch, so a poisoned
  discovery document cannot redefine the JWT trust anchor (ranks 21, 22).
- token_introspection.go: when a distinct API audience is configured, fail
  closed on a missing or mismatched introspection audience; aud parsed as
  string-or-array per RFC 7662 (rank 19).
- logout.go: front-channel logout requires a matching issuer; an empty iss is
  rejected (blocks unauthenticated forced-logout via a known sid) (rank 30).
- token_validation_rs.go: an opaque access token with no ID token and no
  successful introspection fails closed (re-auth) instead of authenticating
  (ranks 18, 20).
- tests: realistic same-host provider mocks; regression tests for the header
  sanitization distinction and the fail-closed paths.

* chore(security): remove unwired dead code with latent footguns

Batch 10 of security audit remediation — delete confirmed-dead, unwired
subsystems (ranks 26, 35, 50). None had a production caller (grep-verified);
removal eliminates the latent footguns and ~2.1k lines of dead code.

- token_validator.go (deleted): an unused *TokenValidator whose validateJWT set
  Valid=true with NO signature verification — a severe footgun if ever wired
  (rank 50). The wired RS-aware validators are unaffected.
- security_monitoring.go (deleted): an unused *SecurityMonitor / ExtractClientIP
  that trusted spoofable X-Forwarded-For / X-Real-IP. The live bearer throttle
  uses clientIPForBearer (RemoteAddr-only), unchanged (rank 35).
- dynamic_client_registration.go: removed the RFC 7592 management methods
  (Update/Read/DeleteClientRegistration) that dereferenced an attacker-
  influenced RegistrationClientURI with the registration token attached and no
  HTTPS/SSRF gate, and had no callers. The wired RFC 7591 RegisterClient and
  credential-store helpers are kept (rank 26).
- tests: removed the tests covering the deleted code.

* chore: add Makefile with yaegi load validation

No Makefile existed. The new `yaegi-validate` target interprets the plugin
under the yaegi interpreter the same way Traefik loads it, catching yaegi-only
incompatibilities (unsupported stdlib symbols, reflection edge cases) that the
native `go build` / `go test` toolchain does not. Importing the plugin forces
yaegi to interpret every file plus its vendored deps; CreateConfig + New
exercise the instantiation path.

- cmd/yaegicheck/main.go: the load driver, marked //go:build ignore so it is
  excluded from `go build ./...` (avoids VCS-stamping a main binary, which
  fails in git-worktree layouts) yet is run explicitly by yaegi.
- Makefile: build / fmt / vet / lint / test / vendor / yaegi-validate / check
  targets; `make check` runs vet + tests + yaegi-validate.

Verified: `make yaegi-validate` passes on this branch — the HKDF cookie
encryption, net-based endpoint validation, and claim sanitizers all interpret
and instantiate cleanly under yaegi.

* ci: bump workflow Go toolchain to 1.25; pin yaegi-validate to v0.16.1

Traefik v3.7.1 (the deployed version) is built with `go 1.25.0`, so the PR and
release workflows now use Go 1.25.x to match the toolchain Traefik uses.

Important distinction: the CI Go version is the build TOOLCHAIN. The plugin's
actual interpreter-compatibility ceiling is the yaegi version Traefik bundles
(v0.16.1, which declares go 1.21 and ships a ~Go 1.22 stdlib symbol surface),
NOT the CI Go version. That ceiling is enforced by `make yaegi-validate` plus
the go.mod language directive — e.g. it is why HKDF is hand-rolled with
hmac+sha256 rather than Go 1.24's crypto/hkdf, which yaegi v0.16.1 lacks.

Also pin Makefile YAEGI_VERSION to v0.16.1 (what Traefik v3.7.1 vendors) so
yaegi-validate exercises the real deployed interpreter instead of @latest,
which could pass on a newer yaegi that supports symbols the deployed one does
not.

* docs: align README/CONFIGURATION with branch behavior changes

- excludedURLs: documented as segment/extension-boundary matching (was
  "prefix-matched") — "/public" no longer also matches "/publicsecret" (rank 14).
- Front-channel logout now requires a matching `iss`; requests without one are
  rejected with 400 (rank 30).
- Add an "Upgrading from an earlier release" note: session cookies are now
  AES-256 encrypted with lifetime tracking sessionMaxAge (one-time re-login on
  upgrade), and invalid configuration (rateLimit < 10, key < 32 bytes, missing
  callbackURL, non-HTTPS remote providerURL) now fails closed at startup.

* fix: remove staticcheck-flagged unused functions; wire staticcheck into make check

CI Static Analysis (standalone staticcheck) failed with U1000 "unused":
- dynamic_client_registration.go: deleteCredentialsFromStore — its only caller
  was the RFC 7592 DeleteClientRegistration removed in the dead-code batch.
- token_test.go: createTestJWTSimple — its only callers were the TokenValidator
  tests removed in the same batch.
Both confirmed to have zero remaining callers and removed. build / vet /
go test ./... / staticcheck ./... all green.

The pre-commit hook runs golangci-lint, but CI runs standalone staticcheck
(which flags U1000). Add a `staticcheck` Makefile target and include it in
`make check` so this class of finding is caught locally before push.

* fix(test): stabilize flaky TestWorkerPool_TaskPanic

tasksFailed is incremented in the worker's deferred recover(), which runs after the panicking task's own defer wg.Done(). wg.Wait() could therefore return before the failure was recorded, so reading the counter immediately raced and flaked on slow CI runners. Poll until the failure lands (2s budget) instead. Verified 200x plain + 50x under -race/GOMAXPROCS=1.
2026-05-30 14:10:32 +01:00

944 lines
25 KiB
Go

package traefikoidc
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"runtime"
"sync"
"sync/atomic"
"testing"
"time"
)
// TestSingletonResourceManager tests the singleton resource manager implementation
func TestSingletonResourceManager(t *testing.T) {
t.Run("SingletonInstance", func(t *testing.T) {
// Test that GetResourceManager returns the same instance
rm1 := GetResourceManager()
rm2 := GetResourceManager()
if rm1 != rm2 {
t.Error("GetResourceManager did not return singleton instance")
}
})
t.Run("ThreadSafeInitialization", func(t *testing.T) {
// Reset singleton for test
resetResourceManagerForTesting()
const numGoroutines = 100
instances := make([]*ResourceManager, numGoroutines)
var wg sync.WaitGroup
for i := 0; i < numGoroutines; i++ {
wg.Add(1)
go func(idx int) {
defer wg.Done()
instances[idx] = GetResourceManager()
}(i)
}
wg.Wait()
// Verify all instances are the same
first := instances[0]
for i := 1; i < numGoroutines; i++ {
if instances[i] != first {
t.Errorf("Instance %d differs from first instance", i)
}
}
})
t.Run("SharedHTTPClient", func(t *testing.T) {
rm := GetResourceManager()
client1 := rm.GetHTTPClient("test-client-1")
client2 := rm.GetHTTPClient("test-client-1")
if client1 != client2 {
t.Error("GetHTTPClient did not return same client for same key")
}
client3 := rm.GetHTTPClient("test-client-2")
if client1 == client3 {
t.Error("GetHTTPClient returned same client for different keys")
}
})
t.Run("SharedCache", func(t *testing.T) {
rm := GetResourceManager()
cache1 := rm.GetCache("test-cache-1")
cache2 := rm.GetCache("test-cache-1")
if cache1 != cache2 {
t.Error("GetCache did not return same cache for same key")
}
})
t.Run("SingletonTaskRegistry", func(t *testing.T) {
rm := GetResourceManager()
err := rm.RegisterBackgroundTask("test-task", 1*time.Second, func() {
// Test task
})
if err != nil {
t.Errorf("Failed to register task: %v", err)
}
// Try to register same task again - should return existing
err = rm.RegisterBackgroundTask("test-task", 1*time.Second, func() {
// Duplicate task
})
if err != nil {
t.Errorf("Failed to handle duplicate task registration: %v", err)
}
})
t.Run("MultiRealmMetadataRefreshTaskNaming", func(t *testing.T) {
// This test verifies that different provider URLs generate different task names
// which is critical for multi-realm Keycloak support (PR #88)
// Reset singletons for clean test state
resetResourceManagerForTesting()
ResetGlobalTaskRegistry()
defer ResetGlobalTaskRegistry()
rm := GetResourceManager()
// Simulate different Keycloak realms
providerURL1 := "https://keycloak.example.com/realms/realm1"
providerURL2 := "https://keycloak.example.com/realms/realm2"
// Generate task names using the same logic as startMetadataRefresh
hash1 := sha256.Sum256([]byte(providerURL1))
taskName1 := "singleton-metadata-refresh-" + hex.EncodeToString(hash1[:])[0:6]
hash2 := sha256.Sum256([]byte(providerURL2))
taskName2 := "singleton-metadata-refresh-" + hex.EncodeToString(hash2[:])[0:6]
// Verify task names are different
if taskName1 == taskName2 {
t.Errorf("Task names should be different for different provider URLs: %s vs %s", taskName1, taskName2)
}
// Register both tasks
task1Called := int32(0)
task2Called := int32(0)
err := rm.RegisterBackgroundTask(taskName1, 100*time.Millisecond, func() {
atomic.AddInt32(&task1Called, 1)
})
if err != nil {
t.Errorf("Failed to register task 1: %v", err)
}
err = rm.RegisterBackgroundTask(taskName2, 100*time.Millisecond, func() {
atomic.AddInt32(&task2Called, 1)
})
if err != nil {
t.Errorf("Failed to register task 2: %v", err)
}
// Start both tasks
_ = rm.StartBackgroundTask(taskName1)
_ = rm.StartBackgroundTask(taskName2)
// Wait for tasks to execute
time.Sleep(250 * time.Millisecond)
// Verify both tasks are running independently
if !rm.IsTaskRunning(taskName1) {
t.Error("Task 1 should be running")
}
if !rm.IsTaskRunning(taskName2) {
t.Error("Task 2 should be running")
}
// Verify both tasks were called (at least once)
if atomic.LoadInt32(&task1Called) == 0 {
t.Error("Task 1 should have been called at least once")
}
if atomic.LoadInt32(&task2Called) == 0 {
t.Error("Task 2 should have been called at least once")
}
// Stop both tasks
_ = rm.StopBackgroundTask(taskName1)
_ = rm.StopBackgroundTask(taskName2)
// Verify tasks are stopped
time.Sleep(50 * time.Millisecond)
if rm.IsTaskRunning(taskName1) {
t.Error("Task 1 should be stopped")
}
if rm.IsTaskRunning(taskName2) {
t.Error("Task 2 should be stopped")
}
t.Logf("Successfully verified multi-realm task isolation: task1=%s, task2=%s", taskName1, taskName2)
})
t.Run("ReferenceCountingCleanup", func(t *testing.T) {
rm := GetResourceManager()
// Add reference
rm.AddReference("test-instance-1")
// Get reference count
if rm.GetReferenceCount("test-instance-1") != 1 {
t.Error("Reference count should be 1")
}
// Add another reference
rm.AddReference("test-instance-1")
if rm.GetReferenceCount("test-instance-1") != 2 {
t.Error("Reference count should be 2")
}
// Remove reference
rm.RemoveReference("test-instance-1")
if rm.GetReferenceCount("test-instance-1") != 1 {
t.Error("Reference count should be 1 after removal")
}
// Remove last reference
rm.RemoveReference("test-instance-1")
if rm.GetReferenceCount("test-instance-1") != 0 {
t.Error("Reference count should be 0 after removing all references")
}
})
t.Run("GracefulShutdown", func(t *testing.T) {
rm := GetResourceManager()
// Register a task with atomic variable to avoid race condition
var taskExecuted int32
err := rm.RegisterBackgroundTask("shutdown-test-task", 100*time.Millisecond, func() {
atomic.StoreInt32(&taskExecuted, 1)
})
if err != nil {
t.Errorf("Failed to register task: %v", err)
}
// Start the task
rm.StartBackgroundTask("shutdown-test-task")
// Wait for task to execute at least once
time.Sleep(150 * time.Millisecond)
if atomic.LoadInt32(&taskExecuted) == 0 {
t.Error("Task was not executed")
}
// Shutdown
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
err = rm.Shutdown(ctx)
if err != nil {
t.Errorf("Shutdown failed: %v", err)
}
// Verify task is stopped
if rm.IsTaskRunning("shutdown-test-task") {
t.Error("Task should be stopped after shutdown")
}
})
}
// createMockOIDCServer creates a mock OIDC server for testing
func createMockOIDCServer() *httptest.Server {
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/.well-known/openid-configuration":
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"issuer": "https://example.com",
"authorization_endpoint": "https://example.com/authorize",
"token_endpoint": "https://example.com/token",
"jwks_uri": "https://example.com/jwks",
"userinfo_endpoint": "https://example.com/userinfo",
})
case "/jwks":
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"keys": []interface{}{},
})
default:
w.WriteHeader(http.StatusNotFound)
}
}))
}
// TestContextAwareGoroutineManagement tests context-aware goroutine management
func TestContextAwareGoroutineManagement(t *testing.T) {
t.Run("GoroutineCleanupOnContextCancel", func(t *testing.T) {
// Reset singletons to ensure clean state
resetResourceManagerForTesting()
ResetUniversalCacheManagerForTesting()
defer ResetUniversalCacheManagerForTesting()
// Create mock OIDC server
mockServer := createMockOIDCServer()
defer mockServer.Close()
initialGoroutines := runtime.NumGoroutine()
ctx, cancel := context.WithCancel(context.Background())
// Create a TraefikOidc instance with context
config := &Config{
ProviderURL: mockServer.URL,
ClientID: "test-client",
ClientSecret: "test-secret",
CallbackURL: "/callback",
SessionEncryptionKey: "test-encryption-key-32-bytes-long",
RateLimit: 100,
}
plugin, err := NewWithContext(ctx, config, nil, "test")
if err != nil {
t.Fatalf("Failed to create plugin: %v", err)
}
// Wait for goroutines to start
time.Sleep(100 * time.Millisecond)
midGoroutines := runtime.NumGoroutine()
if midGoroutines <= initialGoroutines {
t.Error("No goroutines were created")
}
// Cancel context
cancel()
// Close the plugin to trigger cleanup
plugin.Close()
// Wait for cleanup
time.Sleep(500 * time.Millisecond)
finalGoroutines := runtime.NumGoroutine()
// Allow for some singleton background goroutines (caches, pools, etc.)
// These are shared across all instances and persist for the test duration
tolerance := 10
if finalGoroutines > initialGoroutines+tolerance {
t.Errorf("Goroutine leak detected: initial=%d, final=%d", initialGoroutines, finalGoroutines)
}
})
t.Run("NoGoroutineLeakOnMultipleInstances", func(t *testing.T) {
// Reset singletons to ensure clean state
resetResourceManagerForTesting()
ResetUniversalCacheManagerForTesting()
defer ResetUniversalCacheManagerForTesting()
// Create mock OIDC servers
mockServer1 := createMockOIDCServer()
defer mockServer1.Close()
mockServer2 := createMockOIDCServer()
defer mockServer2.Close()
mockServer3 := createMockOIDCServer()
defer mockServer3.Close()
initialGoroutines := runtime.NumGoroutine()
configs := []Config{
{ProviderURL: mockServer1.URL, ClientID: "client1", ClientSecret: "secret1", CallbackURL: "/callback", SessionEncryptionKey: "test-encryption-key-32-bytes-long", RateLimit: 100},
{ProviderURL: mockServer2.URL, ClientID: "client2", ClientSecret: "secret2", CallbackURL: "/callback", SessionEncryptionKey: "test-encryption-key-32-bytes-long", RateLimit: 100},
{ProviderURL: mockServer3.URL, ClientID: "client3", ClientSecret: "secret3", CallbackURL: "/callback", SessionEncryptionKey: "test-encryption-key-32-bytes-long", RateLimit: 100},
}
var plugins []*TraefikOidc
var cancels []context.CancelFunc
// Create multiple instances
for i, config := range configs {
ctx, cancel := context.WithCancel(context.Background())
cancels = append(cancels, cancel)
plugin, err := NewWithContext(ctx, &config, nil, fmt.Sprintf("test-%d", i))
if err != nil {
t.Fatalf("Failed to create plugin %d: %v", i, err)
}
plugins = append(plugins, plugin)
}
// Wait for all goroutines to start
time.Sleep(200 * time.Millisecond)
midGoroutines := runtime.NumGoroutine()
// Cancel all contexts
for _, cancel := range cancels {
cancel()
}
// Close all plugins
for _, plugin := range plugins {
plugin.Close()
}
// Wait for cleanup
time.Sleep(500 * time.Millisecond)
finalGoroutines := runtime.NumGoroutine()
// Check for leaks
tolerance := 5
if finalGoroutines > initialGoroutines+tolerance {
t.Errorf("Goroutine leak with multiple instances: initial=%d, mid=%d, final=%d",
initialGoroutines, midGoroutines, finalGoroutines)
}
})
t.Run("SingletonTasksAcrossInstances", func(t *testing.T) {
// Reset singletons to ensure clean state
ResetGlobalTaskRegistry() // Reset circuit breaker and task registry
resetResourceManagerForTesting()
ResetUniversalCacheManagerForTesting()
defer ResetUniversalCacheManagerForTesting()
// Create mock OIDC servers
mockServers := make([]*httptest.Server, 3)
for i := 0; i < 3; i++ {
mockServers[i] = createMockOIDCServer()
defer mockServers[i].Close()
}
rm := GetResourceManager()
// Register singleton cleanup task
var cleanupCount int32
err := rm.RegisterBackgroundTask("singleton-cleanup", 100*time.Millisecond, func() {
atomic.AddInt32(&cleanupCount, 1)
})
if err != nil {
t.Fatalf("Failed to register singleton task: %v", err)
}
// Start the task
rm.StartBackgroundTask("singleton-cleanup")
// Create multiple plugin instances
var plugins []*TraefikOidc
for i := 0; i < 3; i++ {
ctx := context.Background()
config := &Config{
ProviderURL: mockServers[i].URL,
ClientID: fmt.Sprintf("client%d", i),
ClientSecret: fmt.Sprintf("secret%d", i),
CallbackURL: "/callback",
SessionEncryptionKey: "test-encryption-key-32-bytes-long",
RateLimit: 100,
}
plugin, err := NewWithContext(ctx, config, nil, fmt.Sprintf("test-%d", i))
if err != nil {
t.Fatalf("Failed to create plugin %d: %v", i, err)
}
plugins = append(plugins, plugin)
}
// Wait for cleanup to run at least 2 times with adaptive timeout
// This handles race detector overhead which can slow goroutine scheduling significantly
// When running as part of full test suite, CPU contention is even higher, so use generous timeout
const minExpectedCount = 2
const maxExpectedCount = 5
timeout := time.After(5 * time.Second)
ticker := time.NewTicker(50 * time.Millisecond)
defer ticker.Stop()
var count int32
waitLoop:
for {
select {
case <-ticker.C:
count = atomic.LoadInt32(&cleanupCount)
if count >= minExpectedCount {
// Success: reached minimum threshold
break waitLoop
}
case <-timeout:
count = atomic.LoadInt32(&cleanupCount)
t.Errorf("Timeout waiting for cleanup count to reach %d, got %d (race detector may be slowing execution)", minExpectedCount, count)
break waitLoop
}
}
// Verify count is within expected range (should be singleton, not running excessively)
if count > maxExpectedCount {
t.Errorf("Cleanup count too high: %d (expected max %d for singleton)", count, maxExpectedCount)
}
// Cleanup
for _, plugin := range plugins {
plugin.Close()
}
rm.StopBackgroundTask("singleton-cleanup")
})
}
// TestResourcePooling tests resource pooling implementation
func TestResourcePooling(t *testing.T) {
t.Run("GoroutinePoolLimiting", func(t *testing.T) {
rm := GetResourceManager()
// Configure pool with max workers
pool := rm.GetGoroutinePool("test-pool", 5) // Max 5 workers
if pool == nil {
t.Fatal("Failed to get goroutine pool")
}
// Submit more tasks than pool size
var taskCount int32
var runningCount int32
maxRunning := int32(0)
for i := 0; i < 20; i++ {
err := pool.Submit(func() {
atomic.AddInt32(&taskCount, 1)
current := atomic.AddInt32(&runningCount, 1)
// Track max concurrent tasks
for {
oldMax := atomic.LoadInt32(&maxRunning)
if current <= oldMax || atomic.CompareAndSwapInt32(&maxRunning, oldMax, current) {
break
}
}
time.Sleep(50 * time.Millisecond)
atomic.AddInt32(&runningCount, -1)
})
if err != nil {
t.Errorf("Failed to submit task %d: %v", i, err)
}
}
// Wait for all tasks to complete
pool.Wait()
// Verify all tasks executed
if atomic.LoadInt32(&taskCount) != 20 {
t.Errorf("Expected 20 tasks to execute, got %d", taskCount)
}
// Verify concurrency was limited
if atomic.LoadInt32(&maxRunning) > 5 {
t.Errorf("Max concurrent tasks exceeded pool size: %d > 5", maxRunning)
}
})
t.Run("PoolShutdown", func(t *testing.T) {
rm := GetResourceManager()
pool := rm.GetGoroutinePool("shutdown-pool", 3)
// Submit tasks
var completed int32
for i := 0; i < 10; i++ {
pool.Submit(func() {
time.Sleep(10 * time.Millisecond)
atomic.AddInt32(&completed, 1)
})
}
// Shutdown pool
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
err := pool.Shutdown(ctx)
if err != nil {
t.Errorf("Pool shutdown failed: %v", err)
}
// Try to submit after shutdown - should fail
err = pool.Submit(func() {
t.Error("Task should not execute after shutdown")
})
if err == nil {
t.Error("Expected error when submitting to shutdown pool")
}
})
t.Run("ResourceReuse", func(t *testing.T) {
rm := GetResourceManager()
// Get same pool multiple times
pool1 := rm.GetGoroutinePool("reuse-pool", 3)
pool2 := rm.GetGoroutinePool("reuse-pool", 3)
if pool1 != pool2 {
t.Error("Expected same pool instance for same key")
}
// Get HTTP client multiple times
client1 := rm.GetHTTPClient("reuse-client")
client2 := rm.GetHTTPClient("reuse-client")
if client1 != client2 {
t.Error("Expected same HTTP client instance for same key")
}
})
}
// TestBackwardCompatibility verifies the changes maintain backward compatibility
func TestBackwardCompatibility(t *testing.T) {
t.Run("LegacyNewFunction", func(t *testing.T) {
// Test that the original New function still works
config := &Config{
ProviderURL: "https://example.com",
ClientID: "test-client",
ClientSecret: "test-secret",
CallbackURL: "/callback",
SessionEncryptionKey: "test-encryption-key-32-bytes-long",
RateLimit: 100,
}
handler, err := New(context.Background(), nil, config, "test")
if err != nil {
t.Fatalf("Legacy New function failed: %v", err)
}
if handler == nil {
t.Fatal("Handler should not be nil")
}
// Cleanup - cast to TraefikOidc if needed
if plugin, ok := handler.(*TraefikOidc); ok {
plugin.Close()
}
})
t.Run("ExistingAPICompatibility", func(t *testing.T) {
config := &Config{
ProviderURL: "https://example.com",
ClientID: "test-client",
ClientSecret: "test-secret",
CallbackURL: "/callback",
SessionEncryptionKey: "test-encryption-key-32-bytes-long",
RateLimit: 100,
}
handler, _ := New(context.Background(), nil, config, "test")
// Test that the handler works
if handler == nil {
t.Error("Handler should not be nil")
}
// Cleanup - cast to TraefikOidc if needed
if plugin, ok := handler.(*TraefikOidc); ok {
plugin.Close()
}
})
}
// TestGoroutinePoolConditionVariable tests the condition variable-based Wait implementation
func TestGoroutinePoolConditionVariable(t *testing.T) {
t.Run("WaitDoesNotBusyPoll", func(t *testing.T) {
// This test verifies that Wait() uses condition variable instead of busy-polling
pool := NewGoroutinePool(2, nil)
defer pool.Shutdown(context.Background())
// Submit a slow task
var taskStarted, taskFinished int32
pool.Submit(func() {
atomic.StoreInt32(&taskStarted, 1)
time.Sleep(100 * time.Millisecond)
atomic.StoreInt32(&taskFinished, 1)
})
// Give task time to start
time.Sleep(10 * time.Millisecond)
// Measure CPU-time before Wait
startCPU := time.Now()
// Wait should block efficiently without consuming CPU
pool.Wait()
elapsed := time.Since(startCPU)
// Verify task completed
if atomic.LoadInt32(&taskFinished) != 1 {
t.Error("Task should have finished")
}
// Wait should have taken ~90ms (task was already running for ~10ms)
// If it was busy-polling, we would see much higher CPU usage
// This is a sanity check - the real proof is in profiling
if elapsed < 50*time.Millisecond {
t.Errorf("Wait returned too quickly: %v", elapsed)
}
})
t.Run("WaitReturnsImmediatelyWhenEmpty", func(t *testing.T) {
pool := NewGoroutinePool(2, nil)
defer pool.Shutdown(context.Background())
// Wait on empty pool should return immediately
start := time.Now()
pool.Wait()
elapsed := time.Since(start)
// Should return almost immediately
if elapsed > 10*time.Millisecond {
t.Errorf("Wait on empty pool took too long: %v", elapsed)
}
})
t.Run("ConcurrentSubmitAndWait", func(t *testing.T) {
pool := NewGoroutinePool(4, nil)
defer pool.Shutdown(context.Background())
var completed int32
const numTasks = 100
// Submit tasks concurrently
var wg sync.WaitGroup
for i := 0; i < numTasks; i++ {
wg.Add(1)
go func() {
defer wg.Done()
pool.Submit(func() {
time.Sleep(1 * time.Millisecond)
atomic.AddInt32(&completed, 1)
})
}()
}
wg.Wait() // Wait for all submissions
// Wait for all tasks to complete
pool.Wait()
if atomic.LoadInt32(&completed) != numTasks {
t.Errorf("Expected %d tasks completed, got %d", numTasks, completed)
}
})
t.Run("WaitWithTimeoutSuccess", func(t *testing.T) {
pool := NewGoroutinePool(2, nil)
defer pool.Shutdown(context.Background())
pool.Submit(func() {
time.Sleep(50 * time.Millisecond)
})
// Should complete within timeout
success := pool.WaitWithTimeout(1 * time.Second)
if !success {
t.Error("WaitWithTimeout should have succeeded")
}
})
t.Run("WaitWithTimeoutExpired", func(t *testing.T) {
pool := NewGoroutinePool(1, nil)
defer pool.Shutdown(context.Background())
pool.Submit(func() {
time.Sleep(500 * time.Millisecond)
})
// Should timeout
success := pool.WaitWithTimeout(50 * time.Millisecond)
if success {
t.Error("WaitWithTimeout should have timed out")
}
// Wait for actual completion to avoid goroutine leak in test
pool.Wait()
})
t.Run("PendingTasksCounter", func(t *testing.T) {
// Use pool with larger buffer (maxWorkers=2, buffer=4)
pool := NewGoroutinePool(2, nil)
defer pool.Shutdown(context.Background())
// Initially no pending tasks
if pool.PendingTasks() != 0 {
t.Errorf("Expected 0 pending tasks, got %d", pool.PendingTasks())
}
// Block both workers with signals that tasks have started
blocker1 := make(chan struct{})
blocker2 := make(chan struct{})
started1 := make(chan struct{})
started2 := make(chan struct{})
pool.Submit(func() {
close(started1)
<-blocker1
})
pool.Submit(func() {
close(started2)
<-blocker2
})
// Wait for both blocking tasks to actually start
<-started1
<-started2
// Submit 2 more tasks that will queue up (buffer can hold 4)
for i := 0; i < 2; i++ {
pool.Submit(func() {
time.Sleep(1 * time.Millisecond)
})
}
// Should have pending tasks (2 running + 2 queued = 4)
pending := pool.PendingTasks()
if pending != 4 {
t.Errorf("Expected 4 pending tasks, got %d", pending)
}
// Release blockers
close(blocker1)
close(blocker2)
// Wait for completion
pool.Wait()
// Should have no pending tasks
if pool.PendingTasks() != 0 {
t.Errorf("Expected 0 pending tasks after Wait, got %d", pool.PendingTasks())
}
})
t.Run("MultipleWaiters", func(t *testing.T) {
pool := NewGoroutinePool(2, nil)
defer pool.Shutdown(context.Background())
// Submit a slow task
pool.Submit(func() {
time.Sleep(100 * time.Millisecond)
})
// Multiple goroutines waiting
var waiters sync.WaitGroup
var waitCount int32
for i := 0; i < 5; i++ {
waiters.Add(1)
go func() {
defer waiters.Done()
pool.Wait()
atomic.AddInt32(&waitCount, 1)
}()
}
// All waiters should complete
waiters.Wait()
if atomic.LoadInt32(&waitCount) != 5 {
t.Errorf("Expected all 5 waiters to complete, got %d", waitCount)
}
})
t.Run("SubmitFailureDoesNotIncrementPending", func(t *testing.T) {
pool := NewGoroutinePool(1, nil)
// Shutdown the pool
pool.Shutdown(context.Background())
// Submit should fail
err := pool.Submit(func() {})
if err == nil {
t.Error("Submit should fail on shutdown pool")
}
// Pending tasks should still be 0
if pool.PendingTasks() != 0 {
t.Errorf("Pending tasks should be 0 after failed submit, got %d", pool.PendingTasks())
}
})
t.Run("PanicRecoveryDecrementsPending", func(t *testing.T) {
pool := NewGoroutinePool(2, nil)
defer pool.Shutdown(context.Background())
// Submit a task that panics
pool.Submit(func() {
panic("test panic")
})
// Submit a normal task
var normalCompleted int32
pool.Submit(func() {
atomic.StoreInt32(&normalCompleted, 1)
})
// Wait should still work (panic is recovered)
pool.Wait()
// Normal task should have completed
if atomic.LoadInt32(&normalCompleted) != 1 {
t.Error("Normal task should have completed despite panic in other task")
}
// Pending should be 0
if pool.PendingTasks() != 0 {
t.Errorf("Pending tasks should be 0 after Wait, got %d", pool.PendingTasks())
}
})
}
// BenchmarkGoroutinePoolWait benchmarks the Wait implementation
func BenchmarkGoroutinePoolWait(b *testing.B) {
pool := NewGoroutinePool(4, nil)
defer pool.Shutdown(context.Background())
b.ResetTimer()
for i := 0; i < b.N; i++ {
// Submit a quick task
pool.Submit(func() {})
pool.Wait()
}
}
// BenchmarkGoroutinePoolHighThroughput benchmarks high throughput scenario
func BenchmarkGoroutinePoolHighThroughput(b *testing.B) {
pool := NewGoroutinePool(8, nil)
defer pool.Shutdown(context.Background())
b.ResetTimer()
for i := 0; i < b.N; i++ {
for j := 0; j < 100; j++ {
pool.Submit(func() {
// Minimal work
_ = 1 + 1
})
}
pool.Wait()
}
}
// Helper function to reset singleton for testing
func resetResourceManagerForTesting() {
resourceManagerMutex.Lock()
defer resourceManagerMutex.Unlock()
if globalResourceManager != nil {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
globalResourceManager.Shutdown(ctx)
}
resourceManagerOnce = sync.Once{}
globalResourceManager = nil
}