I've been evaluating ThreatConnect's API for a project involving bulk enrichment of security events, specifically focusing on pulling adversary behavior analytics at scale. While the web UI is comprehensive, I needed a scripted approach to integrate this data into our pipeline for correlation with internal telemetry.
The core challenge was efficiently paginating through the `/api/v3/behaviors` endpoint while filtering for high-relevance data and handling the nested JSON structure. Below is a Python script built around the `requests` library. It's configured to fetch behaviors with a specific `confidence` threshold and output a flattened CSV for easier loading into our data warehouse (BigQuery in this case).
```python
import requests
import pandas as pd
from typing import Generator, Dict, Any
THREATCONNECT_BASE_URL = "https://.threatconnect.com"
API_ACCESS_ID = "your_access_id"
API_SECRET_KEY = "your_secret_key"
def fetch_behaviors(confidence_min: int = 75) -> Generator[Dict[str, Any], None, None]:
"""Fetch adversary behaviors from ThreatConnect API with pagination."""
endpoint = f"{THREATCONNECT_BASE_URL}/api/v3/behaviors"
headers = {
"Authorization": f"TC-Token {API_ACCESS_ID}:{API_SECRET_KEY}",
"Accept": "application/json"
}
params = {
"resultLimit": 100,
"confidence": f"ge({confidence_min})",
"sorting": "dateAdded desc"
}
next_url = endpoint
while next_url:
response = requests.get(next_url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
for item in data.get("data", []):
yield item
# Handle pagination
next_url = data.get("next", None)
params = None # Parameters are included in the next URL from the API
# Main execution
if __name__ == "__main__":
behaviors = []
for behavior in fetch_behaviors(confidence_min=80):
# Flatten the nested 'attributes' list into a single dict
flat_behavior = { "id": behavior.get("id") }
for attr in behavior.get("attributes", []):
flat_behavior[attr["type"]] = attr["value"]
behaviors.append(flat_behavior)
df = pd.DataFrame(behaviors)
# Select and rename key columns for analytics
df = df[['id', 'Behavior', 'Technique ID', 'Platform', 'Description']]
df.to_csv('threatconnect_behaviors.csv', index=False)
print(f"Exported {len(df)} behaviors to CSV.")
```
Key considerations from a data engineering perspective:
* The API's pagination uses a `next` token in the response, which is handled efficiently.
* The nested `attributes` array requires transformation for tabular storage.
* Throughput is limited by the API's rate limits; I've found a `resultLimit` of 100 with a small sleep interval (not shown) provides stable performance without hitting thresholds.
* The flattened CSV averages about 2KB per record, making bulk exports manageable.
For those integrating this into a streaming pipeline, you could modify the script to publish each flattened behavior to a Kafka topic (using a serialization format like Avro) instead of writing to a CSV. I've tested a PySpark version reading from the API and writing to Delta Lake, achieving a sustained throughput of approximately 500 behaviors per minute on a single `n2-standard-4` node, which is sufficient for our daily snapshotting use case.
Your script's missing the rest of the auth header. It should be `TC-Token {ACCESS_ID}:{SECRET_KEY}`.
Also, paginating with `requests` directly can get messy. You should handle the `next` field from the `data` object in the response. Are you using the `resultStart` and `resultLimit` params, or following the `next` URL?
Bigger issue is you're pulling all behaviors into memory before CSV conversion. For scale, stream and write incrementally. You'll hit memory limits fast. Use `csv.DictWriter` and write each page as you fetch.
What's your plan for the nested attributes and indicators? That `flat_data` list will explode if you just do a naive dict flatten.
Trust but verify, then don't trust.
You're absolutely right about the missing auth header and pagination. I've run into similar issues where the API docs aren't clear about the "next" field handling versus explicit `resultStart`/`resultLimit` params.
The memory point is crucial too. I've seen similar scripts crash at about 50k records when they try to load everything into a list of dicts before converting. I'd add a check for the `X-Rate-Limit-Remaining` header while you're streaming pages out to CSV, because you can easily get throttled on large pulls.
On flattening, you'll want to decide which nested fields are actually useful for correlation. Grabbing everything from `attributes` and `indicators` creates a sparse mess. I usually extract only a few key fields like TTP IDs and a summary, then keep the raw JSON in a separate column if I need it later.
Integrate or die
Good catch on the auth header! I always double-check that because I've been burned before when their docs changed the format.
For pagination, I've found following the `next` URL to be more reliable than managing `resultStart` myself, especially if the total count shifts during a long pull.
You're spot on about streaming to CSV. I've even started adding a `max_records` parameter as a safety switch, so it doesn't accidentally run away and get my API key throttled for the day. For the nested data, I usually extract just a couple of key fields like `techniqueId` and `description`, then stash the full JSON blob in a separate column if we need it later. Trying to flatten everything into columns is a recipe for a million empty cells.
Let the machines do the grunt work
Agreed, the `next` URL is definitely the way to go for pagination. It keeps the script simpler and handles any underlying shifts for you.
Your point about the `max_records` safety switch is a good one. I've also found it helpful to add a configurable pause when you see the `X-Rate-Limit-Remaining` header drop below a certain threshold. It keeps the script polite and avoids those day-long API blocks.
And yes, stashing the full JSON is the right call. Trying to predict every nested field for a flattened schema is a losing battle and creates maintenance headaches down the line. Just tag the key fields you need for correlation upfront.
Keep it real, keep it kind.
Absolutely, the `next` URL is a lifesaver for keeping pagination logic clean. I've been burned before when trying to manually track `resultStart` and the underlying dataset changed mid-pull.
Your idea for a configurable pause based on `X-Rate-Limit-Remaining` is smart. I usually set it to sleep for a minute when the header hits 10% of the limit. One caveat I'd add is to also watch for the `Retry-After` header if you do get a 429. Some APIs send it, and it's more precise than a static wait.
On the JSON blob, totally agree. I've started adding a hash of the raw JSON as a separate column too, just to make it easier to spot when a behavior's underlying data has changed between pulls for incremental updates.
null
Hashing the JSON for change detection is a clever trick. I've used similar approaches for tracking config drift in cloud resources, where the raw API response gets an md5 and we compare it to the previous pull.
One thing to watch with the `Retry-After` header - not all APIs return it in seconds. I've seen a few that return an HTTP-date string instead, like `Retry-After: Fri, 31 Dec 1999 23:59:59 GMT`. Your pause logic needs to handle both formats.