feat: multi-issuer OIDC support with config consolidation

- Support multiple OAuth providers (core-api, tatlock-ui, tatlock)
- Changed oidc_issuer (string) to oidc_issuers (list)
- Per-issuer JWKS caching
- Validates token issuer against allowed list
- Consolidated config files (removed deprecated src/config.py, src/security.py)
- Updated imports to use src/shared/config and src/shared/security

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -5,7 +5,23 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [1.10.5] - 2026-01-07
+## [1.10.7] - 2026-01-08
+
+### Added
+
+- **Multi-issuer OIDC support** - Accept tokens from multiple OAuth providers
+  - Changed `oidc_issuer` (string) to `oidc_issuers` (list)
+  - Each issuer has its own JWKS endpoint, now cached per-issuer
+  - Validates token issuer against allowed list before fetching JWKS
+  - Supports tokens from: `core-api`, `tatlock-ui`, `tatlock` OAuth applications
+  - Completes fix for environment endpoint user resolution
+
+### Removed
+
+- Deprecated `src/config.py` - consolidated to `src/shared/config.py`
+- Deprecated `src/security.py` - consolidated to `src/shared/security.py`
+
+## [1.10.6] - 2026-01-08
 
 ### Fixed
 
@@ -13,6 +29,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - Changed `oidc_audience` (string) to `oidc_audiences` (list)
   - Now accepts tokens with audience: `core-api`, `tatlock-ui`, or `tatlock`
   - Fixes environment endpoint returning "default" user instead of authenticated username
+  - Fixed main.py to use `oidc_audiences[0]` for Swagger UI OAuth client
+
+### Changed
+
+- Documentation cleanup in README
 
 ## [1.10.4] - 2026-01-07
 

README.md

@@ -97,10 +97,6 @@ src/
 # Install dependencies
 pip install -r requirements.txt
 
-# Copy credentials template
-cp src/credentials.example.py src/credentials.py
-# Edit src/credentials.py with your values
-
 # Run locally
 uvicorn src.main:app --reload --host 0.0.0.0 --port 8083
 ```
@@ -164,8 +160,8 @@ docker run -p 8083:8083 core-code:latest
 | `HOMEASSISTANT_URL` | Home Assistant URL | `http://localhost:8123` |
 | `HOMEASSISTANT_TOKEN` | HA long-lived access token | - |
 | `OIDC_ENABLED` | Enable OIDC auth | `false` |
-| `OIDC_ISSUER` | OIDC issuer URL | - |
-| `OIDC_AUDIENCE` | OIDC audience | - |
+| `OIDC_ISSUERS` | OIDC issuer URLs (comma-separated) | See config.py |
+| `OIDC_AUDIENCES` | OIDC audiences (comma-separated) | See config.py |
 
 ## API Documentation
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "core-api"
-version = "1.10.5"
+version = "1.10.7"
 description = "Core Code API - Infrastructure management and tools API"
 readme = "README.md"
 requires-python = ">=3.12"

src/auth/oidc.py

@@ -22,29 +22,42 @@ class OIDCConfig:
     def __init__(self):
         # These will be set from environment variables in config.py
         self.enabled = False
-        self.issuer = ""
+        self.issuers: list[str] = []
         self.audiences: list[str] = []
-        self.jwks_uri = ""
 
-    def configure(self, enabled: bool, issuer: str, audiences: list[str]):
+    def configure(self, enabled: bool, issuers: list[str], audiences: list[str]):
         """Configure OIDC settings"""
         self.enabled = enabled
-        self.issuer = issuer
+        self.issuers = [iss.rstrip('/') for iss in issuers]  # Normalize without trailing slash
         self.audiences = audiences
-        self.jwks_uri = f"{issuer.rstrip('/')}/jwks/"
-        logger.info(f"OIDC configured: enabled={enabled}, issuer={issuer}, audiences={audiences}")
+        logger.info(f"OIDC configured: enabled={enabled}, issuers={self.issuers}, audiences={audiences}")
+
+    def get_jwks_uri(self, issuer: str) -> str:
+        """Get JWKS URI for a specific issuer"""
+        return f"{issuer.rstrip('/')}/jwks/"
+
+    def is_valid_issuer(self, issuer: str) -> bool:
+        """Check if issuer is in the allowed list"""
+        normalized = issuer.rstrip('/')
+        return normalized in self.issuers
 
 
 # Global OIDC config instance
 oidc_config = OIDCConfig()
 
 
