GovOPlaN/govoplan-dashboard
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(dashboard): add governed DSAR coverage

Browse Source
This commit is contained in:
zemion
2026-08-21 03:50:08 +02:00
parent b3f612bfa0
commit f8305f11f2
3 changed files with 755 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_dashboard/backend/dsar_provider.py
+335
View File
@@ -0,0 +1,335 @@
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarExecutionResultRef,
DsarRecordRef,
DsarSubjectRef,
dsar_capability_name,
)
from govoplan_dashboard.backend.db.models import DashboardLayout
DASHBOARD_DSAR_CAPABILITY = dsar_capability_name("dashboard")
_MAX_RECORDS = 500
_CONFLICT = object()
@dataclass(frozen=True, slots=True)
class _SubjectSelectors:
account_id: str
layout_id: str | None
view_id: str | None
class DashboardDsarProvider:
provider_id = "dashboard"
module_id = "dashboard"
def search_subject(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
) -> Sequence[DsarRecordRef]:
db = _session(session)
selectors = _subject_selectors(subject)
if selectors is None:
return ()
query = db.query(DashboardLayout).filter(
DashboardLayout.tenant_id == tenant_id,
DashboardLayout.account_id == selectors.account_id,
)
if selectors.layout_id:
query = query.filter(DashboardLayout.id == selectors.layout_id)
if selectors.view_id:
query = query.filter(DashboardLayout.view_id == selectors.view_id)
rows = (
query.order_by(DashboardLayout.context_key, DashboardLayout.id)
.limit(_MAX_RECORDS + 1)
.all()
)
if len(rows) > _MAX_RECORDS:
raise ValueError(
"Dashboard DSAR result limit exceeded; narrow the selectors."
)
return tuple(_record(row) for row in rows)
def plan_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
records: Sequence[DsarRecordRef],
) -> Sequence[DsarErasureActionRef]:
db = _session(session)
selectors = _subject_selectors(subject)
if selectors is None:
raise ValueError("Dashboard DSAR requires one corroborated account.")
actions: list[DsarErasureActionRef] = []
for record in records:
_validate_record(record)
layout = (
db.query(DashboardLayout)
.filter(
DashboardLayout.tenant_id == tenant_id,
DashboardLayout.id == record.resource_id,
)
.one_or_none()
)
if (
layout is None
or layout.account_id != selectors.account_id
or (selectors.layout_id and layout.id != selectors.layout_id)
or (selectors.view_id and layout.view_id != selectors.view_id)
):
actions.append(_manual_action(record))
continue
actions.append(
DsarErasureActionRef(
action_id=f"dashboard:delete:dashboard_layout:{layout.id}:r{layout.revision}",
provider_id=self.provider_id,
module_id=self.module_id,
kind="delete",
resource_type="dashboard_layout",
resource_id=layout.id,
title="Delete personal Dashboard layout",
rationale=(
"The layout is an account-owned display preference. Its "
"deletion does not change widgets or their source data."
),
executable=True,
irreversible=True,
metadata={
"account_id": layout.account_id,
"revision": layout.revision,
},
)
)
return tuple(actions)
def execute_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
actions: Sequence[DsarErasureActionRef],
request_id: str,
) -> Sequence[DsarExecutionResultRef]:
db = _session(session)
selectors = _subject_selectors(subject)
if selectors is None:
raise ValueError("Dashboard DSAR requires one corroborated account.")
results: list[DsarExecutionResultRef] = []
for action in actions:
_validate_action(action)
if not action.executable or action.kind != "delete":
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary="The Dashboard action is not an executable deletion.",
)
)
continue
if action.resource_type != "dashboard_layout":
raise ValueError("Unsupported executable Dashboard DSAR action.")
layout = (
db.query(DashboardLayout)
.filter(
DashboardLayout.tenant_id == tenant_id,
DashboardLayout.id == action.resource_id,
)
.one_or_none()
)
if layout is None:
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="unchanged",
summary="The personal Dashboard layout was already absent.",
evidence={"request_id": request_id},
)
)
continue
expected_account = str(action.metadata.get("account_id") or "")
expected_revision = action.metadata.get("revision")
if (
layout.account_id != selectors.account_id
or expected_account != selectors.account_id
or (selectors.layout_id and layout.id != selectors.layout_id)
or (selectors.view_id and layout.view_id != selectors.view_id)
):
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"The layout is not owned by the corroborated subject "
"account and selector context."
),
)
)
continue
if expected_revision != layout.revision:
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"The layout changed after the erasure plan; create a "
"new plan before deleting it."
),
)
)
continue
layout_id = layout.id
db.delete(layout)
db.flush()
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="executed",
summary=(
"Deleted the personal Dashboard layout without changing "
"widget or domain data."
),
evidence={
"request_id": request_id,
"layout_id": layout_id,
},
)
)
return tuple(results)
def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
references = subject.external_references
account_id = _coalesce(
subject.account_id,
references.get("dashboard.account"),
references.get("access.account"),
)
layout_id = _coalesce(
references.get("dashboard.layout"),
references.get("dashboard.layout_id"),
)
view_id = _coalesce(
references.get("dashboard.view"),
references.get("views.view"),
)
if _CONFLICT in (account_id, layout_id, view_id):
return None
normalized_account = _optional_string(account_id)
if normalized_account is None:
return None
return _SubjectSelectors(
account_id=normalized_account,
layout_id=_optional_string(layout_id),
view_id=_optional_string(view_id),
)
def _coalesce(*values: str | None) -> str | None | object:
normalized = {str(value).strip() for value in values if str(value or "").strip()}
if len(normalized) > 1:
return _CONFLICT
return next(iter(normalized), None)
def _optional_string(value: object) -> str | None:
return value if isinstance(value, str) and value else None
def _record(layout: DashboardLayout) -> DsarRecordRef:
placements = []
for value in layout.placements[:100]:
if not isinstance(value, dict):
continue
placements.append(
{
"instance_id": str(value.get("instance_id") or "")[:120],
"widget_id": str(value.get("widget_id") or "")[:160],
"size": str(value.get("size") or "")[:20],
"column_start": value.get("column_start"),
}
)
return DsarRecordRef(
provider_id="dashboard",
module_id="dashboard",
resource_type="dashboard_layout",
resource_id=layout.id,
category="personal_interface_preference",
title="Personal Dashboard layout",
data={
"view_id": layout.view_id,
"layout_version": layout.layout_version,
"revision": layout.revision,
"placements": placements,
"known_widget_ids": [
str(value)[:160] for value in layout.known_widget_ids[:500]
],
},
observed_at=_aware(layout.updated_at),
)
def _manual_action(record: DsarRecordRef) -> DsarErasureActionRef:
return DsarErasureActionRef(
action_id=f"dashboard:manual_review:{record.resource_type}:{record.resource_id}",
provider_id="dashboard",
module_id="dashboard",
kind="manual_review",
resource_type=record.resource_type,
resource_id=record.resource_id,
title=f"Review {record.title}",
rationale=(
"The selected layout is absent or no longer belongs to the exact "
"tenant, account, and narrowing selectors."
),
executable=False,
)
def _aware(value: datetime | None) -> datetime | None:
if value is None or value.tzinfo is not None:
return value
return value.replace(tzinfo=timezone.utc)
def _session(value: object) -> Session:
if not isinstance(value, Session):
raise TypeError("Dashboard DSAR requires a SQLAlchemy Session.")
return value
def _validate_record(record: DsarRecordRef) -> None:
if record.provider_id != "dashboard" or record.module_id != "dashboard":
raise ValueError("Dashboard DSAR cannot plan a foreign provider record.")
if record.resource_type != "dashboard_layout" or not record.resource_id:
raise ValueError("Dashboard DSAR record identity is invalid.")
def _validate_action(action: DsarErasureActionRef) -> None:
if action.provider_id != "dashboard" or action.module_id != "dashboard":
raise ValueError("Dashboard DSAR cannot execute a foreign provider action.")
if not action.action_id.startswith("dashboard:"):
raise ValueError("Dashboard DSAR action identity is invalid.")
__all__ = ["DASHBOARD_DSAR_CAPABILITY", "DashboardDsarProvider"]
src/govoplan_dashboard/backend/manifest.py
+64
View File
@@ -8,10 +8,13 @@ from govoplan_core.core.module_guards import (
persistent_table_uninstall_guard,
)
from govoplan_core.core.modules import (
CapabilityDocumentation,
DocumentationTopic,
FrontendModule,
FrontendRoute,
MigrationSpec,
ModuleContext,
ModuleInterfaceProvider,
ModuleManifest,
NavItem,
)
@@ -19,6 +22,10 @@ from govoplan_core.core.provider_governance import declared_module_architecture
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
from govoplan_dashboard.backend.db.models import DashboardLayout
from govoplan_dashboard.backend.dsar_provider import (
DASHBOARD_DSAR_CAPABILITY,
DashboardDsarProvider,
)
def _dashboard_router(_context):
@@ -27,6 +34,10 @@ def _dashboard_router(_context):
return router
def _dsar_provider(_context: ModuleContext) -> DashboardDsarProvider:
return DashboardDsarProvider()
def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
return {
"dashboard_layouts": (
@@ -43,7 +54,21 @@ manifest = ModuleManifest(
version="0.1.18",
required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
optional_dependencies=("ops", "campaigns", "files", "mail", "tasks", "notifications", "reporting"),
provides_interfaces=(
ModuleInterfaceProvider(name=DASHBOARD_DSAR_CAPABILITY, version="0.1.0"),
),
route_factory=_dashboard_router,
capability_factories={DASHBOARD_DSAR_CAPABILITY: _dsar_provider},
capability_documentation={
DASHBOARD_DSAR_CAPABILITY: CapabilityDocumentation(
label="Dashboard data-subject request provider",
summary=(
"Exports and deletes exact account-owned Dashboard layout "
"preferences without traversing widget data."
),
contract_version="0.1.0",
),
},
tenant_summary_providers=(_tenant_summary,),
migration_spec=MigrationSpec(
module_id="dashboard",
@@ -122,6 +147,45 @@ manifest = ModuleManifest(
),
),
documentation=(
DocumentationTopic(
id="dashboard.data-subject-requests",
title="Dashboard data-subject requests",
summary=(
"Export or delete personal Dashboard layouts without changing "
"the data displayed by their widgets."
),
body=(
"Dashboard contributes layouts only when the request contains one "
"exact, corroborated account identifier in the active tenant. A "
"layout or View reference can narrow that result. The access package "
"contains layout version, revision, View, widget identities, sizes, "
"and positions. It deliberately omits arbitrary widget configuration "
"values and never follows a widget into its owning module. Erasure "
"deletes the selected account-owned layout after verifying its owner "
"and revision; a repeated execution is unchanged. The operation does "
"not delete widgets, reports, files, tasks, or any other domain data."
),
layer="configured",
documentation_types=("admin", "user"),
audience=("user", "tenant_admin", "operator", "auditor"),
related_modules=("core", "views"),
order=19,
metadata={
"help_contexts": [
"dashboard.page",
"dashboard.action.reset",
"privacy.data-subject-requests",
],
"consequence_classes": {
"export_layout": (
"Returns minimized layout structure, never widget-domain data."
),
"delete_layout": (
"Irreversibly removes the selected personal layout only."
),
},
},
),
DocumentationTopic(
id="dashboard.configurable-home",
title="Configurable user dashboard",
tests/test_dsar_provider.py
+356
View File
@@ -0,0 +1,356 @@
from __future__ import annotations
import json
import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarProvider,
DsarRecordRef,
DsarSubjectRef,
)
from govoplan_core.db.base import Base
from govoplan_core.privacy.dsar_workflow import (
create_data_subject_request,
search_data_subject_request,
)
from govoplan_dashboard.backend.db.models import DashboardLayout
from govoplan_dashboard.backend.dsar_provider import (
DASHBOARD_DSAR_CAPABILITY,
DashboardDsarProvider,
)
from govoplan_dashboard.backend.manifest import manifest
class _Registry:
def __init__(
self,
provider: DashboardDsarProvider,
*,
active: bool = True,
) -> None:
self.provider = provider
self.active = active
def capability_names(self):
return (DASHBOARD_DSAR_CAPABILITY,)
def capability_owner(self, name):
self._assert_capability(name)
return "dashboard"
def tenant_entitlement_resolver(self):
active = self.active
class _Resolver:
@staticmethod
def resolve(session, tenant_id):
del session, tenant_id
return type(
"State",
(),
{"effective_modules": ("dashboard",) if active else ()},
)()
return _Resolver()
def require_tenant_capability(self, name, session, **kwargs):
del session, kwargs
self._assert_capability(name)
return self.provider
def manifests(self):
return (type("Manifest", (), {"id": "dashboard"})(),)
@staticmethod
def _assert_capability(name: str) -> None:
if name != DASHBOARD_DSAR_CAPABILITY:
raise KeyError(name)
class DashboardDsarProviderTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(self.engine)
self.session = Session(self.engine)
self.provider = DashboardDsarProvider()
self.assertIsInstance(self.provider, DsarProvider)
self._seed()
self.session.commit()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def _seed(self) -> None:
def layout(
layout_id: str,
*,
tenant_id: str = "tenant-1",
account_id: str = "account-1",
view_id: str | None = None,
) -> DashboardLayout:
return DashboardLayout(
id=layout_id,
tenant_id=tenant_id,
account_id=account_id,
context_key=f"view:{view_id}" if view_id else "full",
view_id=view_id,
layout_version=1,
revision=3,
placements=[
{
"instance_id": f"instance-{layout_id}",
"widget_id": "reporting.metric",
"size": "medium",
"column_start": 2,
"configuration": {
"subjectFilter": f"private-{layout_id}-do-not-export"
},
}
],
known_widget_ids=["reporting.metric"],
)
self.session.add_all(
(
layout("layout-full"),
layout("layout-view", view_id="view-1"),
layout("layout-other-account", account_id="account-other"),
layout("layout-other-tenant", tenant_id="tenant-2"),
)
)
def test_search_is_tenant_and_account_scoped_and_minimized(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(account_id="account-1"),
)
self.assertEqual(
["layout-full", "layout-view"],
[record.resource_id for record in records],
)
exported = json.dumps([record.to_dict() for record in records])
self.assertNotIn("private-layout-full-do-not-export", exported)
self.assertNotIn("private-layout-view-do-not-export", exported)
self.assertNotIn("layout-other-account", exported)
self.assertNotIn("layout-other-tenant", exported)
self.assertIn("reporting.metric", exported)
def test_references_narrow_and_conflicts_fail_closed(self) -> None:
by_layout = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"dashboard.layout": "layout-view"},
),
)
by_view = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"dashboard.view": "view-1"},
),
)
conflict = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"dashboard.account": "account-other"},
),
)
without_account = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
external_references={"dashboard.layout": "layout-view"}
),
)
self.assertEqual(["layout-view"], [item.resource_id for item in by_layout])
self.assertEqual(["layout-view"], [item.resource_id for item in by_view])
self.assertEqual((), conflict)
self.assertEqual((), without_account)
def test_erasure_is_owner_scoped_revision_safe_and_idempotent(self) -> None:
subject = DsarSubjectRef(
account_id="account-1",
external_references={"dashboard.layout": "layout-view"},
)
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=subject,
)
actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=records,
)
self.assertEqual(1, len(actions))
self.assertEqual("delete", actions[0].kind)
first = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=actions,
request_id="dsar-1",
)
second = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=actions,
request_id="dsar-1",
)
self.assertEqual("executed", first[0].status)
self.assertEqual("unchanged", second[0].status)
self.assertIsNone(self.session.get(DashboardLayout, "layout-view"))
self.assertIsNotNone(self.session.get(DashboardLayout, "layout-full"))
self.assertIsNotNone(
self.session.get(DashboardLayout, "layout-other-account")
)
self.assertIsNotNone(
self.session.get(DashboardLayout, "layout-other-tenant")
)
def test_changed_or_foreign_resources_are_blocked(self) -> None:
subject = DsarSubjectRef(account_id="account-1")
record = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"dashboard.layout": "layout-full"},
),
)[0]
action = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=(record,),
)[0]
self.session.get(DashboardLayout, "layout-full").revision += 1
changed = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=(action,),
request_id="dsar-1",
)
self.assertEqual("blocked", changed[0].status)
with self.assertRaisesRegex(ValueError, "foreign provider record"):
self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=(
DsarRecordRef(
provider_id="views",
module_id="views",
resource_type="dashboard_layout",
resource_id="layout-full",
category="preference",
title="Foreign layout",
),
),
)
with self.assertRaisesRegex(ValueError, "foreign provider action"):
self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=(
DsarErasureActionRef(
action_id="views:delete:dashboard_layout:layout-full",
provider_id="views",
module_id="views",
kind="delete",
resource_type="dashboard_layout",
resource_id="layout-full",
title="Delete layout",
rationale="Foreign action",
executable=True,
),
),
request_id="dsar-1",
)
def test_core_workflow_reports_active_and_inactive_provider(self) -> None:
row = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-DASHBOARD-1",
request_kind="access",
subject=DsarSubjectRef(account_id="account-1"),
purpose="Respond to a verified request.",
legal_basis="Article 15 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
search_data_subject_request(
self.session,
registry=_Registry(self.provider),
row=row,
expected_revision=1,
)
self.assertEqual(
[DASHBOARD_DSAR_CAPABILITY], row.coverage["provider_capabilities"]
)
self.assertEqual(2, row.search_result["record_count"])
inactive = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-DASHBOARD-2",
request_kind="access",
subject=DsarSubjectRef(account_id="account-1"),
purpose="Respond to a verified request.",
legal_basis="Article 15 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
search_data_subject_request(
self.session,
registry=_Registry(self.provider, active=False),
row=inactive,
expected_revision=1,
)
self.assertEqual([], inactive.coverage["provider_capabilities"])
self.assertEqual(
[DASHBOARD_DSAR_CAPABILITY],
inactive.coverage["inactive_provider_capabilities"],
)
def test_manifest_registers_and_documents_the_capability(self) -> None:
self.assertIn(DASHBOARD_DSAR_CAPABILITY, manifest.capability_factories)
self.assertIn(DASHBOARD_DSAR_CAPABILITY, manifest.capability_documentation)
self.assertIn(
DASHBOARD_DSAR_CAPABILITY,
{item.name for item in manifest.provides_interfaces},
)
self.assertTrue(
any(
topic.id == "dashboard.data-subject-requests"
and {"admin", "user"}.issubset(topic.documentation_types)
for topic in manifest.documentation
)
)
if __name__ == "__main__":
unittest.main()