diff --git a/src/govoplan_docs/backend/api/v1/routes.py b/src/govoplan_docs/backend/api/v1/routes.py index 5ecc00d..a04171e 100644 --- a/src/govoplan_docs/backend/api/v1/routes.py +++ b/src/govoplan_docs/backend/api/v1/routes.py @@ -7,6 +7,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from govoplan_core.auth import ApiPrincipal, has_scope, require_any_scope from govoplan_core.core.modules import ( + DocumentationConfigurationDecision, DocumentationCondition, DocumentationContext, DocumentationLink, @@ -22,10 +23,18 @@ from govoplan_core.core.registry import PlatformRegistry from govoplan_core.db.session import get_database from govoplan_docs.backend.manifest import DOCS_ADMIN_READ_SCOPES, DOCS_READ_SCOPES +from govoplan_docs.backend.sources import ( + RegisteredDocumentationSource, + build_documentation_source_registry, +) router = APIRouter(prefix="/docs", tags=["docs"]) TOPIC_KINDS = ("workflow", "reference", "pattern", "system") +_CONFIGURATION_ACTIVE_STATES = frozenset({"enabled", "inherited"}) +_CONFIGURATION_STATES = frozenset( + {"enabled", "disabled", "inherited", "unavailable"} +) @router.get("/context") @@ -46,6 +55,17 @@ def docs_context( route_items = _route_items(registry.manifests(), principal) visible_route_items = [item for item in route_items if item["visible"]] documentation_layers = _documentation_layers(request, registry, principal, documentation_type=documentation_type, locale=resolved_locale) + evidence_sources = ( + _documentation_source_summaries( + request, + registry, + principal, + documentation_type=documentation_type, + locale=resolved_locale, + ) + if documentation_type == "admin" + else [] + ) if documentation_type == "admin": catalog = _admin_documentation_catalog( registry, @@ -76,11 +96,81 @@ def docs_context( "layers": _documentation_layer_payload( catalog, documentation_layers=documentation_layers, - include_evidence_sources=documentation_type == "admin", + evidence_sources=evidence_sources, ), } +@router.get("/sources") +def list_documentation_sources( + request: Request, + documentation_type: DocumentationType = Query( + default="admin", + alias="type", + pattern="^(admin|user)$", + ), + locale: str | None = Query(default=None, min_length=2, max_length=20), + principal: ApiPrincipal = Depends(require_any_scope(*DOCS_READ_SCOPES)), +) -> dict[str, Any]: + _require_documentation_type_access(principal, documentation_type) + registry = _registry(request) + items = _documentation_source_items( + request, + registry, + principal, + documentation_type=documentation_type, + locale=_preferred_locale(request, locale), + ) + return { + "items": [source.item.summary() for source in items], + "total": len(items), + } + + +@router.get("/sources/{source_id}") +def inspect_documentation_source( + source_id: str, + request: Request, + documentation_type: DocumentationType = Query( + default="admin", + alias="type", + pattern="^(admin|user)$", + ), + locale: str | None = Query(default=None, min_length=2, max_length=20), + principal: ApiPrincipal = Depends(require_any_scope(*DOCS_READ_SCOPES)), +) -> dict[str, Any]: + _require_documentation_type_access(principal, documentation_type) + registry = _registry(request) + items = _documentation_source_items( + request, + registry, + principal, + documentation_type=documentation_type, + locale=_preferred_locale(request, locale), + ) + source = next((item for item in items if item.item.id == source_id), None) + if source is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Documentation source not found", + ) + return source.item.model_dump(mode="json") + + +def _require_documentation_type_access( + principal: ApiPrincipal, + documentation_type: DocumentationType, +) -> None: + if documentation_type == "admin" and not _has_any_scope( + principal, + DOCS_ADMIN_READ_SCOPES, + ): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Administrative documentation requires documentation-administrator authority", + ) + + def _admin_documentation_catalog( registry: PlatformRegistry, principal: ApiPrincipal, @@ -186,7 +276,7 @@ def _documentation_layer_payload( catalog: Mapping[str, list[dict[str, Any]]], *, documentation_layers: Mapping[str, list[dict[str, Any]]], - include_evidence_sources: bool, + evidence_sources: list[dict[str, Any]], ) -> dict[str, dict[str, object]]: permissions = catalog["permissions"] return { @@ -206,7 +296,7 @@ def _documentation_layer_payload( }, "evidence": { "optional_modules": catalog["optional_modules"], - "sources": _evidence_sources() if include_evidence_sources else [], + "sources": evidence_sources, "documentation": documentation_layers["evidence"], }, } @@ -398,6 +488,138 @@ def _settings(request: Request) -> object | None: return getattr(lifecycle, "settings", None) +def _documentation_source_summaries( + request: Request, + registry: PlatformRegistry, + principal: ApiPrincipal, + *, + documentation_type: DocumentationType, + locale: str, +) -> list[dict[str, Any]]: + return [ + source.item.summary() + for source in _documentation_source_items( + request, + registry, + principal, + documentation_type=documentation_type, + locale=locale, + ) + ] + + +def _documentation_source_items( + request: Request, + registry: PlatformRegistry, + principal: ApiPrincipal, + *, + documentation_type: DocumentationType, + locale: str, +) -> list[RegisteredDocumentationSource]: + source_registry = build_documentation_source_registry(registry.manifests()) + settings = _settings(request) + try: + with get_database().SessionLocal() as session: + return _visible_documentation_sources( + source_registry.sources(), + registry, + principal, + settings=settings, + session=session, + documentation_type=documentation_type, + locale=locale, + ) + except RuntimeError: + return _visible_documentation_sources( + source_registry.sources(), + registry, + principal, + settings=settings, + session=None, + documentation_type=documentation_type, + locale=locale, + ) + + +def _visible_documentation_sources( + sources: tuple[RegisteredDocumentationSource, ...], + registry: PlatformRegistry, + principal: ApiPrincipal, + *, + settings: object | None, + session: object | None, + documentation_type: DocumentationType, + locale: str, +) -> list[RegisteredDocumentationSource]: + installed = {manifest.id for manifest in registry.manifests()} + context = DocumentationContext( + registry=registry, + principal=principal, + settings=settings, + session=session, + documentation_type=documentation_type, + locale=locale, + ) + visible: list[RegisteredDocumentationSource] = [] + for source in sources: + if documentation_type not in source.documentation_types: + continue + keys = tuple(dict.fromkeys(( + *source.condition.configuration_keys, + *([source.configuration_key] if source.configuration_key else []), + ))) + configuration = _resolve_documentation_configuration( + registry, + source.item.owner_module_id, + keys, + context=context, + ) + active, reason, blockers = _condition_visibility( + source.condition, + installed, + registry, + principal, + configuration=configuration, + ) + if blockers["scopes"]: + continue + state = source.item.state + source_configuration = ( + configuration.get(source.configuration_key) + if source.configuration_key + else None + ) + if state == "configured" and source_configuration is not None: + if source_configuration.state == "unavailable": + state = "unavailable" + reason = source_configuration.reason or reason + elif source_configuration.state == "disabled": + state = "disabled" + reason = source_configuration.reason or reason + if state == "configured" and not active: + state = ( + "unavailable" + if any( + decision.state == "unavailable" + for decision in configuration.values() + ) + or blockers["modules"] + or blockers["capabilities"] + else "disabled" + ) + item = source.item.model_copy(update={ + "state": state, + "state_reason": None if state == "configured" else reason, + }) + visible.append(RegisteredDocumentationSource( + item=item, + condition=source.condition, + documentation_types=source.documentation_types, + configuration_key=source.configuration_key, + )) + return visible + + def _classify_documentation( registry: PlatformRegistry, principal: ApiPrincipal, @@ -417,14 +639,40 @@ def _classify_documentation( if item["visible"] ), ]) if documentation_type == "user" else frozenset() + context = DocumentationContext( + registry=registry, + principal=principal, + settings=settings, + session=session, + documentation_type=documentation_type, + locale=locale, + ) topics = _collect_documentation_topics(registry, principal, settings=settings, session=session, documentation_type=documentation_type, locale=locale) for source_module_id, topic in sorted(topics, key=lambda item: (item[1].order, item[0], item[1].id)): if not _topic_matches_documentation_type(topic, documentation_type): continue - active, reason, blockers = _documentation_visibility(topic, installed, registry, principal) + configuration_keys = _documentation_configuration_keys(topic) + configuration = _resolve_documentation_configuration( + registry, + topic.source_module_id or source_module_id, + configuration_keys, + context=context, + ) + active, reason, blockers = _documentation_visibility( + topic, + installed, + registry, + principal, + configuration=configuration, + ) if documentation_type == "user" and not active: continue - target_layer = _documentation_target_layer(topic, active, blockers) + target_layer = _documentation_target_layer( + topic, + active, + blockers, + configuration=configuration, + ) if target_layer not in layers: target_layer = "evidence" layers[target_layer].append(_documentation_topic_payload( @@ -437,6 +685,7 @@ def _classify_documentation( locale=locale, documentation_type=documentation_type, visible_runtime_paths=visible_runtime_paths, + configuration=configuration, )) return layers @@ -494,21 +743,130 @@ def _topic_matches_documentation_type(topic: DocumentationTopic, documentation_t return documentation_type in (topic.documentation_types or ("admin",)) +def _documentation_configuration_keys( + topic: DocumentationTopic, +) -> tuple[str, ...]: + return tuple(dict.fromkeys(( + *topic.configuration_keys, + *( + key + for condition in topic.conditions + for key in condition.configuration_keys + ), + ))) + + +def _resolve_documentation_configuration( + registry: PlatformRegistry, + module_id: str, + keys: tuple[str, ...], + *, + context: DocumentationContext, +) -> dict[str, DocumentationConfigurationDecision]: + decisions = { + key: DocumentationConfigurationDecision( + key=key, + state="unavailable", + reason="No configuration-state provider is registered.", + ) + for key in keys + } + manifest = registry.get(module_id) + if manifest is None or not keys: + return decisions + + for registration in manifest.documentation_configuration_providers: + requested = tuple(key for key in keys if key in registration.keys) + if not requested: + continue + try: + provided = registration.resolve(context, requested) + except Exception as exc: + for key in requested: + decisions[key] = DocumentationConfigurationDecision( + key=key, + state="unavailable", + reason=f"Configuration-state provider failed ({type(exc).__name__}).", + ) + continue + for key in requested: + decision = provided.get(key) + if ( + not isinstance(decision, DocumentationConfigurationDecision) + or decision.key != key + or decision.state not in _CONFIGURATION_STATES + ): + decisions[key] = DocumentationConfigurationDecision( + key=key, + state="unavailable", + reason="Configuration-state provider returned an invalid decision.", + ) + continue + decisions[key] = DocumentationConfigurationDecision( + key=key, + state=decision.state, + source=_bounded_configuration_text(decision.source), + reason=_bounded_configuration_text(decision.reason), + ) + return decisions + + +def _bounded_configuration_text(value: object | None) -> str | None: + if value is None: + return None + clean = str(value).strip() + return clean[:500] if clean else None + + +def _documentation_configuration_payload( + decision: DocumentationConfigurationDecision, +) -> dict[str, str | None]: + return { + "key": decision.key, + "state": decision.state, + "source": decision.source, + "reason": decision.reason, + } + + def _documentation_visibility( topic: DocumentationTopic, installed: set[str], registry: PlatformRegistry, principal: ApiPrincipal, + *, + configuration: Mapping[str, DocumentationConfigurationDecision] | None = None, ) -> tuple[bool, str, dict[str, list[str]]]: if not topic.conditions: - return True, "documented", {"modules": [], "capabilities": [], "scopes": []} + return True, "documented", { + "modules": [], + "capabilities": [], + "scopes": [], + "configuration": [], + } reasons: list[str] = [] - blockers = {"modules": [], "capabilities": [], "scopes": []} + blockers = { + "modules": [], + "capabilities": [], + "scopes": [], + "configuration": [], + } for condition in topic.conditions: - active, reason, condition_blockers = _condition_visibility(condition, installed, registry, principal) + active, reason, condition_blockers = _condition_visibility( + condition, + installed, + registry, + principal, + configuration=configuration, + ) if active: - return True, reason, {"modules": [], "capabilities": [], "scopes": []} + return True, reason, { + "modules": [], + "capabilities": [], + "scopes": [], + "configuration": [], + } reasons.append(reason) for key, values in condition_blockers.items(): blockers[key].extend(value for value in values if value not in blockers[key]) @@ -520,6 +878,8 @@ def _condition_visibility( installed: set[str], registry: PlatformRegistry, principal: ApiPrincipal, + *, + configuration: Mapping[str, DocumentationConfigurationDecision] | None = None, ) -> tuple[bool, str, dict[str, list[str]]]: missing_required_modules = _missing_required_modules(condition, installed) unsatisfied_any_modules = _unsatisfied_any_modules(condition, installed) @@ -527,6 +887,10 @@ def _condition_visibility( missing_capabilities = _missing_required_capabilities(condition, registry) missing_scopes = _missing_required_scopes(condition, principal) unsatisfied_any_scopes = _unsatisfied_any_scopes(condition, principal) + unavailable_configuration = _unavailable_configuration_keys( + condition, + configuration or {}, + ) blockers = _condition_blockers( missing_required_modules=missing_required_modules, unsatisfied_any_modules=unsatisfied_any_modules, @@ -534,6 +898,7 @@ def _condition_visibility( missing_capabilities=missing_capabilities, missing_scopes=missing_scopes, unsatisfied_any_scopes=unsatisfied_any_scopes, + unavailable_configuration=unavailable_configuration, ) if _condition_has_no_blockers(blockers): @@ -546,6 +911,7 @@ def _condition_visibility( missing_capabilities=missing_capabilities, missing_scopes=missing_scopes, unsatisfied_any_scopes=unsatisfied_any_scopes, + unavailable_configuration=unavailable_configuration, ) return False, reason, blockers @@ -582,6 +948,20 @@ def _unsatisfied_any_scopes(condition: DocumentationCondition, principal: ApiPri return list(condition.any_scopes) +def _unavailable_configuration_keys( + condition: DocumentationCondition, + configuration: Mapping[str, DocumentationConfigurationDecision], +) -> list[str]: + return [ + key + for key in condition.configuration_keys + if configuration.get( + key, + DocumentationConfigurationDecision(key=key, state="unavailable"), + ).state not in _CONFIGURATION_ACTIVE_STATES + ] + + def _condition_blockers( *, missing_required_modules: list[str], @@ -590,14 +970,21 @@ def _condition_blockers( missing_capabilities: list[str], missing_scopes: list[str], unsatisfied_any_scopes: list[str], + unavailable_configuration: list[str], ) -> dict[str, list[str]]: - blockers = {"modules": [], "capabilities": [], "scopes": []} + blockers = { + "modules": [], + "capabilities": [], + "scopes": [], + "configuration": [], + } _extend_unique(blockers["modules"], missing_required_modules) _extend_unique(blockers["modules"], unsatisfied_any_modules) _extend_unique(blockers["modules"], conflicting_modules) _extend_unique(blockers["capabilities"], missing_capabilities) _extend_unique(blockers["scopes"], missing_scopes) _extend_unique(blockers["scopes"], unsatisfied_any_scopes) + _extend_unique(blockers["configuration"], unavailable_configuration) return blockers @@ -606,7 +993,7 @@ def _extend_unique(target: list[str], values: list[str]) -> None: def _condition_has_no_blockers(blockers: dict[str, list[str]]) -> bool: - return not blockers["modules"] and not blockers["capabilities"] and not blockers["scopes"] + return not any(blockers.values()) def _condition_blocker_reason( @@ -617,6 +1004,7 @@ def _condition_blocker_reason( missing_capabilities: list[str], missing_scopes: list[str], unsatisfied_any_scopes: list[str], + unavailable_configuration: list[str], ) -> str: parts: list[str] = [] if missing_required_modules: @@ -631,16 +1019,32 @@ def _condition_blocker_reason( parts.append("missing scopes: " + ", ".join(missing_scopes)) if unsatisfied_any_scopes: parts.append("requires one scope from: " + ", ".join(unsatisfied_any_scopes)) + if unavailable_configuration: + parts.append( + "configuration is disabled or unavailable: " + + ", ".join(unavailable_configuration) + ) return "; ".join(parts) -def _documentation_target_layer(topic: DocumentationTopic, active: bool, blockers: dict[str, list[str]]) -> str: +def _documentation_target_layer( + topic: DocumentationTopic, + active: bool, + blockers: dict[str, list[str]], + *, + configuration: Mapping[str, DocumentationConfigurationDecision] | None = None, +) -> str: if topic.layer == "always": return "always" if active: return topic.layer if blockers["modules"] or blockers["capabilities"]: return "evidence" + if any( + decision.state == "unavailable" + for decision in (configuration or {}).values() + ): + return "evidence" return "available" @@ -655,6 +1059,7 @@ def _documentation_topic_payload( locale: str, documentation_type: DocumentationType, visible_runtime_paths: frozenset[str], + configuration: Mapping[str, DocumentationConfigurationDecision], ) -> dict[str, Any]: module_id = topic.source_module_id or source_module_id translation_locale, translation = _translation_for_locale(topic, locale) @@ -676,6 +1081,9 @@ def _documentation_topic_payload( "modules": sorted(dict.fromkeys(blockers.get("modules", ()))), "capabilities": sorted(dict.fromkeys(blockers.get("capabilities", ()))), "scopes": sorted(dict.fromkeys(blockers.get("scopes", ()))), + "configuration": sorted( + dict.fromkeys(blockers.get("configuration", ())) + ), }, "audience": list(topic.audience), "order": topic.order, @@ -687,6 +1095,10 @@ def _documentation_topic_payload( "related_modules": list(topic.related_modules), "unlocks": list(topic.unlocks), "configuration_keys": sorted({*topic.configuration_keys, *(key for condition in topic.conditions for key in condition.configuration_keys)}), + "configuration_states": [ + _documentation_configuration_payload(configuration[key]) + for key in sorted(configuration) + ], "metadata": dict(topic.metadata), } if documentation_type == "admin": @@ -704,7 +1116,12 @@ def _documentation_topic_payload( "documentation_types": ["user"], "active": True, "reason": "documented", - "blockers": {"modules": [], "capabilities": [], "scopes": []}, + "blockers": { + "modules": [], + "capabilities": [], + "scopes": [], + "configuration": [], + }, "audience": [], "order": payload["order"], "i18n_key": "", @@ -719,6 +1136,7 @@ def _documentation_topic_payload( "related_modules": [], "unlocks": list(topic.unlocks), "configuration_keys": [], + "configuration_states": [], "metadata": _user_topic_metadata(kind, topic.metadata), } @@ -909,18 +1327,3 @@ def _documentation_link_payload(link: DocumentationLink) -> dict[str, str]: "href": link.href, "kind": link.kind, } - - -def _evidence_sources() -> list[dict[str, str]]: - return [ - { - "label": "Documentation layer concept", - "source": "govoplan-docs/docs/DOCUMENTATION_LAYER_CONCEPT.md", - "layer": "evidence", - }, - { - "label": "Master roadmap", - "source": "govoplan-core/docs/GOVOPLAN_MASTER_ROADMAP.md", - "layer": "evidence", - }, - ] diff --git a/src/govoplan_docs/backend/manifest.py b/src/govoplan_docs/backend/manifest.py index 6c1ec21..e2c69fe 100644 --- a/src/govoplan_docs/backend/manifest.py +++ b/src/govoplan_docs/backend/manifest.py @@ -4,6 +4,7 @@ from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPA from govoplan_core.core.modules import ( DocumentationCondition, DocumentationLink, + DocumentationSourceDefinition, DocumentationTopic, FrontendModule, FrontendRoute, @@ -193,6 +194,39 @@ manifest = ModuleManifest( }, ), ), + documentation_sources=( + DocumentationSourceDefinition( + id="docs.configuration.packages", + kind="configuration_package", + label="Configuration package catalog", + condition=DocumentationCondition( + any_scopes=("system:settings:read", "admin:settings:read"), + ), + provenance={ + "source": "core_configuration_package_contract", + "version": "1", + }, + inspection={ + "package_id": "govoplan.configuration-packages", + "schema_version": "1", + "description": "Signed, preflighted platform configuration packages.", + }, + ), + DocumentationSourceDefinition( + id="docs.project.wiki", + kind="wiki", + label="GovOPlaN Docs wiki", + provenance={"source": "gitea_wiki"}, + link=DocumentationLink( + label="GovOPlaN Docs wiki", + href="https://git.add-ideas.de/GovOPlaN/govoplan-docs/wiki", + kind="wiki", + ), + inspection={ + "href": "https://git.add-ideas.de/GovOPlaN/govoplan-docs/wiki", + }, + ), + ), ) diff --git a/src/govoplan_docs/backend/sources.py b/src/govoplan_docs/backend/sources.py new file mode 100644 index 0000000..ff4a3a1 --- /dev/null +++ b/src/govoplan_docs/backend/sources.py @@ -0,0 +1,391 @@ +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from typing import Annotated, Any, Literal, Mapping + +from pydantic import BaseModel, ConfigDict, Field + +from govoplan_core.core.modules import ( + DocumentationCondition, + DocumentationLink, + DocumentationSourceDefinition, + DocumentationType, + ModuleManifest, +) +from govoplan_core.security.redaction import redact_secret_values + + +class _SourceModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class DocumentationSourceProvenance(_SourceModel): + source: str + version: str | None = None + revision: str | None = None + published_at: str | None = None + checksum: str | None = None + + +class DocumentationSourceVisibility(_SourceModel): + documentation_types: list[DocumentationType] + required_modules: list[str] = Field(default_factory=list) + any_modules: list[str] = Field(default_factory=list) + missing_modules: list[str] = Field(default_factory=list) + required_capabilities: list[str] = Field(default_factory=list) + required_scopes: list[str] = Field(default_factory=list) + any_scopes: list[str] = Field(default_factory=list) + configuration_keys: list[str] = Field(default_factory=list) + + +class ManifestInspection(_SourceModel): + kind: Literal["manifest"] = "manifest" + module_id: str + name: str + version: str + dependencies: list[str] + optional_dependencies: list[str] + + +class RouteInspection(_SourceModel): + kind: Literal["route"] = "route" + path: str + component: str | None = None + order: int + + +class CapabilityInspection(_SourceModel): + kind: Literal["capability"] = "capability" + capability: str + + +class PolicyInspection(_SourceModel): + kind: Literal["policy"] = "policy" + capability: str + + +class ConfigurationPackageInspection(_SourceModel): + kind: Literal["configuration_package"] = "configuration_package" + package_id: str + schema_version: str | None = None + description: str | None = None + + +class WikiInspection(_SourceModel): + kind: Literal["wiki"] = "wiki" + href: str + published_at: str | None = None + revision: str | None = None + + +class RepositoryInspection(_SourceModel): + kind: Literal["repository"] = "repository" + href: str + revision: str | None = None + + +DocumentationSourceInspection = Annotated[ + ManifestInspection + | RouteInspection + | CapabilityInspection + | PolicyInspection + | ConfigurationPackageInspection + | WikiInspection + | RepositoryInspection, + Field(discriminator="kind"), +] + + +class DocumentationSourceItem(_SourceModel): + id: str + kind: str + owner_module_id: str + label: str + state: Literal["configured", "disabled", "unavailable"] + state_reason: str | None = None + inspection_url: str + provenance: DocumentationSourceProvenance + visibility: DocumentationSourceVisibility + inspection: DocumentationSourceInspection + + def summary(self) -> dict[str, Any]: + return self.model_dump(mode="json", exclude={"inspection"}) + + +@dataclass(frozen=True, slots=True) +class RegisteredDocumentationSource: + item: DocumentationSourceItem + condition: DocumentationCondition + documentation_types: tuple[DocumentationType, ...] + configuration_key: str | None = None + + +class DocumentationSourceRegistry: + def __init__(self) -> None: + self._sources: dict[str, RegisteredDocumentationSource] = {} + + def register(self, source: RegisteredDocumentationSource) -> None: + if source.item.id in self._sources: + raise ValueError(f"Duplicate documentation source id: {source.item.id}") + self._sources[source.item.id] = source + + def sources(self) -> tuple[RegisteredDocumentationSource, ...]: + return tuple( + self._sources[source_id] + for source_id in sorted(self._sources) + ) + + def get(self, source_id: str) -> RegisteredDocumentationSource | None: + return self._sources.get(source_id) + + +def build_documentation_source_registry( + manifests: tuple[ModuleManifest, ...], +) -> DocumentationSourceRegistry: + registry = DocumentationSourceRegistry() + for manifest in manifests: + registry.register(_manifest_source(manifest)) + for source in _route_sources(manifest): + registry.register(source) + for source in _capability_sources(manifest): + registry.register(source) + for definition in manifest.documentation_sources: + registry.register(_defined_source(manifest, definition)) + for source in _linked_sources(manifest): + if registry.get(source.item.id) is None: + registry.register(source) + return registry + + +def _manifest_source(manifest: ModuleManifest) -> RegisteredDocumentationSource: + return _registered_source( + source_id=f"{manifest.id}.manifest", + kind="manifest", + owner_module_id=manifest.id, + label=f"{manifest.name} module manifest", + provenance={"source": "module_manifest", "version": manifest.version}, + inspection=ManifestInspection( + module_id=manifest.id, + name=manifest.name, + version=manifest.version, + dependencies=list(manifest.dependencies), + optional_dependencies=list(manifest.optional_dependencies), + ), + ) + + +def _route_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSource, ...]: + if manifest.frontend is None: + return () + return tuple( + _registered_source( + source_id=_derived_source_id(manifest.id, "route", route.path), + kind="route", + owner_module_id=manifest.id, + label=f"{manifest.name} route {route.path}", + provenance={"source": "frontend_manifest", "version": manifest.version}, + inspection=RouteInspection( + path=route.path, + component=route.component, + order=route.order, + ), + condition=DocumentationCondition( + required_modules=(manifest.id,), + required_scopes=route.required_all, + any_scopes=route.required_any, + ), + ) + for route in manifest.frontend.routes + ) + + +def _capability_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSource, ...]: + sources: list[RegisteredDocumentationSource] = [] + for capability in sorted(manifest.capability_factories): + is_policy = capability.startswith("policy.") + kind = "policy" if is_policy else "capability" + inspection: DocumentationSourceInspection = ( + PolicyInspection(capability=capability) + if is_policy + else CapabilityInspection(capability=capability) + ) + sources.append(_registered_source( + source_id=_derived_source_id(manifest.id, kind, capability), + kind=kind, + owner_module_id=manifest.id, + label=f"{manifest.name} {kind} {capability}", + provenance={"source": "module_manifest", "version": manifest.version}, + inspection=inspection, + condition=DocumentationCondition(required_modules=(manifest.id,)), + )) + return tuple(sources) + + +def _defined_source( + manifest: ModuleManifest, + definition: DocumentationSourceDefinition, +) -> RegisteredDocumentationSource: + inspection = _defined_inspection(definition) + return _registered_source( + source_id=definition.id, + kind=definition.kind, + owner_module_id=manifest.id, + label=definition.label, + provenance={ + "source": str(definition.provenance.get("source") or "module_manifest"), + **_safe_source_fields(definition.provenance), + }, + inspection=inspection, + condition=definition.condition, + documentation_types=definition.documentation_types, + state=definition.state, + configuration_key=definition.configuration_key, + ) + + +def _defined_inspection(definition: DocumentationSourceDefinition) -> DocumentationSourceInspection: + safe = _safe_source_fields(definition.inspection) + if definition.kind == "configuration_package": + return ConfigurationPackageInspection( + package_id=str(safe.get("package_id") or definition.id), + schema_version=_optional_text(safe.get("schema_version")), + description=_optional_text(safe.get("description")), + ) + href = definition.link.href if definition.link else str(safe.get("href") or "") + if definition.kind == "wiki": + return WikiInspection( + href=href, + published_at=_optional_text(safe.get("published_at")), + revision=_optional_text(safe.get("revision")), + ) + if definition.kind == "repository": + return RepositoryInspection( + href=href, + revision=_optional_text(safe.get("revision")), + ) + if definition.kind == "route": + return RouteInspection( + path=str(safe.get("path") or ""), + component=_optional_text(safe.get("component")), + order=int(safe.get("order") or 0), + ) + if definition.kind == "policy": + return PolicyInspection(capability=str(safe.get("capability") or definition.id)) + if definition.kind == "capability": + return CapabilityInspection(capability=str(safe.get("capability") or definition.id)) + return ManifestInspection( + module_id=str(safe.get("module_id") or definition.id.split(".", 1)[0]), + name=str(safe.get("name") or definition.label), + version=str(safe.get("version") or ""), + dependencies=_string_list(safe.get("dependencies")), + optional_dependencies=_string_list(safe.get("optional_dependencies")), + ) + + +def _linked_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSource, ...]: + sources: list[RegisteredDocumentationSource] = [] + for topic in manifest.documentation: + for link in topic.links: + if link.kind not in {"wiki", "repository"}: + continue + source_id = _derived_source_id(manifest.id, link.kind, link.href) + inspection: DocumentationSourceInspection = ( + WikiInspection(href=link.href) + if link.kind == "wiki" + else RepositoryInspection(href=link.href) + ) + sources.append(_registered_source( + source_id=source_id, + kind=link.kind, + owner_module_id=manifest.id, + label=link.label, + provenance={"source": "documentation_link", "version": manifest.version}, + inspection=inspection, + condition=topic.conditions[0] if len(topic.conditions) == 1 else DocumentationCondition(), + documentation_types=topic.documentation_types, + )) + return tuple(sources) + + +def _registered_source( + *, + source_id: str, + kind: str, + owner_module_id: str, + label: str, + provenance: Mapping[str, Any], + inspection: DocumentationSourceInspection, + condition: DocumentationCondition | None = None, + documentation_types: tuple[DocumentationType, ...] = ("admin",), + state: Literal["configured", "disabled", "unavailable"] = "configured", + configuration_key: str | None = None, +) -> RegisteredDocumentationSource: + clean_provenance = _safe_source_fields(provenance) + return RegisteredDocumentationSource( + item=DocumentationSourceItem( + id=source_id, + kind=kind, + owner_module_id=owner_module_id, + label=label, + state=state, + inspection_url=f"/api/v1/docs/sources/{source_id}", + provenance=DocumentationSourceProvenance( + source=str(clean_provenance.get("source") or "unknown"), + version=_optional_text(clean_provenance.get("version")), + revision=_optional_text(clean_provenance.get("revision")), + published_at=_optional_text(clean_provenance.get("published_at")), + checksum=_optional_text(clean_provenance.get("checksum")), + ), + visibility=_visibility_payload(condition or DocumentationCondition(), documentation_types), + inspection=inspection, + ), + condition=condition or DocumentationCondition(), + documentation_types=documentation_types, + configuration_key=configuration_key, + ) + + +def _visibility_payload( + condition: DocumentationCondition, + documentation_types: tuple[DocumentationType, ...], +) -> DocumentationSourceVisibility: + return DocumentationSourceVisibility( + documentation_types=list(documentation_types), + required_modules=list(condition.required_modules), + any_modules=list(condition.any_modules), + missing_modules=list(condition.missing_modules), + required_capabilities=list(condition.required_capabilities), + required_scopes=list(condition.required_scopes), + any_scopes=list(condition.any_scopes), + configuration_keys=list(condition.configuration_keys), + ) + + +def _derived_source_id(module_id: str, kind: str, value: str) -> str: + digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:16] + return f"{module_id}.{kind}.{digest}" + + +def _safe_source_fields(value: Mapping[str, Any]) -> dict[str, Any]: + redacted = redact_secret_values(dict(value)) + if not isinstance(redacted, Mapping): + return {} + return { + str(key): item + for key, item in redacted.items() + if isinstance(item, (str, int, float, bool, list, tuple)) or item is None + } + + +def _optional_text(value: object) -> str | None: + if value is None: + return None + text = str(value).strip() + return text[:2_000] if text else None + + +def _string_list(value: object) -> list[str]: + if not isinstance(value, (list, tuple)): + return [] + return [str(item)[:500] for item in value[:100]] diff --git a/tests/test_docs_context.py b/tests/test_docs_context.py index a4d83b0..b8d19c9 100644 --- a/tests/test_docs_context.py +++ b/tests/test_docs_context.py @@ -9,6 +9,13 @@ from fastapi import HTTPException from govoplan_access.backend.manifest import get_manifest as get_access_manifest from govoplan_core.core.modules import DocumentationCondition, DocumentationLink, DocumentationTopic, ModuleManifest, NavItem +from govoplan_core.core.modules import ( + DocumentationConfigurationDecision, + DocumentationConfigurationProviderRegistration, + DocumentationSourceDefinition, + FrontendModule, + FrontendRoute, +) from govoplan_core.core.registry import PlatformRegistry from govoplan_tenancy.backend.manifest import get_manifest as get_tenancy_manifest from govoplan_docs.backend.api.v1.routes import ( @@ -16,9 +23,11 @@ from govoplan_docs.backend.api.v1.routes import ( _condition_visibility, _documentation_topic_anchor, _documentation_topic_groups, + _visible_documentation_sources, docs_context, ) from govoplan_docs.backend.manifest import get_manifest as get_docs_manifest +from govoplan_docs.backend.sources import build_documentation_source_registry class FakePrincipal: @@ -208,7 +217,12 @@ class DocsContextTests(unittest.TestCase): ) self.assertEqual(topic["conditions"], []) self.assertEqual(topic["configuration_keys"], []) - self.assertEqual(topic["blockers"], {"modules": [], "capabilities": [], "scopes": []}) + self.assertEqual(topic["blockers"], { + "modules": [], + "capabilities": [], + "scopes": [], + "configuration": [], + }) self.assertNotIn("raw_policy", topic["metadata"]) self.assertNotIn("api_path", topic["metadata"]) self.assertEqual(topic["metadata"]["help_contexts"], ["example.list"]) @@ -327,6 +341,7 @@ class DocsContextTests(unittest.TestCase): self.assertEqual(blockers["modules"], ["mail", "files", "campaigns", "legacy"]) self.assertEqual(blockers["capabilities"], ["mail.delivery"]) self.assertEqual(blockers["scopes"], ["mail:profile:read", "admin:policies:read"]) + self.assertEqual(blockers["configuration"], []) def test_condition_visibility_accepts_any_module_scope_and_capability(self) -> None: registry = PlatformRegistry() @@ -342,7 +357,213 @@ class DocsContextTests(unittest.TestCase): self.assertTrue(active) self.assertEqual(reason, "conditions satisfied") - self.assertEqual(blockers, {"modules": [], "capabilities": [], "scopes": []}) + self.assertEqual(blockers, { + "modules": [], + "capabilities": [], + "scopes": [], + "configuration": [], + }) + + def test_configuration_states_gate_conditions_without_disclosing_values(self) -> None: + states = { + "feature.enabled": "enabled", + "feature.inherited": "inherited", + "feature.disabled": "disabled", + "feature.unavailable": "unavailable", + } + + def resolve(_context, keys): + return { + key: DocumentationConfigurationDecision( + key=key, + state=states[key], # type: ignore[arg-type] + source="tenant" if states[key] == "enabled" else "system", + reason="State metadata only.", + ) + for key in keys + } + + topics = tuple( + DocumentationTopic( + id=f"example.{key}", + title=key, + summary="Configuration-aware topic.", + conditions=( + DocumentationCondition( + required_scopes=("docs:documentation:read",), + configuration_keys=(key,), + ), + ), + configuration_keys=(key,), + ) + for key in states + ) + registry = PlatformRegistry() + registry.register(ModuleManifest( + id="example", + name="Example", + version="1.0.0", + documentation=topics, + documentation_configuration_providers=( + DocumentationConfigurationProviderRegistration( + keys=tuple(states), + resolve=resolve, + ), + ), + )) + + layers = _classify_documentation( + registry, + FakePrincipal({"docs:documentation:read"}), + settings=None, + session=None, + documentation_type="admin", + locale="en", + ) + + self.assertEqual( + {topic["id"] for topic in layers["configured"]}, + {"example.feature.enabled", "example.feature.inherited"}, + ) + self.assertEqual( + [topic["id"] for topic in layers["available"]], + ["example.feature.disabled"], + ) + self.assertEqual( + [topic["id"] for topic in layers["evidence"]], + ["example.feature.unavailable"], + ) + rendered = repr(layers) + self.assertNotIn("raw value", rendered.casefold()) + self.assertNotIn("password", rendered.casefold()) + + def test_source_registry_covers_every_kind_and_redacts_explicit_payloads(self) -> None: + manifest = ModuleManifest( + id="example", + name="Example", + version="1.2.3", + capability_factories={ + "example.lookup": lambda _context: object(), + "policy.example": lambda _context: object(), + }, + frontend=FrontendModule( + module_id="example", + routes=( + FrontendRoute( + path="/example", + component="ExamplePage", + required_all=("example:item:read",), + ), + ), + ), + documentation_sources=( + DocumentationSourceDefinition( + id="example.configuration.package", + kind="configuration_package", + label="Example package", + inspection={ + "package_id": "example.package", + "schema_version": "2", + "password": "must-not-leak", + }, + ), + DocumentationSourceDefinition( + id="example.project.wiki", + kind="wiki", + label="Example wiki", + link=DocumentationLink( + label="Wiki", + href="https://example.invalid/wiki", + kind="wiki", + ), + ), + DocumentationSourceDefinition( + id="example.repository.handbook", + kind="repository", + label="Example handbook", + link=DocumentationLink( + label="Handbook", + href="example/docs/HANDBOOK.md", + kind="repository", + ), + ), + ), + ) + + sources = build_documentation_source_registry((manifest,)).sources() + + self.assertEqual( + {source.item.kind for source in sources}, + { + "manifest", + "route", + "capability", + "policy", + "configuration_package", + "wiki", + "repository", + }, + ) + self.assertNotIn("must-not-leak", repr(sources)) + for source in sources: + self.assertTrue(source.item.id.startswith("example.")) + self.assertEqual("example", source.item.owner_module_id) + self.assertEqual( + f"/api/v1/docs/sources/{source.item.id}", + source.item.inspection_url, + ) + + def test_source_visibility_hides_unauthorized_ids(self) -> None: + manifest = ModuleManifest( + id="example", + name="Example", + version="1.0.0", + documentation_sources=( + DocumentationSourceDefinition( + id="example.protected.handbook", + kind="repository", + label="Protected handbook", + condition=DocumentationCondition( + required_scopes=("example:handbook:read",), + ), + link=DocumentationLink( + label="Protected", + href="example/docs/PROTECTED.md", + kind="repository", + ), + ), + ), + ) + registry = PlatformRegistry() + registry.register(manifest) + sources = build_documentation_source_registry((manifest,)).sources() + + denied = _visible_documentation_sources( + sources, + registry, + FakePrincipal({"docs:documentation:admin"}), + settings=None, + session=None, + documentation_type="admin", + locale="en", + ) + allowed = _visible_documentation_sources( + sources, + registry, + FakePrincipal({ + "docs:documentation:admin", + "example:handbook:read", + }), + settings=None, + session=None, + documentation_type="admin", + locale="en", + ) + + denied_ids = {source.item.id for source in denied} + allowed_ids = {source.item.id for source in allowed} + self.assertNotIn("example.protected.handbook", denied_ids) + self.assertIn("example.protected.handbook", allowed_ids) if __name__ == "__main__": diff --git a/webui/src/api/docs.ts b/webui/src/api/docs.ts index 7bd7ba1..2fd84e1 100644 --- a/webui/src/api/docs.ts +++ b/webui/src/api/docs.ts @@ -54,9 +54,20 @@ export type DocsOptionalModuleEvidence = { }; export type DocsSource = { + id: string; + kind: string; + owner_module_id: string; label: string; - source: string; - layer: string; + state: "configured" | "disabled" | "unavailable"; + state_reason?: string | null; + inspection_url: string; + provenance: { + source: string; + version?: string | null; + revision?: string | null; + published_at?: string | null; + checksum?: string | null; + }; }; export type DocsDocumentationCondition = { @@ -94,6 +105,7 @@ export type DocsDocumentationTopic = { modules: string[]; capabilities: string[]; scopes: string[]; + configuration: string[]; }; audience: string[]; order: number; @@ -105,6 +117,12 @@ export type DocsDocumentationTopic = { related_modules: string[]; unlocks: string[]; configuration_keys: string[]; + configuration_states: Array<{ + key: string; + state: "enabled" | "disabled" | "inherited" | "unavailable"; + source?: string | null; + reason?: string | null; + }>; metadata: Record; }; diff --git a/webui/src/features/docs/DocsPage.tsx b/webui/src/features/docs/DocsPage.tsx index a9893ee..1f91c75 100644 --- a/webui/src/features/docs/DocsPage.tsx +++ b/webui/src/features/docs/DocsPage.tsx @@ -559,7 +559,8 @@ function UnavailableDocumentationReason({ topic }: { topic: DocsDocumentationTop const rows = [ ["i18n:govoplan-docs.modules.04e9462c", topic.blockers.modules], ["i18n:govoplan-docs.capabilities.ca09c54b", topic.blockers.capabilities], - ["i18n:govoplan-docs.permissions.d06d5557", topic.blockers.scopes] + ["i18n:govoplan-docs.permissions.d06d5557", topic.blockers.scopes], + ["Configuration", topic.blockers.configuration] ].filter(([, values]) => Array.isArray(values) && values.length); return (
@@ -975,9 +976,9 @@ function EvidenceList({ modules, sources }: { modules: DocsOptionalModuleEvidenc
)} {sources.map((item) => -
-
{item.layer}
-
{item.label} · {item.source}
+
+
+
{item.label} · {item.owner_module_id} · {item.kind} · {item.provenance.source}
)}