refactor(dataset): validate declared KnowledgeFS operations

Fixes WTA-1411

api/dev/generate_knowledge_fs_contract.py

@@ -0,0 +1,229 @@
+"""Validate Dify Console KnowledgeFS declarations against a pinned OpenAPI document.
+
+The OpenAPI document is exported only during development and CI. Runtime declarations live with Dify product policy;
+this module validates their transport metadata without generating a complete operation catalog.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Any, Literal, TypedDict
+
+API_ROOT = Path(__file__).resolve().parents[1]
+WORKSPACE_ROOT = API_ROOT.parent
+LOCK_PATH = API_ROOT / "knowledge-fs-contract.lock.json"
+DEFAULT_REPOSITORY = WORKSPACE_ROOT.parent / "knowledge-fs"
+OPENAPI_METHODS = ("delete", "get", "head", "options", "patch", "post", "put", "trace")
+PROXY_METHODS = frozenset({"delete", "get", "patch", "post", "put"})
+
+
+class ContractDeclaration(TypedDict):
+    """KnowledgeFS transport contract declared by one Dify Console registry entry."""
+
+    operation_id: str
+    method: str
+    path: str
+    required_scope: str | None
+    response_kind: str
+    max_response_bytes: int
+    request_headers: tuple[str, ...]
+    response_headers: tuple[str, ...]
+    response_media_types: tuple[str, ...]
+
+
+type DeclarationField = Literal[
+    "method",
+    "path",
+    "required_scope",
+    "response_kind",
+    "max_response_bytes",
+    "request_headers",
+    "response_headers",
+    "response_media_types",
+]
+
+DECLARATION_FIELDS: tuple[DeclarationField, ...] = (
+    "method",
+    "path",
+    "required_scope",
+    "response_kind",
+    "max_response_bytes",
+    "request_headers",
+    "response_headers",
+    "response_media_types",
+)
+
+
+def main() -> None:
+    """Update or verify the pinned KnowledgeFS commit and OpenAPI hash."""
+    parser = argparse.ArgumentParser()
+    mode = parser.add_mutually_exclusive_group()
+    mode.add_argument("--check", action="store_true")
+    mode.add_argument("--update-lock", action="store_true")
+    args = parser.parse_args()
+
+    repository = Path(os.environ.get("KNOWLEDGE_FS_REPO", DEFAULT_REPOSITORY)).resolve()
+    lock = json.loads(LOCK_PATH.read_text())
+    tracked_changes = run("git", "status", "--porcelain", "--untracked-files=no", cwd=repository).strip()
+    if tracked_changes:
+        raise RuntimeError("KnowledgeFS checkout must not contain tracked changes during contract export")
+
+    commit = run("git", "rev-parse", "HEAD", cwd=repository).strip()
+    if not args.update_lock and commit != lock["commit"]:
+        raise RuntimeError(
+            f"KnowledgeFS checkout mismatch: expected {lock['commit']}, received {commit}. "
+            "Use the pinned commit or pass --update-lock intentionally."
+        )
+
+    with tempfile.TemporaryDirectory(prefix="dify-knowledge-fs-contract-") as directory:
+        openapi_path = Path(directory) / "knowledge-fs.openapi.json"
+        subprocess.run(
+            ["pnpm", "openapi:export", "--", "--output", str(openapi_path)],
+            cwd=repository,
+            check=True,
+        )
+        openapi_content = openapi_path.read_bytes()
+
+    openapi_sha256 = sha256(openapi_content)
+    if args.update_lock:
+        LOCK_PATH.write_text(
+            json.dumps(
+                {
+                    "commit": commit,
+                    "openapiSha256": openapi_sha256,
+                    "repository": lock["repository"],
+                },
+                indent=2,
+            )
+            + "\n"
+        )
+        return
+
+    if openapi_sha256 != lock["openapiSha256"]:
+        raise RuntimeError(
+            f"KnowledgeFS OpenAPI hash mismatch: expected {lock['openapiSha256']}, received {openapi_sha256}"
+        )
+
+
+def validate_declarations(document: dict[str, Any], declarations: tuple[ContractDeclaration, ...]) -> None:
+    """Validate Dify Console declarations against matching pinned OpenAPI operations."""
+    operations_by_id: dict[str, list[tuple[str, str, dict[str, Any], dict[str, Any]]]] = {}
+    for path, path_item in document.get("paths", {}).items():
+        for method in OPENAPI_METHODS:
+            operation = path_item.get(method)
+            if operation is None:
+                continue
+            operation_id = operation.get("operationId")
+            if isinstance(operation_id, str) and operation_id:
+                operations_by_id.setdefault(operation_id, []).append((method, path, path_item, operation))
+
+    declared_ids: set[str] = set()
+    for declaration in declarations:
+        operation_id = declaration["operation_id"]
+        if operation_id in declared_ids:
+            raise ValueError(f"Dify Console registry has duplicate operationId: {operation_id}")
+        declared_ids.add(operation_id)
+
+        matches = operations_by_id.get(operation_id, [])
+        if not matches:
+            raise ValueError(f"KnowledgeFS OpenAPI has no operationId: {operation_id}")
+        if len(matches) > 1:
+            raise ValueError(f"KnowledgeFS OpenAPI has duplicate operationId: {operation_id}")
+
+        method, path, path_item, operation = matches[0]
+        if not path.startswith("/"):
+            raise ValueError(f"KnowledgeFS OpenAPI path must be absolute: {path}")
+        if method not in PROXY_METHODS:
+            raise ValueError(f"KnowledgeFS proxy does not support {method.upper()} {path}")
+        expected: ContractDeclaration = {
+            "operation_id": operation_id,
+            "method": method.upper(),
+            "path": path[1:],
+            "required_scope": required_scope(operation),
+            "response_kind": response_kind(operation),
+            "max_response_bytes": required_max_response_bytes(operation),
+            "request_headers": request_header_names(path_item, operation),
+            "response_headers": response_header_names(operation),
+            "response_media_types": response_media_types(operation),
+        }
+        for field in DECLARATION_FIELDS:
+            expected_value = expected[field]
+            received_value = declaration[field]
+            if received_value != expected_value:
+                raise ValueError(
+                    f"KnowledgeFS operation {operation_id} field {field} drifted: "
+                    f"expected {expected_value!r}, received {received_value!r}"
+                )
+
+
+def response_kind(operation: dict[str, Any]) -> str:
+    media_types = response_media_types(operation)
+    if "text/event-stream" in media_types:
+        return "stream"
+    if "application/octet-stream" in media_types:
+        return "binary"
+    return "buffered"
+
+
+def response_media_types(operation: dict[str, Any]) -> tuple[str, ...]:
+    media_types: set[str] = set()
+    for status, response in operation.get("responses", {}).items():
+        if status == "2XX" or (len(status) == 3 and status.startswith("2") and status.isdigit()):
+            media_types.update(response.get("content", {}))
+    return tuple(sorted(media_types))
+
+
+def required_scope(operation: dict[str, Any]) -> str | None:
+    scope = operation.get("x-knowledge-fs-required-scope")
+    if scope in ("knowledge-spaces:read", "knowledge-spaces:write"):
+        return scope
+    if operation.get("security") == []:
+        return None
+    raise ValueError(f"KnowledgeFS operation has no supported required scope: {scope}")
+
+
+def required_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(f"KnowledgeFS operation has no valid response byte limit: {value}")
+    return value
+
+
+def request_header_names(path_item: dict[str, Any], operation: dict[str, Any]) -> tuple[str, ...]:
+    names: set[str] = set()
+    for parameter in [*path_item.get("parameters", []), *operation.get("parameters", [])]:
+        if "$ref" in parameter:
+            raise ValueError(f"KnowledgeFS request header references are not supported: {parameter['$ref']}")
+        if parameter.get("in") == "header":
+            names.add(parameter["name"].lower())
+    return tuple(sorted(names))
+
+
+def response_header_names(operation: dict[str, Any]) -> tuple[str, ...]:
+    return tuple(
+        sorted(
+            {
+                name.lower()
+                for response in operation.get("responses", {}).values()
+                for name in response.get("headers", {})
+            }
+        )
+    )
+
+
+def sha256(content: bytes) -> str:
+    return hashlib.sha256(content).hexdigest()
+
+
+def run(*command: str, cwd: Path) -> str:
+    return subprocess.run(command, cwd=cwd, check=True, capture_output=True, text=True).stdout
+
+
+if __name__ == "__main__":
+    main()

api/knowledge-fs-contract.lock.json

@@ -0,0 +1,5 @@
+{
+  "commit": "4310e2d582d25e7de58183f27720afab01e123cf",
+  "openapiSha256": "5827ca930ce38462bfd1b2bef387efbf37eb7ffcaedde4558af2fbaeccbfbc4b",
+  "repository": "https://github.com/langgenius/knowledge-fs"
+}

api/tests/unit_tests/dev/test_generate_knowledge_fs_contract.py

@@ -0,0 +1,266 @@
+"""Tests for pinned KnowledgeFS declaration validation."""
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+from typing import cast
+
+import pytest
+
+from dev import generate_knowledge_fs_contract as contract_validator
+from dev.generate_knowledge_fs_contract import ContractDeclaration, validate_declarations
+
+
+def test_contract_cli_updates_checks_and_detects_openapi_drift(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    repository = tmp_path / "knowledge-fs"
+    repository.mkdir()
+    subprocess.run(["git", "init", "--quiet"], cwd=repository, check=True)
+    subprocess.run(
+        [
+            "git",
+            "-c",
+            "user.email=contract-test@example.com",
+            "-c",
+            "user.name=Contract Test",
+            "commit",
+            "--allow-empty",
+            "--quiet",
+            "-m",
+            "fixture",
+        ],
+        cwd=repository,
+        check=True,
+    )
+    commit = subprocess.run(
+        ["git", "rev-parse", "HEAD"], cwd=repository, check=True, capture_output=True, text=True
+    ).stdout.strip()
+
+    document = {"paths": {"/health": {"get": operation(None, "getHealth", security=[])}}}
+    executable_directory = tmp_path / "bin"
+    executable_directory.mkdir()
+    fake_pnpm = executable_directory / "pnpm"
+    write_fake_pnpm(fake_pnpm, document)
+
+    lock_path = tmp_path / "knowledge-fs-contract.lock.json"
+    lock_path.write_text(
+        json.dumps(
+            {
+                "commit": "",
+                "openapiSha256": "",
+                "repository": "https://github.com/langgenius/knowledge-fs",
+            }
+        )
+    )
+    monkeypatch.setattr(contract_validator, "LOCK_PATH", lock_path)
+    monkeypatch.setenv("KNOWLEDGE_FS_REPO", str(repository))
+    monkeypatch.setenv("PATH", f"{executable_directory}{os.pathsep}{os.environ['PATH']}")
+
+    monkeypatch.setattr(sys, "argv", ["generate_knowledge_fs_contract.py", "--update-lock"])
+    contract_validator.main()
+
+    updated_lock = json.loads(lock_path.read_text())
+    assert updated_lock["commit"] == commit
+    assert set(updated_lock) == {"commit", "openapiSha256", "repository"}
+
+    monkeypatch.setattr(sys, "argv", ["generate_knowledge_fs_contract.py", "--check"])
+    contract_validator.main()
+
+    write_fake_pnpm(fake_pnpm, {"paths": {}})
+    with pytest.raises(RuntimeError, match="OpenAPI hash mismatch"):
+        contract_validator.main()
+
+
+def test_validate_declarations_accepts_matching_contract() -> None:
+    route = operation("knowledge-spaces:read", "listKnowledgeSpaces")
+    route["parameters"] = [{"in": "header", "name": "X-Trace-Id"}]
+    route["responses"] = {
+        "200": {
+            "content": {"application/json": {}},
+            "headers": {"X-Trace-Id": {}},
+        }
+    }
+    document = {"paths": {"/knowledge-spaces": {"get": route}}}
+
+    validate_declarations(
+        document,
+        (
+            declaration(
+                request_headers=("x-trace-id",),
+                response_headers=("x-trace-id",),
+            ),
+        ),
+    )
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("method", "POST"),
+        ("path", "spaces"),
+        ("required_scope", "knowledge-spaces:write"),
+        ("response_kind", "stream"),
+        ("max_response_bytes", 2_097_152),
+        ("request_headers", ("authorization",)),
+        ("response_headers", ("cache-control",)),
+        ("response_media_types", ("text/event-stream",)),
+    ],
+)
+def test_validate_declarations_reports_contract_field_drift(field: str, value: object) -> None:
+    document = {
+        "paths": {
+            "/knowledge-spaces": {
+                "get": operation("knowledge-spaces:read", "listKnowledgeSpaces"),
+            }
+        }
+    }
+
+    with pytest.raises(ValueError, match=rf"listKnowledgeSpaces.*{field}.*expected.*received"):
+        validate_declarations(document, (declaration(**{field: value}),))
+
+
+def test_validate_declarations_rejects_unknown_operation_id() -> None:
+    with pytest.raises(ValueError, match="no operationId: listKnowledgeSpaces"):
+        validate_declarations({"paths": {}}, (declaration(),))
+
+
+def test_validate_declarations_rejects_duplicate_declared_operation_ids() -> None:
+    document = {
+        "paths": {
+            "/knowledge-spaces": {
+                "get": operation("knowledge-spaces:read", "listKnowledgeSpaces"),
+            }
+        }
+    }
+
+    with pytest.raises(ValueError, match="registry has duplicate operationId: listKnowledgeSpaces"):
+        validate_declarations(document, (declaration(), declaration()))
+
+
+def test_validate_declarations_rejects_duplicate_upstream_operation_ids() -> None:
+    document = {
+        "paths": {
+            "/knowledge-spaces": {
+                "get": operation("knowledge-spaces:read", "listKnowledgeSpaces"),
+            },
+            "/spaces": {
+                "get": operation("knowledge-spaces:read", "listKnowledgeSpaces"),
+            },
+        }
+    }
+
+    with pytest.raises(ValueError, match="OpenAPI has duplicate operationId: listKnowledgeSpaces"):
+        validate_declarations(document, (declaration(),))
+
+
+def test_validate_declarations_ignores_undeclared_operations() -> None:
+    document = {
+        "paths": {
+            "/knowledge-spaces": {
+                "get": operation("knowledge-spaces:read", "listKnowledgeSpaces"),
+            },
+            "/internal-maintenance": {
+                "head": {"responses": {"200": {}}},
+            },
+        }
+    }
+
+    validate_declarations(document, (declaration(),))
+
+
+def test_validate_declarations_preserves_public_operation_scope() -> None:
+    document = {"paths": {"/health": {"get": operation(None, "getHealth", security=[])}}}
+
+    validate_declarations(
+        document,
+        (
+            declaration(
+                operation_id="getHealth",
+                path="health",
+                required_scope=None,
+            ),
+        ),
+    )
+
+
+def test_validate_declarations_rejects_unsupported_declared_method() -> None:
+    document = {
+        "paths": {
+            "/knowledge-spaces": {
+                "head": operation("knowledge-spaces:read", "listKnowledgeSpaces"),
+            }
+        }
+    }
+
+    with pytest.raises(ValueError, match="does not support HEAD /knowledge-spaces"):
+        validate_declarations(document, (declaration(method="HEAD"),))
+
+
+def test_validate_declarations_rejects_non_absolute_upstream_path() -> None:
+    document = {
+        "paths": {
+            "knowledge-spaces": {
+                "get": operation("knowledge-spaces:read", "listKnowledgeSpaces"),
+            }
+        }
+    }
+
+    with pytest.raises(ValueError, match="path must be absolute: knowledge-spaces"):
+        validate_declarations(document, (declaration(),))
+
+
+@pytest.mark.parametrize("value", [None, True, 0, "1048576"])
+def test_validate_declarations_rejects_invalid_response_byte_limits(value: object) -> None:
+    route = operation("knowledge-spaces:read", "listKnowledgeSpaces")
+    route["x-knowledge-fs-max-response-bytes"] = value
+
+    with pytest.raises(ValueError, match="no valid response byte limit"):
+        validate_declarations({"paths": {"/knowledge-spaces": {"get": route}}}, (declaration(),))
+
+
+def test_validate_declarations_rejects_request_header_references() -> None:
+    route = operation("knowledge-spaces:read", "listKnowledgeSpaces")
+    route["parameters"] = [{"$ref": "#/components/parameters/TraceId"}]
+
+    with pytest.raises(ValueError, match="request header references are not supported"):
+        validate_declarations({"paths": {"/knowledge-spaces": {"get": route}}}, (declaration(),))
+
+
+def operation(scope: str | None, operation_id: str, **overrides: object) -> dict[str, object]:
+    value: dict[str, object] = {
+        "operationId": operation_id,
+        "responses": {"200": {"content": {"application/json": {}}}},
+        "x-knowledge-fs-max-response-bytes": 1_048_576,
+    }
+    if scope is not None:
+        value["x-knowledge-fs-required-scope"] = scope
+    value.update(overrides)
+    return value
+
+
+def declaration(**overrides: object) -> ContractDeclaration:
+    value: dict[str, object] = {
+        "operation_id": "listKnowledgeSpaces",
+        "method": "GET",
+        "path": "knowledge-spaces",
+        "required_scope": "knowledge-spaces:read",
+        "response_kind": "buffered",
+        "max_response_bytes": 1_048_576,
+        "request_headers": (),
+        "response_headers": (),
+        "response_media_types": ("application/json",),
+    }
+    value.update(overrides)
+    return cast(ContractDeclaration, value)
+
+
+def write_fake_pnpm(path: Path, document: dict[str, object]) -> None:
+    path.write_text(
+        "#!/usr/bin/env python3\n"
+        "import sys\n"
+        "from pathlib import Path\n"
+        "output = Path(sys.argv[sys.argv.index('--output') + 1])\n"
+        f"output.write_text({json.dumps(document)!r})\n"
+    )
+    path.chmod(0o755)