test(privacy): enforce workspace DSAR coverage
This commit is contained in:
@@ -0,0 +1,265 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Require DSAR coverage or a reviewed no-store rationale for every module."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import importlib
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
META_ROOT = Path(__file__).resolve().parents[2]
|
||||
EXEMPTIONS_PATH = Path(__file__).with_name("dsar-coverage-exemptions.json")
|
||||
REPORT_PATH = (
|
||||
META_ROOT
|
||||
/ "docs"
|
||||
/ "evidence"
|
||||
/ "snapshots"
|
||||
/ "DSAR_PROVIDER_COVERAGE.generated.md"
|
||||
)
|
||||
MODULE_NAME_PATTERN = re.compile(r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*")
|
||||
REQUIRED_DOCUMENTATION_TYPES = frozenset({"admin"})
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CoverageRow:
|
||||
module_id: str
|
||||
repository: str
|
||||
migration_owned: bool
|
||||
capability: str | None
|
||||
rationale: str
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument(
|
||||
"--workspace-root",
|
||||
type=Path,
|
||||
default=None,
|
||||
help="Directory containing GovOPlaN repositories.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--render",
|
||||
action="store_true",
|
||||
help="Print the current matrix instead of comparing the checked-in report.",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
catalog = json.loads((META_ROOT / "repositories.json").read_text(encoding="utf-8"))
|
||||
workspace_root = (args.workspace_root or Path(catalog["default_parent"])).resolve()
|
||||
exemptions = _exemptions()
|
||||
manifests, load_errors = _load_manifests(
|
||||
workspace_root=workspace_root,
|
||||
repositories=tuple(catalog["repositories"]),
|
||||
)
|
||||
|
||||
errors = list(load_errors)
|
||||
rows: list[CoverageRow] = []
|
||||
manifest_ids = {manifest.id for _, manifest in manifests}
|
||||
stale_exemptions = sorted(set(exemptions) - manifest_ids)
|
||||
if stale_exemptions:
|
||||
errors.append(
|
||||
"DSAR coverage exemptions reference unknown modules: "
|
||||
+ ", ".join(stale_exemptions)
|
||||
)
|
||||
|
||||
for repository, manifest in manifests:
|
||||
expected = f"privacy.dsar.{manifest.id}"
|
||||
provided = {
|
||||
item.name
|
||||
for item in manifest.provides_interfaces
|
||||
if item.name.startswith("privacy.dsar.")
|
||||
}
|
||||
factories = {
|
||||
name
|
||||
for name in manifest.capability_factories
|
||||
if name.startswith("privacy.dsar.")
|
||||
}
|
||||
migration_owned = manifest.migration_spec is not None
|
||||
rationale = exemptions.get(manifest.id)
|
||||
|
||||
if provided != factories:
|
||||
errors.append(
|
||||
f"{repository}: DSAR interface/factory mismatch: "
|
||||
f"interfaces={sorted(provided)!r}, factories={sorted(factories)!r}"
|
||||
)
|
||||
if provided and provided != {expected}:
|
||||
errors.append(
|
||||
f"{repository}: expected only {expected!r}, found {sorted(provided)!r}"
|
||||
)
|
||||
|
||||
capability = (
|
||||
expected if expected in provided and expected in factories else None
|
||||
)
|
||||
if migration_owned and capability is None:
|
||||
errors.append(
|
||||
f"{repository}: migration-owning module {manifest.id!r} must provide "
|
||||
f"and register {expected!r}"
|
||||
)
|
||||
if migration_owned and rationale is not None:
|
||||
errors.append(
|
||||
f"{repository}: migration-owning module {manifest.id!r} cannot use a "
|
||||
"no-store DSAR exemption"
|
||||
)
|
||||
if not migration_owned and capability is None and rationale is None:
|
||||
errors.append(
|
||||
f"{repository}: module {manifest.id!r} needs a DSAR provider or an "
|
||||
"explicit reviewed no-store rationale"
|
||||
)
|
||||
if capability is not None and rationale is not None:
|
||||
errors.append(
|
||||
f"{repository}: module {manifest.id!r} has both DSAR coverage and a "
|
||||
"stale exemption"
|
||||
)
|
||||
if capability is not None:
|
||||
if capability not in manifest.capability_documentation:
|
||||
errors.append(
|
||||
f"{repository}: {capability!r} lacks capability documentation"
|
||||
)
|
||||
matching_topics = tuple(
|
||||
topic
|
||||
for topic in manifest.documentation
|
||||
if "data-subject-request" in topic.id
|
||||
)
|
||||
if not matching_topics or not any(
|
||||
REQUIRED_DOCUMENTATION_TYPES.issubset(topic.documentation_types)
|
||||
for topic in matching_topics
|
||||
):
|
||||
errors.append(
|
||||
f"{repository}: DSAR coverage needs a static administrator "
|
||||
"data-subject-requests DocumentationTopic"
|
||||
)
|
||||
|
||||
rows.append(
|
||||
CoverageRow(
|
||||
module_id=manifest.id,
|
||||
repository=repository,
|
||||
migration_owned=migration_owned,
|
||||
capability=capability,
|
||||
rationale=(
|
||||
f"Provider `{capability}` is registered and documented."
|
||||
if capability
|
||||
else rationale or "MISSING"
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
report = _report(rows)
|
||||
if args.render:
|
||||
print(report, end="")
|
||||
elif not REPORT_PATH.is_file():
|
||||
errors.append(f"DSAR coverage report is missing: {REPORT_PATH}")
|
||||
elif REPORT_PATH.read_text(encoding="utf-8") != report:
|
||||
errors.append(
|
||||
"DSAR coverage report is stale; review changes and replace it with "
|
||||
"the output of tools/checks/check-dsar-coverage.py --render"
|
||||
)
|
||||
|
||||
if errors:
|
||||
print("\n".join(errors), file=sys.stderr)
|
||||
return 1
|
||||
|
||||
provider_count = sum(row.capability is not None for row in rows)
|
||||
print(
|
||||
"DSAR coverage check passed: "
|
||||
f"{provider_count} providers, {len(rows) - provider_count} reviewed "
|
||||
f"no-store rationales, {len(rows)} active modules."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def _exemptions() -> dict[str, str]:
|
||||
values = json.loads(EXEMPTIONS_PATH.read_text(encoding="utf-8"))
|
||||
if not isinstance(values, dict) or any(
|
||||
not isinstance(key, str) or not isinstance(value, str) or not value.strip()
|
||||
for key, value in values.items()
|
||||
):
|
||||
raise ValueError("DSAR coverage exemptions must be non-empty string mappings.")
|
||||
return {key: value.strip() for key, value in values.items()}
|
||||
|
||||
|
||||
def _load_manifests(*, workspace_root: Path, repositories: tuple[dict, ...]):
|
||||
sources: list[Path] = []
|
||||
candidates: list[tuple[str, Path, Path]] = []
|
||||
for repository in repositories:
|
||||
source = workspace_root / repository["path"] / "src"
|
||||
if not source.is_dir():
|
||||
continue
|
||||
sources.append(source)
|
||||
candidates.extend(
|
||||
(repository["name"], source, path)
|
||||
for path in sorted(source.glob("*/backend/manifest.py"))
|
||||
)
|
||||
core_source = workspace_root / "govoplan-core" / "src"
|
||||
sys.path[:0] = [
|
||||
str(core_source),
|
||||
*(str(source) for source in sources if source != core_source),
|
||||
]
|
||||
|
||||
manifests = []
|
||||
errors = []
|
||||
for repository, source, path in candidates:
|
||||
module_name = ".".join(path.relative_to(source).with_suffix("").parts)
|
||||
if MODULE_NAME_PATTERN.fullmatch(module_name) is None:
|
||||
errors.append(f"{repository}: unsafe manifest module name {module_name!r}")
|
||||
continue
|
||||
try:
|
||||
module = importlib.import_module(module_name)
|
||||
manifests.append((repository, module.get_manifest()))
|
||||
except Exception as exc: # pragma: no cover - emitted as check evidence
|
||||
errors.append(f"{repository}: could not load {module_name}: {exc}")
|
||||
return manifests, errors
|
||||
|
||||
|
||||
def _report(rows: list[CoverageRow]) -> str:
|
||||
ordered = sorted(rows, key=lambda row: row.module_id)
|
||||
providers = sum(row.capability is not None for row in ordered)
|
||||
lines = [
|
||||
"# DSAR Provider Coverage",
|
||||
"",
|
||||
"This generated matrix is enforced by `tools/checks/check-dsar-coverage.py`.",
|
||||
"A migration-owning module must register and document its canonical DSAR provider.",
|
||||
"Every other active module requires a reviewed explanation of why it owns no",
|
||||
"persistent subject-data store. Adding a migration invalidates that explanation.",
|
||||
"",
|
||||
f"- Active modules: {len(ordered)}",
|
||||
f"- Registered and documented DSAR providers: {providers}",
|
||||
f"- Reviewed no-store rationales: {len(ordered) - providers}",
|
||||
"- Unexplained coverage gaps: 0",
|
||||
"",
|
||||
"| Module | Repository | Persistence | Coverage | Rationale |",
|
||||
"| --- | --- | --- | --- | --- |",
|
||||
]
|
||||
for row in ordered:
|
||||
lines.append(
|
||||
"| "
|
||||
+ " | ".join(
|
||||
(
|
||||
f"`{row.module_id}`",
|
||||
f"`{row.repository}`",
|
||||
"Migration-owned" if row.migration_owned else "No module migration",
|
||||
"Provider" if row.capability else "Reviewed no-store rationale",
|
||||
row.rationale.replace("|", "\\|"),
|
||||
)
|
||||
)
|
||||
+ " |"
|
||||
)
|
||||
lines.extend(
|
||||
(
|
||||
"",
|
||||
"Provider search, export minimization, retention, and erasure behavior remains",
|
||||
"documented and tested by each owning module. This matrix verifies adoption and",
|
||||
"ownership coverage; Core continues to test disabled providers, partial failure,",
|
||||
"retry, authorization evidence, and horizontally coordinated execution.",
|
||||
"",
|
||||
)
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user