Python SDK

pip install oathnet

from oathnet import OathNetClient

# Initialize client with API key
client = OathNetClient(api_key="your-api-key")

# Search breaches
result = client.search.breach("user@example.com")
print(f"Found {result.data.results_found} results")

for record in result.data.results:
    print(record.email, "password_present=", bool(record.password))

import os
from oathnet import OathNetClient

# Explicit API key (required)
client = OathNetClient(api_key="your-api-key")

# From environment variable
client = OathNetClient(api_key=os.environ["OATHNET_API_KEY"])

# Basic search
result = client.search.breach("user@example.com")

# With filters
result = client.search.breach(
    "user@example.com",
    dbnames="linkedin_2012,adobe_2013"
)

# Access results
print(f"Total: {result.data.results_found}")
print(f"Shown: {result.data.results_shown}")

for record in result.data.results:
    print(f"Email: {record.email}")
    print(f"Password present: {bool(record.password)}")
    print(f"Database: {record.dbname}")
    print("---")

# Pagination with cursor
if result.data.cursor:
    next_page = client.search.breach(
        "user@example.com",
        cursor=result.data.cursor
    )

# Create a search session for quota optimization.
result = client.search.init_session("user@example.com", search_type="email")

session_id = result.data.session.id
print(f"Session ID: {session_id}")
print(f"Status: {result.data.session.status}")
print(f"Search Type: {result.data.session.search_type}")
print(f"Expires: {result.data.session.expires_at}")

if result.data.summary:
    print(f"Available services: {result.data.summary.available_services}")

# Reuse the session ID on current V2 search calls.
breach_results = client.breach.search_breach_v2(
    q="user@example.com",
    search_id=session_id,
    page_size=25,
)
print(f"Breach items: {len(breach_results.data.items)}")

ai_filter = client.ai_filter.create(
    query="US gmail users with LinkedIn records after 2020",
    index="breach",
)

print(ai_filter.filter_id)
print(ai_filter.filter)

context = client.ai_filter.get_context(ai_filter.filter_id)
print(context.index_type)
print(context.query)

oathnet ai-filter create --index breach --query "US gmail users"
oathnet ai-filter context 0123456789abcdef01234567

# Simple pivots stay readable and shareable
result = client.breach.search(
    q="user@example.com",
    dbname=["linkedin_2012"],
    fields=["email", "password", "dbname"],
    page_size=25,
)

# The same method accepts structured filters
filtered = client.breach.search(
    filter={
        "and": [
            {"field": "email_domain", "operator": "eq", "value": "example.com"},
            {"field": "country", "operator": "eq", "value": "US"},
        ]
    },
    fields=["email", "username", "dbname"],
)

# Use autocomplete before building a narrow query or scanner query_config
values = client.breach.autocomplete_values(
    field="email_domain",
    q="example",
    limit=10,
    include_info=True,
)
dbnames = client.breach.autocomplete_dbnames(q="link", limit=10)
field_coverage = client.breach.autocomplete_fields(
    field="discord_id",
    limit=10,
)

# Basic search
result = client.stealer.search("user@example.com")

# Advanced filtering
result = client.stealer.search(
    "user@example.com",
    domain=["google.com", "facebook.com"],
    has_log_id=True,
    page_size=50
)

# Access items
for item in result.data.items:
    print(f"URL: {item.url}")
    print(f"Username: {item.username}")
    print(f"Password present: {bool(item.password)}")
    print(f"Log ID: {item.log_id}")
    print("---")

# Pagination
if result.data.next_cursor:
    next_page = client.stealer.search(
        "user@example.com",
        cursor=result.data.next_cursor
    )

# Subdomain extraction with optional live checks and session reuse
result = client.stealer.subdomain(
    domain="example.com",
    q="mail",
    alive=True,
    search_id="sess_0123456789abcdef",
)
print(f"Found {result.data.count} subdomains")
for sub in result.data.subdomains:
    print(f"  {sub}")

# Structured filters use the same search method
filtered = client.stealer.search(
    filter={"field": "domain", "operator": "eq", "value": "example.com"},
    fields=["log_id", "url_str", "username"],
    view="enriched",
)

investigation = client.stealer.investigate(
    q="example.com",
    scope="all",
    include=["credentials", "victims", "files"],
    page_size=25,
    view="enriched",
)

investigation_from_body = client.stealer.investigate(
    q="example.com",
    scope="all",
    include=["credentials"],
    filters={"credentials": {"domain": "example.com"}},
)

for item in investigation.data.sections.credentials.items:
    print(item.log_id, item.username)

phonebook = client.stealer.phonebook(
    domain="example.com",
    alive=True,
)
print(len(phonebook.data.subdomains), phonebook.data.email_count)

# Search victims
result = client.victims.search("user@example.com")

