fix: OIDC audience validation - use string not list

python-jose jwt.decode() requires audience as string or None, not list.
Now extract and validate audience from unverified claims first,
then use token's actual audience for JWT decode.

Fixes "audience must be a string or None" error.

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

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

CHANGELOG.md

@@ -5,7 +5,32 @@ 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.8] - 2026-01-08
+
+### Fixed
+
+- **OIDC audience validation** - python-jose requires string audience, not list
+  - Extract and validate audience from unverified claims first
+  - Use token's actual audience for JWT decode (after validating it's allowed)
+  - Fixes "audience must be a string or None" error
+
+## [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 +38,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.8"
 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,21 @@ async def get_current_user(
     token = credentials.credentials
 
     try:
+        # First, extract issuer and audience from unverified claims
+        unverified_claims = jwt.get_unverified_claims(token)
+        token_issuer = unverified_claims.get("iss", "")
+        token_audience = unverified_claims.get("aud", "")
+
+        # 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")
+
+        # Validate audience is in our allowed list
+        if token_audience not in oidc_config.audiences:
+            logger.warning(f"Invalid token audience: {token_audience}")
+            raise HTTPException(status_code=401, detail="Invalid token audience")
+
         # Decode token header to get key ID
         unverified_header = jwt.get_unverified_header(token)
         kid = unverified_header.get("kid")
@@ -116,8 +153,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", []):
@@ -129,17 +166,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 and audience
         payload = jwt.decode(
             token,
             rsa_key,
             algorithms=["RS256"],
-            audience=oidc_config.audiences,
-            issuer=oidc_config.issuer,
+            audience=token_audience,  # Use the token's audience (already validated)
+            issuer=token_issuer,  # Use the token's issuer (already validated)
         )
 
         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/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,30 @@ 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", "")
+        token_audience = unverified_claims.get("aud", "")
 
         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
+        logger.debug(f"Token issuer: {token_issuer}, allowed issuers: {oidc_config.issuers}")
+        if not oidc_config.is_valid_issuer(token_issuer):
+            logger.warning(f"Invalid token issuer: {token_issuer} (allowed: {oidc_config.issuers})")
+            raise HTTPException(status_code=401, detail="Invalid token issuer")
+
+        # Validate audience is in allowed list
+        if token_audience not in oidc_config.audiences:
+            logger.warning(f"Invalid token audience: {token_audience} (allowed: {oidc_config.audiences})")
+            raise HTTPException(status_code=401, detail="Invalid token audience")
+
+        # Get JWKS for this specific issuer
+        jwks = get_jwks_for_issuer(token_issuer)
         rsa_key = None
 
         for key in jwks.get("keys", []):
@@ -177,17 +214,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 and audience
         payload = jwt.decode(
             token,
             rsa_key,
             algorithms=["RS256"],
-            audience=oidc_config.audiences,
-            issuer=oidc_config.issuer,
+            audience=token_audience,  # Use the token's audience (already validated)
+            issuer=token_issuer,  # Use the token's issuer (already validated)
         )
 
         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
 
@@ -285,12 +322,14 @@ async def get_optional_user(
         }
 
     if not credentials:
+        logger.debug("No credentials provided for optional auth")
         return None
 
     try:
         return await get_current_user(credentials)
-    except HTTPException:
-        # Invalid token - return None instead of raising
+    except HTTPException as e:
+        # Invalid token - log and return None instead of raising
+        logger.warning(f"Optional auth failed: {e.detail}")
         return None
 
 

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,