obsidian-api/app/policy.py

from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
import json

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"}
DAILY_NOTE_COMMANDS = {"daily:path", "daily:read", "daily:append", "daily:prepend"}

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 moment_to_strftime(format_string:str) -> str:
  replacements = [
    ("YYYY", "%Y"),
    ("YY", "%y"),
    ("MMMM", "%B"),
    ("MMM", "%b"),
    ("MM", "%m"),
    ("DDDD", "%j"),
    ("DD", "%d"),
    ("dddd", "%A"),
    ("ddd", "%a"),
  ]
  converted = format_string or "YYYY-MM-DD"
  for moment_token, strftime_token in replacements:
    converted = converted.replace(moment_token, strftime_token)
  return converted

def daily_note_path_for_vault(vault_root:Path, now:datetime|None = None) -> str:
  settings_path = vault_root / ".obsidian" / "daily-notes.json"
  folder = ""
  format_string = "YYYY-MM-DD"

  if settings_path.exists() and settings_path.is_file():
    try:
      settings = json.loads(settings_path.read_text(encoding="utf-8"))
      if isinstance(settings, dict):
        folder_value = settings.get("folder")
        format_value = settings.get("format")
        if isinstance(folder_value, str):
          folder = folder_value.strip().strip("/")
        if isinstance(format_value, str) and format_value.strip():
          format_string = format_value.strip()
    except (OSError, json.JSONDecodeError):
      pass

  note_name = (now or datetime.now()).strftime(moment_to_strftime(format_string))
  if not note_name.endswith(".md"):
    note_name += ".md"
  return f"{folder}/{note_name}" if folder else note_name

def vault_root_for_target(target:CommandTarget, cwd:Path) -> Path:
  if target.vault and target.vault in REGISTERED_VAULTS:
    return Path(REGISTERED_VAULTS[target.vault]).resolve()

  vault_name = vault_for_path(cwd) or DEFAULT_VAULT_NAME
  return Path(REGISTERED_VAULTS.get(vault_name, cwd)).resolve()

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

  paths = list(target.paths)
  if not paths and command in DAILY_NOTE_COMMANDS:
    paths.append(daily_note_path_for_vault(vault_root_for_target(target, cwd)))

  if not 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 paths if not is_path_allowed(path, policy.paths)]
  if denied:
    raise PermissionError(f"token is not allowed to access path: {denied[0]}")