feat(properties): prepare property commands before execution
Build and Push Docker Container / build-and-push (push) Successful in 4m28s
Build and Push Docker Container / build-and-push (push) Successful in 4m28s
- Create missing note files before running property:set. - Default date-only datetime values to midnight. - Keep property names out of path authorization checks. - Split policy, vault, daily note, and command target helpers. - Split n8n node args, description, output, and API response helpers. - Split API tests by auth, command, daily note, and property behavior. - Bump API and n8n node versions.
This commit is contained in:
+5
-139
@@ -1,10 +1,12 @@
|
||||
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
|
||||
from app.command_target import CommandTarget, extract_command_target, parse_kv_arg
|
||||
from app.config import DEFAULT_VAULT_NAME
|
||||
from app.daily_notes import daily_note_path_for_vault
|
||||
from app.path_policy import is_path_allowed, normalize_policy_path
|
||||
from app.vaults import vault_for_path, vault_root_for_target
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TokenPolicy:
|
||||
@@ -23,16 +25,6 @@ class TokenPolicy:
|
||||
def allows_all_commands(self) -> bool:
|
||||
return "*" in self.commands
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CommandTarget:
|
||||
command:str|None
|
||||
vault:str|None
|
||||
paths:tuple[str, ...]
|
||||
|
||||
DEFAULT_PATH_KEYS = {"path", "file", "folder", "to"}
|
||||
PATH_KEYS_BY_COMMAND = {
|
||||
"rename": DEFAULT_PATH_KEYS | {"name"},
|
||||
}
|
||||
PATH_REQUIRED_WHEN_RESTRICTED = {
|
||||
"append", "prepend", "read", "create", "delete", "rename", "move", "file", "folder", "open", "diff",
|
||||
"search", "search:context", "files", "folders",
|
||||
@@ -51,132 +43,6 @@ def tuple_claim(value:Any, default:tuple[str, ...]) -> tuple[str, ...]:
|
||||
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 path_keys_for_command(command:str|None) -> set[str]:
|
||||
return PATH_KEYS_BY_COMMAND.get(command or "", DEFAULT_PATH_KEYS)
|
||||
|
||||
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
|
||||
|
||||
path_keys = path_keys_for_command(command)
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user