GraphQL Connector with OpenCTI Schema Translator

.gitignore

@@ -166,4 +166,5 @@ cython_debug/
 
 .DS_Store
 .claude/
-.vscode
+.vscode
+docs/opencti_schema.json

README.md

@@ -19,6 +19,7 @@ New development and releases are published under the `pyapiary` package name.
   - [Customizing API Requests with `**kwargs`](#customizing-api-requests-with-kwargs)
   - [Proxy Awareness](#proxy-awareness)
   - [SSL Verification and Per-Request Options](#ssl-verification-and-per-request-options)
+  - [GraphQL & OpenCTI](#-graphql--opencti)
 - [DBMS Connectors](#-dbms-connectors)
   - [MongoDB](#mongodb)
   - [Elasticsearch](#elasticsearch)
@@ -189,6 +190,51 @@ print(response.http_version)
 
 ---
 
+### 🧬 GraphQL & OpenCTI
+
+`GraphQLConnector` is a generic, schema-agnostic GraphQL executor built on the same `Broker` (so it inherits retries, proxies, timeouts, logging). Like the DBMS connectors, it does not care what your query is — it just runs it. It returns the raw `httpx.Response` (house convention) and raises `GraphQLError` when a `200` response carries a top-level `errors` array (GraphQL reports query errors with HTTP 200, so `raise_for_status` never catches them).
+
+```python
+from pyapiary.api_connectors.graphql import GraphQLConnector
+
+gql = GraphQLConnector(base_url="https://api.example.com", endpoint="/graphql")
+resp = gql.execute("query($n: Int) { things(first: $n) { id } }", {"n": 5})
+data = resp.json()["data"]
+# Pass raise_on_errors=False to inspect resp.json()["errors"] yourself.
+```
+
+`OpenCTIConnector` is a thick, curated connector (urlscan-style) layered on `GraphQLConnector`, with query field selections pinned to OpenCTI **6.9.x** (verified against `docs/opencti-6.9.6.graphql`). It reads `OPENCTI_URL` and `OPENCTI_TOKEN` from the environment (or accepts them explicitly). Access is governed entirely by the permissions on the token's user — so scope the token appropriately, including for mutations run via `execute`.
+
+```python
+from pyapiary.api_connectors.opencti import OpenCTIConnector
+
+with OpenCTIConnector() as octi:                       # env: OPENCTI_URL / OPENCTI_TOKEN
+    # filter helper builds the 6.9.x FilterGroup for you
+    flt = octi.filter_group("name", "Cobalt Strike")
+
+    # generic entity search (any type) — `representative.main` is a universal label
+    resp = octi.search_entities(types=["Malware"], filters=flt, first=50)
+    conn = resp.json()["data"]["stixCoreObjects"]
+
+    # dedicated methods for type-specific fields
+    octi.get_indicators(search="1.2.3.4")
+    octi.get_observables(types=["IPv4-Addr"])
+    octi.get_relationships(from_id="<id>", relationship_type="uses")
+
+    # anything outside the catalog: drop to the generic executor
+    octi.execute("query { me { name } }")
+```
+
+> **Existence check** — there is no `exists()` method (no connector returns a synthesized value). Use `search_entities(..., first=1)` and check whether the result's `edges` is empty.
+>
+> **Pagination** — connectors don't auto-paginate (same as `URLScanConnector`). Pass `after` with the previous page's `pageInfo.endCursor` and loop yourself; `pageInfo.globalCount` gives the total.
+>
+> **Async** — `AsyncGraphQLConnector` / `AsyncOpenCTIConnector` mirror the sync API.
+
+The query catalog (`opencti_queries.py`) is validated offline against the committed SDL in CI. To refresh the schema for a new OpenCTI version, run a GraphQL introspection query against the instance (requires the server's `APP__GRAPHQL__PLAYGROUND__FORCE_DISABLED_INTROSPECTION=false`), save the result as `docs/opencti_schema.json`, and convert it to SDL with `graphql-core`'s `build_client_schema` + `print_schema`. Then use `dev_env/opencti/gen_opencti_fields.py <Type>` to author field selections from the SDL.
+
+---
+
 ## 🗃️ DBMS Connectors
 
 Each database connector follows a class-based pattern and supports reusable sessions, query helpers, and in some cases bulk helpers (e.g., `insert_many`, `bulk_insert`, etc.).

dev_env/opencti/gen_opencti_fields.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Dev helper: generate GraphQL field selections from the pinned OpenCTI SDL.
+
+This is a *development-time* tool, NOT imported by the library at runtime (so
+the runtime stays dependency-free apart from httpx). Use it to author or expand
+the static query field lists in ``pyapiary.api_connectors.opencti_queries``
+without hand-maintaining them or marrying to a pydantic model -- the SDL is the
+source of truth.
+
+GraphQL has no ``SELECT *``; this emits all *leaf* (scalar/enum) fields of a
+type. Object/relationship fields are skipped (they would recurse); add those by
+hand where you need them.
+
+Usage:
+    pip install graphql-core
+    python dev_env/opencti/gen_opencti_fields.py Indicator
+    python dev_env/opencti/gen_opencti_fields.py Malware --fragment
+"""
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+from graphql import build_schema
+from graphql.type import GraphQLEnumType, GraphQLScalarType
+
+
+def _find_sdl() -> Path | None:
+    """Walk up from this file to find docs/opencti-6.9.6.graphql."""
+    for parent in Path(__file__).resolve().parents:
+        candidate = parent / "docs" / "opencti-6.9.6.graphql"
+        if candidate.exists():
+            return candidate
+    return None
+
+
+def _unwrap(t):
+    while hasattr(t, "of_type"):
+        t = t.of_type
+    return t
+
+
+def scalar_fields(schema, type_name: str) -> list[str]:
+    """All leaf (scalar/enum) field names of a type, straight from the schema."""
+    t = schema.type_map.get(type_name)
+    if t is None or not hasattr(t, "fields"):
+        raise SystemExit(f"Type not found or has no fields: {type_name}")
+    return [
+        name
+        for name, f in t.fields.items()
+        if isinstance(_unwrap(f.type), (GraphQLScalarType, GraphQLEnumType))
+    ]
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("type_name", help="GraphQL type name, e.g. Indicator")
+    ap.add_argument("--fragment", action="store_true",
+                    help="wrap as an inline fragment: '... on Type { ... }'")
+    ap.add_argument("--sdl", default=None, help="path to SDL file (auto-detected by default)")
+    args = ap.parse_args()
+
+    sdl_path = Path(args.sdl) if args.sdl else _find_sdl()
+    if sdl_path is None or not sdl_path.exists():
+        raise SystemExit("Could not find docs/opencti-6.9.6.graphql; pass --sdl explicitly.")
+
+    schema = build_schema(sdl_path.read_text())
+    fields = scalar_fields(schema, args.type_name)
+    body = " ".join(fields)
+    if args.fragment:
+        print(f"... on {args.type_name} {{ {body} }}")
+    else:
+        print(body)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())