daniel156161
c34dab2cca
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.
124 lines
3.6 KiB
Python
124 lines
3.6 KiB
Python
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)
|