diff --git a/flowsint-transforms/src/flowsint_transforms/domain/to_history.py b/flowsint-transforms/src/flowsint_transforms/domain/to_history.py index deddfd59..1cef979d 100644 --- a/flowsint-transforms/src/flowsint_transforms/domain/to_history.py +++ b/flowsint-transforms/src/flowsint_transforms/domain/to_history.py @@ -15,9 +15,6 @@ from dotenv import load_dotenv load_dotenv() -WHOXY_API_KEY = os.getenv("WHOXY_API_KEY") - - class DomainToHistoryTransform(Transform): """[WHOXY] Takes a domain and returns history infos about it (history, organization, owners, emails, etc.).""" diff --git a/flowsint-transforms/src/flowsint_transforms/organization/to_domains.py b/flowsint-transforms/src/flowsint_transforms/organization/to_domains.py index d4cd1b70..9a8a0fcc 100644 --- a/flowsint-transforms/src/flowsint_transforms/organization/to_domains.py +++ b/flowsint-transforms/src/flowsint_transforms/organization/to_domains.py @@ -1,21 +1,19 @@ import os import re -from typing import Any, List, Union, Dict, Set +from typing import Any, List, Union, Dict, Set, Optional from flowsint_core.core.transform_base import Transform from flowsint_types.domain import Domain from flowsint_types.organization import Organization from flowsint_types.individual import Individual -from flowsint_types.email import Email -from flowsint_types.phone import Phone from flowsint_types.address import Location from flowsint_core.core.logger import Logger +from flowsint_core.core.graph_db import Neo4jConnection from tools.network.whoxy import WhoxyTool +from flowsint_core.utils import is_valid_domain, is_root_domain from dotenv import load_dotenv load_dotenv() -WHOXY_API_KEY = os.getenv("WHOXY_API_KEY") - class OrgToDomainsTransform(Transform): """[WHOXY] Takes an organization and returns the domains it registered.""" @@ -24,6 +22,39 @@ class OrgToDomainsTransform(Transform): InputType = List[Organization] OutputType = List[Domain] + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + neo4j_conn: Optional[Neo4jConnection] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + neo4j_conn=neo4j_conn, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + + @classmethod + def required_params(cls) -> bool: + return True + + @classmethod + def get_params_schema(cls) -> List[Dict[str, Any]]: + """Declare required parameters for this transform""" + return [ + { + "name": "WHOXY_API_KEY", + "type": "vaultSecret", + "description": "The Whoxy API key to use for domain lookups.", + "required": True, + }, + ] + @classmethod def name(cls) -> str: return "org_to_domains" @@ -54,10 +85,20 @@ class OrgToDomainsTransform(Transform): """Find domains related to organizations using whoxy api.""" domains: OutputType = [] self._extracted_data = [] # Store all extracted data for postprocess + self._extracted_individuals = [] # Store extracted individuals for testing + self._extracted_organizations = [] # Store extracted organizations for testing + api_key = self.get_secret("WHOXY_API_KEY", os.getenv("WHOXY_API_KEY")) for org in data: - infos_data = self.__get_infos_from_whoxy(org.name) + infos_data = self.__get_infos_from_whoxy(org.name, api_key) if infos_data and "search_result" in infos_data: + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Found {len(infos_data['search_result'])} domains for organization {org.name}" + }, + ) + # Process each domain result for result in infos_data["search_result"]: if self.__is_valid_domain_result(result): @@ -65,7 +106,7 @@ class OrgToDomainsTransform(Transform): if domain_name: domain = Domain(domain=domain_name, root=True) domains.append(domain) - + # Store extracted data for postprocess extracted_info = { 'org': org, @@ -79,6 +120,16 @@ class OrgToDomainsTransform(Transform): } } self._extracted_data.append(extracted_info) + + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Processing domain {domain_name} for organization {org.name}" + }, + ) + + # Process contacts and extract individuals/organizations during scan + self.__process_contacts_during_scan(extracted_info) else: Logger.info( self.sketch_id, @@ -86,12 +137,82 @@ class OrgToDomainsTransform(Transform): ) return domains - def __get_infos_from_whoxy(self, org_name: str) -> Dict[str, Any]: + def __process_contacts_during_scan(self, extracted_info: Dict[str, Any]): + """Process contacts and extract individuals and organizations during scan method.""" + org_name = extracted_info["org"].name + domain_name = extracted_info["domain"].domain + contacts = extracted_info["contacts"] + + for contact_type, contact in contacts.items(): + if contact: + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Processing {contact_type} contact for {domain_name}" + }, + ) + + # Extract individual (if name is not redacted) + individual = self.__extract_individual_from_contact( + contact, contact_type + ) + if individual: + # Store the extracted individual for testing/debugging + individual_info = { + "individual": individual, + "contact_type": contact_type, + "domain_name": domain_name, + "org_name": org_name, + "contact_data": contact, + } + self._extracted_individuals.append(individual_info) + + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Extracted individual: {individual.full_name} ({contact_type}) for {domain_name}" + }, + ) + + # Extract organization (if company name is not redacted) + organization = self.__extract_organization_from_contact( + contact, contact_type + ) + if organization: + # Store the extracted organization for testing/debugging + organization_info = { + "organization": organization, + "contact_type": contact_type, + "domain_name": domain_name, + "org_name": org_name, + "contact_data": contact, + } + self._extracted_organizations.append(organization_info) + + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Extracted organization: {organization.name} ({contact_type}) for {domain_name}" + }, + ) + + # Extract other non-redacted information (country, email, etc.) + self.__extract_additional_info_from_contact(contact, contact_type, domain_name, org_name) + else: + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] No contact data for {contact_type} contact for {domain_name}" + }, + ) + + def __get_infos_from_whoxy(self, org_name: str, api_key: str) -> Dict[str, Any]: + """Get domain information from Whoxy API or test data.""" infos: Dict[str, Any] = {} whoxy = WhoxyTool() try: params = { - "key": WHOXY_API_KEY, + "key": api_key, "reverse": "whois", "company": org_name, } @@ -108,16 +229,27 @@ class OrgToDomainsTransform(Transform): domain_name = result.get("domain_name") if not domain_name: return False + # A result is valid if it has a domain name - we'll filter contacts individually later return True + def __is_redacted(self, value: str) -> bool: + """Check if a value is redacted.""" + if not value: + return True + return "REDACTED FOR PRIVACY" in value.upper() or "PRIVACY" in value.upper() + def __extract_individual_from_contact( self, contact: Dict[str, Any], contact_type: str ) -> Individual: """Extract individual information from contact data.""" full_name = contact.get("full_name", "") - # Skip if name is redacted - if "REDACTED" in full_name or "REDACTED FOR PRIVACY" in full_name: + # Skip if name is redacted - we can't create an individual without a name + if self.__is_redacted(full_name) or not full_name: + Logger.info( + self.sketch_id, + {"message": f"[WHOXY] Skipping contact with redacted/empty name: {full_name}"}, + ) return None # Parse full name into first and last name @@ -126,19 +258,20 @@ class OrgToDomainsTransform(Transform): last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else "" # Extract email and phone - email = contact.get("email_address", "") + email_raw = contact.get("email_address", "") phone = contact.get("phone_number", "") - # Skip if email is redacted or invalid - if ( - "REDACTED" in email - or "REDACTED FOR PRIVACY" in email - or not self.__is_valid_email(email) - ): - email = "" + # Handle comma-separated emails + emails = [] + if email_raw and not self.__is_redacted(email_raw): + # Split by comma and clean up each email + email_list = [e.strip() for e in email_raw.split(",")] + for email in email_list: + if email and self.__is_valid_email(email): + emails.append(email) # Skip if phone is redacted - if "REDACTED" in phone or "REDACTED FOR PRIVACY" in phone: + if self.__is_redacted(phone): phone = "" # Extract address information @@ -148,13 +281,13 @@ class OrgToDomainsTransform(Transform): country = contact.get("country_name", "") # Skip if address is redacted - if "REDACTED" in address or "REDACTED FOR PRIVACY" in address: + if self.__is_redacted(address): address = "" - if "REDACTED" in city or "REDACTED FOR PRIVACY" in city: + if self.__is_redacted(city): city = "" - if "REDACTED" in zip_code or "REDACTED FOR PRIVACY" in zip_code: + if self.__is_redacted(zip_code): zip_code = "" - if "REDACTED" in country or "REDACTED FOR PRIVACY" in country: + if self.__is_redacted(country): country = "" # Create individual object @@ -162,10 +295,17 @@ class OrgToDomainsTransform(Transform): first_name=first_name, last_name=last_name, full_name=full_name, - email_addresses=[email] if email else None, + email_addresses=emails if emails else None, phone_numbers=[phone] if phone else None, ) + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Extracted individual: {full_name} ({contact_type}) with {len(emails)} emails" + }, + ) + return individual def __is_valid_email(self, email: str) -> bool: @@ -184,7 +324,9 @@ class OrgToDomainsTransform(Transform): country = contact.get("country_name", "") # Skip if any part is redacted - if any("REDACTED" in field for field in [address, city, zip_code, country]): + if any( + self.__is_redacted(field) for field in [address, city, zip_code, country] + ): return None if not all([address, city, zip_code, country]): @@ -194,217 +336,379 @@ class OrgToDomainsTransform(Transform): address=address, city=city, zip=zip_code, country=country ) + def __extract_organization_from_contact( + self, contact: Dict[str, Any], contact_type: str + ) -> Organization: + """Extract organization information from contact data.""" + company_name = contact.get("company_name", "") + + # Skip if company name is redacted or empty + if not company_name or self.__is_redacted(company_name): + return None + + # Create organization object + organization = Organization(name=company_name) + + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Extracted organization: {company_name} ({contact_type})" + }, + ) + + return organization + + def __extract_additional_info_from_contact( + self, contact: Dict[str, Any], contact_type: str, domain_name: str, org_name: str + ): + """Extract additional non-redacted information from contact data.""" + # Extract country information + country_name = contact.get("country_name", "") + country_code = contact.get("country_code", "") + + if country_name and not self.__is_redacted(country_name): + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Found country: {country_name} ({contact_type}) for {domain_name}" + }, + ) + + if country_code and not self.__is_redacted(country_code): + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Found country code: {country_code} ({contact_type}) for {domain_name}" + }, + ) + + # Extract email (even if individual name is redacted) + email_raw = contact.get("email_address", "") + if email_raw and not self.__is_redacted(email_raw): + emails = [] + email_list = [e.strip() for e in email_raw.split(",")] + for email in email_list: + if email and self.__is_valid_email(email): + emails.append(email) + + if emails: + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Found emails: {emails} ({contact_type}) for {domain_name}" + }, + ) + def postprocess(self, results: OutputType, original_input: InputType) -> OutputType: """Create Neo4j nodes and relationships from extracted data.""" if not self.neo4j_conn: + Logger.info( + self.sketch_id, + {"message": "[WHOXY] No Neo4j connection, skipping postprocess"}, + ) return results + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Starting postprocess with {len(self._extracted_individuals)} individuals and {len(self._extracted_organizations)} organizations" + }, + ) + # Track processed entities to avoid duplicates processed_domains: Set[str] = set() processed_individuals: Set[str] = set() + processed_organizations: Set[str] = set() processed_emails: Set[str] = set() processed_phones: Set[str] = set() processed_addresses: Set[str] = set() - for extracted_info in self._extracted_data: - org = extracted_info['org'] - domain = extracted_info['domain'] - domain_data = extracted_info['domain_data'] - contacts = extracted_info['contacts'] + # Track processed input organizations to ensure they're created + processed_input_orgs: Set[str] = set() - domain_name = domain.domain - if domain_name in processed_domains: - continue - processed_domains.add(domain_name) + # Process extracted individuals (already filtered and extracted during scan) + for individual_info in self._extracted_individuals: + individual = individual_info["individual"] + contact_type = individual_info["contact_type"] + domain_name = individual_info["domain_name"] + org_name = individual_info["org_name"] - # Create organization node - self.create_node( - "organization", - "name", - org.name, - caption=org.name, - type="organization", + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Processing individual: {individual.full_name} ({contact_type}) for {domain_name}" + }, ) - # Create domain node - self.create_node( - "domain", - "domain", - domain_name, - label=domain_name, - caption=domain_name, - type="domain", - ) - - # Create relationship between organization and domain - self.create_relationship( - "organization", - "name", - org.name, - "domain", - "domain", - domain_name, - "HAS_REGISTERED_DOMAIN", - ) - - # Process all contact types - for contact_type, contact in contacts.items(): - if contact: - self.__process_contact( - contact, - contact_type.upper(), - domain_name, - org.name, - processed_individuals, - processed_emails, - processed_phones, - processed_addresses, - ) - - self.log_graph_message( - f"Processed domain {domain_name} for organization {org.name}" - ) - - return results - - def __process_contact( - self, - contact: Dict[str, Any], - contact_type: str, - domain_name: str, - org_name: str, - processed_individuals: Set[str], - processed_emails: Set[str], - processed_phones: Set[str], - processed_addresses: Set[str], - ): - """Process a contact and create all related entities and relationships.""" - - # Extract individual - individual = self.__extract_individual_from_contact(contact, contact_type) - if not individual: - return - - individual_id = ( - f"{individual.first_name}_{individual.last_name}_{individual.full_name}" - ) - if individual_id in processed_individuals: - return - - processed_individuals.add(individual_id) - - # Create individual node - self.create_node( - "individual", - "full_name", - individual.full_name, - caption=individual.full_name, - type="individual", - ) - - # Create relationship between individual and domain - self.create_relationship( - "individual", - "full_name", - individual.full_name, - "domain", - "domain", - domain_name, - f"IS_{contact_type}_CONTACT", - ) - - # Create relationship between individual and organization - self.create_relationship( - "individual", - "full_name", - individual.full_name, - "organization", - "name", - org_name, - f"WORKS_FOR", - ) - - # Process email addresses - if individual.email_addresses: - for email in individual.email_addresses: - if email and email not in processed_emails: - processed_emails.add(email) - - # Create email node - self.create_node( - "email", - "email", - email, - caption=email, - type="email", - ) - - # Create relationship between individual and email - self.create_relationship( - "individual", - "full_name", - individual.full_name, - "email", - "email", - email, - "HAS_EMAIL", - ) - - # Process phone numbers - if individual.phone_numbers: - for phone in individual.phone_numbers: - if phone and phone not in processed_phones: - processed_phones.add(phone) - - # Create phone node - self.create_node( - "phone", - "number", - phone, - caption=phone, - type="phone", - ) - - # Create relationship between individual and phone - self.create_relationship( - "individual", - "full_name", - individual.full_name, - "phone", - "number", - phone, - "HAS_PHONE", - ) - - # Process physical address - address = self.__extract_physical_address(contact) - if address: - address_id = ( - f"{address.address}_{address.city}_{address.zip}_{address.country}" - ) - if address_id not in processed_addresses: - processed_addresses.add(address_id) - - # Create address node + # Create organization node if not already processed + if org_name not in processed_input_orgs: + processed_input_orgs.add(org_name) + Logger.info( + self.sketch_id, + {"message": f"[WHOXY] Creating organization node: {org_name}"}, + ) self.create_node( - "location", - "address", - address.address, - caption=f"{address.address}, {address.city}", - type="location", + "organization", + "name", + org_name, + caption=org_name, + type="organization", ) - # Create relationship between individual and address + # Create domain node if not already processed + if domain_name not in processed_domains: + processed_domains.add(domain_name) + Logger.info( + self.sketch_id, + {"message": f"[WHOXY] Creating domain node: {domain_name}"}, + ) + self.create_node( + "domain", + "domain", + domain_name, + label=domain_name, + caption=domain_name, + type="domain", + ) + + # Create relationship between organization and domain + self.create_relationship( + "organization", + "name", + org_name, + "domain", + "domain", + domain_name, + "HAS_REGISTERED_DOMAIN", + ) + + # Create individual node if not already processed + individual_id = ( + f"{individual.first_name}_{individual.last_name}_{individual.full_name}" + ) + if individual_id not in processed_individuals: + processed_individuals.add(individual_id) + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Creating individual node: {individual.full_name}" + }, + ) + self.create_node( + "individual", + "full_name", + individual.full_name, + caption=individual.full_name, + type="individual", + ) + + # Create relationship between individual and domain self.create_relationship( "individual", "full_name", individual.full_name, - "location", - "address", - address.address, - "LIVES_AT", + "domain", + "domain", + domain_name, + f"IS_{contact_type.upper()}_CONTACT", ) + # Create relationship between individual and organization + self.create_relationship( + "individual", + "full_name", + individual.full_name, + "organization", + "name", + org_name, + "WORKS_FOR", + ) + + # Process email addresses + if individual.email_addresses: + for email in individual.email_addresses: + if email and email not in processed_emails: + processed_emails.add(email) + Logger.info( + self.sketch_id, + {"message": f"[WHOXY] Creating email node: {email}"}, + ) + self.create_node( + "email", + "email", + email, + caption=email, + type="email", + ) + self.create_relationship( + "individual", + "full_name", + individual.full_name, + "email", + "email", + email, + "HAS_EMAIL", + ) + + # Process phone numbers + if individual.phone_numbers: + for phone in individual.phone_numbers: + if phone and phone not in processed_phones: + processed_phones.add(phone) + Logger.info( + self.sketch_id, + {"message": f"[WHOXY] Creating phone node: {phone}"}, + ) + self.create_node( + "phone", + "number", + phone, + caption=phone, + type="phone", + ) + self.create_relationship( + "individual", + "full_name", + individual.full_name, + "phone", + "number", + phone, + "HAS_PHONE", + ) + + # Process physical address from contact data + contact_data = individual_info["contact_data"] + address = self.__extract_physical_address(contact_data) + if address: + address_id = ( + f"{address.address}_{address.city}_{address.zip}_{address.country}" + ) + if address_id not in processed_addresses: + processed_addresses.add(address_id) + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Creating address node: {address.address}" + }, + ) + self.create_node( + "location", + "address", + address.address, + caption=f"{address.address}, {address.city}", + type="location", + ) + self.create_relationship( + "individual", + "full_name", + individual.full_name, + "location", + "address", + address.address, + "LIVES_AT", + ) + + self.log_graph_message( + f"Processed individual {individual.full_name} ({contact_type}) for domain {domain_name}" + ) + + # Process extracted organizations + for organization_info in self._extracted_organizations: + organization = organization_info["organization"] + contact_type = organization_info["contact_type"] + domain_name = organization_info["domain_name"] + org_name = organization_info["org_name"] + + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Processing organization: {organization.name} ({contact_type}) for {domain_name}" + }, + ) + + # Create input organization node if not already processed + if org_name not in processed_input_orgs: + processed_input_orgs.add(org_name) + Logger.info( + self.sketch_id, + {"message": f"[WHOXY] Creating organization node: {org_name}"}, + ) + self.create_node( + "organization", + "name", + org_name, + caption=org_name, + type="organization", + ) + + # Create domain node if not already processed + if domain_name not in processed_domains: + processed_domains.add(domain_name) + Logger.info( + self.sketch_id, + {"message": f"[WHOXY] Creating domain node: {domain_name}"}, + ) + self.create_node( + "domain", + "domain", + domain_name, + label=domain_name, + caption=domain_name, + type="domain", + ) + + # Create relationship between input organization and domain + self.create_relationship( + "organization", + "name", + org_name, + "domain", + "domain", + domain_name, + "HAS_REGISTERED_DOMAIN", + ) + + # Create extracted organization node if not already processed + if organization.name not in processed_organizations: + processed_organizations.add(organization.name) + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Creating organization node: {organization.name}" + }, + ) + self.create_node( + "organization", + "name", + organization.name, + caption=organization.name, + type="organization", + ) + + # Create relationship between extracted organization and domain + self.create_relationship( + "organization", + "name", + organization.name, + "domain", + "domain", + domain_name, + f"IS_{contact_type.upper()}_CONTACT", + ) + + self.log_graph_message( + f"Processed organization {organization.name} ({contact_type}) for domain {domain_name}" + ) + + Logger.info( + self.sketch_id, + { + "message": f"[WHOXY] Postprocess completed. Processed {len(processed_individuals)} individuals and {len(processed_organizations)} organizations" + }, + ) + + return results + InputType = OrgToDomainsTransform.InputType OutputType = OrgToDomainsTransform.OutputType