obsidian-api/app/main.py

from pathlib import Path
from typing import Any
import json

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from app import auth
from app.auth import require_auth
from app.config import CLI_BIN, DEFAULT_TIMEOUT, HOST, MAX_TIMEOUT, PORT, VAULT_PATH
from app.policy import authorize_command
from app.vaults import vault_for_path
from app.runner import execute_command

def error(status_code:int, detail:str) -> JSONResponse:
  return JSONResponse({"detail": detail}, status_code=status_code)

def parse_args(value:Any) -> list[str]:
  if value is None:
    return []
  if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
    raise ValueError("args must be a list of strings")
  return value

def parse_timeout(value:Any) -> float:
  if value is None:
    return DEFAULT_TIMEOUT
  if not isinstance(value, int | float) or value <= 0:
    raise ValueError("timeout must be a positive number")
  return min(float(value), MAX_TIMEOUT)

def safe_cwd(cwd:str|None) -> Path:
  path = Path(cwd).resolve() if cwd else VAULT_PATH
  if not path.exists() or not path.is_dir():
    raise ValueError(f"cwd does not exist or is not a directory: {path}")
  if vault_for_path(path) is None:
    raise ValueError(f"cwd must be inside a registered vault: {path}")
  return path

async def health(request:Request) -> JSONResponse:
  return JSONResponse(
    {
      "ok": True,
      "vault": str(VAULT_PATH),
      "cli": CLI_BIN,
      "auth": bool(auth.JWT_SECRET),
      "jwt": bool(auth.JWT_SECRET),
    }
  )

async def run_command(request:Request) -> JSONResponse:
  auth_context = require_auth(request)
  if isinstance(auth_context, JSONResponse):
    return auth_context

  try:
    payload = await request.json()
  except json.JSONDecodeError:
    return error(400, "invalid JSON body")

  try:
    if "command" in payload:
      raise ValueError("command strings are not allowed; use args as a list of strings")
    if "mode" in payload:
      raise ValueError("mode is not allowed; only the configured Obsidian CLI can be executed")

    args = parse_args(payload.get("args"))
    stdin = payload.get("stdin")
    if stdin is not None and not isinstance(stdin, str):
      raise ValueError("stdin must be a string")

    cwd_value = payload.get("cwd")
    if cwd_value is not None and not isinstance(cwd_value, str):
      raise ValueError("cwd must be a string")

    cwd = safe_cwd(cwd_value)
    timeout = parse_timeout(payload.get("timeout"))
    authorize_command(auth_context.policy, args, cwd)
  except ValueError as exc:
    return error(400, str(exc))
  except PermissionError as exc:
    return error(403, str(exc))

  return await execute_command(args=args, cwd=cwd, stdin=stdin, timeout=timeout)

async def typescript_sdk_example(request:Request) -> JSONResponse:
  return JSONResponse(
    {
      "example": """
import { ObsidianCliClient } from '@obsidian-api/sdk';

const obsidian = new ObsidianCliClient({
  baseUrl: 'http://localhost:8080',
  token: '<jwt>',
});

await obsidian.create({
  path: 'Inbox/Hello.md',
  content: '# Hello from n8n/Prism',
  overwrite: true,
});

const note = await obsidian.read({ path: 'Inbox/Hello.md' });
console.log(note.stdout);
""".strip()
    }
  )

app = Starlette(
  debug=False,
  routes=[
    Route("/health", health, methods=["GET"]),
    Route("/commands", run_command, methods=["POST"]),
    Route("/sdk/typescript", typescript_sdk_example, methods=["GET"]),
  ],
)

if __name__ == "__main__":
  import uvicorn

  uvicorn.run("app.main:app", host=HOST, port=PORT)