c7d3ec57c3
Commitd2217531(google-labs-jules SSRF fix) catastrophically deleted 11,500+ lines across 100+ files while fixing an unrelated IMAP issue. Restored from d2217531^ (pre-bad-commit state): Deleted files (fully restored): - app/api/{automation,classification_rules,comments,sharing}.py - app/middleware/upload_rate_limit.py - app/tasks/{automation_tasks,classify_document}.py - app/utils/{automation_hooks,classification_rules}.py - docs/AppleAppStoreCompliance.md - frontend/input.css, package.json, package-lock.json, tailwind.config.js - frontend/static/js/{annotations,claim,comments,sharing}.js - frontend/templates/{admin_connections,file_annotations,file_summary}.html - tests/{test_api_files_comprehensive,test_auth_extended,test_sharing, test_comments,test_connections,test_imap_profiles,test_api_sessions, test_automation,test_classification_rules,test_api_advanced_filters, test_api_classification_rules,test_upload_rate_limit,test_api_dropbox, test_classify_document,test_comments_ui,test_upload_to_icloud, test_api_onedrive_comprehensive,test_frontend_build,test_sentry, test_diagnostic,test_database,test_views_dropbox,test_local_auth}.py Truncated files (content restored): - app/{auth,config,main,models,celery_worker,database}.py - app/api/{__init__,api_tokens,diagnostic,dropbox,files,google_drive, integrations,local_auth,mobile,onedrive,pipelines,qr_auth, settings,url_upload}.py - app/middleware/upload_rate_limit.py - app/tasks/upload_to_nextcloud.py - app/utils/{allowed_types,settings_service,settings_sync,user_scope,webhook}.py - app/views/{base,dropbox,files,google_drive,onedrive,settings}.py - docs/{API,AuthenticationSetup,ConfigurationGuide,DatabaseConfiguration, DeploymentGuide,DropboxSetup,GoogleDriveSetup,KubernetesDeployment, MobileApp,OneDriveSetup,ProductionReadiness,SentrySetup, SocialLoginSetup,UserGuide}.md - frontend/static/{js/upload.js,styles.css} - frontend/templates/{api_tokens,base,devices,dropbox,dropbox_callback, file_view,files,google_drive,onedrive,onedrive_callback, signup}.html - frontend/translations/en.json - migrations/env.py - tests/{conftest,test_api_integrations,test_api_mobile,test_api_settings, test_api_tokens,test_audit_logs,test_duplicates,test_imap_tasks, test_setup_wizard,test_views_files_comprehensive}.py Security fixes kept from post-d2217531 commits: - app/utils/network.py: DNS SSRF fail-secure fix (06b0fced) - app/utils/file_operations.py: path traversal fix (1018ea17) - tests/test_imap_tasks.py: re-applied 4 is_private_ip mock patches Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> Agent-Logs-Url: https://github.com/christianlouis/DocuElevate/sessions/51133dd8-9bec-41ab-aa10-3de753634187
495 lines
20 KiB
Python
495 lines
20 KiB
Python
"""Tests for the Connections admin page and new authentication providers."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import status
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestDropboxComplianceFix:
|
|
"""Tests for the Dropbox userinfo compliance fix."""
|
|
|
|
def test_dropbox_compliance_fix_adds_sub(self):
|
|
"""Test that compliance fix adds 'sub' from account_id."""
|
|
from app.auth import _dropbox_userinfo_compliance_fix
|
|
|
|
data = {"account_id": "dbid:abc123", "email": "test@example.com"}
|
|
result = _dropbox_userinfo_compliance_fix(None, None, None, data)
|
|
assert result["sub"] == "dbid:abc123"
|
|
|
|
def test_dropbox_compliance_fix_does_not_overwrite_sub(self):
|
|
"""Test that compliance fix preserves existing 'sub'."""
|
|
from app.auth import _dropbox_userinfo_compliance_fix
|
|
|
|
data = {"account_id": "dbid:abc123", "sub": "existing-sub"}
|
|
result = _dropbox_userinfo_compliance_fix(None, None, None, data)
|
|
assert result["sub"] == "existing-sub"
|
|
|
|
def test_dropbox_compliance_fix_normalizes_name(self):
|
|
"""Test that compliance fix normalizes nested name object."""
|
|
from app.auth import _dropbox_userinfo_compliance_fix
|
|
|
|
data = {"name": {"display_name": "John Doe", "given_name": "John"}}
|
|
result = _dropbox_userinfo_compliance_fix(None, None, None, data)
|
|
assert result["name"] == "John Doe"
|
|
|
|
def test_dropbox_compliance_fix_handles_no_name(self):
|
|
"""Test compliance fix works when no name is present."""
|
|
from app.auth import _dropbox_userinfo_compliance_fix
|
|
|
|
data = {"email": "test@example.com"}
|
|
result = _dropbox_userinfo_compliance_fix(None, None, None, data)
|
|
assert "email" in result
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGitHubNormalization:
|
|
"""Tests for GitHub userinfo normalization."""
|
|
|
|
def test_github_normalize_standard(self):
|
|
"""Test GitHub userinfo normalization with standard response."""
|
|
from app.auth import _normalize_social_userinfo
|
|
|
|
raw = {
|
|
"id": 12345,
|
|
"login": "octocat",
|
|
"name": "The Octocat",
|
|
"email": "octocat@github.com",
|
|
"avatar_url": "https://avatars.githubusercontent.com/u/12345",
|
|
}
|
|
result = _normalize_social_userinfo("github", {}, raw)
|
|
assert result["sub"] == "12345"
|
|
assert result["email"] == "octocat@github.com"
|
|
assert result["name"] == "The Octocat"
|
|
assert result["preferred_username"] == "octocat"
|
|
assert result["picture"] == "https://avatars.githubusercontent.com/u/12345"
|
|
|
|
def test_github_normalize_no_name_uses_login(self):
|
|
"""Test GitHub normalization falls back to login when name is empty."""
|
|
from app.auth import _normalize_social_userinfo
|
|
|
|
raw = {"id": 12345, "login": "octocat", "name": "", "email": "octocat@github.com"}
|
|
result = _normalize_social_userinfo("github", {}, raw)
|
|
assert result["name"] == "octocat"
|
|
|
|
def test_github_normalize_missing_fields(self):
|
|
"""Test GitHub normalization handles missing fields gracefully."""
|
|
from app.auth import _normalize_social_userinfo
|
|
|
|
result = _normalize_social_userinfo("github", {}, {})
|
|
assert result["sub"] == ""
|
|
assert result["email"] == ""
|
|
assert result["name"] == ""
|
|
assert result["preferred_username"] == ""
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSSOAutoLogin:
|
|
"""Tests for SSO Auto Login configuration."""
|
|
|
|
def test_sso_auto_login_default_false(self):
|
|
"""Test that SSO auto login defaults to False."""
|
|
from app.config import Settings
|
|
|
|
s = Settings(
|
|
_env_file=None,
|
|
auth_enabled=True,
|
|
)
|
|
assert s.sso_auto_login is False
|
|
|
|
def test_sso_auto_login_can_be_enabled(self):
|
|
"""Test that SSO auto login can be set to True."""
|
|
from app.config import Settings
|
|
|
|
s = Settings(
|
|
_env_file=None,
|
|
auth_enabled=True,
|
|
sso_auto_login=True,
|
|
)
|
|
assert s.sso_auto_login is True
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestNewConfigFields:
|
|
"""Tests for new configuration fields."""
|
|
|
|
def test_github_config_defaults(self):
|
|
"""Test GitHub social auth config defaults."""
|
|
from app.config import Settings
|
|
|
|
s = Settings(_env_file=None, auth_enabled=True)
|
|
assert s.social_auth_github_enabled is False
|
|
assert s.social_auth_github_client_id is None
|
|
assert s.social_auth_github_client_secret is None
|
|
|
|
def test_keycloak_config_defaults(self):
|
|
"""Test Keycloak social auth config defaults."""
|
|
from app.config import Settings
|
|
|
|
s = Settings(_env_file=None, auth_enabled=True)
|
|
assert s.social_auth_keycloak_enabled is False
|
|
assert s.social_auth_keycloak_client_id is None
|
|
assert s.social_auth_keycloak_server_url is None
|
|
assert s.social_auth_keycloak_realm is None
|
|
|
|
def test_generic_oauth2_config_defaults(self):
|
|
"""Test Generic OAuth2 config defaults."""
|
|
from app.config import Settings
|
|
|
|
s = Settings(_env_file=None, auth_enabled=True)
|
|
assert s.social_auth_generic_oauth2_enabled is False
|
|
assert s.social_auth_generic_oauth2_scope == "openid profile email"
|
|
assert s.social_auth_generic_oauth2_name == "OAuth2"
|
|
|
|
def test_saml2_config_defaults(self):
|
|
"""Test SAML2 config defaults."""
|
|
from app.config import Settings
|
|
|
|
s = Settings(_env_file=None, auth_enabled=True)
|
|
assert s.social_auth_saml2_enabled is False
|
|
assert s.social_auth_saml2_name == "SAML2"
|
|
|
|
def test_telegram_config_defaults(self):
|
|
"""Test Telegram config defaults."""
|
|
from app.config import Settings
|
|
|
|
s = Settings(_env_file=None, auth_enabled=True)
|
|
assert s.telegram_enabled is False
|
|
assert s.telegram_bot_token is None
|
|
assert s.telegram_chat_id is None
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSettingsMetadata:
|
|
"""Tests that new settings have metadata entries."""
|
|
|
|
def test_github_settings_have_metadata(self):
|
|
"""Test GitHub settings are in SETTING_METADATA."""
|
|
from app.utils.settings_service import SETTING_METADATA
|
|
|
|
assert "social_auth_github_enabled" in SETTING_METADATA
|
|
assert "social_auth_github_client_id" in SETTING_METADATA
|
|
assert "social_auth_github_client_secret" in SETTING_METADATA
|
|
|
|
def test_keycloak_settings_have_metadata(self):
|
|
"""Test Keycloak settings are in SETTING_METADATA."""
|
|
from app.utils.settings_service import SETTING_METADATA
|
|
|
|
assert "social_auth_keycloak_enabled" in SETTING_METADATA
|
|
assert "social_auth_keycloak_client_id" in SETTING_METADATA
|
|
assert "social_auth_keycloak_server_url" in SETTING_METADATA
|
|
assert "social_auth_keycloak_realm" in SETTING_METADATA
|
|
|
|
def test_generic_oauth2_settings_have_metadata(self):
|
|
"""Test Generic OAuth2 settings are in SETTING_METADATA."""
|
|
from app.utils.settings_service import SETTING_METADATA
|
|
|
|
assert "social_auth_generic_oauth2_enabled" in SETTING_METADATA
|
|
assert "social_auth_generic_oauth2_authorize_url" in SETTING_METADATA
|
|
assert "social_auth_generic_oauth2_token_url" in SETTING_METADATA
|
|
|
|
def test_saml2_settings_have_metadata(self):
|
|
"""Test SAML2 settings are in SETTING_METADATA."""
|
|
from app.utils.settings_service import SETTING_METADATA
|
|
|
|
assert "social_auth_saml2_enabled" in SETTING_METADATA
|
|
assert "social_auth_saml2_sso_url" in SETTING_METADATA
|
|
assert "social_auth_saml2_entity_id" in SETTING_METADATA
|
|
|
|
def test_telegram_settings_have_metadata(self):
|
|
"""Test Telegram settings are in SETTING_METADATA."""
|
|
from app.utils.settings_service import SETTING_METADATA
|
|
|
|
assert "telegram_enabled" in SETTING_METADATA
|
|
assert "telegram_bot_token" in SETTING_METADATA
|
|
assert "telegram_chat_id" in SETTING_METADATA
|
|
|
|
def test_sso_auto_login_has_metadata(self):
|
|
"""Test SSO auto login has metadata."""
|
|
from app.utils.settings_service import SETTING_METADATA
|
|
|
|
assert "sso_auto_login" in SETTING_METADATA
|
|
meta = SETTING_METADATA["sso_auto_login"]
|
|
assert meta["category"] == "Authentication"
|
|
assert meta["type"] == "boolean"
|
|
|
|
def test_github_category_is_social_login(self):
|
|
"""Test GitHub settings are in Social Login category."""
|
|
from app.utils.settings_service import SETTING_METADATA
|
|
|
|
assert SETTING_METADATA["social_auth_github_enabled"]["category"] == "Social Login"
|
|
|
|
def test_github_secret_is_sensitive(self):
|
|
"""Test GitHub client secret is marked sensitive."""
|
|
from app.utils.settings_service import SETTING_METADATA
|
|
|
|
assert SETTING_METADATA["social_auth_github_client_secret"]["sensitive"] is True
|
|
|
|
def test_github_has_help_link(self):
|
|
"""Test GitHub has a help link to developer settings."""
|
|
from app.utils.settings_service import SETTING_METADATA
|
|
|
|
assert "help_link" in SETTING_METADATA["social_auth_github_enabled"]
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestConnectionsPageRoute:
|
|
"""Tests for the /admin/connections route."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connections_page_non_admin_redirected(self):
|
|
"""Test that non-admin users are redirected from connections page."""
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.session = {"user": {"is_admin": False}}
|
|
mock_db = MagicMock()
|
|
|
|
# The require_admin_access decorator should handle this, so we test the decorator
|
|
from app.views.settings import require_admin_access
|
|
|
|
@require_admin_access
|
|
async def dummy_view(request):
|
|
return "success"
|
|
|
|
result = await dummy_view(mock_request)
|
|
assert result.status_code == status.HTTP_302_FOUND
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connections_page_returns_services(self):
|
|
"""Test that connections page includes expected services in context."""
|
|
from app.views.settings import connections_page
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.session = {"user": {"is_admin": True}}
|
|
mock_db = MagicMock()
|
|
|
|
with (
|
|
patch("app.views.settings.get_all_settings_from_db", return_value={}),
|
|
patch("app.views.settings.templates") as mock_templates,
|
|
patch("app.views.settings.SETTING_METADATA", {}),
|
|
patch("app.views.settings.get_setting_metadata", return_value={}),
|
|
):
|
|
mock_templates.TemplateResponse.return_value = "response"
|
|
result = await connections_page(mock_request, db=mock_db)
|
|
|
|
# Check TemplateResponse was called
|
|
mock_templates.TemplateResponse.assert_called_once()
|
|
call_args = mock_templates.TemplateResponse.call_args
|
|
template_name = call_args[0][0]
|
|
context = call_args[0][1]
|
|
|
|
assert template_name == "admin_connections.html"
|
|
assert "services" in context
|
|
assert "service_settings" in context
|
|
assert "sso_auto_login" in context
|
|
|
|
# Verify expected service keys
|
|
service_keys = [s["key"] for s in context["services"]]
|
|
assert "google" in service_keys
|
|
assert "github" in service_keys
|
|
assert "keycloak" in service_keys
|
|
assert "generic_oauth2" in service_keys
|
|
assert "saml2" in service_keys
|
|
assert "smtp" in service_keys
|
|
assert "telegram" in service_keys
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connections_page_linked_status_from_db(self):
|
|
"""Linked status is derived from DB/effective settings, not SOCIAL_PROVIDERS."""
|
|
from app.views.settings import connections_page
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.session = {"user": {"is_admin": True}}
|
|
mock_db = MagicMock()
|
|
|
|
# Simulate GitHub configured only in DB (not in SOCIAL_PROVIDERS yet)
|
|
db_values = {
|
|
"social_auth_github_enabled": "true",
|
|
"social_auth_github_client_id": "gh-id",
|
|
"social_auth_github_client_secret": "gh-secret",
|
|
}
|
|
|
|
with (
|
|
patch("app.views.settings.get_all_settings_from_db", return_value=db_values),
|
|
patch("app.views.settings.templates") as mock_templates,
|
|
patch("app.views.settings.SETTING_METADATA", {}),
|
|
patch("app.views.settings.get_setting_metadata", return_value={}),
|
|
):
|
|
mock_templates.TemplateResponse.return_value = "response"
|
|
await connections_page(mock_request, db=mock_db)
|
|
|
|
context = mock_templates.TemplateResponse.call_args[0][1]
|
|
services_by_key = {s["key"]: s for s in context["services"]}
|
|
|
|
# GitHub should be linked because DB values say so
|
|
assert services_by_key["github"]["linked"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connections_page_unlinked_when_credentials_missing(self):
|
|
"""Provider is unlinked when enabled=true but credentials are absent."""
|
|
from app.views.settings import connections_page
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.session = {"user": {"is_admin": True}}
|
|
mock_db = MagicMock()
|
|
|
|
# enabled but no credentials
|
|
db_values = {"social_auth_github_enabled": "true"}
|
|
|
|
with (
|
|
patch("app.views.settings.get_all_settings_from_db", return_value=db_values),
|
|
patch("app.views.settings.templates") as mock_templates,
|
|
patch("app.views.settings.SETTING_METADATA", {}),
|
|
patch("app.views.settings.get_setting_metadata", return_value={}),
|
|
):
|
|
mock_templates.TemplateResponse.return_value = "response"
|
|
await connections_page(mock_request, db=mock_db)
|
|
|
|
context = mock_templates.TemplateResponse.call_args[0][1]
|
|
services_by_key = {s["key"]: s for s in context["services"]}
|
|
assert services_by_key["github"]["linked"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connections_page_oidc_linked_from_db(self):
|
|
"""OIDC linked status derives from DB effective settings."""
|
|
from app.views.settings import connections_page
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.session = {"user": {"is_admin": True}}
|
|
mock_db = MagicMock()
|
|
|
|
db_values = {
|
|
"authentik_client_id": "my-client-id",
|
|
"authentik_client_secret": "my-secret",
|
|
"oauth_provider_name": "My SSO",
|
|
}
|
|
|
|
with (
|
|
patch("app.views.settings.get_all_settings_from_db", return_value=db_values),
|
|
patch("app.views.settings.templates") as mock_templates,
|
|
patch("app.views.settings.SETTING_METADATA", {}),
|
|
patch("app.views.settings.get_setting_metadata", return_value={}),
|
|
):
|
|
mock_templates.TemplateResponse.return_value = "response"
|
|
await connections_page(mock_request, db=mock_db)
|
|
|
|
context = mock_templates.TemplateResponse.call_args[0][1]
|
|
services_by_key = {s["key"]: s for s in context["services"]}
|
|
assert services_by_key["oidc"]["linked"] is True
|
|
assert services_by_key["oidc"]["name"] == "My SSO"
|
|
# oauth_configured template var should also reflect the DB state
|
|
assert context["oauth_configured"] is True
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestRefreshSocialProviders:
|
|
"""Tests for the refresh_social_providers() mechanism."""
|
|
|
|
def test_refresh_social_providers_exists(self):
|
|
"""refresh_social_providers is importable from app.auth."""
|
|
from app.auth import refresh_social_providers
|
|
|
|
assert callable(refresh_social_providers)
|
|
|
|
def test_refresh_social_providers_clears_and_repopulates(self):
|
|
"""After refresh, SOCIAL_PROVIDERS reflects current settings."""
|
|
import app.auth as auth_module
|
|
|
|
with (
|
|
patch.object(auth_module, "AUTH_ENABLED", True),
|
|
patch.object(auth_module, "settings") as mock_settings,
|
|
):
|
|
mock_settings.authentik_client_id = None
|
|
mock_settings.authentik_client_secret = None
|
|
mock_settings.social_auth_google_enabled = True
|
|
mock_settings.social_auth_google_client_id = "gid"
|
|
mock_settings.social_auth_google_client_secret = "gsecret"
|
|
mock_settings.social_auth_google_use_global_credentials = False
|
|
# All other providers disabled
|
|
for attr in (
|
|
"social_auth_microsoft_enabled",
|
|
"social_auth_apple_enabled",
|
|
"social_auth_dropbox_enabled",
|
|
"social_auth_github_enabled",
|
|
"social_auth_keycloak_enabled",
|
|
"social_auth_generic_oauth2_enabled",
|
|
):
|
|
setattr(mock_settings, attr, False)
|
|
|
|
with patch.object(auth_module, "_register_oauth_client"):
|
|
auth_module._setup_social_providers()
|
|
|
|
assert "google" in auth_module.SOCIAL_PROVIDERS
|
|
assert auth_module.OAUTH_CONFIGURED is False
|
|
|
|
def test_refresh_clears_previous_providers(self):
|
|
"""Providers removed from settings are cleared after refresh."""
|
|
import app.auth as auth_module
|
|
|
|
# Pre-populate with a stale entry
|
|
auth_module.SOCIAL_PROVIDERS["stale_provider"] = {"name": "Stale", "icon": "", "color": ""}
|
|
|
|
with (
|
|
patch.object(auth_module, "AUTH_ENABLED", True),
|
|
patch.object(auth_module, "settings") as mock_settings,
|
|
):
|
|
mock_settings.authentik_client_id = None
|
|
mock_settings.authentik_client_secret = None
|
|
for attr in (
|
|
"social_auth_google_enabled",
|
|
"social_auth_microsoft_enabled",
|
|
"social_auth_apple_enabled",
|
|
"social_auth_dropbox_enabled",
|
|
"social_auth_github_enabled",
|
|
"social_auth_keycloak_enabled",
|
|
"social_auth_generic_oauth2_enabled",
|
|
):
|
|
setattr(mock_settings, attr, False)
|
|
|
|
with patch.object(auth_module, "_register_oauth_client"):
|
|
auth_module._setup_social_providers()
|
|
|
|
assert "stale_provider" not in auth_module.SOCIAL_PROVIDERS
|
|
|
|
def test_register_oauth_client_clears_cache(self):
|
|
"""_register_oauth_client removes the cached client before re-registering."""
|
|
import app.auth as auth_module
|
|
|
|
# Inject a fake cached client
|
|
auth_module.oauth._clients["test_provider"] = object()
|
|
|
|
with patch.object(auth_module.oauth, "register"):
|
|
auth_module._register_oauth_client("test_provider", client_id="x", client_secret="y")
|
|
assert "test_provider" not in auth_module.oauth._clients
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestTranslationKeys:
|
|
"""Tests for new translation keys."""
|
|
|
|
def test_connections_translation_keys_exist(self):
|
|
"""Test that connections translation keys are in en.json."""
|
|
import json
|
|
from pathlib import Path
|
|
|
|
en_path = Path(__file__).parents[1] / "frontend" / "translations" / "en.json"
|
|
translations = json.loads(en_path.read_text())
|
|
|
|
expected_keys = [
|
|
"connections.title",
|
|
"connections.description",
|
|
"connections.configure",
|
|
"connections.linked",
|
|
"connections.unlinked",
|
|
"connections.sso_auto_login",
|
|
"connections.sso_auto_login_title",
|
|
"connections.sso_auto_login_description",
|
|
"connections.mobile_upload_title",
|
|
"connections.qr_code_enabled",
|
|
"connections.unlinked_services",
|
|
"nav.connections",
|
|
]
|
|
for key in expected_keys:
|
|
assert key in translations, f"Missing translation key: {key}"
|