API Reference

LeakIX Class

class leakpy.leakix.LeakIX(api_key=None, silent=True)[source]

Bases: object

Main class for interacting with LeakIX API.

DEFAULT_PAGES = 2
log(message, level='info')[source]

Log messages with proper formatting.

Parameters:

  • message (str) – Message to log.

  • level (str) – Log level (info, debug, warning, error).

has_api_key()[source]

Check if an API key is available and valid (48 characters).

save_api_key(api_key)[source]

Save the API key securely for later use.

Parameters:

api_key (str) – The API key to be saved.

delete_api_key()[source]

Delete the stored API key from secure storage.

get_plugins()[source]

Retrieve the list of available plugins from LeakIX.

get_cache_stats()[source]

Get cache statistics.

Returns:

Object containing cache statistics with attributes:

  • total_entries (int): Total number of cache entries

  • active_entries (int): Number of active (non-expired) entries

  • expired_entries (int): Number of expired entries

  • ttl_seconds (int): Current TTL in seconds

  • ttl_minutes (int): Current TTL in minutes

  • configured_ttl_minutes (int or None): Configured TTL in minutes

  • cache_file_size (int): Size of cache file in bytes

  • cache_file_path (str): Path to cache file

  • oldest_entry_age (float): Age of oldest entry in seconds

  • newest_entry_age (float): Age of newest entry in seconds

Return type:

DotNotationObject

Example

>>> client = LeakIX()
>>> stats = client.get_cache_stats()
>>> print(f"Cache has {stats.total_entries} entries")
clear_cache()[source]

Clear all cache entries.

Example

>>> client = LeakIX()
>>> client.clear_cache()
get_cache_ttl()[source]

Get the current cache TTL in minutes.

Returns:

TTL in minutes, or None if using default.

Return type:

int or None

Example

>>> client = LeakIX()
>>> ttl = client.get_cache_ttl()
>>> print(f"Cache TTL: {ttl} minutes")
set_cache_ttl(minutes)[source]

Set the cache TTL (Time To Live) in minutes.

Parameters:

minutes (int) – TTL in minutes (must be positive).

Raises:

ValueError – If minutes is not a positive integer.

Example

>>> client = LeakIX()
>>> client.set_cache_ttl(10)  # Set TTL to 10 minutes
analyze_query_stats(results, fields=None, top=10, all_fields=False)[source]

Analyze query results and extract statistics.

Parameters:

  • results (iterable) – Iterable of results (list, generator, etc.) containing dicts or L9Event objects to analyze.

  • fields (str or list, optional) – Fields to analyze. Can be: - str: Comma-separated field paths (e.g., “protocol,geoip.country_name”) - list: List of field paths (strings) - None: Uses default fields

  • top (int, optional) – Maximum number of top values to return per field. Defaults to 10.

  • all_fields (bool, optional) – If True, analyze all available fields. Defaults to False.

Returns:

QueryStats object with dot notation access:

  • total (int): Total number of results

  • fields (QueryStats): QueryStats object mapping field names to their statistics. Each field contains a dict mapping values to their counts. Access with: stats.fields.protocol, stats.fields.port, etc.

Return type:

QueryStats

Examples

Using string field paths:

stats = client.analyze_query_stats(events, fields='protocol,port')

Using list of field paths:

stats = client.analyze_query_stats(events, fields=['protocol', 'geoip.country_name'])
get_host(ip, fields='full', output=None)[source]

Get host details for a specific IP address.

Parameters:

  • ip (str) – The IP address to query.

  • fields (str, optional) – Comma-separated list of fields to extract. Defaults to “full” to get all details.

  • output (str or file object, optional) – Path to file or file object to write results. If None, results are only returned. Defaults to None.

Returns:

Object with ‘services’ and ‘leaks’ attributes, each containing a list of L9Event objects.

Both can be None if no information was found.

Return type:

DotNotationObject

