admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: drop OAuth-refresh background-sync path from oauth_sync.py

Browse Source 
Follow-up to #787/#789 (ADR-022 cleanup). After
`oauth_enabled ↔ enable_login_flow` became an invariant, the
`use_basic_auth=False` branch in `vector/oauth_sync.py` — and the
parameter wiring that fed it — was no longer reachable from any
supported deployment mode. This commit removes the dead code.

- nextcloud_mcp_server/vector/oauth_sync.py:
  - Deleted `get_user_client_oauth` (the OAuth-token refresh helper) and
    its `VECTOR_SYNC_SCOPES` constant.
  - Deleted the `get_user_client` dispatcher. Internal callers now call
    `get_user_client_basic_auth` directly.
  - Dropped the `use_basic_auth: bool` parameter from `user_scanner_task`,
    `multi_user_processor_task`, `_run_user_scanner_with_scope`, and
    `user_manager_task`.
  - Dropped the `token_broker` parameter from the same four functions —
    they no longer need it now that the OAuth-refresh path is gone. The
    `TokenBrokerService` constructed in `app.py` is still used by the
    management API revoke endpoint, just not by background sync.
  - Simplified the user-list query in `user_manager_task` to always read
    from the `app_passwords` table.
  - Replaced all `mode_label = "BasicAuth" if use_basic_auth else "OAuth"`
    with a literal `[BasicAuth]` log prefix (keeps existing log filters
    working).
  - Updated the module docstring to describe the post-cleanup shape.
  - Dropped the now-unused `TYPE_CHECKING` import of `TokenBrokerService`.

- nextcloud_mcp_server/app.py: dropped the `use_basic_auth = True` block
  and the now-stale `token_broker if not use_basic_auth else None` /
  `use_basic_auth` positional args from the two `tg.start(...)` calls in
  the multi-user vector-sync lifespan. Token broker construction stays —
  still consumed by the management API revoke endpoint via
  `app.state.oauth_context["token_broker"]`.

- tests/integration/test_app_password_provisioning.py: deleted four tests
  that exercised the now-removed OAuth-refresh path
  (`test_oauth_mode_uses_refresh_token_only`,
  `test_oauth_mode_raises_error_without_token`,
  `test_get_user_client_oauth_function`,
  `test_oauth_mode_requires_token_broker`) plus the
  `test_get_user_client_dispatches_to_basic_auth` test for the deleted
  dispatcher. Updated the module docstring + imports accordingly. The
  BasicAuth-mode tests (`test_basic_auth_mode_uses_local_storage`,
  `test_multiple_users_basic_auth_mode`, etc.) all remain.

