958 lines
36 KiB
Python
958 lines
36 KiB
Python
"""Resource-scoped Dify Capability v2 issuance and public JWKS publication.
|
|
|
|
This module is deliberately independent from the legacy Console proxy and from database models.
|
|
Callers must construct requests from already-authorized Dify control-plane state and provide a
|
|
durable audit sink. Only the current private key signs; published JWKS contain current and previous
|
|
public keys so KnowledgeFS can verify an overlap window without ever receiving signing material.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import secrets
|
|
from collections.abc import Callable, Mapping
|
|
from datetime import UTC, datetime
|
|
from types import MappingProxyType
|
|
from typing import Final, Literal, NamedTuple, NoReturn, Protocol, TypedDict, cast
|
|
|
|
import jwt
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, RSAPublicKey
|
|
from jwt.algorithms import RSAAlgorithm
|
|
from pydantic import BaseModel, ConfigDict, Field, NonNegativeInt, PositiveInt, field_validator, model_validator
|
|
|
|
from configs import dify_config
|
|
from services.knowledge_fs.observability import (
|
|
KnowledgeFSCapabilityIssuanceMetric,
|
|
KnowledgeFSOperationalMetricsPort,
|
|
get_knowledge_fs_operational_metrics,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
type CapabilityCallerKind = Literal[
|
|
"interactive",
|
|
"service",
|
|
"agent",
|
|
"workflow",
|
|
"internal_worker",
|
|
"mcp",
|
|
]
|
|
type CapabilityResourceType = Literal[
|
|
"namespace",
|
|
"knowledge_space",
|
|
"document",
|
|
"job",
|
|
"query",
|
|
"research_task",
|
|
"source",
|
|
"upload_session",
|
|
]
|
|
|
|
_ALGORITHM: Final = "RS256"
|
|
_AUDIENCE: Final = "knowledge-fs"
|
|
_ISSUER: Final = "dify-control-plane"
|
|
_MAX_CONTENT_SCOPE_IDS: Final = 1_000
|
|
_MAX_TTL_SECONDS: Final = 60
|
|
|
|
|
|
class CapabilityPublicJwk(TypedDict):
|
|
alg: Literal["RS256"]
|
|
e: str
|
|
kid: str
|
|
kty: Literal["RSA"]
|
|
n: str
|
|
use: Literal["sig"]
|
|
|
|
|
|
class CapabilityPublicJwks(TypedDict):
|
|
keys: list[CapabilityPublicJwk]
|
|
|
|
|
|
class CapabilityAuthzRevision(BaseModel):
|
|
"""Monotonic Dify authorization revisions frozen into one capability."""
|
|
|
|
membership_epoch: NonNegativeInt
|
|
space_acl_epoch: NonNegativeInt
|
|
external_access_epoch: NonNegativeInt
|
|
credential_revision: NonNegativeInt | None
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
|
|
class CapabilityResource(BaseModel):
|
|
"""One signed resource; child resources always retain their parent KnowledgeSpace id."""
|
|
|
|
type: CapabilityResourceType
|
|
id: str = Field(min_length=1, max_length=255)
|
|
parent_id: str | None = Field(default=None, min_length=1, max_length=255)
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
@field_validator("id", "parent_id", mode="before")
|
|
@classmethod
|
|
def normalize_identifier(cls, value: object) -> object:
|
|
return value.strip() if isinstance(value, str) else value
|
|
|
|
@model_validator(mode="after")
|
|
def validate_parent_binding(self) -> CapabilityResource:
|
|
root_resource = self.type in {"namespace", "knowledge_space"}
|
|
if root_resource and self.parent_id is not None:
|
|
raise ValueError("Root capability resources must not declare parent_id")
|
|
if not root_resource and self.parent_id is None:
|
|
raise ValueError("Child capability resources require parent_id")
|
|
return self
|
|
|
|
|
|
class DifyCapabilityV2Claims(BaseModel):
|
|
"""Exact signed Capability v2 payload consumed by KnowledgeFS."""
|
|
|
|
iss: str
|
|
aud: str
|
|
sub: str
|
|
actor: str
|
|
namespace_id: str
|
|
azp: str
|
|
caller_kind: CapabilityCallerKind
|
|
cap_ver: Literal[2]
|
|
jti: str
|
|
iat: NonNegativeInt
|
|
nbf: NonNegativeInt
|
|
exp: PositiveInt
|
|
action: str
|
|
grant_id: str
|
|
authz_revision: CapabilityAuthzRevision
|
|
content_scope_ids: tuple[str, ...]
|
|
content_policy_revision: NonNegativeInt
|
|
resource: CapabilityResource
|
|
control_space_id: str
|
|
trace_id: str
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
@field_validator(
|
|
"iss",
|
|
"aud",
|
|
"sub",
|
|
"actor",
|
|
"namespace_id",
|
|
"azp",
|
|
"jti",
|
|
"action",
|
|
"grant_id",
|
|
"control_space_id",
|
|
"trace_id",
|
|
mode="before",
|
|
)
|
|
@classmethod
|
|
def normalize_required_string(cls, value: object) -> object:
|
|
return value.strip() if isinstance(value, str) else value
|
|
|
|
@field_validator("content_scope_ids")
|
|
@classmethod
|
|
def validate_content_scope_ids(cls, values: tuple[str, ...]) -> tuple[str, ...]:
|
|
normalized = tuple(value.strip() for value in values)
|
|
if len(normalized) > _MAX_CONTENT_SCOPE_IDS:
|
|
raise ValueError("content_scope_ids exceeds the bounded limit")
|
|
if any(not value for value in normalized) or len(set(normalized)) != len(normalized):
|
|
raise ValueError("content_scope_ids must contain unique non-empty ids")
|
|
return normalized
|
|
|
|
@model_validator(mode="after")
|
|
def validate_temporal_and_namespace_binding(self) -> DifyCapabilityV2Claims:
|
|
if not self.iat <= self.nbf < self.exp:
|
|
raise ValueError("Capability time claims must satisfy iat <= nbf < exp")
|
|
if self.resource.type == "namespace" and self.resource.id != self.namespace_id:
|
|
raise ValueError("Namespace capability resource must match namespace_id")
|
|
return self
|
|
|
|
|
|
class CapabilityIssueRequest(BaseModel):
|
|
"""Trusted, already-authorized control-plane input for one operation capability."""
|
|
|
|
actor: str = Field(min_length=1, max_length=255)
|
|
authz_revision: CapabilityAuthzRevision
|
|
caller_kind: CapabilityCallerKind
|
|
content_policy_revision: NonNegativeInt
|
|
content_scope_ids: tuple[str, ...] = ()
|
|
control_space_id: str = Field(min_length=1, max_length=255)
|
|
grant_id: str = Field(min_length=1, max_length=255)
|
|
namespace_id: str = Field(min_length=1, max_length=255)
|
|
operation_id: str = Field(min_length=1, max_length=128)
|
|
principal_id: str = Field(min_length=1, max_length=255)
|
|
resource: CapabilityResource
|
|
trace_id: str = Field(min_length=1, max_length=128)
|
|
ttl_seconds: PositiveInt | None = Field(default=None, le=_MAX_TTL_SECONDS)
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
@field_validator(
|
|
"actor",
|
|
"control_space_id",
|
|
"grant_id",
|
|
"namespace_id",
|
|
"operation_id",
|
|
"principal_id",
|
|
"trace_id",
|
|
mode="before",
|
|
)
|
|
@classmethod
|
|
def normalize_required_string(cls, value: object) -> object:
|
|
return value.strip() if isinstance(value, str) else value
|
|
|
|
|
|
class CapabilityIssuanceAuditEvent(BaseModel):
|
|
"""Non-secret issuance evidence. Raw JWTs and raw jti values are intentionally absent."""
|
|
|
|
action: str
|
|
actor: str
|
|
authz_revision: CapabilityAuthzRevision
|
|
caller_kind: CapabilityCallerKind
|
|
content_policy_revision: NonNegativeInt
|
|
content_scope_ids: tuple[str, ...]
|
|
control_space_id: str
|
|
expires_at: datetime
|
|
grant_id: str
|
|
issued_at: datetime
|
|
jti_hash: str
|
|
namespace_id: str
|
|
operation_id: str
|
|
resource_id: str
|
|
resource_parent_id: str | None
|
|
resource_type: CapabilityResourceType
|
|
subject: str
|
|
trace_id: str
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
|
|
class CapabilityIssuanceAuditor(Protocol):
|
|
def record(self, event: CapabilityIssuanceAuditEvent) -> None:
|
|
"""Persist one sanitized issuance event before the token leaves the issuer."""
|
|
|
|
|
|
class CapabilityIssuanceProfile(NamedTuple):
|
|
azp: str
|
|
enabled: bool
|
|
subject_prefix: str
|
|
|
|
|
|
CAPABILITY_ISSUANCE_PROFILES: Final[Mapping[CapabilityCallerKind, CapabilityIssuanceProfile]] = MappingProxyType(
|
|
{
|
|
"interactive": CapabilityIssuanceProfile("dify-console", True, "dify-account"),
|
|
"service": CapabilityIssuanceProfile("dify-service-api", True, "dify-kfs-credential"),
|
|
"agent": CapabilityIssuanceProfile("dify-agent", True, "dify-app"),
|
|
"workflow": CapabilityIssuanceProfile("dify-workflow", True, "dify-app"),
|
|
"internal_worker": CapabilityIssuanceProfile("dify-worker", True, "dify-worker"),
|
|
# The architecture plan requires an independent MCP policy before issuance.
|
|
"mcp": CapabilityIssuanceProfile("dify-mcp", False, "dify-mcp-session"),
|
|
}
|
|
)
|
|
|
|
|
|
class KnowledgeFSCapabilityOperation(NamedTuple):
|
|
action: str
|
|
allowed_caller_kinds: tuple[CapabilityCallerKind, ...]
|
|
method: Literal["DELETE", "GET", "PATCH", "POST", "PUT"]
|
|
path: str
|
|
resource_type: CapabilityResourceType
|
|
|
|
|
|
_STANDARD_CALLERS: Final[tuple[CapabilityCallerKind, ...]] = (
|
|
"interactive",
|
|
"service",
|
|
"agent",
|
|
"workflow",
|
|
)
|
|
_CONTROL_PLANE_CALLERS: Final[tuple[CapabilityCallerKind, ...]] = ("interactive", "service")
|
|
_LIST_CALLERS: Final[tuple[CapabilityCallerKind, ...]] = (*_CONTROL_PLANE_CALLERS, "internal_worker")
|
|
_PROVISION_CALLERS: Final[tuple[CapabilityCallerKind, ...]] = ("service", "internal_worker")
|
|
_INTERNAL_WORKER_CALLERS: Final[tuple[CapabilityCallerKind, ...]] = ("internal_worker",)
|
|
|
|
KNOWLEDGE_FS_CAPABILITY_OPERATIONS: Final[Mapping[str, KnowledgeFSCapabilityOperation]] = MappingProxyType(
|
|
{
|
|
"listKnowledgeSpaces": KnowledgeFSCapabilityOperation(
|
|
"knowledge_spaces.list", _LIST_CALLERS, "GET", "/knowledge-spaces", "namespace"
|
|
),
|
|
"batchKnowledgeSpaceProductSummaries": KnowledgeFSCapabilityOperation(
|
|
"knowledge_spaces.status.batch",
|
|
_CONTROL_PLANE_CALLERS,
|
|
"POST",
|
|
"/internal/knowledge-spaces/product-summaries/batch",
|
|
"namespace",
|
|
),
|
|
"provisionIntegratedKnowledgeSpace": KnowledgeFSCapabilityOperation(
|
|
"knowledge_spaces.provision",
|
|
_PROVISION_CALLERS,
|
|
"POST",
|
|
"/internal/knowledge-spaces/provision",
|
|
"namespace",
|
|
),
|
|
"activateDifyWorkspaceIntegration": KnowledgeFSCapabilityOperation(
|
|
"dify_integration.activate",
|
|
_INTERNAL_WORKER_CALLERS,
|
|
"POST",
|
|
"/internal/dify-integration/activate",
|
|
"namespace",
|
|
),
|
|
"freezeDifyWorkspaceIntegration": KnowledgeFSCapabilityOperation(
|
|
"dify_integration.freeze",
|
|
_INTERNAL_WORKER_CALLERS,
|
|
"POST",
|
|
"/internal/dify-integration/freeze",
|
|
"namespace",
|
|
),
|
|
"revokeCapabilityGrant": KnowledgeFSCapabilityOperation(
|
|
"capability_grants.revoke",
|
|
_PROVISION_CALLERS,
|
|
"POST",
|
|
"/internal/capability-grants/{grantId}/revoke",
|
|
"knowledge_space",
|
|
),
|
|
"fenceCapabilityKnowledgeSpace": KnowledgeFSCapabilityOperation(
|
|
"knowledge_spaces.fence",
|
|
_PROVISION_CALLERS,
|
|
"POST",
|
|
"/internal/knowledge-spaces/{id}/capability-fence",
|
|
"knowledge_space",
|
|
),
|
|
"deleteIntegratedKnowledgeSpace": KnowledgeFSCapabilityOperation(
|
|
"knowledge_spaces.delete",
|
|
_INTERNAL_WORKER_CALLERS,
|
|
"POST",
|
|
"/internal/knowledge-spaces/{id}/delete",
|
|
"knowledge_space",
|
|
),
|
|
"getKnowledgeSpace": KnowledgeFSCapabilityOperation(
|
|
"knowledge_spaces.read",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}",
|
|
"knowledge_space",
|
|
),
|
|
"updateKnowledgeSpace": KnowledgeFSCapabilityOperation(
|
|
"knowledge_spaces.update",
|
|
_STANDARD_CALLERS,
|
|
"PATCH",
|
|
"/knowledge-spaces/{id}",
|
|
"knowledge_space",
|
|
),
|
|
"getKnowledgeSpaceProductSettings": KnowledgeFSCapabilityOperation(
|
|
"knowledge_spaces.settings.read",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/product-settings",
|
|
"knowledge_space",
|
|
),
|
|
"updateKnowledgeSpaceProductSettings": KnowledgeFSCapabilityOperation(
|
|
"knowledge_spaces.settings.update",
|
|
_STANDARD_CALLERS,
|
|
"PATCH",
|
|
"/knowledge-spaces/{id}/product-settings",
|
|
"knowledge_space",
|
|
),
|
|
"listDocuments": KnowledgeFSCapabilityOperation(
|
|
"documents.list",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/documents",
|
|
"knowledge_space",
|
|
),
|
|
"uploadDocument": KnowledgeFSCapabilityOperation(
|
|
"documents.create",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/knowledge-spaces/{id}/documents",
|
|
"knowledge_space",
|
|
),
|
|
"getDocument": KnowledgeFSCapabilityOperation(
|
|
"documents.read",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/documents/{documentId}",
|
|
"document",
|
|
),
|
|
"getDocumentOutline": KnowledgeFSCapabilityOperation(
|
|
"documents.outline.read",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/documents/{documentId}/outline",
|
|
"document",
|
|
),
|
|
"listDocumentRevisions": KnowledgeFSCapabilityOperation(
|
|
"documents.revisions.list",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/documents/{documentId}/revisions",
|
|
"document",
|
|
),
|
|
"patchDocumentMetadata": KnowledgeFSCapabilityOperation(
|
|
"documents.metadata.update",
|
|
_STANDARD_CALLERS,
|
|
"PATCH",
|
|
"/knowledge-spaces/{id}/documents/{documentId}/metadata",
|
|
"document",
|
|
),
|
|
"listDocumentChunks": KnowledgeFSCapabilityOperation(
|
|
"documents.chunks.list",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/chunks",
|
|
"document",
|
|
),
|
|
"getDocumentChunk": KnowledgeFSCapabilityOperation(
|
|
"documents.chunks.read",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/chunks/{chunkId}",
|
|
"document",
|
|
),
|
|
"requestDocumentDeletion": KnowledgeFSCapabilityOperation(
|
|
"documents.delete",
|
|
_STANDARD_CALLERS,
|
|
"DELETE",
|
|
"/knowledge-spaces/{id}/documents/{documentId}",
|
|
"document",
|
|
),
|
|
"requestBulkDocumentDeletion": KnowledgeFSCapabilityOperation(
|
|
"documents.bulk.delete",
|
|
_STANDARD_CALLERS,
|
|
"DELETE",
|
|
"/knowledge-spaces/{id}/documents/bulk",
|
|
"knowledge_space",
|
|
),
|
|
"bulkReindexDocuments": KnowledgeFSCapabilityOperation(
|
|
"documents.bulk.reindex",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/knowledge-spaces/{id}/documents/bulk/reindex",
|
|
"knowledge_space",
|
|
),
|
|
"getDocumentCompilationJob": KnowledgeFSCapabilityOperation(
|
|
"document_jobs.read", _STANDARD_CALLERS, "GET", "/jobs/{id}", "job"
|
|
),
|
|
"cancelDocumentCompilationJob": KnowledgeFSCapabilityOperation(
|
|
"document_jobs.cancel", _STANDARD_CALLERS, "DELETE", "/jobs/{id}", "job"
|
|
),
|
|
"retryDocumentCompilationJob": KnowledgeFSCapabilityOperation(
|
|
"document_jobs.retry", _STANDARD_CALLERS, "POST", "/jobs/{id}/retry", "job"
|
|
),
|
|
"getBulkOperation": KnowledgeFSCapabilityOperation(
|
|
"bulk_jobs.read", _STANDARD_CALLERS, "GET", "/bulk-jobs/{id}", "job"
|
|
),
|
|
"listKnowledgeSpaceSources": KnowledgeFSCapabilityOperation(
|
|
"sources.list",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/sources",
|
|
"knowledge_space",
|
|
),
|
|
"createKnowledgeSpaceSource": KnowledgeFSCapabilityOperation(
|
|
"sources.create",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/knowledge-spaces/{id}/sources",
|
|
"knowledge_space",
|
|
),
|
|
"getKnowledgeSpaceSource": KnowledgeFSCapabilityOperation(
|
|
"sources.read",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/sources/{sourceId}",
|
|
"source",
|
|
),
|
|
"updateKnowledgeSpaceSource": KnowledgeFSCapabilityOperation(
|
|
"sources.update",
|
|
_STANDARD_CALLERS,
|
|
"PATCH",
|
|
"/knowledge-spaces/{id}/sources/{sourceId}",
|
|
"source",
|
|
),
|
|
"requestSourceDeletion": KnowledgeFSCapabilityOperation(
|
|
"sources.delete",
|
|
_STANDARD_CALLERS,
|
|
"DELETE",
|
|
"/knowledge-spaces/{id}/sources/{sourceId}",
|
|
"source",
|
|
),
|
|
"testKnowledgeSpaceSource": KnowledgeFSCapabilityOperation(
|
|
"sources.test",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/knowledge-spaces/{id}/sources/{sourceId}/test",
|
|
"source",
|
|
),
|
|
"crawlKnowledgeSpaceSource": KnowledgeFSCapabilityOperation(
|
|
"sources.crawl",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/knowledge-spaces/{id}/sources/{sourceId}/crawl",
|
|
"source",
|
|
),
|
|
"listKnowledgeSpaceSourcePages": KnowledgeFSCapabilityOperation(
|
|
"sources.pages.list",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/sources/{sourceId}/pages",
|
|
"source",
|
|
),
|
|
"importKnowledgeSpaceSourcePages": KnowledgeFSCapabilityOperation(
|
|
"sources.pages.import",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/knowledge-spaces/{id}/sources/{sourceId}/import",
|
|
"source",
|
|
),
|
|
"listKnowledgeSpaceSourceFiles": KnowledgeFSCapabilityOperation(
|
|
"sources.files.list",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/sources/{sourceId}/files",
|
|
"source",
|
|
),
|
|
"importKnowledgeSpaceSourceFiles": KnowledgeFSCapabilityOperation(
|
|
"sources.files.import",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/knowledge-spaces/{id}/sources/{sourceId}/import-files",
|
|
"source",
|
|
),
|
|
"listKnowledgeSpaceResearchTasks": KnowledgeFSCapabilityOperation(
|
|
"research_tasks.list",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/research-tasks",
|
|
"knowledge_space",
|
|
),
|
|
"listKnowledgeSpaceQualityTraces": KnowledgeFSCapabilityOperation(
|
|
"quality.traces.list",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/knowledge-spaces/{id}/quality/traces",
|
|
"knowledge_space",
|
|
),
|
|
"createQuery": KnowledgeFSCapabilityOperation(
|
|
"queries.create", _STANDARD_CALLERS, "POST", "/queries", "knowledge_space"
|
|
),
|
|
"getAnswerTrace": KnowledgeFSCapabilityOperation(
|
|
"queries.read", _STANDARD_CALLERS, "GET", "/queries/{traceId}", "query"
|
|
),
|
|
"listQueryEvidence": KnowledgeFSCapabilityOperation(
|
|
"queries.evidence.list", _STANDARD_CALLERS, "GET", "/queries/{traceId}/evidence", "query"
|
|
),
|
|
"listQueryConflicts": KnowledgeFSCapabilityOperation(
|
|
"queries.conflicts.list", _STANDARD_CALLERS, "GET", "/queries/{traceId}/conflicts", "query"
|
|
),
|
|
"listQueryMissing": KnowledgeFSCapabilityOperation(
|
|
"queries.missing.list", _STANDARD_CALLERS, "GET", "/queries/{traceId}/missing", "query"
|
|
),
|
|
"planResearchTask": KnowledgeFSCapabilityOperation(
|
|
"research_tasks.plan", _STANDARD_CALLERS, "POST", "/research-tasks/plan", "knowledge_space"
|
|
),
|
|
"createResearchTask": KnowledgeFSCapabilityOperation(
|
|
"research_tasks.create",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/research-tasks",
|
|
"knowledge_space",
|
|
),
|
|
"streamResearchTaskProgress": KnowledgeFSCapabilityOperation(
|
|
"research_tasks.stream",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/research-tasks/{id}/events",
|
|
"research_task",
|
|
),
|
|
"getResearchTask": KnowledgeFSCapabilityOperation(
|
|
"research_tasks.read", _STANDARD_CALLERS, "GET", "/research-tasks/{id}", "research_task"
|
|
),
|
|
"listResearchTaskPartials": KnowledgeFSCapabilityOperation(
|
|
"research_tasks.partials.list",
|
|
_STANDARD_CALLERS,
|
|
"GET",
|
|
"/research-tasks/{id}/partials",
|
|
"research_task",
|
|
),
|
|
"createUploadSession": KnowledgeFSCapabilityOperation(
|
|
"upload_sessions.create",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/knowledge-spaces/{id}/upload-sessions",
|
|
"knowledge_space",
|
|
),
|
|
"presignUploadSessionPart": KnowledgeFSCapabilityOperation(
|
|
"upload_sessions.write",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/upload-sessions/{id}/parts/{partNumber}/presign",
|
|
"upload_session",
|
|
),
|
|
"uploadSmallFile": KnowledgeFSCapabilityOperation(
|
|
"upload_sessions.write",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/upload-sessions/{id}/small-file",
|
|
"upload_session",
|
|
),
|
|
"completeUploadSession": KnowledgeFSCapabilityOperation(
|
|
"upload_sessions.complete",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/upload-sessions/{id}/complete",
|
|
"upload_session",
|
|
),
|
|
"abortUploadSession": KnowledgeFSCapabilityOperation(
|
|
"upload_sessions.abort",
|
|
_STANDARD_CALLERS,
|
|
"POST",
|
|
"/upload-sessions/{id}/abort",
|
|
"upload_session",
|
|
),
|
|
"cancelResearchTask": KnowledgeFSCapabilityOperation(
|
|
"research_tasks.cancel",
|
|
_STANDARD_CALLERS,
|
|
"DELETE",
|
|
"/research-tasks/{id}",
|
|
"research_task",
|
|
),
|
|
}
|
|
)
|
|
|
|
|
|
class CapabilityVerificationKey:
|
|
"""One public RSA verification key published during a rotation overlap."""
|
|
|
|
kid: str
|
|
public_key: RSAPublicKey
|
|
|
|
def __init__(self, *, kid: str, public_key: RSAPublicKey) -> None:
|
|
self.kid = _required_string(kid, "kid")
|
|
self.public_key = public_key
|
|
|
|
|
|
class CapabilitySigningKey:
|
|
"""The sole current private signing key; it is never serialized into JWKS or audit."""
|
|
|
|
kid: str
|
|
private_key: RSAPrivateKey
|
|
|
|
def __init__(self, *, kid: str, private_key: RSAPrivateKey) -> None:
|
|
self.kid = _required_string(kid, "kid")
|
|
self.private_key = private_key
|
|
|
|
def public_verification_key(self) -> CapabilityVerificationKey:
|
|
return CapabilityVerificationKey(kid=self.kid, public_key=self.private_key.public_key())
|
|
|
|
|
|
class RotatingCapabilityKeyRing:
|
|
"""Current signer plus bounded previous public keys for non-disruptive verification overlap."""
|
|
|
|
current: CapabilitySigningKey
|
|
previous: tuple[CapabilityVerificationKey, ...]
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
current: CapabilitySigningKey,
|
|
previous: tuple[CapabilityVerificationKey, ...] = (),
|
|
) -> None:
|
|
kids = [current.kid, *(key.kid for key in previous)]
|
|
if len(kids) != len(set(kids)):
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability key ids must be unique")
|
|
if len(previous) > 2:
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability key overlap exceeds two previous keys")
|
|
self.current = current
|
|
self.previous = previous
|
|
|
|
def sign(self, claims: DifyCapabilityV2Claims) -> str:
|
|
return jwt.encode(
|
|
claims.model_dump(mode="json"),
|
|
self.current.private_key,
|
|
algorithm=_ALGORITHM,
|
|
headers={"kid": self.current.kid, "typ": "JWT"},
|
|
)
|
|
|
|
def public_jwks(self) -> CapabilityPublicJwks:
|
|
keys = [self.current.public_verification_key(), *self.previous]
|
|
return {"keys": [_public_jwk(key) for key in keys]}
|
|
|
|
|
|
class IssuedKnowledgeFSCapability(NamedTuple):
|
|
claims: DifyCapabilityV2Claims
|
|
jti_hash: str
|
|
token: str
|
|
|
|
|
|
class KnowledgeFSCapabilityConfigurationError(RuntimeError):
|
|
"""Capability signing or publication configuration is invalid."""
|
|
|
|
|
|
class KnowledgeFSCapabilityPolicyError(RuntimeError):
|
|
"""A trusted issuance request violates the declared operation/profile contract."""
|
|
|
|
|
|
class KnowledgeFSCapabilityIssuer:
|
|
"""Issue exactly one short-lived action/resource capability and its sanitized audit event."""
|
|
|
|
_audit: CapabilityIssuanceAuditor
|
|
_audience: str
|
|
_issuer: str
|
|
_key_ring: RotatingCapabilityKeyRing
|
|
_max_ttl_seconds: int
|
|
_metrics: KnowledgeFSOperationalMetricsPort
|
|
_now: Callable[[], datetime]
|
|
_random_jti: Callable[[], str]
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
audit: CapabilityIssuanceAuditor,
|
|
key_ring: RotatingCapabilityKeyRing,
|
|
metrics: KnowledgeFSOperationalMetricsPort | None = None,
|
|
audience: str = _AUDIENCE,
|
|
issuer: str = _ISSUER,
|
|
max_ttl_seconds: int = _MAX_TTL_SECONDS,
|
|
now: Callable[[], datetime] = lambda: datetime.now(UTC),
|
|
random_jti: Callable[[], str] = lambda: secrets.token_urlsafe(24),
|
|
) -> None:
|
|
if max_ttl_seconds <= 0 or max_ttl_seconds > _MAX_TTL_SECONDS:
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability max TTL must be between 1 and 60 seconds")
|
|
self._audit = audit
|
|
self._audience = _required_string(audience, "audience")
|
|
self._issuer = _required_string(issuer, "issuer")
|
|
self._key_ring = key_ring
|
|
self._max_ttl_seconds = max_ttl_seconds
|
|
self._metrics = metrics or get_knowledge_fs_operational_metrics()
|
|
self._now = now
|
|
self._random_jti = random_jti
|
|
|
|
def issue(self, request: CapabilityIssueRequest) -> IssuedKnowledgeFSCapability:
|
|
"""Sign and audit one pre-authorized operation or fail before returning a token."""
|
|
operation = KNOWLEDGE_FS_CAPABILITY_OPERATIONS.get(request.operation_id)
|
|
if operation is None:
|
|
self._deny(request, "operation_not_registered", "Capability operation is not registered")
|
|
profile = CAPABILITY_ISSUANCE_PROFILES[request.caller_kind]
|
|
if not profile.enabled:
|
|
self._deny(request, "caller_profile_disabled", "MCP Capability issuance requires a dedicated policy")
|
|
if request.caller_kind not in operation.allowed_caller_kinds:
|
|
self._deny(
|
|
request,
|
|
"caller_kind_not_allowed",
|
|
"Caller profile is not allowed for this capability operation",
|
|
)
|
|
if request.resource.type != operation.resource_type:
|
|
self._deny(
|
|
request,
|
|
"resource_type_mismatch",
|
|
"Capability resource type does not match the operation",
|
|
)
|
|
if request.resource.type == "namespace" and request.resource.id != request.namespace_id:
|
|
self._deny(
|
|
request,
|
|
"namespace_mismatch",
|
|
"Namespace resource does not match the Dify workspace",
|
|
)
|
|
|
|
subject = f"{profile.subject_prefix}:{request.principal_id}"
|
|
if request.caller_kind == "interactive" and request.actor != subject:
|
|
self._deny(
|
|
request,
|
|
"interactive_actor_mismatch",
|
|
"Interactive capability actor must match its subject",
|
|
)
|
|
now = self._now()
|
|
if now.tzinfo is None or now.utcoffset() is None:
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability clock must return a timezone-aware datetime")
|
|
issued_at = int(now.timestamp())
|
|
ttl_seconds = request.ttl_seconds or self._max_ttl_seconds
|
|
if ttl_seconds > self._max_ttl_seconds:
|
|
self._deny(request, "ttl_exceeded", "Capability TTL exceeds the issuer maximum")
|
|
jti = _required_string(self._random_jti(), "jti")
|
|
claims = DifyCapabilityV2Claims(
|
|
iss=self._issuer,
|
|
aud=self._audience,
|
|
sub=subject,
|
|
actor=request.actor,
|
|
namespace_id=request.namespace_id,
|
|
azp=profile.azp,
|
|
caller_kind=request.caller_kind,
|
|
cap_ver=2,
|
|
jti=jti,
|
|
iat=issued_at,
|
|
nbf=issued_at,
|
|
exp=issued_at + ttl_seconds,
|
|
action=operation.action,
|
|
grant_id=request.grant_id,
|
|
authz_revision=request.authz_revision,
|
|
content_scope_ids=request.content_scope_ids,
|
|
content_policy_revision=request.content_policy_revision,
|
|
resource=request.resource,
|
|
control_space_id=request.control_space_id,
|
|
trace_id=request.trace_id,
|
|
)
|
|
try:
|
|
token = self._key_ring.sign(claims)
|
|
except Exception:
|
|
self._record_metric(request, outcome="failed", reason="signing_failure")
|
|
raise
|
|
jti_hash = hash_capability_jti(jti)
|
|
try:
|
|
self._audit.record(
|
|
CapabilityIssuanceAuditEvent(
|
|
action=claims.action,
|
|
actor=claims.actor,
|
|
authz_revision=claims.authz_revision,
|
|
caller_kind=claims.caller_kind,
|
|
content_policy_revision=claims.content_policy_revision,
|
|
content_scope_ids=claims.content_scope_ids,
|
|
control_space_id=claims.control_space_id,
|
|
expires_at=datetime.fromtimestamp(claims.exp, tz=UTC),
|
|
grant_id=claims.grant_id,
|
|
issued_at=datetime.fromtimestamp(claims.iat, tz=UTC),
|
|
jti_hash=jti_hash,
|
|
namespace_id=claims.namespace_id,
|
|
operation_id=request.operation_id,
|
|
resource_id=claims.resource.id,
|
|
resource_parent_id=claims.resource.parent_id,
|
|
resource_type=claims.resource.type,
|
|
subject=claims.sub,
|
|
trace_id=claims.trace_id,
|
|
)
|
|
)
|
|
except Exception:
|
|
self._record_metric(request, outcome="failed", reason="audit_failure")
|
|
raise
|
|
self._record_metric(request, outcome="issued", reason="success")
|
|
return IssuedKnowledgeFSCapability(claims=claims, jti_hash=jti_hash, token=token)
|
|
|
|
def _deny(self, request: CapabilityIssueRequest, reason: str, message: str) -> NoReturn:
|
|
self._record_metric(request, outcome="denied", reason=reason)
|
|
raise KnowledgeFSCapabilityPolicyError(message)
|
|
|
|
def _record_metric(
|
|
self,
|
|
request: CapabilityIssueRequest,
|
|
*,
|
|
outcome: Literal["denied", "failed", "issued"],
|
|
reason: str,
|
|
) -> None:
|
|
try:
|
|
self._metrics.record_capability_issuance(
|
|
KnowledgeFSCapabilityIssuanceMetric(
|
|
caller_kind=request.caller_kind,
|
|
operation_id=(
|
|
request.operation_id
|
|
if request.operation_id in KNOWLEDGE_FS_CAPABILITY_OPERATIONS
|
|
else "unknown"
|
|
),
|
|
outcome=outcome,
|
|
reason=reason,
|
|
)
|
|
)
|
|
except Exception:
|
|
logger.warning("KnowledgeFS capability metric export failed", exc_info=True)
|
|
|
|
def public_jwks(self) -> CapabilityPublicJwks:
|
|
"""Publish current and previous public keys; private RSA parameters are never returned."""
|
|
return self._key_ring.public_jwks()
|
|
|
|
|
|
def create_configured_knowledge_fs_capability_issuer(
|
|
*,
|
|
audit: CapabilityIssuanceAuditor,
|
|
) -> KnowledgeFSCapabilityIssuer | None:
|
|
"""Assemble the issuer only from validated Dify server configuration.
|
|
|
|
This factory deliberately never reads process environment directly. Rotation overlap accepts
|
|
only public RSA JWKs; KnowledgeFS receives the resulting public JWKS, never this private key.
|
|
"""
|
|
if not dify_config.KNOWLEDGE_FS_CAPABILITY_V2_ENABLED:
|
|
return None
|
|
private_key_secret = dify_config.KNOWLEDGE_FS_CAPABILITY_V2_PRIVATE_KEY_PEM
|
|
signing_kid = dify_config.KNOWLEDGE_FS_CAPABILITY_V2_SIGNING_KID
|
|
if private_key_secret is None or signing_kid is None:
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability v2 signing configuration is incomplete")
|
|
try:
|
|
private_key = serialization.load_pem_private_key(
|
|
private_key_secret.get_secret_value().encode(),
|
|
password=None,
|
|
)
|
|
except (TypeError, ValueError) as exc:
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability v2 private key PEM is invalid") from exc
|
|
if not isinstance(private_key, RSAPrivateKey):
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability v2 requires an RSA private key")
|
|
|
|
previous = _load_previous_public_keys(dify_config.KNOWLEDGE_FS_CAPABILITY_V2_PREVIOUS_PUBLIC_JWKS)
|
|
return KnowledgeFSCapabilityIssuer(
|
|
audit=audit,
|
|
audience=dify_config.KNOWLEDGE_FS_CAPABILITY_V2_AUDIENCE,
|
|
issuer=dify_config.KNOWLEDGE_FS_CAPABILITY_V2_ISSUER,
|
|
key_ring=RotatingCapabilityKeyRing(
|
|
current=CapabilitySigningKey(kid=signing_kid, private_key=private_key),
|
|
previous=previous,
|
|
),
|
|
max_ttl_seconds=dify_config.KNOWLEDGE_FS_CAPABILITY_V2_MAX_TTL_SECONDS,
|
|
)
|
|
|
|
|
|
def hash_capability_jti(jti: str) -> str:
|
|
"""Return the only jti representation allowed in cross-service audit records."""
|
|
return f"sha256:{hashlib.sha256(jti.encode()).hexdigest()}"
|
|
|
|
|
|
def _public_jwk(key: CapabilityVerificationKey) -> CapabilityPublicJwk:
|
|
raw = cast(dict[str, object], RSAAlgorithm.to_jwk(key.public_key, as_dict=True))
|
|
modulus = raw.get("n")
|
|
exponent = raw.get("e")
|
|
if not isinstance(modulus, str) or not isinstance(exponent, str):
|
|
raise KnowledgeFSCapabilityConfigurationError("RSA public key could not be serialized")
|
|
return {
|
|
"alg": _ALGORITHM,
|
|
"e": exponent,
|
|
"kid": key.kid,
|
|
"kty": "RSA",
|
|
"n": modulus,
|
|
"use": "sig",
|
|
}
|
|
|
|
|
|
def _load_previous_public_keys(raw_jwks: str | None) -> tuple[CapabilityVerificationKey, ...]:
|
|
if raw_jwks is None:
|
|
return ()
|
|
try:
|
|
decoded = json.loads(raw_jwks)
|
|
except json.JSONDecodeError as exc:
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability rotation JWKS is invalid JSON") from exc
|
|
if not isinstance(decoded, dict) or set(decoded) != {"keys"} or not isinstance(decoded["keys"], list):
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability rotation JWKS must contain only a keys array")
|
|
verification_keys: list[CapabilityVerificationKey] = []
|
|
for item in decoded["keys"]:
|
|
if not isinstance(item, dict):
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability rotation JWKS contains an invalid key")
|
|
if any(parameter in item for parameter in ("d", "p", "q", "dp", "dq", "qi", "oth")):
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability rotation JWKS must contain public keys only")
|
|
if (
|
|
item.get("alg") != _ALGORITHM
|
|
or item.get("kty") != "RSA"
|
|
or item.get("use") != "sig"
|
|
or not isinstance(item.get("kid"), str)
|
|
):
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability rotation JWKS key metadata is invalid")
|
|
try:
|
|
key = RSAAlgorithm.from_jwk(item)
|
|
except (TypeError, ValueError) as exc:
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability rotation public key is invalid") from exc
|
|
if not isinstance(key, RSAPublicKey):
|
|
raise KnowledgeFSCapabilityConfigurationError("Capability rotation JWKS must contain RSA public keys")
|
|
verification_keys.append(CapabilityVerificationKey(kid=item["kid"], public_key=key))
|
|
return tuple(verification_keys)
|
|
|
|
|
|
def _required_string(value: str, name: str) -> str:
|
|
normalized = value.strip()
|
|
if not normalized:
|
|
raise KnowledgeFSCapabilityConfigurationError(f"Capability {name} must be non-empty")
|
|
return normalized
|