Automate Dark Web Alerts Into Your Security Stack

< 100 Lines

Code needed to build a fully functional dark web monitoring pipeline with automated alerting

Dark web monitoring APIs provide the raw threat intelligence. Your job as a developer is to transform that data into automated detection and response. This tutorial walks you through building a complete threat detection pipeline from scratch.

The examples show how to poll for records, filter relevant results, and pass them into an existing response workflow.

Prerequisites

Python 3.8+ or Node.js 16+
Dark web monitoring API key (we'll use AdverseMonitor as an example)
Basic understanding of REST APIs
Optional: Slack webhook URL for notifications

Step 1: API Authentication

Start with API authentication. Keep API keys in environment variables rather than hardcoding them.

Python

import os
import requests
from datetime import datetime, timedelta

# Load API key from environment
API_KEY = os.environ.get('DARKWEB_API_KEY')
API_BASE = 'https://api.adversemonitor.com/v1'

def get_headers():
    return {
        'Authorization': f'Bearer {API_KEY}',
        'Content-Type': 'application/json'
    }

# Test connection
def test_connection():
    response = requests.get(
        f'{API_BASE}/status',
        headers=get_headers()
    )
    return response.status_code == 200

JavaScript

const axios = require('axios');

const API_KEY = process.env.DARKWEB_API_KEY;
const API_BASE = 'https://api.adversemonitor.com/v1';

const apiClient = axios.create({
    baseURL: API_BASE,
    headers: {
        'Authorization': `Bearer ${API_KEY}`,
        'Content-Type': 'application/json'
    }
});

async function testConnection() {
    const response = await apiClient.get('/status');
    return response.status === 200;
}

Step 2: Building the Poller

The poller fetches new threats since the last check. We'll use a timestamp to track what we've already processed.

Python

import json
from pathlib import Path

STATE_FILE = Path('last_poll.json')

def get_last_poll_time():
    if STATE_FILE.exists():
        data = json.loads(STATE_FILE.read_text())
        return data.get('last_poll')
    # Default: 1 hour ago
    return (datetime.utcnow() - timedelta(hours=1)).isoformat()

def save_poll_time(timestamp):
    STATE_FILE.write_text(json.dumps({'last_poll': timestamp}))

def fetch_new_threats():
    since = get_last_poll_time()

    response = requests.get(
        f'{API_BASE}/threats',
        headers=get_headers(),
        params={'since': since, 'limit': 100}
    )

    if response.status_code == 200:
        threats = response.json().get('threats', [])

        if threats:
            # Save timestamp of newest threat
            newest = max(t['detected_at'] for t in threats)
            save_poll_time(newest)

        return threats

    return []

Step 3: Filtering Relevant Threats

Not every threat requires action. Filter based on your organization's domains and risk tolerance.

Python

# Configuration
MONITORED_DOMAINS = ['company.com', 'company.co.uk']
MIN_SEVERITY = 'medium'  # low, medium, high, critical

SEVERITY_LEVELS = {
    'low': 1,
    'medium': 2,
    'high': 3,
    'critical': 4
}

def is_relevant(threat):
    # Check domain match
    affected_domains = threat.get('affected_domains', [])
    domain_match = any(
        any(d in affected for d in MONITORED_DOMAINS)
        for affected in affected_domains
    )

    # Check severity threshold
    severity = threat.get('severity', 'low')
    meets_severity = SEVERITY_LEVELS.get(severity, 0) >= SEVERITY_LEVELS[MIN_SEVERITY]

    return domain_match and meets_severity

def filter_threats(threats):
    return [t for t in threats if is_relevant(t)]

Step 4: Sending Notifications

When a record meets the configured criteria, route it to the owner and urgency defined in your response workflow.

Python - Slack

SLACK_WEBHOOK = os.environ.get('SLACK_WEBHOOK_URL')

def send_slack_alert(threat):
    severity_emoji = {
        'critical': '🚨',
        'high': '⚠️',
        'medium': '📢',
        'low': 'ℹ️'
    }

    emoji = severity_emoji.get(threat['severity'], '📢')

    message = {
        'blocks': [
            {
                'type': 'header',
                'text': {
                    'type': 'plain_text',
                    'text': f'{emoji} Dark Web Alert: {threat["type"]}'
                }
            },
            {
                'type': 'section',
                'fields': [
                    {'type': 'mrkdwn', 'text': f'*Severity:* {threat["severity"]}'},
                    {'type': 'mrkdwn', 'text': f'*Detected:* {threat["detected_at"]}'},
                    {'type': 'mrkdwn', 'text': f'*Source:* {threat["source"]}'}
                ]
            },
            {
                'type': 'section',
                'text': {
                    'type': 'mrkdwn',
                    'text': threat.get('summary', 'No summary available')
                }
            }
        ]
    }

    requests.post(SLACK_WEBHOOK, json=message)

Step 5: Automated Response

The real power is automated response. Here's an example that forces password resets for exposed credentials.

Python - Okta Integration

OKTA_DOMAIN = os.environ.get('OKTA_DOMAIN')
OKTA_TOKEN = os.environ.get('OKTA_API_TOKEN')

def force_password_reset(email):
    # Find user in Okta
    response = requests.get(
        f'https://{OKTA_DOMAIN}/api/v1/users',
        params={'search': f'profile.email eq "{email}"'},
        headers={'Authorization': f'SSWS {OKTA_TOKEN}'}
    )

    users = response.json()
    if not users:
        return False

    user_id = users[0]['id']

    # Expire password
    requests.post(
        f'https://{OKTA_DOMAIN}/api/v1/users/{user_id}/lifecycle/expire_password',
        headers={'Authorization': f'SSWS {OKTA_TOKEN}'}
    )

    return True

def handle_credential_leak(threat):
    if threat['type'] != 'credential_leak':
        return

    exposed_emails = threat.get('exposed_credentials', [])

    for cred in exposed_emails:
        email = cred.get('email')
        if email and any(d in email for d in MONITORED_DOMAINS):
            success = force_password_reset(email)
            if success:
                print(f'Password reset forced for: {email}')

Step 6: Complete Pipeline

Here's the complete pipeline that runs on a schedule:

Python - Full Pipeline

import time
import schedule

def run_pipeline():
    print(f'[{datetime.utcnow()}] Running threat detection pipeline...')

    # 1. Fetch new threats
    threats = fetch_new_threats()
    print(f'  Found {len(threats)} new threats')

    # 2. Filter relevant threats
    relevant = filter_threats(threats)
    print(f'  {len(relevant)} threats match our criteria')

    # 3. Process each threat
    for threat in relevant:
        # Send notification
        send_slack_alert(threat)

        # Handle credential leaks specially
        if threat['type'] == 'credential_leak':
            handle_credential_leak(threat)

    print('  Pipeline complete')

# Run every 5 minutes
schedule.every(5).minutes.do(run_pipeline)

# Initial run
run_pipeline()

# Keep running
while True:
    schedule.run_pending()
    time.sleep(1)

Webhook Alternative

If the API supports webhooks, it can deliver eligible events without a polling loop. Measure the full path from source collection to receipt before setting a latency expectation.

Python - Flask Webhook

from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)
WEBHOOK_SECRET = os.environ.get('WEBHOOK_SECRET')

