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 <[email protected]> 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(), } }