-@lru_cache(maxsize=1)
-def get_jwks() -> Dict:
-    """
-    Fetch JSON Web Key Set (JWKS) from Authentik
+# Per-issuer JWKS cache
+_jwks_cache: Dict[str, Dict] = {}
 
-    Cached to avoid repeated requests. Cache is cleared on server restart.
+
+def get_jwks_for_issuer(issuer: str) -> Dict:
+    """
+    Fetch JSON Web Key Set (JWKS) for a specific issuer.
+
+    Cached per-issuer to avoid repeated requests. Cache is cleared on server restart.
+
+    Args:
+        issuer: The token issuer URL
 
     Returns:
         JWKS dictionary containing public keys for token verification
@@ -55,15 +68,24 @@ def get_jwks() -> Dict:
     if not oidc_config.enabled:
         return {}
 
+    normalized_issuer = issuer.rstrip('/')
+
+    # Return cached JWKS if available
+    if normalized_issuer in _jwks_cache:
+        return _jwks_cache[normalized_issuer]
+
+    jwks_uri = oidc_config.get_jwks_uri(normalized_issuer)
+
     try:
-        logger.debug(f"Fetching JWKS from {oidc_config.jwks_uri}")
-        response = httpx.get(oidc_config.jwks_uri, timeout=10.0)
+        logger.debug(f"Fetching JWKS from {jwks_uri}")
+        response = httpx.get(jwks_uri, timeout=10.0)
         response.raise_for_status()
         jwks = response.json()
-        logger.info(f"JWKS fetched successfully ({len(jwks.get('keys', []))} keys)")
+        logger.info(f"JWKS fetched successfully for {normalized_issuer} ({len(jwks.get('keys', []))} keys)")
+        _jwks_cache[normalized_issuer] = jwks
         return jwks
     except Exception as e:
