coulomb/kontextual-engine
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

profile-scoped ACL policy and redaction

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-07 01:51:44 +02:00
parent e02f78d7e3
commit 88f9df6288
7 changed files with 136 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
docs/cmis-profiled-access-points-implementation.md
View File

@@ -105,6 +105,21 @@ These routes delegate to existing engine services:
Read-only profiles reject the same mutations with CMIS-shaped authorization
diagnostics before touching engine services.
## ACL And Redaction Slice
The Browser Binding adapter now projects profile-derived ACLs through
`GET /cmis/{access_point_id}/browser/acl/{object_id}`. ACL entries are derived
from the access profile and actor context:
- visible objects grant the current actor `cmis:read`,
- authoring profiles also project `cmis:write` and `cmis:delete`,
- public objects include a read-only `anyone` ACE,
- hidden objects return `not found` rather than partial metadata.
Relationship listings and change logs now apply the same asset visibility gates
as object reads. This prevents indirect leakage of confidential or restricted
asset IDs through relationship targets or audit-backed change entries.
Route-level tests are present but skip when the optional FastAPI/httpx service
dependencies are not installed. Runtime-level Browser Binding tests cover the
same behavior in the default Python test suite.

53
src/kontextual_engine/api/app.py
View File

@@ -421,6 +421,26 @@ class ServiceRuntime:
)
return content_stream
def cmis_acl(
self,
access_point_id: str,
object_id: str,
context: OperationContext,
) -> dict[str, Any]:
mapper = self._cmis_mapper(access_point_id)
decision = mapper.access_point.decide_action(CMISAction.GET_ACL, context, resource=object_id)
if not decision.allowed:
raise _cmis_authorization_error(decision, "getACL")
asset_id = _cmis_asset_id(object_id)
asset = self.repository.get_asset(asset_id)
acl = mapper.acl_for_asset(asset, context)
if acl is None:
raise NotFoundError(
"CMIS object not found",
details={"object_id": object_id, "access_point_id": access_point_id},
)
return acl
def cmis_create_document(
self,
access_point_id: str,
@@ -593,6 +613,7 @@ class ServiceRuntime:
projections = [
projection.to_dict()
for relationship in self.repository.list_relationships(source_id=source_id)
if self._cmis_relationship_visible(mapper, relationship, context)
if (projection := mapper.map_relationship(relationship, context))
]
return {"items": projections, "count": len(projections)}
@@ -621,6 +642,7 @@ class ServiceRuntime:
}
for event in events
if event.target.startswith("asset:")
if self._cmis_asset_visible(mapper, event.target.removeprefix("asset:"), context)
]
paged = changes[max(skip_count, 0) : max(skip_count, 0) + max(max_items, 0)]
return {
@@ -640,6 +662,29 @@ class ServiceRuntime:
details={"access_point_id": access_point_id, "available": [profile.name for profile in _cmis_profiles()]},
)
def _cmis_asset_visible(
self,
mapper: CMISDomainMapper,
asset_id: str,
context: OperationContext,
) -> bool:
try:
return mapper.access_point.exposes_asset(self.repository.get_asset(asset_id), context)
except NotFoundError:
return False
def _cmis_relationship_visible(
self,
mapper: CMISDomainMapper,
relationship: Any,
context: OperationContext,
) -> bool:
if not self._cmis_asset_visible(mapper, relationship.source_id, context):
return False
if relationship.target_kind == RelationshipTargetKind.ASSET:
return self._cmis_asset_visible(mapper, relationship.target_id, context)
return True
def create_asset(self, payload: dict[str, Any], context: OperationContext) -> dict[str, Any]:
classification = Classification.from_dict(payload["classification"])
result = self.asset_service().create_asset(
@@ -2087,6 +2132,14 @@ def create_app(runtime: ServiceRuntime | None = None):
) -> dict[str, Any]:
return response(runtime.cmis_content_stream, access_point_id, object_id, context)
@app.get("/cmis/{access_point_id}/browser/acl/{object_id:path}", tags=["cmis"])
def cmis_acl(
access_point_id: str,
object_id: str,
context: OperationContext = Depends(context_from_headers),
) -> dict[str, Any]:
return response(runtime.cmis_acl, access_point_id, object_id, context)
@app.post("/cmis/{access_point_id}/browser/document", tags=["cmis"])
def cmis_create_document(
access_point_id: str,

36
src/kontextual_engine/core/cmis.py
View File

@@ -50,6 +50,7 @@ class CMISAction(str, Enum):
GET_CHILDREN = "get_children"
GET_OBJECT = "get_object"
GET_CONTENT_STREAM = "get_content_stream"
GET_ACL = "get_acl"
QUERY = "query"
GET_RELATIONSHIPS = "get_relationships"
GET_CHANGE_LOG = "get_change_log"
@@ -77,6 +78,7 @@ ACTION_CAPABILITIES: dict[CMISAction, CMISCapability] = {
CMISAction.GET_CHILDREN: CMISCapability.NAVIGATION,
CMISAction.GET_OBJECT: CMISCapability.OBJECT_READ,
CMISAction.GET_CONTENT_STREAM: CMISCapability.CONTENT_STREAM_READ,
CMISAction.GET_ACL: CMISCapability.ACL,
CMISAction.QUERY: CMISCapability.DISCOVERY_QUERY,
CMISAction.GET_RELATIONSHIPS: CMISCapability.RELATIONSHIPS,
CMISAction.GET_CHANGE_LOG: CMISCapability.CHANGE_LOG,
@@ -431,6 +433,7 @@ class CMISObjectProjection:
content_stream: dict[str, Any] | None = None
version: dict[str, Any] | None = None
relationships: tuple[str, ...] = ()
acl: dict[str, Any] | None = None
def to_dict(self) -> dict[str, Any]:
return compact_dict(
@@ -445,6 +448,7 @@ class CMISObjectProjection:
"content_stream": dict(self.content_stream or {}),
"version": dict(self.version or {}),
"relationships": list(self.relationships),
"acl": dict(self.acl or {}),
}
)
@@ -536,6 +540,7 @@ class CMISDomainMapper:
content_stream=content_stream,
version=self.version_properties(asset, current_version, versions),
relationships=tuple(relationship_ids),
acl=self.acl_for_asset(asset, context),
)
def map_relationship(
@@ -575,6 +580,36 @@ class CMISDomainMapper:
allowable_actions=(CMISAction.GET_OBJECT, CMISAction.GET_RELATIONSHIPS),
)
def acl_for_asset(self, asset: KnowledgeAsset, context: OperationContext) -> dict[str, Any] | None:
visibility = self.access_point.profile.decide_asset_visibility(asset, context)
if not visibility.allowed:
return None
permissions = ["cmis:read"]
if self.access_point.profile.allow_mutations:
permissions.extend(["cmis:write", "cmis:delete"])
entries = [
{
"principal_id": context.actor.id,
"permissions": permissions,
"direct": True,
}
]
if asset.classification.sensitivity == Sensitivity.PUBLIC:
entries.append(
{
"principal_id": "anyone",
"permissions": ["cmis:read"],
"direct": False,
}
)
return {
"object_id": self.asset_object_id(asset.id),
"is_exact": True,
"aces": entries,
"derived_from": "kontextual-profile-policy",
"profile": self.access_point.profile.name,
}
def asset_object_id(self, asset_id: str) -> str:
return f"cmis:asset:{asset_id}"
@@ -679,6 +714,7 @@ class CMISDomainMapper:
candidates = [
CMISAction.GET_OBJECT,
CMISAction.GET_CONTENT_STREAM,
CMISAction.GET_ACL,
CMISAction.GET_RELATIONSHIPS,
CMISAction.UPDATE_PROPERTIES,
CMISAction.DELETE_OBJECT,

1
tests/cmis/test_cmis_browser_binding_api.py
View File

@@ -83,6 +83,7 @@ def test_cmis_browser_binding_routes_are_advertised_in_openapi(cmis_client) -> N
assert "/cmis/{access_point_id}/browser/children" in paths
assert "/cmis/{access_point_id}/browser/object/{object_id}" in paths
assert "/cmis/{access_point_id}/browser/content/{object_id}" in paths
assert "/cmis/{access_point_id}/browser/acl/{object_id}" in paths
assert "/cmis/{access_point_id}/browser/query" in paths
assert "/cmis/{access_point_id}/browser/relationships" in paths
assert "/cmis/{access_point_id}/browser/changes" in paths

29
tests/cmis/test_cmis_runtime_browser_binding.py
View File

@@ -65,6 +65,16 @@ def cmis_runtime() -> tuple[ServiceRuntime, object]:
},
context,
)
runtime.create_relationship(
{
"source_asset_id": "asset-runtime-source",
"target_id": "asset-runtime-confidential",
"predicate": "mentions_sensitive",
"target_kind": "asset",
"confidence": 0.5,
},
context,
)
return runtime, context
@@ -105,6 +115,7 @@ def test_runtime_cmis_browser_content_query_relationships_and_changes(cmis_runti
assert relationships["count"] == 1
assert relationships["items"][0]["properties"]["cmis:targetId"] == "cmis:asset:asset-runtime-public"
assert changes["total_num_items"] >= 3
assert all(change["object_id"] != "cmis:asset:asset-runtime-confidential" for change in changes["changes"])
def test_runtime_cmis_browser_rejects_unsupported_query_subset(cmis_runtime) -> None:
@@ -173,3 +184,21 @@ def test_runtime_cmis_readonly_profile_rejects_mutations(cmis_runtime) -> None:
)
assert "CMIS operation denied" in str(exc_info.value)
def test_runtime_cmis_acl_projection_and_redaction(cmis_runtime) -> None:
runtime, context = cmis_runtime
public_acl = runtime.cmis_acl("readonly-browser", "cmis:asset:asset-runtime-public", context)
internal_acl = runtime.cmis_acl("governed-authoring", "cmis:asset:asset-runtime-source", context)
assert public_acl["is_exact"] is True
assert {entry["principal_id"] for entry in public_acl["aces"]} == {"cmis-runtime", "anyone"}
assert ["cmis:read", "cmis:write", "cmis:delete"] in [
entry["permissions"] for entry in internal_acl["aces"]
]
with pytest.raises(Exception) as exc_info:
runtime.cmis_acl("readonly-browser", "cmis:asset:asset-runtime-confidential", context)
assert "CMIS object not found" in str(exc_info.value)

1
tests/test_service_api.py
View File

@@ -657,6 +657,7 @@ def test_service_health_readiness_version_and_openapi_contracts(client) -> None:
assert "/cmis" in paths
assert "/cmis/{access_point_id}/browser" in paths
assert "/cmis/{access_point_id}/browser/children" in paths
assert "/cmis/{access_point_id}/browser/acl/{object_id}" in paths
assert "/cmis/{access_point_id}/browser/document" in paths
assert "/cmis/{access_point_id}/browser/object/{object_id}/properties" in paths
assert "/api/v1/assets" in paths

2
workplans/KONT-WP-0012-cmis-profiled-access-points.md
View File

@@ -125,7 +125,7 @@ Acceptance:
```task
id: KONT-WP-0012-T005
status: todo
status: done
priority: high
state_hub_task_id: "64289d84-d7a2-4c03-8fa6-5f439bc233fe"
```