From de0553a7114d78df8e4f7da4f72168bc8a0f8d0a Mon Sep 17 00:00:00 2001 From: Maojia Sheng Date: Tue, 3 Nov 2026 21:58:39 +0800 Subject: [PATCH] feat(auth): scope GLOBAL reads to owner policy Co-authored-by: TRAE CLI --- openviking/core/global_policy.py | 118 +++++++++++++ openviking/core/namespace.py | 3 +- openviking/server/identity.py | 128 +++++++++++++++ openviking/service/core.py | 1 + openviking/service/fs_service.py | 26 ++- openviking/storage/acl.py | 8 +- openviking/storage/viking_fs/__init__.py | 5 + openviking/storage/viking_fs/_access.py | 155 +++++++++++++++++- openviking/storage/viking_fs/_grep.py | 32 ++-- openviking/storage/viking_fs/_ops.py | 65 +++++++- openviking/storage/viking_fs/_semantic.py | 55 ++++++- .../storage/viking_vector_index_backend.py | 9 +- 12 files changed, 568 insertions(+), 37 deletions(-) diff --git a/openviking/core/global_policy.py b/openviking/core/global_policy.py index 98bb28f7bf..eea2e20034 100644 --- a/openviking/core/global_policy.py +++ b/openviking/core/global_policy.py @@ -13,6 +13,8 @@ GLOBAL_POLICY_INVALID = "GLOBAL_POLICY_INVALID" GLOBAL_POLICY_SCHEMA_UNSUPPORTED = "GLOBAL_POLICY_SCHEMA_UNSUPPORTED" +# Keep the ordinary-content boundary aligned with Rust PROTECTED_ROOTS. +PROTECTED_ROOTS = ("privacy", "sessions") class GlobalPolicyError(ValueError): @@ -34,6 +36,100 @@ class GlobalPolicyRule: recursive: bool +@dataclass(frozen=True) +class PolicyPathScope: + path: str | None + recursive: bool + + +def _scope_contains(scope: PolicyPathScope, path: str | None) -> bool: + return scope.path == path or ( + scope.recursive + and (scope.path is None or (path is not None and path.startswith(scope.path + "/"))) + ) + + +def _scope_intersection(a: PolicyPathScope, b: PolicyPathScope) -> PolicyPathScope | None: + if a.path == b.path: + return PolicyPathScope(a.path, a.recursive and b.recursive) + if a.recursive and _scope_contains(a, b.path): + return b + if b.recursive and _scope_contains(b, a.path): + return a + return None + + +def _finish_scopes( + includes: list[PolicyPathScope], excludes: list[PolicyPathScope] +) -> PolicyScopePlan: + exact = {s.path for s in excludes if not s.recursive} + recursive = {s.path for s in excludes if s.recursive} + + def covered(scope: PolicyPathScope) -> bool: + if None in recursive or scope.path in recursive: + return True + if not scope.recursive and scope.path in exact: + return True + parts = scope.path.split("/") if scope.path is not None else [] + return any("/".join(parts[:i]) in recursive for i in range(1, len(parts))) + + def ordered(scopes): + return tuple(sorted(set(scopes), key=lambda s: (s.path or "", s.recursive))) + + remaining = [s for s in includes if not covered(s)] + if not remaining: + return PolicyScopePlan("denied", (), ()) + return PolicyScopePlan("allowed", ordered(remaining), ordered(excludes)) + + +@dataclass(frozen=True) +class PolicyScopePlan: + status: Literal["allowed", "denied"] + includes: tuple[PolicyPathScope, ...] + excludes: tuple[PolicyPathScope, ...] + + def allows(self, path: str) -> bool: + """Membership requires both the includes and excludes. + + Paths the policy grammar cannot represent fail closed, matching + ``is_global_policy_path_allowed``, so enumeration never aborts on an + ordinary filename the validator rejects. + """ + try: + validate_global_policy_path(path) + except GlobalPolicyError: + return False + return ( + self.status == "allowed" + and any(_scope_contains(s, path) for s in self.includes) + and not any(_scope_contains(s, path) for s in self.excludes) + ) + + def restrict_scope(self, path: str | None, *, recursive: bool = True) -> PolicyScopePlan: + request = _validated_scope(path, recursive) + if self.status != "allowed": + return PolicyScopePlan("denied", (), ()) + return _finish_scopes( + [s for item in self.includes if (s := _scope_intersection(item, request)) is not None], + [s for item in self.excludes if (s := _scope_intersection(item, request)) is not None], + ) + + def has_allowed_descendant(self, path: str | None) -> bool: + try: + restricted = self.restrict_scope(path) + except GlobalPolicyError: + return False + return any(s.recursive or s.path != path for s in restricted.includes) + + +def _validated_scope(path: str | None, recursive: bool) -> PolicyPathScope: + if not isinstance(recursive, bool) or (path is None and not recursive): + raise GlobalPolicyError("The owner root requires recursive=True") + if path is not None: + validate_global_policy_path(path) + return PolicyPathScope(path, recursive) + + @dataclass(frozen=True) class GlobalAccessPolicy: """Validated parser output. The body is preserved, not sanitized or executed. @@ -55,6 +151,28 @@ def allows_path(self, path: str) -> bool: """Evaluate one owner-relative path, failing closed for non-valid policies.""" return is_global_policy_path_allowed(self, path) + def restrict_scope(self, path: str | None = None, *, recursive: bool = True) -> PolicyScopePlan: + """Clip one user's ordinary-content scope; apply includes AND excludes. + + This computes a request plan, not filesystem existence or final ACL authorization. + """ + request = _validated_scope(path, recursive) + if self.status != "valid" or self._native is None: + return PolicyScopePlan("denied", (), ()) + includes = [request] if self.default == "allow" else [] + excludes = [] + for rule in self.rules: + if rule.effect == "allow" and self.default == "allow": + continue + overlap = _scope_intersection(request, PolicyPathScope(rule.path, rule.recursive)) + if overlap is not None: + (includes if rule.effect == "allow" else excludes).append(overlap) + for root in PROTECTED_ROOTS: + overlap = _scope_intersection(request, PolicyPathScope(root, True)) + if overlap is not None: + excludes.append(overlap) + return _finish_scopes(includes, excludes) + def _native_types(): from openviking.pyagfs import get_user_global_policy_types diff --git a/openviking/core/namespace.py b/openviking/core/namespace.py index e469f0a133..2f496475c4 100644 --- a/openviking/core/namespace.py +++ b/openviking/core/namespace.py @@ -186,7 +186,7 @@ def visible_roots(ctx: RequestContext) -> list[str]: return [ "viking://resources", "viking://agent", - canonical_user_root(ctx), + *(ctx.data_access.effective_targets if ctx.data_access else [canonical_user_root(ctx)]), ] @@ -349,6 +349,7 @@ def is_content_root_uri( and len(parts) == classification.content_index + 1 ) + def _validate_peer_id_segments(parts: list[str]) -> None: if len(parts) >= 4 and parts[0] == "user" and parts[2] == "peers": _require_peer_id_segment(parts[3]) diff --git a/openviking/server/identity.py b/openviking/server/identity.py index f028ca0909..04af647c95 100644 --- a/openviking/server/identity.py +++ b/openviking/server/identity.py @@ -10,8 +10,11 @@ from typing import TYPE_CHECKING, Any, List, Optional if TYPE_CHECKING: + from openviking.storage.acl import AclAction + from openviking.storage.expr import FilterExpr from openviking.storage.viking_fs import VikingFS +from openviking.core.global_policy import GlobalPolicyError, PolicyScopePlan from openviking_cli.session.user_id import UserIdentifier @@ -151,6 +154,121 @@ def __post_init__(self) -> None: self.role = Role.parse(self.role) +class AccessDecision(str, Enum): + ALLOWED = "allowed" + CONTAINER = "container" + DENIED = "denied" + + +@dataclass(frozen=True) +class DataAccessScope: + owner_user_id: str + operation: AclAction + plan: PolicyScopePlan + policy_hash: str | None + request_root: str + requested_targets: tuple[str, ...] + effective_targets: tuple[str, ...] + + def query_filter(self) -> FilterExpr: + """Filter before query limits; public paths retain actor ACL filtering.""" + from openviking.storage.expr import And, Or, PathScope, RawDSL + + public_filters = [ + PathScope("uri", "viking://resources", depth=-1), + PathScope("uri", "viking://agent", depth=-1), + ] + if self.plan.status != "allowed" or not self.plan.includes: + # No owner content is in scope (e.g. a public target survived while + # every owner target was denied); keep only the public roots. An + # empty include list must not degrade to an unfiltered scan. + return Or(public_filters) + + def scope_uri(scope): + return self.request_root + ("/" + scope.path if scope.path else "") + + owner_filters = [ + Or( + [ + PathScope("uri", scope_uri(s), depth=-1 if s.recursive else 0) + for s in self.plan.includes + ] + ) + ] + owner_filters.append( + RawDSL({"op": "must_not", "field": "uri", "conds": [self.request_root], "para": "-d=0"}) + ) + owner_filters.extend( + RawDSL( + { + "op": "must_not", + "field": "uri", + "conds": [scope_uri(s)], + "para": "-d=-1" if s.recursive else "-d=0", + } + ) + for s in self.plan.excludes + ) + return Or([*public_filters, And(owner_filters)]) + + def _is_fully_allowed_directory(self, path: str | None) -> bool: + """Whether the policy grants the directory and every descendant.""" + try: + restricted = self.plan.restrict_scope(path) + except GlobalPolicyError: + return False + return ( + restricted.status == "allowed" + and not restricted.excludes + and any(scope.path == path and scope.recursive for scope in restricted.includes) + ) + + def allows_overview(self, uri: str) -> bool: + """L1 overview is exposed only for directories with no denied descendants. + + A partially allowed container may still expose its L0 abstract, but its + overview aggregates child summaries and would otherwise leak the + denied subtree. + """ + uri = uri.rstrip("/") + if uri.endswith("/.overview.md"): + uri = uri[: -len("/.overview.md")] + if uri != self.request_root and not uri.startswith(self.request_root + "/"): + return False + path = uri[len(self.request_root) + 1 :] or None + return self._is_fully_allowed_directory(path) + + def decide(self, uri: str) -> AccessDecision: + uri = uri.rstrip("/") + if uri != self.request_root and not uri.startswith(self.request_root + "/"): + return AccessDecision.DENIED + if not any(uri == t or uri.startswith(t + "/") for t in self.requested_targets): + return AccessDecision.DENIED + path = uri[len(self.request_root) + 1 :] or None + if path == "GLOBAL_POLICY.md" or (path and path.startswith("GLOBAL_POLICY.md/")): + return AccessDecision.DENIED + if path is not None and (path == ".overview.md" or path.endswith("/.overview.md")): + if not self.allows_overview(uri): + return AccessDecision.DENIED + if path is not None and self.plan.allows(path): + return AccessDecision.ALLOWED + if self.plan.has_allowed_descendant(path): + return AccessDecision.CONTAINER + return AccessDecision.DENIED + + +def data_owner_id(ctx: RequestContext) -> str: + return ctx.data_access.owner_user_id if ctx.data_access else ctx.user.user_id + + +def data_user_root(ctx: RequestContext) -> str: + return f"viking://user/{data_owner_id(ctx)}" + + +def is_delegated(ctx: RequestContext) -> bool: + return ctx.data_access is not None + + @dataclass class RequestContext: """Request-level context, flows through Router -> Service -> VikingFS.""" @@ -174,9 +292,19 @@ class RequestContext: # Trusted background workers may maintain derived content without replaying # the triggering user's resource ACL. Tenant and namespace checks still apply. bypass_acl: bool = False + data_access: DataAccessScope | None = field(default=None, repr=False) def __post_init__(self) -> None: self.role = Role.parse(self.role) + if self.data_access is not None: + from openviking.storage.acl import AclAction + + if ( + self.role != Role.GLOBAL + or self.bypass_acl + or self.data_access.operation is not AclAction.READ + ): + raise ValueError("Delegated data access requires GLOBAL READ without ACL bypass") @property def account_id(self) -> str: diff --git a/openviking/service/core.py b/openviking/service/core.py index 0a8232f88e..d5adcdc32b 100644 --- a/openviking/service/core.py +++ b/openviking/service/core.py @@ -572,6 +572,7 @@ async def initialize( user_count, ) self._privacy_config_service = UserPrivacyConfigService(self._viking_fs) + self._viking_fs.set_global_policy_service(self._privacy_config_service.global_policies) self._privacy_sanitizer = PrivacySanitizer(self._privacy_config_service) self._viking_fs.set_privacy_sanitizer(self._privacy_sanitizer) self._resource_processor.set_privacy_sanitizer(self._privacy_sanitizer) diff --git a/openviking/service/fs_service.py b/openviking/service/fs_service.py index 6fce1c7912..e8b68983b2 100644 --- a/openviking/service/fs_service.py +++ b/openviking/service/fs_service.py @@ -249,7 +249,11 @@ async def _attach_and_filter_tags( tags_by_uri: Dict[str, List[str]] = {} if self._vikingdb: uris = list( - dict.fromkeys(str(entry.get("uri") or "") for entry in entries if entry.get("uri")) + dict.fromkeys( + str(entry["uri"]) + for entry in entries + if entry.get("uri") and entry.get("access") != "container" + ) ) if uris: records = await VikingDBManagerProxy(self._vikingdb, ctx).filter( @@ -263,6 +267,8 @@ async def _attach_and_filter_tags( for record in records: records_by_uri.setdefault(str(record.get("uri") or ""), []).append(record) for entry in entries: + if entry.get("access") == "container": + continue uri = str(entry.get("uri") or "") levels = {0, 1} if entry.get("isDir", False) else {2} found: List[str] = [] @@ -278,6 +284,17 @@ async def _attach_and_filter_tags( tags_by_uri[uri] = found result_entries: List[Dict[str, Any]] = [] for entry in entries: + if entry.get("access") == "container": + # Navigation does not grant access to indexed tags or tag matching. + if not normalized_tags: + result_entries.append( + { + key: value + for key, value in entry.items() + if key not in {"tags", "search_tags"} + } + ) + continue entry_tags = tags_by_uri.get(str(entry.get("uri") or ""), []) if normalized_tags and not set(normalized_tags).issubset(entry_tags): continue @@ -1503,8 +1520,10 @@ async def _read_content( async def read(self, uri: str, ctx: RequestContext, offset: int = 0, limit: int = -1) -> str: """Read file content, restoring Privacy references for Skills and Sessions.""" + uri = await self._resolve_uri(uri, ctx) + ctx = self._ensure_initialized().authorize_read(ctx, [uri]) resolved_uri, content = await self._read_content(uri, ctx) - if self._privacy_config_service is not None: + if self._privacy_config_service is not None and ctx.data_access is None: skill_name = get_skill_name_from_uri(resolved_uri) session_id = get_session_id_from_uri(resolved_uri, canonical_session_root(ctx)) if skill_name is not None: @@ -1542,6 +1561,7 @@ async def read_visible( # Resolve id to URI before classify_uri / visible_content which expect # a real viking:// URI. resolved_uri = await self._resolve_uri(uri, ctx) + ctx = self._ensure_initialized().authorize_read(ctx, [resolved_uri]) max_bytes = ( self._privacy_sanitizer.max_content_bytes if self._privacy_sanitizer is not None @@ -1567,7 +1587,7 @@ async def _make_content_visible(self, uri: str, content: str, ctx: RequestContex """Hide memory metadata and sanitize private values in already-read text.""" if classify_uri(uri).is_memory: content = visible_content(content, uri=uri) - if self._privacy_sanitizer is not None: + if self._privacy_sanitizer is not None and ctx.data_access is None: content = ( await self._privacy_sanitizer.sanitize( content, diff --git a/openviking/storage/acl.py b/openviking/storage/acl.py index 1f4ad37fb1..75f7079972 100644 --- a/openviking/storage/acl.py +++ b/openviking/storage/acl.py @@ -367,9 +367,7 @@ async def _records_for_uris( records: list[dict[str, Any]] = [] for offset in range(0, len(unique), 100): conditions = [PathScope("uri", uri, depth=0) for uri in unique[offset : offset + 100]] - records.extend( - await self._scroll_all(Or(conditions), _ACL_OUTPUT_FIELDS, ctx) - ) + records.extend(await self._scroll_all(Or(conditions), _ACL_OUTPUT_FIELDS, ctx)) return records async def _scroll_all( @@ -387,7 +385,7 @@ async def _scroll_all( cursor=cursor, output_fields=output_fields, # ACL resolution must see records the caller's own grants hide. - ctx=replace(ctx, bypass_acl=True), + ctx=replace(ctx, bypass_acl=True, data_access=None), ) records.extend(page) if cursor is None: @@ -412,7 +410,7 @@ async def iter_denied_uris( limit=1, cursor=cursor, output_fields=["uri"], - ctx=replace(ctx, bypass_acl=True), + ctx=replace(ctx, bypass_acl=True, data_access=None), ) for record in records: candidate = record["uri"] diff --git a/openviking/storage/viking_fs/__init__.py b/openviking/storage/viking_fs/__init__.py index 25fcb8bd27..543d232eec 100644 --- a/openviking/storage/viking_fs/__init__.py +++ b/openviking/storage/viking_fs/__init__.py @@ -163,6 +163,7 @@ def __init__( self._background_tasks: set = set() self._deletion_guard: Optional[Callable[[str, str], bool]] = None self._privacy_sanitizer: Optional[Any] = None + self._global_policy_service: Optional[Any] = None self._post_write_hook: Optional[Callable[[str, Optional[RequestContext], int], Any]] = None def set_vlm_resolver(self, resolver: "VLMResolver") -> None: @@ -173,6 +174,10 @@ def set_privacy_sanitizer(self, sanitizer: Optional[Any]) -> None: """Install the account service's text sanitizer for direct writes.""" self._privacy_sanitizer = sanitizer + def set_global_policy_service(self, service: Any) -> None: + """Install the owner policy registry supplied by the account service.""" + self._global_policy_service = service + def set_post_write_hook( self, hook: Optional[Callable[[str, Optional[RequestContext], int], Any]], diff --git a/openviking/storage/viking_fs/_access.py b/openviking/storage/viking_fs/_access.py index 85247fefe5..f90a339d63 100644 --- a/openviking/storage/viking_fs/_access.py +++ b/openviking/storage/viking_fs/_access.py @@ -6,6 +6,7 @@ import inspect import re from contextlib import contextmanager +from dataclasses import replace from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Union from openviking.core.context import ContextLevel @@ -19,7 +20,7 @@ from openviking.core.namespace_policy import is_system_custom_asset_uri, is_system_uri from openviking.resource.watch_storage import is_watch_task_control_uri from openviking.server.error_mapping import is_not_found_error -from openviking.server.identity import RequestContext, Role +from openviking.server.identity import AccessDecision, DataAccessScope, RequestContext, Role from openviking.storage.acl import ( AclAction, AclEntry, @@ -104,9 +105,7 @@ def _ctx_or_default(self, ctx: Optional[RequestContext]) -> RequestContext: def _require_request_context(self, ctx: Optional[RequestContext]) -> RequestContext: """Resolve an account context without inventing the default account.""" if ctx is None and self._bound_ctx.get() is None: - raise RuntimeError( - "Account request context is required for account-scoped operations" - ) + raise RuntimeError("Account request context is required for account-scoped operations") return self._ctx_or_default(ctx) @contextmanager @@ -168,6 +167,125 @@ def _encrypted_temp_path(self, path: str) -> str: digest = hashlib.sha256(relative_path.encode("utf-8")).hexdigest() return f"{mount_prefix}{temp_root}/{digest}.encrypt" + def authorize_read( + self, ctx: RequestContext, targets: Sequence[str], *, recursive: bool = True + ) -> RequestContext: + """Derive one owner scope, preserving actor identity and public ACLs.""" + canonical = tuple( + dict.fromkeys((t if t == "viking://" else t.rstrip("/")) for t in targets) + ) + owners = set() + for target in canonical: + parts = self._safe_uri_parts(target) + if parts[:1] == ["user"] and len(parts) >= 2: + owners.add(parts[1]) + elif parts == ["user"] and ctx.role == Role.GLOBAL: + raise PermissionDeniedError("GLOBAL_ACCESS_DENIED: explicit owner required") + if ctx.role != Role.GLOBAL: + return ctx + if ctx.bypass_acl or len(owners) > 1: + raise PermissionDeniedError( + "GLOBAL_ACCESS_DENIED: one owner and no ACL bypass required" + ) + if ctx.data_access is not None: + if owners - {ctx.data_access.owner_user_id}: + raise PermissionDeniedError("GLOBAL_ACCESS_DENIED: owner scope cannot change") + return ctx + if not owners or owners == {ctx.user.user_id}: + return ctx + owner = next(iter(owners)) + service = getattr(self, "_global_policy_service", None) + policy = service.for_owner(ctx, owner) if service is not None else None + if policy is None or policy.status != "valid": + raise PermissionDeniedError("GLOBAL_ACCESS_DENIED: owner policy unavailable") + from openviking.core.global_policy import ( + GlobalPolicyError, + PolicyPathScope, + PolicyScopePlan, + ) + + root = f"viking://user/{owner}" + plan = policy.restrict_scope() + plan = PolicyScopePlan( + plan.status, plan.includes, (*plan.excludes, PolicyPathScope("GLOBAL_POLICY.md", True)) + ) + effective = [] + scoped_includes = [] + for target in canonical: + if target != root and not target.startswith(root + "/"): + effective.append(target) + continue + sub_path = target[len(root) + 1 :] or None + try: + restricted = plan.restrict_scope(sub_path, recursive=recursive) + except GlobalPolicyError: + if sub_path is None: + raise + # The policy grammar cannot represent this path, so deny it. + continue + if restricted.status != "allowed": + continue + if sub_path == ".overview.md" or ( + sub_path is not None and sub_path.endswith("/.overview.md") + ): + # Keep the source directory scope for the overview check; the + # requested and effective targets still name only the sidecar. + parent_path = sub_path.rsplit("/", 1)[0] if "/" in sub_path else None + scoped_includes.extend(plan.restrict_scope(parent_path).includes) + else: + scoped_includes.extend(restricted.includes) + effective.extend( + root + ("/" + item.path if item.path else "") for item in restricted.includes + ) + if not effective: + raise PermissionDeniedError("GLOBAL_ACCESS_DENIED: targets have no allowed content") + plan = PolicyScopePlan("allowed", tuple(dict.fromkeys(scoped_includes)), plan.excludes) + return replace( + ctx, + data_access=DataAccessScope( + owner, + AclAction.READ, + plan, + policy.policy_hash, + root, + canonical, + tuple(dict.fromkeys(effective)), + ), + ) + + def _access_decision(self, uri: str, ctx: RequestContext) -> AccessDecision: + parts = self._safe_uri_parts(uri) + if ctx.data_access is not None and parts[:1] == ["user"]: + return ctx.data_access.decide(uri) + return AccessDecision.ALLOWED if self._is_accessible(uri, ctx) else AccessDecision.DENIED + + def _is_traversable(self, uri: str, ctx: RequestContext) -> bool: + return self._access_decision(uri, ctx) is not AccessDecision.DENIED + + async def _access_decision_many( + self, uris: Sequence[str], ctx: Optional[RequestContext] + ) -> Dict[str, AccessDecision]: + real_ctx = self._ctx_or_default(ctx) + access = await self._can_access_many(uris, real_ctx) + return { + uri: ( + AccessDecision.ALLOWED + if access.get(uri, False) + else AccessDecision.CONTAINER + if real_ctx.data_access is not None + and self._access_decision(uri, real_ctx) is AccessDecision.CONTAINER + else AccessDecision.DENIED + ) + for uri in uris + } + + async def _ensure_traversable(self, uri: str, ctx: RequestContext) -> None: + if ctx.data_access is not None and self._safe_uri_parts(uri)[:1] == ["user"]: + if not self._is_traversable(uri, ctx): + raise PermissionDeniedError("GLOBAL_ACCESS_DENIED", resource=uri) + return + await self._ensure_access(uri, ctx) + @staticmethod def _system_permission_allows( uri: str, @@ -193,6 +311,8 @@ async def _can_access_many( if not isinstance(action, AclAction): raise TypeError(f"action must be AclAction, got {type(action).__name__}") real_ctx = self._ctx_or_default(ctx) + if real_ctx.data_access is not None and action is not AclAction.READ: + return dict.fromkeys(uris, False) result: Dict[str, bool] = {} valid: List[str] = [] for uri in dict.fromkeys(uris): @@ -242,6 +362,8 @@ async def _ensure_access( *, action: AclAction = AclAction.READ, ) -> None: + if action is AclAction.READ: + ctx = self.authorize_read(self._ctx_or_default(ctx), [uri]) await self._ensure_access_many([uri], ctx, action=action) async def _ensure_destination_access(self, uri: str, ctx: Optional[RequestContext]) -> bool: @@ -280,6 +402,8 @@ async def _ensure_access_many( action: AclAction, ) -> None: real_ctx = self._ctx_or_default(ctx) + if real_ctx.data_access is not None and action is not AclAction.READ: + raise PermissionDeniedError("Delegated GLOBAL access is read-only") access = await self._can_access_many(uris, real_ctx, action=action) denied = next((uri for uri in uris if not access.get(uri, False)), None) if denied is not None: @@ -344,6 +468,10 @@ async def _ensure_descendant_access( async def _ensure_retrieval_scope(self, uri: str, ctx: Optional[RequestContext]) -> None: self._safe_uri_parts(uri) + real_ctx = self._ctx_or_default(ctx) + if real_ctx.data_access is not None: + await self._ensure_traversable(uri, real_ctx) + return if await self._acl_enabled(ctx) and is_acl_uri(uri): return await self._ensure_access(uri, ctx) @@ -609,6 +737,10 @@ def _is_tree_entry_visible( entry_path = entry["path"] entry_info = entry.get("info", {}) name = entry_info.get("name") or entry_path.rstrip("/").rsplit("/", 1)[-1] + if ctx.data_access is not None and entry_info.get("isDir"): + uri = self._path_to_uri(entry_path, ctx=ctx) + if self._access_decision(uri, ctx) is AccessDecision.CONTAINER: + return not self._ancestor_is_filtered(entry_path, base_path) return self._is_path_entry_visible( entry_path, name, @@ -689,7 +821,18 @@ async def _iter_visible_tree_entries( ): break - if not acl_enabled: + if real_ctx.data_access is not None: + decisions = await self._access_decision_many([u for _, u in candidates], real_ctx) + visible = [] + for entry, entry_uri in candidates: + if decisions[entry_uri] is AccessDecision.ALLOWED: + visible.append((entry, entry_uri)) + elif ( + entry.get("info", {}).get("isDir") + and decisions[entry_uri] is AccessDecision.CONTAINER + ): + visible.append((dict(entry, access="container", extra={}), entry_uri)) + elif not acl_enabled: visible = candidates else: access = await self._can_access_many( @@ -877,6 +1020,8 @@ def _is_legacy_temp_uri_parts(self, parts: List[str]) -> bool: def _is_accessible(self, uri: str, ctx: RequestContext) -> bool: """Check whether a URI is visible/accessible under current request context.""" parts = self._safe_uri_parts(uri) + if ctx.data_access is not None and parts[:1] == ["user"]: + return ctx.data_access.decide(uri) is AccessDecision.ALLOWED if ctx.role == Role.ROOT: return True if is_hidden_by_actor_peer_view(uri, ctx): diff --git a/openviking/storage/viking_fs/_grep.py b/openviking/storage/viking_fs/_grep.py index 4a3f4d489f..1580abbade 100644 --- a/openviking/storage/viking_fs/_grep.py +++ b/openviking/storage/viking_fs/_grep.py @@ -103,6 +103,8 @@ async def grep( # stat() authorizes the requested path. # Skip vector_store.count() — the count field is not needed for grep, # and avoiding it saves one VikingDB API call. + ctx = self.authorize_read(self._ctx_or_default(ctx), [uri]) + await self._ensure_traversable(uri, ctx) await self.stat(uri, ctx=ctx, skip_count=True) # Read engine and threshold from grep_config (ov.conf) @@ -115,7 +117,10 @@ async def grep( # persisted raw content, so it cannot safely recall projected results. resolved_engine = ( "fs" - if force_fs or content_transform is not None or is_session_uri(uri) + if force_fs + or ctx.data_access is not None + or content_transform is not None + or is_session_uri(uri) else await self._resolve_grep_engine(engine, uri, ctx, switch_to_remote_threshold) ) tags_by_uri: Dict[str, List[str]] = {} @@ -326,7 +331,11 @@ async def _grep_fs( after_context=0, ): """Filesystem grep path: prefer native agfs grep and fall back if unavailable.""" - native_safe = content_transform is None and allowed_uris is None + native_safe = ( + content_transform is None + and allowed_uris is None + and self._ctx_or_default(ctx).data_access is None + ) if native_safe: try: # Session grep historically used the Python fallback, where @@ -385,7 +394,9 @@ async def _grep_vikingdb_then_fs( """VikingDB bm25 recall + local fs precise matching.""" vector_store = self._get_vector_store() tags_by_uri: Dict[str, List[str]] = {} - output_fields = ["uri", "search_tags"] if tag_filter is not None or include_tags else ["uri"] + output_fields = ( + ["uri", "search_tags"] if tag_filter is not None or include_tags else ["uri"] + ) # Split regex alternation (e.g. "error|warning|fail") and join as a # single query string for bm25 search. VikingDB's standard tokenizer @@ -394,7 +405,8 @@ async def _grep_vikingdb_then_fs( excluded_prefix = None if exclude_uri: excluded_prefix = exclude_uri.rstrip("/") - await self._ensure_access(excluded_prefix, ctx) + if self._ctx_or_default(ctx).data_access is None: + await self._ensure_access(excluded_prefix, ctx) filter_expr = _build_grep_scope_filter( uri, level_limit, @@ -635,7 +647,8 @@ async def _grep_with_agfs( excluded_path = None if exclude_uri: normalized_excluded_uri = exclude_uri.rstrip("/") - await self._ensure_access(normalized_excluded_uri, ctx) + if self._ctx_or_default(ctx).data_access is None: + await self._ensure_access(normalized_excluded_uri, ctx) excluded_path = self._uri_to_path(normalized_excluded_uri, ctx=ctx) try: @@ -756,7 +769,8 @@ async def _grep_encrypted( excluded_prefix = None if exclude_uri: excluded_prefix = exclude_uri.rstrip("/") - await self._ensure_access(excluded_prefix, ctx) + if self._ctx_or_default(ctx).data_access is None: + await self._ensure_access(excluded_prefix, ctx) file_uris = await self._collect_grep_files( uri, excluded_prefix=excluded_prefix, @@ -950,14 +964,12 @@ def _build_grep_match( if before_context > 0: start = max(0, line_index - before_context) match["before_context"] = [ - {"line": index + 1, "content": lines[index]} - for index in range(start, line_index) + {"line": index + 1, "content": lines[index]} for index in range(start, line_index) ] if after_context > 0: end = min(len(lines), line_index + after_context + 1) match["after_context"] = [ - {"line": index + 1, "content": lines[index]} - for index in range(line_index + 1, end) + {"line": index + 1, "content": lines[index]} for index in range(line_index + 1, end) ] return match diff --git a/openviking/storage/viking_fs/_ops.py b/openviking/storage/viking_fs/_ops.py index 105acc94b5..5f0967e367 100644 --- a/openviking/storage/viking_fs/_ops.py +++ b/openviking/storage/viking_fs/_ops.py @@ -24,7 +24,7 @@ ) from openviking.resource.watch_storage import is_watch_task_control_uri from openviking.server.error_mapping import is_not_found_error, map_exception -from openviking.server.identity import RequestContext, Role +from openviking.server.identity import AccessDecision, RequestContext, Role from openviking.storage.abstract_overview import ( ABSTRACT_OVERVIEW_FILENAMES, rewrite_abstract_overview_for_transfer, @@ -149,6 +149,7 @@ async def read( """Read file. Accepts a Viking URI or a 32-char hex vector record id.""" real_ctx = self._ctx_or_default(ctx) uri = await self.resolve_uri(uri, real_ctx) + ctx = self.authorize_read(real_ctx, [uri]) await self._ensure_access(uri, ctx) path = self._uri_to_path(uri, ctx=ctx) @@ -1174,7 +1175,9 @@ async def stat( """ real_ctx = self._ctx_or_default(ctx) uri = await self.resolve_uri(uri, real_ctx) - await self._ensure_access(uri, ctx) + ctx = self.authorize_read(real_ctx, [uri]) + real_ctx = ctx + await self._ensure_traversable(uri, ctx) path = self._uri_to_path(uri, ctx=ctx) try: result = await self._async_agfs.stat(path) @@ -1182,6 +1185,14 @@ async def stat( if is_not_found_error(exc): raise NotFoundError(uri, "file") from exc raise + if ( + ctx.data_access is not None + and self._access_decision(uri, ctx) is AccessDecision.CONTAINER + and not result.get("isDir", False) + ): + raise PermissionDeniedError( + "GLOBAL_ACCESS_DENIED: only directories are traversable", resource=uri + ) if isinstance(result, dict): result["uri"] = uri if include_lock_status: @@ -1215,6 +1226,11 @@ async def exists(self, uri: str, ctx: Optional[RequestContext] = None) -> bool: actor-peer, upload, and internal paths. """ real_ctx = self._ctx_or_default(ctx) + if real_ctx.role == Role.GLOBAL: + try: + real_ctx = self.authorize_read(real_ctx, [uri]) + except PermissionDeniedError: + return False self._safe_uri_parts(uri) if not self._is_accessible(uri, real_ctx): return False @@ -1238,7 +1254,8 @@ async def glob( (with name/uri/size/mode/mtime/isDir populated from stat) for CLI table rendering. """ _ensure_non_empty_search_query(pattern) - await self._ensure_access(uri, ctx) + ctx = self.authorize_read(self._ctx_or_default(ctx), [uri]) + await self._ensure_traversable(uri, ctx) real_ctx = self._ctx_or_default(ctx) return_entries = extra_fields is not None aug_fields = list(extra_fields) if extra_fields else [] @@ -1363,6 +1380,8 @@ async def _try_glob_vikingdb( remote_limit = self._remote_glob_limit(node_limit) filter_expr = self._remote_glob_filter(uri, pattern) + if ctx.data_access is not None: + filter_expr = And([filter_expr, ctx.data_access.query_filter()]) if tag_filter: filter_expr = And([filter_expr, RawDSL(tag_filter)]) full_pattern = _uri_to_remote_path_pattern(uri, pattern) @@ -1651,6 +1670,12 @@ async def _batch_fetch_directory_summaries( """Load one summary field for directories with a fixed-size worker pool.""" dir_jobs = [] for index, entry in enumerate(entries): + if entry.get("access") in {"denied", "container"} or ( + self._ctx_or_default(ctx).data_access is not None + and not self._is_accessible(entry["uri"], self._ctx_or_default(ctx)) + ): + entry[field] = "" + continue if not entry.get("isDir", False): entry[field] = "" continue @@ -1676,6 +1701,8 @@ async def worker() -> None: try: summary = await self._read_directory_summary(entry["uri"], field, ctx=ctx) + except PermissionDeniedError: + summary = "" except Exception: summary = fallback @@ -1760,7 +1787,9 @@ async def _finalize_listing_entries( } if recursive: item["rel_path"] = entry.get("rel_path", "") - if "tags" in entry: + if entry.get("access") == "container": + item["access"] = "container" + if "tags" in entry and entry.get("access") != "container": item["tags"] = entry["tags"] result.append(item) @@ -1809,7 +1838,8 @@ async def tree( """ if offset < 0: raise ValueError("offset must be non-negative") - await self._ensure_access(uri, ctx) + ctx = self.authorize_read(self._ctx_or_default(ctx), [uri]) + await self._ensure_traversable(uri, ctx) extra_fields = extra_fields or [] if output not in {"original", "agent"}: raise ValueError(f"Invalid output format: {output}") @@ -1885,6 +1915,8 @@ async def _tree_original( "uri": entry_uri, } ) + if entry.get("access") == "container": + new_entry["access"] = "container" result.append(new_entry) return result @@ -2063,6 +2095,7 @@ async def read_file( """ real_ctx = self._ctx_or_default(ctx) uri = await self.resolve_uri(uri, real_ctx) + ctx = self.authorize_read(real_ctx, [uri]) await self._ensure_access(uri, ctx) path = self._uri_to_path(uri, ctx=ctx) # Verify the file exists before reading, because AGFS read returns @@ -2111,6 +2144,7 @@ async def read_file_bytes( raise InvalidArgumentError("max_bytes must be non-negative") real_ctx = self._ctx_or_default(ctx) uri = await self.resolve_uri(uri, real_ctx) + ctx = self.authorize_read(real_ctx, [uri]) await self._ensure_access(uri, ctx) path = self._uri_to_path(uri, ctx=ctx) try: @@ -2276,7 +2310,8 @@ async def ls( output="agent" [{'name': '.abstract.md', 'size': 100, 'modTime': '2026-02-11T08:52:16.256Z', 'isDir': False, 'uri': 'viking://resources/.abstract.md', 'abstract': "..."}] """ - await self._ensure_access(uri, ctx) + ctx = self.authorize_read(self._ctx_or_default(ctx), [uri]) + await self._ensure_traversable(uri, ctx) extra_fields = extra_fields or [] if sort_by not in {None, "name", "mtime"}: raise ValueError("sort_by must be 'name' or 'mtime'") @@ -2406,10 +2441,22 @@ async def _ls_browsable_items( or not str(item[0].get("name", "")).startswith(".") or show_all_hidden ] - access = await self._can_access_many([entry_uri for _, entry_uri in entry_items], ctx) + decisions = await self._access_decision_many( + [entry_uri for _, entry_uri in entry_items], ctx + ) for entry, entry_uri in entry_items: - if access.get(entry_uri, False): + if decisions[entry_uri] is AccessDecision.ALLOWED: item = (entry, entry_uri) + elif entry.get("isDir") and decisions[entry_uri] is AccessDecision.CONTAINER: + item = ( + { + k: v + for k, v in entry.items() + if k in {"name", "size", "modTime", "mtime", "mode", "isDir"} + } + | {"access": "container"}, + entry_uri, + ) elif expose_resource_names: item = ( { @@ -2451,7 +2498,7 @@ async def _augment_entries_extra_fields( for i, entry in enumerate(entries): # ACL directory enumeration may expose only a name/type placeholder. # Do not enrich denied entries with metadata from inaccessible paths. - if entry.get("access") == "denied": + if entry.get("access") in {"denied", "container"}: continue entry_uri = entry.get("uri", "") is_dir = entry.get("isDir", False) diff --git a/openviking/storage/viking_fs/_semantic.py b/openviking/storage/viking_fs/_semantic.py index 6eb6200c04..cc6e7c5b33 100644 --- a/openviking/storage/viking_fs/_semantic.py +++ b/openviking/storage/viking_fs/_semantic.py @@ -10,7 +10,7 @@ from openviking.core.retrieval_targets import resolve_retrieval_targets from openviking.core.retrieval_types import SearchType from openviking.server.error_mapping import is_not_found_error, map_exception -from openviking.server.identity import RequestContext +from openviking.server.identity import RequestContext, Role from openviking.storage.abstract_overview import ( AbstractOverviewFormatError, body_for_preview, @@ -41,6 +41,11 @@ async def _read_directory_summary( self, uri: str, field: Literal["abstract", "overview"], ctx: Optional[RequestContext] = None ) -> str: """Read a fixed summary file from an already authorized, known directory.""" + real_ctx = self._ctx_or_default(ctx) + if real_ctx.role == Role.GLOBAL: + ctx = self.authorize_read(real_ctx, [uri]) + if ctx.data_access is not None: + await self._ensure_access(f"{uri.rstrip('/')}/.{field}.md", ctx) path = self._uri_to_path(uri, ctx=ctx) try: raw = self._handle_agfs_read(await self._async_agfs.read(f"{path}/.{field}.md")) @@ -67,6 +72,8 @@ async def abstract( If the caller points to a file, its parent directory is used instead so the endpoint remains usable for both file and directory URIs. """ + real_ctx = self._ctx_or_default(ctx) + ctx = self.authorize_read(real_ctx, [uri]) await self._ensure_access(uri, ctx) path = self._uri_to_path(uri, ctx=ctx) try: @@ -84,7 +91,7 @@ async def abstract( uri, parent_uri, ) - return await self.abstract(parent_uri, ctx=ctx) + return await self.abstract(parent_uri, ctx=real_ctx) return await self._read_directory_summary(uri, "abstract", ctx=ctx) async def overview( @@ -97,6 +104,8 @@ async def overview( If the caller points to a file, its parent directory is used instead so the endpoint remains usable for both file and directory URIs. """ + real_ctx = self._ctx_or_default(ctx) + ctx = self.authorize_read(real_ctx, [uri]) await self._ensure_access(uri, ctx) path = self._uri_to_path(uri, ctx=ctx) try: @@ -114,7 +123,7 @@ async def overview( uri, parent_uri, ) - return await self.overview(parent_uri, ctx=ctx) + return await self.overview(parent_uri, ctx=real_ctx) return await self._read_directory_summary(uri, "overview", ctx=ctx) async def find( @@ -181,6 +190,14 @@ async def find( real_ctx = self._require_request_context(ctx) retrieval_targets = resolve_retrieval_targets(target_uri, real_ctx) + real_ctx = self.authorize_read(real_ctx, retrieval_targets.target_directories) + if real_ctx.data_access is not None: + from dataclasses import replace + + retrieval_targets = replace( + retrieval_targets, target_directories=list(real_ctx.data_access.effective_targets) + ) + if filter_only: return await self._find_by_filter( filter=filter, @@ -239,7 +256,20 @@ async def find( # Convert QueryResult to FindResult memories, resources, skills = [], [], [] + access = ( + await self._can_access_many([hit.uri for hit in result.matched_contexts], real_ctx) + if real_ctx.data_access is not None + else None + ) for ctx in result.matched_contexts: + if access is not None and not access.get(ctx.uri, False): + continue + if ( + ctx.level == ContextLevel.OVERVIEW + and real_ctx.data_access is not None + and not real_ctx.data_access.allows_overview(ctx.uri) + ): + continue if ctx.context_type == ContextType.MEMORY: memories.append(ctx) elif ctx.context_type == ContextType.RESOURCE: @@ -299,10 +329,23 @@ async def _find_by_filter( # and L1 record and would otherwise be returned twice — inflating the # total and eating two of the caller's limit slots for one result. seen_uris: set = set() + access = ( + await self._can_access_many([str(r.get("uri", "")) for r in records], ctx) + if ctx.data_access is not None + else None + ) for record in records: + if access is not None and not access.get(str(record.get("uri", "")), False): + continue matched = build_matched_context_from_record(record) if matched is None or matched.uri in seen_uris: continue + if ( + matched.level == ContextLevel.OVERVIEW + and ctx.data_access is not None + and not ctx.data_access.allows_overview(matched.uri) + ): + continue seen_uris.add(matched.uri) if matched.context_type == ContextType.MEMORY: memories.append(matched) @@ -480,6 +523,12 @@ async def _execute(tq: TypedQuery): memories, resources, skills = [], [], [] for result in query_results: for ctx in result.matched_contexts: + if ( + ctx.level == ContextLevel.OVERVIEW + and real_ctx.data_access is not None + and not real_ctx.data_access.allows_overview(ctx.uri) + ): + continue if ctx.context_type == ContextType.MEMORY: memories.append(ctx) elif ctx.context_type == ContextType.RESOURCE: diff --git a/openviking/storage/viking_vector_index_backend.py b/openviking/storage/viking_vector_index_backend.py index 16fe6efbf0..f33f4dcf24 100644 --- a/openviking/storage/viking_vector_index_backend.py +++ b/openviking/storage/viking_vector_index_backend.py @@ -3164,7 +3164,12 @@ def _build_scope_filter( ) -> Optional[FilterExpr]: targets = [target_dir for target_dir in target_directories or [] if target_dir] tenant_filter = self._tenant_filter(ctx, acl_enabled=acl_enabled) - if tenant_filter and not acl_enabled and self._targets_within_visible_roots(ctx, targets): + if ( + tenant_filter + and not acl_enabled + and ctx.data_access is None + and self._targets_within_visible_roots(ctx, targets) + ): # The target scopes are already narrower than the tenant-visible # roots. Keep account isolation, but avoid recursively evaluating # the broader path scopes as an additional filter. @@ -3205,6 +3210,8 @@ def _tenant_filter( return None account_filter = Eq("account_id", ctx.account_id) + if ctx.data_access is not None: + account_filter = And([account_filter, ctx.data_access.query_filter()]) if not acl_enabled: return And( [