for victim in result.data.items:
    print(f"Log ID: {victim.log_id}")
    print(f"Users: {victim.device_users}")
    print(f"IPs: {victim.device_ips}")
    print(f"Documents: {victim.total_docs}")
    print("---")

# Use the same method when the pivot comes from an AI/manual structured filter
victims = client.victims.search(
    filter={"field": "service", "operator": "eq", "value": "discord"},
    view="enriched",
)

# Search victim properties globally, then inspect one selected log
properties = client.victims.search_properties(
    q="example.com",
    service="discord",
    confidence="high",
    exclude_cookie_evidence=True,
)
properties_from_body = client.victims.search_properties(
    q="example.com",
    service="discord",
    confidence=["high", "medium"],
)
print(len(properties.data.items), len(properties_from_body.data.items))

details = client.victims.get_properties(
    "log_id_here",
    service="discord",
)
print(len(details.data.items))

summary = client.victims.get_summary("log_id_here")
cookies = client.victims.get_cookies(
    "log_id_here",
    domain="example.com",
    status="active",
)

# Raw cookie-domain inspection returns copyable text
cookie_text = client.victims.inspect_cookie_domain(
    "log_id_here",
    domain="example.com",
)
print(f"Cookie export characters: {len(cookie_text.copy_text or '')}")

# Get victim manifest (file tree) for the selected log
search_id = "sess_0123456789abcdef"
log_id = "log_id_here"
manifest = client.victims.get_manifest(log_id, search_id=search_id)
print(f"Root: {manifest.data.victim_tree.name}")

for child in (manifest.data.victim_tree.children or [])[:10]:
    print(f"  {child.name} ({child.type})")

# Get one file from the manifest. The SDK returns decoded text.
file_content = client.victims.get_file(
    log_id,
    "Browser Passwords/file 1.txt",
    search_id=search_id,
)
print(f"Downloaded file characters: {len(file_content.data.content or '')}")

# Download the full victim archive as a ZIP
client.victims.download_archive(
    log_id,
    "./victim_archive.zip",
    search_id=search_id,
)

# Search victim file metadata before fetching any file bytes
files = client.file_search.search_metadata(
    q="cookies",
    kind="cookies",
    page_size=25,
)
for file in files.data.items:
    print(file.log_id, file.file_id, file.path)

# Create a file search job
result = client.file_search.create(
    expression="password",
    log_ids=["log_id_1", "log_id_2"],
    search_mode="literal"  # literal, regex, or wildcard
)
job_id = result.data.job_id
print(f"Job ID: {job_id}")

# Check job status
status = client.file_search.get_status(job_id)
print(f"Status: {status.data.status}")
if status.data.summary:
    print(f"Files scanned: {status.data.summary.files_scanned}")
    print(f"Matches: {status.data.summary.matches}")

# Wait for completion and get results
result = client.file_search.search(
    expression="password",
    log_ids=["log_id_1", "log_id_2"],
    search_mode="literal",
    timeout=300  # seconds
)

if result.data.matches:
    for match in result.data.matches:
        print(f"File: {match.file_name}")
        print(f"Log ID: {match.log_id}")
        print(f"Match: {match.match_text}")
        print("---")

# List recent export jobs before creating a duplicate
exports = client.exports.list_exports_v2(page=1, page_size=20)
for job in exports.results:
    print(job.job_id, job.status)

# Create an export job from a structured breach query
result = client.exports.create_export_v2(
    export_type="breach",
    service="breach",
    limit=100,
    format="jsonl",  # json, jsonl, csv, txt, or html
    fields=["email", "password", "dbname"],
    query_config={
        "filter": {
            "field": "email_domain",
            "operator": "eq",
            "value": "example.com",
        }
    },
)
job_id = result.data.job_id
print(f"Export Job ID: {job_id}")

# Check export status
status = client.exports.get_export_v2(job_id)
print(f"Status: {status.data.status}")
if status.data.progress:
    print(f"Progress: {status.data.progress.percent}%")

# Wait for completion and download
result = client.exports.wait_for_completion(job_id, timeout=600)
if result.data.status == "completed":
    # Download the export file
    client.exports.download_export_v2(job_id, "./export.jsonl")

job = client.bulk_search.create(
    service="breach",
    terms=["alice@example.com", "bob@example.com"],
    format="jsonl",
    query_config={
        "filter": {
            "field": "email_domain",
            "operator": "eq",
            "value": "example.com",
        }
    },
    fields=["email", "password", "dbname"],
)

job_id = job.data.job_id
status = client.bulk_search.get_status(job_id)

jobs = client.bulk_search.list(page=1, page_size=20)
for item in jobs.results:
    print(item.job_id, item.status)

completed = client.bulk_search.wait_for_completion(job_id, timeout=600)
if completed.data.status == "completed":
    client.bulk_search.download(job_id, "./bulk-results.jsonl")

