495 lines
21 KiB
Python
495 lines
21 KiB
Python
"""Cross-service KnowledgeFS product-operation contract validation.
|
|
|
|
The checked product manifest is the reviewable boundary, while the Dify product registry, Dify
|
|
Capability issuer registry, exported KnowledgeFS OpenAPI, and exported TypeScript request-guard
|
|
policy remain executable sources of truth. A product operation must appear exactly once as ready
|
|
or as an explicit gap; internal KFS-only operations require a named exclusion.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable, Mapping
|
|
from typing import Any, Literal, NamedTuple, TypedDict, cast
|
|
|
|
|
|
class ProductOperationLimits(TypedDict):
|
|
kfsMaxResponseBytes: int
|
|
productMaxRequestBytes: int
|
|
productMaxResponseBytes: int
|
|
|
|
|
|
class ProductOperationStream(TypedDict):
|
|
kfsResponseKind: str
|
|
productKind: str
|
|
|
|
|
|
class ProductOperationManifestEntry(TypedDict):
|
|
action: str
|
|
kfsOperationId: str
|
|
limits: ProductOperationLimits
|
|
method: str
|
|
path: str
|
|
productOperationId: str
|
|
resource: str
|
|
stream: ProductOperationStream
|
|
transport: str
|
|
|
|
|
|
type ProductOperationManifestField = Literal[
|
|
"action",
|
|
"kfsOperationId",
|
|
"limits",
|
|
"method",
|
|
"path",
|
|
"productOperationId",
|
|
"resource",
|
|
"stream",
|
|
"transport",
|
|
]
|
|
|
|
PRODUCT_OPERATION_MANIFEST_FIELDS: tuple[ProductOperationManifestField, ...] = (
|
|
"action",
|
|
"kfsOperationId",
|
|
"limits",
|
|
"method",
|
|
"path",
|
|
"productOperationId",
|
|
"resource",
|
|
"stream",
|
|
"transport",
|
|
)
|
|
|
|
|
|
class ProductOperationManifest(TypedDict):
|
|
operations: list[ProductOperationManifestEntry]
|
|
schemaVersion: int
|
|
|
|
|
|
class ProductOperationGapEntry(ProductOperationManifestEntry):
|
|
reason: str
|
|
reasonCode: str
|
|
replacementProductOperationIds: list[str]
|
|
|
|
|
|
class InternalKfsOperationExclusion(TypedDict):
|
|
kfsOperationId: str
|
|
reason: str
|
|
reasonCode: str
|
|
|
|
|
|
class ProductOperationGapManifest(TypedDict):
|
|
gaps: list[ProductOperationGapEntry]
|
|
internalKfsOperationExclusions: list[InternalKfsOperationExclusion]
|
|
schemaVersion: int
|
|
|
|
|
|
class ProductOperationRuntimeContract(NamedTuple):
|
|
action: str
|
|
kfs_operation_id: str
|
|
max_request_bytes: int
|
|
max_response_bytes: int
|
|
method: str
|
|
path: str
|
|
product_operation_id: str
|
|
ready: bool
|
|
resource: str
|
|
stream_kind: str
|
|
transport: str
|
|
|
|
|
|
class CapabilityOperationRuntimeContract(NamedTuple):
|
|
action: str
|
|
allowed_caller_kinds: tuple[str, ...]
|
|
method: str
|
|
operation_id: str
|
|
path: str
|
|
resource: str
|
|
|
|
|
|
class CapabilityOperationPolicy(NamedTuple):
|
|
action: str
|
|
allowed_caller_kinds: tuple[str, ...]
|
|
method: str
|
|
operation_id: str
|
|
parent_resource_binding: dict[str, object] | None
|
|
path: str
|
|
resource: str
|
|
resource_binding: dict[str, object]
|
|
|
|
|
|
def product_operation_runtime_contracts() -> tuple[ProductOperationRuntimeContract, ...]:
|
|
"""Read the complete typed Dify product registry without normalizing away gaps."""
|
|
from services.knowledge_fs.product_operations import (
|
|
KNOWLEDGE_FS_PRODUCT_OPERATIONS,
|
|
is_product_operation_ready,
|
|
)
|
|
|
|
contracts: list[ProductOperationRuntimeContract] = []
|
|
for product_operation_id, operation in KNOWLEDGE_FS_PRODUCT_OPERATIONS.items():
|
|
if operation.capability_operation_id is None or operation.kfs_path is None or operation.action is None:
|
|
raise ValueError(f"Dify product operation {product_operation_id} has no Capability contract")
|
|
contracts.append(
|
|
ProductOperationRuntimeContract(
|
|
action=operation.action,
|
|
kfs_operation_id=operation.capability_operation_id,
|
|
max_request_bytes=operation.max_request_bytes,
|
|
max_response_bytes=operation.max_response_bytes,
|
|
method=operation.method,
|
|
path=operation.kfs_path,
|
|
product_operation_id=product_operation_id,
|
|
ready=is_product_operation_ready(product_operation_id),
|
|
resource=operation.resource_resolver,
|
|
stream_kind=operation.stream_kind,
|
|
transport=operation.transport,
|
|
)
|
|
)
|
|
return tuple(contracts)
|
|
|
|
|
|
def capability_operation_runtime_contracts() -> tuple[CapabilityOperationRuntimeContract, ...]:
|
|
"""Read the Python issuer policy used to mint each operation Capability."""
|
|
from services.knowledge_fs_capability import KNOWLEDGE_FS_CAPABILITY_OPERATIONS
|
|
|
|
return tuple(
|
|
CapabilityOperationRuntimeContract(
|
|
action=operation.action,
|
|
allowed_caller_kinds=operation.allowed_caller_kinds,
|
|
method=operation.method,
|
|
operation_id=operation_id,
|
|
path=operation.path,
|
|
resource=operation.resource_type,
|
|
)
|
|
for operation_id, operation in KNOWLEDGE_FS_CAPABILITY_OPERATIONS.items()
|
|
)
|
|
|
|
|
|
def parse_product_operation_manifest(value: dict[str, Any]) -> ProductOperationManifest:
|
|
"""Parse the ready product-operation inventory with a closed schema."""
|
|
if set(value) != {"operations", "schemaVersion"} or value.get("schemaVersion") != 2:
|
|
raise ValueError("KnowledgeFS product operation manifest must use schemaVersion 2")
|
|
operations = value.get("operations")
|
|
if not isinstance(operations, list) or not operations:
|
|
raise ValueError("KnowledgeFS product operation manifest operations must be a non-empty list")
|
|
return {
|
|
"operations": [_parse_product_entry(entry, gap=False) for entry in operations],
|
|
"schemaVersion": 2,
|
|
}
|
|
|
|
|
|
def parse_product_operation_gap_manifest(value: dict[str, Any]) -> ProductOperationGapManifest:
|
|
"""Parse explicit product gaps and KFS-only operation exclusions with a closed schema."""
|
|
expected_fields = {"gaps", "internalKfsOperationExclusions", "schemaVersion"}
|
|
if set(value) != expected_fields or value.get("schemaVersion") != 2:
|
|
raise ValueError("KnowledgeFS product operation gap manifest must use schemaVersion 2")
|
|
gaps = value.get("gaps")
|
|
exclusions = value.get("internalKfsOperationExclusions")
|
|
if not isinstance(gaps, list) or not gaps or not isinstance(exclusions, list):
|
|
raise ValueError("KnowledgeFS product operation gap manifest lists are invalid")
|
|
parsed_exclusions: list[InternalKfsOperationExclusion] = []
|
|
for raw in exclusions:
|
|
if not isinstance(raw, dict) or set(raw) != {"kfsOperationId", "reason", "reasonCode"}:
|
|
raise ValueError("KnowledgeFS internal KFS operation exclusion fields are invalid")
|
|
for field in ("kfsOperationId", "reason", "reasonCode"):
|
|
_required_string(raw.get(field), f"internal exclusion {field}")
|
|
parsed_exclusions.append(cast(InternalKfsOperationExclusion, raw))
|
|
return {
|
|
"gaps": [cast(ProductOperationGapEntry, _parse_product_entry(entry, gap=True)) for entry in gaps],
|
|
"internalKfsOperationExclusions": parsed_exclusions,
|
|
"schemaVersion": 2,
|
|
}
|
|
|
|
|
|
def parse_capability_operation_policy(value: dict[str, Any]) -> tuple[CapabilityOperationPolicy, ...]:
|
|
"""Parse the runtime-exported TypeScript request-guard registry."""
|
|
if set(value) != {"operations", "schemaVersion"} or value.get("schemaVersion") != 1:
|
|
raise ValueError("KnowledgeFS Capability operation policy must use schemaVersion 1")
|
|
operations = value.get("operations")
|
|
if not isinstance(operations, list) or not operations:
|
|
raise ValueError("KnowledgeFS Capability operation policy must contain operations")
|
|
parsed: list[CapabilityOperationPolicy] = []
|
|
for raw in operations:
|
|
expected_fields = {
|
|
"action",
|
|
"allowedCallerKinds",
|
|
"method",
|
|
"operationId",
|
|
"parentResourceBinding",
|
|
"path",
|
|
"resourceBinding",
|
|
"resourceType",
|
|
}
|
|
if not isinstance(raw, dict) or set(raw) != expected_fields:
|
|
raise ValueError("KnowledgeFS Capability operation policy fields are invalid")
|
|
callers = raw.get("allowedCallerKinds")
|
|
resource_binding = raw.get("resourceBinding")
|
|
parent_binding = raw.get("parentResourceBinding")
|
|
if (
|
|
not isinstance(callers, list)
|
|
or not callers
|
|
or not all(isinstance(caller, str) and caller for caller in callers)
|
|
or not isinstance(resource_binding, dict)
|
|
or not resource_binding
|
|
or (parent_binding is not None and not isinstance(parent_binding, dict))
|
|
):
|
|
raise ValueError("KnowledgeFS Capability operation request binding is invalid")
|
|
for field in ("action", "method", "operationId", "path", "resourceType"):
|
|
_required_string(raw.get(field), f"Capability operation {field}")
|
|
parsed.append(
|
|
CapabilityOperationPolicy(
|
|
action=cast(str, raw["action"]),
|
|
allowed_caller_kinds=tuple(callers),
|
|
method=cast(str, raw["method"]),
|
|
operation_id=cast(str, raw["operationId"]),
|
|
parent_resource_binding=cast(dict[str, object] | None, parent_binding),
|
|
path=cast(str, raw["path"]),
|
|
resource=cast(str, raw["resourceType"]),
|
|
resource_binding=cast(dict[str, object], resource_binding),
|
|
)
|
|
)
|
|
_unique((operation.operation_id for operation in parsed), "Capability operation policy")
|
|
return tuple(parsed)
|
|
|
|
|
|
def validate_product_operation_contracts(
|
|
*,
|
|
capability_operations: tuple[CapabilityOperationRuntimeContract, ...],
|
|
capability_policy: tuple[CapabilityOperationPolicy, ...],
|
|
document: dict[str, Any],
|
|
gap_manifest: ProductOperationGapManifest,
|
|
manifest: ProductOperationManifest,
|
|
product_operations: tuple[ProductOperationRuntimeContract, ...],
|
|
) -> None:
|
|
"""Require exact agreement across manifests, Dify registries, KFS guard policy, and OpenAPI."""
|
|
ready_entries = {entry["productOperationId"]: entry for entry in manifest["operations"]}
|
|
gap_entries = {entry["productOperationId"]: entry for entry in gap_manifest["gaps"]}
|
|
if len(ready_entries) != len(manifest["operations"]):
|
|
raise ValueError("KnowledgeFS product operation manifest contains duplicate product ids")
|
|
if len(gap_entries) != len(gap_manifest["gaps"]):
|
|
raise ValueError("KnowledgeFS product operation gap manifest contains duplicate product ids")
|
|
if set(ready_entries) & set(gap_entries):
|
|
raise ValueError("KnowledgeFS product operation cannot be both ready and a gap")
|
|
runtime_by_product = {operation.product_operation_id: operation for operation in product_operations}
|
|
if set(ready_entries) | set(gap_entries) != set(runtime_by_product):
|
|
missing = sorted(set(runtime_by_product) - set(ready_entries) - set(gap_entries))
|
|
extra = sorted((set(ready_entries) | set(gap_entries)) - set(runtime_by_product))
|
|
raise ValueError(f"KnowledgeFS product operation completeness drifted: missing={missing}, extra={extra}")
|
|
|
|
python_capabilities = {operation.operation_id: operation for operation in capability_operations}
|
|
guard_capabilities = {operation.operation_id: operation for operation in capability_policy}
|
|
openapi_operations = _openapi_operations(document)
|
|
for product_operation_id, runtime in runtime_by_product.items():
|
|
entry = ready_entries.get(product_operation_id) or gap_entries[product_operation_id]
|
|
if runtime.ready != (product_operation_id in ready_entries):
|
|
state = "ready" if runtime.ready else "gap"
|
|
raise ValueError(f"Dify product operation {product_operation_id} must be declared as {state}")
|
|
_validate_product_entry(entry, runtime, openapi_operations, python_capabilities, guard_capabilities)
|
|
if product_operation_id in gap_entries:
|
|
_required_string(gap_entries[product_operation_id]["reason"], "product gap reason")
|
|
_required_string(gap_entries[product_operation_id]["reasonCode"], "product gap reasonCode")
|
|
replacements = gap_entries[product_operation_id]["replacementProductOperationIds"]
|
|
if not replacements or any(replacement not in ready_entries for replacement in replacements):
|
|
raise ValueError(f"KnowledgeFS product gap {product_operation_id} has invalid replacements")
|
|
|
|
exclusions = gap_manifest["internalKfsOperationExclusions"]
|
|
exclusion_by_id = {entry["kfsOperationId"]: entry for entry in exclusions}
|
|
if len(exclusion_by_id) != len(exclusions):
|
|
raise ValueError("KnowledgeFS internal KFS operation exclusions contain duplicate ids")
|
|
lifecycle_operation_ids = {
|
|
"activateDifyWorkspaceIntegration",
|
|
"freezeDifyWorkspaceIntegration",
|
|
}
|
|
if set(exclusion_by_id) != lifecycle_operation_ids:
|
|
raise ValueError("KnowledgeFS Dify integration lifecycle operations must be explicit internal exclusions")
|
|
product_capability_ids = {operation.kfs_operation_id for operation in product_operations}
|
|
for operation_id in exclusion_by_id:
|
|
if operation_id in product_capability_ids:
|
|
raise ValueError(f"Internal KFS operation {operation_id} cannot also be a product operation")
|
|
_validate_capability_alignment(
|
|
operation_id,
|
|
openapi_operations,
|
|
python_capabilities,
|
|
guard_capabilities,
|
|
)
|
|
python_operation = python_capabilities[operation_id]
|
|
if python_operation.allowed_caller_kinds != ("internal_worker",) or python_operation.resource != "namespace":
|
|
raise ValueError("Dify integration lifecycle operations must remain internal-worker namespace-only")
|
|
|
|
|
|
def _parse_product_entry(raw: object, *, gap: bool) -> ProductOperationManifestEntry:
|
|
base_fields = {
|
|
"action",
|
|
"kfsOperationId",
|
|
"limits",
|
|
"method",
|
|
"path",
|
|
"productOperationId",
|
|
"resource",
|
|
"stream",
|
|
"transport",
|
|
}
|
|
expected_fields = base_fields | ({"reason", "reasonCode", "replacementProductOperationIds"} if gap else set())
|
|
if not isinstance(raw, dict) or set(raw) != expected_fields:
|
|
raise ValueError("KnowledgeFS product operation entry fields are invalid")
|
|
for field in ("action", "kfsOperationId", "method", "path", "productOperationId", "resource", "transport"):
|
|
_required_string(raw.get(field), f"product operation {field}")
|
|
limits = raw.get("limits")
|
|
stream = raw.get("stream")
|
|
if not isinstance(limits, dict) or set(limits) != {
|
|
"kfsMaxResponseBytes",
|
|
"productMaxRequestBytes",
|
|
"productMaxResponseBytes",
|
|
}:
|
|
raise ValueError("KnowledgeFS product operation limits are invalid")
|
|
if not all(isinstance(limit, int) and not isinstance(limit, bool) and limit >= 0 for limit in limits.values()):
|
|
raise ValueError("KnowledgeFS product operation limits must be nonnegative integers")
|
|
if not isinstance(stream, dict) or set(stream) != {"kfsResponseKind", "productKind"}:
|
|
raise ValueError("KnowledgeFS product operation stream contract is invalid")
|
|
_required_string(stream.get("kfsResponseKind"), "KFS response kind")
|
|
_required_string(stream.get("productKind"), "product stream kind")
|
|
if gap:
|
|
_required_string(raw.get("reason"), "product gap reason")
|
|
_required_string(raw.get("reasonCode"), "product gap reasonCode")
|
|
replacements = raw.get("replacementProductOperationIds")
|
|
if not isinstance(replacements, list) or not all(isinstance(item, str) and item for item in replacements):
|
|
raise ValueError("KnowledgeFS product operation gap replacements are invalid")
|
|
return cast(ProductOperationManifestEntry, raw)
|
|
|
|
|
|
def _validate_product_entry(
|
|
entry: ProductOperationManifestEntry,
|
|
runtime: ProductOperationRuntimeContract,
|
|
openapi_operations: Mapping[str, tuple[str, str, dict[str, Any]]],
|
|
python_capabilities: Mapping[str, CapabilityOperationRuntimeContract],
|
|
guard_capabilities: Mapping[str, CapabilityOperationPolicy],
|
|
) -> None:
|
|
_validate_capability_alignment(
|
|
runtime.kfs_operation_id,
|
|
openapi_operations,
|
|
python_capabilities,
|
|
guard_capabilities,
|
|
)
|
|
_, _, openapi_operation = openapi_operations[runtime.kfs_operation_id]
|
|
expected: ProductOperationManifestEntry = {
|
|
"action": runtime.action,
|
|
"kfsOperationId": runtime.kfs_operation_id,
|
|
"limits": {
|
|
"kfsMaxResponseBytes": _openapi_max_response_bytes(openapi_operation),
|
|
"productMaxRequestBytes": runtime.max_request_bytes,
|
|
"productMaxResponseBytes": runtime.max_response_bytes,
|
|
},
|
|
"method": runtime.method,
|
|
"path": runtime.path,
|
|
"productOperationId": runtime.product_operation_id,
|
|
"resource": runtime.resource,
|
|
"stream": {
|
|
"kfsResponseKind": _openapi_response_kind(openapi_operation),
|
|
"productKind": runtime.stream_kind,
|
|
},
|
|
"transport": runtime.transport,
|
|
}
|
|
for field in PRODUCT_OPERATION_MANIFEST_FIELDS:
|
|
expected_value = expected[field]
|
|
if entry[field] != expected_value:
|
|
raise ValueError(
|
|
f"KnowledgeFS product operation {runtime.product_operation_id} field {field} drifted: "
|
|
f"expected {expected_value!r}, received {entry[field]!r}"
|
|
)
|
|
|
|
|
|
def _validate_capability_alignment(
|
|
operation_id: str,
|
|
openapi_operations: Mapping[str, tuple[str, str, dict[str, Any]]],
|
|
python_capabilities: Mapping[str, CapabilityOperationRuntimeContract],
|
|
guard_capabilities: Mapping[str, CapabilityOperationPolicy],
|
|
) -> None:
|
|
python_operation = python_capabilities.get(operation_id)
|
|
guard_operation = guard_capabilities.get(operation_id)
|
|
openapi_operation = openapi_operations.get(operation_id)
|
|
if python_operation is None or guard_operation is None or openapi_operation is None:
|
|
raise ValueError(
|
|
f"KnowledgeFS operation {operation_id} must exist in Python issuer, TypeScript guard, and OpenAPI"
|
|
)
|
|
method, path, _ = openapi_operation
|
|
expected = (
|
|
python_operation.action,
|
|
python_operation.method,
|
|
python_operation.path,
|
|
python_operation.resource,
|
|
)
|
|
received_guard = (guard_operation.action, guard_operation.method, guard_operation.path, guard_operation.resource)
|
|
if received_guard != expected:
|
|
raise ValueError(f"KnowledgeFS operation {operation_id} TypeScript guard drifted")
|
|
if (method, path) != (python_operation.method, python_operation.path):
|
|
raise ValueError(f"KnowledgeFS operation {operation_id} OpenAPI method/path drifted")
|
|
if guard_operation.allowed_caller_kinds != python_operation.allowed_caller_kinds:
|
|
raise ValueError(f"KnowledgeFS operation {operation_id} caller policy drifted")
|
|
|
|
|
|
def _openapi_operations(document: dict[str, Any]) -> dict[str, tuple[str, str, dict[str, Any]]]:
|
|
operations: dict[str, tuple[str, str, dict[str, Any]]] = {}
|
|
duplicates: set[str] = set()
|
|
for path, path_item in document.get("paths", {}).items():
|
|
if not isinstance(path, str) or not isinstance(path_item, dict):
|
|
continue
|
|
for method in ("delete", "get", "patch", "post", "put"):
|
|
operation = path_item.get(method)
|
|
if not isinstance(operation, dict):
|
|
continue
|
|
operation_id = operation.get("operationId")
|
|
if not isinstance(operation_id, str) or not operation_id:
|
|
continue
|
|
if operation_id in operations:
|
|
duplicates.add(operation_id)
|
|
operations[operation_id] = (method.upper(), path, operation)
|
|
if duplicates:
|
|
raise ValueError(f"KnowledgeFS OpenAPI contains duplicate operation ids: {sorted(duplicates)}")
|
|
return operations
|
|
|
|
|
|
def _openapi_response_kind(operation: dict[str, Any]) -> Literal["binary", "buffered", "stream"]:
|
|
media_types = {
|
|
media_type
|
|
for status, response in operation.get("responses", {}).items()
|
|
if status == "2XX" or (len(status) == 3 and status.startswith("2") and status.isdigit())
|
|
for media_type in response.get("content", {})
|
|
}
|
|
if "text/event-stream" in media_types:
|
|
return "stream"
|
|
if "application/octet-stream" in media_types:
|
|
return "binary"
|
|
return "buffered"
|
|
|
|
|
|
def _openapi_max_response_bytes(operation: dict[str, Any]) -> int:
|
|
value = operation.get("x-knowledge-fs-max-response-bytes")
|
|
if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
|
|
raise ValueError("KnowledgeFS product OpenAPI operation has no positive response byte limit")
|
|
return value
|
|
|
|
|
|
def _required_string(value: object, field: str) -> str:
|
|
if not isinstance(value, str) or not value.strip() or value != value.strip():
|
|
raise ValueError(f"KnowledgeFS {field} must be a non-empty trimmed string")
|
|
return value
|
|
|
|
|
|
def _unique(values: Iterable[str], label: str) -> None:
|
|
sequence = tuple(values)
|
|
if len(sequence) != len(set(sequence)):
|
|
raise ValueError(f"KnowledgeFS {label} contains duplicate ids")
|
|
|
|
|
|
__all__ = [
|
|
"CapabilityOperationPolicy",
|
|
"CapabilityOperationRuntimeContract",
|
|
"ProductOperationGapManifest",
|
|
"ProductOperationManifest",
|
|
"ProductOperationRuntimeContract",
|
|
"capability_operation_runtime_contracts",
|
|
"parse_capability_operation_policy",
|
|
"parse_product_operation_gap_manifest",
|
|
"parse_product_operation_manifest",
|
|
"product_operation_runtime_contracts",
|
|
"validate_product_operation_contracts",
|
|
]
|