Example

>>> client = LeakIX()
>>> host_info = client.get_host("157.90.211.37")
>>> print(f"Services: {len(host_info.services or [])}")
>>> print(f"Leaks: {len(host_info.leaks or [])}")
>>>
>>> # Access service details
>>> if host_info.services:
...     for service in host_info.services:
...         print(f"{service.protocol}://{service.ip}:{service.port}")
get_domain(domain, fields='full', output=None)[source]

Get domain details for a specific domain name.

Parameters:

  • domain (str) – The domain name to query.

  • fields (str, optional) – Comma-separated list of fields to extract. Defaults to “full” to get all details.

  • output (str or file object, optional) – Path to file or file object to write results. If None, results are only returned. Defaults to None.

Returns:

Object with ‘services’ and ‘leaks’ attributes, each containing a list of L9Event objects.

Both can be None if no information was found.

Return type:

DotNotationObject

Example

>>> client = LeakIX()
>>> domain_info = client.get_domain("leakix.net")
>>> print(f"Services: {len(domain_info.services or [])}")
>>> print(f"Leaks: {len(domain_info.leaks or [])}")
>>>
>>> # Access service details
>>> if domain_info.services:
...     for service in domain_info.services:
...         print(f"{service.protocol}://{service.host}:{service.port}")
get_subdomains(domain, output=None)[source]

Get subdomains for a specific domain name.

Parameters:

  • domain (str) – The domain name to query.

  • output (str or file object, optional) – Path to file or file object to write results. If None, results are only returned. Defaults to None.

Returns:

List of DotNotationObject containing subdomain information.

Each object has ‘subdomain’, ‘distinct_ips’, and ‘last_seen’ attributes. Returns empty list if no subdomains found or on error.

Return type:

list

Example

>>> client = LeakIX()
>>> subdomains = client.get_subdomains("leakix.net")
>>> print(f"Found {len(subdomains)} subdomains")
>>>
>>> # Access subdomain details with dot notation
>>> for subdomain in subdomains:
...     print(f"{subdomain.subdomain} - {subdomain.distinct_ips} IPs")
process_and_print_data(data, fields, suppress_debug=False)[source]

Process the provided data, extract desired fields, and print the results.

Parameters:

  • data (list or dict) – Either a list of dictionaries or a single dictionary.

  • fields (str or None) – Comma-separated list of fields to extract.

  • suppress_debug (bool) – If True, suppress debug logs during processing.

Returns:

A list of dictionaries containing the extracted fields from the data.

Return type:

list

get_all_fields(data=None, current_path=None)[source]

Get all available field paths.

If no data is provided, returns all fields from the l9format schema (no API call needed). If data is provided, extracts fields from that data.

Parameters:

  • data (dict, optional) – The nested dictionary to extract field paths from. If None, returns all fields from l9format schema.

  • current_path (list, optional) – Current path being traversed. Defaults to None.

Returns:

A list of field paths sorted alphabetically.

Return type:

list[str]

Examples

>>> # Get all fields from schema (no API call needed)
>>> fields = client.get_all_fields()
>>> for field in sorted(fields):
...     print(field)

>>> # Get fields from a specific event
>>> events = list(client.search(scope="leak", query='+country:"France"', pages=1))
>>> if events:
...     fields = client.get_all_fields(events[0])
...     for field in sorted(fields):
...         print(field)
search(scope='leak', query='', pages=None, plugin='', fields='full', use_bulk=False, output=None)[source]

Search LeakIX for data based on provided parameters.

This is the main function for querying LeakIX. It returns a generator that streams events in real-time using HTTP streaming. Events are yielded as soon as they are received from the API, allowing you to process results immediately without waiting for all pages to be fetched.