# Check scanner quota before creating a new monitor
quota = client.scanners.get_quota()
print(f"Remaining scanners: {quota.remaining}")

query_config = {
    "filter": {
        "and": [
            {"field": "email_domain", "operator": "eq", "value": "example.com"},
            {"field": "dbname", "operator": "in", "value": ["twitter.com", "linkedin.com"]},
        ]
    }
}

# Validate webhook or Discord delivery before saving the scanner.
# A 200 response can still have success=False when the remote endpoint rejects
# the test delivery, so inspect the returned object.
test = client.scanners.test_delivery(
    scanner_type="breach",
    query_config=query_config,
    notification_type="webhook",
    webhook_url="https://alerts.example.com/oathnet",
    webhook_security_mode="signed_json",
)
if not test.success:
    print(f"Delivery test failed: {test.message}")

scanner = client.scanners.create(
    name="Example breach monitor",
    scanner_type="breach",
    query_config=query_config,
    notification_type="webhook",
    webhook_url="https://alerts.example.com/oathnet",
    webhook_security_mode="signed_json",
    notify_on_zero_results=False,
)

security = client.scanners.get_webhook_security(scanner.uid)
if security.data:
    print(security.data.verification_method)

client.scanners.update(scanner.uid, name="Example breach monitor v2")
client.scanners.trigger(scanner.uid)

runs = client.scanners.list_runs(scanner.uid, limit=10)
for run in runs:
    print(f"{run.uid}: {run.status} ({run.results_count} results)")

client.scanners.pause(scanner.uid)
client.scanners.resume(scanner.uid)

# IP Info
result = client.osint.ip_info("8.8.8.8")
print(f"Location: {result.data.city}, {result.data.country}")
print(f"ISP: {result.data.isp}")
print(f"Proxy: {result.data.proxy}")

# Steam
result = client.osint.steam("76561198012345678")
print(f"Username: {result.data.username}")
print(f"Avatar: {result.data.avatar}")

# Xbox
result = client.osint.xbox("GamerTag123")
print(f"Username: {result.data.username}")
print(f"Avatar: {result.data.avatar}")

# Discord User Info
result = client.osint.discord_userinfo("123456789012345678")
print(f"Username: {result.data.username}")
print(f"Global Name: {result.data.global_name}")
print(f"Created: {result.data.creation_date}")

# Discord Username History
result = client.osint.discord_username_history("123456789012345678")
for entry in result.data.history:
    if entry.name and entry.time:
        print(f"{entry.name[0]} at {entry.time[0]}")

# Roblox User Info
result = client.osint.roblox_userinfo(user_id="123456789")
# Or by username:
result = client.osint.roblox_userinfo(username="PlayerName")
print(f"Username: {result.data.username}")
print(f"Display Name: {result.data.display_name}")

# Holehe - Email account detection
result = client.osint.holehe("user@example.com")
print(f"Found on {len(result.data.domains)} services:")
for domain in result.data.domains:
    print(f"  {domain}")

# GHunt - Google account lookup
result = client.osint.ghunt("user@gmail.com")
profile = (result.data.data or {}).get("profile", {}) if result.data else {}
if profile:
    print(f"Name: {profile.get('Name')}")

# Subdomain extraction
result = client.osint.extract_subdomain("example.com", is_alive=True)
for sub in result.data.subdomains:
    if isinstance(sub, str):
        print(sub)
    else:
        print(sub.get("subdomain", sub))

# Minecraft username history
result = client.osint.minecraft_history("PlayerName")
for entry in result.data.history:
    print(f"{entry.username} - {entry.changed_at}")

# Database name autocomplete
result = client.utility.dbname_autocomplete("link")
for name in result:
    print(name)  # linkedin_2012, linkedin_2021, etc.

from oathnet.exceptions import (
    OathNetError,
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    QuotaExceededError,
    ServiceUnavailableError
)

try:
    result = client.search.breach("user@example.com")

except AuthenticationError:
    print("Invalid API key")

except QuotaExceededError:
    print("Daily quota exceeded")

except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")

except NotFoundError as e:
    print(f"Resource not found: {e.message}")

except ValidationError as e:
    print(f"Invalid input: {e.message}")

except ServiceUnavailableError:
    print("Server error. Try again later.")

except OathNetError as e:
    print(f"API error: {e.message}")

from oathnet import OathNetClient

client = OathNetClient(
    api_key="your-api-key",
    base_url="https://oathnet.org/api",  # Custom base URL
    timeout=30.0                          # Request timeout in seconds
)

# Use context manager for automatic cleanup
with OathNetClient(api_key="your-api-key") as client:
    result = client.search.breach("user@example.com")
    print(result.data.results_found)
# Connection automatically closed