Sumsub Api Generic

"""Local cache for the Sumsub OpenAPI schema.

The schema is publicly served (no authentication required) at:

    https://api.sumsub.com/openapi.json

We fetch it on first use, cache locally for 24 h, and refresh transparently
on the next call once the cache goes stale. On a refresh that fails (network
down, server 5xx) we fall back to the stale cache with a stderr warning.

Environment:
  SUMSUB_OPENAPI            optional — absolute path to a local schema file
                            (bypasses cache + fetch entirely)
  SUMSUB_OPENAPI_URL        optional — override the source URL
  SUMSUB_SCHEMA_REFRESH=1   optional — force re-fetch even if cache is fresh
  XDG_CACHE_HOME            optional — override base cache dir
                            (default: ~/.cache)
"""

from __future__ import annotations

import json
import os
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

DEFAULT_SCHEMA_URL = "https://api.sumsub.com/openapi.json"
CACHE_TTL_SECONDS = 24 * 60 * 60  # 1 day


def schema_url() -> str:
    return os.environ.get("SUMSUB_OPENAPI_URL") or DEFAULT_SCHEMA_URL


def cache_path() -> Path:
    base = os.environ.get("XDG_CACHE_HOME") or str(Path.home() / ".cache")
    return Path(base) / "sumsub" / "openapi.json"


def _is_fresh(path: Path) -> bool:
    if not path.is_file():
        return False
    return (time.time() - path.stat().st_mtime) < CACHE_TTL_SECONDS


def _fetch_or_none() -> bytes | None:
    url = schema_url()
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return resp.read()
    except (urllib.error.URLError, urllib.error.HTTPError) as exc:
        print(f"warning: schema fetch from {url} failed: {exc}", file=sys.stderr)
        return None


