mirror of
https://github.com/HKUDS/OpenSpace.git
synced 2026-08-28 05:15:00 +00:00
1066 lines
39 KiB
Python
1066 lines
39 KiB
Python
"""Permission settings loader.
|
|
|
|
Key persistence behavior:
|
|
|
|
* "Always allow" UI updates default to ``localSettings`` at
|
|
``<cwd>/.openspace/settings.local.json``.
|
|
* ``session`` / ``cliArg`` — runtime-only stores (process-lifetime singletons,
|
|
never persisted). See :class:`_BehaviorRuleStore`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple
|
|
|
|
from openspace.utils.logging import Logger
|
|
from openspace.services.runtime_support.settings import (
|
|
get_settings_for_source as _settings_get_settings_for_source,
|
|
get_settings_path_for_source as _settings_get_settings_path_for_source,
|
|
save_settings_for_source as _settings_save_settings_for_source,
|
|
)
|
|
|
|
from .types import (
|
|
EXTERNAL_PERMISSION_MODES,
|
|
PERMISSION_RULE_SOURCES,
|
|
AddDirectoriesUpdate,
|
|
AdditionalWorkingDirectory,
|
|
AddRulesUpdate,
|
|
ExternalPermissionMode,
|
|
PermissionBehavior,
|
|
PermissionMode,
|
|
PermissionRule,
|
|
PermissionRuleSource,
|
|
PermissionRuleValue,
|
|
PermissionUpdate,
|
|
PermissionUpdateDestination,
|
|
RemoveDirectoriesUpdate,
|
|
RemoveRulesUpdate,
|
|
ReplaceRulesUpdate,
|
|
SetModeUpdate,
|
|
ToolPermissionContext,
|
|
ToolPermissionRulesBySource,
|
|
format_rule_value,
|
|
parse_rule_value,
|
|
)
|
|
|
|
logger = Logger.get_logger(__name__)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §1 Constants
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
_SUPPORTED_RULE_BEHAVIORS: Tuple[PermissionBehavior, ...] = ("allow", "deny", "ask")
|
|
|
|
_EDITABLE_SOURCES: Tuple[PermissionUpdateDestination, ...] = (
|
|
"userSettings",
|
|
"projectSettings",
|
|
"localSettings",
|
|
)
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §2 Settings file paths
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _settings_path_for_source(
|
|
source: PermissionRuleSource, cwd: str
|
|
) -> Optional[Path]:
|
|
"""Return the on-disk path for a source, or None if the source is
|
|
runtime-only (``session`` / ``cliArg``).
|
|
|
|
Implementation: ``getSettingsFilePathForSource`` in settings.ts.
|
|
"""
|
|
if source in ("userSettings", "projectSettings", "localSettings"):
|
|
return _settings_get_settings_path_for_source(source, cwd)
|
|
# session / cliArg: runtime stores, handled separately.
|
|
return None
|
|
|
|
|
|
def get_settings_for_source(
|
|
source: PermissionRuleSource, cwd: str
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""OpenSpace ``getSettingsForSource`` — return parsed JSON for a source.
|
|
|
|
Returns ``None`` when:
|
|
* the source is runtime-only (``session`` / ``cliArg``);
|
|
* the source has no on-disk path;
|
|
* the file does not exist;
|
|
* the file is unreadable or contains invalid JSON (a warning is
|
|
logged but no exception propagates — parity with OpenSpace's lenient
|
|
loader used for editing).
|
|
|
|
An empty file returns ``{}`` (same as OpenSpace).
|
|
"""
|
|
if source in ("session", "cliArg"):
|
|
return None
|
|
|
|
try:
|
|
return _settings_get_settings_for_source(source, cwd)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Failed to read settings for source %s at %s: %s",
|
|
source,
|
|
_settings_path_for_source(source, cwd),
|
|
exc,
|
|
)
|
|
return None
|
|
|
|
|
|
def save_settings_for_source(
|
|
source: PermissionRuleSource, cwd: str, settings: Mapping[str, Any]
|
|
) -> None:
|
|
"""OpenSpace ``updateSettingsForSource`` — atomically write ``settings`` JSON.
|
|
|
|
Raises :class:`ValueError` when ``source`` is not writable
|
|
(runtime stores). Parent
|
|
directory is created on demand. Write is atomic (temp-file + rename)
|
|
to survive process crashes mid-write.
|
|
"""
|
|
if source in ("session", "cliArg"):
|
|
raise ValueError(
|
|
f"source {source!r} is a runtime store; use the store APIs instead"
|
|
)
|
|
if source not in _EDITABLE_SOURCES:
|
|
raise ValueError(f"source {source!r} is not editable in OpenSpace")
|
|
|
|
path = _settings_path_for_source(source, cwd)
|
|
if path is None:
|
|
raise ValueError(f"no settings path defined for source {source!r}")
|
|
|
|
_settings_save_settings_for_source(source, dict(settings), cwd=cwd)
|
|
|
|
|
|
def _atomic_write_json(path: Path, data: Mapping[str, Any]) -> None:
|
|
"""Write ``data`` as JSON to ``path`` via a tempfile + rename.
|
|
|
|
The tempfile is created in the same directory as ``path`` so the
|
|
rename is atomic on POSIX; if the write fails, the tempfile is
|
|
cleaned up.
|
|
"""
|
|
parent = path.parent
|
|
parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
fd, tmp_name = tempfile.mkstemp(
|
|
prefix=".settings-",
|
|
suffix=".json.tmp",
|
|
dir=str(parent),
|
|
)
|
|
tmp_path = Path(tmp_name)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
|
json.dump(data, fh, indent=2, ensure_ascii=False, sort_keys=False)
|
|
fh.write("\n")
|
|
os.replace(tmp_path, path)
|
|
except Exception:
|
|
try:
|
|
tmp_path.unlink()
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §3 Runtime stores (session / cliArg)
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class _BehaviorRuleStore:
|
|
"""Process-level singleton for runtime-only permission sources.
|
|
|
|
Used by the ``session`` and ``cliArg`` rule sources, which OpenSpace keeps
|
|
in memory only — they never touch the disk. Thread-safe via a
|
|
single coarse-grained lock (permissions are updated at most a few
|
|
times per turn, so contention is negligible).
|
|
|
|
Stores three things:
|
|
|
|
* per-behavior rule list (``allow`` / ``deny`` / ``ask``);
|
|
* additional working directories (paths added via ``addDirectories``
|
|
updates with this destination);
|
|
* a default-mode override (used when cli flag ``--permission-mode``
|
|
is passed — OpenSpace maps the flag to a ``session``-scoped ``setMode``
|
|
update).
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self._rules: Dict[PermissionBehavior, List[PermissionRuleValue]] = {
|
|
"allow": [],
|
|
"deny": [],
|
|
"ask": [],
|
|
}
|
|
self._directories: List[str] = []
|
|
self._mode: Optional[ExternalPermissionMode] = None
|
|
|
|
def add_rule(
|
|
self, behavior: PermissionBehavior, rule: PermissionRuleValue
|
|
) -> None:
|
|
with self._lock:
|
|
existing = {format_rule_value(r) for r in self._rules[behavior]}
|
|
if format_rule_value(rule) not in existing:
|
|
self._rules[behavior].append(rule)
|
|
|
|
def remove_rule(
|
|
self, behavior: PermissionBehavior, rule: PermissionRuleValue
|
|
) -> None:
|
|
target = format_rule_value(rule)
|
|
with self._lock:
|
|
self._rules[behavior] = [
|
|
r for r in self._rules[behavior] if format_rule_value(r) != target
|
|
]
|
|
|
|
def replace_rules(
|
|
self,
|
|
behavior: PermissionBehavior,
|
|
rules: Iterable[PermissionRuleValue],
|
|
) -> None:
|
|
deduped: List[PermissionRuleValue] = []
|
|
seen: set = set()
|
|
for r in rules:
|
|
s = format_rule_value(r)
|
|
if s not in seen:
|
|
seen.add(s)
|
|
deduped.append(r)
|
|
with self._lock:
|
|
self._rules[behavior] = deduped
|
|
|
|
def get_rules(
|
|
self, behavior: PermissionBehavior
|
|
) -> List[PermissionRuleValue]:
|
|
with self._lock:
|
|
return list(self._rules[behavior])
|
|
|
|
def add_directory(self, path: str) -> None:
|
|
with self._lock:
|
|
if path not in self._directories:
|
|
self._directories.append(path)
|
|
|
|
def remove_directory(self, path: str) -> None:
|
|
with self._lock:
|
|
self._directories = [d for d in self._directories if d != path]
|
|
|
|
def get_directories(self) -> List[str]:
|
|
with self._lock:
|
|
return list(self._directories)
|
|
|
|
def set_mode(self, mode: Optional[ExternalPermissionMode]) -> None:
|
|
with self._lock:
|
|
self._mode = mode
|
|
|
|
def get_mode(self) -> Optional[ExternalPermissionMode]:
|
|
with self._lock:
|
|
return self._mode
|
|
|
|
def clear(self) -> None:
|
|
with self._lock:
|
|
self._rules = {"allow": [], "deny": [], "ask": []}
|
|
self._directories = []
|
|
self._mode = None
|
|
|
|
|
|
_SESSION_STORE = _BehaviorRuleStore()
|
|
_CLIARG_STORE = _BehaviorRuleStore()
|
|
|
|
|
|
def _store_for_source(
|
|
source: PermissionRuleSource,
|
|
) -> Optional[_BehaviorRuleStore]:
|
|
if source == "session":
|
|
return _SESSION_STORE
|
|
if source == "cliArg":
|
|
return _CLIARG_STORE
|
|
return None
|
|
|
|
|
|
def get_session_store() -> _BehaviorRuleStore:
|
|
"""Return the live session rule store (for tests / TUI inspection)."""
|
|
return _SESSION_STORE
|
|
|
|
|
|
def get_cliarg_store() -> _BehaviorRuleStore:
|
|
"""Return the live cliArg rule store."""
|
|
return _CLIARG_STORE
|
|
|
|
|
|
def reset_runtime_stores() -> None:
|
|
"""Clear both runtime stores — used by tests for isolation."""
|
|
_SESSION_STORE.clear()
|
|
_CLIARG_STORE.clear()
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §4 Parse helpers (OpenSpace settingsJsonToRules + PermissionUpdate roundtrip)
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _normalize_rule_string(raw: str) -> Optional[str]:
|
|
"""Roundtrip parse→format so legacy names (``Bash(npm:*)``) compare
|
|
equal to their canonical OS form (``bash(npm:*)``).
|
|
|
|
Returns ``None`` when the string is malformed. Callers drop ``None``
|
|
entries (OpenSpace does the same via ``settingsJsonToRules`` silently
|
|
skipping invalid rules).
|
|
"""
|
|
try:
|
|
return format_rule_value(parse_rule_value(raw))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _rules_from_settings(
|
|
settings: Optional[Mapping[str, Any]], source: PermissionRuleSource
|
|
) -> List[PermissionRule]:
|
|
"""OpenSpace ``settingsJsonToRules`` — permissionsLoader.ts L91-L114.
|
|
|
|
Extracts allow/deny/ask arrays from the ``permissions`` object and
|
|
turns each string into a :class:`PermissionRule` tagged with the
|
|
given ``source``. Malformed strings are logged-and-skipped (OpenSpace
|
|
also silently skips them).
|
|
"""
|
|
if not settings:
|
|
return []
|
|
perms = settings.get("permissions")
|
|
if not isinstance(perms, dict):
|
|
return []
|
|
|
|
rules: List[PermissionRule] = []
|
|
for behavior in _SUPPORTED_RULE_BEHAVIORS:
|
|
arr = perms.get(behavior)
|
|
if not isinstance(arr, list):
|
|
continue
|
|
for raw in arr:
|
|
if not isinstance(raw, str):
|
|
continue
|
|
try:
|
|
value = parse_rule_value(raw)
|
|
except ValueError as exc:
|
|
logger.warning(
|
|
"Skipping malformed rule %r in source %s: %s",
|
|
raw,
|
|
source,
|
|
exc,
|
|
)
|
|
continue
|
|
rules.append(
|
|
PermissionRule(
|
|
source=source,
|
|
rule_behavior=behavior,
|
|
rule_value=value,
|
|
)
|
|
)
|
|
return rules
|
|
|
|
|
|
def _additional_dirs_from_settings(
|
|
settings: Optional[Mapping[str, Any]],
|
|
) -> List[str]:
|
|
"""Read ``permissions.additionalDirectories`` (OpenSpace's canonical place).
|
|
|
|
For backward-compat with OS docs that show it at top-level, we also
|
|
check top-level ``additionalDirectories`` if the nested one is
|
|
missing. Only string entries are kept.
|
|
"""
|
|
if not settings:
|
|
return []
|
|
perms = settings.get("permissions")
|
|
if isinstance(perms, dict):
|
|
nested = perms.get("additionalDirectories")
|
|
if isinstance(nested, list):
|
|
return [d for d in nested if isinstance(d, str)]
|
|
top_level = settings.get("additionalDirectories")
|
|
if isinstance(top_level, list):
|
|
return [d for d in top_level if isinstance(d, str)]
|
|
return []
|
|
|
|
|
|
def _default_mode_from_settings(
|
|
settings: Optional[Mapping[str, Any]],
|
|
) -> Optional[ExternalPermissionMode]:
|
|
"""Read ``permissions.defaultMode`` (OpenSpace canonical) with top-level
|
|
fallback. Invalid enum values are ignored.
|
|
"""
|
|
if not settings:
|
|
return None
|
|
perms = settings.get("permissions")
|
|
if isinstance(perms, dict):
|
|
mode = perms.get("defaultMode")
|
|
if isinstance(mode, str) and mode in EXTERNAL_PERMISSION_MODES:
|
|
return mode # type: ignore[return-value]
|
|
mode_top = settings.get("defaultMode")
|
|
if isinstance(mode_top, str) and mode_top in EXTERNAL_PERMISSION_MODES:
|
|
return mode_top # type: ignore[return-value]
|
|
return None
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §5 load_permissions_from_source / load_all_permission_rules_from_disk
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _store_as_rules(
|
|
store: _BehaviorRuleStore, source: PermissionRuleSource
|
|
) -> List[PermissionRule]:
|
|
out: List[PermissionRule] = []
|
|
for behavior in _SUPPORTED_RULE_BEHAVIORS:
|
|
for value in store.get_rules(behavior):
|
|
out.append(
|
|
PermissionRule(
|
|
source=source,
|
|
rule_behavior=behavior,
|
|
rule_value=value,
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def load_permissions_from_source(
|
|
source: PermissionRuleSource, cwd: str
|
|
) -> List[PermissionRule]:
|
|
"""OpenSpace ``getPermissionRulesForSource`` — permissionsLoader.ts L140-L145.
|
|
|
|
Load rules from a *single* source. Disk sources read via
|
|
:func:`get_settings_for_source`; runtime stores (``session`` /
|
|
``cliArg``) read from the process singletons.
|
|
"""
|
|
store = _store_for_source(source)
|
|
if store is not None:
|
|
return _store_as_rules(store, source)
|
|
settings = get_settings_for_source(source, cwd)
|
|
return _rules_from_settings(settings, source)
|
|
|
|
|
|
def load_all_permission_rules_from_disk(
|
|
cwd: str,
|
|
) -> Tuple[
|
|
ToolPermissionRulesBySource,
|
|
ToolPermissionRulesBySource,
|
|
ToolPermissionRulesBySource,
|
|
]:
|
|
"""OpenSpace ``loadAllPermissionRulesFromDisk`` — permissionsLoader.ts L120-L133.
|
|
|
|
Returns a triple ``(allow, deny, ask)`` of source-keyed dicts. Each
|
|
dict maps :data:`PermissionRuleSource` → tuple of formatted rule
|
|
strings (canonical ``tool(content)`` form). A source whose
|
|
behavior bucket is empty is omitted from that dict.
|
|
|
|
OpenSpace iterates :data:`PERMISSION_RULE_SOURCES`, which contains all
|
|
sources in priority order.
|
|
"""
|
|
allow: Dict[PermissionRuleSource, Tuple[str, ...]] = {}
|
|
deny: Dict[PermissionRuleSource, Tuple[str, ...]] = {}
|
|
ask: Dict[PermissionRuleSource, Tuple[str, ...]] = {}
|
|
|
|
for source in PERMISSION_RULE_SOURCES:
|
|
rules = load_permissions_from_source(source, cwd)
|
|
if not rules:
|
|
continue
|
|
|
|
bucketed: Dict[PermissionBehavior, List[str]] = {
|
|
"allow": [],
|
|
"deny": [],
|
|
"ask": [],
|
|
}
|
|
for rule in rules:
|
|
bucketed[rule.rule_behavior].append(format_rule_value(rule.rule_value))
|
|
|
|
if bucketed["allow"]:
|
|
allow[source] = tuple(bucketed["allow"])
|
|
if bucketed["deny"]:
|
|
deny[source] = tuple(bucketed["deny"])
|
|
if bucketed["ask"]:
|
|
ask[source] = tuple(bucketed["ask"])
|
|
|
|
return allow, deny, ask
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §6 add / delete rules in settings files
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def add_permission_rules_to_settings(
|
|
destination: PermissionUpdateDestination,
|
|
rules: Iterable[PermissionRuleValue],
|
|
behavior: PermissionBehavior,
|
|
cwd: str,
|
|
) -> bool:
|
|
"""OpenSpace ``addPermissionRulesToSettings`` — permissionsLoader.ts L229-L296.
|
|
|
|
Appends ``rules`` (deduped via normalized roundtrip) to the
|
|
destination's rule list for the given ``behavior``. Behavior by
|
|
destination:
|
|
|
|
* ``session`` / ``cliArg`` — append to runtime store (in-memory only);
|
|
* ``userSettings`` / ``projectSettings`` / ``localSettings`` —
|
|
read current settings JSON, dedupe, write back atomically;
|
|
* any other destination — returns ``False`` with a warning.
|
|
|
|
Existing JSON keys are preserved (parity with OpenSpace's spread-copy of
|
|
``settingsData``).
|
|
"""
|
|
rule_list = list(rules)
|
|
if not rule_list:
|
|
return True # OpenSpace returns true for no-op
|
|
|
|
if destination == "session":
|
|
for rule in rule_list:
|
|
_SESSION_STORE.add_rule(behavior, rule)
|
|
return True
|
|
if destination == "cliArg":
|
|
for rule in rule_list:
|
|
_CLIARG_STORE.add_rule(behavior, rule)
|
|
return True
|
|
|
|
if destination not in _EDITABLE_SOURCES:
|
|
logger.warning(
|
|
"Cannot add permission rules to non-editable source %r",
|
|
destination,
|
|
)
|
|
return False
|
|
|
|
settings = get_settings_for_source(destination, cwd)
|
|
if settings is None:
|
|
settings = {}
|
|
|
|
permissions_raw = settings.get("permissions")
|
|
permissions = dict(permissions_raw) if isinstance(permissions_raw, dict) else {}
|
|
|
|
existing_raw = permissions.get(behavior)
|
|
existing_list: List[Any] = list(existing_raw) if isinstance(existing_raw, list) else []
|
|
|
|
# Build the set of already-present canonical rule strings so we
|
|
# don't re-append duplicates (OpenSpace does the same via Set + normalize).
|
|
existing_canonical: set = set()
|
|
for raw in existing_list:
|
|
if isinstance(raw, str):
|
|
canon = _normalize_rule_string(raw)
|
|
if canon is not None:
|
|
existing_canonical.add(canon)
|
|
|
|
additions: List[str] = []
|
|
for rule in rule_list:
|
|
s = format_rule_value(rule)
|
|
if s not in existing_canonical:
|
|
existing_canonical.add(s)
|
|
additions.append(s)
|
|
|
|
if not additions:
|
|
return True # nothing to do after dedupe
|
|
|
|
permissions[behavior] = existing_list + additions
|
|
updated_settings = dict(settings)
|
|
updated_settings["permissions"] = permissions
|
|
|
|
try:
|
|
save_settings_for_source(destination, cwd, updated_settings)
|
|
return True
|
|
except Exception as exc: # pragma: no cover — exercised in error paths
|
|
logger.warning(
|
|
"Failed to persist permission rules to %s: %s", destination, exc
|
|
)
|
|
return False
|
|
|
|
|
|
def delete_permission_rule_from_settings(
|
|
destination: PermissionUpdateDestination,
|
|
rule: PermissionRuleValue,
|
|
behavior: PermissionBehavior,
|
|
cwd: str,
|
|
) -> bool:
|
|
"""OpenSpace ``deletePermissionRuleFromSettings`` — permissionsLoader.ts L163-L216.
|
|
|
|
Deletes a single rule from the destination's bucket for ``behavior``.
|
|
Returns ``True`` iff a rule was actually removed (OpenSpace semantics: no-op
|
|
when the rule isn't there). Legacy names match their canonical
|
|
form thanks to the parse→format normalisation.
|
|
"""
|
|
target = format_rule_value(rule)
|
|
|
|
if destination == "session":
|
|
existing = [format_rule_value(r) for r in _SESSION_STORE.get_rules(behavior)]
|
|
if target not in existing:
|
|
return False
|
|
_SESSION_STORE.remove_rule(behavior, rule)
|
|
return True
|
|
if destination == "cliArg":
|
|
existing = [format_rule_value(r) for r in _CLIARG_STORE.get_rules(behavior)]
|
|
if target not in existing:
|
|
return False
|
|
_CLIARG_STORE.remove_rule(behavior, rule)
|
|
return True
|
|
|
|
if destination not in _EDITABLE_SOURCES:
|
|
return False
|
|
|
|
settings = get_settings_for_source(destination, cwd)
|
|
if not settings:
|
|
return False
|
|
|
|
permissions_raw = settings.get("permissions")
|
|
if not isinstance(permissions_raw, dict):
|
|
return False
|
|
behavior_raw = permissions_raw.get(behavior)
|
|
if not isinstance(behavior_raw, list):
|
|
return False
|
|
|
|
new_list: List[Any] = []
|
|
removed = False
|
|
for raw in behavior_raw:
|
|
if isinstance(raw, str):
|
|
canon = _normalize_rule_string(raw)
|
|
if canon == target:
|
|
removed = True
|
|
continue
|
|
new_list.append(raw)
|
|
|
|
if not removed:
|
|
return False
|
|
|
|
permissions = dict(permissions_raw)
|
|
permissions[behavior] = new_list
|
|
updated_settings = dict(settings)
|
|
updated_settings["permissions"] = permissions
|
|
|
|
try:
|
|
save_settings_for_source(destination, cwd, updated_settings)
|
|
return True
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning(
|
|
"Failed to delete permission rule from %s: %s", destination, exc
|
|
)
|
|
return False
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §7 apply_permission_update — context mutation (no disk)
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _bucket_name_for_behavior(behavior: PermissionBehavior) -> str:
|
|
"""Map ``allow``/``deny``/``ask`` → ``always_*_rules`` attribute."""
|
|
if behavior == "allow":
|
|
return "always_allow_rules"
|
|
if behavior == "deny":
|
|
return "always_deny_rules"
|
|
return "always_ask_rules"
|
|
|
|
|
|
def _clone_context_with(
|
|
context: ToolPermissionContext, **overrides: Any
|
|
) -> ToolPermissionContext:
|
|
"""Return a shallow copy of ``context`` with the given fields replaced.
|
|
|
|
Needed because :class:`ToolPermissionContext` is a frozen dataclass
|
|
— every mutation produces a new instance (OpenSpace does the same via the
|
|
``{...context, field: new}`` spread pattern).
|
|
"""
|
|
fields = dict(
|
|
mode=context.mode,
|
|
additional_working_directories=context.additional_working_directories,
|
|
always_allow_rules=context.always_allow_rules,
|
|
always_deny_rules=context.always_deny_rules,
|
|
always_ask_rules=context.always_ask_rules,
|
|
is_bypass_permissions_mode_available=context.is_bypass_permissions_mode_available,
|
|
stripped_dangerous_rules=context.stripped_dangerous_rules,
|
|
should_avoid_permission_prompts=context.should_avoid_permission_prompts,
|
|
await_automated_checks_before_dialog=context.await_automated_checks_before_dialog,
|
|
pre_plan_mode=context.pre_plan_mode,
|
|
)
|
|
fields.update(overrides)
|
|
return ToolPermissionContext(**fields)
|
|
|
|
|
|
def apply_permission_update(
|
|
update: PermissionUpdate,
|
|
cwd: str,
|
|
context: ToolPermissionContext,
|
|
) -> ToolPermissionContext:
|
|
"""OpenSpace ``applyPermissionUpdate`` — PermissionUpdate.ts L55-L188.
|
|
|
|
Apply a single :class:`PermissionUpdate` to the context and return a
|
|
new immutable context. No disk I/O (persistence is handled by
|
|
:func:`persist_permission_updates`). ``cwd`` is currently unused —
|
|
it is accepted for API symmetry with the persist family so callers
|
|
can pass the same signature through both stages.
|
|
"""
|
|
del cwd # unused; kept for interface parity with the persist family
|
|
|
|
if isinstance(update, SetModeUpdate):
|
|
return _clone_context_with(context, mode=update.mode)
|
|
|
|
if isinstance(update, AddRulesUpdate):
|
|
bucket_name = _bucket_name_for_behavior(update.behavior)
|
|
bucket: ToolPermissionRulesBySource = getattr(context, bucket_name)
|
|
rule_strings = tuple(format_rule_value(r) for r in update.rules)
|
|
existing = bucket.get(update.destination, ())
|
|
new_bucket = dict(bucket)
|
|
new_bucket[update.destination] = tuple(existing) + rule_strings
|
|
return _clone_context_with(context, **{bucket_name: new_bucket})
|
|
|
|
if isinstance(update, ReplaceRulesUpdate):
|
|
bucket_name = _bucket_name_for_behavior(update.behavior)
|
|
bucket = getattr(context, bucket_name)
|
|
rule_strings = tuple(format_rule_value(r) for r in update.rules)
|
|
new_bucket = dict(bucket)
|
|
new_bucket[update.destination] = rule_strings
|
|
return _clone_context_with(context, **{bucket_name: new_bucket})
|
|
|
|
if isinstance(update, RemoveRulesUpdate):
|
|
bucket_name = _bucket_name_for_behavior(update.behavior)
|
|
bucket = getattr(context, bucket_name)
|
|
to_remove = {format_rule_value(r) for r in update.rules}
|
|
existing = bucket.get(update.destination, ())
|
|
filtered = tuple(s for s in existing if s not in to_remove)
|
|
new_bucket = dict(bucket)
|
|
new_bucket[update.destination] = filtered
|
|
return _clone_context_with(context, **{bucket_name: new_bucket})
|
|
|
|
if isinstance(update, AddDirectoriesUpdate):
|
|
new_dirs: Dict[str, AdditionalWorkingDirectory] = dict(
|
|
context.additional_working_directories
|
|
)
|
|
for directory in update.directories:
|
|
new_dirs[directory] = AdditionalWorkingDirectory(
|
|
path=directory,
|
|
source=update.destination,
|
|
)
|
|
return _clone_context_with(context, additional_working_directories=new_dirs)
|
|
|
|
if isinstance(update, RemoveDirectoriesUpdate):
|
|
new_dirs = dict(context.additional_working_directories)
|
|
for directory in update.directories:
|
|
new_dirs.pop(directory, None)
|
|
return _clone_context_with(context, additional_working_directories=new_dirs)
|
|
|
|
logger.warning("Unknown permission update type: %r", type(update).__name__)
|
|
return context
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §8 persist_permission_updates — disk writes
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _persist_add_rules(update: AddRulesUpdate, cwd: str) -> None:
|
|
add_permission_rules_to_settings(
|
|
update.destination,
|
|
list(update.rules),
|
|
update.behavior,
|
|
cwd,
|
|
)
|
|
|
|
|
|
def _persist_replace_rules(update: ReplaceRulesUpdate, cwd: str) -> None:
|
|
dest = update.destination
|
|
if dest in ("session", "cliArg"):
|
|
store = _store_for_source(dest)
|
|
if store is not None:
|
|
store.replace_rules(update.behavior, list(update.rules))
|
|
return
|
|
if dest not in _EDITABLE_SOURCES:
|
|
return
|
|
|
|
settings = get_settings_for_source(dest, cwd) or {}
|
|
permissions_raw = settings.get("permissions")
|
|
permissions = dict(permissions_raw) if isinstance(permissions_raw, dict) else {}
|
|
permissions[update.behavior] = [format_rule_value(r) for r in update.rules]
|
|
|
|
updated = dict(settings)
|
|
updated["permissions"] = permissions
|
|
try:
|
|
save_settings_for_source(dest, cwd, updated)
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning(
|
|
"Failed to replace %s rules in %s: %s",
|
|
update.behavior,
|
|
dest,
|
|
exc,
|
|
)
|
|
|
|
|
|
def _persist_remove_rules(update: RemoveRulesUpdate, cwd: str) -> None:
|
|
dest = update.destination
|
|
if dest in ("session", "cliArg"):
|
|
store = _store_for_source(dest)
|
|
if store is not None:
|
|
for rule in update.rules:
|
|
store.remove_rule(update.behavior, rule)
|
|
return
|
|
if dest not in _EDITABLE_SOURCES:
|
|
return
|
|
|
|
settings = get_settings_for_source(dest, cwd)
|
|
if not settings:
|
|
return
|
|
permissions_raw = settings.get("permissions")
|
|
if not isinstance(permissions_raw, dict):
|
|
return
|
|
behavior_raw = permissions_raw.get(update.behavior)
|
|
if not isinstance(behavior_raw, list):
|
|
return
|
|
|
|
to_remove = {format_rule_value(r) for r in update.rules}
|
|
filtered: List[Any] = []
|
|
for raw in behavior_raw:
|
|
if isinstance(raw, str):
|
|
canon = _normalize_rule_string(raw)
|
|
if canon in to_remove:
|
|
continue
|
|
filtered.append(raw)
|
|
|
|
permissions = dict(permissions_raw)
|
|
permissions[update.behavior] = filtered
|
|
updated = dict(settings)
|
|
updated["permissions"] = permissions
|
|
try:
|
|
save_settings_for_source(dest, cwd, updated)
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning(
|
|
"Failed to remove %s rules from %s: %s",
|
|
update.behavior,
|
|
dest,
|
|
exc,
|
|
)
|
|
|
|
|
|
def _persist_add_directories(update: AddDirectoriesUpdate, cwd: str) -> None:
|
|
dest = update.destination
|
|
if dest in ("session", "cliArg"):
|
|
store = _store_for_source(dest)
|
|
if store is not None:
|
|
for directory in update.directories:
|
|
store.add_directory(directory)
|
|
return
|
|
if dest not in _EDITABLE_SOURCES:
|
|
return
|
|
|
|
settings = get_settings_for_source(dest, cwd) or {}
|
|
permissions_raw = settings.get("permissions")
|
|
permissions = dict(permissions_raw) if isinstance(permissions_raw, dict) else {}
|
|
existing_raw = permissions.get("additionalDirectories")
|
|
existing_list: List[str] = (
|
|
[d for d in existing_raw if isinstance(d, str)]
|
|
if isinstance(existing_raw, list)
|
|
else []
|
|
)
|
|
|
|
to_add = [d for d in update.directories if d not in existing_list]
|
|
if not to_add:
|
|
return
|
|
|
|
permissions["additionalDirectories"] = existing_list + to_add
|
|
updated = dict(settings)
|
|
updated["permissions"] = permissions
|
|
try:
|
|
save_settings_for_source(dest, cwd, updated)
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning("Failed to add directories to %s: %s", dest, exc)
|
|
|
|
|
|
def _persist_remove_directories(
|
|
update: RemoveDirectoriesUpdate, cwd: str
|
|
) -> None:
|
|
dest = update.destination
|
|
if dest in ("session", "cliArg"):
|
|
store = _store_for_source(dest)
|
|
if store is not None:
|
|
for directory in update.directories:
|
|
store.remove_directory(directory)
|
|
return
|
|
if dest not in _EDITABLE_SOURCES:
|
|
return
|
|
|
|
settings = get_settings_for_source(dest, cwd)
|
|
if not settings:
|
|
return
|
|
permissions_raw = settings.get("permissions")
|
|
if not isinstance(permissions_raw, dict):
|
|
return
|
|
existing_raw = permissions_raw.get("additionalDirectories")
|
|
if not isinstance(existing_raw, list):
|
|
return
|
|
|
|
to_remove = set(update.directories)
|
|
filtered = [d for d in existing_raw if not (isinstance(d, str) and d in to_remove)]
|
|
|
|
permissions = dict(permissions_raw)
|
|
permissions["additionalDirectories"] = filtered
|
|
updated = dict(settings)
|
|
updated["permissions"] = permissions
|
|
try:
|
|
save_settings_for_source(dest, cwd, updated)
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning("Failed to remove directories from %s: %s", dest, exc)
|
|
|
|
|
|
def _persist_set_mode(update: SetModeUpdate, cwd: str) -> None:
|
|
dest = update.destination
|
|
if dest in ("session", "cliArg"):
|
|
store = _store_for_source(dest)
|
|
if store is not None:
|
|
store.set_mode(update.mode)
|
|
return
|
|
if dest not in _EDITABLE_SOURCES:
|
|
return
|
|
|
|
settings = get_settings_for_source(dest, cwd) or {}
|
|
permissions_raw = settings.get("permissions")
|
|
permissions = dict(permissions_raw) if isinstance(permissions_raw, dict) else {}
|
|
permissions["defaultMode"] = update.mode
|
|
|
|
updated = dict(settings)
|
|
updated["permissions"] = permissions
|
|
try:
|
|
save_settings_for_source(dest, cwd, updated)
|
|
except Exception as exc: # pragma: no cover
|
|
logger.warning("Failed to persist defaultMode to %s: %s", dest, exc)
|
|
|
|
|
|
def _persist_permission_update(update: PermissionUpdate, cwd: str) -> None:
|
|
"""OpenSpace ``persistPermissionUpdate`` — PermissionUpdate.ts L222-L342."""
|
|
if isinstance(update, AddRulesUpdate):
|
|
_persist_add_rules(update, cwd)
|
|
elif isinstance(update, ReplaceRulesUpdate):
|
|
_persist_replace_rules(update, cwd)
|
|
elif isinstance(update, RemoveRulesUpdate):
|
|
_persist_remove_rules(update, cwd)
|
|
elif isinstance(update, AddDirectoriesUpdate):
|
|
_persist_add_directories(update, cwd)
|
|
elif isinstance(update, RemoveDirectoriesUpdate):
|
|
_persist_remove_directories(update, cwd)
|
|
elif isinstance(update, SetModeUpdate):
|
|
_persist_set_mode(update, cwd)
|
|
else: # pragma: no cover
|
|
logger.warning(
|
|
"Unknown permission update type: %r", type(update).__name__
|
|
)
|
|
|
|
|
|
def persist_permission_updates(
|
|
updates: Iterable[PermissionUpdate], cwd: str
|
|
) -> None:
|
|
"""OpenSpace ``persistPermissionUpdates`` — PermissionUpdate.ts L349-L353."""
|
|
for update in updates:
|
|
_persist_permission_update(update, cwd)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §9 load_tool_permission_context — top-level aggregator
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _resolve_default_mode(cwd: str) -> PermissionMode:
|
|
"""Resolve effective default mode across all sources.
|
|
|
|
Priority (later wins), matching OpenSpace's "later sources override earlier"
|
|
convention declared in ``SETTING_SOURCES`` (constants.ts L7-L22) and
|
|
then amended by runtime stores (cli / session):
|
|
|
|
userSettings < projectSettings < localSettings
|
|
< cliArg store < session store
|
|
|
|
Returns ``"default"`` when no source specifies one.
|
|
"""
|
|
selected: Optional[ExternalPermissionMode] = None
|
|
|
|
for source in (
|
|
"userSettings",
|
|
"projectSettings",
|
|
"localSettings",
|
|
):
|
|
data = get_settings_for_source(source, cwd)
|
|
mode = _default_mode_from_settings(data)
|
|
if mode is not None:
|
|
selected = mode
|
|
|
|
# Runtime overrides (cliArg first so session wins on explicit /mode).
|
|
for store in (_CLIARG_STORE, _SESSION_STORE):
|
|
mode = store.get_mode()
|
|
if mode is not None:
|
|
selected = mode
|
|
|
|
return selected if selected is not None else "default"
|
|
|
|
|
|
def load_tool_permission_context(
|
|
cwd: str, mode: Optional[PermissionMode] = None
|
|
) -> ToolPermissionContext:
|
|
"""Aggregate every source into a :class:`ToolPermissionContext`.
|
|
|
|
* Rules: merged via :func:`load_all_permission_rules_from_disk`.
|
|
* Additional working directories: cwd (always, ``session``-sourced)
|
|
plus any ``permissions.additionalDirectories`` found on disk plus
|
|
runtime-store directories. The first source to contribute a path
|
|
wins — later sources don't overwrite the :class:`AdditionalWorkingDirectory`
|
|
attribution.
|
|
* Mode: explicit ``mode`` arg wins; otherwise :func:`_resolve_default_mode`.
|
|
"""
|
|
allow, deny, ask = load_all_permission_rules_from_disk(cwd)
|
|
|
|
working_dirs: Dict[str, AdditionalWorkingDirectory] = {
|
|
cwd: AdditionalWorkingDirectory(path=cwd, source="session")
|
|
}
|
|
|
|
for source in PERMISSION_RULE_SOURCES:
|
|
store = _store_for_source(source)
|
|
if store is not None:
|
|
for directory in store.get_directories():
|
|
if directory not in working_dirs:
|
|
working_dirs[directory] = AdditionalWorkingDirectory(
|
|
path=directory, source=source
|
|
)
|
|
continue
|
|
|
|
settings = get_settings_for_source(source, cwd)
|
|
for directory in _additional_dirs_from_settings(settings):
|
|
if directory not in working_dirs:
|
|
working_dirs[directory] = AdditionalWorkingDirectory(
|
|
path=directory, source=source
|
|
)
|
|
|
|
resolved_mode: PermissionMode = mode if mode is not None else _resolve_default_mode(cwd)
|
|
|
|
return ToolPermissionContext(
|
|
mode=resolved_mode,
|
|
additional_working_directories=working_dirs,
|
|
always_allow_rules=allow,
|
|
always_deny_rules=deny,
|
|
always_ask_rules=ask,
|
|
# Reaching the effective bypass mode means the caller explicitly
|
|
# selected it or opted into it as a settings default. Bash keeps a
|
|
# separate availability guard, so carry that opt-in into the context.
|
|
is_bypass_permissions_mode_available=(
|
|
resolved_mode == "bypassPermissions"
|
|
),
|
|
)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
# §10 Public API surface
|
|
# ════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
__all__ = [
|
|
# Top-level loaders
|
|
"load_tool_permission_context",
|
|
"load_all_permission_rules_from_disk",
|
|
"load_permissions_from_source",
|
|
# Low-level settings I/O
|
|
"get_settings_for_source",
|
|
"save_settings_for_source",
|
|
# Rule mutations (single-source, disk)
|
|
"add_permission_rules_to_settings",
|
|
"delete_permission_rule_from_settings",
|
|
# Update dispatch
|
|
"apply_permission_update",
|
|
"persist_permission_updates",
|
|
# Runtime stores
|
|
"get_session_store",
|
|
"get_cliarg_store",
|
|
"reset_runtime_stores",
|
|
]
|