release: v3.3.0 — platform hardening & admin unification (Track F.1–F.8)
Plugin visibility control, UI/UX fixes, unified admin panel, database-backed settings, and security hardening. Highlights: - ENABLED_PLUGINS env var for plugin visibility (F.1) - Admin designation via ADMIN_EMAILS (F.4a) - Master key scope control (F.4b) - Panel unification + settings from UI (F.4c) - exec() removal, shell injection fix, bcrypt migration (F.8) - 5 new test suites Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
0
tests/legacy/test_oauth_registration_security.py
Normal file → Executable file
0
tests/legacy/test_oauth_registration_security.py
Normal file → Executable file
188
tests/test_admin_system.py
Normal file
188
tests/test_admin_system.py
Normal file
@@ -0,0 +1,188 @@
|
||||
"""Tests for Admin System Unification (F.4)."""
|
||||
|
||||
from core.dashboard.auth import (
|
||||
DashboardSession,
|
||||
get_session_display_info,
|
||||
is_admin_session,
|
||||
)
|
||||
|
||||
|
||||
class TestIsAdminEmail:
|
||||
"""Test is_admin_email() helper."""
|
||||
|
||||
def test_admin_email_matches(self, monkeypatch):
|
||||
monkeypatch.setenv("ADMIN_EMAILS", "admin@example.com,boss@corp.com")
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
assert is_admin_email("admin@example.com") is True
|
||||
|
||||
def test_admin_email_case_insensitive(self, monkeypatch):
|
||||
monkeypatch.setenv("ADMIN_EMAILS", "Admin@Example.com")
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
assert is_admin_email("admin@example.com") is True
|
||||
|
||||
def test_non_admin_email(self, monkeypatch):
|
||||
monkeypatch.setenv("ADMIN_EMAILS", "admin@example.com")
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
assert is_admin_email("user@example.com") is False
|
||||
|
||||
def test_no_env_var_set(self, monkeypatch):
|
||||
monkeypatch.delenv("ADMIN_EMAILS", raising=False)
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
assert is_admin_email("anyone@example.com") is False
|
||||
|
||||
def test_empty_env_var(self, monkeypatch):
|
||||
monkeypatch.setenv("ADMIN_EMAILS", "")
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
assert is_admin_email("anyone@example.com") is False
|
||||
|
||||
def test_whitespace_in_env_var(self, monkeypatch):
|
||||
monkeypatch.setenv("ADMIN_EMAILS", " admin@example.com , boss@corp.com ")
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
assert is_admin_email("admin@example.com") is True
|
||||
|
||||
def test_none_email_returns_false(self, monkeypatch):
|
||||
monkeypatch.setenv("ADMIN_EMAILS", "admin@example.com")
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
assert is_admin_email(None) is False
|
||||
|
||||
|
||||
class TestOAuthAdminRole:
|
||||
"""Test that OAuth callback assigns admin role based on ADMIN_EMAILS."""
|
||||
|
||||
def test_admin_email_gets_admin_role(self, monkeypatch):
|
||||
monkeypatch.setenv("ADMIN_EMAILS", "admin@example.com")
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
email = "admin@example.com"
|
||||
db_role = "user"
|
||||
effective_role = "admin" if is_admin_email(email) else db_role
|
||||
assert effective_role == "admin"
|
||||
|
||||
def test_normal_email_gets_user_role(self, monkeypatch):
|
||||
monkeypatch.setenv("ADMIN_EMAILS", "admin@example.com")
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
email = "user@example.com"
|
||||
db_role = "user"
|
||||
effective_role = "admin" if is_admin_email(email) else db_role
|
||||
assert effective_role == "user"
|
||||
|
||||
def test_no_admin_emails_env_keeps_user_role(self, monkeypatch):
|
||||
monkeypatch.delenv("ADMIN_EMAILS", raising=False)
|
||||
from core.admin_utils import is_admin_email
|
||||
|
||||
email = "anyone@example.com"
|
||||
db_role = "user"
|
||||
effective_role = "admin" if is_admin_email(email) else db_role
|
||||
assert effective_role == "user"
|
||||
|
||||
|
||||
class TestIsAdminSession:
|
||||
"""Test is_admin_session with different session types."""
|
||||
|
||||
def test_master_session_is_admin(self):
|
||||
session = DashboardSession(
|
||||
session_id="test", created_at=None, expires_at=None, user_type="master"
|
||||
)
|
||||
assert is_admin_session(session) is True
|
||||
|
||||
def test_api_key_session_is_admin(self):
|
||||
session = DashboardSession(
|
||||
session_id="test", created_at=None, expires_at=None, user_type="api_key"
|
||||
)
|
||||
assert is_admin_session(session) is True
|
||||
|
||||
def test_oauth_admin_is_admin(self):
|
||||
session = {"user_id": "123", "email": "a@b.com", "role": "admin", "type": "oauth_user"}
|
||||
assert is_admin_session(session) is True
|
||||
|
||||
def test_oauth_user_is_not_admin(self):
|
||||
session = {"user_id": "123", "email": "a@b.com", "role": "user", "type": "oauth_user"}
|
||||
assert is_admin_session(session) is False
|
||||
|
||||
|
||||
class TestGetSessionDisplayInfo:
|
||||
"""Test display info for admin OAuth users."""
|
||||
|
||||
def test_oauth_admin_shows_admin_type(self):
|
||||
session = {
|
||||
"user_id": "123",
|
||||
"email": "a@b.com",
|
||||
"name": "Admin User",
|
||||
"role": "admin",
|
||||
"type": "oauth_user",
|
||||
}
|
||||
info = get_session_display_info(session)
|
||||
assert info["type"] == "admin"
|
||||
assert info["name"] == "Admin User"
|
||||
assert info["email"] == "a@b.com"
|
||||
|
||||
def test_oauth_user_shows_user_type(self):
|
||||
session = {
|
||||
"user_id": "123",
|
||||
"email": "a@b.com",
|
||||
"name": "Normal User",
|
||||
"role": "user",
|
||||
"type": "oauth_user",
|
||||
}
|
||||
info = get_session_display_info(session)
|
||||
assert info["type"] == "user"
|
||||
|
||||
def test_master_session_shows_admin(self):
|
||||
session = DashboardSession(
|
||||
session_id="test", created_at=None, expires_at=None, user_type="master"
|
||||
)
|
||||
info = get_session_display_info(session)
|
||||
assert info["type"] == "admin"
|
||||
assert info["name"] == "Admin"
|
||||
|
||||
|
||||
class TestDisableMasterKeyLogin:
|
||||
"""Test DISABLE_MASTER_KEY_LOGIN env var."""
|
||||
|
||||
def test_master_key_works_by_default(self, monkeypatch):
|
||||
monkeypatch.delenv("DISABLE_MASTER_KEY_LOGIN", raising=False)
|
||||
monkeypatch.setenv("MASTER_API_KEY", "test-key-123")
|
||||
|
||||
from core.dashboard.auth import DashboardAuth
|
||||
|
||||
auth = DashboardAuth(master_api_key="test-key-123")
|
||||
is_valid, user_type, key_id = auth.validate_api_key("test-key-123")
|
||||
assert is_valid is True
|
||||
assert user_type == "master"
|
||||
|
||||
def test_master_key_blocked_when_disabled(self, monkeypatch):
|
||||
monkeypatch.setenv("DISABLE_MASTER_KEY_LOGIN", "true")
|
||||
monkeypatch.setenv("MASTER_API_KEY", "test-key-123")
|
||||
|
||||
from core.dashboard.auth import DashboardAuth
|
||||
|
||||
auth = DashboardAuth(master_api_key="test-key-123")
|
||||
is_valid, user_type, key_id = auth.validate_api_key("test-key-123")
|
||||
assert is_valid is False
|
||||
|
||||
def test_master_key_works_when_explicitly_enabled(self, monkeypatch):
|
||||
monkeypatch.setenv("DISABLE_MASTER_KEY_LOGIN", "false")
|
||||
monkeypatch.setenv("MASTER_API_KEY", "test-key-123")
|
||||
|
||||
from core.dashboard.auth import DashboardAuth
|
||||
|
||||
auth = DashboardAuth(master_api_key="test-key-123")
|
||||
is_valid, user_type, key_id = auth.validate_api_key("test-key-123")
|
||||
assert is_valid is True
|
||||
|
||||
def test_api_key_still_works_when_master_disabled(self, monkeypatch):
|
||||
monkeypatch.setenv("DISABLE_MASTER_KEY_LOGIN", "true")
|
||||
|
||||
from core.dashboard.auth import DashboardAuth
|
||||
|
||||
auth = DashboardAuth(master_api_key="test-key-123")
|
||||
is_valid, user_type, key_id = auth.validate_api_key("test-key-123")
|
||||
assert is_valid is False # Master key blocked
|
||||
17
tests/test_f3_admin_stats.py
Normal file
17
tests/test_f3_admin_stats.py
Normal file
@@ -0,0 +1,17 @@
|
||||
"""Tests for admin dashboard stats (Phase F.3)."""
|
||||
|
||||
from core.dashboard.routes import get_dashboard_stats
|
||||
|
||||
|
||||
class TestAdminDashboardStats:
|
||||
"""Test admin dashboard statistics include platform data."""
|
||||
|
||||
async def test_stats_include_users_count(self):
|
||||
"""Verify users_count is present in admin stats."""
|
||||
stats = await get_dashboard_stats()
|
||||
assert "users_count" in stats
|
||||
|
||||
async def test_stats_include_user_sites_count(self):
|
||||
"""Verify user_sites_count is present in admin stats."""
|
||||
stats = await get_dashboard_stats()
|
||||
assert "user_sites_count" in stats
|
||||
77
tests/test_f3_last_tested.py
Normal file
77
tests/test_f3_last_tested.py
Normal file
@@ -0,0 +1,77 @@
|
||||
"""Tests for last_tested_at site feature (Phase F.3)."""
|
||||
|
||||
import base64
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from core.database import Database
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _set_encryption_key(monkeypatch):
|
||||
"""Ensure ENCRYPTION_KEY is set and singleton is reset for tests."""
|
||||
key = base64.b64encode(os.urandom(32)).decode()
|
||||
monkeypatch.setenv("ENCRYPTION_KEY", key)
|
||||
# Reset the module-level singleton so it picks up the new key
|
||||
import core.encryption as enc_mod
|
||||
|
||||
monkeypatch.setattr(enc_mod, "_credential_encryption", None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def db(tmp_path):
|
||||
db = Database(str(tmp_path / "test.db"))
|
||||
await db.initialize()
|
||||
yield db
|
||||
await db.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def test_user(db):
|
||||
return await db.create_user(
|
||||
email="test@example.com",
|
||||
name="Test User",
|
||||
provider="github",
|
||||
provider_id="12345",
|
||||
)
|
||||
|
||||
|
||||
class TestLastTestedAt:
|
||||
async def test_new_site_has_no_last_tested(self, db, test_user):
|
||||
from core.encryption import get_credential_encryption
|
||||
|
||||
enc = get_credential_encryption()
|
||||
creds = enc.encrypt_credentials({"username": "admin"}, "test-context")
|
||||
site = await db.create_site(
|
||||
user_id=test_user["id"],
|
||||
plugin_type="wordpress",
|
||||
alias="myblog",
|
||||
url="https://example.com",
|
||||
credentials=creds,
|
||||
)
|
||||
assert site is not None
|
||||
assert site["last_tested_at"] is None
|
||||
|
||||
async def test_update_status_sets_last_tested(self, db, test_user):
|
||||
from core.encryption import get_credential_encryption
|
||||
|
||||
enc = get_credential_encryption()
|
||||
creds = enc.encrypt_credentials({"username": "admin"}, "test-context-2")
|
||||
site = await db.create_site(
|
||||
user_id=test_user["id"],
|
||||
plugin_type="wordpress",
|
||||
alias="testblog",
|
||||
url="https://example.com",
|
||||
credentials=creds,
|
||||
)
|
||||
site_id = site["id"]
|
||||
await db.update_site_status(site_id, "active", "Connection OK", user_id=test_user["id"])
|
||||
updated = await db.get_site(site_id, test_user["id"])
|
||||
assert updated["last_tested_at"] is not None
|
||||
|
||||
async def test_schema_version_is_current(self, db):
|
||||
from core.database import SCHEMA_VERSION
|
||||
|
||||
row = await db.fetchone("SELECT MAX(version) AS v FROM schema_version")
|
||||
assert row["v"] == SCHEMA_VERSION
|
||||
35
tests/test_f3_service_pages.py
Normal file
35
tests/test_f3_service_pages.py
Normal file
@@ -0,0 +1,35 @@
|
||||
"""Tests for MCP service pages (Phase F.3)."""
|
||||
|
||||
from core.dashboard.routes import get_service_page_data
|
||||
|
||||
|
||||
class TestServicePageData:
|
||||
async def test_wordpress_service_data(self):
|
||||
data = await get_service_page_data("wordpress")
|
||||
assert data is not None
|
||||
assert data["plugin_type"] == "wordpress"
|
||||
assert data["display_name"] == "WordPress"
|
||||
assert "tools" in data
|
||||
assert len(data["tools"]) > 0
|
||||
assert "credential_fields" in data
|
||||
|
||||
async def test_supabase_service_data(self):
|
||||
data = await get_service_page_data("supabase")
|
||||
assert data is not None
|
||||
assert data["display_name"] == "Supabase"
|
||||
|
||||
async def test_unknown_plugin_returns_none(self):
|
||||
data = await get_service_page_data("nonexistent")
|
||||
assert data is None
|
||||
|
||||
async def test_tools_have_name_and_description(self):
|
||||
data = await get_service_page_data("wordpress")
|
||||
for tool in data["tools"]:
|
||||
assert "name" in tool
|
||||
assert "description" in tool
|
||||
|
||||
async def test_tools_have_scope(self):
|
||||
data = await get_service_page_data("wordpress")
|
||||
for tool in data["tools"]:
|
||||
assert "scope" in tool
|
||||
assert tool["scope"] in ("read", "write", "admin")
|
||||
148
tests/test_plugin_visibility.py
Normal file
148
tests/test_plugin_visibility.py
Normal file
@@ -0,0 +1,148 @@
|
||||
"""Tests for plugin visibility control (Track F.1)."""
|
||||
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
from core.plugin_visibility import (
|
||||
DEFAULT_PUBLIC_PLUGINS,
|
||||
get_public_plugin_types,
|
||||
is_plugin_public,
|
||||
)
|
||||
|
||||
|
||||
class TestGetPublicPluginTypes:
|
||||
"""Tests for get_public_plugin_types()."""
|
||||
|
||||
def test_default_when_env_not_set(self):
|
||||
"""Returns default set when ENABLED_PLUGINS is not set."""
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("ENABLED_PLUGINS", None)
|
||||
result = get_public_plugin_types()
|
||||
assert result == DEFAULT_PUBLIC_PLUGINS
|
||||
|
||||
def test_default_when_env_empty(self):
|
||||
"""Returns default set when ENABLED_PLUGINS is empty string."""
|
||||
with patch.dict(os.environ, {"ENABLED_PLUGINS": ""}):
|
||||
result = get_public_plugin_types()
|
||||
assert result == DEFAULT_PUBLIC_PLUGINS
|
||||
|
||||
def test_custom_plugins_from_env(self):
|
||||
"""Reads custom plugin list from ENABLED_PLUGINS."""
|
||||
with patch.dict(os.environ, {"ENABLED_PLUGINS": "wordpress,gitea"}):
|
||||
result = get_public_plugin_types()
|
||||
assert result == {"wordpress", "gitea"}
|
||||
|
||||
def test_whitespace_handling(self):
|
||||
"""Strips whitespace from plugin names."""
|
||||
with patch.dict(os.environ, {"ENABLED_PLUGINS": " wordpress , supabase "}):
|
||||
result = get_public_plugin_types()
|
||||
assert result == {"wordpress", "supabase"}
|
||||
|
||||
def test_case_insensitive(self):
|
||||
"""Plugin names are lowercased."""
|
||||
with patch.dict(os.environ, {"ENABLED_PLUGINS": "WordPress,SUPABASE"}):
|
||||
result = get_public_plugin_types()
|
||||
assert result == {"wordpress", "supabase"}
|
||||
|
||||
def test_single_plugin(self):
|
||||
"""Works with a single plugin."""
|
||||
with patch.dict(os.environ, {"ENABLED_PLUGINS": "wordpress"}):
|
||||
result = get_public_plugin_types()
|
||||
assert result == {"wordpress"}
|
||||
|
||||
def test_ignores_empty_entries(self):
|
||||
"""Trailing commas or double commas are ignored."""
|
||||
with patch.dict(os.environ, {"ENABLED_PLUGINS": "wordpress,,supabase,"}):
|
||||
result = get_public_plugin_types()
|
||||
assert result == {"wordpress", "supabase"}
|
||||
|
||||
|
||||
class TestIsPluginPublic:
|
||||
"""Tests for is_plugin_public()."""
|
||||
|
||||
def test_default_wordpress_public(self):
|
||||
"""WordPress is public by default."""
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("ENABLED_PLUGINS", None)
|
||||
assert is_plugin_public("wordpress") is True
|
||||
|
||||
def test_default_woocommerce_public(self):
|
||||
"""WooCommerce is public by default."""
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("ENABLED_PLUGINS", None)
|
||||
assert is_plugin_public("woocommerce") is True
|
||||
|
||||
def test_default_supabase_public(self):
|
||||
"""Supabase is public by default."""
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("ENABLED_PLUGINS", None)
|
||||
assert is_plugin_public("supabase") is True
|
||||
|
||||
def test_default_gitea_not_public(self):
|
||||
"""Gitea is not public by default."""
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("ENABLED_PLUGINS", None)
|
||||
assert is_plugin_public("gitea") is False
|
||||
|
||||
def test_default_wordpress_advanced_not_public(self):
|
||||
"""WordPress Advanced is not public by default."""
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("ENABLED_PLUGINS", None)
|
||||
assert is_plugin_public("wordpress_advanced") is False
|
||||
|
||||
def test_case_insensitive_check(self):
|
||||
"""Plugin type check is case insensitive."""
|
||||
with patch.dict(os.environ, {"ENABLED_PLUGINS": "wordpress"}):
|
||||
assert is_plugin_public("WordPress") is True
|
||||
assert is_plugin_public("WORDPRESS") is True
|
||||
|
||||
def test_custom_env_overrides_defaults(self):
|
||||
"""Custom ENABLED_PLUGINS overrides the default set."""
|
||||
with patch.dict(os.environ, {"ENABLED_PLUGINS": "gitea,n8n"}):
|
||||
assert is_plugin_public("gitea") is True
|
||||
assert is_plugin_public("n8n") is True
|
||||
assert is_plugin_public("wordpress") is False
|
||||
|
||||
|
||||
class TestSiteApiIntegration:
|
||||
"""Tests that site_api uses plugin_visibility correctly."""
|
||||
|
||||
def test_get_user_credential_fields_filtered(self):
|
||||
"""get_user_credential_fields only returns enabled plugins."""
|
||||
from core.site_api import get_user_credential_fields
|
||||
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("ENABLED_PLUGINS", None)
|
||||
fields = get_user_credential_fields()
|
||||
|
||||
assert "wordpress" in fields
|
||||
assert "woocommerce" in fields
|
||||
assert "supabase" in fields
|
||||
assert "wordpress_advanced" not in fields
|
||||
assert "gitea" not in fields
|
||||
assert "n8n" not in fields
|
||||
|
||||
def test_get_user_plugin_names_filtered(self):
|
||||
"""get_user_plugin_names only returns enabled plugins."""
|
||||
from core.site_api import get_user_plugin_names
|
||||
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("ENABLED_PLUGINS", None)
|
||||
names = get_user_plugin_names()
|
||||
|
||||
assert "wordpress" in names
|
||||
assert "woocommerce" in names
|
||||
assert "supabase" in names
|
||||
assert "wordpress_advanced" not in names
|
||||
assert "gitea" not in names
|
||||
|
||||
def test_custom_env_changes_fields(self):
|
||||
"""Custom ENABLED_PLUGINS changes which credential fields are returned."""
|
||||
from core.site_api import get_user_credential_fields
|
||||
|
||||
with patch.dict(os.environ, {"ENABLED_PLUGINS": "wordpress"}):
|
||||
fields = get_user_credential_fields()
|
||||
|
||||
assert "wordpress" in fields
|
||||
assert "woocommerce" not in fields
|
||||
assert "supabase" not in fields
|
||||
Reference in New Issue
Block a user