feat(auth): scope GLOBAL reads to owner policy

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

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])

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:

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)

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,