You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by la...@apache.org on 2026/06/23 08:03:17 UTC
(airavata-custos) 02/02: Add identity resolver and core.auth configuration
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake pushed a commit to branch auth
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit 9f397b914ca7e0da8e03b69617ca3dfe5853bf83
Author: lahiruj <la...@gmail.com>
AuthorDate: Tue Jun 23 04:03:01 2026 -0400
Add identity resolver and core.auth configuration
---
config/custos.yaml | 8 +++
docs/CONFIG.md | 47 ++++++++++++++++
internal/config/config.go | 47 +++++++++++++++-
internal/config/config_test.go | 96 +++++++++++++++++++++++++++++++
internal/store/store.go | 3 +
internal/store/user_store.go | 33 +++++++++++
pkg/identity/identity.go | 4 +-
pkg/identity/verifier.go | 3 +-
pkg/service/identity_resolver.go | 119 +++++++++++++++++++++++++++++++++++++++
pkg/service/service.go | 3 +
10 files changed, 357 insertions(+), 6 deletions(-)
diff --git a/config/custos.yaml b/config/custos.yaml
index 6296d21a7..7568b510e 100644
--- a/config/custos.yaml
+++ b/config/custos.yaml
@@ -4,6 +4,14 @@ core:
api:
port: 8080
log_level: "info"
+ auth:
+ oidc:
+ issuer: "${OIDC_ISSUER_URL}"
+ audience: "${OIDC_AUDIENCE}"
+ cache_ttl: "30s"
+ public_paths:
+ - "/healthz"
+ - "/ready"
connectors:
slurm-mapper:
diff --git a/docs/CONFIG.md b/docs/CONFIG.md
new file mode 100644
index 000000000..1667acb59
--- /dev/null
+++ b/docs/CONFIG.md
@@ -0,0 +1,47 @@
+# Custos configuration
+
+The server reads `config/custos.yaml` at boot. `${VAR}` substitution keeps
+secrets out of the file; an unset var stays as the literal `${VAR}` so
+missing-secret failures surface at the consumer, not at load time.
+
+## Core
+
+| Key | Default | Notes |
+|-----|---------|-------|
+| `core.database.url` | *(required)* | MariaDB / MySQL DSN. `parseTime=true` is mandatory. |
+| `core.api.port` | `8080` | HTTP API port. |
+| `core.log_level` | `info` | One of `debug`, `info`, `warn`, `error`. |
+
+## Auth
+
+The verifier + caller resolver read this block. The loader refuses to boot
+when either OIDC field is empty; the cache TTL is capped at 60s.
+
+| Key | Default | Notes |
+|-----|---------|-------|
+| `core.auth.oidc.issuer` | *(required)* | OIDC issuer URL. The verifier discovers JWKS from `<issuer>/.well-known/openid-configuration`. Empty → boot fails with `core.auth.oidc.issuer is required`. |
+| `core.auth.oidc.audience` | *(required)* | Expected `aud` claim. Usually the client_id registered with the IdP. Empty → boot fails with `core.auth.oidc.audience is required`. |
+| `core.auth.cache_ttl` | `30s` | TTL of the in-process caller + privilege cache, keyed by OIDC sub. Units: any `time.Duration` string (`s`, `m`). Zero or negative → `30s`. Values above `60s` are capped at `60s` at boot and a WARN is logged. |
+| `core.auth.public_paths` | `[]` | List of path components that bypass JWT verification entirely. Typically `/healthz` and `/ready`. Routes registered via `router.Public(...)` are merged with this list at server bootstrap. |
+
+### Example
+
+```yaml
+core:
+ auth:
+ oidc:
+ issuer: "${OIDC_ISSUER_URL}"
+ audience: "${OIDC_AUDIENCE}"
+ cache_ttl: "30s"
+ public_paths:
+ - "/healthz"
+ - "/ready"
+```
+
+## Environment overrides
+
+| Variable | Default | Purpose |
+|----------|---------|---------|
+| `DB_MAX_OPEN_CONNS` | `25` | Maximum open database connections. |
+| `DB_MAX_IDLE_CONNS` | `5` | Maximum idle database connections. |
+| `CONFIG_PATH` | `config/custos.yaml` | Override the config file location. |
diff --git a/internal/config/config.go b/internal/config/config.go
index bbffcfb92..ecceb6a5e 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -19,6 +19,7 @@ package config
import (
"fmt"
+ "log/slog"
"os"
"regexp"
"time"
@@ -26,8 +27,16 @@ import (
"gopkg.in/yaml.v3"
)
+// IdentityCacheMaxTTL caps how long the caller resolver may hold a verified
+// identity in process memory.
+const IdentityCacheMaxTTL = 60 * time.Second
+
+// IdentityCacheDefaultTTL is the TTL applied when core.auth.cache_ttl is unset
+// or non-positive.
+const IdentityCacheDefaultTTL = 30 * time.Second
+
type Config struct {
- Core CoreConfig `yaml:"core"`
+ Core CoreConfig `yaml:"core"`
Connectors map[string]*ConnectorConfig `yaml:"connectors"`
}
@@ -35,6 +44,7 @@ type CoreConfig struct {
Database DatabaseConfig `yaml:"database"`
API APIConfig `yaml:"api"`
LogLevel string `yaml:"log_level"`
+ Auth AuthConfig `yaml:"auth"`
}
type DatabaseConfig struct {
@@ -45,6 +55,17 @@ type APIConfig struct {
Port int `yaml:"port"`
}
+type AuthConfig struct {
+ OIDC OIDCConfig `yaml:"oidc"`
+ CacheTTL time.Duration `yaml:"cache_ttl"`
+ PublicPaths []string `yaml:"public_paths"`
+}
+
+type OIDCConfig struct {
+ Issuer string `yaml:"issuer"`
+ Audience string `yaml:"audience"`
+}
+
type ConnectorConfig struct {
Type string `yaml:"type"`
Enabled bool `yaml:"enabled"`
@@ -64,9 +85,33 @@ func LoadConfig(path string) (*Config, error) {
return nil, fmt.Errorf("failed to parse config file: %w", err)
}
+ if err := applyAuthDefaults(&cfg.Core.Auth); err != nil {
+ return nil, err
+ }
+
return &cfg, nil
}
+// applyAuthDefaults enforces the auth invariants:
+// - Issuer and Audience are required.
+// - CacheTTL <= 0 collapses to IdentityCacheDefaultTTL.
+// - CacheTTL > IdentityCacheMaxTTL caps at the ceiling and logs a warning.
+func applyAuthDefaults(a *AuthConfig) error {
+ if a.OIDC.Issuer == "" {
+ return fmt.Errorf("core.auth.oidc.issuer is required")
+ }
+ if a.OIDC.Audience == "" {
+ return fmt.Errorf("core.auth.oidc.audience is required")
+ }
+ if a.CacheTTL <= 0 {
+ a.CacheTTL = IdentityCacheDefaultTTL
+ } else if a.CacheTTL > IdentityCacheMaxTTL {
+ slog.Warn("core.auth.cache_ttl above cap; capping", "configured", a.CacheTTL, "cap", IdentityCacheMaxTTL)
+ a.CacheTTL = IdentityCacheMaxTTL
+ }
+ return nil
+}
+
func expandEnvVars(input string) string {
re := regexp.MustCompile(`\$\{([^}]+)\}`)
return re.ReplaceAllStringFunc(input, func(match string) string {
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index 787fbe078..3ab363569 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -19,12 +19,25 @@ package config
import (
"os"
+ "path/filepath"
"testing"
+ "time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
+// writeTempYAML writes contents to a temp file and returns its path.
+func writeTempYAML(t *testing.T, contents string) string {
+ t.Helper()
+ dir := t.TempDir()
+ p := filepath.Join(dir, "custos.yaml")
+ if err := os.WriteFile(p, []byte(contents), 0o600); err != nil {
+ t.Fatalf("write temp yaml: %v", err)
+ }
+ return p
+}
+
func TestLoadConfig(t *testing.T) {
cfg, err := LoadConfig("../../config/custos.yaml")
require.NoError(t, err)
@@ -100,3 +113,86 @@ func TestConnectorConfigGetters(t *testing.T) {
}
})
}
+
+func TestLoadConfig_AuthDefaults(t *testing.T) {
+ cfg, err := LoadConfig("../../config/custos.yaml")
+ require.NoError(t, err)
+
+ assert.Equal(t, 30*time.Second, cfg.Core.Auth.CacheTTL)
+ assert.Equal(t, []string{"/healthz", "/ready"}, cfg.Core.Auth.PublicPaths)
+ assert.NotEmpty(t, cfg.Core.Auth.OIDC.Issuer, "issuer should not be empty (env-unset stays as ${...} literal)")
+ assert.NotEmpty(t, cfg.Core.Auth.OIDC.Audience)
+}
+
+func TestLoadConfig_AuthRejectsEmptyIssuer(t *testing.T) {
+ yaml := `core:
+ database:
+ url: "dsn"
+ api:
+ port: 8080
+ log_level: "info"
+ auth:
+ oidc:
+ issuer: ""
+ audience: "aud"
+`
+ _, err := LoadConfig(writeTempYAML(t, yaml))
+ if err == nil {
+ t.Fatal("expected error for empty issuer, got nil")
+ }
+ assert.Contains(t, err.Error(), "core.auth.oidc.issuer")
+}
+
+func TestLoadConfig_AuthRejectsEmptyAudience(t *testing.T) {
+ yaml := `core:
+ database:
+ url: "dsn"
+ api:
+ port: 8080
+ log_level: "info"
+ auth:
+ oidc:
+ issuer: "https://idp.example"
+ audience: ""
+`
+ _, err := LoadConfig(writeTempYAML(t, yaml))
+ if err == nil {
+ t.Fatal("expected error for empty audience, got nil")
+ }
+ assert.Contains(t, err.Error(), "core.auth.oidc.audience")
+}
+
+func TestLoadConfig_AuthClampsTTLAboveMax(t *testing.T) {
+ yaml := `core:
+ database:
+ url: "dsn"
+ api:
+ port: 8080
+ log_level: "info"
+ auth:
+ oidc:
+ issuer: "https://idp.example"
+ audience: "aud"
+ cache_ttl: "5m"
+`
+ cfg, err := LoadConfig(writeTempYAML(t, yaml))
+ require.NoError(t, err)
+ assert.Equal(t, IdentityCacheMaxTTL, cfg.Core.Auth.CacheTTL)
+}
+
+func TestLoadConfig_AuthDefaultsTTLWhenZero(t *testing.T) {
+ yaml := `core:
+ database:
+ url: "dsn"
+ api:
+ port: 8080
+ log_level: "info"
+ auth:
+ oidc:
+ issuer: "https://idp.example"
+ audience: "aud"
+`
+ cfg, err := LoadConfig(writeTempYAML(t, yaml))
+ require.NoError(t, err)
+ assert.Equal(t, IdentityCacheDefaultTTL, cfg.Core.Auth.CacheTTL)
+}
diff --git a/internal/store/store.go b/internal/store/store.go
index b473c7bc8..604e1d58e 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -31,6 +31,9 @@ type UserStore interface {
FindByID(ctx context.Context, id string) (*models.User, error)
// FindByEmail returns the user with the given email, or nil if not found.
FindByEmail(ctx context.Context, email string) (*models.User, error)
+ // GetUserByOIDCSub returns the user owning the user_identities row whose
+ // oidc_sub matches or nil if no row links the given subject.
+ GetUserByOIDCSub(ctx context.Context, oidcSub string) (*models.User, error)
// FindByOrganization returns all users belonging to the given organization.
FindByOrganization(ctx context.Context, organizationID string) ([]models.User, error)
// Create inserts a new user within the provided transaction.
diff --git a/internal/store/user_store.go b/internal/store/user_store.go
index 32d89da1f..9b2c065f3 100644
--- a/internal/store/user_store.go
+++ b/internal/store/user_store.go
@@ -21,6 +21,7 @@ import (
"context"
"database/sql"
"errors"
+ "strings"
"github.com/jmoiron/sqlx"
@@ -64,6 +65,38 @@ func (s *mysqlUserStore) FindByEmail(ctx context.Context, email string) (*models
return &u, nil
}
+// GetUserByOIDCSub returns the user owning the user_identities row whose
+// oidc_sub matches. Returns nil when the OIDC subject is empty or no row
+// links it to a Custos user.
+func (s *mysqlUserStore) GetUserByOIDCSub(ctx context.Context, oidcSub string) (*models.User, error) {
+ if oidcSub == "" {
+ return nil, nil
+ }
+ var u models.User
+ err := s.db.GetContext(ctx, &u,
+ `SELECT `+prefixed("u", userColumns)+`
+ FROM users u
+ JOIN user_identities ui ON ui.user_id = u.id
+ WHERE ui.oidc_sub = ?`, oidcSub)
+ if err != nil {
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, nil
+ }
+ return nil, err
+ }
+ return &u, nil
+}
+
+// prefixed returns the comma-separated column list with each bare column
+// prefixed by alias. Used to disambiguate joined queries.
+func prefixed(alias, columns string) string {
+ parts := strings.Split(columns, ", ")
+ for i, p := range parts {
+ parts[i] = alias + "." + p
+ }
+ return strings.Join(parts, ", ")
+}
+
func (s *mysqlUserStore) FindByOrganization(ctx context.Context, organizationID string) ([]models.User, error) {
var users []models.User
err := s.db.SelectContext(ctx, &users,
diff --git a/pkg/identity/identity.go b/pkg/identity/identity.go
index 2564392d7..d50bfbef3 100644
--- a/pkg/identity/identity.go
+++ b/pkg/identity/identity.go
@@ -29,9 +29,7 @@ import (
// Caller is the verified identity behind an HTTP request.
type Caller struct {
- UserID string
- OIDCSub string
- Email string
+ UserID string
}
// ErrNotLinked is returned by a UserResolver when the OIDC sub has no row in
diff --git a/pkg/identity/verifier.go b/pkg/identity/verifier.go
index 8044730ad..24314d063 100644
--- a/pkg/identity/verifier.go
+++ b/pkg/identity/verifier.go
@@ -29,8 +29,7 @@ type JWTVerifier struct{ inner *oidc.IDTokenVerifier }
// Claims are the minimal subset the middleware needs after verification.
type Claims struct {
- Sub string `json:"sub"`
- Email string `json:"email"`
+ Sub string `json:"sub"`
}
// NewJWTVerifier resolves the provider's discovery doc and builds a verifier
diff --git a/pkg/service/identity_resolver.go b/pkg/service/identity_resolver.go
new file mode 100644
index 000000000..84512e6ff
--- /dev/null
+++ b/pkg/service/identity_resolver.go
@@ -0,0 +1,119 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 service
+
+import (
+ "context"
+ "sync"
+ "time"
+
+ "github.com/apache/airavata-custos/pkg/identity"
+ "github.com/apache/airavata-custos/pkg/models"
+)
+
+// identityCacheDefaultTTL is the fallback TTL when SetIdentityCacheTTL is never
+// invoked or receives a non-positive value. The config loader applies the
+// same default and enforces a 60s ceiling before the value reaches the cache.
+const identityCacheDefaultTTL = 30 * time.Second
+
+type identityCacheEntry struct {
+ caller *identity.Caller
+ privs []models.PrivilegeKey
+ expires time.Time
+}
+
+// identityCache is the sub-keyed in-process cache the resolver consults before
+// hitting the database. Entries expire on read after TTL.
+type identityCache struct {
+ mu sync.Mutex
+ entries map[string]identityCacheEntry
+ ttl time.Duration
+}
+
+func newIdentityCache() *identityCache {
+ return &identityCache{entries: map[string]identityCacheEntry{}, ttl: identityCacheDefaultTTL}
+}
+
+func (c *identityCache) get(sub string, now time.Time) (identityCacheEntry, bool) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ e, ok := c.entries[sub]
+ if !ok {
+ return identityCacheEntry{}, false
+ }
+ if !now.Before(e.expires) {
+ delete(c.entries, sub)
+ return identityCacheEntry{}, false
+ }
+ return e, true
+}
+
+func (c *identityCache) set(sub string, caller *identity.Caller, privs []models.PrivilegeKey, now time.Time) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.entries[sub] = identityCacheEntry{caller: caller, privs: privs, expires: now.Add(c.ttl)}
+}
+
+func (c *identityCache) invalidateAll() {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.entries = map[string]identityCacheEntry{}
+}
+
+func (c *identityCache) setTTL(d time.Duration) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if d <= 0 {
+ d = identityCacheDefaultTTL
+ }
+ c.ttl = d
+}
+
+// SetIdentityCacheTTL configures the resolver cache's TTL. The config loader
+// caps the value at the 60s ceiling before it reaches the cache.
+func (s *Service) SetIdentityCacheTTL(d time.Duration) {
+ s.identityCache.setTTL(d)
+}
+
+// InvalidateAllIdentities empties the cache.
+func (s *Service) InvalidateAllIdentities() {
+ s.identityCache.invalidateAll()
+}
+
+// ResolveCaller maps a verified OIDC sub to a Caller + effective privilege set.
+// Returns ErrNotLinked when no user_identities row references sub.
+func (s *Service) ResolveCaller(ctx context.Context, oidcSub string) (*identity.Caller, []models.PrivilegeKey, error) {
+ now := time.Now()
+ if hit, ok := s.identityCache.get(oidcSub, now); ok {
+ return hit.caller, hit.privs, nil
+ }
+ user, err := s.users.GetUserByOIDCSub(ctx, oidcSub)
+ if err != nil {
+ return nil, nil, err
+ }
+ if user == nil {
+ return nil, nil, identity.ErrNotLinked
+ }
+ privs, err := s.EffectivePrivileges(ctx, user.ID)
+ if err != nil {
+ return nil, nil, err
+ }
+ caller := &identity.Caller{UserID: user.ID}
+ s.identityCache.set(oidcSub, caller, privs, now)
+ return caller, privs, nil
+}
diff --git a/pkg/service/service.go b/pkg/service/service.go
index af2635aab..c1737d7ab 100644
--- a/pkg/service/service.go
+++ b/pkg/service/service.go
@@ -57,6 +57,7 @@ type Service struct {
privileges store.UserPrivilegeStore
roles store.RoleStore
userRoles store.UserRoleStore
+ identityCache *identityCache
}
// New constructs a Service backed by the supplied database handle.
@@ -86,6 +87,7 @@ func New(database *sqlx.DB, eventBus *events.Bus) *Service {
privileges: store.NewUserPrivilegeStore(database),
roles: store.NewRoleStore(database),
userRoles: store.NewUserRoleStore(database),
+ identityCache: newIdentityCache(),
}
}
@@ -141,6 +143,7 @@ func NewWithStores(
privileges: privileges,
roles: roles,
userRoles: userRoles,
+ identityCache: newIdentityCache(),
}
}