-        logger.error(f"Failed to fetch JWKS: {e}")
+        logger.error(f"Failed to fetch JWKS from {jwks_uri}: {e}")
         raise HTTPException(
             status_code=503,
             detail="Authentication service unavailable"
@@ -109,6 +131,15 @@ async def get_current_user(
     token = credentials.credentials
 
     try:
+        # First, extract issuer from unverified claims to know which JWKS to use
+        unverified_claims = jwt.get_unverified_claims(token)
+        token_issuer = unverified_claims.get("iss", "")
+
+        # Validate issuer is in our allowed list
+        if not oidc_config.is_valid_issuer(token_issuer):
+            logger.warning(f"Invalid token issuer: {token_issuer}")
+            raise HTTPException(status_code=401, detail="Invalid token issuer")
+
         # Decode token header to get key ID
         unverified_header = jwt.get_unverified_header(token)
         kid = unverified_header.get("kid")
@@ -116,8 +147,8 @@ async def get_current_user(
         if not kid:
             raise HTTPException(status_code=401, detail="Invalid token format")
 
-        # Find matching key in JWKS
-        jwks = get_jwks()
+        # Find matching key in JWKS for this specific issuer
+        jwks = get_jwks_for_issuer(token_issuer)
         rsa_key = None
 
         for key in jwks.get("keys", []):
@@ -130,12 +161,13 @@ async def get_current_user(
             raise HTTPException(status_code=401, detail="Invalid token key")
 
         # Verify and decode token (accepts any of the configured audiences)
+        # Use the token's issuer for validation (already verified it's in our allowed list)
         payload = jwt.decode(
             token,
             rsa_key,
             algorithms=["RS256"],
             audience=oidc_config.audiences,
-            issuer=oidc_config.issuer,
+            issuer=token_issuer,
         )
 
         user_email = payload.get("email", "unknown")

src/auth/service.py

@@ -13,7 +13,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
-from src.config import get_settings
+from src.shared.config import get_settings
 from src.logging_config import get_logger
 from src.db.models import User, Role, UserPreferences, Group
 from src.auth.schemas import TokenInfoSchema, UserSchema, RoleSchema, UserPreferencesSchema, UserListItemSchema, BulkSyncResultSchema, GroupListItemSchema

src/clients/homeassistant_client.py

@@ -10,7 +10,7 @@ import json
 from typing import Optional, Dict, List, Any
 from datetime import datetime, timedelta, timezone
 from src.logging_config import get_logger
-from src.config import get_settings
+from src.shared.config import get_settings
 
 logger = get_logger(__name__)
 settings = get_settings()

src/clients/npm_client.py

@@ -7,7 +7,7 @@ import httpx
 from typing import Optional, Dict, List, Any
 from datetime import datetime, timedelta
 from src.logging_config import get_logger
-from src.config import get_settings
+from src.shared.config import get_settings
 
 logger = get_logger(__name__)
 settings = get_settings()

src/clients/portainer_client.py

@@ -8,7 +8,7 @@ import httpx
 import json
 from typing import Optional, Dict, List, Any
 from src.logging_config import get_logger
-from src.config import get_settings
+from src.shared.config import get_settings
 
 logger = get_logger(__name__)
 settings = get_settings()

src/config.py

@@ -1,105 +0,0 @@
-"""
-Global configuration for Core Code API
-
-All configuration is loaded from environment variables or .env file.
-See .env.example for available settings.
-"""
-import tomllib
-from pathlib import Path
-from pydantic_settings import BaseSettings
-from functools import lru_cache
-
-
-def _get_version_from_pyproject() -> str:
-    """Load version from pyproject.toml"""
-    pyproject_path = Path(__file__).parent.parent / "pyproject.toml"
-    try:
-        with open(pyproject_path, "rb") as f:
-            data = tomllib.load(f)
-            return data.get("project", {}).get("version", "0.0.0")
-    except FileNotFoundError:
-        return "0.0.0"
-
-
-__version__ = _get_version_from_pyproject()
-
-
-class Settings(BaseSettings):
-    """Global application settings"""
-
-    # Application
-    app_name: str = "Core Code API"
-    app_version: str = __version__
-    debug: bool = False
-
-    # Server
-    host: str = "0.0.0.0"
-    port: int = 8083
-
-    # CORS
-    cors_origins: list[str] = ["*"]
-    cors_credentials: bool = True
-    cors_methods: list[str] = ["*"]
-    cors_headers: list[str] = ["*"]
-
-    # Logging
-    log_level: str = "DEBUG"
-
-    # Qdrant Configuration
-    qdrant_host: str = "qdrant"
-    qdrant_port: int = 6333
-    qdrant_collection_conversations: str = "core_api_conversations"
-    qdrant_collection_documents: str = "core_api_documents"
-    qdrant_collection_user_facts: str = "core_api_user_facts"
-
-    # Search Configuration
-    search_provider: str = "searxng"
-    searxng_url: str  # Required - set SEARXNG_URL in .env
-
-    # Infrastructure Management (Portainer)
-    portainer_url: str  # Required - set PORTAINER_URL in .env
-    portainer_api_key: str  # Required - set PORTAINER_API_KEY in .env
-
-    # Infrastructure Management (Nginx Proxy Manager)
-    npm_url: str  # Required - set NPM_URL in .env
-    npm_email: str  # Required - set NPM_EMAIL in .env
-    npm_password: str  # Required - set NPM_PASSWORD in .env
-
-    # Home Assistant Configuration
-    homeassistant_url: str  # Required - set HOMEASSISTANT_URL in .env
-    homeassistant_token: str  # Required - set HOMEASSISTANT_TOKEN in .env
-    homeassistant_timeout: int = 30
-
-    # PostgreSQL Database
-    postgres_host: str  # Required - set POSTGRES_HOST in .env (e.g., localhost:5432)
-    postgres_user: str = "core_api"
-    postgres_password: str  # Required - set POSTGRES_PASSWORD in .env
-    postgres_database: str = "core_api"
-
-    @property
-    def database_url(self) -> str:
-        """Construct database URL from components"""
-        return f"postgresql://{self.postgres_user}:{self.postgres_password}@{self.postgres_host}/{self.postgres_database}"
-
-    # OIDC Authentication (Authentik)
-    oidc_enabled: bool = False  # Set to True to require authentication
-    oidc_issuer: str = "https://auth.schweitz.net/application/o/core-api/"
-    # Accept tokens from multiple clients (core-api, tatlock-ui, tatlock)
-    oidc_audiences: list[str] = ["core-api", "tatlock-ui", "tatlock"]
-
-    # Authentik API (for token validation and user management)
-    # Must use domain name (not IP) when AUTHENTIK_COOKIE_DOMAIN is set
-    authentik_url: str = "https://auth.schweitz.net"  # Authentik base URL
-    authentik_username: str = ""  # Admin username for API access (AUTHENTIK_USERNAME env var)
-    authentik_password: str = ""  # Admin password for API access (AUTHENTIK_PASSWORD env var)
-
-    class Config:
-        env_file = ".env"
-        case_sensitive = False
-        extra = "ignore"  # Ignore extra env vars not defined in Settings
-
-
-@lru_cache()
-def get_settings() -> Settings:
-    """Cached settings instance"""
-    return Settings()

src/controllers/health_controller.py

@@ -7,7 +7,7 @@ from fastapi import APIRouter, Response
 from fastapi.responses import JSONResponse
 
 from src.controllers.base import BaseController
-from src.config import get_settings
+from src.shared.config import get_settings
 from src.logging_config import get_logger
 from src.db import get_database
 

src/db/database.py

@@ -15,7 +15,7 @@ from sqlalchemy.ext.asyncio import (
 from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.pool import NullPool
 
-from src.config import get_settings
+from src.shared.config import get_settings
 from src.logging_config import get_logger
 
 logger = get_logger(__name__)

src/domains/auth/oidc.py

@@ -63,29 +63,42 @@ class OIDCConfig:
     def __init__(self):
         # These will be set from environment variables in config.py
         self.enabled = False
-        self.issuer = ""
+        self.issuers: list[str] = []
         self.audiences: list[str] = []
-        self.jwks_uri = ""
 
-    def configure(self, enabled: bool, issuer: str, audiences: list[str]):
+    def configure(self, enabled: bool, issuers: list[str], audiences: list[str]):
         """Configure OIDC settings"""
         self.enabled = enabled
-        self.issuer = issuer
+        self.issuers = [iss.rstrip('/') for iss in issuers]  # Normalize without trailing slash
         self.audiences = audiences
-        self.jwks_uri = f"{issuer.rstrip('/')}/jwks/"
-        logger.info(f"OIDC configured: enabled={enabled}, issuer={issuer}, audiences={audiences}")
+        logger.info(f"OIDC configured: enabled={enabled}, issuers={self.issuers}, audiences={audiences}")
+
+    def get_jwks_uri(self, issuer: str) -> str:
+        """Get JWKS URI for a specific issuer"""
+        return f"{issuer.rstrip('/')}/jwks/"
+
+    def is_valid_issuer(self, issuer: str) -> bool:
+        """Check if issuer is in the allowed list"""
+        normalized = issuer.rstrip('/')
+        return normalized in self.issuers
 
 
 # Global OIDC config instance
 oidc_config = OIDCConfig()
 
 
-@lru_cache(maxsize=1)
-def get_jwks() -> Dict:
-    """
-    Fetch JSON Web Key Set (JWKS) from Authentik
+# Per-issuer JWKS cache
+_jwks_cache: Dict[str, Dict] = {}
 
-    Cached to avoid repeated requests. Cache is cleared on server restart.
+
+def get_jwks_for_issuer(issuer: str) -> Dict:
+    """
+    Fetch JSON Web Key Set (JWKS) for a specific issuer.
+
+    Cached per-issuer to avoid repeated requests. Cache is cleared on server restart.
+
+    Args:
+        issuer: The token issuer URL
 
     Returns:
         JWKS dictionary containing public keys for token verification
@@ -96,15 +109,24 @@ def get_jwks() -> Dict:
     if not oidc_config.enabled:
         return {}
 
+    normalized_issuer = issuer.rstrip('/')
+
+    # Return cached JWKS if available
+    if normalized_issuer in _jwks_cache:
+        return _jwks_cache[normalized_issuer]
+
+    jwks_uri = oidc_config.get_jwks_uri(normalized_issuer)
+
     try:
-        logger.debug(f"Fetching JWKS from {oidc_config.jwks_uri}")
-        response = httpx.get(oidc_config.jwks_uri, timeout=10.0)
+        logger.debug(f"Fetching JWKS from {jwks_uri}")
+        response = httpx.get(jwks_uri, timeout=10.0)
         response.raise_for_status()
         jwks = response.json()
-        logger.info(f"JWKS fetched successfully ({len(jwks.get('keys', []))} keys)")
+        logger.info(f"JWKS fetched successfully for {normalized_issuer} ({len(jwks.get('keys', []))} keys)")
+        _jwks_cache[normalized_issuer] = jwks
         return jwks
     except Exception as e:
-        logger.error(f"Failed to fetch JWKS: {e}")
+        logger.error(f"Failed to fetch JWKS from {jwks_uri}: {e}")
         raise HTTPException(
             status_code=503,
             detail="Authentication service unavailable"
@@ -157,15 +179,23 @@ async def get_current_user(
     token = credentials.credentials
 
     try:
-        # Decode token header to get key ID
+        # First, decode token without verification to get issuer and key ID
         unverified_header = jwt.get_unverified_header(token)
+        unverified_claims = jwt.get_unverified_claims(token)
+
         kid = unverified_header.get("kid")
+        token_issuer = unverified_claims.get("iss", "")
 
         if not kid:
             raise HTTPException(status_code=401, detail="Invalid token format")
 
-        # Find matching key in JWKS
-        jwks = get_jwks()
+        # Validate issuer is in allowed list
+        if not oidc_config.is_valid_issuer(token_issuer):
+            logger.warning(f"Invalid token issuer: {token_issuer}")
+            raise HTTPException(status_code=401, detail="Invalid token issuer")
+
+        # Get JWKS for this specific issuer
+        jwks = get_jwks_for_issuer(token_issuer)
         rsa_key = None
 
         for key in jwks.get("keys", []):
@@ -177,17 +207,17 @@ async def get_current_user(
             logger.warning(f"No matching key found for kid: {kid}")
             raise HTTPException(status_code=401, detail="Invalid token key")
 
-        # Verify and decode token (accepts any of the configured audiences)
+        # Verify and decode token using the token's actual issuer
         payload = jwt.decode(
             token,
             rsa_key,
             algorithms=["RS256"],
             audience=oidc_config.audiences,
-            issuer=oidc_config.issuer,
+            issuer=token_issuer,  # Use the token's issuer for validation
         )
 
         user_email = payload.get("email", "unknown")
-        logger.info(f"Authenticated user: {user_email}")
+        logger.info(f"Authenticated user: {user_email} (issuer: {token_issuer})")
 
         return payload
 

src/security.py

@@ -1,32 +0,0 @@
-"""
-Security initialization module
-
-Handles OIDC configuration and authentication setup
-"""
-from src.config import Settings
-from src.auth.oidc import oidc_config
-from src.logging_config import get_logger
-
-logger = get_logger(__name__)
-
-
-def initialize_oidc(settings: Settings) -> None:
-    """
-    Initialize OIDC authentication configuration
-
-    Configures the global oidc_config instance with settings from environment.
-    If OIDC is enabled, logs the issuer URL for verification.
-
-    Args:
-        settings: Application settings containing OIDC configuration
-    """
-    oidc_config.configure(
-        enabled=settings.oidc_enabled,
-        issuer=settings.oidc_issuer,
-        audience=settings.oidc_audience
-    )
-
-    if settings.oidc_enabled:
-        logger.info(f"✓ OIDC authentication enabled (issuer: {settings.oidc_issuer})")
-    else:
-        logger.info("○ OIDC authentication disabled - API is publicly accessible")

src/shared/config.py

@@ -90,8 +90,13 @@ class Settings(BaseSettings):
 
     # OIDC Authentication (Authentik)
     oidc_enabled: bool = False
-    oidc_issuer: str = "https://auth.schweitz.net/application/o/core-api/"
-    # Accept tokens from multiple clients (core-api, tatlock-ui, tatlock)
+    # Accept tokens from multiple OAuth providers (each has its own issuer/JWKS)
+    oidc_issuers: list[str] = [
+        "https://auth.schweitz.net/application/o/core-api/",
+        "https://auth.schweitz.net/application/o/tatlock-ui/",
+        "https://auth.schweitz.net/application/o/tatlock/",
+    ]
+    # Accept tokens from multiple clients
     oidc_audiences: list[str] = ["core-api", "tatlock-ui", "tatlock"]
 
     # Authentik API (for token validation and user management)

src/shared/security.py

@@ -23,17 +23,17 @@ def initialize_oidc(settings: Settings) -> None:
 
     auth_oidc_config.configure(
         enabled=settings.oidc_enabled,
-        issuer=settings.oidc_issuer,
+        issuers=settings.oidc_issuers,
         audiences=settings.oidc_audiences
     )
 
     domains_oidc_config.configure(
         enabled=settings.oidc_enabled,
-        issuer=settings.oidc_issuer,
+        issuers=settings.oidc_issuers,
         audiences=settings.oidc_audiences
     )
 
     if settings.oidc_enabled:
-        logger.info(f"OIDC authentication enabled (issuer: {settings.oidc_issuer})")
+        logger.info(f"OIDC authentication enabled (issuers: {settings.oidc_issuers})")
     else:
         logger.info("OIDC authentication disabled - API is publicly accessible")

tests/test_config.py

@@ -1,6 +1,6 @@
 """Tests for config module."""
 import pytest
-from src.config import (
+from src.shared.config import (
     __version__,
     Settings,
     get_settings,