Agent Zero
1b80090ad2
Check Cross-Plugin Imports / check (push) Has been cancelled
Der vorherige Patch hatte permissions=["system:read"] versehentlich in die PluginRouteDef-kwargs gesetzt statt auf Manifest-Ebene — der Key blieb dadurch unregistriert (live bewiesen: nur 11/12 Keys im Produktionskatalog sichtbar). Korrigiert; Test f9 prueft jetzt die ECHTEN Manifeste statt manueller Registrierung, so haette der Fehler ab sofort gefangen werden muessen. Verifikation: tests/test_audit_architecture_fixes.py 17/17, manifest.permissions=['system:read'], is_core=False, routes=1.
519 lines
20 KiB
Python
519 lines
20 KiB
Python
"""Regression tests for the external architecture audit (leocrm-full.zip, 2026-09).
|
|
|
|
Covers the confirmed findings with live counterchecks (AGENTS.md rule —
|
|
no finding is marked done without a measurement):
|
|
|
|
- F1 list_workspaces() returns modules (WorkspaceManager overwrite bug)
|
|
- F2 active-manifests respects tenant deactivation (signature level)
|
|
- F4 ContractRegistry fails closed for DB-inactive plugins
|
|
(restart edge case) and reopens on activation
|
|
- F5 field definitions have a full lifecycle (register/unregister)
|
|
- F9 permission catalog: every used key is registered
|
|
- F10 contact field definitions are plugin-owned (not in core list)
|
|
- F11 forgejo_error_reporter is not is_core
|
|
- F12 dms is a platform core plugin (ADR-020)
|
|
- F13 core worker has no contact import; contacts trash cleanup is a
|
|
plugin job discovered via get_job_modules()
|
|
- F14 the legacy dsgvo-export route delegates to the single DSAR collector
|
|
- F15 get_entity_read_permission fails closed instead of contacts:read
|
|
|
|
The frontend architecture items (static plugin routes, STATIC_COMPONENT_MAP,
|
|
widgetRegistry) are tracked as Phase Q — not covered here.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.tenant import Tenant
|
|
from app.models.user import User, UserTenant
|
|
from app.services import workspace_service
|
|
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _restore_global_registries():
|
|
"""Snapshot & restore global registries mutated by these tests.
|
|
|
|
The F5 lifecycle test deactivates the real 'contacts' plugin through
|
|
PluginService, which (since the audit fix) centrally unregisters its
|
|
CONTRACT and clears permission-registry state. The conftest resets
|
|
entity models but NOT the contract registry — without this restore,
|
|
get_contract('contacts') stays fail-closed for every later test in the
|
|
same process (suite-order dependence).
|
|
"""
|
|
from app.plugins.builtins.contracts import get_contract_registry
|
|
from app.core.permission_registry import get_permission_registry
|
|
from app.services.entity_permission_service import (
|
|
ENTITY_MODELS,
|
|
ENTITY_PLUGIN_OWNERS,
|
|
)
|
|
|
|
perm_reg = get_permission_registry()
|
|
saved_active = set(perm_reg._active_plugins)
|
|
saved_plugin_perms = {k: list(v) for k, v in perm_reg._plugin_permissions.items()}
|
|
saved_field_defs = {k: list(v) for k, v in perm_reg._field_definitions.items()}
|
|
|
|
cr = get_contract_registry()
|
|
saved_contracts = dict(cr._contracts)
|
|
saved_loaded = set(cr._loaded)
|
|
saved_unreg = set(cr._unregistered)
|
|
saved_db_inactive = set(cr._db_inactive)
|
|
|
|
saved_entity_models = dict(ENTITY_MODELS)
|
|
saved_owners = dict(ENTITY_PLUGIN_OWNERS)
|
|
|
|
yield
|
|
|
|
perm_reg._active_plugins.clear()
|
|
perm_reg._active_plugins.update(saved_active)
|
|
perm_reg._plugin_permissions.clear()
|
|
perm_reg._plugin_permissions.update(saved_plugin_perms)
|
|
perm_reg._field_definitions.clear()
|
|
perm_reg._field_definitions.update(saved_field_defs)
|
|
|
|
cr._contracts.clear()
|
|
cr._contracts.update(saved_contracts)
|
|
cr._loaded.clear()
|
|
cr._loaded.update(saved_loaded)
|
|
cr._unregistered.clear()
|
|
cr._unregistered.update(saved_unreg)
|
|
cr._db_inactive.clear()
|
|
cr._db_inactive.update(saved_db_inactive)
|
|
|
|
ENTITY_MODELS.clear()
|
|
ENTITY_MODELS.update(saved_entity_models)
|
|
ENTITY_PLUGIN_OWNERS.clear()
|
|
ENTITY_PLUGIN_OWNERS.update(saved_owners)
|
|
|
|
# ── F1: list_workspaces returns modules ─────────────────────────────────────
|
|
|
|
|
|
async def _seed_tenant_and_user(db: AsyncSession) -> dict:
|
|
"""Seed a tenant and a user, return IDs (test_workspaces.py pattern)."""
|
|
from app.core.auth import hash_password
|
|
|
|
tenant = Tenant(name="Audit Tenant", slug=f"audit-{uuid.uuid4().hex[:8]}")
|
|
db.add(tenant)
|
|
await db.flush()
|
|
|
|
user = User(
|
|
email=f"audit-{uuid.uuid4().hex[:8]}@example.com",
|
|
name="Audit User",
|
|
password_hash=hash_password("TestPass123!"),
|
|
is_active=True,
|
|
preferences={},
|
|
)
|
|
db.add(user)
|
|
await db.flush()
|
|
|
|
db.add(UserTenant(user_id=user.id, tenant_id=tenant.id, is_default=True, role="admin"))
|
|
await db.flush()
|
|
return {"tenant": tenant, "user": user}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_f1_list_workspaces_returns_modules(db_session: AsyncSession):
|
|
"""Audit F1: list_workspaces must include configured modules.
|
|
|
|
Previously every workspace came back with modules: [] and the
|
|
WorkspaceManager editor overwrote existing config as all-hidden.
|
|
"""
|
|
seed = await _seed_tenant_and_user(db_session)
|
|
created = await workspace_service.create_workspace(
|
|
db_session, seed["tenant"].id, seed["user"].id, name="AuditWS",
|
|
)
|
|
ws_id = uuid.UUID(created["id"])
|
|
await workspace_service.set_workspace_modules(
|
|
db_session,
|
|
seed["tenant"].id,
|
|
ws_id,
|
|
[
|
|
{"module_key": "contacts", "is_visible": True, "menu_order": 10, "config": {}},
|
|
{"module_key": "calendar", "is_visible": False, "menu_order": 20, "config": {}},
|
|
],
|
|
)
|
|
|
|
result = await workspace_service.list_workspaces(db_session, seed["tenant"].id)
|
|
assert result["total"] == 1
|
|
listed = result["items"][0]
|
|
assert listed["name"] == "AuditWS"
|
|
# THE regression: modules must be present with their is_visible flags
|
|
mods = {m["module_key"]: m["is_visible"] for m in listed["modules"]}
|
|
assert mods == {"contacts": True, "calendar": False}
|
|
|
|
|
|
# ── F2: active-manifests tenant filtering (signature level) ────────────────
|
|
|
|
|
|
def test_f2_get_active_manifests_accepts_tenant_id():
|
|
"""Audit F2: the registry manifest API accepts and honours tenant_id."""
|
|
import inspect
|
|
|
|
from app.plugins.registry import PluginRegistry
|
|
|
|
sig = inspect.signature(PluginRegistry.get_active_manifests)
|
|
assert "tenant_id" in sig.parameters, (
|
|
"get_active_manifests must accept tenant_id so tenant-deactivated "
|
|
"plugins are hidden from the UI manifests"
|
|
)
|
|
|
|
# Service passes it through
|
|
from app.services.plugin_service import PluginService
|
|
|
|
sig = inspect.signature(PluginService.get_active_manifests)
|
|
assert "tenant_id" in sig.parameters
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_f2_manifests_exclude_tenant_disabled_plugins(db_session: AsyncSession):
|
|
"""Audit F2: a plugin deactivated for the tenant is filtered out."""
|
|
from sqlalchemy import text
|
|
|
|
seed = await _seed_tenant_and_user(db_session)
|
|
|
|
# tenant_plugin_activation is created by migration 0066 only (no ORM
|
|
# model) — test DBs built via create_all lack it, so create it if needed.
|
|
await db_session.execute(
|
|
text(
|
|
"CREATE TABLE IF NOT EXISTS tenant_plugin_activation ("
|
|
"id UUID PRIMARY KEY DEFAULT gen_random_uuid(), "
|
|
"tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, "
|
|
"plugin_name VARCHAR(100) NOT NULL, "
|
|
"is_active BOOLEAN NOT NULL DEFAULT true, "
|
|
"activated_by UUID, "
|
|
"created_at TIMESTAMPTZ NOT NULL DEFAULT now(), "
|
|
"updated_at TIMESTAMPTZ NOT NULL DEFAULT now())"
|
|
)
|
|
)
|
|
await db_session.execute(
|
|
text(
|
|
"INSERT INTO tenant_plugin_activation (tenant_id, plugin_name, is_active) "
|
|
"VALUES (:tid, 'forgejo_error_reporter', false)"
|
|
),
|
|
{"tid": str(seed["tenant"].id)},
|
|
)
|
|
await db_session.flush()
|
|
|
|
from app.models.plugin import Plugin as PluginModel
|
|
from app.plugins.registry import get_registry
|
|
|
|
registry = get_registry()
|
|
if not registry.list_discovered():
|
|
registry.discover_builtins()
|
|
|
|
# Seed a globally-ACTIVE plugin record so the manifest is served at all
|
|
# (test DBs have no plugin rows unless installed through the registry).
|
|
db_session.add(PluginModel(
|
|
name="forgejo_error_reporter",
|
|
display_name="Forgejo Error Reporter",
|
|
version="1.0.0",
|
|
status="active",
|
|
installed=True,
|
|
active=True,
|
|
is_core=False,
|
|
))
|
|
await db_session.flush()
|
|
|
|
all_manifests = await registry.get_active_manifests(db_session, tenant_id=None)
|
|
assert any(m["name"] == "forgejo_error_reporter" for m in all_manifests), (
|
|
"globally active plugin must appear in manifests (test premise)"
|
|
)
|
|
|
|
filtered = await registry.get_active_manifests(
|
|
db_session, tenant_id=seed["tenant"].id
|
|
)
|
|
assert all(m["name"] != "forgejo_error_reporter" for m in filtered), (
|
|
"tenant-deactivated plugin must not appear in the UI manifests"
|
|
)
|
|
|
|
|
|
# ── F4: contract registry restart guard ────────────────────────────────────
|
|
|
|
|
|
def test_f4_contract_registry_fails_closed_for_db_inactive():
|
|
"""Audit F4: contracts of DB-inactive plugins are not resurrected."""
|
|
from app.plugins.builtins.contracts import get_contract, get_contract_registry
|
|
|
|
reg = get_contract_registry()
|
|
reg._reset_for_testing()
|
|
try:
|
|
reg.mark_db_inactive({"kommunikation"})
|
|
# Even after lazy-load attempts the contract must stay unavailable
|
|
assert get_contract("kommunikation") is None
|
|
finally:
|
|
reg._reset_for_testing()
|
|
|
|
|
|
def test_f4_contract_registry_reopens_on_activation():
|
|
"""Audit F4: activating a plugin clears its inactive markers."""
|
|
from app.plugins.builtins.contracts import get_contract_registry
|
|
|
|
reg = get_contract_registry()
|
|
reg._reset_for_testing()
|
|
try:
|
|
reg.mark_db_inactive({"some_plugin"})
|
|
reg._unregistered.add("some_plugin")
|
|
reg.mark_plugin_active("some_plugin")
|
|
assert "some_plugin" not in reg._db_inactive
|
|
assert "some_plugin" not in reg._unregistered
|
|
finally:
|
|
reg._reset_for_testing()
|
|
|
|
|
|
# ── F5: field definitions lifecycle ─────────────────────────────────────────
|
|
|
|
|
|
def test_f5_field_definitions_full_lifecycle():
|
|
"""Audit F5: register → present, unregister → gone."""
|
|
from app.core.permission_registry import PermissionRegistry
|
|
|
|
reg = PermissionRegistry()
|
|
reg.initialize()
|
|
defs = [{"module": "x", "field": "y", "label": "Y", "sensitivity": "normal"}]
|
|
reg.register_field_definitions("audit_plugin", defs)
|
|
assert any(d.get("plugin_key", d.get("field")) or True for d in defs)
|
|
all_defs = reg.get_all_field_definitions()
|
|
assert any(d.get("field") == "y" for d in all_defs)
|
|
|
|
reg.unregister_field_definitions("audit_plugin")
|
|
all_defs = reg.get_all_field_definitions()
|
|
assert not any(d.get("field") == "y" and d.get("module") == "x" for d in all_defs)
|
|
|
|
|
|
def test_f5_deactivate_plugin_unregisters_field_definitions():
|
|
"""Audit F5: PluginService.deactivate_plugin removes field definitions."""
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from app.core.permission_registry import get_permission_registry, init_permission_registry
|
|
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
|
from app.services.plugin_service import PluginService
|
|
|
|
init_permission_registry(set())
|
|
perm_reg = get_permission_registry()
|
|
perm_reg._active_plugins.add("contacts")
|
|
perm_reg.register_field_definitions("contacts", ContactsPlugin().get_field_definitions())
|
|
assert any(
|
|
d.get("module") == "contacts"
|
|
for d in perm_reg.get_all_field_definitions()
|
|
)
|
|
|
|
service = PluginService()
|
|
# deactivate_plugin resolves the record via registry._get_plugin_record —
|
|
# a MagicMock plugin record with active=True triggers the cleanup path.
|
|
# NOTE: service._registry IS the global singleton — save & restore the
|
|
# patched methods so the conftest bootstrap stays intact for later tests
|
|
# (order-dependence guard).
|
|
mock_record = MagicMock()
|
|
mock_record.active = True
|
|
mock_record.status = "active"
|
|
_orig_get_record = service._registry._get_plugin_record
|
|
_orig_deactivate = service._registry.deactivate
|
|
_orig_get_plugin = service._registry.get_plugin
|
|
service._registry._get_plugin_record = AsyncMock(return_value=mock_record)
|
|
service._registry.deactivate = AsyncMock(return_value=mock_record)
|
|
service._registry.get_plugin = MagicMock(return_value=ContactsPlugin())
|
|
|
|
try:
|
|
asyncio.get_event_loop().run_until_complete(
|
|
service.deactivate_plugin(AsyncMock(), "contacts")
|
|
)
|
|
finally:
|
|
service._registry._get_plugin_record = _orig_get_record
|
|
service._registry.deactivate = _orig_deactivate
|
|
service._registry.get_plugin = _orig_get_plugin
|
|
assert not any(
|
|
d.get("module") == "contacts"
|
|
for d in perm_reg.get_all_field_definitions()
|
|
), "deactivation must unregister the plugin's field definitions"
|
|
|
|
|
|
# ── F9: permission catalog completeness ─────────────────────────────────────
|
|
|
|
|
|
def test_f9_all_audit_permission_keys_registered():
|
|
"""Audit F9: the 12 previously missing keys are grantable now."""
|
|
from app.core.permission_registry import (
|
|
CORE_PERMISSIONS,
|
|
get_permission_registry,
|
|
init_permission_registry,
|
|
)
|
|
|
|
init_permission_registry(set())
|
|
# Register from the REAL manifests (not hand-picked lists) — this catches
|
|
# kwargs-level corruption like the bug where permissions=[...] had landed
|
|
# inside PluginRouteDef kwargs instead of the manifest level.
|
|
from app.plugins.builtins.forgejo_error_reporter.plugin import (
|
|
ForgejoErrorReporterPlugin,
|
|
)
|
|
from app.plugins.builtins.permissions.plugin import PermissionsPlugin
|
|
|
|
assert PermissionsPlugin().manifest.permissions == [
|
|
"permissions:read", "permissions:admin",
|
|
]
|
|
assert ForgejoErrorReporterPlugin().manifest.permissions == ["system:read"], (
|
|
"system:read must be declared at MANIFEST level (not PluginRouteDef kwargs)"
|
|
)
|
|
|
|
get_permission_registry().register_plugin_permissions(
|
|
"permissions", PermissionsPlugin().manifest.permissions
|
|
)
|
|
get_permission_registry().register_plugin_permissions(
|
|
"forgejo_error_reporter", ForgejoErrorReporterPlugin().manifest.permissions
|
|
)
|
|
|
|
known = get_permission_registry().get_all()
|
|
keys = {p["key"] for p in known}
|
|
|
|
core_expected = [
|
|
"automation:admin",
|
|
"bank-accounts:read",
|
|
"bank-accounts:write",
|
|
"delegations:read",
|
|
"delegations:write",
|
|
"policies:read",
|
|
"policies:write",
|
|
"templates:read",
|
|
"templates:write",
|
|
]
|
|
for key in core_expected:
|
|
assert key in {p["key"] for p in CORE_PERMISSIONS}, f"{key} missing in CORE_PERMISSIONS"
|
|
|
|
for key in ["permissions:read", "permissions:admin", "system:read"]:
|
|
assert key in keys, f"{key} must be grantable"
|
|
|
|
assert get_permission_registry().is_valid("delegations:read")
|
|
|
|
|
|
# ── F10: contact field definitions are plugin-owned ────────────────────────
|
|
|
|
|
|
def test_f10_contact_field_definitions_plugin_owned():
|
|
"""Audit F10: contacts:* field defs come from the plugin manifest."""
|
|
from app.core.permission_registry import CORE_FIELD_DEFINITIONS
|
|
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
|
|
|
core_modules = {fd["module"] for fd in CORE_FIELD_DEFINITIONS}
|
|
assert "contacts" not in core_modules, "core must not own contacts field definitions"
|
|
|
|
plugin_defs = ContactsPlugin().get_field_definitions()
|
|
assert len(plugin_defs) >= 30, "contacts plugin should carry its ~39 field definitions"
|
|
assert all(fd["module"] == "contacts" for fd in plugin_defs)
|
|
# Spot-check sensitive fields are still classified sensitive after the move
|
|
by_field = {fd["field"]: fd for fd in plugin_defs}
|
|
assert by_field["mobilephone"]["sensitivity"] == "sensitive"
|
|
assert by_field["bic"]["sensitivity"] == "sensitive"
|
|
|
|
|
|
# ── F11/F12: plugin classification ─────────────────────────────────────────
|
|
|
|
|
|
def test_f11_forgejo_error_reporter_not_core():
|
|
"""Audit F11: the test/staging-only reporter is deactivatable."""
|
|
from app.plugins.builtins.forgejo_error_reporter.plugin import ForgejoErrorReporterPlugin
|
|
|
|
assert ForgejoErrorReporterPlugin().manifest.is_core is False
|
|
|
|
|
|
def test_f12_dms_is_platform_core_plugin():
|
|
"""Audit F12/ADR-020: the core schema builds on the DMS files table."""
|
|
from app.plugins.builtins.dms.plugin import DmsPlugin
|
|
|
|
assert DmsPlugin().manifest.is_core is True
|
|
|
|
|
|
# ── F13: worker decoupled from contacts ────────────────────────────────────
|
|
|
|
|
|
def test_f13_core_worker_has_no_contact_import():
|
|
"""Audit F13: app/core/worker.py must not import contact models."""
|
|
import inspect
|
|
|
|
import app.core.worker as worker
|
|
|
|
src = inspect.getsource(worker)
|
|
assert "from app.models.contact import" not in src
|
|
|
|
|
|
def test_f13_contacts_trash_job_discovered_via_job_modules():
|
|
"""Audit F13: contacts plugin owns its trash cleanup job."""
|
|
from app.core.job_registry import get_job
|
|
from app.plugins.builtins.contacts.jobs import cleanup_contacts_trash_job
|
|
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
|
|
|
modules = ContactsPlugin().get_job_modules()
|
|
assert "app.plugins.builtins.contacts.jobs" in modules
|
|
|
|
import importlib
|
|
|
|
importlib.import_module("app.plugins.builtins.contacts.jobs")
|
|
assert get_job("cleanup_contacts_trash") is cleanup_contacts_trash_job
|
|
|
|
|
|
# ── F14: single DSAR export path ───────────────────────────────────────────
|
|
|
|
|
|
def test_f14_dsgvo_route_delegates_to_dsar_collector():
|
|
"""Audit F14: the legacy export route delegates, it does not reimplement."""
|
|
import inspect
|
|
|
|
import app.routes.system_settings as ss
|
|
|
|
src = inspect.getsource(ss)
|
|
assert "_dsar_collect_user_data" in src, "route must delegate to the DSAR collector"
|
|
assert "from app.models.contact import" not in src, (
|
|
"the export route must not know contacts internals"
|
|
)
|
|
|
|
|
|
# ── F15: entity permission fallback fails closed ───────────────────────────
|
|
|
|
|
|
def test_f15_entity_read_permission_fails_closed():
|
|
"""Audit F15: unmapped entities no longer default to contacts:read."""
|
|
from app.services.entity_permission_service import (
|
|
ENTITY_PLUGIN_OWNERS,
|
|
get_entity_read_permission,
|
|
)
|
|
|
|
saved = ENTITY_PLUGIN_OWNERS.pop("zzz_audit_unknown", None)
|
|
try:
|
|
perm = get_entity_read_permission("zzz_audit_unknown")
|
|
assert perm != "contacts:read"
|
|
assert perm == "__unmapped__:read"
|
|
finally:
|
|
if saved is not None:
|
|
ENTITY_PLUGIN_OWNERS["zzz_audit_unknown"] = saved
|
|
|
|
|
|
def test_f15_contact_folder_registered_by_plugin():
|
|
"""Audit F15: contact_folder comes from ContactsPlugin, not core."""
|
|
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
|
|
|
models = ContactsPlugin().get_entity_models()
|
|
assert "contact_folder" in models
|
|
|
|
from app.services.entity_permission_service import get_entity_read_permission
|
|
|
|
assert get_entity_read_permission("contact_folder") == "contacts:read"
|
|
|
|
|
|
# ── Uninstall lifecycle (audit P1) ─────────────────────────────────────────
|
|
|
|
|
|
def test_f3_uninstall_deactivates_via_service_first():
|
|
"""Audit P1: uninstall_plugin runs the full service deactivation first."""
|
|
import inspect
|
|
|
|
from app.services.plugin_service import PluginService
|
|
|
|
src = inspect.getsource(PluginService.uninstall_plugin)
|
|
assert "self.deactivate_plugin(" in src, (
|
|
"uninstall must run the full service-level deactivation (permissions, "
|
|
"active set, entity models) before registry.uninstall()"
|
|
)
|