diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 3bfffb60..8eba83dd 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -16,6 +16,19 @@ on: - '*.md' jobs: + GoTests: + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-go@v6 + with: + go-version-file: services/iam-cache/go.mod + cache-dependency-path: services/iam-cache/go.mod + - name: Test IAM cache + working-directory: services/iam-cache + run: | + go test -race ./... + go vet ./... Lint: runs-on: ubuntu-24.04 strategy: diff --git a/etc/exordos_core/iam_cache.json.example b/etc/exordos_core/iam_cache.json.example new file mode 100644 index 00000000..fef363c1 --- /dev/null +++ b/etc/exordos_core/iam_cache.json.example @@ -0,0 +1,10 @@ +{ + "public_listen_address": "127.0.0.1:11110", + "internal_listen_address": "127.0.0.1:11111", + "core_url": "http://127.0.0.1:11010", + "request_timeout": "5s", + "introspection_cache_ttl": "15s", + "introspection_cache_max_entries": 100000, + "jwks_cache_ttl": "1m", + "jwks_cache_max_entries": 1000 +} diff --git a/etc/systemd/exordos-iam-cache.service b/etc/systemd/exordos-iam-cache.service new file mode 100644 index 00000000..8e0728ed --- /dev/null +++ b/etc/systemd/exordos-iam-cache.service @@ -0,0 +1,13 @@ +[Unit] +Description=Exordos IAM Cache Service +After=network-online.target ec-user-api.service + +[Service] +TimeoutStopSec=10 +Restart=always +RestartSec=5s +KillSignal=SIGINT +ExecStart=/usr/bin/exordos-iam-cache -config /etc/exordos_core/iam_cache.json + +[Install] +WantedBy=multi-user.target diff --git a/exordos/images/bootstrap.sh b/exordos/images/bootstrap.sh index 8d290d15..acbd349a 100755 --- a/exordos/images/bootstrap.sh +++ b/exordos/images/bootstrap.sh @@ -85,6 +85,18 @@ if [[ -n "$PERSISTENT_DISK" ]]; then persist_migrate_complete fi +# Existing persistent installations predate the IAM cache configuration. Add +# its default config only when it is absent so operator changes survive future +# image updates. +if [[ ! -f "$GC_CFG_DIR/iam_cache.json" ]]; then + sudo install \ + -o root \ + -g root \ + -m 0644 \ + "$GC_PATH/etc/exordos_core/iam_cache.json.example" \ + "$GC_CFG_DIR/iam_cache.json" +fi + # Create deprecated path mkdir -p /var/lib/exordos/data @@ -159,6 +171,7 @@ fi log "systemctl enable --now ec-services" sudo systemctl enable --now \ ec-user-api \ + exordos-iam-cache \ ec-orch-api \ ec-status-api \ ec-boot-api \ diff --git a/exordos/images/install.sh b/exordos/images/install.sh index d098d9f2..c96a9f11 100644 --- a/exordos/images/install.sh +++ b/exordos/images/install.sh @@ -32,6 +32,7 @@ GC_PG_PASS="exordos_core" GC_PG_DB="exordos_core" SYSTEMD_SERVICE_DIR=/etc/systemd/system/ +IAM_CACHE_GO_VERSION="1.23.12" DEV_SDK_PATH="/opt/gcl_sdk" SDK_DEV_MODE=$([ -d "$DEV_SDK_PATH" ] && echo "true" || echo "false") @@ -141,12 +142,73 @@ sudo systemctl enable nginx # Install exordos core sudo mkdir -p $GC_CFG_DIR sudo cp "$GC_PATH/etc/exordos_core/logging.yaml" $GC_CFG_DIR/ +sudo install \ + -o root \ + -g root \ + -m 0644 \ + "$GC_PATH/etc/exordos_core/iam_cache.json.example" \ + "$GC_CFG_DIR/iam_cache.json" # Drop-in config dir loaded by ec-user-api via --config-dir. The notification # element lands its [events] override and event_type_mapping.yaml here; must # exist (oslo --config-dir errors on a missing directory). sudo mkdir -p $GC_CFG_DIR/exordos_core.d sudo cp "$GC_PATH/exordos/images/bootstrap.sh" $BOOTSTRAP_PATH/0100-ec-bootstrap.sh +# Build the IAM cache with a temporary Go toolchain. Only the stripped static +# binary is installed into the image; the toolchain and every build cache are +# removed both after a successful build and if the build fails. +case "$(dpkg --print-architecture)" in + amd64) + IAM_CACHE_GO_ARCH="amd64" + IAM_CACHE_GO_SHA256="d3847fef834e9db11bf64e3fb34db9c04db14e068eeb064f49af747010454f90" + ;; + arm64) + IAM_CACHE_GO_ARCH="arm64" + IAM_CACHE_GO_SHA256="52ce172f96e21da53b1ae9079808560d49b02ac86cecfa457217597f9bc28ab3" + ;; + *) + echo "Unsupported architecture for the IAM cache: $(dpkg --print-architecture)" >&2 + exit 1 + ;; +esac + +IAM_CACHE_BUILD_DIR=$(mktemp -d) +cleanup_iam_cache_build() { + if [[ -n "${IAM_CACHE_BUILD_DIR:-}" && -d "$IAM_CACHE_BUILD_DIR" ]]; then + rm -rf -- "$IAM_CACHE_BUILD_DIR" + fi +} +trap cleanup_iam_cache_build EXIT + +curl -fsSLo "$IAM_CACHE_BUILD_DIR/go.tar.gz" \ + "https://go.dev/dl/go${IAM_CACHE_GO_VERSION}.linux-${IAM_CACHE_GO_ARCH}.tar.gz" +echo "$IAM_CACHE_GO_SHA256 $IAM_CACHE_BUILD_DIR/go.tar.gz" \ + | sha256sum --check - +tar -xzf "$IAM_CACHE_BUILD_DIR/go.tar.gz" -C "$IAM_CACHE_BUILD_DIR" + +( + cd "$GC_PATH/services/iam-cache" + CGO_ENABLED=0 \ + GOCACHE="$IAM_CACHE_BUILD_DIR/go-cache" \ + GOPATH="$IAM_CACHE_BUILD_DIR/gopath" \ + "$IAM_CACHE_BUILD_DIR/go/bin/go" build \ + -buildvcs=false \ + -trimpath \ + -ldflags="-s -w" \ + -o "$IAM_CACHE_BUILD_DIR/exordos-iam-cache" \ + ./cmd/exordos-iam-cache +) +sudo install \ + -o root \ + -g root \ + -m 0755 \ + "$IAM_CACHE_BUILD_DIR/exordos-iam-cache" \ + /usr/bin/exordos-iam-cache + +cleanup_iam_cache_build +trap - EXIT +unset IAM_CACHE_BUILD_DIR + cd "$GC_PATH" uv sync source "$GC_PATH"/.venv/bin/activate @@ -206,6 +268,7 @@ sudo cp "$GC_PATH/etc/systemd/ec-core-agent.service" $SYSTEMD_SERVICE_DIR sudo cp "$GC_PATH/etc/systemd/exordos-universal-agent.service" $SYSTEMD_SERVICE_DIR sudo cp "$GC_PATH/etc/systemd/exordos-universal-scheduler.service" $SYSTEMD_SERVICE_DIR sudo cp "$GC_PATH/etc/systemd/exordos-repo-proxy-gservice.service" $SYSTEMD_SERVICE_DIR +sudo cp "$GC_PATH/etc/systemd/exordos-iam-cache.service" $SYSTEMD_SERVICE_DIR # Prepare DNSaaS sudo systemctl disable --now pdns dnsdist@public dnsdist@private diff --git a/exordos/manifests/core.yaml.j2 b/exordos/manifests/core.yaml.j2 index b31d8be9..141a6b49 100644 --- a/exordos/manifests/core.yaml.j2 +++ b/exordos/manifests/core.yaml.j2 @@ -215,6 +215,14 @@ resources: host: 127.0.0.1 port: 11010 weight: 1 + core_lb_iam_cache_backend_http: + project_id: "12345678-c625-4fee-81d5-f691897b8142" + parent: $core.network.lb.$core_lb:uuid + endpoints: + - kind: host + host: 127.0.0.1 + port: 11110 + weight: 1 $core.network.lb.$core_lb.vhosts: core_lb_core_http: project_id: "12345678-c625-4fee-81d5-f691897b8142" @@ -248,6 +256,30 @@ resources: - kind: rewrite_url regex: "^/api/core/(.*)" replacement: "/$1" + core_lb_iam_clients: + project_id: "12345678-c625-4fee-81d5-f691897b8142" + parent: $core.network.lb.$core_lb.vhosts.$core_lb_core_http:uuid + condition: + kind: prefix + value: /api/core/v1/iam/clients/ + allowed_ips: + - 0.0.0.0/0 + actions: + - kind: backend + pool: $core.network.lb.$core_lb.backend_pools.$core_lb_iam_cache_backend_http:uuid + protocol: + kind: http + modifiers: + - kind: auto_header + headers: + - 'Host' + - 'X-Forwarded-For' + - 'X-Forwarded-Port' + - 'X-Forwarded-Proto' + - 'X-Forwarded-Prefix' + - kind: rewrite_url + regex: "^/api/core/(.*)" + replacement: "/$1" core_lb_iam_default_client: project_id: "12345678-c625-4fee-81d5-f691897b8142" parent: $core.network.lb.$core_lb.vhosts.$core_lb_core_http:uuid @@ -258,7 +290,7 @@ resources: - 0.0.0.0/0 actions: - kind: backend - pool: $core.network.lb.$core_lb.backend_pools.$core_lb_core_backend_http:uuid + pool: $core.network.lb.$core_lb.backend_pools.$core_lb_iam_cache_backend_http:uuid protocol: kind: http modifiers: diff --git a/services/iam-cache/README.md b/services/iam-cache/README.md new file mode 100644 index 00000000..2ad68976 --- /dev/null +++ b/services/iam-cache/README.md @@ -0,0 +1,53 @@ +# Exordos IAM Cache + +`exordos-iam-cache` is an in-memory caching proxy for the Exordos Core IAM +introspection and JWKS endpoints. + +The public listener preserves the existing Core routes: + +- `GET /v1/iam/clients/{client_uuid}/actions/introspect` +- `GET /v1/iam/clients/{client_uuid}/actions/jwks` + +All other IAM client requests are forwarded unchanged to Core and are never +cached. Any request carrying `X-OTP`, including a token request, also bypasses +the cache. Successful introspection responses without `X-OTP` are cached by +access token. The token UUID is read from the validated access token's `jti` +claim and is used by the reverse index. + +The internal listener exposes an idempotent invalidation endpoint: + +```text +DELETE /internal/v1/cache/introspection/{token_uuid} +``` + +Core does not call this endpoint in the first implementation. Until that +integration is added, introspection entries expire only by their configured +TTL, the access token expiration, or capacity eviction. + +JWKS responses use a separate cache keyed by IAM client UUID and a separate +TTL. + +## Configuration + +The deployment example is +[`../../etc/exordos_core/iam_cache.json.example`](../../etc/exordos_core/iam_cache.json.example). +Cache lifetimes and the upstream request timeout use Go duration syntax such +as `15s`, `5m`, or `1h`. The deployed defaults are 15 seconds for +introspection and one minute for JWKS. + +The internal listener defaults to loopback. If it is exposed outside the host, +protect it with the deployment's service-to-service authentication layer. + +## Run + +```bash +go run ./cmd/exordos-iam-cache \ + -config ../../etc/exordos_core/iam_cache.json.example +``` + +## Test + +```bash +go test -race ./... +go vet ./... +``` diff --git a/services/iam-cache/cmd/exordos-iam-cache/main.go b/services/iam-cache/cmd/exordos-iam-cache/main.go new file mode 100644 index 00000000..246117bc --- /dev/null +++ b/services/iam-cache/cmd/exordos-iam-cache/main.go @@ -0,0 +1,115 @@ +// Copyright 2026 Genesis Corporation +// +// All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/exordos/exordos_core/services/iam-cache/internal/app" +) + +func main() { + if err := run(); err != nil { + log.Fatal(err) + } +} + +func run() error { + configPath := flag.String( + "config", + "/etc/exordos_core/iam_cache.json", + "path to the JSON configuration file", + ) + flag.Parse() + + config, err := app.LoadConfig(*configPath) + if err != nil { + return err + } + proxy := app.NewProxy(config) + + publicServer := newHTTPServer( + config.PublicListenAddress, + proxy.PublicHandler(), + ) + internalServer := newHTTPServer( + config.InternalListenAddress, + proxy.InternalHandler(), + ) + + runContext, stop := signal.NotifyContext( + context.Background(), + syscall.SIGINT, + syscall.SIGTERM, + ) + defer stop() + + serverErrors := make(chan error, 2) + startServer("public", publicServer, serverErrors) + startServer("internal", internalServer, serverErrors) + + var runErr error + select { + case <-runContext.Done(): + case runErr = <-serverErrors: + stop() + } + + shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + publicErr := publicServer.Shutdown(shutdownContext) + internalErr := internalServer.Shutdown(shutdownContext) + return errors.Join(runErr, publicErr, internalErr) +} + +func newHTTPServer(address string, handler http.Handler) *http.Server { + return &http.Server{ + Addr: address, + Handler: handler, + ReadHeaderTimeout: 5 * time.Second, + IdleTimeout: 60 * time.Second, + } +} + +func startServer( + name string, + server *http.Server, + errorsChannel chan<- error, +) { + go func() { + log.Printf("%s listener started on %s", name, server.Addr) + err := server.ListenAndServe() + if err != nil && !errors.Is(err, http.ErrServerClosed) { + errorsChannel <- fmt.Errorf("%s listener: %w", name, err) + } + }() +} + +func init() { + log.SetOutput(os.Stderr) + log.SetFlags(log.Ldate | log.Ltime | log.LUTC) +} diff --git a/services/iam-cache/go.mod b/services/iam-cache/go.mod new file mode 100644 index 00000000..d5be37d1 --- /dev/null +++ b/services/iam-cache/go.mod @@ -0,0 +1,3 @@ +module github.com/exordos/exordos_core/services/iam-cache + +go 1.23.0 diff --git a/services/iam-cache/internal/app/cache.go b/services/iam-cache/internal/app/cache.go new file mode 100644 index 00000000..3be271c5 --- /dev/null +++ b/services/iam-cache/internal/app/cache.go @@ -0,0 +1,260 @@ +// Copyright 2026 Genesis Corporation +// +// All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package app + +import ( + "container/list" + "crypto/sha256" + "net/http" + "sync" + "time" +) + +type cachedResponse struct { + statusCode int + header http.Header + body []byte +} + +func (response cachedResponse) clone() cachedResponse { + return cachedResponse{ + statusCode: response.statusCode, + header: response.header.Clone(), + body: append([]byte(nil), response.body...), + } +} + +type accessTokenKey [sha256.Size]byte + +func makeAccessTokenKey(accessToken string) accessTokenKey { + return sha256.Sum256([]byte(accessToken)) +} + +type introspectionEntry struct { + key accessTokenKey + tokenUUID string + clientUUID string + response cachedResponse + expiresAt time.Time + element *list.Element +} + +type introspectionCache struct { + mu sync.Mutex + ttl time.Duration + maxEntries int + now func() time.Time + epoch uint64 + items map[accessTokenKey]*introspectionEntry + byTokenUUID map[string]map[accessTokenKey]struct{} + lru list.List +} + +func newIntrospectionCache(ttl time.Duration, maxEntries int) *introspectionCache { + return &introspectionCache{ + ttl: ttl, + maxEntries: maxEntries, + now: time.Now, + items: make(map[accessTokenKey]*introspectionEntry), + byTokenUUID: make(map[string]map[accessTokenKey]struct{}), + } +} + +func (cache *introspectionCache) get( + accessToken string, + clientUUID string, +) (cachedResponse, bool) { + key := makeAccessTokenKey(accessToken) + + cache.mu.Lock() + defer cache.mu.Unlock() + + entry, ok := cache.items[key] + if !ok { + return cachedResponse{}, false + } + if !cache.now().Before(entry.expiresAt) { + cache.removeLocked(entry) + return cachedResponse{}, false + } + if entry.clientUUID != clientUUID { + return cachedResponse{}, false + } + + cache.lru.MoveToFront(entry.element) + return entry.response.clone(), true +} + +func (cache *introspectionCache) currentEpoch() uint64 { + cache.mu.Lock() + defer cache.mu.Unlock() + return cache.epoch +} + +func (cache *introspectionCache) put( + accessToken string, + tokenUUID string, + clientUUID string, + tokenExpiresAt time.Time, + response cachedResponse, + expectedEpoch uint64, +) bool { + key := makeAccessTokenKey(accessToken) + + cache.mu.Lock() + defer cache.mu.Unlock() + + if cache.epoch != expectedEpoch { + return false + } + + expiresAt := cache.now().Add(cache.ttl) + if tokenExpiresAt.Before(expiresAt) { + expiresAt = tokenExpiresAt + } + if !cache.now().Before(expiresAt) { + return false + } + + if existing, ok := cache.items[key]; ok { + cache.removeLocked(existing) + } + + entry := &introspectionEntry{ + key: key, + tokenUUID: tokenUUID, + clientUUID: clientUUID, + response: response.clone(), + expiresAt: expiresAt, + } + entry.element = cache.lru.PushFront(entry) + cache.items[key] = entry + + tokenEntries := cache.byTokenUUID[tokenUUID] + if tokenEntries == nil { + tokenEntries = make(map[accessTokenKey]struct{}) + cache.byTokenUUID[tokenUUID] = tokenEntries + } + tokenEntries[key] = struct{}{} + + for len(cache.items) > cache.maxEntries { + oldest := cache.lru.Back() + if oldest == nil { + break + } + cache.removeLocked(oldest.Value.(*introspectionEntry)) + } + return true +} + +func (cache *introspectionCache) invalidate(tokenUUID string) int { + cache.mu.Lock() + defer cache.mu.Unlock() + + cache.epoch++ + keys := cache.byTokenUUID[tokenUUID] + evicted := len(keys) + for key := range keys { + if entry, ok := cache.items[key]; ok { + cache.removeLocked(entry) + } + } + return evicted +} + +func (cache *introspectionCache) removeLocked(entry *introspectionEntry) { + delete(cache.items, entry.key) + cache.lru.Remove(entry.element) + + tokenEntries := cache.byTokenUUID[entry.tokenUUID] + delete(tokenEntries, entry.key) + if len(tokenEntries) == 0 { + delete(cache.byTokenUUID, entry.tokenUUID) + } +} + +type jwksEntry struct { + clientUUID string + response cachedResponse + expiresAt time.Time + element *list.Element +} + +type jwksCache struct { + mu sync.Mutex + ttl time.Duration + maxEntries int + now func() time.Time + items map[string]*jwksEntry + lru list.List +} + +func newJWKSCache(ttl time.Duration, maxEntries int) *jwksCache { + return &jwksCache{ + ttl: ttl, + maxEntries: maxEntries, + now: time.Now, + items: make(map[string]*jwksEntry), + } +} + +func (cache *jwksCache) get(clientUUID string) (cachedResponse, bool) { + cache.mu.Lock() + defer cache.mu.Unlock() + + entry, ok := cache.items[clientUUID] + if !ok { + return cachedResponse{}, false + } + if !cache.now().Before(entry.expiresAt) { + cache.removeLocked(entry) + return cachedResponse{}, false + } + + cache.lru.MoveToFront(entry.element) + return entry.response.clone(), true +} + +func (cache *jwksCache) put(clientUUID string, response cachedResponse) { + cache.mu.Lock() + defer cache.mu.Unlock() + + if existing, ok := cache.items[clientUUID]; ok { + cache.removeLocked(existing) + } + + entry := &jwksEntry{ + clientUUID: clientUUID, + response: response.clone(), + expiresAt: cache.now().Add(cache.ttl), + } + entry.element = cache.lru.PushFront(entry) + cache.items[clientUUID] = entry + + for len(cache.items) > cache.maxEntries { + oldest := cache.lru.Back() + if oldest == nil { + break + } + cache.removeLocked(oldest.Value.(*jwksEntry)) + } +} + +func (cache *jwksCache) removeLocked(entry *jwksEntry) { + delete(cache.items, entry.clientUUID) + cache.lru.Remove(entry.element) +} diff --git a/services/iam-cache/internal/app/cache_test.go b/services/iam-cache/internal/app/cache_test.go new file mode 100644 index 00000000..7c02489f --- /dev/null +++ b/services/iam-cache/internal/app/cache_test.go @@ -0,0 +1,188 @@ +// Copyright 2026 Genesis Corporation +// +// All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package app + +import ( + "net/http" + "testing" + "time" +) + +func TestIntrospectionCacheUsesBothIndexes(t *testing.T) { + t.Parallel() + + cache := newIntrospectionCache(time.Minute, 10) + now := time.Date(2026, 7, 26, 12, 0, 0, 0, time.UTC) + cache.now = func() time.Time { return now } + response := testCachedResponse(`{"permissions":["read"]}`) + + epoch := cache.currentEpoch() + for _, token := range []string{"access-one", "access-two"} { + if !cache.put( + token, + "6ba7b810-9dad-11d1-80b4-00c04fd430c8", + "client-one", + now.Add(time.Hour), + response, + epoch, + ) { + t.Fatalf("put(%q) returned false", token) + } + } + + if _, ok := cache.get("access-one", "client-one"); !ok { + t.Fatal("first access token was not found") + } + if _, ok := cache.get("access-two", "client-one"); !ok { + t.Fatal("second access token was not found") + } + if _, ok := cache.get("access-one", "client-two"); ok { + t.Fatal("entry was returned for a different IAM client") + } + + if got, want := cache.invalidate( + "6ba7b810-9dad-11d1-80b4-00c04fd430c8", + ), 2; got != want { + t.Fatalf("invalidate() = %d, want %d", got, want) + } + if _, ok := cache.get("access-one", "client-one"); ok { + t.Fatal("first access token survived invalidation") + } + if _, ok := cache.get("access-two", "client-one"); ok { + t.Fatal("second access token survived invalidation") + } +} + +func TestIntrospectionCacheHonorsTTLAndTokenExpiration(t *testing.T) { + t.Parallel() + + now := time.Date(2026, 7, 26, 12, 0, 0, 0, time.UTC) + response := testCachedResponse(`{"permissions":[]}`) + + tests := map[string]struct { + cacheTTL time.Duration + tokenExpiration time.Time + advance time.Duration + }{ + "configured TTL": { + cacheTTL: 30 * time.Second, + tokenExpiration: now.Add(time.Hour), + advance: 31 * time.Second, + }, + "token expiration": { + cacheTTL: time.Hour, + tokenExpiration: now.Add(10 * time.Second), + advance: 11 * time.Second, + }, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + cache := newIntrospectionCache(test.cacheTTL, 10) + currentTime := now + cache.now = func() time.Time { return currentTime } + + if !cache.put( + "access", + "6ba7b810-9dad-11d1-80b4-00c04fd430c8", + "client", + test.tokenExpiration, + response, + cache.currentEpoch(), + ) { + t.Fatal("put() returned false") + } + currentTime = currentTime.Add(test.advance) + if _, ok := cache.get("access", "client"); ok { + t.Fatal("expired entry was returned") + } + }) + } +} + +func TestIntrospectionInvalidationFencesInFlightStore(t *testing.T) { + t.Parallel() + + cache := newIntrospectionCache(time.Minute, 10) + now := time.Date(2026, 7, 26, 12, 0, 0, 0, time.UTC) + cache.now = func() time.Time { return now } + + epoch := cache.currentEpoch() + cache.invalidate("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + if cache.put( + "access", + "6ba7b810-9dad-11d1-80b4-00c04fd430c8", + "client", + now.Add(time.Hour), + testCachedResponse(`{"permissions":[]}`), + epoch, + ) { + t.Fatal("stale in-flight result was stored after invalidation") + } +} + +func TestCachesEvictLeastRecentlyUsedEntries(t *testing.T) { + t.Parallel() + + now := time.Date(2026, 7, 26, 12, 0, 0, 0, time.UTC) + introspection := newIntrospectionCache(time.Hour, 2) + introspection.now = func() time.Time { return now } + epoch := introspection.currentEpoch() + for _, token := range []string{"one", "two"} { + introspection.put( + token, + token+"-uuid", + "client", + now.Add(time.Hour), + testCachedResponse(token), + epoch, + ) + } + if _, ok := introspection.get("one", "client"); !ok { + t.Fatal("recently used entry not found") + } + introspection.put( + "three", + "three-uuid", + "client", + now.Add(time.Hour), + testCachedResponse("three"), + epoch, + ) + if _, ok := introspection.get("two", "client"); ok { + t.Fatal("least recently used introspection entry was not evicted") + } + + jwks := newJWKSCache(time.Hour, 1) + jwks.now = func() time.Time { return now } + jwks.put("client-one", testCachedResponse("one")) + jwks.put("client-two", testCachedResponse("two")) + if _, ok := jwks.get("client-one"); ok { + t.Fatal("least recently used JWKS entry was not evicted") + } + if _, ok := jwks.get("client-two"); !ok { + t.Fatal("newest JWKS entry not found") + } +} + +func testCachedResponse(body string) cachedResponse { + return cachedResponse{ + statusCode: http.StatusOK, + header: http.Header{"Content-Type": []string{"application/json"}}, + body: []byte(body), + } +} diff --git a/services/iam-cache/internal/app/config.go b/services/iam-cache/internal/app/config.go new file mode 100644 index 00000000..26a3de5a --- /dev/null +++ b/services/iam-cache/internal/app/config.go @@ -0,0 +1,182 @@ +// Copyright 2026 Genesis Corporation +// +// All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package app + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/url" + "os" + "time" +) + +const ( + defaultPublicListenAddress = "127.0.0.1:11110" + defaultInternalListenAddress = "127.0.0.1:11111" + defaultRequestTimeout = 5 * time.Second + defaultIntrospectionCacheTTL = 15 * time.Second + defaultIntrospectionCacheEntries = 100000 + defaultJWKSCacheTTL = time.Minute + defaultJWKSCacheEntries = 1000 +) + +type fileConfig struct { + PublicListenAddress string `json:"public_listen_address"` + InternalListenAddress string `json:"internal_listen_address"` + CoreURL string `json:"core_url"` + RequestTimeout string `json:"request_timeout"` + IntrospectionCacheTTL string `json:"introspection_cache_ttl"` + IntrospectionCacheMaxEntries int `json:"introspection_cache_max_entries"` + JWKSCacheTTL string `json:"jwks_cache_ttl"` + JWKSCacheMaxEntries int `json:"jwks_cache_max_entries"` +} + +// Config contains validated runtime configuration. +type Config struct { + PublicListenAddress string + InternalListenAddress string + CoreURL *url.URL + RequestTimeout time.Duration + IntrospectionCacheTTL time.Duration + IntrospectionCacheMaxEntries int + JWKSCacheTTL time.Duration + JWKSCacheMaxEntries int +} + +// LoadConfig reads and validates a JSON configuration file. +func LoadConfig(path string) (Config, error) { + file, err := os.Open(path) + if err != nil { + return Config{}, fmt.Errorf("open config: %w", err) + } + defer file.Close() + + decoder := json.NewDecoder(file) + decoder.DisallowUnknownFields() + + var raw fileConfig + if err := decoder.Decode(&raw); err != nil { + return Config{}, fmt.Errorf("decode config: %w", err) + } + if err := ensureSingleJSONValue(decoder); err != nil { + return Config{}, err + } + + return parseConfig(raw) +} + +func ensureSingleJSONValue(decoder *json.Decoder) error { + var extra any + if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) { + if err == nil { + return errors.New("decode config: multiple JSON values") + } + return fmt.Errorf("decode config: %w", err) + } + return nil +} + +func parseConfig(raw fileConfig) (Config, error) { + if raw.PublicListenAddress == "" { + raw.PublicListenAddress = defaultPublicListenAddress + } + if raw.InternalListenAddress == "" { + raw.InternalListenAddress = defaultInternalListenAddress + } + if raw.PublicListenAddress == raw.InternalListenAddress { + return Config{}, errors.New("public and internal listen addresses must differ") + } + + coreURL, err := url.Parse(raw.CoreURL) + if err != nil { + return Config{}, fmt.Errorf("parse core_url: %w", err) + } + if coreURL.Scheme != "http" && coreURL.Scheme != "https" { + return Config{}, errors.New("core_url must use http or https") + } + if coreURL.Host == "" { + return Config{}, errors.New("core_url must include a host") + } + if coreURL.User != nil || coreURL.RawQuery != "" || coreURL.Fragment != "" { + return Config{}, errors.New("core_url must not include credentials, query, or fragment") + } + + requestTimeout, err := parseDuration( + "request_timeout", + raw.RequestTimeout, + defaultRequestTimeout, + ) + if err != nil { + return Config{}, err + } + introspectionTTL, err := parseDuration( + "introspection_cache_ttl", + raw.IntrospectionCacheTTL, + defaultIntrospectionCacheTTL, + ) + if err != nil { + return Config{}, err + } + jwksTTL, err := parseDuration( + "jwks_cache_ttl", + raw.JWKSCacheTTL, + defaultJWKSCacheTTL, + ) + if err != nil { + return Config{}, err + } + + if raw.IntrospectionCacheMaxEntries == 0 { + raw.IntrospectionCacheMaxEntries = defaultIntrospectionCacheEntries + } + if raw.IntrospectionCacheMaxEntries < 0 { + return Config{}, errors.New("introspection_cache_max_entries must be positive") + } + if raw.JWKSCacheMaxEntries == 0 { + raw.JWKSCacheMaxEntries = defaultJWKSCacheEntries + } + if raw.JWKSCacheMaxEntries < 0 { + return Config{}, errors.New("jwks_cache_max_entries must be positive") + } + + return Config{ + PublicListenAddress: raw.PublicListenAddress, + InternalListenAddress: raw.InternalListenAddress, + CoreURL: coreURL, + RequestTimeout: requestTimeout, + IntrospectionCacheTTL: introspectionTTL, + IntrospectionCacheMaxEntries: raw.IntrospectionCacheMaxEntries, + JWKSCacheTTL: jwksTTL, + JWKSCacheMaxEntries: raw.JWKSCacheMaxEntries, + }, nil +} + +func parseDuration(name, value string, fallback time.Duration) (time.Duration, error) { + if value == "" { + return fallback, nil + } + duration, err := time.ParseDuration(value) + if err != nil { + return 0, fmt.Errorf("parse %s: %w", name, err) + } + if duration <= 0 { + return 0, fmt.Errorf("%s must be positive", name) + } + return duration, nil +} diff --git a/services/iam-cache/internal/app/config_test.go b/services/iam-cache/internal/app/config_test.go new file mode 100644 index 00000000..27cf8dfe --- /dev/null +++ b/services/iam-cache/internal/app/config_test.go @@ -0,0 +1,131 @@ +// Copyright 2026 Genesis Corporation +// +// All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package app + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestLoadConfigAppliesDefaults(t *testing.T) { + t.Parallel() + + path := writeTestConfig(t, `{"core_url":"https://core.example/api/core"}`) + config, err := LoadConfig(path) + if err != nil { + t.Fatalf("LoadConfig returned an error: %v", err) + } + + if got, want := config.PublicListenAddress, defaultPublicListenAddress; got != want { + t.Errorf("PublicListenAddress = %q, want %q", got, want) + } + if got, want := config.InternalListenAddress, defaultInternalListenAddress; got != want { + t.Errorf("InternalListenAddress = %q, want %q", got, want) + } + if got, want := config.RequestTimeout, defaultRequestTimeout; got != want { + t.Errorf("RequestTimeout = %s, want %s", got, want) + } + if got, want := config.IntrospectionCacheTTL, defaultIntrospectionCacheTTL; got != want { + t.Errorf("IntrospectionCacheTTL = %s, want %s", got, want) + } + if got, want := config.IntrospectionCacheTTL, 15*time.Second; got != want { + t.Errorf("IntrospectionCacheTTL = %s, want deployed default %s", got, want) + } + if got, want := config.JWKSCacheTTL, defaultJWKSCacheTTL; got != want { + t.Errorf("JWKSCacheTTL = %s, want %s", got, want) + } + if got, want := config.JWKSCacheTTL, time.Minute; got != want { + t.Errorf("JWKSCacheTTL = %s, want deployed default %s", got, want) + } +} + +func TestLoadConfigReadsIndependentCacheTTLs(t *testing.T) { + t.Parallel() + + path := writeTestConfig(t, `{ + "core_url":"http://core.example:8080/api/core", + "request_timeout":"2s", + "introspection_cache_ttl":"17s", + "jwks_cache_ttl":"23m" + }`) + config, err := LoadConfig(path) + if err != nil { + t.Fatalf("LoadConfig returned an error: %v", err) + } + + if got, want := config.RequestTimeout, 2*time.Second; got != want { + t.Errorf("RequestTimeout = %s, want %s", got, want) + } + if got, want := config.IntrospectionCacheTTL, 17*time.Second; got != want { + t.Errorf("IntrospectionCacheTTL = %s, want %s", got, want) + } + if got, want := config.JWKSCacheTTL, 23*time.Minute; got != want { + t.Errorf("JWKSCacheTTL = %s, want %s", got, want) + } +} + +func TestLoadConfigRejectsUnknownFields(t *testing.T) { + t.Parallel() + + path := writeTestConfig(t, `{ + "core_url":"https://core.example", + "introspection_ttl":"15s" + }`) + _, err := LoadConfig(path) + if err == nil || !strings.Contains(err.Error(), "unknown field") { + t.Fatalf("LoadConfig error = %v, want unknown field error", err) + } +} + +func TestLoadConfigRejectsInvalidValues(t *testing.T) { + t.Parallel() + + tests := map[string]string{ + "missing Core URL": `{}`, + "unsupported Core scheme": `{"core_url":"ftp://core.example"}`, + "non-positive cache size": `{"core_url":"https://core.example","jwks_cache_max_entries":-1}`, + "non-positive cache lifetime": `{"core_url":"https://core.example","jwks_cache_ttl":"0s"}`, + "shared listener": `{ + "core_url":"https://core.example", + "public_listen_address":":8080", + "internal_listen_address":":8080" + }`, + } + + for name, contents := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + path := writeTestConfig(t, contents) + if _, err := LoadConfig(path); err == nil { + t.Fatal("LoadConfig returned no error") + } + }) + } +} + +func writeTestConfig(t *testing.T, contents string) string { + t.Helper() + + path := filepath.Join(t.TempDir(), "config.json") + if err := os.WriteFile(path, []byte(contents), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + return path +} diff --git a/services/iam-cache/internal/app/flight.go b/services/iam-cache/internal/app/flight.go new file mode 100644 index 00000000..105a5cc4 --- /dev/null +++ b/services/iam-cache/internal/app/flight.go @@ -0,0 +1,67 @@ +// Copyright 2026 Genesis Corporation +// +// All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package app + +import ( + "context" + "sync" +) + +type flightCall struct { + done chan struct{} + response cachedResponse + err error +} + +type flightGroup struct { + mu sync.Mutex + calls map[string]*flightCall +} + +func newFlightGroup() *flightGroup { + return &flightGroup{calls: make(map[string]*flightCall)} +} + +func (group *flightGroup) do( + ctx context.Context, + key string, + call func() (cachedResponse, error), +) (cachedResponse, error) { + group.mu.Lock() + if running, ok := group.calls[key]; ok { + group.mu.Unlock() + select { + case <-running.done: + return running.response.clone(), running.err + case <-ctx.Done(): + return cachedResponse{}, ctx.Err() + } + } + + running := &flightCall{done: make(chan struct{})} + group.calls[key] = running + group.mu.Unlock() + + running.response, running.err = call() + close(running.done) + + group.mu.Lock() + delete(group.calls, key) + group.mu.Unlock() + + return running.response.clone(), running.err +} diff --git a/services/iam-cache/internal/app/server.go b/services/iam-cache/internal/app/server.go new file mode 100644 index 00000000..3bd2e98b --- /dev/null +++ b/services/iam-cache/internal/app/server.go @@ -0,0 +1,440 @@ +// Copyright 2026 Genesis Corporation +// +// All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package app + +import ( + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +const ( + iamRoutePrefix = "/v1/iam/clients/" + invalidationPrefix = "/internal/v1/cache/introspection/" + maxUpstreamBodyBytes = 4 << 20 +) + +var hopByHopHeaders = map[string]struct{}{ + "Connection": {}, + "Keep-Alive": {}, + "Proxy-Authenticate": {}, + "Proxy-Authorization": {}, + "Te": {}, + "Trailer": {}, + "Transfer-Encoding": {}, + "Upgrade": {}, +} + +// Proxy serves the public caching API and the internal invalidation API. +type Proxy struct { + coreURL *url.URL + client *http.Client + introspectionCache *introspectionCache + jwksCache *jwksCache + introspectionCalls *flightGroup + jwksCalls *flightGroup +} + +// NewProxy constructs a proxy from validated configuration. +func NewProxy(config Config) *Proxy { + return &Proxy{ + coreURL: config.CoreURL, + client: &http.Client{ + Timeout: config.RequestTimeout, + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { + return http.ErrUseLastResponse + }, + }, + introspectionCache: newIntrospectionCache( + config.IntrospectionCacheTTL, + config.IntrospectionCacheMaxEntries, + ), + jwksCache: newJWKSCache( + config.JWKSCacheTTL, + config.JWKSCacheMaxEntries, + ), + introspectionCalls: newFlightGroup(), + jwksCalls: newFlightGroup(), + } +} + +// PublicHandler returns the handler for consumers of the cached IAM API. +func (proxy *Proxy) PublicHandler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/health/live", healthHandler) + mux.HandleFunc("/health/ready", healthHandler) + mux.HandleFunc(iamRoutePrefix, proxy.handleIAM) + return mux +} + +// InternalHandler returns the handler intended only for Core. +func (proxy *Proxy) InternalHandler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/health/live", healthHandler) + mux.HandleFunc(invalidationPrefix, proxy.handleInvalidation) + return mux +} + +func healthHandler(response http.ResponseWriter, request *http.Request) { + if request.Method != http.MethodGet { + response.Header().Set("Allow", http.MethodGet) + http.Error(response, "method not allowed", http.StatusMethodNotAllowed) + return + } + response.Header().Set("Content-Type", "application/json") + response.WriteHeader(http.StatusOK) + _, _ = response.Write([]byte(`{"status":"ok"}`)) +} + +func (proxy *Proxy) handleIAM(response http.ResponseWriter, request *http.Request) { + clientUUID, action, cacheableRoute := parseIAMRoute(request.URL.Path) + if request.Method != http.MethodGet || + !cacheableRoute || + hasHeader(request.Header, "X-OTP") { + proxy.forward(response, request) + return + } + + switch action { + case "introspect": + proxy.handleIntrospection(response, request, clientUUID) + case "jwks": + proxy.handleJWKS(response, request, clientUUID) + } +} + +func parseIAMRoute(path string) (string, string, bool) { + if !strings.HasPrefix(path, iamRoutePrefix) { + return "", "", false + } + parts := strings.Split(strings.TrimPrefix(path, iamRoutePrefix), "/") + if len(parts) != 3 || parts[0] == "" || parts[1] != "actions" { + return "", "", false + } + if parts[2] != "introspect" && parts[2] != "jwks" { + return "", "", false + } + return parts[0], parts[2], true +} + +func (proxy *Proxy) handleIntrospection( + writer http.ResponseWriter, + request *http.Request, + clientUUID string, +) { + accessToken, hasBearer := bearerToken(request.Header.Get("Authorization")) + if !hasBearer { + proxy.forward(writer, request) + return + } + + if response, ok := proxy.introspectionCache.get(accessToken, clientUUID); ok { + writeCachedResponse(writer, response) + return + } + + claims, claimsOK := parseAccessTokenClaims(accessToken) + flightKey := "introspection:" + clientUUID + ":" + tokenKeyString(accessToken) + upstreamResponse, err := proxy.introspectionCalls.do( + request.Context(), + flightKey, + func() (cachedResponse, error) { + if response, ok := proxy.introspectionCache.get( + accessToken, + clientUUID, + ); ok { + return response, nil + } + + epoch := proxy.introspectionCache.currentEpoch() + response, err := proxy.fetchUpstream(request) + if err != nil { + return cachedResponse{}, err + } + if response.statusCode == http.StatusOK && claimsOK { + proxy.introspectionCache.put( + accessToken, + claims.TokenUUID, + clientUUID, + claims.ExpiresAt, + response, + epoch, + ) + } + return response, nil + }, + ) + if err != nil { + writeUpstreamError(writer, err) + return + } + writeCachedResponse(writer, upstreamResponse) +} + +func (proxy *Proxy) handleJWKS( + writer http.ResponseWriter, + request *http.Request, + clientUUID string, +) { + if response, ok := proxy.jwksCache.get(clientUUID); ok { + writeCachedResponse(writer, response) + return + } + + upstreamResponse, err := proxy.jwksCalls.do( + request.Context(), + "jwks:"+clientUUID, + func() (cachedResponse, error) { + if response, ok := proxy.jwksCache.get(clientUUID); ok { + return response, nil + } + + response, err := proxy.fetchUpstream(request) + if err != nil { + return cachedResponse{}, err + } + if response.statusCode == http.StatusOK { + proxy.jwksCache.put(clientUUID, response) + } + return response, nil + }, + ) + if err != nil { + writeUpstreamError(writer, err) + return + } + writeCachedResponse(writer, upstreamResponse) +} + +func (proxy *Proxy) forward(writer http.ResponseWriter, request *http.Request) { + upstreamRequest, err := proxy.newUpstreamRequest(request, request.Body) + if err != nil { + writeUpstreamError(writer, err) + return + } + + upstreamResponse, err := proxy.client.Do(upstreamRequest) + if err != nil { + writeUpstreamError(writer, fmt.Errorf("request Core: %w", err)) + return + } + defer upstreamResponse.Body.Close() + + copyEndToEndHeaders(writer.Header(), upstreamResponse.Header) + writer.WriteHeader(upstreamResponse.StatusCode) + if _, err := io.Copy(writer, upstreamResponse.Body); err != nil { + log.Printf("stream IAM upstream response: %v", err) + } +} + +func (proxy *Proxy) fetchUpstream(request *http.Request) (cachedResponse, error) { + upstreamRequest, err := proxy.newUpstreamRequest(request, nil) + if err != nil { + return cachedResponse{}, err + } + + upstreamResponse, err := proxy.client.Do(upstreamRequest) + if err != nil { + return cachedResponse{}, fmt.Errorf("request Core: %w", err) + } + defer upstreamResponse.Body.Close() + + body, err := io.ReadAll(io.LimitReader( + upstreamResponse.Body, + maxUpstreamBodyBytes+1, + )) + if err != nil { + return cachedResponse{}, fmt.Errorf("read Core response: %w", err) + } + if len(body) > maxUpstreamBodyBytes { + return cachedResponse{}, errors.New("Core response exceeds size limit") + } + + return cachedResponse{ + statusCode: upstreamResponse.StatusCode, + header: sanitizedHeaders(upstreamResponse.Header), + body: body, + }, nil +} + +func (proxy *Proxy) newUpstreamRequest( + request *http.Request, + body io.Reader, +) (*http.Request, error) { + upstreamURL := strings.TrimRight(proxy.coreURL.String(), "/") + + request.URL.EscapedPath() + if request.URL.RawQuery != "" { + upstreamURL += "?" + request.URL.RawQuery + } + + upstreamRequest, err := http.NewRequestWithContext( + request.Context(), + request.Method, + upstreamURL, + body, + ) + if err != nil { + return nil, fmt.Errorf("build upstream request: %w", err) + } + copyEndToEndHeaders(upstreamRequest.Header, request.Header) + upstreamRequest.Host = request.Host + if body != nil { + upstreamRequest.ContentLength = request.ContentLength + upstreamRequest.TransferEncoding = append( + []string(nil), + request.TransferEncoding..., + ) + } + + return upstreamRequest, nil +} + +func copyEndToEndHeaders(destination, source http.Header) { + for name, values := range source { + if _, skip := hopByHopHeaders[http.CanonicalHeaderKey(name)]; skip { + continue + } + for _, value := range values { + destination.Add(name, value) + } + } +} + +func sanitizedHeaders(source http.Header) http.Header { + result := make(http.Header) + copyEndToEndHeaders(result, source) + result.Del("Content-Length") + return result +} + +func writeCachedResponse(writer http.ResponseWriter, response cachedResponse) { + copyEndToEndHeaders(writer.Header(), response.header) + writer.WriteHeader(response.statusCode) + _, _ = writer.Write(response.body) +} + +func writeUpstreamError(writer http.ResponseWriter, err error) { + log.Printf("IAM upstream request failed: %v", err) + http.Error(writer, "IAM upstream unavailable", http.StatusBadGateway) +} + +func bearerToken(value string) (string, bool) { + parts := strings.Fields(value) + if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") { + return "", false + } + return parts[1], true +} + +func hasHeader(header http.Header, name string) bool { + for key := range header { + if strings.EqualFold(key, name) { + return true + } + } + return false +} + +type accessTokenClaims struct { + TokenUUID string + ExpiresAt time.Time +} + +func parseAccessTokenClaims(accessToken string) (accessTokenClaims, bool) { + parts := strings.Split(accessToken, ".") + if len(parts) != 3 { + return accessTokenClaims{}, false + } + + payload, err := base64.RawURLEncoding.DecodeString(parts[1]) + if err != nil { + return accessTokenClaims{}, false + } + var claims struct { + JTI string `json:"jti"` + Exp json.RawMessage `json:"exp"` + } + if err := json.Unmarshal(payload, &claims); err != nil || claims.JTI == "" { + return accessTokenClaims{}, false + } + + expiration, err := strconv.ParseInt(string(claims.Exp), 10, 64) + if err != nil { + return accessTokenClaims{}, false + } + return accessTokenClaims{ + TokenUUID: claims.JTI, + ExpiresAt: time.Unix(expiration, 0), + }, true +} + +func tokenKeyString(accessToken string) string { + key := makeAccessTokenKey(accessToken) + return hex.EncodeToString(key[:]) +} + +func (proxy *Proxy) handleInvalidation( + writer http.ResponseWriter, + request *http.Request, +) { + if request.Method != http.MethodDelete { + writer.Header().Set("Allow", http.MethodDelete) + http.Error(writer, "method not allowed", http.StatusMethodNotAllowed) + return + } + + tokenUUID := strings.TrimPrefix(request.URL.Path, invalidationPrefix) + if strings.Contains(tokenUUID, "/") { + http.NotFound(writer, request) + return + } + if !validUUID(tokenUUID) { + http.Error(writer, "invalid token UUID", http.StatusBadRequest) + return + } + + proxy.introspectionCache.invalidate(tokenUUID) + writer.WriteHeader(http.StatusNoContent) +} + +func validUUID(value string) bool { + if len(value) != 36 { + return false + } + for index, char := range value { + switch index { + case 8, 13, 18, 23: + if char != '-' { + return false + } + default: + if !strings.ContainsRune("0123456789abcdefABCDEF", char) { + return false + } + } + } + return true +} diff --git a/services/iam-cache/internal/app/server_test.go b/services/iam-cache/internal/app/server_test.go new file mode 100644 index 00000000..984e68ab --- /dev/null +++ b/services/iam-cache/internal/app/server_test.go @@ -0,0 +1,544 @@ +// Copyright 2026 Genesis Corporation +// +// All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +// License for the specific language governing permissions and limitations +// under the License. + +package app + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +const testTokenUUID = "6ba7b810-9dad-11d1-80b4-00c04fd430c8" + +func TestIntrospectionCachesSuccessfulResponse(t *testing.T) { + t.Parallel() + + token := testAccessToken(testTokenUUID, time.Now().Add(time.Hour)) + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func( + writer http.ResponseWriter, + request *http.Request, + ) { + calls.Add(1) + if got, want := request.URL.Path, "/api/core/v1/iam/clients/client-one/actions/introspect"; got != want { + t.Errorf("upstream path = %q, want %q", got, want) + } + if got, want := request.Header.Get("Authorization"), "Bearer "+token; got != want { + t.Errorf("Authorization header = %q, want %q", got, want) + } + writer.Header().Set("Content-Type", "application/json") + _, _ = writer.Write([]byte(`{"permissions":["read"]}`)) + })) + defer upstream.Close() + + proxy := newTestProxy(t, upstream.URL+"/api/core", time.Minute, time.Minute) + server := httptest.NewServer(proxy.PublicHandler()) + defer server.Close() + + for range 2 { + response := makePublicRequest( + t, + server.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + "", + ) + assertResponse(t, response, http.StatusOK, `{"permissions":["read"]}`) + } + + if got, want := calls.Load(), int32(1); got != want { + t.Fatalf("upstream calls = %d, want %d", got, want) + } +} + +func TestXOTPAlwaysBypassesIntrospectionCache(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + var receivedOTP []string + var mu sync.Mutex + upstream := httptest.NewServer(http.HandlerFunc(func( + writer http.ResponseWriter, + request *http.Request, + ) { + call := calls.Add(1) + mu.Lock() + receivedOTP = append(receivedOTP, request.Header.Get("X-OTP")) + mu.Unlock() + writer.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(writer, `{"call":%d}`, call) + })) + defer upstream.Close() + + proxy := newTestProxy(t, upstream.URL, time.Minute, time.Minute) + server := httptest.NewServer(proxy.PublicHandler()) + defer server.Close() + + token := testAccessToken(testTokenUUID, time.Now().Add(time.Hour)) + for _, otp := range []string{"111111", "222222"} { + response := makePublicRequest( + t, + server.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + otp, + ) + assertResponse(t, response, http.StatusOK, "") + } + response := makePublicRequest( + t, + server.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + "", + ) + assertResponse(t, response, http.StatusOK, `{"call":3}`) + response = makePublicRequest( + t, + server.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + "", + ) + assertResponse(t, response, http.StatusOK, `{"call":3}`) + + if got, want := calls.Load(), int32(3); got != want { + t.Fatalf("upstream calls = %d, want %d", got, want) + } + mu.Lock() + defer mu.Unlock() + if got, want := strings.Join(receivedOTP, ","), "111111,222222,"; got != want { + t.Fatalf("received OTP values = %q, want %q", got, want) + } +} + +func TestXOTPAlwaysBypassesJWKSCache(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func( + writer http.ResponseWriter, + request *http.Request, + ) { + call := calls.Add(1) + if got, want := request.Header.Get("X-OTP"), "123456"; got != want { + t.Errorf("X-OTP = %q, want %q", got, want) + } + _, _ = fmt.Fprintf(writer, `{"call":%d}`, call) + })) + defer upstream.Close() + + proxy := newTestProxy(t, upstream.URL, time.Hour, time.Hour) + server := httptest.NewServer(proxy.PublicHandler()) + defer server.Close() + + endpoint := server.URL + "/v1/iam/clients/client-one/actions/jwks" + for call := 1; call <= 2; call++ { + request, err := http.NewRequest(http.MethodGet, endpoint, nil) + if err != nil { + t.Fatalf("create JWKS request: %v", err) + } + request.Header.Set("X-OTP", "123456") + + response, err := http.DefaultClient.Do(request) + if err != nil { + t.Fatalf("perform JWKS request: %v", err) + } + assertResponse( + t, + response, + http.StatusOK, + fmt.Sprintf(`{"call":%d}`, call), + ) + } + + if got, want := calls.Load(), int32(2); got != want { + t.Fatalf("upstream calls = %d, want %d", got, want) + } +} + +func TestTokenRequestPassesThroughWithoutCaching(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func( + writer http.ResponseWriter, + request *http.Request, + ) { + call := calls.Add(1) + if got, want := request.Method, http.MethodPost; got != want { + t.Errorf("method = %q, want %q", got, want) + } + if got, want := request.URL.Path, "/api/core/v1/iam/clients/client-one/actions/get_token/invoke"; got != want { + t.Errorf("upstream path = %q, want %q", got, want) + } + if got, want := request.URL.RawQuery, "source=test"; got != want { + t.Errorf("query = %q, want %q", got, want) + } + if got, want := request.Header.Get("X-OTP"), "123456"; got != want { + t.Errorf("X-OTP = %q, want %q", got, want) + } + body, err := io.ReadAll(request.Body) + if err != nil { + t.Errorf("read request body: %v", err) + } + if got, want := string(body), "grant_type=password"; got != want { + t.Errorf("body = %q, want %q", got, want) + } + writer.Header().Set("X-Upstream-Call", fmt.Sprint(call)) + writer.WriteHeader(http.StatusCreated) + _, _ = writer.Write([]byte(`{"access_token":"token"}`)) + })) + defer upstream.Close() + + proxy := newTestProxy(t, upstream.URL+"/api/core", time.Hour, time.Hour) + server := httptest.NewServer(proxy.PublicHandler()) + defer server.Close() + + endpoint := server.URL + + "/v1/iam/clients/client-one/actions/get_token/invoke?source=test" + for range 2 { + request, err := http.NewRequest( + http.MethodPost, + endpoint, + strings.NewReader("grant_type=password"), + ) + if err != nil { + t.Fatalf("create token request: %v", err) + } + request.Header.Set("Content-Type", "application/x-www-form-urlencoded") + request.Header.Set("X-OTP", "123456") + + response, err := http.DefaultClient.Do(request) + if err != nil { + t.Fatalf("perform token request: %v", err) + } + assertResponse( + t, + response, + http.StatusCreated, + `{"access_token":"token"}`, + ) + } + + if got, want := calls.Load(), int32(2); got != want { + t.Fatalf("upstream calls = %d, want %d", got, want) + } +} + +func TestIntrospectionCacheExpires(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func( + writer http.ResponseWriter, + _ *http.Request, + ) { + call := calls.Add(1) + _, _ = fmt.Fprintf(writer, `{"call":%d}`, call) + })) + defer upstream.Close() + + proxy := newTestProxy(t, upstream.URL, 20*time.Millisecond, time.Minute) + server := httptest.NewServer(proxy.PublicHandler()) + defer server.Close() + + token := testAccessToken(testTokenUUID, time.Now().Add(time.Hour)) + first := makePublicRequest( + t, + server.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + "", + ) + assertResponse(t, first, http.StatusOK, `{"call":1}`) + time.Sleep(30 * time.Millisecond) + second := makePublicRequest( + t, + server.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + "", + ) + assertResponse(t, second, http.StatusOK, `{"call":2}`) +} + +func TestJWKSUsesIndependentCache(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func( + writer http.ResponseWriter, + request *http.Request, + ) { + if !strings.HasSuffix(request.URL.Path, "/actions/jwks") { + http.NotFound(writer, request) + return + } + call := calls.Add(1) + _, _ = fmt.Fprintf(writer, `{"keys":[{"call":%d}]}`, call) + })) + defer upstream.Close() + + proxy := newTestProxy(t, upstream.URL, time.Hour, 20*time.Millisecond) + server := httptest.NewServer(proxy.PublicHandler()) + defer server.Close() + + endpoint := server.URL + "/v1/iam/clients/client-one/actions/jwks" + first := makePublicRequest(t, endpoint, "", "") + assertResponse(t, first, http.StatusOK, `{"keys":[{"call":1}]}`) + second := makePublicRequest(t, endpoint, "", "") + assertResponse(t, second, http.StatusOK, `{"keys":[{"call":1}]}`) + time.Sleep(30 * time.Millisecond) + third := makePublicRequest(t, endpoint, "", "") + assertResponse(t, third, http.StatusOK, `{"keys":[{"call":2}]}`) +} + +func TestInternalInvalidationRemovesAllAccessTokens(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func( + writer http.ResponseWriter, + _ *http.Request, + ) { + call := calls.Add(1) + _, _ = fmt.Fprintf(writer, `{"call":%d}`, call) + })) + defer upstream.Close() + + proxy := newTestProxy(t, upstream.URL, time.Hour, time.Hour) + publicServer := httptest.NewServer(proxy.PublicHandler()) + defer publicServer.Close() + internalServer := httptest.NewServer(proxy.InternalHandler()) + defer internalServer.Close() + + tokens := []string{ + testAccessToken(testTokenUUID, time.Now().Add(time.Hour)), + testAccessTokenWithNonce(testTokenUUID, time.Now().Add(time.Hour), "second"), + } + for _, token := range tokens { + response := makePublicRequest( + t, + publicServer.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + "", + ) + assertResponse(t, response, http.StatusOK, "") + } + + request, err := http.NewRequest( + http.MethodDelete, + internalServer.URL+invalidationPrefix+testTokenUUID, + nil, + ) + if err != nil { + t.Fatalf("create invalidation request: %v", err) + } + invalidationResponse, err := http.DefaultClient.Do(request) + if err != nil { + t.Fatalf("perform invalidation request: %v", err) + } + assertResponse(t, invalidationResponse, http.StatusNoContent, "") + + for _, token := range tokens { + response := makePublicRequest( + t, + publicServer.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + "", + ) + assertResponse(t, response, http.StatusOK, "") + } + + if got, want := calls.Load(), int32(4); got != want { + t.Fatalf("upstream calls = %d, want %d", got, want) + } +} + +func TestUpstreamErrorsAreNotCached(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func( + writer http.ResponseWriter, + _ *http.Request, + ) { + calls.Add(1) + http.Error(writer, "denied", http.StatusUnauthorized) + })) + defer upstream.Close() + + proxy := newTestProxy(t, upstream.URL, time.Hour, time.Hour) + server := httptest.NewServer(proxy.PublicHandler()) + defer server.Close() + + token := testAccessToken(testTokenUUID, time.Now().Add(time.Hour)) + for range 2 { + response := makePublicRequest( + t, + server.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + "", + ) + assertResponse(t, response, http.StatusUnauthorized, "") + } + if got, want := calls.Load(), int32(2); got != want { + t.Fatalf("upstream calls = %d, want %d", got, want) + } +} + +func TestConcurrentMissesAreCoalesced(t *testing.T) { + t.Parallel() + + var calls atomic.Int32 + release := make(chan struct{}) + upstream := httptest.NewServer(http.HandlerFunc(func( + writer http.ResponseWriter, + _ *http.Request, + ) { + calls.Add(1) + <-release + _, _ = writer.Write([]byte(`{"permissions":[]}`)) + })) + defer upstream.Close() + + proxy := newTestProxy(t, upstream.URL, time.Hour, time.Hour) + server := httptest.NewServer(proxy.PublicHandler()) + defer server.Close() + token := testAccessToken(testTokenUUID, time.Now().Add(time.Hour)) + + const requests = 8 + results := make(chan int, requests) + for range requests { + go func() { + response := makePublicRequest( + t, + server.URL+"/v1/iam/clients/client-one/actions/introspect", + token, + "", + ) + results <- response.StatusCode + _, _ = io.Copy(io.Discard, response.Body) + _ = response.Body.Close() + }() + } + + deadline := time.Now().Add(time.Second) + for calls.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + close(release) + for range requests { + if status := <-results; status != http.StatusOK { + t.Errorf("status = %d, want %d", status, http.StatusOK) + } + } + if got, want := calls.Load(), int32(1); got != want { + t.Fatalf("upstream calls = %d, want %d", got, want) + } +} + +func newTestProxy( + t *testing.T, + coreURL string, + introspectionTTL time.Duration, + jwksTTL time.Duration, +) *Proxy { + t.Helper() + + parsedURL, err := url.Parse(coreURL) + if err != nil { + t.Fatalf("parse Core URL: %v", err) + } + return NewProxy(Config{ + CoreURL: parsedURL, + RequestTimeout: time.Second, + IntrospectionCacheTTL: introspectionTTL, + IntrospectionCacheMaxEntries: 100, + JWKSCacheTTL: jwksTTL, + JWKSCacheMaxEntries: 100, + }) +} + +func makePublicRequest( + t *testing.T, + endpoint string, + accessToken string, + otp string, +) *http.Response { + t.Helper() + + request, err := http.NewRequest(http.MethodGet, endpoint, nil) + if err != nil { + t.Fatalf("create request: %v", err) + } + if accessToken != "" { + request.Header.Set("Authorization", "Bearer "+accessToken) + } + if otp != "" { + request.Header.Set("X-OTP", otp) + } + response, err := http.DefaultClient.Do(request) + if err != nil { + t.Fatalf("perform request: %v", err) + } + return response +} + +func assertResponse( + t *testing.T, + response *http.Response, + wantStatus int, + wantBody string, +) { + t.Helper() + defer response.Body.Close() + + body, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("read response: %v", err) + } + if got := response.StatusCode; got != wantStatus { + t.Fatalf("status = %d, want %d; body = %q", got, wantStatus, body) + } + if wantBody != "" && strings.TrimSpace(string(body)) != wantBody { + t.Fatalf("body = %q, want %q", body, wantBody) + } +} + +func testAccessToken(tokenUUID string, expiration time.Time) string { + return testAccessTokenWithNonce(tokenUUID, expiration, "") +} + +func testAccessTokenWithNonce( + tokenUUID string, + expiration time.Time, + nonce string, +) string { + header, _ := json.Marshal(map[string]string{"alg": "none"}) + payload, _ := json.Marshal(map[string]any{ + "jti": tokenUUID, + "exp": expiration.Unix(), + "nonce": nonce, + }) + return base64.RawURLEncoding.EncodeToString(header) + "." + + base64.RawURLEncoding.EncodeToString(payload) + ".signature" +}