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

refactor: change OAuth scope separator from colon to dot for IDP compatibility

Browse Source 
Many identity providers (AWS Cognito, Okta, Azure AD) reject or mishandle
colons in OAuth scope names. This migrates all custom scopes from
`resource:action` to `resource.action` format (e.g., `notes:read` →
`notes.read`), which is universally accepted and aligns with industry
conventions (Microsoft, Google).

Includes Alembic migration 004 for stored scope strings and ADR-024
documenting the rationale and RFC references.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-04-07 10:07:02 +02:00
co-authored by Claude Opus 4.6
parent 899b9c7191
commit 29fd0486c9
44 changed files with 724 additions and 520 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+33 -33
View File
@@ -29,43 +29,43 @@ logger = logging.getLogger(__name__)
# Default scopes for OAuth testing - all app-specific read/write scopes
DEFAULT_FULL_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"
"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"
)
# Read-only scopes (all read scopes across apps) - should match DEFAULT_FULL_SCOPES read portion
DEFAULT_READ_SCOPES = (
"openid profile email "
"notes:read "
"calendar:read "
"todo:read "
"contacts:read "
"cookbook:read "
"deck:read "
"tables:read "
"files:read "
"sharing:read"
"notes.read "
"calendar.read "
"todo.read "
"contacts.read "
"cookbook.read "
"deck.read "
"tables.read "
"files.read "
"sharing.read"
)
# Write-only scopes (all write scopes across apps) - should match DEFAULT_FULL_SCOPES write portion
DEFAULT_WRITE_SCOPES = (
"openid profile email "
"notes:write "
"calendar:write "
"todo:write "
"contacts:write "
"cookbook:write "
"deck:write "
"tables:write "
"files:write "
"sharing:write"
"notes.write "
"calendar.write "
"todo.write "
"contacts.write "
"cookbook.write "
"deck.write "
"tables.write "
"files.write "
"sharing.write"
)
@@ -545,7 +545,7 @@ async def nc_mcp_oauth_client_no_custom_scopes(
Connects to the OAuth-enabled MCP server on port 8001.
This client has only OIDC default scopes (openid, profile, email) without
application-specific scopes (notes:read, notes:write, etc.).
application-specific scopes (notes.read, notes.write, etc.).
Expected behavior: Should see 0 tools (all tools require custom scopes).
@@ -1671,7 +1671,7 @@ async def no_custom_scopes_oauth_client_credentials(
Fixture for OAuth client with NO custom scopes (only OIDC defaults).
Tests the security behavior when a user grants only the default OIDC scopes
(openid, profile, email) but declines custom application scopes (notes:read, notes:write, etc.).
(openid, profile, email) but declines custom application scopes (notes.read, notes.write, etc.).
The client is automatically deleted from Nextcloud after the test session completes.
@@ -1808,7 +1808,7 @@ async def playwright_oauth_token(
f"client_id={client_id}&"
f"redirect_uri={quote(callback_url, safe='')}&"
f"state={state}&"
f"scope=openid%20profile%20email%20notes:read%20notes:write%20calendar:read%20calendar:write%20contacts:read%20contacts:write%20cookbook:read%20cookbook:write%20deck:read%20deck:write%20tables:read%20tables:write%20files:read%20files:write%20sharing:read%20sharing:write"
f"scope=openid%20profile%20email%20notes.read%20notes.write%20calendar.read%20calendar.write%20contacts.read%20contacts.write%20cookbook.read%20cookbook.write%20deck.read%20deck.write%20tables.read%20tables.write%20files.read%20files.write%20sharing.read%20sharing.write"
)
# Add resource parameter (RFC 8707) if available
@@ -2060,7 +2060,7 @@ async def _get_oauth_token_with_scopes(
browser: Playwright browser instance
shared_oauth_client_credentials: Tuple of OAuth client credentials
oauth_callback_server: OAuth callback server fixture
scopes: Space-separated list of scopes (e.g., "openid profile email notes:read")
scopes: Space-separated list of scopes (e.g., "openid profile email notes.read")
resource: Optional resource parameter (RFC 8707) for token audience
mcp_server_base_url: Base URL of the MCP server for resource metadata discovery
@@ -2521,7 +2521,7 @@ async def _get_oauth_token_for_user(
f"redirect_uri={quote(callback_url, safe='')}&"
f"state={state}&"
f"resource={quote(mcp_server_resource, safe='')}&" # Resource URI from PRM
f"scope=openid%20profile%20email%20notes:read%20notes:write%20calendar:read%20calendar:write%20contacts:read%20contacts:write%20cookbook:read%20cookbook:write%20deck:read%20deck:write%20tables:read%20tables:write%20files:read%20files:write%20sharing:read%20sharing:write"
f"scope=openid%20profile%20email%20notes.read%20notes.write%20calendar.read%20calendar.write%20contacts.read%20contacts.write%20cookbook.read%20cookbook.write%20deck.read%20deck.write%20tables.read%20tables.write%20files.read%20files.write%20sharing.read%20sharing.write"
)
logger.info(f"Performing browser OAuth flow for {username}...")
@@ -3038,7 +3038,7 @@ async def configure_astrolabe_for_mcp_server(nc_client):
"--resource_url",
mcp_server_public_url,
"--allowed_scopes",
"openid profile email offline_access notes:read notes:write calendar:read calendar:write contacts:read contacts:write cookbook:read cookbook:write deck:read deck:write tables:read tables:write files:read files:write",
"openid profile email offline_access notes.read notes.write calendar.read calendar.write contacts.read contacts.write cookbook.read cookbook.write deck.read deck.write tables.read tables.write files.read files.write",
],
check=True,
capture_output=True,
tests/server/login_flow/test_dcr_lifecycle.py
+4 -4
View File
@@ -37,7 +37,7 @@ async def get_oauth_token_with_client(
authorization_endpoint: str,
callback_url: str,
auth_states: dict,
scopes: str = "openid profile email notes:read notes:write",
scopes: str = "openid profile email notes.read notes.write",
) -> str:
"""
Helper to obtain OAuth access token using existing client credentials.
@@ -187,7 +187,7 @@ async def test_dcr_register_and_delete_lifecycle(
"token_endpoint_auth_method": "client_secret_post",
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"scope": "openid profile email notes:read",
"scope": "openid profile email notes.read",
"token_type": "Bearer",
}
@@ -209,7 +209,7 @@ async def test_dcr_register_and_delete_lifecycle(
registration_endpoint=registration_endpoint,
client_name="DCR Lifecycle Test Client 2",
redirect_uris=[callback_url],
scopes="openid profile email notes:read",
scopes="openid profile email notes.read",
token_type="Bearer",
)
@@ -235,7 +235,7 @@ async def test_dcr_register_and_delete_lifecycle(
authorization_endpoint=authorization_endpoint,
callback_url=callback_url,
auth_states=auth_states,
scopes="openid profile email notes:read",
scopes="openid profile email notes.read",
)
assert access_token, "Failed to obtain access token"
tests/server/login_flow/test_dcr_token_type.py
+5 -5
View File
@@ -93,7 +93,7 @@ async def get_oauth_token_with_client(
authorization_endpoint: str,
callback_url: str,
auth_states: dict,
scopes: str = "openid profile email notes:read notes:write",
scopes: str = "openid profile email notes.read notes.write",
) -> str:
"""
Helper to obtain OAuth access token using existing client credentials.
@@ -241,7 +241,7 @@ async def test_dcr_respects_jwt_token_type(
registration_endpoint=registration_endpoint,
client_name="DCR Test - JWT Token Type",
redirect_uris=[callback_url],
scopes="openid profile email notes:read notes:write",
scopes="openid profile email notes.read notes.write",
token_type="jwt",
)
@@ -276,8 +276,8 @@ async def test_dcr_respects_jwt_token_type(
# Verify scope claim exists (critical for MCP tool filtering)
assert "scope" in payload, "JWT payload missing 'scope' claim"
scopes = payload["scope"].split()
assert "notes:read" in scopes, "JWT scope claim missing notes:read"
assert "notes:write" in scopes, "JWT scope claim missing notes:write"
assert "notes.read" in scopes, "JWT scope claim missing notes.read"
assert "notes.write" in scopes, "JWT scope claim missing notes.write"
logger.info(
f"✅ DCR with token_type=jwt works correctly! "
@@ -325,7 +325,7 @@ async def test_dcr_respects_bearer_token_type(
registration_endpoint=registration_endpoint,
client_name="DCR Test - Opaque Token Type",
redirect_uris=[callback_url],
scopes="openid profile email notes:read notes:write",
scopes="openid profile email notes.read notes.write",
token_type="opaque",
)
tests/server/login_flow/test_scope_authorization.py
+62 -62
View File
@@ -31,8 +31,8 @@ async def test_prm_endpoint():
prm_data = response.json()
assert prm_data["resource"] == "http://localhost:8004/mcp"
assert "notes:read" in prm_data["scopes_supported"]
assert "notes:write" in prm_data["scopes_supported"]
assert "notes.read" in prm_data["scopes_supported"]
assert "notes.write" in prm_data["scopes_supported"]
assert "http://localhost:8004" in prm_data["authorization_servers"]
assert "header" in prm_data["bearer_methods_supported"]
assert "RS256" in prm_data["resource_signing_alg_values_supported"]
@@ -67,7 +67,7 @@ async def test_read_only_token_filters_write_tools(nc_mcp_login_flow_client_read
logger = logging.getLogger(__name__)
# Connect with token that has only "notes:read" scope
# Connect with token that has only "notes.read" scope
result = await nc_mcp_login_flow_client_read_only.list_tools()
assert result is not None
assert len(result.tools) > 0
@@ -76,13 +76,13 @@ async def test_read_only_token_filters_write_tools(nc_mcp_login_flow_client_read
logger.info(f"Read-only token sees {len(tool_names)} tools")
# Verify read tools are present (only for apps with :read scopes)
# Read-only token has: notes:read, calendar:read, contacts:read,
# cookbook:read, deck:read, tables:read, files:read, sharing:read
# Read-only token has: notes.read, calendar.read, contacts.read,
# cookbook.read, deck.read, tables.read, files.read, sharing.read
expected_read_tools = [
"nc_notes_get_note", # notes:read
"nc_notes_search_notes", # notes:read
"nc_calendar_list_calendars", # calendar:read
"nc_calendar_get_event", # calendar:read
"nc_notes_get_note", # notes.read
"nc_notes_search_notes", # notes.read
"nc_calendar_list_calendars", # calendar.read
"nc_calendar_get_event", # calendar.read
]
for tool in expected_read_tools:
@@ -90,12 +90,12 @@ async def test_read_only_token_filters_write_tools(nc_mcp_login_flow_client_read
# Verify write tools are NOT present (filtered out)
write_tools_should_be_filtered = [
"nc_notes_create_note", # notes:write
"nc_notes_update_note", # notes:write
"nc_notes_delete_note", # notes:write
"nc_calendar_create_event", # calendar:write
"nc_calendar_update_event", # calendar:write
"nc_calendar_delete_event", # calendar:write
"nc_notes_create_note", # notes.write
"nc_notes_update_note", # notes.write
"nc_notes_delete_note", # notes.write
"nc_calendar_create_event", # calendar.write
"nc_calendar_update_event", # calendar.write
"nc_calendar_delete_event", # calendar.write
]
for tool in write_tools_should_be_filtered:
@@ -116,7 +116,7 @@ async def test_write_only_token_filters_read_tools(nc_mcp_login_flow_client_writ
logger = logging.getLogger(__name__)
# Connect with token that has only "notes:write" scope
# Connect with token that has only "notes.write" scope
result = await nc_mcp_login_flow_client_write_only.list_tools()
assert result is not None
assert len(result.tools) > 0
@@ -125,15 +125,15 @@ async def test_write_only_token_filters_read_tools(nc_mcp_login_flow_client_writ
logger.info(f"Write-only token sees {len(tool_names)} tools")
# Verify write tools are present
# Write-only token has: notes:write, calendar:write, contacts:write,
# cookbook:write, deck:write, tables:write, files:write, sharing:write
# Write-only token has: notes.write, calendar.write, contacts.write,
# cookbook.write, deck.write, tables.write, files.write, sharing.write
expected_write_tools = [
"nc_notes_create_note", # notes:write
"nc_notes_update_note", # notes:write
"nc_notes_delete_note", # notes:write
"nc_calendar_create_event", # calendar:write
"nc_calendar_update_event", # calendar:write
"nc_calendar_delete_event", # calendar:write
"nc_notes_create_note", # notes.write
"nc_notes_update_note", # notes.write
"nc_notes_delete_note", # notes.write
"nc_calendar_create_event", # calendar.write
"nc_calendar_update_event", # calendar.write
"nc_calendar_delete_event", # calendar.write
]
for tool in expected_write_tools:
@@ -141,10 +141,10 @@ async def test_write_only_token_filters_read_tools(nc_mcp_login_flow_client_writ
# Verify read-only tools are NOT present (write-only scope)
read_tools_should_be_filtered = [
"nc_notes_get_note", # notes:read
"nc_notes_search_notes", # notes:read
"nc_calendar_list_calendars", # calendar:read
"nc_calendar_get_event", # calendar:read
"nc_notes_get_note", # notes.read
"nc_notes_search_notes", # notes.read
"nc_calendar_list_calendars", # calendar.read
"nc_calendar_get_event", # calendar.read
]
for tool in read_tools_should_be_filtered:
@@ -165,7 +165,7 @@ async def test_full_access_token_shows_all_tools(nc_mcp_login_flow_client_full_a
logger = logging.getLogger(__name__)
# Connect with token that has both "notes:read" and "notes:write" scopes
# Connect with token that has both "notes.read" and "notes.write" scopes
result = await nc_mcp_login_flow_client_full_access.list_tools()
assert result is not None
assert len(result.tools) > 0
@@ -177,14 +177,14 @@ async def test_full_access_token_shows_all_tools(nc_mcp_login_flow_client_full_a
# Verify both read and write tools are present
# Full access has all *read and *write scopes
expected_read_tools = [
"nc_notes_get_note", # notes:read
"nc_notes_search_notes", # notes:read
"nc_calendar_list_calendars", # calendar:read
"nc_notes_get_note", # notes.read
"nc_notes_search_notes", # notes.read
"nc_calendar_list_calendars", # calendar.read
]
expected_write_tools = [
"nc_notes_create_note", # notes:write
"nc_calendar_create_event", # calendar:write
"nc_notes_create_note", # notes.write
"nc_calendar_create_event", # calendar.write
]
for tool in expected_read_tools:
@@ -217,17 +217,17 @@ async def test_scope_helper_functions():
pass
# Add scope metadata
mock_read_tool._required_scopes = ["notes:read"] # type: ignore
mock_write_tool._required_scopes = ["notes:write"] # type: ignore
mock_read_tool._required_scopes = ["notes.read"] # type: ignore
mock_write_tool._required_scopes = ["notes.write"] # type: ignore
# Test get_required_scopes
assert get_required_scopes(mock_read_tool) == ["notes:read"]
assert get_required_scopes(mock_write_tool) == ["notes:write"]
assert get_required_scopes(mock_read_tool) == ["notes.read"]
assert get_required_scopes(mock_write_tool) == ["notes.write"]
assert get_required_scopes(mock_no_scope_tool) == []
# Test has_required_scopes
read_only_scopes = {"notes:read"}
full_scopes = {"notes:read", "notes:write"}
read_only_scopes = {"notes.read"}
full_scopes = {"notes.read", "notes.write"}
no_scopes = set()
# User with only read scope
@@ -251,13 +251,13 @@ async def test_scope_decorator_stores_metadata():
"""Test that @require_scopes decorator properly stores metadata."""
from nextcloud_mcp_server.auth import require_scopes
@require_scopes("notes:read", "notes:write")
@require_scopes("notes.read", "notes.write")
async def test_function():
pass
# Check that metadata was stored
assert hasattr(test_function, "_required_scopes")
assert test_function._required_scopes == ["notes:read", "notes:write"]
assert test_function._required_scopes == ["notes.read", "notes.write"]
@pytest.mark.integration
@@ -308,28 +308,28 @@ async def test_scope_classification():
from scripts.add_scope_decorators_simple import classify_function
# Test read operations
assert classify_function("nc_notes_get_note") == "notes:read"
assert classify_function("nc_notes_search_notes") == "notes:read"
assert classify_function("nc_calendar_list_events") == "calendar:read"
assert classify_function("nc_webdav_read_file") == "files:read"
assert classify_function("nc_calendar_find_availability") == "calendar:read"
assert classify_function("nc_calendar_get_upcoming_events") == "notes:read"
assert classify_function("nc_notes_get_note") == "notes.read"
assert classify_function("nc_notes_search_notes") == "notes.read"
assert classify_function("nc_calendar_list_events") == "calendar.read"
assert classify_function("nc_webdav_read_file") == "files.read"
assert classify_function("nc_calendar_find_availability") == "calendar.read"
assert classify_function("nc_calendar_get_upcoming_events") == "notes.read"
# Test write operations
assert classify_function("nc_notes_create_note") == "notes:write"
assert classify_function("nc_notes_update_note") == "notes:write"
assert classify_function("nc_notes_delete_note") == "notes:write"
assert classify_function("nc_notes_append_content") == "notes:write"
assert classify_function("nc_calendar_create_event") == "calendar:write"
assert classify_function("nc_calendar_update_event") == "notes:write"
assert classify_function("nc_calendar_manage_calendar") == "notes:write"
assert classify_function("nc_webdav_write_file") == "files:write"
assert classify_function("nc_webdav_move_resource") == "notes:write"
assert classify_function("nc_contacts_create_contact") == "notes:write"
assert classify_function("nc_cookbook_import_recipe") == "notes:write"
assert classify_function("nc_tables_insert_row") == "notes:write"
assert classify_function("deck_archive_card") == "notes:write"
assert classify_function("deck_assign_label_to_card") == "notes:write"
assert classify_function("nc_notes_create_note") == "notes.write"
assert classify_function("nc_notes_update_note") == "notes.write"
assert classify_function("nc_notes_delete_note") == "notes.write"
assert classify_function("nc_notes_append_content") == "notes.write"
assert classify_function("nc_calendar_create_event") == "calendar.write"
assert classify_function("nc_calendar_update_event") == "notes.write"
assert classify_function("nc_calendar_manage_calendar") == "notes.write"
assert classify_function("nc_webdav_write_file") == "files.write"
assert classify_function("nc_webdav_move_resource") == "notes.write"
assert classify_function("nc_contacts_create_contact") == "notes.write"
assert classify_function("nc_cookbook_import_recipe") == "notes.write"
assert classify_function("nc_tables_insert_row") == "notes.write"
assert classify_function("deck_archive_card") == "notes.write"
assert classify_function("deck_assign_label_to_card") == "notes.write"
@pytest.mark.skip(reason="Script no longer exists - decorators are already in place")
tests/test_cli.py
+9 -9
View File
@@ -251,15 +251,15 @@ def test_default_values(runner, clean_env, monkeypatch):
# Verify default values
assert captured_env["NEXTCLOUD_OIDC_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"
"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"
)
assert captured_env["NEXTCLOUD_OIDC_TOKEN_TYPE"] == "bearer"
assert captured_env["NEXTCLOUD_MCP_SERVER_URL"] == "http://localhost:8000"
tests/unit/api/test_access.py
+8 -8
View File
@@ -100,7 +100,7 @@ class TestGetUserAccess:
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="test-app-pw",
scopes=["notes:read", "calendar:write"],
scopes=["notes.read", "calendar.write"],
username="alice_nc",
)
@@ -115,7 +115,7 @@ class TestGetUserAccess:
data = resp.json()
assert data["success"] is True
assert data["provisioned"] is True
assert set(data["scopes"]) == {"notes:read", "calendar:write"}
assert set(data["scopes"]) == {"notes.read", "calendar.write"}
assert data["username"] == "alice_nc"
async def test_missing_auth_header(self, temp_storage):
@@ -146,7 +146,7 @@ class TestUpdateUserScopes:
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="test-app-pw",
scopes=["notes:read"],
scopes=["notes.read"],
username="alice_nc",
)
@@ -156,19 +156,19 @@ class TestUpdateUserScopes:
resp = client.patch(
"/api/v1/users/alice/scopes",
headers={"Authorization": create_basic_auth_header("alice", "pw")},
json={"scopes": ["notes:read", "notes:write", "calendar:read"]},
json={"scopes": ["notes.read", "notes.write", "calendar.read"]},
)
assert resp.status_code == 200
data = resp.json()
assert data["success"] is True
assert set(data["scopes"]) == {"notes:read", "notes:write", "calendar:read"}
assert set(data["scopes"]) == {"notes.read", "notes.write", "calendar.read"}
async def test_invalid_scopes(self, temp_storage):
"""Returns 400 for invalid scope names."""
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="test-app-pw",
scopes=["notes:read"],
scopes=["notes.read"],
)
app = create_test_app(temp_storage)
@@ -177,7 +177,7 @@ class TestUpdateUserScopes:
resp = client.patch(
"/api/v1/users/alice/scopes",
headers={"Authorization": create_basic_auth_header("alice", "pw")},
json={"scopes": ["notes:read", "invalid:scope"]},
json={"scopes": ["notes.read", "invalid:scope"]},
)
assert resp.status_code == 400
data = resp.json()
@@ -192,7 +192,7 @@ class TestUpdateUserScopes:
resp = client.patch(
"/api/v1/users/alice/scopes",
headers={"Authorization": create_basic_auth_header("alice", "pw")},
json={"scopes": ["notes:read"]},
json={"scopes": ["notes.read"]},
)
assert resp.status_code == 404
data = resp.json()
tests/unit/test_auth_tools.py
+12 -12
View File
@@ -38,14 +38,14 @@ async def test_store_app_password_with_scopes(temp_storage):
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="aaaaa-bbbbb-ccccc-ddddd-eeeee",
scopes=["notes:read", "notes:write"],
scopes=["notes.read", "notes.write"],
username="alice_nc",
)
data = await temp_storage.get_app_password_with_scopes("alice")
assert data is not None
assert data["app_password"] == "aaaaa-bbbbb-ccccc-ddddd-eeeee"
assert data["scopes"] == ["notes:read", "notes:write"]
assert data["scopes"] == ["notes.read", "notes.write"]
assert data["username"] == "alice_nc"
assert data["created_at"] is not None
assert data["updated_at"] is not None
@@ -70,18 +70,18 @@ async def test_store_app_password_with_scopes_replaces(temp_storage):
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="aaaaa-bbbbb-ccccc-ddddd-eeeee",
scopes=["notes:read"],
scopes=["notes.read"],
)
await temp_storage.store_app_password_with_scopes(
user_id="alice",
app_password="xxxxx-yyyyy-zzzzz-aaaaa-bbbbb",
scopes=["notes:read", "calendar:read"],
scopes=["notes.read", "calendar.read"],
username="alice_nc",
)
data = await temp_storage.get_app_password_with_scopes("alice")
assert data["app_password"] == "xxxxx-yyyyy-zzzzz-aaaaa-bbbbb"
assert data["scopes"] == ["notes:read", "calendar:read"]
assert data["scopes"] == ["notes.read", "calendar.read"]
async def test_get_app_password_with_scopes_nonexistent(temp_storage):
@@ -99,14 +99,14 @@ async def test_store_and_get_login_flow_session(temp_storage):
user_id="alice",
poll_token="secret-poll-token",
poll_endpoint="https://cloud.example.com/login/v2/poll",
requested_scopes=["notes:read", "notes:write"],
requested_scopes=["notes.read", "notes.write"],
)
session = await temp_storage.get_login_flow_session("alice")
assert session is not None
assert session["poll_token"] == "secret-poll-token"
assert session["poll_endpoint"] == "https://cloud.example.com/login/v2/poll"
assert session["requested_scopes"] == ["notes:read", "notes:write"]
assert session["requested_scopes"] == ["notes.read", "notes.write"]
assert session["created_at"] is not None
assert session["expires_at"] is not None
@@ -187,11 +187,11 @@ async def test_delete_expired_login_flow_sessions(temp_storage):
def test_all_supported_scopes():
"""Test that ALL_SUPPORTED_SCOPES contains expected scopes."""
assert "notes:read" in ALL_SUPPORTED_SCOPES
assert "notes:write" in ALL_SUPPORTED_SCOPES
assert "calendar:read" in ALL_SUPPORTED_SCOPES
assert "files:read" in ALL_SUPPORTED_SCOPES
assert "deck:read" in ALL_SUPPORTED_SCOPES
assert "notes.read" in ALL_SUPPORTED_SCOPES
assert "notes.write" in ALL_SUPPORTED_SCOPES
assert "calendar.read" in ALL_SUPPORTED_SCOPES
assert "files.read" in ALL_SUPPORTED_SCOPES
assert "deck.read" in ALL_SUPPORTED_SCOPES
# Scopes should be in pairs (read/write)
read_scopes = [s for s in ALL_SUPPORTED_SCOPES if s.endswith(":read")]
write_scopes = [s for s in ALL_SUPPORTED_SCOPES if s.endswith(":write")]
tests/unit/test_scope_authorization_stored.py
+2 -2
View File
@@ -29,7 +29,7 @@ async def test_get_stored_scopes_with_scopes():
mock_storage = AsyncMock()
mock_storage.get_app_password_with_scopes.return_value = {
"app_password": "xxxxx",
"scopes": ["notes:read", "calendar:read"],
"scopes": ["notes.read", "calendar.read"],
"username": "alice",
"created_at": 1000,
"updated_at": 1000,
@@ -41,7 +41,7 @@ async def test_get_stored_scopes_with_scopes():
):
result = await _get_stored_scopes("alice")
assert result == ["notes:read", "calendar:read"]
assert result == ["notes.read", "calendar.read"]
async def test_get_stored_scopes_null_scopes():
tests/unit/test_scope_decorator.py
+8 -8
View File
@@ -12,24 +12,24 @@ from nextcloud_mcp_server.auth.scope_authorization import (
def test_scope_decorator_stores_metadata():
"""Test that @require_scopes decorator stores scope requirements as function metadata."""
@require_scopes("notes:read", "notes:write")
@require_scopes("notes.read", "notes.write")
async def example_function():
pass
# Verify metadata is stored
assert hasattr(example_function, "_required_scopes")
assert example_function._required_scopes == ["notes:read", "notes:write"]
assert example_function._required_scopes == ["notes.read", "notes.write"]
@pytest.mark.unit
def test_scope_decorator_with_single_scope():
"""Test decorator with a single scope requirement."""
@require_scopes("calendar:read")
@require_scopes("calendar.read")
async def example_function():
pass
assert example_function._required_scopes == ["calendar:read"]
assert example_function._required_scopes == ["calendar.read"]
@pytest.mark.unit
@@ -46,18 +46,18 @@ def test_scope_decorator_with_no_scopes():
@pytest.mark.unit
def test_insufficient_scope_error():
"""Test InsufficientScopeError exception structure."""
missing = ["notes:write", "calendar:write"]
missing = ["notes.write", "calendar.write"]
error = InsufficientScopeError(missing)
assert error.missing_scopes == missing
assert "notes:write" in str(error)
assert "calendar:write" in str(error)
assert "notes.write" in str(error)
assert "calendar.write" in str(error)
@pytest.mark.unit
def test_insufficient_scope_error_with_custom_message():
"""Test InsufficientScopeError with custom message."""
missing = ["files:write"]
missing = ["files.write"]
custom_msg = "You need more permissions"
error = InsufficientScopeError(missing, custom_msg)
tests/unit/test_token_broker.py
+3 -3
View File
@@ -379,7 +379,7 @@ class TestRefreshTokenRotation:
expires_in,
) = await broker._refresh_access_token_with_scopes(
refresh_token="old_refresh_token_123",
required_scopes=["notes:read"],
required_scopes=["notes.read"],
user_id="admin",
)
@@ -424,7 +424,7 @@ class TestRefreshTokenRotation:
):
await broker._refresh_access_token_with_scopes(
refresh_token="same_refresh_token",
required_scopes=["notes:read"],
required_scopes=["notes.read"],
user_id="admin",
)
@@ -460,7 +460,7 @@ class TestRefreshTokenRotation:
):
await broker._refresh_access_token_with_scopes(
refresh_token="old_token",
required_scopes=["notes:read"],
required_scopes=["notes.read"],
user_id=None, # No user_id
)