From 0e7dd21c6638d759a5a8171ae160c05a328061e7 Mon Sep 17 00:00:00 2001 From: Fizzlepoof Date: Fri, 15 May 2026 17:44:30 +0000 Subject: [PATCH] Add vault-backed Karakeep API helper --- automation/README.md | 16 ++++ automation/bin/karakeep_api.py | 139 +++++++++++++++++++++++++++++++++ 2 files changed, 155 insertions(+) create mode 100755 automation/bin/karakeep_api.py diff --git a/automation/README.md b/automation/README.md index 37899c1..eefa3f8 100644 --- a/automation/README.md +++ b/automation/README.md @@ -12,6 +12,22 @@ This directory includes repo-side helpers for homelab operational tasks. See `docs/operations/KITCHENOWL_RECIPE_IMPORT.md` for setup and usage. +## Karakeep API helper + +- `bin/karakeep_api.py` — tiny Karakeep REST helper backed by the local encrypted vault entry + - reads `.secrets/karakeep_secrets.json.enc` + - uses `baseUrl` + `apiKey` + - targets `/api/v1/...` routes by default + - supports ad hoc GET/POST/PATCH/DELETE calls with query/body args + +Examples: + +```bash +automation/bin/karakeep_api.py bookmarks --query limit=5 --pretty +automation/bin/karakeep_api.py tags --pretty +automation/bin/karakeep_api.py /api/health --raw-path --pretty +``` + ## Backup scripts - `bin/pd_backup_postgres.sh` — creates gzipped `pg_dump` exports for the configured Postgres DB list (quote space-separated DB names in `.env`, e.g. `POSTGRES_DB_LIST="n8n paperless"`) diff --git a/automation/bin/karakeep_api.py b/automation/bin/karakeep_api.py new file mode 100755 index 0000000..d69bba2 --- /dev/null +++ b/automation/bin/karakeep_api.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +"""Small Karakeep API helper that reads its token from the local encrypted vault. + +Default behavior: +- reads `.secrets/karakeep_secrets.json.enc` +- uses `baseUrl` + `apiKey` +- targets `/api/v1/...` routes unless an absolute path is given + +Examples: + automation/bin/karakeep_api.py bookmarks --pretty + automation/bin/karakeep_api.py tags --pretty + automation/bin/karakeep_api.py /api/health --raw-path --pretty + automation/bin/karakeep_api.py bookmarks --query limit=5 --pretty +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +from pathlib import Path +from typing import Any +from urllib import error, parse, request + +SCRIPT_DIR = Path(__file__).resolve().parent +REPO_ROOT = SCRIPT_DIR.parent.parent +WORKSPACE = Path('/home/fizzlepoof/.openclaw/workspace') +DEFAULT_SECRETS_DIR = WORKSPACE / '.secrets' +DEFAULT_SECRET_FILE = DEFAULT_SECRETS_DIR / 'karakeep_secrets.json.enc' +DEFAULT_KEY_FILE = DEFAULT_SECRETS_DIR / 'encryption.key' +DEFAULT_TIMEOUT = 30 + + +def decrypt_json(path: Path, key_file: Path) -> dict[str, Any]: + raw = subprocess.check_output([ + 'openssl', 'enc', '-d', '-aes-256-cbc', '-pbkdf2', + '-in', str(path), + '-pass', f'file:{key_file}', + ]) + return json.loads(raw) + + +def parse_kv(items: list[str]) -> dict[str, str]: + out: dict[str, str] = {} + for item in items: + if '=' not in item: + raise SystemExit(f'Expected key=value, got: {item}') + key, value = item.split('=', 1) + out[key] = value + return out + + +def build_url(base_url: str, endpoint: str, raw_path: bool) -> str: + base = base_url.rstrip('/') + if raw_path: + if not endpoint.startswith('/'): + endpoint = '/' + endpoint + return base + endpoint + endpoint = endpoint.strip('/') + return f'{base}/api/v1/{endpoint}' + + +def api_request(method: str, url: str, api_key: str, *, query: dict[str, str] | None = None, + body: dict[str, Any] | list[Any] | None = None, timeout: int = DEFAULT_TIMEOUT) -> tuple[int, Any, dict[str, str]]: + if query: + url += ('&' if '?' in url else '?') + parse.urlencode(query, doseq=True) + data = None + headers = { + 'Authorization': f'Bearer {api_key}', + 'Accept': 'application/json', + 'User-Agent': 'doris-karakeep-helper/1.0', + } + if body is not None: + data = json.dumps(body).encode('utf-8') + headers['Content-Type'] = 'application/json' + req = request.Request(url, data=data, method=method.upper(), headers=headers) + try: + with request.urlopen(req, timeout=timeout) as resp: + raw = resp.read().decode('utf-8', 'replace') + ctype = resp.headers.get('Content-Type', '') + payload = json.loads(raw) if 'json' in ctype else raw + return resp.status, payload, dict(resp.headers.items()) + except error.HTTPError as exc: + raw = exc.read().decode('utf-8', 'replace') + try: + payload = json.loads(raw) + except Exception: + payload = raw + return exc.code, payload, dict(exc.headers.items()) + + +def main() -> int: + parser = argparse.ArgumentParser(description='Query Karakeep using the encrypted local vault secret.') + parser.add_argument('endpoint', help='Endpoint alias/path. Default base is /api/v1/. Example: bookmarks, tags, lists') + parser.add_argument('--method', default='GET', help='HTTP method (default: GET)') + parser.add_argument('--query', action='append', default=[], metavar='KEY=VALUE', help='Query params') + parser.add_argument('--data', action='append', default=[], metavar='KEY=VALUE', help='JSON body fields') + parser.add_argument('--json-body', help='Raw JSON body string') + parser.add_argument('--raw-path', action='store_true', help='Use endpoint as-is instead of prefixing /api/v1/') + parser.add_argument('--pretty', action='store_true', help='Pretty-print JSON output') + parser.add_argument('--headers', action='store_true', help='Print response headers to stderr') + parser.add_argument('--timeout', type=int, default=DEFAULT_TIMEOUT) + parser.add_argument('--secret-file', default=str(DEFAULT_SECRET_FILE)) + parser.add_argument('--key-file', default=str(DEFAULT_KEY_FILE)) + args = parser.parse_args() + + secret = decrypt_json(Path(args.secret_file), Path(args.key_file)) + base_url = secret['baseUrl'] + api_key = secret['apiKey'] + + query = parse_kv(args.query) + body: dict[str, Any] | list[Any] | None = None + if args.json_body: + body = json.loads(args.json_body) + elif args.data: + body = parse_kv(args.data) + + url = build_url(base_url, args.endpoint, args.raw_path) + status, payload, headers = api_request(args.method, url, api_key, query=query, body=body, timeout=args.timeout) + + if args.headers: + for k, v in headers.items(): + print(f'{k}: {v}', file=sys.stderr) + + if isinstance(payload, (dict, list)): + if args.pretty: + print(json.dumps(payload, indent=2, sort_keys=True)) + else: + print(json.dumps(payload)) + else: + print(payload) + + return 0 if 200 <= status < 300 else 1 + + +if __name__ == '__main__': + raise SystemExit(main())