Streaming Behavior:

  • Pagination mode (default): Events are streamed page by page. Each page is fetched, processed, and events are yielded immediately. You don’t wait for all pages before processing the first events. The pages parameter controls how many pages to fetch.

  • Bulk mode (use_bulk=True): Events are streamed line by line from the HTTP response using true HTTP streaming. Events are parsed and yielded as they arrive from the server. Note: In bulk mode, the pages parameter is ignored as all results are retrieved automatically.

Connection Management:

When you stop consuming the generator early (e.g., using break), the HTTP connection is automatically closed when the generator is garbage collected (which happens automatically in Python). This prevents unnecessary bandwidth and API quota consumption. For immediate closure, you can explicitly call generator.close() after stopping the loop.

Optionally writes them to a file if output is specified.

Parameters:

  • scope (str, optional) – The scope of the search. Defaults to “leak”.

  • query (str, optional) – Search query string. Defaults to an empty string.

  • pages (int, optional) – Number of pages to scrape. Defaults to 2 (or LeakIX.DEFAULT_PAGES). Note: This parameter is ignored when use_bulk=True (bulk mode retrieves all results automatically).

  • plugin (str, optional) – Comma-separated plugin names. Defaults to an empty string.

  • fields (str, optional) – Comma-separated list of fields to extract from results. Defaults to “full” (all fields).

  • use_bulk (bool, optional) – Whether to use the bulk functionality (requires Pro API). When True, retrieves all results automatically and ignores the pages parameter. Defaults to False.

  • output (str or file object, optional) – Path to file or file object (e.g., sys.stdout) to write results. If None, results are only returned. Defaults to None.

Returns:

A generator that yields L9Event objects in real-time via HTTP streaming.

Events are yielded as soon as they are received and processed. If you need a list, use: list(client.search(…))

Return type:

generator

Examples

>>> # Stream events page by page (events processed as they arrive)
>>> events = client.search(scope="leak", query='+country:"France"', pages=5)
>>> for event in events:
...     print(event.ip)  # Events are streamed in real-time, no waiting for all pages

>>> # Stream events with bulk mode (HTTP streaming, line by line)
>>> events = client.search(scope="leak", query='plugin:TraccarPlugin', use_bulk=True)
>>> for event in events:
...     print(event.ip)  # Events streamed directly from HTTP response

>>> # Get all events as a list (if needed)
>>> events = list(client.search(scope="leak", query='+country:"France"', pages=2))

>>> # Write to file while streaming (events still yielded)
>>> events = client.search(scope="leak", query='+country:"France"', pages=2, output="results.txt")
>>> for event in events:
...     print(event.ip)  # Events streamed and written to file simultaneously

>>> # Stream to stdout (for piping) - need to consume generator
>>> list(client.search(scope="leak", query='+country:"France"', pages=2, output=sys.stdout))

CLI Mapping

CLI to Library Method Mapping

CLI Command

Library Method

leakpy search

client.search()

leakpy lookup host

client.get_host()

leakpy lookup domain

client.get_domain()

leakpy lookup subdomains

client.get_subdomains()

leakpy list plugins

client.get_plugins()

leakpy list fields

client.get_all_fields()

leakpy config set

client.save_api_key()

leakpy cache clear

client.clear_cache()

leakpy cache set-ttl

client.set_cache_ttl(minutes)

leakpy cache show-ttl

client.get_cache_ttl()

leakpy stats cache

client.get_cache_stats()

leakpy stats query

client.analyze_query_stats()

L9Event Objects

Results are returned as L9Event objects with dot notation access:

from leakpy import LeakIX
client = LeakIX()
# Events are streamed in real-time - process them as they arrive
events = client.search(scope="leak", query='+country:"France"', pages=1)

for event in events:
    # Events are processed immediately, no waiting for all pages
    if event.ip and event.port and event.protocol:
        print(f"{event.protocol}://{event.ip}:{event.port}")
    if event.geoip and event.geoip.country_name:
        print(f"  Country: {event.geoip.country_name}")

See Available Fields for all available fields.