No runtime-behaviour change in any supported deployment mode — the deleted
branches were already unreachable post-PR #787. 3 files changed,
+59 / -301; 1010 unit tests pass; integration jobs for
`mcp-login-flow` and `mcp-multi-user-basic` are the critical regression
gates before merge.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-05-13 01:13:53 +02:00
co-authored by Claude Opus 4.7
parent 735a4019ed
commit 65345fd6eb
3 changed files with 59 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/vector/oauth_sync.py
+46 -164
View File
@@ -5,25 +5,22 @@ Manages background vector sync for multi-user deployments:
- Per-User Scanners: One scanner task per provisioned user
- Shared Processor Pool: Processes documents from all users
Authentication strategies are mutually exclusive by deployment mode:
Background sync authenticates as each provisioned user via locally-stored
Nextcloud app passwords (BasicAuth), retrieved through the management API
after the user completes Login Flow v2 (or, in multi-user BasicAuth mode,
the per-user Astrolabe provisioning flow).
Multi-user BasicAuth mode (MCP_DEPLOYMENT_MODE=multi_user_basic):
- Uses app passwords stored locally in MCP server's database
- Users provision via Astrolabe personal settings, which sends to MCP API
- OAuth is NOT used
OAuth mode (with external IdP like Keycloak):
- Uses OAuth refresh tokens via TokenBrokerService
- Users provision via browser OAuth flow
- App passwords are NOT used
These are separate concerns - no fallback between them.
The earlier OAuth refresh-token path was removed in the ADR-022 follow-up:
it depended on unmerged Nextcloud `user_oidc` patches for Bearer-token
validation on non-OCS endpoints, and was never reachable from any
supported deployment mode. The `TokenBrokerService` constructed in
`app.py` is retained for the management API revoke endpoint, not for
background sync.
"""
import logging
import time
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
import anyio
from anyio.abc import TaskGroup, TaskStatus
@@ -39,19 +36,8 @@ from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.vector.processor import process_document
from nextcloud_mcp_server.vector.scanner import DocumentTask, scan_user_documents
if TYPE_CHECKING:
from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
logger = logging.getLogger(__name__)
# Scopes required for vector sync operations
VECTOR_SYNC_SCOPES = [
"notes.read",
"files.read",
"deck.read",
# "news.read", # News app may not be installed
]
class NotProvisionedError(Exception):
"""User has not provisioned offline access or has revoked it."""
@@ -113,84 +99,13 @@ async def get_user_client_basic_auth(
)
async def get_user_client_oauth(
user_id: str,
token_broker: "TokenBrokerService",
nextcloud_host: str,
) -> NextcloudClient:
"""Get an authenticated NextcloudClient using OAuth refresh token.
For OAuth deployments with external IdP where users provision via
browser OAuth flow. App passwords are NOT used in this mode.
Args:
user_id: User identifier
token_broker: Token broker for obtaining access tokens
nextcloud_host: Nextcloud base URL
Returns:
Authenticated NextcloudClient with Bearer token
Raises:
NotProvisionedError: If user has not provisioned offline access
"""
token = await token_broker.get_background_token(user_id, VECTOR_SYNC_SCOPES)
if not token:
raise NotProvisionedError(
f"User {user_id} has not provisioned offline access. "
f"User must complete the OAuth provisioning flow."
)
logger.info(f"Using OAuth refresh token for background sync: {user_id}")
return NextcloudClient.from_token(
base_url=nextcloud_host,
token=token,
username=user_id,
)
async def get_user_client(
user_id: str,
token_broker: "TokenBrokerService | None",
nextcloud_host: str,
*,
use_basic_auth: bool = False,
) -> NextcloudClient:
"""Get an authenticated NextcloudClient for a user.
Dispatches to the appropriate authentication strategy based on mode.
These are mutually exclusive - no fallback between them.
Args:
user_id: User identifier
token_broker: Token broker for OAuth mode (can be None for BasicAuth mode)
nextcloud_host: Nextcloud base URL
use_basic_auth: If True, use app passwords via Astrolabe (BasicAuth mode).
If False, use OAuth refresh tokens (OAuth mode).
Returns:
Authenticated NextcloudClient
Raises:
NotProvisionedError: If user has not provisioned access for the mode
"""
if use_basic_auth:
return await get_user_client_basic_auth(user_id, nextcloud_host)
else:
if token_broker is None:
raise ValueError("token_broker required for OAuth mode")
return await get_user_client_oauth(user_id, token_broker, nextcloud_host)
async def user_scanner_task(
user_id: str,
send_stream: MemoryObjectSendStream[DocumentTask],
shutdown_event: anyio.Event,
wake_event: anyio.Event,
token_broker: "TokenBrokerService | None",
nextcloud_host: str,
*,
use_basic_auth: bool = False,
task_status: TaskStatus = anyio.TASK_STATUS_IGNORED,
) -> None:
"""Scanner task for a single user.
@@ -202,13 +117,10 @@ async def user_scanner_task(
send_stream: Stream to send changed documents to processors
shutdown_event: Event signaling shutdown
wake_event: Event to trigger immediate scan
token_broker: Token broker for OAuth mode (None for BasicAuth mode)
nextcloud_host: Nextcloud base URL
use_basic_auth: If True, use app passwords; if False, use OAuth tokens
task_status: Status object for signaling task readiness
"""
mode_label = "BasicAuth" if use_basic_auth else "OAuth"
logger.info(f"[{mode_label}] Scanner started for user: {user_id}")
logger.info(f"[BasicAuth] Scanner started for user: {user_id}")
settings = get_settings()
max_consecutive_errors = 5
@@ -216,16 +128,14 @@ async def user_scanner_task(
# Pre-validate credentials before entering scan loop
try:
nc_client = await get_user_client(
user_id, token_broker, nextcloud_host, use_basic_auth=use_basic_auth
)
nc_client = await get_user_client_basic_auth(user_id, nextcloud_host)
try:
await nc_client.capabilities() # Lightweight OCS call to validate creds
logger.info(f"[{mode_label}] Credentials validated for {user_id}")
logger.info(f"[BasicAuth] Credentials validated for {user_id}")
except HTTPStatusError as e:
if e.response.status_code in (401, 403):
logger.warning(
f"[{mode_label}] Credential validation failed for {user_id} "
f"[BasicAuth] Credential validation failed for {user_id} "
f"(HTTP {e.response.status_code}), not starting scan loop"
)
return
@@ -234,12 +144,12 @@ async def user_scanner_task(
await nc_client.close()
except NotProvisionedError:
logger.warning(
f"[{mode_label}] User {user_id} not provisioned, not starting scan loop"
f"[BasicAuth] User {user_id} not provisioned, not starting scan loop"
)
return
except Exception as e:
logger.warning(
f"[{mode_label}] Pre-validation failed for {user_id}: {e}. "
f"[BasicAuth] Pre-validation failed for {user_id}: {e}. "
f"Proceeding to scan loop (has its own error handling)."
)
@@ -249,9 +159,7 @@ async def user_scanner_task(
nc_client = None
try:
# Get fresh credentials for this scan cycle
nc_client = await get_user_client(
user_id, token_broker, nextcloud_host, use_basic_auth=use_basic_auth
)
nc_client = await get_user_client_basic_auth(user_id, nextcloud_host)
# Scan user's documents
await scan_user_documents(
@@ -264,7 +172,7 @@ async def user_scanner_task(
except NotProvisionedError:
logger.warning(
f"[{mode_label}] User {user_id} no longer provisioned, stopping scanner"
f"[BasicAuth] User {user_id} no longer provisioned, stopping scanner"
)
break
@@ -272,7 +180,7 @@ async def user_scanner_task(
status_code = e.response.status_code
if status_code in (401, 403):
logger.warning(
f"[{mode_label}] Scanner auth failed for {user_id} "
f"[BasicAuth] Scanner auth failed for {user_id} "
f"(HTTP {status_code}), stopping scanner. "
f"User may need to re-provision credentials."
)
@@ -280,7 +188,7 @@ async def user_scanner_task(
elif status_code == 429:
retry_after = min(int(e.response.headers.get("Retry-After", "60")), 300)
logger.warning(
f"[{mode_label}] Scanner rate-limited for {user_id}, "
f"[BasicAuth] Scanner rate-limited for {user_id}, "
f"backing off {retry_after}s"
)
try:
@@ -294,7 +202,7 @@ async def user_scanner_task(
else:
consecutive_errors += 1
logger.error(
f"[{mode_label}] Scanner HTTP error for {user_id}: {e} "
f"[BasicAuth] Scanner HTTP error for {user_id}: {e} "
f"({consecutive_errors}/{max_consecutive_errors})",
exc_info=True,
)
@@ -302,7 +210,7 @@ async def user_scanner_task(
except Exception as e:
consecutive_errors += 1
logger.error(
f"[{mode_label}] Scanner error for {user_id}: {e} "
f"[BasicAuth] Scanner error for {user_id}: {e} "
f"({consecutive_errors}/{max_consecutive_errors})",
exc_info=True,
)
@@ -313,7 +221,7 @@ async def user_scanner_task(
if consecutive_errors >= max_consecutive_errors:
logger.error(
f"[{mode_label}] Scanner for {user_id} hit {max_consecutive_errors} "
f"[BasicAuth] Scanner for {user_id} hit {max_consecutive_errors} "
f"consecutive errors, stopping scanner"
)
break
@@ -325,16 +233,14 @@ async def user_scanner_task(
except anyio.get_cancelled_exc_class():
break
logger.info(f"[{mode_label}] Scanner stopped for user: {user_id}")
logger.info(f"[BasicAuth] Scanner stopped for user: {user_id}")
async def multi_user_processor_task(
worker_id: int,
receive_stream: MemoryObjectReceiveStream[DocumentTask],
shutdown_event: anyio.Event,
token_broker: "TokenBrokerService | None",
nextcloud_host: str,
use_basic_auth: bool = False,
*,
task_status: TaskStatus = anyio.TASK_STATUS_IGNORED,
) -> None:
@@ -346,13 +252,10 @@ async def multi_user_processor_task(
worker_id: Worker identifier for logging
receive_stream: Stream to receive documents from
shutdown_event: Event signaling shutdown
token_broker: Token broker for OAuth mode (None for BasicAuth mode)
nextcloud_host: Nextcloud base URL
use_basic_auth: If True, use app passwords; if False, use OAuth tokens
task_status: Status object for signaling task readiness
"""
mode_label = "BasicAuth" if use_basic_auth else "OAuth"
logger.info(f"[{mode_label}] Processor {worker_id} started")
logger.info(f"[BasicAuth] Processor {worker_id} started")
task_status.started()
while not shutdown_event.is_set():
@@ -364,11 +267,8 @@ async def multi_user_processor_task(
doc_task = await receive_stream.receive()
# Get credentials for THIS document's user
nc_client = await get_user_client(
doc_task.user_id,
token_broker,
nextcloud_host,
use_basic_auth=use_basic_auth,
nc_client = await get_user_client_basic_auth(
doc_task.user_id, nextcloud_host
)
# Process the document
@@ -378,13 +278,13 @@ async def multi_user_processor_task(
continue
except anyio.EndOfStream:
logger.info(f"[{mode_label}] Processor {worker_id}: Stream closed, exiting")
logger.info(f"[BasicAuth] Processor {worker_id}: Stream closed, exiting")
break
except NotProvisionedError:
if doc_task:
logger.warning(
f"[{mode_label}] User {doc_task.user_id} not provisioned, "
f"[BasicAuth] User {doc_task.user_id} not provisioned, "
f"skipping {doc_task.doc_type}_{doc_task.doc_id}"
)
continue
@@ -392,20 +292,20 @@ async def multi_user_processor_task(
except Exception as e:
if doc_task:
logger.error(
f"[{mode_label}] Processor {worker_id} error processing "
f"[BasicAuth] Processor {worker_id} error processing "
f"{doc_task.doc_type}_{doc_task.doc_id}: {e}",
exc_info=True,
)
else:
logger.error(
f"[{mode_label}] Processor {worker_id} error: {e}", exc_info=True
f"[BasicAuth] Processor {worker_id} error: {e}", exc_info=True
)
finally:
if nc_client:
await nc_client.close()
logger.info(f"[{mode_label}] Processor {worker_id} stopped")
logger.info(f"[BasicAuth] Processor {worker_id} stopped")
# Backward compatibility alias
@@ -418,10 +318,8 @@ async def _run_user_scanner_with_scope(
send_stream: MemoryObjectSendStream[DocumentTask],
shutdown_event: anyio.Event,
wake_event: anyio.Event,
token_broker: "TokenBrokerService | None",
nextcloud_host: str,
user_states: dict[str, UserSyncState],
use_basic_auth: bool = False,
) -> None:
"""Wrapper to run scanner with cancellation scope.
@@ -435,9 +333,7 @@ async def _run_user_scanner_with_scope(
send_stream=cloned_stream,
shutdown_event=shutdown_event,
wake_event=wake_event,
token_broker=token_broker,
nextcloud_host=nextcloud_host,
use_basic_auth=use_basic_auth,
)
finally:
# Clean up on exit
@@ -450,12 +346,10 @@ async def user_manager_task(
send_stream: MemoryObjectSendStream[DocumentTask],
shutdown_event: anyio.Event,
wake_event: anyio.Event,
token_broker: "TokenBrokerService | None",
refresh_token_storage: "RefreshTokenStorage",
nextcloud_host: str,
user_states: dict[str, UserSyncState],
tg: TaskGroup,
use_basic_auth: bool = False,
*,
task_status: TaskStatus = anyio.TASK_STATUS_IGNORED,
) -> None:
@@ -469,41 +363,33 @@ async def user_manager_task(
send_stream: Stream to send documents to processors
shutdown_event: Event signaling shutdown
wake_event: Event to wake scanners for immediate scan
token_broker: Token broker for OAuth mode (None for BasicAuth mode)
refresh_token_storage: Storage for tracking provisioned users
nextcloud_host: Nextcloud base URL
user_states: Shared dict tracking active user scanners
tg: Task group for spawning scanner tasks
use_basic_auth: If True, use app passwords; if False, use OAuth tokens
task_status: Status object for signaling task readiness
"""
settings = get_settings()
poll_interval = settings.vector_sync_user_poll_interval
mode_label = "BasicAuth" if use_basic_auth else "OAuth"
logger.info(
f"[{mode_label}] User manager started (poll interval: {poll_interval}s)"
)
logger.info(f"[BasicAuth] User manager started (poll interval: {poll_interval}s)")
task_status.started()
while not shutdown_event.is_set():
try:
# Get current provisioned users based on mode
if use_basic_auth:
# BasicAuth / Login Flow v2 mode: query app_passwords table
provisioned_users = set(
await refresh_token_storage.get_all_app_password_user_ids()
)
else:
# OAuth mode: query refresh_tokens table
provisioned_users = set(await refresh_token_storage.get_all_user_ids())
# Query the app_passwords table — background sync always
# authenticates as the user via locally-stored Nextcloud app
# passwords (Login Flow v2 / multi-user BasicAuth).
provisioned_users = set(
await refresh_token_storage.get_all_app_password_user_ids()
)
active_users = set(user_states.keys())
# Start scanners for new users
new_users = provisioned_users - active_users
for user_id in new_users:
logger.info(
f"[{mode_label}] Starting scanner for newly provisioned user: {user_id}"
f"[BasicAuth] Starting scanner for newly provisioned user: {user_id}"
)
cancel_scope = anyio.CancelScope()
user_states[user_id] = UserSyncState(
@@ -519,30 +405,26 @@ async def user_manager_task(
send_stream,
shutdown_event,
wake_event,
token_broker,
nextcloud_host,
user_states,
use_basic_auth, # Positional after user_states
)
# Cancel scanners for revoked users
revoked_users = active_users - provisioned_users
for user_id in revoked_users:
logger.info(
f"[{mode_label}] Stopping scanner for revoked user: {user_id}"
)
logger.info(f"[BasicAuth] Stopping scanner for revoked user: {user_id}")
state = user_states.get(user_id)
if state:
state.cancel_scope.cancel()
# Note: state will be removed by _run_user_scanner_with_scope on exit
if new_users:
logger.info(f"[{mode_label}] Started {len(new_users)} new scanner(s)")
logger.info(f"[BasicAuth] Started {len(new_users)} new scanner(s)")
if revoked_users:
logger.info(f"[{mode_label}] Stopped {len(revoked_users)} scanner(s)")
logger.info(f"[BasicAuth] Stopped {len(revoked_users)} scanner(s)")
except Exception as e:
logger.error(f"[{mode_label}] User manager error: {e}", exc_info=True)
logger.error(f"[BasicAuth] User manager error: {e}", exc_info=True)
# Sleep until next poll
try:
@@ -553,9 +435,9 @@ async def user_manager_task(
# Cancel all remaining scanners on shutdown
logger.info(
f"[{mode_label}] User manager shutting down, cancelling {len(user_states)} scanner(s)"
f"[BasicAuth] User manager shutting down, cancelling {len(user_states)} scanner(s)"
)
for state in list(user_states.values()):
state.cancel_scope.cancel()
logger.info(f"[{mode_label}] User manager stopped")
logger.info("[BasicAuth] User manager stopped")