89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
"""Authenticated, secret-free service compatibility diagnostics."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Literal, cast
|
|
|
|
from gewu.application.identity import Principal
|
|
from gewu.application.uploads import ALLOWED_MEDIA_TYPES, CHUNK_BYTES, MAX_ARTIFACT_BYTES
|
|
from gewu.domain.contracts import Scope, ServiceDiagnostics
|
|
from gewu.domain.operations import Transport, catalog_digest, operations_for
|
|
from gewu.version import SERVICE_VERSION
|
|
|
|
|
|
def build_service_diagnostics(
|
|
*,
|
|
principal: Principal,
|
|
transport: Literal["http", "mcp"],
|
|
authorized_projects: tuple[str, ...],
|
|
available_operations: frozenset[str],
|
|
index_status: tuple[str, str | None, str, int | None],
|
|
projects: tuple[str, ...] | None = None,
|
|
include_shared: bool = False,
|
|
) -> ServiceDiagnostics:
|
|
"""Build caller-specific diagnostics without exposing credential identifiers."""
|
|
|
|
visible_projects = tuple(sorted(set(authorized_projects)))
|
|
if projects is None:
|
|
if include_shared:
|
|
raise ValueError("Select at least one project before including shared knowledge")
|
|
selected_scope = None
|
|
else:
|
|
if not projects or len(set(projects)) != len(projects):
|
|
raise ValueError("Selected projects must be non-empty and unique")
|
|
if any(project not in visible_projects for project in projects):
|
|
raise PermissionError("Selected scope includes an unreadable project")
|
|
if include_shared and not principal.can_read_project("shared"):
|
|
raise PermissionError("Credential cannot read shared knowledge")
|
|
selected_scope = Scope(projects=projects, include_shared=include_shared)
|
|
|
|
capabilities: set[str] = {"service:diagnostics"}
|
|
if "admin" in principal.roles and "agent" not in principal.roles:
|
|
capabilities.update({"project:admin", "credential:admin"})
|
|
if principal.roles.intersection({"admin", "editor", "viewer", "agent"}):
|
|
capabilities.add("knowledge:read")
|
|
if principal.auth_method == "bearer" and "agent" in principal.roles:
|
|
capabilities.update(
|
|
{
|
|
"proposal:submit",
|
|
"proposal:evaluate",
|
|
"proposal:read",
|
|
"proposal:withdraw",
|
|
"run:read",
|
|
"run:write",
|
|
}
|
|
)
|
|
if principal.roles.intersection({"admin", "editor", "viewer"}):
|
|
capabilities.add("identity:session")
|
|
|
|
catalog_transport = Transport(transport)
|
|
authorized_operations = tuple(
|
|
sorted(
|
|
operation.name
|
|
for operation in operations_for(catalog_transport)
|
|
if operation.name in available_operations and operation.capability in capabilities
|
|
)
|
|
)
|
|
canonical, index_revision, index_state, index_schema_version = index_status
|
|
return ServiceDiagnostics(
|
|
service_version=SERVICE_VERSION,
|
|
catalog_sha256=catalog_digest(),
|
|
transport=transport,
|
|
authentication=cast(Literal["bearer", "browser_session"], principal.auth_method),
|
|
roles=tuple(sorted(principal.roles)),
|
|
authorized_projects=visible_projects,
|
|
selected_scope=selected_scope,
|
|
capabilities=tuple(sorted(capabilities)),
|
|
authorized_operations=authorized_operations,
|
|
max_upload_bytes=MAX_ARTIFACT_BYTES,
|
|
upload_chunk_bytes=CHUNK_BYTES,
|
|
supported_media_types=tuple(sorted(ALLOWED_MEDIA_TYPES)),
|
|
canonical_revision=canonical,
|
|
index_state=cast(
|
|
Literal["current", "lagging", "not_built", "not_configured", "unavailable"],
|
|
index_state,
|
|
),
|
|
index_revision=index_revision,
|
|
index_schema_version=index_schema_version,
|
|
)
|