maltego-transforms

pip install canari maltego-trx requests

my-transforms/
├── project.conf
├── src/
│   └── mytransforms/
│       ├── __init__.py
│       ├── transforms/
│       │   ├── __init__.py
│       │   ├── domain_to_emails.py
│       │   └── ip_to_asn.py
│       └── entities.py

# transforms/domain_to_emails.py
from maltego_trx.maltego import MaltegoMsg, MaltegoTransform
from maltego_trx.transform import DiscoverableTransform
import requests

class DomainToEmails(DiscoverableTransform):
    """Find email addresses associated with a domain using Hunter.io."""

    @classmethod
    def create_entities(cls, request: MaltegoMsg, response: MaltegoTransform):
        domain = request.Value
        api_key = request.getTransformSetting("hunter_api_key")

        url = f"https://api.hunter.io/v2/domain-search"
        params = {
            "domain": domain,
            "api_key": api_key,
            "limit": 100
        }

        try:
            r = requests.get(url, params=params, timeout=10)
            data = r.json()

            for email_data in data.get("data", {}).get("emails", []):
                email = response.addEntity("maltego.EmailAddress", email_data["value"])
                email.addProperty("confidence", "Confidence", "loose",
                                  str(email_data.get("confidence", 0)))
                if email_data.get("first_name"):
                    email.addProperty("firstname", "First Name", "loose",
                                      email_data["first_name"])
                if email_data.get("last_name"):
                    email.addProperty("lastname", "Last Name", "loose",
                                      email_data["last_name"])
                if email_data.get("position"):
                    email.addProperty("title", "Title", "loose",
                                      email_data["position"])

        except Exception as e:
            response.addUIMessage(f"Error: {str(e)}", messageType="PartialError")

# transforms/ip_to_asn.py
from maltego_trx.maltego import MaltegoMsg, MaltegoTransform
from maltego_trx.transform import DiscoverableTransform
import requests

class IPToASN(DiscoverableTransform):
    """Lookup ASN and network owner for an IP address."""

    @classmethod
    def create_entities(cls, request: MaltegoMsg, response: MaltegoTransform):
        ip = request.Value

        # Use ip-api.com (free, no key required)
        url = f"http://ip-api.com/json/{ip}?fields=status,org,as,asname,isp,country,city"

        try:
            r = requests.get(url, timeout=10)
            data = r.json()

            if data.get("status") == "success":
                # Add ASN as organization entity
                org_name = data.get("org", data.get("isp", "Unknown"))
                org = response.addEntity("maltego.Organization", org_name)
                org.addProperty("as", "AS Number", "loose", data.get("as", ""))
                org.addProperty("country", "Country", "loose", data.get("country", ""))
                org.addProperty("city", "City", "loose", data.get("city", ""))

                # Add location note
                response.addUIMessage(
                    f"ASN: {data.get('as')} | ISP: {data.get('isp')} | {data.get('country')}"
                )

        except Exception as e:
            response.addUIMessage(f"Error: {str(e)}", messageType="PartialError")

# transforms/domain_to_subdomains.py
from maltego_trx.maltego import MaltegoMsg, MaltegoTransform
from maltego_trx.transform import DiscoverableTransform
import requests

class DomainToSubdomains(DiscoverableTransform):
    """Find subdomains via certificate transparency logs (crt.sh)."""

    @classmethod
    def create_entities(cls, request: MaltegoMsg, response: MaltegoTransform):
        domain = request.Value
        url = f"https://crt.sh/?q=%.{domain}&output=json"

        try:
            r = requests.get(url, timeout=15)
            entries = r.json()

            seen = set()
            for entry in entries:
                name = entry.get("name_value", "").strip()
                for subdomain in name.split("\n"):
                    subdomain = subdomain.strip().lstrip("*.")
                    if subdomain and subdomain not in seen and domain in subdomain:
                        seen.add(subdomain)
                        entity = response.addEntity("maltego.Domain", subdomain)
                        entity.addProperty("issuer", "Issuer", "loose",
                                           entry.get("issuer_name", ""))

            response.addUIMessage(f"Found {len(seen)} unique subdomains")

        except Exception as e:
            response.addUIMessage(f"Error: {str(e)}", messageType="PartialError")

# runner.py
from maltego_trx.registry import register_transform_function
from maltego_trx.server import application
from transforms.domain_to_emails import DomainToEmails
from transforms.ip_to_asn import IPToASN
from transforms.domain_to_subdomains import DomainToSubdomains

register_transform_function(DomainToEmails)
register_transform_function(IPToASN)
register_transform_function(DomainToSubdomains)

if __name__ == "__main__":
    application.run(host="0.0.0.0", port=8080, debug=True)

python runner.py
# Transforms available at http://localhost:8080

import requests
import json

def run_transform_batch(targets: list, transform_url: str) -> dict:
    """Run a transform against multiple targets without Maltego UI."""
    results = {}

    for target in targets:
        payload = {
            "Value": target,
            "Type": "maltego.Domain",
            "Properties": {}
        }
        r = requests.post(transform_url, json=payload, timeout=30)
        results[target] = r.json()

    return results

# Example usage
domains = ["example.com", "target.org", "company.io"]
emails_found = run_transform_batch(
    domains,
    "http://localhost:8080/DomainToEmails"
)

for domain, result in emails_found.items():
    entities = result.get("MaltegoMessage", {}).get("MaltegoTransformResponseMessage", {}).get("UIMessages", [])
    print(f"{domain}: {len(entities)} results")