def verify_signature(payload, signature):
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)

@app.route('/webhook/threats', methods=['POST'])
def receive_threat():
    # Verify webhook signature
    signature = request.headers.get('X-Signature')
    if not verify_signature(request.data, signature):
        return jsonify({'error': 'Invalid signature'}), 401

    threat = request.json

    # Route the record through your review workflow
    if is_relevant(threat):
        send_slack_alert(threat)
        handle_credential_leak(threat)

    return jsonify({'status': 'processed'}), 200

if __name__ == '__main__':
    app.run(port=5000)

Ready to Build Your Pipeline?

Check a domain against the current index first, review the returned records, and evaluate API access only when it fits a verified workflow.

Check a Domain

Next Steps

Add SIEM integration: Send events to Splunk, Sentinel, or Elastic
Create tickets: Automatically open Jira or ServiceNow tickets
Build dashboards: Store threats in a database for trend analysis
Add enrichment: Cross-reference with VirusTotal or Shodan
Deploy to cloud: Run on AWS Lambda, Azure Functions, or GCP

Conclusion

Dark web API automation connects several security systems and should be tested as an integration pattern, not copied directly into production. Adapt authentication, filtering, error handling, and response controls to your environment.

Automation can route a new credential-leak record into an existing response workflow instead of relying on someone to notice a dashboard update.

Building Automated Threat Detection with Dark Web Monitoring APIs

Executive summary

Prerequisites

Step 1: API Authentication

Step 2: Building the Poller

Step 3: Filtering Relevant Threats

Step 4: Sending Notifications

Step 5: Automated Response

Step 6: Complete Pipeline

Webhook Alternative

Ready to Build Your Pipeline?

Next Steps

Conclusion