feat(search): add governed DSAR coverage

This commit is contained in:
2026-08-21 03:01:01 +02:00
parent cc13bcf45e
commit 016136f56c
4 changed files with 1119 additions and 0 deletions
@@ -0,0 +1,576 @@
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from sqlalchemy import tuple_
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarExecutionResultRef,
DsarRecordRef,
DsarSubjectRef,
dsar_capability_name,
)
from govoplan_search.backend.db.models import (
SearchIndexAclToken,
SearchIndexChangeQueue,
SearchIndexDocument,
)
SEARCH_DSAR_CAPABILITY = dsar_capability_name("search")
_MAX_RECORDS = 5_000
_CONFLICT = object()
_RESERVED_REFERENCE_NAMES = frozenset(
{
"search.account",
"search.identity",
"search.membership",
"access.account",
"identity.id",
"tenancy.membership",
"search.document",
"search.index_document",
"search.change",
"search.queued_change",
}
)
@dataclass(frozen=True, slots=True)
class _SourceReference:
module_id: str
resource_type: str
resource_id: str
@dataclass(frozen=True, slots=True)
class _SubjectSelectors:
account_id: str | None
identity_id: str | None
membership_id: str | None
document_id: str | None
change_id: str | None
source_references: tuple[_SourceReference, ...]
@property
def acl_tokens(self) -> tuple[str, ...]:
return tuple(
token
for token in (
f"account:{self.account_id}" if self.account_id else None,
f"identity:{self.identity_id}" if self.identity_id else None,
f"membership:{self.membership_id}" if self.membership_id else None,
)
if token
)
@property
def has_direct_selector(self) -> bool:
return bool(self.document_id or self.change_id or self.source_references)
@dataclass(frozen=True, slots=True)
class _ResourceIdentity:
module_id: str
resource_type: str
resource_id: str
class SearchDsarProvider:
provider_id = "search"
module_id = "search"
def search_subject(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
) -> Sequence[DsarRecordRef]:
db = _session(session)
selectors = _subject_selectors(subject)
if selectors is None or not (
selectors.acl_tokens or selectors.has_direct_selector
):
return ()
direct = _direct_rows(db, tenant_id=tenant_id, selectors=selectors)
if direct is None:
return ()
documents, changes = direct
records: list[DsarRecordRef] = []
seen: set[tuple[str, str]] = set()
def append(record: DsarRecordRef) -> None:
key = (record.resource_type, record.resource_id)
if key in seen:
return
if len(records) >= _MAX_RECORDS:
raise ValueError(
"Search DSAR result limit exceeded; narrow the selectors."
)
seen.add(key)
records.append(record)
for row in documents:
append(_document_record(row))
for row in changes:
append(_change_record(row))
for row, document in _acl_rows(
db,
tenant_id=tenant_id,
selectors=selectors,
direct_identities={_identity(item) for item in (*documents, *changes)},
):
append(_acl_record(row, document=document))
return tuple(records)
def plan_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
records: Sequence[DsarRecordRef],
) -> Sequence[DsarErasureActionRef]:
del tenant_id
_session(session)
if _subject_selectors(subject) is None:
raise ValueError("Search DSAR subject selectors conflict.")
actions: list[DsarErasureActionRef] = []
for record in records:
_validate_record(record)
executable = record.resource_type in {
"search_index_document",
"search_index_change",
}
kind = "delete" if executable else "manual_review"
actions.append(
DsarErasureActionRef(
action_id=(
f"search:{kind}:{record.resource_type}:{record.resource_id}"
),
provider_id=self.provider_id,
module_id=self.module_id,
kind=kind,
resource_type=record.resource_type,
resource_id=record.resource_id,
title=(
f"Delete {record.title}"
if executable
else f"Review {record.title}"
),
rationale=(
"The row is a derived Search cache or queued copy and can "
"be removed without changing the authoritative source."
if executable
else "The ACL token is a derived authorization projection; correct or revoke authority at the source before rebuilding Search."
),
executable=executable,
)
)
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)
if _subject_selectors(subject) is None:
raise ValueError("Search DSAR subject selectors conflict.")
results: list[DsarExecutionResultRef] = []
for action in actions:
_validate_action(action)
if not action.executable:
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"Correct the source-owned authorization before the "
"next Search rebuild."
),
evidence={"request_id": request_id},
)
)
continue
model = {
"search_index_document": SearchIndexDocument,
"search_index_change": SearchIndexChangeQueue,
}.get(action.resource_type)
if model is None or action.kind != "delete":
raise ValueError("Search DSAR executable action is not supported.")
row = (
db.query(model)
.filter(
model.tenant_id == tenant_id,
model.id == action.resource_id,
)
.one_or_none()
)
if row is None:
status = "unchanged"
summary = "Derived Search row was already absent."
else:
db.delete(row)
db.flush()
status = "executed"
summary = (
"Derived Search row removed; the source owner remains "
"authoritative and must be corrected before any rebuild."
)
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status=status,
summary=summary,
evidence={"request_id": request_id},
)
)
return tuple(results)
def _direct_rows(
session: Session,
*,
tenant_id: str,
selectors: _SubjectSelectors,
) -> tuple[list[SearchIndexDocument], list[SearchIndexChangeQueue]] | None:
documents: dict[str, SearchIndexDocument] = {}
changes: dict[str, SearchIndexChangeQueue] = {}
direct_identities: list[_ResourceIdentity] = []
source_identities: set[_ResourceIdentity] = set()
if selectors.document_id:
row = (
session.query(SearchIndexDocument)
.filter(
SearchIndexDocument.tenant_id == tenant_id,
SearchIndexDocument.id == selectors.document_id,
)
.one_or_none()
)
if row is None:
return None
documents[row.id] = row
direct_identities.append(_identity(row))
if selectors.change_id:
row = (
session.query(SearchIndexChangeQueue)
.filter(
SearchIndexChangeQueue.tenant_id == tenant_id,
SearchIndexChangeQueue.id == selectors.change_id,
)
.one_or_none()
)
if row is None:
return None
changes[row.id] = row
direct_identities.append(_identity(row))
if len(set(direct_identities)) > 1:
return None
for reference in selectors.source_references:
document_rows = (
session.query(SearchIndexDocument)
.filter(
SearchIndexDocument.tenant_id == tenant_id,
SearchIndexDocument.module_id == reference.module_id,
SearchIndexDocument.resource_type == reference.resource_type,
SearchIndexDocument.resource_id == reference.resource_id,
)
.order_by(SearchIndexDocument.id)
.limit(_MAX_RECORDS + 1)
.all()
)
change_rows = (
session.query(SearchIndexChangeQueue)
.filter(
SearchIndexChangeQueue.tenant_id == tenant_id,
SearchIndexChangeQueue.module_id == reference.module_id,
SearchIndexChangeQueue.resource_type == reference.resource_type,
SearchIndexChangeQueue.resource_id == reference.resource_id,
)
.order_by(SearchIndexChangeQueue.id)
.limit(_MAX_RECORDS + 1)
.all()
)
if not document_rows and not change_rows:
return None
source_identities.update(
_identity(row) for row in (*document_rows, *change_rows)
)
documents.update((row.id, row) for row in document_rows)
changes.update((row.id, row) for row in change_rows)
if direct_identities and selectors.source_references:
if not set(direct_identities).issubset(source_identities):
return None
if len(documents) + len(changes) > _MAX_RECORDS:
raise ValueError("Search DSAR result limit exceeded; narrow the selectors.")
return (
sorted(documents.values(), key=lambda row: row.id),
sorted(changes.values(), key=lambda row: row.id),
)
def _acl_rows(
session: Session,
*,
tenant_id: str,
selectors: _SubjectSelectors,
direct_identities: set[_ResourceIdentity],
) -> list[tuple[SearchIndexAclToken, SearchIndexDocument]]:
if not selectors.acl_tokens:
return []
query = (
session.query(SearchIndexAclToken, SearchIndexDocument)
.join(
SearchIndexDocument,
SearchIndexDocument.id == SearchIndexAclToken.document_id,
)
.filter(
SearchIndexDocument.tenant_id == tenant_id,
SearchIndexAclToken.token.in_(selectors.acl_tokens),
)
)
if direct_identities:
query = query.filter(
tuple_(
SearchIndexDocument.module_id,
SearchIndexDocument.resource_type,
SearchIndexDocument.resource_id,
).in_(
sorted(
(
identity.module_id,
identity.resource_type,
identity.resource_id,
)
for identity in direct_identities
)
)
)
rows = (
query.order_by(SearchIndexDocument.id, SearchIndexAclToken.id)
.limit(_MAX_RECORDS + 1)
.all()
)
if len(rows) > _MAX_RECORDS:
raise ValueError("Search DSAR result limit exceeded; narrow the selectors.")
return rows
def _identity(row: object) -> _ResourceIdentity:
return _ResourceIdentity(
module_id=str(getattr(row, "module_id")),
resource_type=str(getattr(row, "resource_type")),
resource_id=str(getattr(row, "resource_id")),
)
def _document_record(row: SearchIndexDocument) -> DsarRecordRef:
return _record(
"search_index_document",
row.id,
"derived_search_document",
"Derived Search document",
{
"module_id": row.module_id,
"provider_id": row.provider_id,
"resource_type": row.resource_type,
"resource_id": row.resource_id,
"source_revision": row.source_revision,
"source_updated_at": _iso(row.source_updated_at),
"visibility": row.visibility,
"index_version": row.index_version,
"requires_authorization_recheck": row.requires_authorization_recheck,
"active": row.active,
"indexed_at": _iso(row.indexed_at),
},
observed_at=row.indexed_at,
)
def _change_record(row: SearchIndexChangeQueue) -> DsarRecordRef:
return _record(
"search_index_change",
row.id,
"derived_search_change",
"Queued Search change",
{
"module_id": row.module_id,
"provider_id": row.provider_id,
"resource_type": row.resource_type,
"resource_id": row.resource_id,
"kind": row.kind,
"source_revision": row.source_revision,
"occurred_at": _iso(row.occurred_at),
"status": row.status,
"attempts": row.attempts,
"available_at": _iso(row.available_at),
"processed_at": _iso(row.processed_at),
},
observed_at=row.occurred_at or row.created_at,
)
def _acl_record(
row: SearchIndexAclToken,
*,
document: SearchIndexDocument,
) -> DsarRecordRef:
return _record(
"search_acl_projection",
row.id,
"derived_access_projection",
"Search access projection",
{
"module_id": document.module_id,
"resource_type": document.resource_type,
"resource_id": document.resource_id,
"token_kind": row.token.partition(":")[0],
"active": document.active,
"indexed_at": _iso(document.indexed_at),
},
observed_at=document.indexed_at,
)
def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
references = subject.external_references
values = {
"account_id": _coalesce(
subject.account_id,
references.get("search.account"),
references.get("access.account"),
),
"identity_id": _coalesce(
subject.identity_id,
references.get("search.identity"),
references.get("identity.id"),
),
"membership_id": _coalesce(
subject.membership_id,
references.get("search.membership"),
references.get("tenancy.membership"),
),
"document_id": _coalesce(
references.get("search.document"),
references.get("search.index_document"),
),
"change_id": _coalesce(
references.get("search.change"),
references.get("search.queued_change"),
),
}
if any(value is _CONFLICT for value in values.values()):
return None
source_references: list[_SourceReference] = []
for key, raw_value in sorted(references.items()):
if key in _RESERVED_REFERENCE_NAMES:
continue
module_id, separator, resource_type = key.partition(".")
value = str(raw_value or "").strip()
if (
not separator
or not module_id
or module_id == "search"
or not resource_type
or not value
):
continue
source_references.append(
_SourceReference(
module_id=module_id,
resource_type=resource_type,
resource_id=value,
)
)
return _SubjectSelectors(
account_id=_optional_string(values["account_id"]),
identity_id=_optional_string(values["identity_id"]),
membership_id=_optional_string(values["membership_id"]),
document_id=_optional_string(values["document_id"]),
change_id=_optional_string(values["change_id"]),
source_references=tuple(source_references),
)
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(
resource_type: str,
resource_id: str,
category: str,
title: str,
data: dict[str, object],
*,
observed_at: datetime | None,
) -> DsarRecordRef:
return DsarRecordRef(
provider_id="search",
module_id="search",
resource_type=resource_type,
resource_id=resource_id,
category=category,
title=title,
data={key: value for key, value in data.items() if value is not None},
observed_at=_aware(observed_at),
source_path="/admin/search",
)
def _iso(value: datetime | None) -> str | None:
aware = _aware(value)
return aware.isoformat() if aware else None
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("Search DSAR requires a SQLAlchemy Session.")
return value
def _validate_record(record: DsarRecordRef) -> None:
if record.provider_id != "search" or record.module_id != "search":
raise ValueError("Search DSAR cannot plan a foreign provider record.")
if not record.resource_type or not record.resource_id:
raise ValueError("Search DSAR record identity is incomplete.")
def _validate_action(action: DsarErasureActionRef) -> None:
if action.provider_id != "search" or action.module_id != "search":
raise ValueError("Search DSAR cannot execute a foreign provider action.")
if not action.action_id.startswith("search:"):
raise ValueError("Search DSAR action identity is invalid.")
__all__ = ["SEARCH_DSAR_CAPABILITY", "SearchDsarProvider"]
+47
View File
@@ -11,6 +11,7 @@ from govoplan_core.core.module_guards import (
persistent_table_uninstall_guard,
)
from govoplan_core.core.modules import (
CapabilityDocumentation,
DocumentationLink,
DocumentationTopic,
FrontendModule,
@@ -30,6 +31,10 @@ from govoplan_core.core.search import (
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
from govoplan_search.backend.db import models as search_models
from govoplan_search.backend.dsar_provider import (
SEARCH_DSAR_CAPABILITY,
SearchDsarProvider,
)
MODULE_ID = "search"
@@ -101,6 +106,11 @@ def _router(_context: ModuleContext):
return router
def _dsar_provider(context: ModuleContext) -> SearchDsarProvider:
del context
return SearchDsarProvider()
manifest = ModuleManifest(
id=MODULE_ID,
name=MODULE_NAME,
@@ -122,6 +132,7 @@ manifest = ModuleManifest(
ModuleInterfaceProvider(name="search.provider", version="1.0.0"),
ModuleInterfaceProvider(name="search.index_writer", version="1.1.0"),
ModuleInterfaceProvider(name="search.source", version="1.0.0"),
ModuleInterfaceProvider(name=SEARCH_DSAR_CAPABILITY, version="0.1.0"),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
@@ -189,6 +200,14 @@ manifest = ModuleManifest(
),
capability_factories={
CAPABILITY_SEARCH_INDEX_WRITER: _service,
SEARCH_DSAR_CAPABILITY: _dsar_provider,
},
capability_documentation={
SEARCH_DSAR_CAPABILITY: CapabilityDocumentation(
label="Search data-subject request provider",
summary="Finds and purges derived index copies while preserving source authority.",
contract_version="0.1.0",
),
},
search_providers=(
SearchProviderRegistration(
@@ -198,6 +217,34 @@ manifest = ModuleManifest(
),
),
documentation=(
DocumentationTopic(
id="search.data-subject-requests",
title="Derived Search data-subject requests",
summary="Remove derived index copies without treating Search as the authoritative data owner.",
body=(
"Search correlates explicit index/change identifiers and provider-owned source references inside the exact tenant. Account, identity, and membership selectors expose only minimized ACL projections and never imply ownership of the indexed source object. Indexed title, summary, body, search text, URL, keywords, metadata, external-reference payloads, ACL token values, hashes, cursors, queued documents, and errors are excluded. "
"Derived document and queued-change rows may be deleted idempotently. ACL-only matches require source-authority review. Source correction or erasure must happen in the owner module before Search is rebuilt; otherwise the source can legitimately republish the derived entry."
),
layer="configured",
documentation_types=("admin", "user"),
audience=("user", "operator", "module_admin", "auditor"),
related_modules=("core", "access"),
metadata={
"kind": "reference",
"help_contexts": [
"search.data-subject-requests",
"search.admin.index",
],
},
links=(
DocumentationLink(
label="Search index lifecycle",
href="govoplan-search/README.md",
kind="repository",
),
),
order=11,
),
DocumentationTopic(
id="search.global-and-contextual",
title="Global and contextual search",