def _atomic_write(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(f".tmp.{os.getpid()}")
    tmp.write_bytes(data)
    os.replace(tmp, path)


def load_schema() -> dict:
    """Return the parsed schema, fetching/caching as needed.

    Resolution order:
    1. $SUMSUB_OPENAPI override (skip cache + fetch).
    2. Local cache, if fresh (≤ 1 day) and SUMSUB_SCHEMA_REFRESH not set.
    3. Fetch from the schema URL, atomic-write to cache, return.
    4. On network failure with stale cache: fall back to stale + stderr warning.
    """
    override = os.environ.get("SUMSUB_OPENAPI")
    if override:
        p = Path(override)
        if not p.is_file():
            sys.exit(f"error: SUMSUB_OPENAPI={override} not a file")
        return json.loads(p.read_text())

    cache = cache_path()
    force = os.environ.get("SUMSUB_SCHEMA_REFRESH") == "1"

    if not force and _is_fresh(cache):
        return json.loads(cache.read_text())

    data = _fetch_or_none()
    if data is None:
        if cache.is_file():
            print(f"warning: using stale schema cache at {cache}", file=sys.stderr)
            return json.loads(cache.read_text())
        sys.exit(
            f"error: could not fetch schema from {schema_url()} and no cache exists at {cache}."
        )

    try:
        parsed = json.loads(data)
    except json.JSONDecodeError as exc:
        sys.exit(f"error: schema URL returned non-JSON ({exc}); first 200 bytes: {data[:200]!r}")

    if not parsed.get("openapi"):
        sys.exit("error: response did not look like OpenAPI (no 'openapi' field)")

    _atomic_write(cache, data)
    return parsed

#!/usr/bin/env python3
"""Search the Sumsub OpenAPI schema for endpoints matching keywords.

Each positional argument is a keyword that must appear (case-insensitive) in
the path, the summary, the tags, or the operationId. Multi-word AND match —
narrow as you go.

Usage:
    find_endpoint.py applicants tags
    find_endpoint.py accessTokens
    find_endpoint.py POST applicants     # mix HTTP method and keywords

The schema is auto-fetched and cached (24 h TTL) — see _schema_cache.py for
the cache location and env-var knobs.
"""

from __future__ import annotations

import sys

from _schema_cache import load_schema

METHODS = {"get", "post", "put", "patch", "delete", "options", "head"}


def main() -> int:
    args = [a.lower() for a in sys.argv[1:]]
    if not args:
        print(__doc__, file=sys.stderr)
        return 2

    method_filters = {a for a in args if a in METHODS}
    keyword_filters = [a for a in args if a not in METHODS]

    schema = load_schema()
    paths = schema.get("paths", {})

    rows: list[tuple[str, str, str, str]] = []
    for path, ops in paths.items():
        for method, op in ops.items():
            if method.lower() not in METHODS:
                continue
            if method_filters and method.lower() not in method_filters:
                continue
            summary = (op.get("summary") or "").strip()
            tags = ", ".join(op.get("tags") or [])
            op_id = (op.get("operationId") or "").strip()
            hay = f"{path}\n{summary}\n{tags}\n{op_id}".lower()
            if all(kw in hay for kw in keyword_filters):
                rows.append((method.upper(), path, summary, op_id))

    if not rows:
        print("(no matches)", file=sys.stderr)
        return 1

    rows.sort(key=lambda r: (r[1], r[0]))
    width = max(len(m) for m, *_ in rows)
    for method, path, summary, op_id in rows:
        line = f"{method.ljust(width)}  {path}"
        if summary:
            line += f"  — {summary}"
        if op_id:
            line += f"  ({op_id})"
        print(line)
    print(f"\n{len(rows)} match(es)", file=sys.stderr)
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Force-refresh the local Sumsub OpenAPI schema cache.

Equivalent to `SUMSUB_SCHEMA_REFRESH=1 ./find_endpoint.py <anything>`, but
prints a friendly summary of what landed.

Use when you know a new endpoint shipped and don't want to wait for the
24-hour TTL to expire.
"""

from __future__ import annotations

import os
import sys

os.environ["SUMSUB_SCHEMA_REFRESH"] = "1"

from _schema_cache import cache_path, load_schema, schema_url

schema = load_schema()
path = cache_path()
paths = len(schema.get("paths", {}))
schemas = len(schema.get("components", {}).get("schemas", {}))
size = path.stat().st_size

print(
    f"OK — refreshed from {schema_url()}\n"
    f"     cached at {path}\n"
    f"     {paths} paths, {schemas} schemas, {size:,} bytes",
    file=sys.stderr,
)

#!/usr/bin/env python3
"""Dump a single Sumsub OpenAPI operation: params, request body, responses.

Usage:
    show_endpoint.py GET /resources/applicants/{applicantId}/one
    show_endpoint.py POST /resources/accessTokens

Resolves $ref pointers one level deep so you can see the actual shape without
chasing references manually. For nested refs you'll still need to look them
up in the schema.

The schema is auto-fetched and cached (24 h TTL) — see _schema_cache.py for
the cache location and env-var knobs.
"""

from __future__ import annotations

import json
import sys
from typing import Any

from _schema_cache import load_schema


def _resolve(schema: dict[str, Any], ref: str) -> Any:
    # Only #/-style local refs; OpenAPI doesn't use anything else here.
    if not ref.startswith("#/"):
        return {"$ref": ref}
    node: Any = schema
    for part in ref[2:].split("/"):
        node = node[part]
    return node


def _inline_refs(schema: dict[str, Any], node: Any, depth: int = 1) -> Any:
    if depth < 0:
        return node
    if isinstance(node, dict):
        if "$ref" in node and isinstance(node["$ref"], str):
            target = _resolve(schema, node["$ref"])
            return _inline_refs(schema, target, depth - 1)
        return {k: _inline_refs(schema, v, depth) for k, v in node.items()}
    if isinstance(node, list):
        return [_inline_refs(schema, x, depth) for x in node]
    return node


def main() -> int:
    if len(sys.argv) != 3:
        print(__doc__, file=sys.stderr)
        return 2

    method = sys.argv[1].lower()
    path = sys.argv[2]

    schema = load_schema()
    paths = schema.get("paths", {})

    if path not in paths:
        print(f"path not found: {path}", file=sys.stderr)
        # Best-effort hint: list similar paths.
        prefix = path.rstrip("/").rsplit("/", 1)[0]
        similar = [p for p in paths if p.startswith(prefix)][:10]
        if similar:
            print("similar paths:", file=sys.stderr)
            for s in similar:
                print("  " + s, file=sys.stderr)
        return 1

    ops = paths[path]
    if method not in ops:
        print(f"{method.upper()} not defined on {path}", file=sys.stderr)
        print("defined methods: " + ", ".join(m.upper() for m in ops), file=sys.stderr)
        return 1

    op = ops[method]
    out = {
        "method": method.upper(),
        "path": path,
        "summary": op.get("summary"),
        "description": op.get("description"),
        "operationId": op.get("operationId"),
        "tags": op.get("tags"),
        "parameters": _inline_refs(schema, op.get("parameters", []), depth=2),
        "requestBody": _inline_refs(schema, op.get("requestBody"), depth=2),
        "responses": _inline_refs(schema, op.get("responses"), depth=2),
        "security": op.get("security"),
    }
    json.dump(out, sys.stdout, indent=2, ensure_ascii=False)
    print()
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env bash
# Sign + send a single Sumsub API request with App Token auth.
#
# Per https://docs.sumsub.com/reference/authentication: HMAC-SHA256 over
# ts + METHOD + path?query + body, hex digest, lowercase.
#
# Usage:
#   SUMSUB_APP_TOKEN=sbx:... SUMSUB_SECRET_KEY=... \
#     sumsub_curl.sh METHOD PATH_WITH_QUERY [BODY_FILE_OR_-]
#
# Refuses non-sandbox tokens unless SUMSUB_ALLOW_PROD=1.
set -euo pipefail

if [[ $# -lt 2 ]]; then
  sed -n '2,12p' "$0" >&2
  exit 2
fi

METHOD="$(printf '%s' "$1" | tr '[:lower:]' '[:upper:]')"
PATH_Q="$2"
BODY_ARG="${3-}"

: "${SUMSUB_APP_TOKEN:?set SUMSUB_APP_TOKEN}"
: "${SUMSUB_SECRET_KEY:?set SUMSUB_SECRET_KEY}"
BASE="${SUMSUB_BASE:-https://api.sumsub.com}"

if [[ "${SUMSUB_APP_TOKEN}" != sbx:* && "${SUMSUB_ALLOW_PROD:-0}" != "1" ]]; then
  echo "error: SUMSUB_APP_TOKEN does not look like a sandbox token (expected 'sbx:' prefix)." >&2
  echo "       Production credentials must not be shared with this skill." >&2
  exit 3
fi

if [[ "${PATH_Q}" != /* ]]; then
  echo "error: PATH must start with '/'" >&2
  exit 2
fi

BODY_FILE="$(mktemp)"
trap 'rm -f "${BODY_FILE}"' EXIT
if [[ -z "${BODY_ARG}" ]]; then
  : >"${BODY_FILE}"
elif [[ "${BODY_ARG}" == "-" ]]; then
  cat >"${BODY_FILE}"
else
  cp "${BODY_ARG}" "${BODY_FILE}"
fi

TS="$(date -u +%s)"

SIG="$(
  { printf '%s%s%s' "${TS}" "${METHOD}" "${PATH_Q}"; cat "${BODY_FILE}"; } \
    | openssl dgst -sha256 -hmac "${SUMSUB_SECRET_KEY}" -hex \
    | awk '{print $NF}'
)"

CURL_ARGS=(
  -sS -X "${METHOD}"
  -H "X-App-Token: ${SUMSUB_APP_TOKEN}"
  -H "X-App-Access-Ts: ${TS}"
  -H "X-App-Access-Sig: ${SIG}"
  -H "X-Agent-Source: sumsub-skills"
  -H "X-Agent-Source-Ver: 1.0.1"
  -H "Accept: application/json"
)

if [[ -s "${BODY_FILE}" ]]; then
  CURL_ARGS+=(-H "Content-Type: application/json" --data-binary "@${BODY_FILE}")
fi

curl "${CURL_ARGS[@]}" -w '\nHTTP %{http_code}\n' "${BASE}${PATH_Q}"