refactor(auth): Decouple BasicAuth and OAuth authentication strategies
Completely separates multi-user BasicAuth mode from OAuth mode with no fallback between them. These are now mutually exclusive authentication strategies based on deployment configuration. Changes: - Create separate functions: get_user_client_basic_auth() and get_user_client_oauth() with clear separation of concerns - Update get_user_client() to dispatch based on use_basic_auth parameter - Pass use_basic_auth through all background sync tasks - Update app.py to determine auth mode at startup - Rewrite integration tests to verify no OAuth fallback in BasicAuth mode - Fix test assertions for response field names and duplicate title handling 🤖 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
@@ -1,17 +1,24 @@
|
||||
"""Integration tests for app password provisioning via Astrolabe.
|
||||
|
||||
Tests the complete flow:
|
||||
Tests the complete flow for multi-user BasicAuth mode:
|
||||
1. User stores app password via Astrolabe API
|
||||
2. MCP server retrieves it via OAuth client credentials
|
||||
3. Background sync uses it to access Nextcloud
|
||||
3. Background sync uses it to access Nextcloud (NOT OAuth refresh tokens)
|
||||
|
||||
These tests verify that BasicAuth and OAuth are completely separate concerns
|
||||
with no fallback between them.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from httpx import BasicAuth
|
||||
|
||||
from nextcloud_mcp_server.auth.astrolabe_client import AstrolabeClient
|
||||
from nextcloud_mcp_server.config import get_settings
|
||||
from nextcloud_mcp_server.vector.oauth_sync import get_user_client
|
||||
from nextcloud_mcp_server.vector.oauth_sync import (
|
||||
NotProvisionedError,
|
||||
get_user_client,
|
||||
get_user_client_basic_auth,
|
||||
get_user_client_oauth,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@@ -77,9 +84,20 @@ async def test_get_user_app_password_returns_none_for_unconfigured_user():
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_dual_credential_support_in_background_sync(mocker):
|
||||
"""Test that background sync tries app password first, then refresh token."""
|
||||
from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
|
||||
async def test_basic_auth_mode_uses_app_password_only(mocker):
|
||||
"""Test that BasicAuth mode uses ONLY app passwords, NOT OAuth tokens.
|
||||
|
||||
In multi-user BasicAuth mode, OAuth refresh tokens are NOT used.
|
||||
This is a complete separation of concerns.
|
||||
"""
|
||||
# Mock settings to have client credentials
|
||||
mock_settings = mocker.MagicMock()
|
||||
mock_settings.oidc_client_id = "test-client-id"
|
||||
mock_settings.oidc_client_secret = "test-client-secret"
|
||||
mocker.patch(
|
||||
"nextcloud_mcp_server.vector.oauth_sync.get_settings",
|
||||
return_value=mock_settings,
|
||||
)
|
||||
|
||||
# Mock AstrolabeClient to return an app password
|
||||
mock_astrolabe = mocker.AsyncMock()
|
||||
@@ -90,35 +108,36 @@ async def test_dual_credential_support_in_background_sync(mocker):
|
||||
return_value=mock_astrolabe,
|
||||
)
|
||||
|
||||
# Mock TokenBrokerService (shouldn't be called if app password works)
|
||||
mock_token_broker = mocker.MagicMock(spec=TokenBrokerService)
|
||||
# Call get_user_client in BasicAuth mode
|
||||
_client = await get_user_client(
|
||||
user_id="test_user",
|
||||
token_broker=None, # No token broker needed for BasicAuth mode
|
||||
nextcloud_host="http://localhost:8080",
|
||||
use_basic_auth=True,
|
||||
)
|
||||
|
||||
# Call get_user_client - should use app password
|
||||
try:
|
||||
_client = await get_user_client(
|
||||
user_id="test_user",
|
||||
token_broker=mock_token_broker,
|
||||
nextcloud_host="http://localhost:8080",
|
||||
)
|
||||
# Verify app password was requested
|
||||
mock_astrolabe.get_user_app_password.assert_called_once_with("test_user")
|
||||
|
||||
# Verify app password was requested
|
||||
mock_astrolabe.get_user_app_password.assert_called_once_with("test_user")
|
||||
|
||||
# Verify token broker was NOT called (app password took priority)
|
||||
mock_token_broker.get_background_token.assert_not_called()
|
||||
|
||||
# Verify client uses BasicAuth
|
||||
assert _client.auth is not None
|
||||
assert isinstance(_client.auth, BasicAuth)
|
||||
except Exception:
|
||||
# May fail in test environment, but we verified the priority logic
|
||||
pass
|
||||
# Verify client was created successfully with correct username
|
||||
assert _client is not None
|
||||
assert _client.username == "test_user"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_background_sync_falls_back_to_refresh_token(mocker):
|
||||
"""Test that background sync falls back to refresh token if no app password."""
|
||||
from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
|
||||
async def test_basic_auth_mode_raises_error_without_app_password(mocker):
|
||||
"""Test that BasicAuth mode raises NotProvisionedError if no app password.
|
||||
|
||||
There is NO fallback to OAuth - if no app password, user must provision one.
|
||||
"""
|
||||
# Mock settings to have client credentials
|
||||
mock_settings = mocker.MagicMock()
|
||||
mock_settings.oidc_client_id = "test-client-id"
|
||||
mock_settings.oidc_client_secret = "test-client-secret"
|
||||
mocker.patch(
|
||||
"nextcloud_mcp_server.vector.oauth_sync.get_settings",
|
||||
return_value=mock_settings,
|
||||
)
|
||||
|
||||
# Mock AstrolabeClient to return None (no app password)
|
||||
mock_astrolabe = mocker.AsyncMock()
|
||||
@@ -129,23 +148,131 @@ async def test_background_sync_falls_back_to_refresh_token(mocker):
|
||||
return_value=mock_astrolabe,
|
||||
)
|
||||
|
||||
# Call get_user_client in BasicAuth mode - should raise NotProvisionedError
|
||||
with pytest.raises(NotProvisionedError) as exc_info:
|
||||
await get_user_client(
|
||||
user_id="test_user",
|
||||
token_broker=None,
|
||||
nextcloud_host="http://localhost:8080",
|
||||
use_basic_auth=True,
|
||||
)
|
||||
|
||||
# Verify error message mentions app password provisioning
|
||||
assert "app password" in str(exc_info.value).lower()
|
||||
assert "test_user" in str(exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_oauth_mode_uses_refresh_token_only(mocker):
|
||||
"""Test that OAuth mode uses ONLY refresh tokens, NOT app passwords.
|
||||
|
||||
In OAuth mode, app passwords are NOT used.
|
||||
This is a complete separation of concerns.
|
||||
"""
|
||||
from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
|
||||
|
||||
# Mock TokenBrokerService to return an access token
|
||||
mock_token_broker = mocker.AsyncMock(spec=TokenBrokerService)
|
||||
mock_token_broker.get_background_token.return_value = "test-access-token"
|
||||
|
||||
# Call get_user_client - should fall back to refresh token
|
||||
try:
|
||||
_client = await get_user_client(
|
||||
# Call get_user_client in OAuth mode
|
||||
_client = await get_user_client(
|
||||
user_id="test_user",
|
||||
token_broker=mock_token_broker,
|
||||
nextcloud_host="http://localhost:8080",
|
||||
use_basic_auth=False, # OAuth mode
|
||||
)
|
||||
|
||||
# Verify token broker was called (NOT Astrolabe)
|
||||
mock_token_broker.get_background_token.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_oauth_mode_raises_error_without_token(mocker):
|
||||
"""Test that OAuth mode raises NotProvisionedError if no refresh token.
|
||||
|
||||
There is NO fallback to app passwords - if no token, user must provision.
|
||||
"""
|
||||
from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
|
||||
|
||||
# Mock TokenBrokerService to return None (no token)
|
||||
mock_token_broker = mocker.AsyncMock(spec=TokenBrokerService)
|
||||
mock_token_broker.get_background_token.return_value = None
|
||||
|
||||
# Call get_user_client in OAuth mode - should raise NotProvisionedError
|
||||
with pytest.raises(NotProvisionedError) as exc_info:
|
||||
await get_user_client(
|
||||
user_id="test_user",
|
||||
token_broker=mock_token_broker,
|
||||
nextcloud_host="http://localhost:8080",
|
||||
use_basic_auth=False,
|
||||
)
|
||||
|
||||
# Verify app password was attempted first
|
||||
mock_astrolabe.get_user_app_password.assert_called_once_with("test_user")
|
||||
# Verify error message mentions OAuth provisioning
|
||||
assert "oauth" in str(exc_info.value).lower()
|
||||
assert "test_user" in str(exc_info.value)
|
||||
|
||||
# Verify token broker was called as fallback
|
||||
mock_token_broker.get_background_token.assert_called_once()
|
||||
except Exception:
|
||||
# May fail in test environment, but we verified the fallback logic
|
||||
pass
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_get_user_client_basic_auth_function(mocker):
|
||||
"""Test the dedicated get_user_client_basic_auth function."""
|
||||
# Mock settings to have client credentials
|
||||
mock_settings = mocker.MagicMock()
|
||||
mock_settings.oidc_client_id = "test-client-id"
|
||||
mock_settings.oidc_client_secret = "test-client-secret"
|
||||
mocker.patch(
|
||||
"nextcloud_mcp_server.vector.oauth_sync.get_settings",
|
||||
return_value=mock_settings,
|
||||
)
|
||||
|
||||
# Mock AstrolabeClient
|
||||
mock_astrolabe = mocker.AsyncMock()
|
||||
mock_astrolabe.get_user_app_password.return_value = "xxxxx-xxxxx-xxxxx-xxxxx-xxxxx"
|
||||
|
||||
mocker.patch(
|
||||
"nextcloud_mcp_server.vector.oauth_sync.AstrolabeClient",
|
||||
return_value=mock_astrolabe,
|
||||
)
|
||||
|
||||
# Call dedicated function
|
||||
client = await get_user_client_basic_auth(
|
||||
user_id="alice",
|
||||
nextcloud_host="http://localhost:8080",
|
||||
)
|
||||
|
||||
assert client is not None
|
||||
assert client.username == "alice"
|
||||
mock_astrolabe.get_user_app_password.assert_called_once_with("alice")
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_get_user_client_oauth_function(mocker):
|
||||
"""Test the dedicated get_user_client_oauth function."""
|
||||
from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
|
||||
|
||||
# Mock TokenBrokerService
|
||||
mock_token_broker = mocker.AsyncMock(spec=TokenBrokerService)
|
||||
mock_token_broker.get_background_token.return_value = "test-bearer-token"
|
||||
|
||||
# Call dedicated function
|
||||
client = await get_user_client_oauth(
|
||||
user_id="alice",
|
||||
token_broker=mock_token_broker,
|
||||
nextcloud_host="http://localhost:8080",
|
||||
)
|
||||
|
||||
assert client is not None
|
||||
assert client.username == "alice"
|
||||
mock_token_broker.get_background_token.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_oauth_mode_requires_token_broker():
|
||||
"""Test that OAuth mode requires a token broker."""
|
||||
with pytest.raises(ValueError, match="token_broker required"):
|
||||
await get_user_client(
|
||||
user_id="test_user",
|
||||
token_broker=None, # Missing token broker
|
||||
nextcloud_host="http://localhost:8080",
|
||||
use_basic_auth=False, # OAuth mode
|
||||
)
|
||||
|
||||
@@ -4,18 +4,26 @@ Tests that BasicAuth credentials are extracted from request headers
|
||||
and passed through to Nextcloud APIs without storage (stateless).
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_basic_auth_pass_through_notes_list(nc_mcp_basic_auth_client):
|
||||
"""Test BasicAuth pass-through with notes list tool."""
|
||||
async def test_basic_auth_pass_through_notes_search(nc_mcp_basic_auth_client):
|
||||
"""Test BasicAuth pass-through with notes search tool."""
|
||||
# Call tool - BasicAuth header is set at connection level by fixture
|
||||
response = await nc_mcp_basic_auth_client.call_tool("nc_notes_list", {})
|
||||
response = await nc_mcp_basic_auth_client.call_tool(
|
||||
"nc_notes_search_notes", {"query": "test"}
|
||||
)
|
||||
|
||||
# Verify tool executed successfully with pass-through auth
|
||||
assert response is not None
|
||||
assert "results" in response or "content" in response
|
||||
assert not response.isError, f"Tool returned error: {response.content}"
|
||||
# Response should have content with results
|
||||
assert len(response.content) > 0
|
||||
data = json.loads(response.content[0].text)
|
||||
assert "results" in data
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@@ -23,7 +31,7 @@ async def test_basic_auth_pass_through_notes_create(nc_mcp_basic_auth_client):
|
||||
"""Test BasicAuth pass-through with notes create tool."""
|
||||
# Create a note using BasicAuth
|
||||
response = await nc_mcp_basic_auth_client.call_tool(
|
||||
"nc_notes_create",
|
||||
"nc_notes_create_note",
|
||||
{
|
||||
"title": "BasicAuth Test Note",
|
||||
"content": "This note was created via BasicAuth pass-through",
|
||||
@@ -32,16 +40,35 @@ async def test_basic_auth_pass_through_notes_create(nc_mcp_basic_auth_client):
|
||||
)
|
||||
|
||||
assert response is not None
|
||||
assert response.get("success") is True or "note_id" in response
|
||||
assert not response.isError, f"Tool returned error: {response.content}"
|
||||
# Parse response and verify note was created
|
||||
data = json.loads(response.content[0].text)
|
||||
assert data.get("success") is True or "note_id" in data
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_basic_auth_pass_through_search(nc_mcp_basic_auth_client):
|
||||
"""Test BasicAuth pass-through with search tool."""
|
||||
# Search notes using BasicAuth
|
||||
async def test_basic_auth_pass_through_get_note(nc_mcp_basic_auth_client):
|
||||
"""Test BasicAuth pass-through with get note tool."""
|
||||
# First create a note to get
|
||||
create_response = await nc_mcp_basic_auth_client.call_tool(
|
||||
"nc_notes_create_note",
|
||||
{
|
||||
"title": "BasicAuth Get Test",
|
||||
"content": "Note to retrieve",
|
||||
"category": "Test",
|
||||
},
|
||||
)
|
||||
assert not create_response.isError
|
||||
create_data = json.loads(create_response.content[0].text)
|
||||
note_id = create_data.get("id")
|
||||
|
||||
# Now get the note using BasicAuth
|
||||
response = await nc_mcp_basic_auth_client.call_tool(
|
||||
"nc_notes_search", {"query": "BasicAuth"}
|
||||
"nc_notes_get_note", {"note_id": note_id}
|
||||
)
|
||||
|
||||
assert response is not None
|
||||
assert "results" in response or "content" in response
|
||||
assert not response.isError, f"Tool returned error: {response.content}"
|
||||
data = json.loads(response.content[0].text)
|
||||
# Nextcloud may append a number to duplicate titles
|
||||
assert data.get("title", "").startswith("BasicAuth Get Test")
|
||||
|
||||
Reference in New Issue
Block a user