from dataclasses import dataclass from pathlib import Path from typing import Any from app.config import DEFAULT_VAULT_NAME, REGISTERED_VAULTS @dataclass(frozen=True) class TokenPolicy: token:str = "" name:str = "jwt" vaults:tuple[str, ...] = ("*",) paths:tuple[str, ...] = ("*",) commands:tuple[str, ...] = ("*",) def allows_all_vaults(self) -> bool: return "*" in self.vaults def allows_all_paths(self) -> bool: return "*" in self.paths def allows_all_commands(self) -> bool: return "*" in self.commands @dataclass(frozen=True) class CommandTarget: command:str|None vault:str|None paths:tuple[str, ...] PATH_KEYS = {"path", "file", "folder", "name", "to"} PATH_REQUIRED_WHEN_RESTRICTED = { "append", "prepend", "read", "create", "delete", "rename", "move", "file", "folder", "open", "diff", "search", "search:context", "files", "folders", "property:set", "property:remove", "property:read", "properties", "aliases", "backlinks", "links", "outlinks", "tags", "tasks", "task", } PATHLESS_ALLOWED_WHEN_RESTRICTED = {"help", "version"} def tuple_claim(value:Any, default:tuple[str, ...]) -> tuple[str, ...]: if value is None: return default if isinstance(value, str): return (value,) if isinstance(value, list) and all(isinstance(item, str) for item in value): return tuple(value) raise ValueError("JWT claims vaults, paths and commands must be strings or string arrays") def vault_for_path(path:Path) -> str|None: resolved = path.resolve() best_name:str|None = None best_len = -1 for name, vault_path in REGISTERED_VAULTS.items(): root = Path(vault_path).resolve() try: resolved.relative_to(root) except ValueError: continue root_len = len(str(root)) if root_len > best_len: best_name = name best_len = root_len return best_name def normalize_policy_path(value:str) -> str: normalized = value.strip().replace("\\", "/").lstrip("/") while "//" in normalized: normalized = normalized.replace("//", "/") return normalized def is_path_allowed(path:str, allowed_patterns:tuple[str, ...]) -> bool: normalized = normalize_policy_path(path) for pattern in allowed_patterns: pattern_normalized = normalize_policy_path(pattern) if pattern_normalized == "*": return True if pattern_normalized.endswith("/**"): prefix = pattern_normalized[:-3].rstrip("/") if normalized == prefix or normalized.startswith(prefix + "/"): return True elif pattern_normalized.endswith("/"): if normalized.startswith(pattern_normalized): return True elif normalized == pattern_normalized or normalized.startswith(pattern_normalized.rstrip("/") + "/"): return True return False def parse_kv_arg(arg:str) -> tuple[str, str]|None: if "=" not in arg: return None key, value = arg.split("=", 1) if not key: return None return key, value def extract_command_target(args:list[str]) -> CommandTarget: vault:str|None = None command:str|None = None paths:list[str] = [] for arg in args: parsed = parse_kv_arg(arg) if parsed and parsed[0] == "vault" and command is None: vault = parsed[1] continue command = arg break for arg in args: parsed = parse_kv_arg(arg) if not parsed: continue key, value = parsed if key in PATH_KEYS and value: paths.append(value) return CommandTarget(command=command, vault=vault, paths=tuple(paths)) def authorize_command(policy:TokenPolicy|None, args:list[str], cwd:Path) -> None: if policy is None: return target = extract_command_target(args) command = target.command or "" if not policy.allows_all_commands() and command not in policy.commands: raise PermissionError(f"token is not allowed to run command: {command}") if not policy.allows_all_vaults(): vault = target.vault or vault_for_path(cwd) or DEFAULT_VAULT_NAME if vault not in policy.vaults: raise PermissionError(f"token is not allowed to access vault: {vault}") if policy.allows_all_paths(): return if not target.paths: if command in PATHLESS_ALLOWED_WHEN_RESTRICTED: return if command in PATH_REQUIRED_WHEN_RESTRICTED: raise PermissionError("token has path restrictions; command must include an allowed path/file/folder/name argument") raise PermissionError(f"token has path restrictions; command is not allowed without an explicit path: {command}") denied = [path for path in target.paths if not is_path_allowed(path, policy.paths)] if denied: raise PermissionError(f"token is not allowed to access path: {denied[0]}")