obsidian-api/app/policy.py

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