example_client.py

from datetime import datetime, timedelta

import decouple

from leakix import Client, Scope
from leakix.field import CountryField, Operator, PluginField, TimeField
from leakix.plugin import Plugin
from leakix.query import MustNotQuery, MustQuery, RawQuery

API_KEY = decouple.config("API_KEY")
CLIENT = Client(api_key=API_KEY)


def example_get_host_filter_plugin():
    response = CLIENT.get_host(ipv4="33.33.33.33")
    assert response.status_code() == 200


def example_get_service_filter_plugin():
    """
    Filter by fields. In this example, we want to have the NTLM services.
    A list of plugins can be found in leakix.plugin
    """
    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
    response = CLIENT.get_service(queries=[query_http_ntlm])
    assert response.status_code() == 200, response.status_code()
    # check we only get NTML related services
    assert all(i.tags == ["ntlm"] for i in response.json())


def example_get_service_filter_plugin_with_pagination():
    """
    Filter by fields. In this example, we want to have the NTLM services.
    A list of plugins can be found in leakix.plugin.
    Ask for page 1 (starts at 0)
    """
    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
    response = CLIENT.get_service(queries=[query_http_ntlm], page=1)
    assert response.status_code() == 200
    # check we only get NTML related services
    assert all(i.tags == ["ntlm"] for i in response.json())


def example_get_leaks_filter_multiple_plugins():
    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
    query_country = MustQuery(field=CountryField("France"))
    response = CLIENT.get_leak(queries=[query_http_ntlm, query_country])
    assert response.status_code() == 200, response.status_code()
    assert all(
        i.geoip.country_name == "France" and i.tags == ["ntlm"] for i in response.json()
    )


def example_get_leaks_multiple_filter_plugins_must_not():
    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
    query_country = MustNotQuery(field=CountryField("France"))
    response = CLIENT.get_leak(queries=[query_http_ntlm, query_country])
    assert response.status_code() == 200, response.status_code()
    assert all(
        i.geoip.country_name != "France" and i.tags == ["ntlm"] for i in response.json()
    )


def example_get_leak_raw_query():
    raw_query = '+plugin:HttpNTLM +country:"France"'
    query = RawQuery(raw_query)
    response = CLIENT.get_leak(queries=[query])
    assert response.status_code() == 200, response.status_code()
    assert all(
        i.geoip.country_name == "France" and i.tags == ["ntlm"] for i in response.json()
    )


def example_get_leak_plugins_with_time():
    query_plugin = MustQuery(field=PluginField(Plugin.GitConfigHttpPlugin))
    today = datetime.now()
    one_month_ago = today - timedelta(days=30)
    query_today = MustQuery(field=TimeField(today, Operator.StrictlySmaller))
    query_yesterday = MustQuery(
        field=TimeField(one_month_ago, Operator.StrictlyGreater)
    )
    queries = [query_today, query_yesterday, query_plugin]
    response = CLIENT.get_leak(queries=queries)
    assert response.status_code() == 200


def example_get_plugins():
    response = CLIENT.get_plugins()
    for p in response.json():
        print(p.name)
        print(p.description)


def example_bulk_export():
    raw_query = '+plugin:"SmbPlugin" +country:"Belgium"'
    query = RawQuery(raw_query)
    response = CLIENT.bulk_export([query])
    print(response.json())


def example_bulk_export_last_event():
    raw_query = '+plugin:"SmbPlugin" +country:"Belgium"'
    query = RawQuery(raw_query)
    response = CLIENT.bulk_export_last_event([query])
    print(response.json())


def example_bulk_service():
    raw_query = "+\"window.onload=function(){ url ='/webui';window.location.href=url;}\" +port:443"
    query = RawQuery(raw_query)
    response = CLIENT.bulk_service([query])
    print(response.json())


def example_get_subdomains():
    domain = "leakix.net"
    response = CLIENT.get_subdomains(domain)
    print(response.json())


def example_search_simple():
    """Simple search using query string syntax (same as the website)."""
    response = CLIENT.search("+plugin:GitConfigHttpPlugin", scope=Scope.LEAK)
    for event in response.json():
        print(event.ip)


def example_search_service():
    """Search for services with multiple filters."""
    response = CLIENT.search("+country:FR +port:22", scope=Scope.SERVICE)
    for event in response.json():
        print(event.ip, event.port)


def example_get_domain():
    """Get services and leaks for a domain."""
    response = CLIENT.get_domain("example.com")
    if response.is_success():
        print("Services:", response.json()["services"])
        print("Leaks:", response.json()["leaks"])


def example_bulk_export_stream():
    """Streaming bulk export - memory efficient for large datasets."""
    query = MustQuery(field=PluginField(Plugin.GitConfigHttpPlugin))
    for aggregation in CLIENT.bulk_export_stream(queries=[query]):
        for event in aggregation.events:
            print(event.ip)


if __name__ == "__main__":
    example_get_host_filter_plugin()
    example_get_service_filter_plugin()
    example_get_service_filter_plugin_with_pagination()
    example_get_leaks_filter_multiple_plugins()
    example_get_leaks_multiple_filter_plugins_must_not()
    example_get_leak_plugins_with_time()
    example_get_leak_raw_query()
    example_get_plugins()
    example_bulk_export()
    example_bulk_service()
    example_bulk_export_last_event()
    example_get_subdomains()
    example_search_simple()
    example_search_service()
    example_get_domain()
    example_bulk_export_stream()