Guide: Monitoring for Brand Impersonation

This guide provides a practical, step-by-step approach to monitoring for brand impersonation and phishing threats using the FullHunt API. We will write a Python script that regularly checks for typosquatting and potentially malicious domains related to your brand.

This guide covers:

Querying multiple API endpoints.
Handling paginated results.
Implementing robust error handling.
Basic logging and output.

Prerequisites

A FullHunt Enterprise API Key.
Python 3.6+ with the requests library installed (pip install requests).
Your company's primary domain to monitor (e.g., acme.com).

The Scenario

As an organization, you want to proactively identify domains that could be used to impersonate your brand and launch phishing attacks against your customers or employees. Two common vectors for this are typosquatting (e.g., acmme.com) and the registration of suspicious domains (e.g., acme-support.com).

Our script will perform the following steps:

Fetch all potential phishing domains targeting our brand.
Fetch all detected typosquatting domains.
Handle pagination to retrieve all results.
Log the findings to the console.

The Script

Here is the complete Python script. We will break it down section by section below.

import requests
import time
import logging

# --- Configuration ---
API_KEY = "YOUR_ENTERPRISE_API_KEY"  # Replace with your actual API key
BASE_URL = "https://fullhunt.io/api/v1/enterprise/darkweb"
HEADERS = {"X-API-KEY": API_KEY}
DOMAIN_TO_MONITOR = "acme.com" # Replace with the domain you want to monitor

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def fetch_paginated_results(endpoint: str, params: dict):
    """
    Fetches all results from a paginated FullHunt API endpoint.
    """
    all_results = []
    page = 1
    while True:
        params['page'] = page
        try:
            response = requests.get(endpoint, headers=HEADERS, params=params, timeout=30)

            # --- Error Handling ---
            if response.status_code == 429:
                logging.warning("Rate limit exceeded. Waiting for 60 seconds...")
                time.sleep(60)
                continue # Retry the same page

            if response.status_code != 200:
                logging.error(f"Error fetching data from {endpoint}. Status: {response.status_code}, Response: {response.text}")
                return None

            data = response.json()
            results = data.get('results', {}).get('items', [])
            
            if not results:
                break # No more results to fetch

            all_results.extend(results)
            
            # --- Check if there are more pages ---
            total_items = data.get('results', {}).get('total', 0)
            per_page = data.get('results', {}).get('per_page', len(results))
            if not per_page or len(all_results) >= total_items:
                break

            page += 1
            time.sleep(1) # Be a good API citizen

        except requests.exceptions.RequestException as e:
            logging.error(f"A network error occurred: {e}")
            return None
            
    return all_results


def main():
    """
    Main function to monitor for brand impersonation.
    """
    logging.info("Starting brand impersonation monitoring...")

    # 1. Fetch potential phishing domains
    logging.info(f"Checking for potential phishing domains targeting '{DOMAIN_TO_MONITOR}'...")
    phishing_params = {'query': DOMAIN_TO_MONITOR}
    phishing_domains = fetch_paginated_results(f"{BASE_URL}/potential-phishing", phishing_params)

    if phishing_domains is not None:
        if phishing_domains:
            logging.info(f"Found {len(phishing_domains)} potential phishing domains:")
            for domain in phishing_domains:
                print(f"  - {domain.get('domain_name')} (Last Seen: {domain.get('last_seen')})")
        else:
            logging.info("No potential phishing domains found.")

    # 2. Fetch typosquatting domains
    logging.info(f"\nChecking for typosquatting domains related to '{DOMAIN_TO_MONITOR}'...")
    typo_params = {'query': DOMAIN_TO_MONITOR}
    typo_domains = fetch_paginated_results(f"{BASE_URL}/typosquatting", typo_params)
    
    if typo_domains is not None:
        if typo_domains:
            logging.info(f"Found {len(typo_domains)} typosquatting domains:")
            for domain in typo_domains:
                print(f"  - {domain.get('domain_name')} (Last Seen: {domain.get('last_seen')})")
        else:
            logging.info("No typosquatting domains found.")

    logging.info("\nBrand impersonation monitoring check complete.")


if __name__ == "__main__":
    main()

Breakdown of the Script

Configuration

We start by defining our API key, the base URL, and the domain we want to monitor. Remember to replace "YOUR_ENTERPRISE_API_KEY" with a real key.

The `fetch_paginated_results` Function

This is the core of our script. It takes an API endpoint and a dictionary of parameters.

Looping and Pagination: It runs a while True loop, incrementing the page parameter with each request.
Error Handling: It explicitly checks for a 429 Too Many Requests status code and waits for a minute before retrying. It also checks for any other non-200 status code and logs the error. A try...except block handles network-level errors.
Result Extraction: It parses the JSON response and extracts the list of items from data['results']['items'].
Termination Condition: The loop breaks when the API returns an empty list of results, or when we have collected all the items indicated by the total count in the response.
Rate Limiting Courtesy: time.sleep(1) is included to avoid hitting the API too aggressively.

The `main` Function

This function orchestrates the process:

It calls fetch_paginated_results for both the /potential-phishing and /typosquatting endpoints, passing the domain to monitor as the query.
It checks if the API call was successful (i.e., not None) before processing results.
It then logs the discovered domains in a clean, human-readable format.

Next Steps and Further Reading

This script provides a strong foundation for an automated monitoring solution. You could expand it by:

Storing the results in a database to track changes over time.
Sending alerts (via email or Slack) when a new domain is discovered.
Integrating it with a SOAR platform to automate takedown requests.

For more information, refer to the following documents:

Prerequisites​

The Scenario​

The Script​

Breakdown of the Script​

Configuration​

The fetch_paginated_results Function​

The main Function​

Next Steps and Further Reading​