Add vault-backed Karakeep API helper

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())