fix: Use settings.enable_offline_access for env var consolidation

Migrate all direct ENABLE_OFFLINE_ACCESS environment variable checks to
use settings.enable_offline_access, which handles both the new
ENABLE_BACKGROUND_OPERATIONS and deprecated ENABLE_OFFLINE_ACCESS vars.

Also fixes JWT issuer validation in Docker by using NEXTCLOUD_PUBLIC_ISSUER_URL
when set, resolving 401 errors caused by internal/external URL mismatch.

Changes:
- app.py: Use settings for offline access checks in setup_oauth_config,
  register_oauth_client, and tool registration
- oauth_tools.py: Use settings in provision_nextcloud_access and check_logged_in
- management.py: Use settings in get_user_session
- scope_authorization.py: Use settings in require_scopes decorator
- Remove unused os imports after migration

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2025-12-24 09:10:01 -06:00
co-authored by Claude Opus 4.5
parent 10a0969138
commit a51376fd5a
4 changed files with 49 additions and 46 deletions
+6 -6
View File
@@ -11,7 +11,6 @@ The PHP app obtains tokens through PKCE flow and uses them to access these endpo
""" """
import logging import logging
import os
import time import time
from importlib.metadata import version from importlib.metadata import version
from typing import Any from typing import Any
@@ -347,11 +346,12 @@ async def get_user_session(request: Request) -> JSONResponse:
) )
# Check if offline access is enabled # Check if offline access is enabled
enable_offline_access = os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() in ( # Use settings.enable_offline_access which handles both ENABLE_BACKGROUND_OPERATIONS (new)
"true", # and ENABLE_OFFLINE_ACCESS (deprecated) environment variables
"1", from nextcloud_mcp_server.config import get_settings
"yes",
) settings = get_settings()
enable_offline_access = settings.enable_offline_access
if not enable_offline_access: if not enable_offline_access:
# Offline access disabled - return minimal session info # Offline access disabled - return minimal session info
+24 -25
View File
@@ -539,11 +539,10 @@ async def load_oauth_client_credentials(
dcr_scopes = "openid profile email notes:read notes:write calendar:read calendar:write todo:read todo:write contacts:read contacts:write cookbook:read cookbook:write deck:read deck:write tables:read tables:write files:read files:write sharing:read sharing:write news:read news:write" dcr_scopes = "openid profile email notes:read notes:write calendar:read calendar:write todo:read todo:write contacts:read contacts:write cookbook:read cookbook:write deck:read deck:write tables:read tables:write files:read files:write sharing:read sharing:write news:read news:write"
# Add offline_access scope if refresh tokens are enabled # Add offline_access scope if refresh tokens are enabled
enable_offline_access = os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() in ( # Use settings.enable_offline_access which handles both ENABLE_BACKGROUND_OPERATIONS (new)
"true", # and ENABLE_OFFLINE_ACCESS (deprecated) environment variables
"1", dcr_settings = get_settings()
"yes", enable_offline_access = dcr_settings.enable_offline_access
)
if enable_offline_access: if enable_offline_access:
dcr_scopes = f"{dcr_scopes} offline_access" dcr_scopes = f"{dcr_scopes} offline_access"
logger.info("✓ offline_access scope enabled for refresh tokens") logger.info("✓ offline_access scope enabled for refresh tokens")
@@ -672,6 +671,10 @@ async def setup_oauth_config():
Returns: Returns:
Tuple of (nextcloud_host, token_verifier, auth_settings, refresh_token_storage, oauth_client, oauth_provider, client_id, client_secret) Tuple of (nextcloud_host, token_verifier, auth_settings, refresh_token_storage, oauth_client, oauth_provider, client_id, client_secret)
""" """
# Get settings for enable_offline_access check (handles both ENABLE_BACKGROUND_OPERATIONS
# and ENABLE_OFFLINE_ACCESS environment variables)
settings = get_settings()
nextcloud_host = os.getenv("NEXTCLOUD_HOST") nextcloud_host = os.getenv("NEXTCLOUD_HOST")
if not nextcloud_host: if not nextcloud_host:
raise ValueError( raise ValueError(
@@ -744,11 +747,9 @@ async def setup_oauth_config():
logger.info("✓ Detected integrated mode (Nextcloud OIDC app)") logger.info("✓ Detected integrated mode (Nextcloud OIDC app)")
# Check if offline access (refresh tokens) is enabled # Check if offline access (refresh tokens) is enabled
enable_offline_access = os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() in ( # Use settings.enable_offline_access which handles both ENABLE_BACKGROUND_OPERATIONS (new)
"true", # and ENABLE_OFFLINE_ACCESS (deprecated) environment variables
"1", enable_offline_access = settings.enable_offline_access
"yes",
)
# Initialize refresh token storage if enabled # Initialize refresh token storage if enabled
refresh_token_storage = None refresh_token_storage = None
@@ -799,8 +800,10 @@ async def setup_oauth_config():
) )
# ADR-005: Unified Token Verifier with proper audience validation # ADR-005: Unified Token Verifier with proper audience validation
# Use discovered issuer for JWT validation # Use public issuer URL for JWT validation if set (handles Docker internal/external URL mismatch)
client_issuer = issuer # Tokens are issued with the public URL, but OIDC discovery returns internal URL
public_issuer_url = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
client_issuer = public_issuer_url if public_issuer_url else issuer
# Get MCP server URL for audience validation # Get MCP server URL for audience validation
mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
nextcloud_resource_uri = os.getenv("NEXTCLOUD_RESOURCE_URI", nextcloud_host) nextcloud_resource_uri = os.getenv("NEXTCLOUD_RESOURCE_URI", nextcloud_host)
@@ -817,10 +820,8 @@ async def setup_oauth_config():
"This should be set explicitly for proper audience validation." "This should be set explicitly for proper audience validation."
) )
# Create settings for UnifiedTokenVerifier # Create settings for UnifiedTokenVerifier (use same settings instance from start of function)
from nextcloud_mcp_server.config import get_settings # settings is already set at the start of setup_oauth_config()
settings = get_settings()
# Override with discovered values if not set in environment # Override with discovered values if not set in environment
if not settings.oidc_client_id: if not settings.oidc_client_id:
settings.oidc_client_id = client_id settings.oidc_client_id = client_id
@@ -1015,8 +1016,10 @@ async def setup_oauth_config_for_multi_user_basic(
mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
nextcloud_resource_uri = os.getenv("NEXTCLOUD_RESOURCE_URI", nextcloud_host) nextcloud_resource_uri = os.getenv("NEXTCLOUD_RESOURCE_URI", nextcloud_host)
# Use discovered issuer for JWT validation # Use public issuer URL for JWT validation if set (handles Docker internal/external URL mismatch)
client_issuer = issuer # Tokens are issued with the public URL, but OIDC discovery returns internal URL
public_issuer_url = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
client_issuer = public_issuer_url if public_issuer_url else issuer
# Update settings with discovered values for UnifiedTokenVerifier # Update settings with discovered values for UnifiedTokenVerifier
if not settings.oidc_client_id: if not settings.oidc_client_id:
@@ -1399,13 +1402,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
enable_token_exchange = ( enable_token_exchange = (
os.getenv("ENABLE_TOKEN_EXCHANGE", "false").lower() == "true" os.getenv("ENABLE_TOKEN_EXCHANGE", "false").lower() == "true"
) )
enable_offline_access_for_tools = os.getenv( # Use settings.enable_offline_access which handles both ENABLE_BACKGROUND_OPERATIONS (new)
"ENABLE_OFFLINE_ACCESS", "false" # and ENABLE_OFFLINE_ACCESS (deprecated) environment variables
).lower() in ( enable_offline_access_for_tools = settings.enable_offline_access
"true",
"1",
"yes",
)
if oauth_enabled and enable_offline_access_for_tools and not enable_token_exchange: if oauth_enabled and enable_offline_access_for_tools and not enable_token_exchange:
logger.info("Registering OAuth provisioning tools for offline access") logger.info("Registering OAuth provisioning tools for offline access")
register_oauth_tools(mcp) register_oauth_tools(mcp)
@@ -1,7 +1,6 @@
"""Scope-based authorization for MCP tools.""" """Scope-based authorization for MCP tools."""
import logging import logging
import os
from functools import wraps from functools import wraps
from typing import Any, Callable from typing import Any, Callable
@@ -131,9 +130,12 @@ def require_scopes(*required_scopes: str):
required_scopes_set = set(required_scopes) required_scopes_set = set(required_scopes)
# Check if offline access is enabled # Check if offline access is enabled
enable_offline_access = ( # Use settings.enable_offline_access which handles both ENABLE_BACKGROUND_OPERATIONS (new)
os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() == "true" # and ENABLE_OFFLINE_ACCESS (deprecated) environment variables
) from nextcloud_mcp_server.config import get_settings
settings = get_settings()
enable_offline_access = settings.enable_offline_access
# In offline access mode, check if Nextcloud scopes require provisioning # In offline access mode, check if Nextcloud scopes require provisioning
if enable_offline_access: if enable_offline_access:
+13 -11
View File
@@ -303,16 +303,17 @@ async def provision_nextcloud_access(
), ),
) )
# Get configuration # Get configuration using settings (handles both ENABLE_BACKGROUND_OPERATIONS
enable_offline_access = ( # and ENABLE_OFFLINE_ACCESS environment variables)
os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() == "true" from nextcloud_mcp_server.config import get_settings
)
if not enable_offline_access: settings = get_settings()
if not settings.enable_offline_access:
return ProvisioningResult( return ProvisioningResult(
success=False, success=False,
message=( message=(
"Offline access is not enabled. " "Offline access is not enabled. "
"Set ENABLE_OFFLINE_ACCESS=true to use this feature." "Set ENABLE_BACKGROUND_OPERATIONS=true to use this feature."
), ),
) )
@@ -488,13 +489,14 @@ async def check_logged_in(ctx: Context, user_id: Optional[str] = None) -> str:
logger.info("=" * 60) logger.info("=" * 60)
# Not logged in - generate OAuth URL for Flow 2 # Not logged in - generate OAuth URL for Flow 2
enable_offline_access = ( # Use settings (handles both ENABLE_BACKGROUND_OPERATIONS and ENABLE_OFFLINE_ACCESS)
os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() == "true" from nextcloud_mcp_server.config import get_settings
)
if not enable_offline_access: settings = get_settings()
if not settings.enable_offline_access:
return ( return (
"Not logged in. Offline access is not enabled. " "Not logged in. Offline access is not enabled. "
"Set ENABLE_OFFLINE_ACCESS=true to use this feature." "Set ENABLE_BACKGROUND_OPERATIONS=true to use this feature."
) )
# Get MCP server's OAuth client credentials # Get MCP server's OAuth client credentials