Threat Feeds¶

Access real-time threat intelligence feeds delivered via API streaming or S3 download. Feeds provide newly observed domains, newly active domains, newly observed hosts, and more.

Understanding feed responses¶

Key difference: Feeds return generators with status=None, not regular response objects with status=200. This is the correct and expected behavior for feed endpoints.

from domaintools import API

api = API('your_username', 'your_api_key')

# Feeds return generators with status=None (this is correct!)
result = api.nod(after=-3600)

# Iterate over generator to get records
for record in result.response():
    print(f"Domain: {record['domain']}")

Authentication (automatic)¶

The SDK automatically switches to header-based authentication for threat feeds. You don't need to configure this manually.

What happens automatically:

When calling any feed endpoint (nod, nad, noh, etc.), the SDK sets header_authentication=True
The SDK disables HMAC signing (always_sign_api_key=False)
Your API key is passed directly in headers instead of being HMAC-signed

Why this matters:

Threat Feeds do NOT support HMAC-signed requests
If you try to force HMAC signing, you'll get a ValueError
This is handled transparently - just use feeds normally

Simple usage (recommended):

# This is all you need - authentication is automatic
api = API('your_username', 'your_api_key')
result = api.nod(after=-3600)

You can still override if needed:

# Explicit control (rarely needed)
api = API('your_username', 'your_api_key', header_authentication=True)

Access patterns¶

Session-based polling (recommended)¶

# Session-based polling automatically tracks position
results = api.nod(sessionID='my-unique-session')

# Subsequent calls return data since last request
results = api.nod(sessionID='my-unique-session')

Time-based queries¶

Query specific time ranges using relative or absolute times:

# Relative time (seconds ago)
results = api.nod(after=-3600)  # Last hour

# Absolute time range (ISO-8601 UTC)
results = api.nod(
    after='2024-01-01T00:00:00Z',
    before='2024-01-01T23:59:59Z'
)

Note: Using 1 hour (-3600 seconds) is more reliable than shorter windows for testing.

Available feeds¶

Newly observed domains (NOD)¶

Domains observed for the first time in DNS or other sources:

import json

# Get newly observed domains from the last hour
api = API('your_username', 'your_api_key')
result = api.nod(after=-3600)

# Feed responses are generators that yield JSON strings (feed-specific behavior)
# Unlike other API endpoints, feeds return raw JSON strings
for json_string in result.response():
    record = json.loads(json_string)
    print(f"Domain: {record['domain']}")
    break  # Show first result

Newly active domains (NAD)¶

Domains reactivated after 10+ days of inactivity:

import json

# Get domains reactivated after dormancy
api = API('your_username', 'your_api_key')
result = api.nad(after=-3600)

# Feed responses are generators that yield JSON strings
for json_string in result.response():
    record = json.loads(json_string)
    print(f"Domain: {record['domain']}")
    break

Newly observed hosts (NOH)¶

Fully qualified hostnames seen for the first time:

import json

# Get newly observed hostnames
api = API('your_username', 'your_api_key')
result = api.noh(after=-3600)

# Feed responses are generators that yield JSON strings
for json_string in result.response():
    record = json.loads(json_string)
    print(f"Host: {record['host']}")
    break

Domain discovery feed¶

New domains from registration, DNS, and third-party sources:

import json

# Get newly discovered domains
api = API('your_username', 'your_api_key')
result = api.domaindiscovery(after=-3600)

# Feed responses are generators that yield JSON strings
for json_string in result.response():
    record = json.loads(json_string)
    print(f"Domain: {record['domain']}")
    break

Domain RDAP feed¶

Registration data via RDAP protocol:

import json

# Get RDAP registration data
api = API('your_username', 'your_api_key')
result = api.domainrdap(after=-3600)

# Feed responses are generators that yield JSON strings
for json_string in result.response():
    record = json.loads(json_string)
    print(f"Domain: {record['domain']}")
    break

Other feeds¶

# Realtime Domain Risk
result = api.realtime_domain_risk(sessionID='my-session')

# Domain Hotlist
result = api.domainhotlist(sessionID='my-session')

Handling feed data¶

Single request (small data)¶

When data fits within the maximum result size:

result = api.nod(after=-60)

# Status is None for feeds (correct behavior)
assert result.status is None

# Iterate over generator
for record in result.response():
    print(f"Domain: {record['domain']}")

Multiple requests (large data)¶

When data exceeds the maximum result size, the feed automatically makes multiple requests:

result = api.nod(after=-7200)  # 2 hours of data

# Iterate over all batches
for partial_result in result.response():
    for record in partial_result:
        print(f"Domain: {record['domain']}")

Filtering results¶

Filter feed data by domain pattern:

# Filter by domain pattern
result = api.nod(domain='*test*', after=-3600)

for record in result.response():
    print(f"Matching domain: {record['domain']}")
    break

Feed endpoints¶

Choose between real-time API streaming or S3 download:

# Real-time API (default)
api.nod(sessionID='my-session')

# S3 download endpoint
api.nod(endpoint='download', sessionID='my-session')

Authentication notes¶

Threat Feeds use header-based authentication automatically - you don't need to configure anything special.

# Standard initialization works for all feeds
api = API('your_username', 'your_api_key')

# Use any feed endpoint
result = api.nod(after=-3600)
result = api.domainrdap(after=-3600)
result = api.domainhotlist(sessionID='my-session')

Technical detail: The SDK automatically detects feed endpoints and switches from HMAC signing to header-based authentication. This happens transparently in the background.

Best practices¶

Use session-based polling¶

Session-based polling is the most efficient way to consume feeds:

# Initialize session
session_id = 'my-app-nod-feed'

# Poll regularly (e.g., every 5 minutes)
while True:
    results = api.nod(sessionID=session_id)

    for record in results.response():
        process_domain(record['domain'])

    time.sleep(300)  # Wait 5 minutes

Handle rate limits¶

Feeds have rate limits based on your account:

from domaintools.exceptions import ServiceUnavailableException
import time

try:
    results = api.nod(sessionID='my-session')
    for record in results.response():
        process_record(record)
except ServiceUnavailableException as e:
    print("Rate limit exceeded, waiting...")
    time.sleep(60)

Process incrementally¶

Process feed data incrementally to avoid memory issues:

# Process records as they arrive
result = api.nod(sessionID='my-session')

for record in result.response():
    # Process immediately
    store_in_database(record)
    # Don't accumulate in memory

Common parameters¶

All feeds support these parameters:

sessionID - Session identifier for tracking position
after - Start time (seconds ago or ISO-8601)
before - End time (ISO-8601)
domain - Domain pattern filter
endpoint - API endpoint type ('api' or 'download')

Next steps¶

Iris Platform - Enrich feed data with Iris
Advanced Features - Feed-to-Enrich workflows
Configuration - Rate limits and authentication
Examples - Complete feed processing examples